Chapter 06 · Memory Architectures
Your agent just had a fascinating conversation. Then you refreshed the page. It remembers nothing. Let's fix that.
Here is the conversation that just happened:
User: I'm building a CLI tool for managing Docker containers. It needs to
start, stop, and list containers. What's the best Python library?
Agent: Use the `docker` Python SDK. Here's a complete implementation:
[produces 80 lines of clean, working code with error handling]
User: Thanks, that's perfect. And also, can you add a feature that shows
resource usage per container?
Agent: I'd be happy to help! What kind of CLI tool are you building, and
what language or framework are you using?
The agent just asked "what language or framework are you using" -- thirty seconds after writing 80 lines of Python for the same project. This is not a bug. It is not a bad model. It is the fundamental architecture of every LLM API on the planet.
LLMs are stateless. Every API call is independent. The model does not remember you, your project, or the code it wrote three seconds ago. The only reason it appeared to remember anything during that first exchange is that YOU were maintaining the conversation array and feeding it back on every call. The moment that array is lost -- page refresh, server restart, new session -- the agent has total amnesia.
Memory is not a feature you get for free. Memory is something you build. And building it well is the difference between an agent that feels like a collaborator and an agent that feels like a search engine with a personality disorder.
What This Chapter Covers
This chapter builds the complete memory architecture for your agent. You will implement three tiers of memory -- short-term, working, and long-term -- and integrate them into the agent loop from Chapter 5. By the end, your agent will remember conversations across sessions, retrieve relevant knowledge from a vector database, and maintain a structured scratchpad for multi-step reasoning.
You will write real code. You will use ChromaDB for vector search, SQLite for structured memory, and a scratchpad pattern for working memory. Every piece is production-grade. Nothing is throwaway.
Section 1: The Memory Stack -- A Mental Model
Before you write code, you need a mental model. The architecture mirrors human cognition -- not because we are simulating brains, but because the same constraints apply: limited attention, fast recall of recent events, slow recall of distant ones, and the need to forget irrelevant information.
Here is the stack:
┌─────────────────────────────────────────────────────────┐
│ SHORT-TERM MEMORY │
│ Conversation history in the context window. │
│ Everything the agent can "see" right now. │
│ Capacity: ~200K tokens. Latency: ~0ms. │
│ Volatile -- gone when the conversation ends. │
├─────────────────────────────────────────────────────────┤
│ WORKING MEMORY │
│ Structured scratchpad for the current task. │
│ Plans, intermediate results, hypotheses. │
│ Capacity: ~10K tokens. Latency: ~0ms. │
│ Volatile -- cleared when the task completes. │
├─────────────────────────────────────────────────────────┤
│ LONG-TERM MEMORY │
│ Persistent storage across conversations. │
│ Vector DB, SQL, key-value stores. │
│ Capacity: unlimited. Latency: 10-100ms. │
│ Durable -- survives restarts and deployments. │
└─────────────────────────────────────────────────────────┘
Three tiers. Each has different characteristics along four dimensions:
| Tier | Capacity | Latency | Persistence | Searchability |
|---|---|---|---|---|
| Short-term | Limited by context window | Instant (in-context) | Volatile | Sequential (LLM reads it) |
| Working | ~10K tokens | Instant (in-context) | Volatile | Structured (keyed access) |
| Long-term | Unlimited | 10-100ms (DB query) | Durable | Semantic + keyword |
Short-term memory is the conversation. Every message -- user, assistant, tool call, tool result -- appended to the messages array and fed into the context window. The model sees everything in short-term memory. When the context fills, older entries get evicted. When the session ends, everything is gone.
Working memory is the agent's scratchpad -- structured state for the current task: plan steps, intermediate results, hypotheses, extracted facts. It lives in the context window but is formatted for the agent's own use. It is the agent thinking on paper.
Long-term memory survives across sessions. User preferences. Facts from past conversations. Documents the agent has read. It lives outside the context window -- in databases, vector stores, and file systems. The agent queries it when it needs information it does not currently have.
The key insight: Short-term memory is what the agent is thinking about right now. Working memory is what it is working on. Long-term memory is what it knows. Each tier serves a different purpose. Each has different performance characteristics. You need all three.
Section 2: Short-Term Memory -- Conversation Management
Short-term memory is the messages array -- the simplest form of memory and the one you have been using since Chapter 4. But "just append everything" stops working when conversations get long.
The Context Budget Problem
Every message consumes tokens. A single turn -- user message, assistant response, tool call and result -- can consume 2,000-5,000 tokens. Ten turns and you are at 50,000. Twenty turns and you have burned through half of a 200K context window. And you still need room for the system prompt, tool definitions, retrieved documents, and the model's response.
The context window is a budget. When you run out, you have three options: drop old messages (losing context), summarize (losing detail), or pay for a bigger window (losing money). There is no fourth option.
Strategy 1: Sliding Window
Keep the last N messages. Drop everything older. Simple, predictable, and brutal.
from typing import Literal
def sliding_window(
messages: list[dict],
max_messages: int = 20,
preserve_system: bool = True,
) -> list[dict]:
"""
Keep only the most recent `max_messages` messages.
Optionally preserve the system message at the front.
"""
if len(messages) <= max_messages:
return messages
system_msg = []
rest = messages
if preserve_system and messages[0]["role"] == "system":
system_msg = [messages[0]]
rest = messages[1:]
# Keep the most recent messages, minus any we reserved for system
kept = rest[-(max_messages - len(system_msg)):]
return system_msg + kept
# Usage: trim conversation to last 20 messages before each API call
conversation = sliding_window(conversation, max_messages=20)
The sliding window works for short-lived tasks where early context does not matter. It fails for long-running tasks where early context is critical -- the user says "I want to build a Docker CLI" on turn one, and on turn thirty the window has dropped that message.
Strategy 2: Summarization
Before the context fills up, summarize the conversation so far. Replace the full history with the summary. The agent retains the gist but loses the details.
def summarize_conversation(
messages: list[dict],
model: str = "claude-haiku-4-20250514",
keep_last_n: int = 6,
) -> list[dict]:
"""
Summarize older messages, keep the most recent ones verbatim.
Strategy:
1. Take all messages except the last `keep_last_n`.
2. Ask a cheap model to summarize them.
3. Replace the old messages with a single summary message.
4. Keep the recent messages intact.
"""
if len(messages) <= keep_last_n + 4:
return messages # Not enough to summarize
old_messages = messages[:-keep_last_n]
recent_messages = messages[-keep_last_n:]
# Build a text representation of the old conversation
old_text = "\n".join(
f"[{m['role']}]: {m['content'][:500]}"
for m in old_messages
if m.get("content")
)
summary_prompt = f"""Summarize this conversation segment. Capture:
- Key facts the user shared (name, preferences, project details)
- Decisions that were made
- Actions the agent took and their results
- Anything the agent learned that it should remember
Keep the summary under 500 words. Write in third person past tense.
Conversation:
{old_text}
"""
# Use a cheap model for summarization
from anthropic import Anthropic
client = Anthropic()
response = client.messages.create(
model=model,
max_tokens=600,
temperature=0.0,
messages=[{"role": "user", "content": summary_prompt}],
)
summary = response.content[0].text
# Build the new message array: system (if present) + summary + recent
result = []
if messages[0]["role"] == "system":
result.append(messages[0])
result.append({
"role": "user",
"content": f"[CONVERSATION SUMMARY]\n{summary}\n[/CONVERSATION SUMMARY]"
})
result.extend(recent_messages)
return result
Summarization is lossy compression. The agent remembers the user is building a Docker CLI, but might forget they specifically wanted Docker Compose v2, not v1. The tradeoff is acceptable for most use cases.
Strategy 3: Selective Retention
Not all messages are equally important. Mark critical messages as permanent. Everything else is eligible for eviction.
def selective_retention(
messages: list[dict],
max_tokens: int = 100_000,
importance_threshold: float = 0.7,
) -> list[dict]:
"""
Retain messages based on importance scoring.
Critical messages (system prompt, user facts, decisions) are kept.
Routine messages (acknowledgments, intermediate tool results) are
candidates for eviction when the token budget is exceeded.
"""
# Always preserve system messages
system_msgs = [m for m in messages if m["role"] == "system"]
rest = [m for m in messages if m["role"] != "system"]
# Score each message for importance
scored = []
for msg in rest:
score = _importance_score(msg)
scored.append((score, msg))
# Sort by importance (highest first), then by original position
# within each importance tier to maintain conversation order
scored.sort(key=lambda x: (-x[0], rest.index(x[1])))
# Keep messages until we hit the token budget
kept = system_msgs.copy()
token_count = sum(_estimate_tokens(m) for m in kept)
for score, msg in scored:
msg_tokens = _estimate_tokens(msg)
if token_count + msg_tokens <= max_tokens:
kept.append(msg)
token_count += msg_tokens
elif score >= importance_threshold:
# Critical message that exceeds budget -- we have a problem
# Keep it anyway and log a warning
kept.append(msg)
token_count += msg_tokens
# Restore original order
kept_system = [m for m in kept if m["role"] == "system"]
kept_rest = [m for m in kept if m["role"] != "system"]
kept_rest.sort(key=lambda m: rest.index(m))
return kept_system + kept_rest
def _importance_score(msg: dict) -> float:
"""Score a message's importance. Higher = more important to retain."""
content = str(msg.get("content", ""))
# Heuristics for importance
score = 0.5 # baseline
# User messages are more important than tool results
if msg["role"] == "user":
score += 0.2
# Longer messages tend to contain more information
if len(content) > 200:
score += 0.1
# Messages containing decisions, facts, or preferences
indicators = [
"prefer", "want", "need", "decided", "agreed",
"my name is", "i am", "project is", "goal is",
"error", "failed", "important", "critical",
]
content_lower = content.lower()
matches = sum(1 for ind in indicators if ind in content_lower)
score += min(matches * 0.05, 0.2)
return min(score, 1.0)
def _estimate_tokens(msg: dict) -> int:
"""Rough token estimate: ~4 chars per token."""
return len(str(msg.get("content", ""))) // 4
For production systems, replace the heuristic _importance_score with an LLM call: "Rate this message's importance to the ongoing conversation on a scale of 0 to 1." A cheap model like Haiku can score hundreds of messages for fractions of a cent.
The Tradeoff Matrix
| Strategy | Simplicity | Information Preservation | Cost | Best For |
|---|---|---|---|---|
| Sliding window | Trivial | Low (loses old context) | Free | Short tasks, stateless agents |
| Summarization | Moderate | Medium (keeps gist) | Cheap (one LLM call) | Long conversations, support agents |
| Selective retention | Complex | High (keeps what matters) | Moderate (scoring cost) | Research agents, coding agents |
Start with sliding window. It works for 80% of use cases. Add summarization when conversations get long. Add selective retention when you have specific information that must survive across turns.
Section 3: Working Memory -- The Agent's Scratchpad
Short-term memory is the conversation. Working memory is the agent's internal monologue -- structured state it maintains to track what it is doing, what it has learned, and what it needs to do next.
Why Working Memory Matters
Consider an agent researching a topic. It searches the web, gets 15 results, reads 5 pages, extracts key facts from each. Without working memory, all of that information is scattered across the conversation history -- tool results interleaved with user messages and assistant responses. The agent has to re-read the entire history to figure out what it knows.
With working memory, the agent maintains a structured scratchpad:
SCRATCHPAD:
Task: Research electric cargo bike market
Plan:
[x] Search for market reports
[x] Read top 3 reports
[ ] Extract pricing data
[ ] Compare top 5 models
[ ] Write summary
Findings:
- Market size: $2.1B in 2025, growing 12% CAGR
- Top players: Rad Power, Tern, Yuba, Riese & Muller
- Average price range: $1,800 - $6,000
Open questions:
- What are the regulatory differences between US and EU?
- Are there battery fire safety standards?
Next step: Search for "electric cargo bike pricing comparison 2025"
The scratchpad is not for the user. It is for the agent. It is the agent's working memory -- the structured representation of what it knows and what it needs to find out.
Implementing the Scratchpad
The scratchpad is a data structure that gets injected into the context on every turn. The agent updates it before responding.
import json
from dataclasses import dataclass, field
from typing import Any
@dataclass
class Scratchpad:
"""The agent's working memory for the current task."""
task: str = ""
plan: list[dict] = field(default_factory=list) # [{"step": "...", "status": "pending|done|failed"}]
findings: list[str] = field(default_factory=list)
open_questions: list[str] = field(default_factory=list)
next_step: str = ""
def to_prompt(self) -> str:
"""Render the scratchpad as a section for the system prompt."""
plan_lines = []
for item in self.plan:
status_icon = {"pending": "[ ]", "done": "[x]", "failed": "[!]"}.get(
item.get("status", "pending"), "[ ]"
)
plan_lines.append(f" {status_icon} {item['step']}")
return f"""SCRATCHPAD:
Task: {self.task}
Plan:
{chr(10).join(plan_lines) if plan_lines else ' (no plan yet)'}
Findings:
{chr(10).join(f' - {f}' for f in self.findings) if self.findings else ' (no findings yet)'}
Open questions:
{chr(10).join(f' - {q}' for q in self.open_questions) if self.open_questions else ' (none)'}
Next step: {self.next_step or '(decide based on above)'}"""
@classmethod
def from_text(cls, text: str) -> "Scratchpad":
"""Parse a scratchpad from the agent's text output."""
sp = cls()
# Simple line-based parser for the scratchpad format
current_section = None
for line in text.split("\n"):
line = line.strip()
if line.startswith("Task:"):
sp.task = line.replace("Task:", "").strip()
elif line.startswith("Next step:"):
sp.next_step = line.replace("Next step:", "").strip()
elif line.startswith("[x]") or line.startswith("[ ]") or line.startswith("[!]"):
status = {"[x]": "done", "[ ]": "pending", "[!]": "failed"}[line[:3]]
sp.plan.append({"step": line[4:].strip(), "status": status})
elif line.startswith("- ") and current_section == "findings":
sp.findings.append(line[2:].strip())
elif line.startswith("- ") and current_section == "questions":
sp.open_questions.append(line[2:].strip())
elif line == "Findings:":
current_section = "findings"
elif line == "Open questions:":
current_section = "questions"
return sp
SCRATCHPAD_INSTRUCTION = """
Before you respond, update your scratchpad. The scratchpad is your working memory.
It tracks what you know, what you need to find out, and what your next step is.
Update the scratchpad by including a SCRATCHPAD block at the start of your response:
SCRATCHPAD:
Task: [one-line description of the current task]
Plan:
[x] Step already completed
[ ] Step not yet started
[!] Step that failed
Findings:
- Fact you have confirmed
- Another confirmed fact
Open questions:
- Question you still need to answer
Next step: [what you will do next]
Then provide your actual response after the scratchpad.
"""
def build_system_prompt_with_scratchpad(
base_system_prompt: str,
scratchpad: Scratchpad,
) -> str:
"""Combine the base system prompt with scratchpad instructions and current state."""
return f"""{base_system_prompt}
{SCRATCHPAD_INSTRUCTION}
CURRENT STATE:
{scratchpad.to_prompt()}
"""
The Scratchpad in Action
Here is a complete agent that uses working memory to research a topic:
from anthropic import Anthropic
client = Anthropic()
BASE_SYSTEM_PROMPT = """You are a research agent. You have access to a web_search tool
and a fetch_page tool. Use them to answer user questions thoroughly.
When you have enough information to answer the user's question, provide a complete
response with citations. Do not end the task until you are confident in your answer."""
def research_agent_with_scratchpad(user_task: str, max_turns: int = 10) -> str:
"""Research agent that maintains a scratchpad across turns."""
scratchpad = Scratchpad(task=user_task)
messages = []
for turn in range(max_turns):
system_prompt = build_system_prompt_with_scratchpad(
BASE_SYSTEM_PROMPT, scratchpad
)
# Build the user message for this turn
if turn == 0:
user_msg = f"Task: {user_task}\n\nStart by planning your approach, then begin researching."
else:
user_msg = "Continue working. Update your scratchpad and take the next step."
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2048,
temperature=0.0,
system=system_prompt,
messages=messages + [{"role": "user", "content": user_msg}],
)
full_response = response.content[0].text
# Parse scratchpad from response
if "SCRATCHPAD:" in full_response:
parts = full_response.split("SCRATCHPAD:", 1)
after_scratchpad = parts[1] if len(parts) > 1 else ""
# Find where the scratchpad ends (next double newline or end of text)
sp_text = after_scratchpad.strip()
scratchpad = Scratchpad.from_text(sp_text)
# Check if the task is complete
if scratchpad.plan and all(
item["status"] == "done" for item in scratchpad.plan
):
return full_response
# Append to conversation
messages.append({"role": "user", "content": user_msg})
messages.append({"role": "assistant", "content": full_response})
return "Agent reached maximum turns without completing the task."
The scratchpad pattern forces the agent to externalize its thinking. Instead of reasoning silently and producing an answer, it writes down what it knows, what it needs, and what it plans to do next. This has three benefits:
-
The agent stays on track. The scratchpad is a constant reminder of the task, the plan, and the open questions. The agent cannot forget what it is doing because the scratchpad is in its context on every turn.
-
You can debug the agent. When the agent goes off the rails, you can read the scratchpad and see exactly what it was thinking at each step. This is invaluable for debugging.
-
The agent can recover from interruptions. If the agent loop crashes mid-task, you can reload the scratchpad and resume. The agent picks up where it left off.
The scratchpad is the difference between an agent that reacts and an agent that thinks. It is the simplest form of planning, and it costs almost nothing to implement.
Section 4: Long-Term Memory -- Vector Stores
Short-term memory and working memory are volatile. They disappear when the session ends. Long-term memory persists. It is what lets your agent remember user preferences across conversations, retrieve relevant documents, and learn from past experiences.
The core technology for long-term memory is embeddings + vector search.
What Embeddings Are
An embedding is a numerical representation of text that captures semantic meaning. It is a vector -- a list of floating-point numbers, typically 768 to 3,072 of them -- that positions the text in a high-dimensional space. Texts with similar meanings have similar vectors. Texts with different meanings are far apart.
"The cat sat on the mat" -> [0.023, -0.451, 0.782, ..., 0.134] (1536 numbers)
"The dog lay on the rug" -> [0.019, -0.447, 0.779, ..., 0.129] (very close)
"Quantum mechanics is hard" -> [-0.612, 0.231, -0.104, ..., 0.891] (very far)
The embedding model -- a neural network trained specifically for this task -- produces these vectors. You do not need to understand how it works internally. You need to understand what it enables: search by meaning, not by keyword.
Keyword search looks for exact word matches. "car" matches "car" but not "automobile." Semantic search via embeddings matches "car" with "automobile," "vehicle," "sedan," and "transportation" -- because they are close in vector space.
How Vector Search Works
The pipeline has four steps:
1. CHUNK documents into manageable pieces (500-1000 tokens each)
2. EMBED each chunk into a vector using an embedding model
3. STORE vectors + original text in a vector database
4. QUERY: embed the query, find nearest neighbors, return matching texts
When the user asks a question, you embed the question, search the vector database for the most similar document chunks, and inject those chunks into the agent's context. The agent now has relevant information it did not have before.
ChromaDB: Vector Search Without the Infrastructure
ChromaDB is an open-source vector database that runs in-process. No server. No Docker. No configuration. You pip install chromadb and you have a working vector store. It is the right choice for learning and for single-machine deployments.
import chromadb
from chromadb.config import Settings
# Create an in-memory client (use PersistentClient for disk storage)
client = chromadb.Client(Settings(anonymized_telemetry=False))
# Create a collection -- think of it as a table
collection = client.create_collection(
name="agent_memory",
metadata={"description": "Long-term memory for the research agent"},
)
Embedding Documents
You need an embedding model to convert text to vectors. OpenAI and Anthropic both provide embedding APIs. Here is a function that uses OpenAI's embedding model (widely available, cheap, and reliable):
from openai import OpenAI
embed_client = OpenAI()
def embed_texts(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
"""Convert a list of texts to embedding vectors."""
# text-embedding-3-small produces 1536-dimensional vectors
# Cost: $0.02 per 1M tokens -- effectively free for most use cases
response = embed_client.embeddings.create(
model=model,
input=texts,
)
return [item.embedding for item in response.data]
def embed_query(query: str) -> list[float]:
"""Embed a single query string."""
return embed_texts([query])[0]
Chunking Documents
You cannot embed an entire book as one vector. The embedding loses too much detail. You need to split documents into chunks -- pieces small enough that each chunk represents a coherent unit of meaning, but large enough to contain useful context.
def chunk_text(
text: str,
chunk_size: int = 500,
chunk_overlap: int = 100,
) -> list[dict]:
"""
Split text into overlapping chunks.
Overlap prevents information from being split across chunk boundaries.
A sentence that starts at the end of chunk N will also appear at the
start of chunk N+1, so it is findable regardless of where the query
embedding lands.
"""
words = text.split()
chunks = []
for i in range(0, len(words), chunk_size - chunk_overlap):
chunk_words = words[i:i + chunk_size]
if not chunk_words:
break
chunk_text = " ".join(chunk_words)
chunks.append({
"text": chunk_text,
"index": len(chunks),
"start_word": i,
"end_word": i + len(chunk_words),
})
return chunks
The Complete Pipeline
Now put it all together: chunk, embed, store, query.
import uuid
def add_to_memory(
collection,
documents: list[str],
metadatas: list[dict] | None = None,
ids: list[str] | None = None,
) -> list[str]:
"""
Chunk documents, embed them, and store in the vector database.
Returns the list of chunk IDs for later retrieval or deletion.
"""
all_chunks = []
all_metadatas = []
all_ids = []
for doc_idx, doc in enumerate(documents):
chunks = chunk_text(doc)
for chunk in chunks:
all_chunks.append(chunk["text"])
meta = {
"document_index": doc_idx,
"chunk_index": chunk["index"],
**(metadatas[doc_idx] if metadatas else {}),
}
all_metadatas.append(meta)
all_ids.append(str(uuid.uuid4()))
# Embed all chunks
embeddings = embed_texts(all_chunks)
# Store in ChromaDB
collection.add(
ids=all_ids,
embeddings=embeddings,
documents=all_chunks,
metadatas=all_metadatas,
)
return all_ids
def query_memory(
collection,
query: str,
n_results: int = 5,
metadata_filter: dict | None = None,
) -> list[dict]:
"""
Search the vector database for chunks relevant to the query.
Returns a list of dicts with 'text', 'metadata', and 'distance'.
"""
query_embedding = embed_query(query)
where_filter = metadata_filter if metadata_filter else None
results = collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=where_filter,
include=["documents", "metadatas", "distances"],
)
# Flatten ChromaDB's nested result structure
output = []
if results["ids"] and results["ids"][0]:
for i in range(len(results["ids"][0])):
output.append({
"id": results["ids"][0][i],
"text": results["documents"][0][i],
"metadata": results["metadatas"][0][i] if results["metadatas"] else {},
"distance": results["distances"][0][i],
})
return output
Semantic Search vs. Keyword Search
Vector search finds semantically similar content. It excels at "find me information about X" when X might be phrased differently in the documents. But it has blind spots:
- Exact matches. "Error code EADDRINUSE" -- vector search might return documents about port conflicts generally, but keyword search finds the exact error code.
- Names and IDs. "User amitk123" -- vector search does not know this is a specific identifier. Keyword search does.
- Dates and numbers. "Revenue in Q3 2025" -- vector search might return Q2 or Q4 results. Keyword search matches the exact quarter.
Production memory systems use hybrid search: vector search for semantic relevance, keyword search for exact matches, and a fusion algorithm that combines the results. ChromaDB does not natively support hybrid search, but you can implement it by running both searches and merging results:
def hybrid_search(
collection,
query: str,
n_results: int = 5,
) -> list[dict]:
"""
Combine vector search and keyword search results.
Simple fusion: interleave results, deduplicate, return top N.
"""
# Vector search
vector_results = query_memory(collection, query, n_results=n_results * 2)
# Keyword search (ChromaDB supports full-text search via 'where_document')
try:
keyword_results = collection.query(
query_texts=[query],
n_results=n_results * 2,
include=["documents", "metadatas", "distances"],
)
kw_output = []
if keyword_results["ids"] and keyword_results["ids"][0]:
for i in range(len(keyword_results["ids"][0])):
kw_output.append({
"id": keyword_results["ids"][0][i],
"text": keyword_results["documents"][0][i],
"metadata": keyword_results["metadatas"][0][i] if keyword_results["metadatas"] else {},
"distance": keyword_results["distances"][0][i],
})
except Exception:
kw_output = []
# Merge: interleave, deduplicate by ID, take top N
seen_ids = set()
merged = []
for pair in zip(vector_results, kw_output + [None] * len(vector_results)):
for result in pair:
if result is None:
continue
if result["id"] not in seen_ids:
seen_ids.add(result["id"])
merged.append(result)
return merged[:n_results]
Section 5: Long-Term Memory -- Beyond Vectors
Vector search is powerful, but it is not the right tool for every kind of memory. Some information is structured. Some is relational. Some needs exact lookup, not fuzzy similarity. A complete memory architecture uses multiple storage backends, each optimized for a different kind of recall.
Key-Value Memory: Facts and Preferences
Some things are just facts. "The user's name is Amit." "The user prefers Celsius." "The default output format is markdown." These do not need semantic search. They need exact lookup by key.
import json
import sqlite3
from datetime import datetime, timezone
class KeyValueMemory:
"""
Simple key-value store backed by SQLite.
For user preferences, facts, and settings.
Why SQLite and not a dict? Because SQLite survives restarts.
"""
def __init__(self, db_path: str = "memory/kv.db"):
import os
os.makedirs(os.path.dirname(db_path), exist_ok=True)
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self.conn.execute("""
CREATE TABLE IF NOT EXISTS kv_store (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
self.conn.commit()
def set(self, key: str, value: any) -> None:
"""Store a value by key. Overwrites if key exists."""
self.conn.execute(
"INSERT OR REPLACE INTO kv_store (key, value, updated_at) VALUES (?, ?, ?)",
(key, json.dumps(value), datetime.now(timezone.utc).isoformat()),
)
self.conn.commit()
def get(self, key: str, default: any = None) -> any:
"""Retrieve a value by key. Returns default if not found."""
row = self.conn.execute(
"SELECT value FROM kv_store WHERE key = ?", (key,)
).fetchone()
if row is None:
return default
return json.loads(row["value"])
def delete(self, key: str) -> None:
"""Remove a key and its value."""
self.conn.execute("DELETE FROM kv_store WHERE key = ?", (key,))
self.conn.commit()
def get_all(self, prefix: str = "") -> dict:
"""Get all keys (optionally filtered by prefix) and their values."""
if prefix:
rows = self.conn.execute(
"SELECT key, value FROM kv_store WHERE key LIKE ?",
(f"{prefix}%",),
).fetchall()
else:
rows = self.conn.execute(
"SELECT key, value FROM kv_store"
).fetchall()
return {row["key"]: json.loads(row["value"]) for row in rows}
# Usage
kv = KeyValueMemory()
kv.set("user:name", "Amit")
kv.set("user:preferences:temperature_unit", "celsius")
kv.set("user:preferences:output_format", "markdown")
name = kv.get("user:name") # "Amit"
all_prefs = kv.get_all("user:preferences:")
# {"user:preferences:temperature_unit": "celsius", "user:preferences:output_format": "markdown"}
Key-value memory is the simplest form of long-term memory. It is fast, predictable, and trivial to implement. Use it for anything that has a clear key and does not need semantic search.
Relational Memory: Structured Data
Some information is relational. "Show me all conversations about Project X." "What tasks were created in the last week?" "Which documents reference the Q3 budget?" This is what SQL databases are for.
class RelationalMemory:
"""
Structured memory for entities and their relationships.
Backed by SQLite for zero-config persistence.
"""
def __init__(self, db_path: str = "memory/relational.db"):
import os
os.makedirs(os.path.dirname(db_path), exist_ok=True)
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
self._init_schema()
self.conn.commit()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS conversations (
id TEXT PRIMARY KEY,
title TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
summary TEXT,
tags TEXT -- JSON array of tags
);
CREATE TABLE IF NOT EXISTS entities (
id TEXT PRIMARY KEY,
type TEXT NOT NULL, -- 'project', 'person', 'document', 'task'
name TEXT NOT NULL,
properties TEXT, -- JSON object for flexible attributes
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS references_rel (
id TEXT PRIMARY KEY,
source_type TEXT NOT NULL,
source_id TEXT NOT NULL,
target_type TEXT NOT NULL,
target_id TEXT NOT NULL,
relationship TEXT NOT NULL, -- 'mentions', 'belongs_to', 'depends_on'
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_entities_type ON entities(type);
CREATE INDEX IF NOT EXISTS idx_refs_source ON references_rel(source_type, source_id);
CREATE INDEX IF NOT EXISTS idx_refs_target ON references_rel(target_type, target_id);
""")
def add_conversation(self, conv_id: str, title: str, tags: list[str] | None = None):
now = datetime.now(timezone.utc).isoformat()
self.conn.execute(
"INSERT OR REPLACE INTO conversations (id, title, created_at, updated_at, tags) VALUES (?, ?, ?, ?, ?)",
(conv_id, title, now, now, json.dumps(tags or [])),
)
self.conn.commit()
def add_entity(self, entity_id: str, entity_type: str, name: str, properties: dict | None = None):
now = datetime.now(timezone.utc).isoformat()
self.conn.execute(
"INSERT OR REPLACE INTO entities (id, type, name, properties, created_at) VALUES (?, ?, ?, ?, ?)",
(entity_id, entity_type, name, json.dumps(properties or {}), now),
)
self.conn.commit()
def link(self, source_type: str, source_id: str, target_type: str, target_id: str, relationship: str):
now = datetime.now(timezone.utc).isoformat()
link_id = f"{source_type}:{source_id}->{target_type}:{target_id}:{relationship}"
self.conn.execute(
"INSERT OR REPLACE INTO references_rel (id, source_type, source_id, target_type, target_id, relationship, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
(link_id, source_type, source_id, target_type, target_id, relationship, now),
)
self.conn.commit()
def find_related(self, entity_type: str, entity_id: str, relationship: str | None = None) -> list[dict]:
"""Find all entities related to the given entity."""
query = """
SELECT e.*, r.relationship
FROM references_rel r
JOIN entities e ON (
(r.target_type = e.type AND r.target_id = e.id)
OR (r.source_type = e.type AND r.source_id = e.id)
)
WHERE (r.source_type = ? AND r.source_id = ?)
OR (r.target_type = ? AND r.target_id = ?)
"""
params = [entity_type, entity_id, entity_type, entity_id]
if relationship:
query += " AND r.relationship = ?"
params.append(relationship)
rows = self.conn.execute(query, params).fetchall()
return [dict(row) for row in rows]
Episodic Memory: Past Conversations
Episodic memory stores past conversations and their outcomes. "Last time we tried approach A, it failed because the API rate limit was 100 requests per minute." This is the most valuable kind of long-term memory -- and the hardest to get right.
The implementation is a vector store where each "episode" is a conversation summary with metadata about the outcome:
def store_episode(
collection,
kv_memory: KeyValueMemory,
conversation_id: str,
summary: str,
outcome: str, # 'success', 'failure', 'incomplete'
tags: list[str],
key_learnings: list[str],
) -> None:
"""Store a conversation episode in long-term memory."""
# Store the full summary in the vector DB for semantic retrieval
episode_text = f"""CONVERSATION {conversation_id}
Outcome: {outcome}
Tags: {', '.join(tags)}
Summary:
{summary}
Key Learnings:
{chr(10).join(f'- {l}' for l in key_learnings)}
"""
add_to_memory(
collection,
documents=[episode_text],
metadatas=[{
"type": "episode",
"conversation_id": conversation_id,
"outcome": outcome,
"tags": json.dumps(tags),
}],
)
# Also store structured metadata for exact lookup
kv.set(f"episode:{conversation_id}:outcome", outcome)
kv.set(f"episode:{conversation_id}:tags", tags)
kv.set(f"episode:{conversation_id}:learnings", key_learnings)
The Memory Router
With multiple memory backends, the agent needs to decide which one to query. The memory router is a lightweight decision layer:
def route_memory_query(
query: str,
vector_collection,
kv_memory: KeyValueMemory,
relational_memory: RelationalMemory,
) -> dict:
"""
Decide which memory backend(s) to query based on the query type.
Returns a dict with results from the appropriate backend(s).
"""
query_lower = query.lower()
results = {"vector": [], "kv": {}, "relational": []}
# Key-value: exact lookups for facts and preferences
kv_indicators = [
"what is my", "what are my", "my name", "my preference",
"my settings", "what do i prefer", "remember my",
]
if any(ind in query_lower for ind in kv_indicators):
# Extract potential key patterns from the query
if "name" in query_lower:
results["kv"]["name"] = kv_memory.get("user:name")
if "prefer" in query_lower:
results["kv"]["preferences"] = kv_memory.get_all("user:preferences:")
# Relational: structured queries about entities and relationships
relational_indicators = [
"project", "task", "conversation about", "related to",
"all my", "list of", "show me",
]
if any(ind in query_lower for ind in relational_indicators):
# This is a simplified router -- a production version would use
# an LLM to parse the query into a structured database query
pass
# Vector: semantic search for everything else
results["vector"] = query_memory(vector_collection, query, n_results=5)
return results
For a production system, replace the keyword-based router with an LLM call: "Given this query, which memory backends should be searched? Options: key-value (facts/preferences), relational (structured data), vector (semantic search). Return a JSON array."
Section 6: Building the Complete Memory Stack
Now integrate all three tiers into the agent loop. The updated flow:
1. Receive user message
2. Query long-term memory for relevant context
3. Load short-term memory (recent conversation)
4. Initialize working memory (scratchpad)
5. Run agent loop with all memory available
6. After completion, save important facts to long-term memory
Here is the complete agent:
import uuid
from datetime import datetime, timezone
from anthropic import Anthropic
class MemoryAgent:
"""
An agent with short-term, working, and long-term memory.
Short-term: conversation history in the messages array.
Working: scratchpad for the current task.
Long-term: vector store (ChromaDB) + key-value store (SQLite).
"""
def __init__(
self,
system_prompt: str,
model: str = "claude-sonnet-4-20250514",
vector_collection=None,
kv_memory: KeyValueMemory | None = None,
):
self.system_prompt = system_prompt
self.model = model
self.client = Anthropic()
self.vector_collection = vector_collection
self.kv_memory = kv_memory or KeyValueMemory()
# Short-term memory
self.messages: list[dict] = []
# Working memory
self.scratchpad = Scratchpad()
# Session identity
self.session_id = str(uuid.uuid4())
def _load_long_term_context(self, user_message: str) -> str:
"""Query long-term memory for context relevant to the current message."""
context_parts = []
# 1. Load user facts from key-value store
user_name = self.kv_memory.get("user:name")
if user_name:
prefs = self.kv_memory.get_all("user:preferences:")
context_parts.append(f"User: {user_name}")
if prefs:
context_parts.append(
"Preferences: " + ", ".join(
f"{k.split(':')[-1]}={v}" for k, v in prefs.items()
)
)
# 2. Search vector store for relevant past episodes and documents
if self.vector_collection:
results = query_memory(
self.vector_collection,
user_message,
n_results=3,
)
if results:
context_parts.append("Relevant past context:")
for r in results:
# Truncate to avoid context pollution
snippet = r["text"][:300]
context_parts.append(f" - {snippet}")
if not context_parts:
return ""
return "LONG-TERM MEMORY CONTEXT:\n" + "\n".join(context_parts) + "\n\n"
def _build_system_prompt(self, long_term_context: str) -> str:
"""Assemble the full system prompt with all memory tiers."""
parts = [self.system_prompt]
if long_term_context:
parts.append(long_term_context)
parts.append(SCRATCHPAD_INSTRUCTION)
parts.append("CURRENT STATE:")
parts.append(self.scratchpad.to_prompt())
return "\n\n".join(parts)
def run(self, user_message: str, max_turns: int = 10) -> str:
"""Run the agent loop with full memory integration."""
# Step 1: Load long-term context
long_term_context = self._load_long_term_context(user_message)
# Step 2: Add user message to short-term memory
self.messages.append({"role": "user", "content": user_message})
# Step 3-5: Agent loop
for turn in range(max_turns):
system_prompt = self._build_system_prompt(long_term_context)
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
temperature=0.0,
system=system_prompt,
messages=self.messages,
)
full_response = response.content[0].text
# Parse scratchpad from response
if "SCRATCHPAD:" in full_response:
parts = full_response.split("SCRATCHPAD:", 1)
sp_text = parts[1].strip() if len(parts) > 1 else ""
self.scratchpad = Scratchpad.from_text(sp_text)
# Add response to short-term memory
self.messages.append({"role": "assistant", "content": full_response})
# Check if task is complete
if response.stop_reason == "end_turn":
# No tool calls -- agent is done
break
# Step 6: Save important facts to long-term memory
self._consolidate_memory(user_message, full_response)
return full_response
def _consolidate_memory(self, user_message: str, agent_response: str) -> None:
"""
After the conversation turn, extract and store important facts.
Uses a cheap model to identify what's worth remembering.
"""
consolidation_prompt = f"""Analyze this conversation turn and extract facts worth remembering.
User message: {user_message[:500]}
Agent response: {agent_response[:500]}
Return a JSON object with these fields:
- user_facts: object with key-value pairs to store (e.g., {{"user:name": "Amit"}})
- episode_summary: string summarizing this interaction (or null if nothing notable)
- episode_tags: array of strings for categorization
- episode_outcome: "success", "failure", or "incomplete"
- key_learnings: array of strings (things the agent learned)
Only include information that would be useful in future conversations.
Return ONLY valid JSON, no other text."""
try:
response = self.client.messages.create(
model="claude-haiku-4-20250514",
max_tokens=500,
temperature=0.0,
messages=[{"role": "user", "content": consolidation_prompt}],
)
import json
extracted = json.loads(response.content[0].text)
# Store user facts
for key, value in extracted.get("user_facts", {}).items():
self.kv_memory.set(key, value)
# Store episode in vector DB
if self.vector_collection and extracted.get("episode_summary"):
store_episode(
self.vector_collection,
self.kv_memory,
self.session_id,
extracted["episode_summary"],
extracted.get("episode_outcome", "success"),
extracted.get("episode_tags", []),
extracted.get("key_learnings", []),
)
except Exception:
# Memory consolidation is best-effort. Never block the agent on it.
pass
def reset_session(self):
"""Start a new session, clearing short-term and working memory."""
self.messages = []
self.scratchpad = Scratchpad()
self.session_id = str(uuid.uuid4())
Multi-Session Conversation Trace
Here is what the agent looks like across two sessions:
=== SESSION 1 ===
User: Hi, I'm Amit. I'm building a data pipeline that processes
weather data. I prefer all code examples in Python with type hints.
Agent: [Scratchpad records: user=Amit, project=weather data pipeline,
preference=Python with type hints]
Nice to meet you, Amit! I'll make sure to use Python with type
hints. What's the first thing you need help with?
User: What's the best way to handle missing temperature values in a
time series?
Agent: [Scratchpad records: task=missing value imputation]
[Searches knowledge, provides detailed answer with code]
Here are three approaches with tradeoffs...
--- SESSION ENDS (browser closed, server restarted) ---
=== SESSION 2 ===
User: Hey, can you remind me what we were working on?
Agent: [Queries long-term memory]
[KV store returns: user:name=Amit,
user:preferences:output_format=python_with_type_hints]
[Vector store returns: episode about weather data pipeline,
episode about missing value imputation]
Welcome back, Amit! In our last session, you were building a
weather data pipeline and we discussed strategies for handling
missing temperature values in time series data. I covered three
approaches: forward fill, interpolation, and model-based imputation.
Would you like to continue with that, or is there something new
you'd like to work on?
The agent remembered. Not because the model is stateful -- it is not. Because you built a memory architecture that persists across sessions. The key-value store held the user's name and preferences. The vector store held the conversation summary. The agent queried both before responding.
This is the transformation. The agent goes from "talking to a search engine" to "working with a collaborator who knows your context."
Section 7: Memory Pitfalls
Memory is powerful. It is also dangerous. Here are the five ways memory systems fail, and how to prevent each one.
1. Context Pollution
The vector store returns 10 chunks. You inject all 10 into the context. The model is now drowning in information, much of it tangentially relevant at best. It confuses details from different chunks. It hallucinates connections between unrelated documents. Its responses get worse, not better.
The fix: Be selective. Return fewer chunks (3-5 is usually enough). Rank by relevance and drop anything below a similarity threshold. Use metadata filtering to narrow the search space before running the vector query. More context is not always better -- it is often worse.
def query_with_relevance_filter(
collection,
query: str,
n_results: int = 5,
min_similarity: float = 0.7,
) -> list[dict]:
"""Query but drop results below a similarity threshold."""
results = query_memory(collection, query, n_results=n_results)
# ChromaDB uses distance; convert to similarity (1 - distance for cosine)
return [r for r in results if (1 - r["distance"]) >= min_similarity]
2. Stale Memory
Facts change. "The CEO is X" -- but X was replaced last month. "The API rate limit is 100/min" -- but it was increased to 500/min last week. Memory that does not expire becomes misinformation.
The fix: Every piece of stored information needs a timestamp. When retrieving, prefer recent information. When facts conflict, prefer the newer one. For critical facts, set explicit expiration:
def set_with_expiry(kv: KeyValueMemory, key: str, value: any, ttl_seconds: int) -> None:
"""Store a value that expires after ttl_seconds."""
expiry = datetime.now(timezone.utc).timestamp() + ttl_seconds
kv.set(key, {"value": value, "expires_at": expiry})
def get_with_expiry(kv: KeyValueMemory, key: str, default: any = None) -> any:
"""Retrieve a value, returning default if expired."""
stored = kv.get(key)
if stored is None:
return default
if isinstance(stored, dict) and "expires_at" in stored:
if datetime.now(timezone.utc).timestamp() > stored["expires_at"]:
kv.delete(key)
return default
return stored["value"]
return stored
3. Memory Conflicts
The vector store returns two chunks. One says "the deployment process uses Kubernetes." The other says "the deployment process uses Docker Compose." Both are from legitimate sources. They contradict each other. What does the agent do?
The fix: Surface the contradiction to the agent explicitly. "Note: retrieved memories contain conflicting information about X. Source A says Y. Source B says Z. Use the most recent source, or ask the user to clarify."
def detect_conflicts(results: list[dict]) -> list[str]:
"""
Simple conflict detection: if two results have the same metadata key
but different values, flag it.
"""
conflicts = []
for i, r1 in enumerate(results):
for r2 in results[i+1:]:
# Check for overlapping metadata with different values
common_keys = set(r1["metadata"].keys()) & set(r2["metadata"].keys())
for key in common_keys:
if key in ("type", "chunk_index", "document_index"):
continue
if r1["metadata"][key] != r2["metadata"][key]:
conflicts.append(
f"Conflict on '{key}': '{r1['metadata'][key]}' vs '{r2['metadata'][key]}'"
)
return conflicts
4. Privacy
Memory persists. What if the user wants to be forgotten? What if sensitive information was stored and should not have been? Memory systems need deletion from day one, not bolted on after a privacy incident.
The fix: Every piece of stored data needs a deletion path. Tag all stored information with a user ID. Provide a delete_all_user_data(user_id) function that wipes everything. Test it.
def delete_user_data(
user_id: str,
vector_collection,
kv_memory: KeyValueMemory,
) -> dict:
"""Delete all stored data for a user. Returns counts of what was deleted."""
result = {"vector_chunks": 0, "kv_entries": 0}
# Delete from vector store
try:
# Get all chunks for this user
existing = vector_collection.get(
where={"user_id": user_id},
include=["metadatas"],
)
if existing["ids"]:
vector_collection.delete(ids=existing["ids"])
result["vector_chunks"] = len(existing["ids"])
except Exception:
pass
# Delete from key-value store
all_kv = kv_memory.get_all(f"user:{user_id}")
for key in all_kv:
kv_memory.delete(key)
result["kv_entries"] += 1
return result
5. Cost
Embedding every message, every document, every tool result gets expensive. At scale, embedding costs can exceed LLM costs. You need to be intentional about what you store.
The fix: Not everything is worth remembering. Use the consolidation step (the cheap model that extracts facts after each turn) as a filter. Only store what it flags as important. Batch embeddings where possible -- embed 100 chunks in one API call instead of 100 separate calls. Use a local embedding model (like all-MiniLM-L6-v2 via sentence-transformers) for high-volume, low-stakes embedding where API costs would be prohibitive.
The Turn
Your agent now has a memory that spans conversations. It remembers who the user is. It remembers what they discussed. It remembers what it learned. It retrieves relevant past context before responding. It maintains a structured scratchpad to track what it is doing. It consolidates important facts after each interaction.
This is not a chatbot with a longer prompt. This is a system with genuine continuity. The user can close the browser, come back a week later, and pick up where they left off. The agent knows their name, their preferences, their project, and the last thing they were working on.
The architecture you built in this chapter -- short-term conversation management, working memory scratchpads, vector search for semantic retrieval, key-value stores for facts, relational databases for structured data -- is the same architecture that powers production agent systems. The implementations are simple, but the patterns are real. You can take the MemoryAgent class from this chapter, swap in a production vector database like Pinecone or Weaviate, add a proper API, and you have a production-ready memory system.
Memory is what turns a tool into a collaborator. Without memory, the agent is a function: input in, output out, nothing carries forward. With memory, the agent is a relationship: it learns, it adapts, it gets better over time. The code you wrote in this chapter is the difference between those two things.
Close
Your agent can now remember. It carries context across conversations. It retrieves relevant knowledge from a growing library of past experiences. It maintains a scratchpad to track its own thinking.
But it still thinks one step at a time. React, respond, react. The user says something, the agent responds. The user says something else, the agent responds again. For simple tasks, this is enough. For complex problems, it is not.
Consider this request: "Research the competitive landscape for electric cargo bikes, compare the top five models on price, range, and payload capacity, and produce a summary with citations." A reactive agent dives in immediately. It searches, reads, searches again, reads more, and eventually produces something. But it never steps back to plan. It never asks: what do I need to know first? What are the sub-questions? What order should I tackle them in? It reacts its way through the task and hopes for the best.
Your agent needs to learn how to think before it acts. It needs to produce a plan, evaluate that plan, execute it step by step, and revise when reality diverges from expectations. It needs reasoning.
In the next chapter, you will teach your agent to plan.
Next: Chapter 7 -- Planning and Reasoning