Memory

Long-term knowledge storage with semantic search for agents

Memory is a long-term knowledge storage system that enables agents to remember facts, preferences, and context across conversations. Unlike Session state (short-term), Memory provides persistent, searchable knowledge that agents build upon over time.

Key Characteristics

Long-Term Persistence - Knowledge survives across sessions and conversations
Semantic Search - Find relevant memories using natural language queries
Smart Ingestion - Automatically extract important facts using LLMs
Multiple Backends - InMemory (dev), Vector (semantic), Database (persistent)
Built on Entity - Inherits durability and consistency
Cross-Session - Shared knowledge accessible to all agents

Basic Usage

Creating Memory

Storing and Retrieving Memories

# Store individual memories
await memory.store(
    key="user_role",
    content="Senior Software Engineer specializing in distributed systems",
    type="user_info",
    confidence=0.95
)

await memory.store(
    key="project_context",
    content="Building a real-time analytics platform for financial data",
    type="project_info"
)

# Retrieve specific memories
memories = await memory.recall(["user_role", "project_context"])
for mem in memories:
    print(f"{mem.key}: {mem.content}")

Semantic Search

# Search using natural language
results = await memory.search(
    query="What does the user know about databases?",
    limit=5
)

for result in results:
    print(f"Score: {result.score:.2f}")
    print(f"Content: {result.content}")
    print(f"Source: {result.metadata.get('source_session')}")

Integration with Agents

from agnt5 import Agent, Session, Memory, LanguageModel

# Create memory
memory = Memory(service=VectorMemoryService())

# Store long-term knowledge
await memory.store("user_expertise", "PhD in Machine Learning, specializes in NLP")
await memory.store("preferred_tools", "Prefers PyTorch over TensorFlow")

# First conversation
session1 = Session(id="conv-001", user_id="researcher-123")
lm = LanguageModel()
agent = Agent(name="assistant", model=lm, memory=memory, session=session1)
await agent.run("Help me implement attention mechanisms")
# Agent recalls user's ML expertise and PyTorch preference

# Later conversation (different session)
session2 = Session(id="conv-042", user_id="researcher-123")
agent2 = Agent(name="assistant", model=lm, memory=memory, session=session2)
await agent2.run("Review my transformer code")
# Agent still remembers user's background and preferences

Smart Ingestion

Automatically extract and store important information from conversations:

# Agent conversation
session = Session(id="consultation-123", user_id="user-456")
agent = Agent(name="advisor", memory=memory, session=session)

await agent.run("I'm building a recommendation system for e-commerce")
await agent.run("We have 10 million users and need sub-100ms latency")
await agent.run("Our team is experienced with Python and Go")

# Extract and store important facts
memory_keys = await memory.ingest_from_session(
    session,
    strategy="smart"  # Uses LLM to identify important facts
)

# Memory now contains:
# - "User building recommendation system for e-commerce"
# - "System requirements: 10M users, <100ms latency"
# - "Team expertise: Python, Go"

# Future conversations automatically recall these facts

Ingestion Strategies

Choose the right strategy for your use case:

Common Patterns

User Profile Memory

Build comprehensive user profiles over time:

class UserProfileMemory:
    def __init__(self, user_id: str, memory: Memory):
        self.user_id = user_id
        self.memory = memory
        self.prefix = f"user_{user_id}_"

    async def store_preference(self, category: str, value: str):
        """Store user preference."""
        await self.memory.store(
            key=f"{self.prefix}pref_{category}",
            content=value,
            type="preference",
            user_id=self.user_id
        )

    async def store_expertise(self, domain: str, level: str, details: str):
        """Store user expertise."""
        await self.memory.store(
            key=f"{self.prefix}expertise_{domain}",
            content=f"{level} expertise in {domain}: {details}",
            type="expertise",
            user_id=self.user_id
        )

    async def get_profile(self) -> Dict[str, Any]:
        """Retrieve complete user profile."""
        results = await self.memory.search(
            query=f"user {self.user_id} profile preferences expertise",
            limit=50
        )

        profile = {
            "preferences": {},
            "expertise": {},
            "context": []
        }

        for result in results:
            if result.metadata.get("type") == "preference":
                category = result.key.replace(f"{self.prefix}pref_", "")
                profile["preferences"][category] = result.content
            elif result.metadata.get("type") == "expertise":
                domain = result.key.replace(f"{self.prefix}expertise_", "")
                profile["expertise"][domain] = result.content

        return profile

# Usage
user_memory = UserProfileMemory("user-123", memory)
await user_memory.store_preference("language", "Python")
await user_memory.store_expertise("ml", "advanced", "10+ years in NLP")

profile = await user_memory.get_profile()

Conversation Summarization

Archive long sessions as summaries:

async def archive_session_to_memory(
    session: Session,
    memory: Memory,
    summary_threshold: int = 50
):
    """Archive long sessions as summaries in memory."""

    history = await session.history()

    if len(history) > summary_threshold:
        # Generate comprehensive summary
        from agnt5 import LanguageModel

        lm = LanguageModel()
        conversation = "\n".join([
            f"{msg.role}: {msg.content}" for msg in history
        ])

        summary = await lm.generate(
            prompt=f"""Summarize this conversation in 2-3 paragraphs.
            Focus on key decisions, important facts, and outcomes.

            Conversation:
            {conversation}

            Summary:""",
            max_tokens=300
        )

        # Store summary in memory
        await memory.store(
            key=f"session_summary_{session.id}",
            content=summary.text,
            type="session_summary",
            session_id=session.id,
            user_id=session.user_id,
            message_count=len(history)
        )

        # Optionally prune session to save space
        await session.prune(strategy="keep_summary")

# Usage
await archive_session_to_memory(session, memory, summary_threshold=100)

Learning Agent

Agent that learns from every interaction:

class LearningAgent:
    def __init__(self, name: str, model, tools, memory: Memory):
        self.agent = Agent(name=name, model=model, tools=tools, memory=memory)
        self.memory = memory

    async def run_and_learn(self, prompt: str, session: Session):
        """Execute task and learn from interaction."""

        # Run agent
        result = await self.agent.run(prompt, session=session)

        # Extract and store learnings
        memory_keys = await self.memory.ingest_from_session(
            session,
            strategy="smart"
        )

        # Store outcome for future reference
        await self.memory.store(
            key=f"interaction_{session.id}",
            content=f"Task: {prompt}\nOutcome: {result.output}",
            type="interaction_history",
            success=result.status == "completed"
        )

        return result

    async def recall_similar_tasks(self, prompt: str) -> List[Dict]:
        """Find similar past interactions."""
        return await self.memory.search(
            query=f"Similar to: {prompt}",
            limit=5
        )

# Usage
learning_agent = LearningAgent("assistant", lm, tools, memory)

# Agent learns from each interaction
result = await learning_agent.run_and_learn(
    "Debug this performance issue",
    session
)

# Later: Recall similar past work
similar = await learning_agent.recall_similar_tasks(
    "Another performance problem"
)

Best Practices

1. Distinguish Memory from Session State

Use Memory for long-term, Session for short-term:

2. Add Metadata for Better Retrieval

Enrich memories with metadata:

await memory.store(
    key="technical_decision_001",
    content="Chose PostgreSQL for transactional data, Redis for caching",
    type="decision",
    category="architecture",
    confidence=0.9,
    source_session="planning-session-789",
    timestamp=datetime.now(),
    decision_maker="tech-lead-123",
    rationale="Need ACID guarantees and high read performance"
)

# Later: Search by metadata
postgres_decisions = await memory.search(
    query="database decisions",
    filter={"type": "decision", "category": "architecture"}
)

3. Implement Memory Maintenance

Manage memory lifecycle:

async def maintain_memory(memory: Memory, user_id: str):
    """Prune old or low-confidence memories."""

    # Remove outdated information
    await memory.forget([
        key for key in await memory.list_keys(user_id=user_id)
        if is_outdated(key)
    ])

    # Update confidence scores based on usage
    for key in await memory.list_keys(user_id=user_id):
        mem = await memory.recall([key])
        if mem[0].metadata.get("last_accessed"):
            days_since_access = calculate_days_since(
                mem[0].metadata["last_accessed"]
            )
            new_confidence = calculate_confidence_decay(
                mem[0].metadata.get("confidence", 1.0),
                days_since_access
            )
            await memory.update(key, confidence=new_confidence)

Session State vs Memory

Aspect	Session State	Memory
Lifetime	Single conversation	Indefinite
Scope	Session-specific	Cross-session
Search	Direct key access	Semantic search
Purpose	Current context	Long-term knowledge
Storage	Entity state	Entity + Vector DB
Example	Shopping cart	User preferences

When to use Session State:

Current conversation context
Temporary workflow state
UI state and navigation

When to use Memory:

User profile and preferences
Historical interactions
Domain knowledge
Learned facts and insights

Next Steps

Session - Short-term conversation state
Agent - Agents use Memory for context
Entity - Underlying primitive for Memory
Context API - Memory context operations