Memory Usage & API¶

Core classes, storage backends, memory tools, search, and common patterns.

See Memory Overview for the memory architecture and how each system works.

Core Classes¶

Memory¶

The main memory interface with simple remember/recall API:

from cogent.memory import Memory

memory = Memory()

# Remember a value
await memory.remember("key", "value")
await memory.remember("user.name", "Alice")
await memory.remember("conversation.topic", "AI research")

# Recall a value
name = await memory.recall("user.name")  # "Alice"
missing = await memory.recall("unknown")  # None
missing = await memory.recall("unknown", default="N/A")  # "N/A"

# Check existence
exists = await memory.exists("user.name")  # True

# Delete a memory
await memory.forget("user.name")

# List all keys
keys = await memory.list_keys()  # ["conversation.topic"]

# Clear all memories
await memory.clear()

Scoped Memory¶

Create isolated memory views for users, teams, or conversations:

from cogent.memory import Memory

memory = Memory()

# Create scoped views
user_mem = memory.scoped("user:alice")
team_mem = memory.scoped("team:research")
conv_mem = memory.scoped("conv:thread-123")

# Each scope is isolated
await user_mem.remember("preference", "compact")
await team_mem.remember("preference", "detailed")

user_pref = await user_mem.recall("preference")  # "compact"
team_pref = await team_mem.recall("preference")  # "detailed"

# Scopes can be nested
project_mem = team_mem.scoped("project:alpha")
await project_mem.remember("status", "active")

Shared Memory Between Agents¶

Wire the same memory to multiple agents for shared knowledge:

from cogent import Agent
from cogent.memory import Memory

# Shared memory instance
shared = Memory()

# Both agents share the same memory
researcher = Agent(name="researcher", model=model, memory=shared)
writer = Agent(name="writer", model=model, memory=shared)

# Researcher stores findings
await shared.remember("findings", "Key insight: AI adoption is growing")

# Writer can access them
findings = await shared.recall("findings")

Storage Backends¶

InMemoryStore (Default)¶

Fast, no-persistence storage for development and testing:

from cogent.memory import Memory, InMemoryStore

# Default - uses InMemoryStore
memory = Memory()

# Explicit
memory = Memory(store=InMemoryStore())

SQLAlchemyStore¶

Persistent storage with SQLAlchemy 2.0 async support:

from cogent.memory import Memory, SQLAlchemyStore

# SQLite (local file)
store = SQLAlchemyStore("sqlite+aiosqlite:///./memory.db")
memory = Memory(store=store)

# PostgreSQL
store = SQLAlchemyStore(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=10,
)
memory = Memory(store=store)

# Initialize tables (run once)
await store.initialize()

# Cleanup
await store.close()

Context manager for cleanup:

async with SQLAlchemyStore("sqlite+aiosqlite:///./data.db") as store:
    memory = Memory(store=store)
    await memory.remember("key", "value")

RedisStore¶

Distributed cache with native TTL support:

from cogent.memory import Memory, RedisStore

store = RedisStore(
    url="redis://localhost:6379",
    prefix="myapp:",  # Key prefix
    default_ttl=3600,  # 1 hour default TTL
)
memory = Memory(store=store)

# With TTL per key
await memory.remember("session", {"user": "alice"}, ttl=1800)

Memory Key Search¶

Memory provides intelligent key search with three methods that automatically cascade:

1. Fuzzy Matching (Default - Fast & Free)¶

The default search method uses fuzzy string matching for instant, offline key discovery:

from cogent import Agent
from cogent.memory import Memory

# No special setup needed - fuzzy matching works out of the box
memory = Memory()

agent = Agent(name="assistant", model=model, memory=memory)

# Save memories
await agent.run("My name is Alice, I prefer dark mode, language is Python")

# Fuzzy matching finds similar keys instantly
await agent.run("What are my preferences?")
# → search_memories("preferences") finds "preferred_mode" and "preferred_language"
# Method: Fuzzy match (0.1ms, free, offline)

Benefits: - ⚡ 2,800× faster than semantic search (0.1ms vs 280ms) - 💰 Free - no API calls - 🔌 Works offline - no network required - 📊 62.5% accuracy - good enough for most use cases - 🧹 Smart normalization - handles underscores, hyphens, word order

How it works:

# String normalization helps matching:
"preferred_mode" → "preferred mode"
"user_timezone" → "user timezone"
"notification-settings" → "notification settings"

# Fuzzy matching finds similarity:
Query: "preferences" → Matches: "preferred mode", "preferred language"
Query: "contact" → Matches: "email", "phone number"
Query: "settings" → Matches: "notification settings"

2. Semantic Search (Optional Fallback)¶

Enable semantic search by adding a vectorstore (used when fuzzy matching unavailable):

from cogent import Agent
from cogent.memory import Memory
from cogent.vectorstore import VectorStore

# Add vectorstore for semantic fallback
memory = Memory(vectorstore=VectorStore())

agent = Agent(name="assistant", model=model, memory=memory)

When semantic search is used: - Fuzzy matching library (rapidfuzz) not installed - Fuzzy matching finds no matches (< 40% similarity)

Trade-offs: - ✅ 75% accuracy - better than fuzzy (but only 12.5% improvement) - ❌ 280ms avg - 2,800× slower than fuzzy - ❌ Costs money - OpenAI API calls - ❌ Requires network - API dependency

3. Keyword Search (Final Fallback)¶

Simple substring matching when all else fails:

# Query: "mode" → Matches keys containing "mode": "preferred_mode", "dark_mode"

Installation¶

Recommended (fuzzy matching):

uv add rapidfuzz  # For fast, free fuzzy matching

Optional (semantic fallback):

from cogent.memory import Memory
from cogent.vectorstore import VectorStore

memory = Memory(vectorstore=VectorStore())  # Enables semantic fallback

Performance Comparison¶

Method	Speed	Accuracy	Cost	Offline
Fuzzy	0.1ms	62.5%	Free	✅ Yes
Semantic	280ms	75.0%	$$ API	❌ No
Keyword	0.1ms	~30%	Free	✅ Yes

Recommendation: Use fuzzy matching (default) for 99% of use cases.

Example¶

See examples/memory/long_term.py for a complete demo.

from cogent import Agent
from cogent.memory import Memory

memory = Memory()  # Fuzzy matching by default

agent = Agent(name="assistant", model="gpt-5.4", memory=memory)

# Save with specific key names
await memory.remember("preferred_mode", "dark")
await memory.remember("preferred_language", "Python")
await memory.remember("email", "alice@example.com")

# Agent finds them with fuzzy matching (instant!)
await agent.run("What are my preferences?")
# → search_memories("preferences") finds "preferred_mode" and "preferred_language"
# ⚡ 0.1ms, free, offline

await agent.run("How can I contact the user?")
# → search_memories("contact") finds "email"

Memory Tools¶

Memory automatically exposes tools to agents for autonomous memory management:

from cogent import Agent
from cogent.memory import Memory

# Memory is always agentic - tools auto-added
memory = Memory()

agent = Agent(
    name="assistant",
    model=model,
    memory=memory,
)

# Agent has 5 memory tools available:
# 1. remember(key, value) - Save facts to long-term memory
# 2. recall(key) - Retrieve specific facts
# 3. forget(key) - Remove facts
# 4. search_memories(query) - Search long-term facts (fuzzy matching by default)
# 5. search_conversation(query) - Search conversation history

# Agent can now use memory tools autonomously
result = await agent.run("Remember that my name is Alice")
result = await agent.run("What's my name?")

Available Tools¶

1. remember(key, value) - Save important facts

# Agent automatically calls when user shares information
await agent.run("My favorite language is Python")
# → Agent calls: remember("favorite_language", "Python")

2. recall(key) - Retrieve specific saved facts

await agent.run("What's my favorite language?")
# → Agent calls: recall("favorite_language")

3. forget(key) - Remove facts (when user requests)

await agent.run("Forget my favorite language")
# → Agent calls: forget("favorite_language")

4. search_memories(query, k=5) - Search long-term facts with intelligent matching

# Default: Fast fuzzy matching (0.1ms, free, offline)
memory = Memory()
await agent.run("What are my preferences?")
# → Agent calls: search_memories("preferences")
# → Finds: "preferred_mode", "preferred_language" via fuzzy matching

# Optional: Add vectorstore for semantic fallback
memory = Memory(vectorstore=VectorStore())
# → Uses fuzzy matching first, falls back to semantic if needed

5. search_conversation(query, max_results=5) - Search conversation history

# Critical for long conversations exceeding context window
await agent.run("What were the three projects I mentioned earlier?")
# → Agent calls: search_conversation("three projects")

When Tools Are Used¶

The agent's system prompt instructs it to:

At conversation start → search_memories("user") to recall context
When user shares info → remember(key, value) immediately
When asked about something → Search before saying "I don't know"
For facts → search_memories(query) or recall(key)
For past conversation → search_conversation(query)
In long conversations → Use search_conversation() to find earlier context

Shorthand - memory=True creates a Memory instance:

# Shorthand for Memory()
agent = Agent(name="assistant", model=model, memory=True)

Usage Patterns¶

Conversation History¶

from cogent.memory import Memory

memory = Memory()

async def chat(user_id: str, message: str) -> str:
    user_mem = memory.scoped(f"user:{user_id}")

    # Load history
    history = await user_mem.recall("history", default=[])
    history.append({"role": "user", "content": message})

    # Get response (using agent)
    response = await agent.run(message, history=history)

    # Save updated history
    history.append({"role": "assistant", "content": response})
    await user_mem.remember("history", history)

    return response

Team Knowledge Base¶

from cogent.memory import Memory, SQLAlchemyStore
from cogent.vectorstore import VectorStore

# Persistent team memory with search
team_memory = Memory(
    store=SQLAlchemyStore("sqlite+aiosqlite:///./team.db"),
    vectorstore=VectorStore(),
)

# Store team knowledge
await team_memory.remember("policy:vacation", "Employees get 20 days PTO")
await team_memory.remember("policy:remote", "Remote work allowed 3 days/week")
await team_memory.remember("contact:hr", "hr@company.com")

# Search policies
results = await team_memory.search("time off work", k=3)

Agent with Persistent Context¶

from cogent import Agent
from cogent.memory import Memory, SQLAlchemyStore

store = SQLAlchemyStore("sqlite+aiosqlite:///./agent.db")
memory = Memory(store=store)

agent = Agent(
    name="assistant",
    model=model,
    memory=memory,
    instructions="""You have access to persistent memory.
    Use it to remember user preferences and context.""",
)

# First conversation
await agent.run("My favorite color is blue")

# Later conversation (same agent)
await agent.run("What's my favorite color?")  # Recalls "blue"

Store Protocol¶

Implement custom storage backends:

from typing import Protocol, Any

class Store(Protocol):
    """Protocol for memory storage backends."""

    async def get(self, key: str) -> Any | None:
        """Get a value by key."""
        ...

    async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
        """Set a value with optional TTL."""
        ...

    async def delete(self, key: str) -> bool:
        """Delete a key. Returns True if existed."""
        ...

    async def exists(self, key: str) -> bool:
        """Check if key exists."""
        ...

    async def keys(self, pattern: str = "*") -> list[str]:
        """List keys matching pattern."""
        ...

    async def clear(self) -> None:
        """Clear all keys."""
        ...

Custom implementation example:

class DynamoDBStore:
    """Custom DynamoDB backend."""

    def __init__(self, table_name: str):
        self.table_name = table_name
        self.client = boto3.resource("dynamodb")
        self.table = self.client.Table(table_name)

    async def get(self, key: str) -> Any | None:
        response = self.table.get_item(Key={"pk": key})
        item = response.get("Item")
        return item["value"] if item else None

    async def set(self, key: str, value: Any, ttl: int | None = None) -> None:
        item = {"pk": key, "value": value}
        if ttl:
            item["ttl"] = int(time.time()) + ttl
        self.table.put_item(Item=item)

    # ... implement other methods

# Use custom store
memory = Memory(store=DynamoDBStore("my-memories"))

API Reference¶

Memory¶

Method	Description
`remember(key, value, ttl?)`	Store a value
`recall(key, default?)`	Retrieve a value
`forget(key)`	Delete a value
`exists(key)`	Check if key exists
`list_keys(pattern?)`	List matching keys
`clear()`	Clear all memories
`scoped(prefix)`	Create scoped view
`search(query, k?)`	Semantic search (requires vectorstore)

Stores¶

Store	Use Case
`InMemoryStore`	Development, testing, ephemeral
`SQLAlchemyStore`	Persistent, ACID, SQL databases
`RedisStore`	Distributed, TTL, high-throughput