Memory Systems for Agents

The memory problem

An agent starts each turn with a context window. It can see the system prompt, the current user message, and previous messages in the conversation. But:

1. Conversation history grows: After 20 turns, the context is mostly old messages, leaving little room for new reasoning.
2. Repetition is wasteful: The agent re-reads the same facts every turn (e.g., user name, project details).
3. Long-term facts are lost: After the conversation ends, the agent forgets everything. If the user returns tomorrow with a follow-up question, the agent has no context.

Memory systems solve these by storing information outside the context window and retrieving it when needed.

Comparing memory system tradeoffs

The table below illustrates the tradeoffs across different memory approaches:

| Memory Type | Retention | Retrieval speed | Update cost | Semantic search | Best for | |-------------|-----------|-----------------|-------------|-----------------|----------| | Conversation buffer | Session only | O(1) | Negligible | No | Short conversations, <10 turns | | Summarized buffer | Session, compressed | O(1) | Summarization LLM call | No | Medium conversations, 10-50 turns | | Key-value store | Persistent | O(1) | Write to disk | No | Structured facts, exact lookups | | Vector-backed store | Persistent | O(n) with similarity | Write + embedding cost | Yes | Unstructured facts, flexible queries | | Hybrid (buffer + vector) | Session + persistent | O(1) + O(n) | Dual write cost | Yes | Long conversations with cross-session context |

Trade-off guidance: Use conversation buffer for single short conversations. For agents that run 20+ turns or span multiple sessions, add a persistent store. If you need semantic search ("find facts similar to X"), vector-backed storage is essential despite the higher embedding cost (~$0.00001-0.0001 per embedding with modern APIs).

Short-term memory: the conversation buffer

Basic approach

The simplest memory is a list of recent messages:

class ConversationMemory:
    def __init__(self, max_messages: int = 10):
        """Keep the last N messages in memory."""
        self.messages = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        """Add a message and trim old ones if needed."""
        self.messages.append({"role": role, "content": content})
        # Keep only the most recent messages
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_context(self) -> list:
        """Return messages to send to the model."""
        return self.messages

# Usage:
memory = ConversationMemory(max_messages=10)

# Turn 1
memory.add("user", "My name is Alice.")
response = call_model(memory.get_context())
memory.add("assistant", response)

# Turn 2
memory.add("user", "What's my name?")
response = call_model(memory.get_context())
# Model sees the earlier "My name is Alice" message
memory.add("assistant", response)

This works for short conversations (up to ~5–10 turns), but quickly hits limits:

• After 20 turns, the context is 90% conversation history and 10% room for new reasoning.
• The agent has no memory of conversations from yesterday.

Windowing and summarization

To extend this, you can summarize old messages:

class SummarizingMemory:
    def __init__(self, max_recent: int = 3, max_total: int = 15):
        """Keep recent messages in full; summarize older ones."""
        self.recent = []  # Full messages from last N turns
        self.summary = ""  # Compressed summary of earlier turns
        self.max_recent = max_recent
        self.max_total = max_total

    def add(self, role: str, content: str):
        """Add a message and summarize old ones if needed."""
        self.recent.append({"role": role, "content": content})

        # If we've exceeded max messages, summarize the oldest turn
        if len(self.recent) > self.max_total:
            # Move oldest message(s) to summary
            turn_to_summarize = self.recent.pop(0)
            summary_text = call_summarizer(
                turn_to_summarize["content"],
                existing_summary=self.summary
            )
            self.summary = summary_text

    def get_context(self) -> list:
        """Return full recent messages plus summary of older context."""
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"Earlier conversation summary:\n{self.summary}"
            })
        context.extend(self.recent)
        return context

def call_summarizer(message: str, existing_summary: str) -> str:
    """Use the model to compress a message into key facts."""
    prompt = f"""Existing summary:
{existing_summary}

New message to add to summary:
{message}

Update the summary to include any new facts from the message, omit repetition, keep under 200 words."""
    return model.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=256,
        messages=[{"role": "user", "content": prompt}]
    ).content[0].text

This approach:

• Keeps recent turns in full detail (full context).
• Compresses older turns into a summary (efficient).
• Balances short-term clarity with long-term context.

Long-term memory: persistent storage

Once a conversation ends, short-term memory vanishes. For agents that interact with returning users, you need persistent storage.

Key-value memory: storing facts

import json
import hashlib

class FactMemory:
    """Store named facts that persist across conversations."""

    def __init__(self, file_path: str = "agent_facts.json"):
        self.file_path = file_path
        self.facts = self._load()

    def _load(self) -> dict:
        """Load facts from disk."""
        try:
            with open(self.file_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}

    def _save(self):
        """Write facts to disk."""
        with open(self.file_path, 'w') as f:
            json.dump(self.facts, f, indent=2)

    def remember(self, key: str, value: str):
        """Store a fact: key -> value."""
        self.facts[key] = {
            "value": value,
            "timestamp": datetime.datetime.now().isoformat()
        }
        self._save()

    def recall(self, key: str) -> str | None:
        """Retrieve a fact."""
        if key in self.facts:
            return self.facts[key]["value"]
        return None

    def recall_all(self) -> str:
        """Return all facts as a formatted string for the prompt."""
        if not self.facts:
            return "(No stored facts)"
        lines = []
        for key, data in self.facts.items():
            lines.append(f"- {key}: {data['value']}")
        return "\n".join(lines)

# Usage:
fact_mem = FactMemory()

# In a conversation where the user says "I'm a software engineer."
fact_mem.remember("user_profession", "software engineer")

# In a later conversation, inject known facts into the prompt:
system_prompt = f"""You are a helpful assistant.

Known facts about the user:
{fact_mem.recall_all()}

Continue the conversation, using these facts to personalize responses."""

Limitations:

• No semantic understanding: You must remember the exact key (e.g., "user_profession"). If the user says "I'm a coder," you might store it separately unless you normalize the key.
• No similarity search: If you want to find facts related to a question, key-value lookup won't help.

Vector-backed memory: semantic search

For richer retrieval, store facts as vectors and search by meaning:

import numpy as np
from typing import List

class VectorMemory:
    """Store facts as embeddings; retrieve by semantic similarity."""

    def __init__(self, file_path: str = "vector_memory.json"):
        self.file_path = file_path
        self.memories = self._load()

    def _embed(self, text: str) -> list[float]:
        """Convert text to an embedding vector."""
        response = embed_client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def remember(self, fact: str):
        """Store a fact as text + embedding."""
        embedding = self._embed(fact)
        self.memories.append({
            "fact": fact,
            "embedding": embedding,
            "timestamp": datetime.datetime.now().isoformat()
        })
        self._save()

    def recall_similar(self, query: str, top_k: int = 3) -> List[str]:
        """Retrieve facts most similar to the query."""
        query_embedding = self._embed(query)

        # Compute similarity (cosine distance) with each stored fact
        similarities = []
        for mem in self.memories:
            similarity = np.dot(query_embedding, mem["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(mem["embedding"]) + 1e-8
            )
            similarities.append((mem["fact"], similarity))

        # Return top K by similarity
        similarities.sort(key=lambda x: x[1], reverse=True)
        return [fact for fact, score in similarities[:top_k]]

    def _load(self):
        try:
            with open(self.file_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return []

    def _save(self):
        with open(self.file_path, 'w') as f:
            json.dump(self.memories, f)

# Usage:
vec_mem = VectorMemory()

# Remember several facts
vec_mem.remember("Alice is a software engineer with 5 years of experience.")
vec_mem.remember("Alice prefers Python and TypeScript for most projects.")
vec_mem.remember("Alice's favorite database is PostgreSQL.")

# Later, retrieve facts related to a question:
query = "What programming languages does the user know?"
related = vec_mem.recall_similar(query, top_k=2)
# Returns: ["Alice prefers Python and TypeScript...", "Alice is a software engineer..."]

# Inject into prompt:
system_prompt = f"""You are a helpful assistant.

Relevant facts about the user:
{chr(10).join(related)}

Answer the user's question, drawing on these facts."""

This approach:

• Scales to many facts: You can store hundreds without worrying about exact key matching.
• Semantic search: "programmer" matches "software engineer" even though the exact words differ.
• Trade-off: Requires embeddings, which add latency and cost (~a few milliseconds per embed).

Combining short-term and long-term memory

A production agent often uses both:

class HybridMemory:
    def __init__(self):
        self.conversation_buffer = ConversationMemory(max_messages=6)
        self.fact_store = VectorMemory()

    def add_user_message(self, text: str):
        """Add to conversation and extract facts if needed."""
        self.conversation_buffer.add("user", text)

        # Optionally extract facts with the model:
        # fact = extract_facts_from_text(text)
        # if fact:
        #     self.fact_store.remember(fact)

    def add_assistant_message(self, text: str):
        self.conversation_buffer.add("assistant", text)

    def build_prompt(self, user_query: str) -> str:
        """Build a prompt with short-term and long-term context."""
        # Retrieve relevant long-term facts
        relevant_facts = self.fact_store.recall_similar(user_query, top_k=3)

        facts_section = ""
        if relevant_facts:
            facts_section = "Relevant facts:\n" + "\n".join(f"- {f}" for f in relevant_facts) + "\n\n"

        # Get recent conversation
        recent_context = "\n".join(
            f"{m['role']}: {m['content']}"
            for m in self.conversation_buffer.get_context()
        )

        return f"""{facts_section}Recent conversation:
{recent_context}"""

# Usage in agent loop:
memory = HybridMemory()
memory.add_user_message("My name is Bob, and I work in finance.")
response = model.messages.create(...)
memory.add_assistant_message(response)

# In a later turn:
memory.add_user_message("Can you help me analyze a spreadsheet?")
prompt = memory.build_prompt("help with spreadsheet")
response = model.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": prompt}]
)

Case study: Customer support agent with session recovery

Problem: A support agent handled customer issues across multiple conversations. When a customer returned after a week with a follow-up, the agent had no context about the original issue. The agent would ask the customer to repeat everything, leading to poor experience and duplicate work.

Solution: Build a hybrid memory system combining conversation buffers (for current session) with vector-backed persistent storage (for historical sessions).

class SupportAgentMemory:
    def __init__(self, customer_id: str):
        self.customer_id = customer_id
        self.current_session = ConversationMemory(max_messages=15)
        self.long_term = VectorMemory(f"customer_{customer_id}_facts.json")

    def start_session(self):
        """Load relevant facts from previous sessions."""
        recent_issues = self.long_term.recall_similar(
            "main issue and resolution",
            top_k=3
        )
        return recent_issues

    def end_session(self):
        """Extract and store key facts from this session."""
        # Use model to extract key facts from conversation
        facts_to_remember = self._extract_session_facts()
        for fact in facts_to_remember:
            self.long_term.remember(fact)

    def _extract_session_facts(self) -> list[str]:
        """Use the model to distill conversation into memorable facts."""
        messages_text = "\n".join([
            f"{m['role']}: {m['content']}"
            for m in self.current_session.get_context()
        ])

        prompt = f"""Extract key facts from this support conversation that would help a support agent in a future conversation with the same customer. Focus on:
- Main problem and root cause
- Solution implemented
- Customer preferences
- Any known issues or workarounds

Keep each fact to one sentence.

Conversation:
{messages_text}"""

        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=512,
            messages=[{"role": "user", "content": prompt}]
        )

        # Parse response into individual facts
        facts_text = response.content[0].text
        facts = [f.strip() for f in facts_text.split("\n") if f.strip()]
        return facts

# Usage in agent loop
def handle_customer_inquiry(customer_id: str, message: str) -> str:
    mem = SupportAgentMemory(customer_id)

    # Get historical context
    previous_sessions = mem.start_session()
    history_context = "\n".join(previous_sessions) if previous_sessions else "No previous interactions."

    # Build system prompt with history
    system_prompt = f"""You are a customer support agent.

CUSTOMER HISTORY:
{history_context}

Current session conversation history:
{chr(10).join(f'{m["role"]}: {m["content"]}' for m in mem.current_session.get_context())}"""

    # Get agent response
    response = model.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": message}
        ]
    )

    reply = response.content[0].text
    mem.current_session.add("user", message)
    mem.current_session.add("assistant", reply)

    return reply

# After session ends
# mem.end_session()  # Extracts and stores facts for next time

Result: When customers returned with follow-ups, the agent could recall previous issues and resolutions in seconds. Support quality improved; average resolution time dropped from 2 exchanges to 1.

Edge case: Memory coherence under concurrent updates

If multiple agent instances access the same persistent memory store (e.g., multiple support agents serving the same customer), concurrent writes can cause incoherent updates. For example:

• Agent A reads vector memory: "Customer prefers phone support"
• Agent B reads vector memory: same
• Agent A writes: "Customer prefers phone support AND email updates"
• Agent B writes: "Customer prefers phone support AND SMS updates" (overwrites A's update)

Result: Email preference is lost.

Mitigation strategies:

1. Last-write-wins (simple, risky): Accept the coherence loss for simplicity. Suitable if rare.

2. Append-only log (more robust): Instead of overwriting, append new facts with timestamps. During retrieval, merge conflicting facts intelligently:

class AppendOnlyMemory:
    def __init__(self, file_path: str):
        self.file_path = file_path
        self.log = self._load_log()

    def remember(self, fact: str, agent_id: str):
        """Append a fact with agent metadata."""
        entry = {
            "fact": fact,
            "agent_id": agent_id,
            "timestamp": datetime.now().isoformat()
        }
        self.log.append(entry)
        self._save_log()

    def recall_similar(self, query: str, top_k: int = 3) -> list[str]:
        """Retrieve non-conflicting facts, preferring recent ones."""
        # De-duplicate: if two agents stored conflicting facts, keep the newer
        seen_topics = {}
        for entry in reversed(self.log):  # Iterate newest first
            topic = self._extract_topic(entry["fact"])
            if topic not in seen_topics:
                seen_topics[topic] = entry["fact"]

        # Now do vector search on deduplicated facts
        facts = list(seen_topics.values())
        query_embedding = self._embed(query)
        similarities = [
            np.dot(query_embedding, self._embed(f)) for f in facts
        ]
        top_indices = sorted(
            range(len(similarities)),
            key=lambda i: similarities[i],
            reverse=True
        )[:top_k]
        return [facts[i] for i in top_indices]

    def _extract_topic(self, fact: str) -> str:
        """Simple topic extraction."""
        return fact.split(":")[0] if ":" in fact else fact[:20]

This prevents facts from being silently overwritten while maintaining retrieval speed.

Common mistake

Storing everything in long-term memory without filtering. Over time, the memory grows to contain redundant, outdated, or irrelevant facts. Before storing a fact, ask: "Will I need this again?" and "Is it still accurate?" Consider periodic cleanup or deprecation (marking facts as outdated and eventually removing them). Memory quality matters more than memory size.