8.5 Memory Systems for Agents¶

Every human expert you have ever admired carries three things into a meeting: working notes they can reference in real time, a personal knowledge base built from years of experience, and an episodic record of what happened last time they worked with you. LLM agents need the same three things — and the context window alone cannot provide all of them. Without external memory, an agent reset to a blank context after 200,000 tokens is not a persistent assistant; it is a brilliant amnesiac.

This chapter maps the full memory stack for LLM agents: from the in-context scratch pad to durable vector stores, file systems, and knowledge graphs. We cover how to architect write, read, and reflect operations; the distinction between episodic and semantic memory; memory compaction strategies that fight context-window limits; and the file-as-memory pattern that many production agents rely on. We close with a fully runnable from-scratch implementation you can drop into your own harness.

Related chapters: this one focuses on the memory layer of agents. For the broader agentic loop, see The Agentic Loop: ReAct, Plan-Execute & Reflection. For how retrieved text gets injected back into the model, see Context Engineering & Management. For the vector database machinery underneath, see Vector Databases & Approximate Nearest Neighbor Search.

8.5.1 The Two Memory Horizons¶

We can draw a clean axis: memory that lives inside the context window versus memory that lives outside it. Everything inside is available on every token prediction; everything outside must be explicitly fetched.

Short-term (in-context) memory is the running conversation: system prompt, prior turns, tool call/result pairs, and any snippets the agent has fetched. Its capacity is bounded by the context length and its cost by the token budget — and note that the whole window is re-encoded as input on every turn. At on the order of $1 per million input tokens, one completely filled 128 k window costs roughly $0.13 per call, so a 200-turn session that keeps the window near-full pays that price 200 times. Prefix caching removes most of the repeat cost for the unchanged prefix (an order of magnitude cheaper per cached token on several 2026 APIs — see Prefix Caching & KV-Cache Reuse), but it does not stop the window from filling up, which is the failure external memory exists to prevent. See Context Engineering & Management for how to manage what goes into this budget.

Long-term (external) memory is any storage system the agent can query and update. It is not bounded by the context window, persists across sessions, and can be shared across many agent instances. Its cost is dominated by embedding latency (for vector retrieval) and I/O rather than per-token model pricing.

The fundamental design challenge: you must decide what to commit to external memory, when, in what form, and how to retrieve it later accurately. Getting any one of these wrong produces either a useless store (nothing relevant comes back on retrieval) or a poisoned store (stale or contradictory facts pollute the context).

8.5.2 Episodic vs Semantic Memory¶

Psychology distinguishes two kinds of long-term memory in humans; the same distinction is useful for agents.

Episodic memory records events with their temporal context — “on 2025-11-04, the user asked me to refactor the authentication module and I found the JWT library had a bug.” Each entry is a timestamped, situated record of a specific interaction. Episodic stores are append-only by nature, grow monotonically, and are queried with temporal or situational cues (“what did we decide last Tuesday?”).

Semantic memory records general knowledge and facts — “the production database uses PostgreSQL 15; the API gateway rate-limits at 1 000 req/s; the user prefers British English.” Semantic entries are facts that should be true across sessions. They are updated (overwritten) when new information supersedes old, and queried with content-based cues (“what do I know about the database?”).

In practice, most memory systems blend both types. A common architecture uses:

The key insight: do not collapse both into a single vector store with no separation. Mixing a “2025-11-04: JWT bug” episodic entry with a “the JWT library version is 4.1.0” semantic fact produces a store where date-specific noise pollutes general-knowledge queries.

8.5.3 Memory Store Types¶

8.5.3.1 Vector Stores¶

A vector store encodes each memory as a dense embedding vector and supports approximate nearest-neighbour (ANN) retrieval. This is the most common substrate for agent memory because natural language is naturally suited to embedding-based similarity.

Given a query \(q\) and a corpus of stored memories \(\{m_1, m_2, \ldots, m_N\}\), the retrieval objective is:

where \(\mathbf{e}(\cdot)\) is an embedding model and \(\cos\) is cosine similarity. In practice we retrieve the top-\(k\) memories and re-rank them. See Embeddings & Representation Learning for embedding model choices and Chunking, Reranking & Hybrid Search for re-ranking strategies.

Strengths: fuzzy matching, handles paraphrase, scales to millions of entries with HNSW or IVF indexes.

Weaknesses: exact-match failure (“what is the API key for service X?”), no structured filtering without metadata, embedding drift if you change embedding models.

8.5.3.2 Key-Value and Structured Stores¶

For structured facts — user preferences, configuration, task state — a simple key-value store or a relational database is superior. SQLite is an excellent single-agent choice: it is serverless, ACID-compliant, and has a negligible setup cost. A Redis or DynamoDB instance suits multi-agent setups.

Pattern: store semantic facts with a canonical key (for example, db.engine, user.language) and retrieve deterministically. No embedding required; latency is sub-millisecond.

8.5.3.3 Property Graphs¶

A property graph (Neo4j, Kuzu, or even NetworkX for small deployments) represents knowledge as nodes and labelled edges: (User)--[PREFERS]-->(BritishEnglish), (Service:AuthAPI)--[RATE_LIMITED_AT]-->(1000 req/s). Graph stores shine when relationships between entities matter — “which services depend on the authentication module?” is a graph traversal, not a similarity search.

Agents that build knowledge graphs over multiple sessions can answer multi-hop queries that purely flat memory architectures cannot, at the cost of a more complex write pipeline that must extract entities and relationships before storing.

8.5.3.4 File System as Memory¶

The simplest form of persistent memory is a structured file system directory. The agent writes markdown files, JSON blobs, or code artefacts; a lightweight indexer (even a plain find + grep) serves retrieval. This is the file-as-memory pattern: the working directory is the agent’s notebook.

Several production agents (including early versions of Devin and Claude Code’s memory feature) use precisely this pattern. It requires no external database, is human-readable and auditable, and integrates with version control. Retrieval is handled either by fuzzy search over the file tree or by maintaining a separate vector index of file contents.

8.5.4 The Write / Read / Reflect Triad¶

Every memory operation falls into one of three categories.

8.5.4.1 Write¶

A memory write commits information from the current context to external storage. Writes happen:

• After a significant discovery — the agent learned something new (a fact, a bug, a user preference).
• At session boundaries — before the context is cleared, the agent saves a summary.
• On task completion — the outcome of a sub-task is logged for future reference.
• On explicit instruction — the user says “remember this.”

The write step must decide what to save, how to format it, and what metadata to attach (timestamp, session ID, confidence, topic tags). Writes that are too verbose pollute the store; writes that are too sparse lose signal.

8.5.4.2 Read¶

A memory read queries external storage and injects the result into the current context. Reads happen:

• At session start — bootstrap the context with relevant prior knowledge.
• Before a tool call — check whether the agent already knows the answer.
• When the agent signals uncertainty — “I don’t remember whether…” triggers a retrieval.
• Proactively, by the harness — the orchestrating code detects topic shifts and retrieves matching memories.

The retrieved snippets must be ranked and filtered: if you inject the top-20 vector matches unfiltered, you fill the context with noise. A conservative top-\(k\) of 3–5 with a relevance threshold works well for most agents. Ranking by cosine similarity alone is also a weak default — it is blind to when a memory was formed and how consequential it was; section 8.5.10 adds recency and importance terms and implements the combined ranker.

8.5.4.3 Reflect¶

Reflection is the highest-value memory operation and the most commonly skipped. A reflect operation asks the agent to read a batch of recent episodes and distil new semantic facts or updated beliefs:

"Given the last 10 interactions with this user, what are the three most important
 things to know about their preferences, working style, and current project state?"

Reflection produces compact, high-signal semantic memories from noisy episodic records. Generative Agents (Park et al., 2023) showed that reflection loops dramatically improve the coherence of long-running simulated agents. The reflect step should run: (a) periodically (every N episodes), (b) before a session ends, or © when a new episode contradicts an existing fact.

8.5.5 Memory Compaction¶

Even with external memory, the in-context portion — the conversation so far — grows unboundedly in a long session. Memory compaction is the process of replacing a long conversation prefix with a shorter, lossier representation while keeping the most important information.

8.5.5.1 Summarisation-Based Compaction¶

The simplest approach: when the conversation exceeds a threshold (say, 80% of the context window), ask the model to summarise the prefix into a bullet list, then discard the prefix and replace it with the summary.

This preserves recent turns verbatim and compresses older turns. The rolling-summary pattern applies this recursively: the next time you compact, you summarise the previous summary plus the new window.

Loss is guaranteed. You must accept that some information will not survive compaction. This is the right trade-off for most tasks, but you should write high-value facts to the external store before compacting, not after.

8.5.5.2 Hierarchical Compaction¶

For very long sessions, a two-level hierarchy works well:

1. Micro-summaries: every 20 turns, summarise into a 200-word snippet and push to the episodic log.
2. Macro-summaries: every 10 micro-summaries, distil into a semantic fact update via a reflect call.

The context then always contains: system prompt + macro-summary + recent micro-summary + last N turns. This is bounded in size regardless of session length.

8.5.5.3 Structured Compaction¶

Rather than free-text summaries, some agents compact into structured formats:

{
  "task_state": "In progress — refactoring auth module",
  "decisions_made": ["Use JWT HS256", "Add rate limiting at 500 req/s"],
  "open_questions": ["How to handle token refresh for mobile clients?"],
  "user_preferences": {"language": "TypeScript", "style_guide": "Google"},
  "next_steps": ["Implement refresh endpoint", "Write integration tests"]
}

This structured compaction is more retrievable and less prone to hallucination than free-text summaries because the agent is filling in a schema rather than generating freely.

8.5.6 Session Persistence: How Agents Survive Restarts¶

A naive agent starts each session with a blank context. A persistent agent carries its prior self into every new session. The architecture has four moving parts:

1. Session ID: a stable identifier for this user/project/thread. Every memory write tags itself with the session ID; every session-start retrieval filters by it.

2. Bootstrap read: the first thing the harness does is query external memory for entries relevant to this session, and prepend them to the system prompt. This gives the agent an immediate sense of continuity.

3. Graceful shutdown write: when a session ends (timeout, user exit, token limit), the harness triggers a compaction pass that writes a session summary to the episodic store and updates any semantic facts that changed.

4. Cross-session reflection: periodically (for example, at the start of a new week), a background job runs a reflection pass over all episodes, updates semantic facts, and prunes redundant episodic entries. This is the equivalent of sleeping — the agent consolidates its knowledge.

The inter-session state that must be persisted is at minimum: the session summary, any newly learned facts, and any open task state (what was in progress when the session ended).

Memory at 100M parameters. Everything above assumes the model can be trusted to decide when to write and what to distil. A ~100M model cannot: it has neither the instruction-following reliability to emit a remember() call at the right moment nor the in-context-learning strength to write a good reflection — the same limitation that forces trajectory distillation in A Narrow Auto-Research Agent. So for Stack-100M the entire write/read/reflect triad moves harness-side: the harness writes structured slots deterministically after each verified tool result, injects a fixed ## Known facts block at a known position in every prompt (position stability also keeps the prefix cache warm), and runs the reflect step offline with a larger teacher model whose distilled facts the small model only ever reads. The budgets shrink accordingly: with the 8 192-token context that mid-training gives us (Mid-Training), the injection budget is a few hundred tokens rather than 2 000 — top-\(k\) of 2–3 short facts, and compaction triggered much earlier than 75%.

8.5.7 A From-Scratch Memory Store Implementation¶

The following implementation is a self-contained, runnable memory system using only Python standard library plus numpy and sentence-transformers. It implements all three memory types: episodic log (append-only), semantic facts (key-value), and a vector store for similarity search. It is designed to be readable and hackable rather than production-hardened.

"""
agent_memory.py  —  A complete from-scratch memory store for LLM agents.

Requires:
    pip install numpy sentence-transformers

The design:
  - EpisodicLog:   append-only JSONL file, timestamped entries.
  - SemanticStore: JSON key-value for structured facts.
  - VectorMemory:  numpy-backed cosine similarity search over embedded entries.
  - AgentMemory:   unified facade that orchestrates all three.
"""

from __future__ import annotations

import json
import os
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any

import numpy as np

# We use sentence-transformers for embeddings; swap in any provider.
# lazy-import to avoid paying the cost when not using vector retrieval.
_embedder = None

def _get_embedder():
    global _embedder
    if _embedder is None:
        from sentence_transformers import SentenceTransformer
        # 'all-MiniLM-L6-v2' produces 384-dim vectors; small but fast.
        _embedder = SentenceTransformer("all-MiniLM-L6-v2")
    return _embedder


# ─────────────────────────── Data structures ───────────────────────────

@dataclass
class MemoryEntry:
    """A single memory item stored in the vector index."""
    id: str                    # UUID or hash
    text: str                  # The content to embed and return
    metadata: dict[str, Any]   # timestamp, session_id, topic, type, etc.
    # The vector is stored separately in the numpy array; not serialised here.


# ─────────────────────────── Episodic log ──────────────────────────────

class EpisodicLog:
    """
    Append-only JSONL log of interaction episodes.
    Each entry has: timestamp, session_id, role, content, metadata.
    """

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        # Create file if it does not exist.
        self.path.touch(exist_ok=True)

    def append(self, session_id: str, role: str, content: str,
               metadata: dict[str, Any] | None = None) -> None:
        """Write one episode entry. Thread-safe via append mode + file lock."""
        entry = {
            "timestamp": time.time(),
            "iso_time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "session_id": session_id,
            "role": role,           # 'user', 'assistant', 'tool', 'system'
            "content": content,
            "metadata": metadata or {},
        }
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")

    def tail(self, n: int = 20, session_id: str | None = None) -> list[dict]:
        """
        Return the last n entries, optionally filtered by session_id.
        We read the whole file; for production, use a proper DB.
        """
        entries = [
            json.loads(line)
            for line in self.path.read_text("utf-8").splitlines()
            if line.strip()
        ]
        if session_id:
            entries = [e for e in entries if e["session_id"] == session_id]
        return entries[-n:]

    def load_session(self, session_id: str) -> list[dict]:
        """Return all entries for a given session, in order."""
        return [
            json.loads(line)
            for line in self.path.read_text("utf-8").splitlines()
            if line.strip() and json.loads(line).get("session_id") == session_id
        ]


# ─────────────────────────── Semantic store ────────────────────────────

class SemanticStore:
    """
    A key-value store for structured semantic facts.
    Backed by a single JSON file. Keys are canonical strings like
    'user.language' or 'project.database.engine'.
    """

    def __init__(self, path: str | Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        if self.path.exists():
            self._data: dict[str, Any] = json.loads(self.path.read_text("utf-8"))
        else:
            self._data = {}

    def set(self, key: str, value: Any, source: str = "agent",
            confidence: float = 1.0) -> None:
        """
        Upsert a semantic fact. Records the update time, source, and
        confidence score alongside the value.
        """
        self._data[key] = {
            "value": value,
            "updated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "source": source,
            "confidence": confidence,
        }
        self._flush()

    def get(self, key: str, default: Any = None) -> Any:
        """Return the value for key, or default if not found."""
        entry = self._data.get(key)
        return entry["value"] if entry else default

    def get_all(self) -> dict[str, Any]:
        """Return all facts as {key: value} (stripping metadata)."""
        return {k: v["value"] for k, v in self._data.items()}

    def delete(self, key: str) -> None:
        """Remove a fact."""
        self._data.pop(key, None)
        self._flush()

    def _flush(self) -> None:
        self.path.write_text(json.dumps(self._data, indent=2), "utf-8")

    def as_context_string(self) -> str:
        """
        Format all facts as a human-readable block suitable for injection
        into a system prompt.
        """
        if not self._data:
            return "(no semantic facts stored)"
        lines = ["## Known facts"]
        for k, v in self._data.items():
            conf_note = f" (confidence: {v['confidence']:.0%})" \
                        if v["confidence"] < 0.9 else ""
            lines.append(f"- **{k}**: {v['value']}{conf_note}")
        return "\n".join(lines)


# ─────────────────────────── Vector memory ─────────────────────────────

class VectorMemory:
    """
    Cosine-similarity vector store backed by numpy arrays.
    For production, replace the numpy backend with FAISS, Qdrant, or Chroma.

    Storage format (one directory):
      vectors.npy    — float32 array, shape [N, D]
      entries.jsonl  — one JSON entry per line (id, text, metadata)
    """

    def __init__(self, directory: str | Path, dim: int = 384):
        self.dir = Path(directory)
        self.dir.mkdir(parents=True, exist_ok=True)
        self.dim = dim
        self._vectors: np.ndarray   # shape [N, D]
        self._entries: list[MemoryEntry]
        self._load()

    # ── Persistence ──────────────────────────────────────────────────

    def _load(self) -> None:
        vec_path = self.dir / "vectors.npy"
        ent_path = self.dir / "entries.jsonl"
        if vec_path.exists() and ent_path.exists():
            self._vectors = np.load(str(vec_path))
            self._entries = [
                MemoryEntry(**json.loads(line))
                for line in ent_path.read_text("utf-8").splitlines()
                if line.strip()
            ]
        else:
            self._vectors = np.empty((0, self.dim), dtype=np.float32)
            self._entries = []

    def _save(self) -> None:
        np.save(str(self.dir / "vectors.npy"), self._vectors)
        lines = [json.dumps(asdict(e)) for e in self._entries]
        (self.dir / "entries.jsonl").write_text("\n".join(lines), "utf-8")

    # ── Write ─────────────────────────────────────────────────────────

    def add(self, text: str, metadata: dict[str, Any] | None = None,
            entry_id: str | None = None) -> str:
        """
        Embed text and add it to the store. Returns the entry ID.
        If an entry with the same ID already exists, it is replaced.
        """
        import hashlib
        metadata = metadata or {}
        entry_id = entry_id or hashlib.sha256(
            (text + str(time.time())).encode()
        ).hexdigest()[:16]

        # Embed the text (normalise to unit sphere for cosine = dot product).
        embedder = _get_embedder()
        vec = embedder.encode([text], normalize_embeddings=True)[0].astype(np.float32)

        # If ID already exists, overwrite.
        existing_idx = next(
            (i for i, e in enumerate(self._entries) if e.id == entry_id), None
        )
        if existing_idx is not None:
            self._vectors[existing_idx] = vec
            self._entries[existing_idx] = MemoryEntry(
                id=entry_id, text=text, metadata=metadata
            )
        else:
            self._vectors = np.vstack([self._vectors, vec[np.newaxis]])
            self._entries.append(
                MemoryEntry(id=entry_id, text=text, metadata=metadata)
            )

        self._save()
        return entry_id

    # ── Read ──────────────────────────────────────────────────────────

    def search(self, query: str, top_k: int = 5,
               min_score: float = 0.0,
               filter_metadata: dict[str, Any] | None = None
               ) -> list[tuple[MemoryEntry, float]]:
        """
        Return the top_k most similar entries to query, with their scores.
        Optionally filter by exact metadata key-value pairs.
        """
        if len(self._entries) == 0:
            return []

        embedder = _get_embedder()
        q_vec = embedder.encode([query], normalize_embeddings=True)[0].astype(np.float32)

        # Dot product = cosine similarity when both vectors are L2-normalised.
        scores = self._vectors @ q_vec  # shape [N]

        # Apply metadata filter.
        if filter_metadata:
            mask = np.array([
                all(e.metadata.get(k) == v for k, v in filter_metadata.items())
                for e in self._entries
            ])
            scores = scores * mask + (mask - 1) * 1e9  # push filtered out to -inf

        # Rank and return.
        ranked_idx = np.argsort(-scores)
        results = []
        for idx in ranked_idx[:top_k]:
            score = float(scores[idx])
            if score < min_score:
                break
            results.append((self._entries[idx], score))
        return results

    # ── Delete ─────────────────────────────────────────────────────────

    def delete(self, entry_id: str) -> bool:
        """Remove an entry by ID. Returns True if found."""
        idx = next(
            (i for i, e in enumerate(self._entries) if e.id == entry_id), None
        )
        if idx is None:
            return False
        self._vectors = np.delete(self._vectors, idx, axis=0)
        self._entries.pop(idx)
        self._save()
        return True

    def __len__(self) -> int:
        return len(self._entries)


# ─────────────────────────── Unified facade ────────────────────────────

class AgentMemory:
    """
    Unified memory facade for an LLM agent.

    Usage pattern:
        mem = AgentMemory(base_dir=".agent/memory", session_id="sess-001")

        # At session start: bootstrap context
        context_str = mem.bootstrap_context()

        # During session: log episodes, store facts
        mem.log("user", "I prefer TypeScript")
        mem.remember("user.language", "TypeScript")

        # After an important finding: add to vector store
        mem.store("JWT library has a bug in token refresh endpoint",
                  metadata={"topic": "auth", "type": "bug"})

        # Retrieve relevant memories
        hits = mem.recall("token refresh", top_k=3)

        # At session end: write summary
        mem.close_session(summary="Refactored auth module; found JWT bug.")
    """

    def __init__(self, base_dir: str | Path, session_id: str):
        self.base_dir = Path(base_dir)
        self.session_id = session_id
        self.episodic = EpisodicLog(self.base_dir / "episodes.jsonl")
        self.semantic  = SemanticStore(self.base_dir / "facts.json")
        self.vectors   = VectorMemory(self.base_dir / "vectors")

    # ── Write ─────────────────────────────────────────────────────────

    def log(self, role: str, content: str,
            metadata: dict[str, Any] | None = None) -> None:
        """Append an episode to the episodic log."""
        meta = {"session_id": self.session_id, **(metadata or {})}
        self.episodic.append(
            session_id=self.session_id, role=role,
            content=content, metadata=meta
        )

    def remember(self, key: str, value: Any,
                 confidence: float = 1.0) -> None:
        """Upsert a semantic fact."""
        self.semantic.set(key, value, source=self.session_id,
                          confidence=confidence)

    def store(self, text: str,
              metadata: dict[str, Any] | None = None) -> str:
        """Add a free-text memory to the vector store."""
        meta = {"session_id": self.session_id,
                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                **(metadata or {})}
        return self.vectors.add(text, metadata=meta)

    # ── Read ──────────────────────────────────────────────────────────

    def recall(self, query: str, top_k: int = 5,
               min_score: float = 0.3) -> list[tuple[str, float]]:
        """
        Query the vector store. Returns list of (text, score) tuples.
        Score is cosine similarity in [0, 1] (since vectors are normalised).
        """
        hits = self.vectors.search(query, top_k=top_k, min_score=min_score)
        return [(entry.text, score) for entry, score in hits]

    def bootstrap_context(self, top_k_memories: int = 5) -> str:
        """
        Build the context injection string for session start.
        Combines semantic facts + recent episodes + vector-recalled memories.
        """
        parts = []

        # 1. Semantic facts (always include all of them if small).
        facts = self.semantic.as_context_string()
        parts.append(facts)

        # 2. Recent episodic entries (last 5 from this session).
        recent = self.episodic.tail(n=5, session_id=self.session_id)
        if recent:
            lines = ["## Recent episode summary"]
            for e in recent:
                lines.append(f"[{e['iso_time']}] {e['role']}: {e['content'][:200]}")
            parts.append("\n".join(lines))

        # 3. Cross-session vector memories (not filtered to session).
        hits = self.vectors.search(
            "recent tasks and important findings",
            top_k=top_k_memories,
            min_score=0.25,
        )
        if hits:
            lines = ["## Recalled memories"]
            for entry, score in hits:
                lines.append(f"- [{score:.2f}] {entry.text}")
            parts.append("\n".join(lines))

        return "\n\n".join(parts)

    def close_session(self, summary: str) -> None:
        """
        Write a session-end summary to both the episodic log and vector store.
        Call this before clearing the context window.
        """
        self.log("system", f"[SESSION SUMMARY] {summary}",
                 metadata={"type": "session_summary"})
        self.store(summary,
                   metadata={"type": "session_summary",
                             "session_id": self.session_id})

This gives you an immediately usable memory layer. The vector backend is pure numpy — swap in faiss.IndexFlatIP or qdrant_client for production with millions of entries.

8.5.8 Worked Example: Memory Budget and Retrieval Magnitudes¶

8.5.9 The Reflect Step: Distilling Episodic to Semantic¶

The reflect step is a scheduled LLM call over recent episodic entries. Here is the pattern as a Python function that fits into the AgentMemory facade:

def reflect(
    mem: AgentMemory,
    llm_call: callable,          # func(prompt: str) -> str
    n_recent_episodes: int = 30,
) -> list[tuple[str, str]]:
    """
    Run a reflection pass: read recent episodes, ask the LLM to distil
    facts, and write them back to the semantic store.

    Returns a list of (key, value) pairs that were upserted.
    """
    episodes = mem.episodic.tail(n=n_recent_episodes,
                                 session_id=mem.session_id)
    if not episodes:
        return []

    # Format episodes as a readable transcript.
    transcript = "\n".join(
        f"[{e['iso_time']}] {e['role']}: {e['content']}"
        for e in episodes
    )

    prompt = f"""You are reviewing the following conversation transcript.
Extract a list of lasting facts, preferences, or decisions that should
be remembered for future sessions. Format your response as JSON:
[
  {{"key": "canonical.key", "value": "concise fact", "confidence": 0.9}},
  ...
]
Only include facts that are stable across sessions, not one-off details.

TRANSCRIPT:
{transcript}

JSON output:"""

    raw = llm_call(prompt)

    # Robust parse: look for the JSON array in the response.
    import re, json as _json
    match = re.search(r"\[.*\]", raw, re.DOTALL)
    if not match:
        return []

    facts = _json.loads(match.group())
    updated = []
    for fact in facts:
        key   = fact.get("key", "").strip()
        value = fact.get("value", "").strip()
        conf  = float(fact.get("confidence", 0.8))
        if key and value:
            mem.remember(key, value, confidence=conf)
            updated.append((key, value))

    return updated

This is the core of the Generative Agents reflection loop made concrete. A typical agent running on a personal software project might call reflect() at the end of each work session, discovering and persisting things like “user prefers composable functions over class hierarchies” from 30 episodes of code review feedback.

8.5.10 Ranked Recall: Recency, Importance, Relevance¶

Cosine similarity answers “which memory is most like this query?” but an agent usually wants “which memory is most useful right now?” Those differ. A three-month-old note about a database that has since been migrated is highly similar and actively harmful; a decision made an hour ago is what you actually need. Generative Agents (Park et al., 2023) scores each candidate memory \(m\) against query \(q\) with a weighted sum of three separately normalised components:

Here \(\Delta t(m)\) is the number of hours since the memory was last accessed (not created — retrieving a memory refreshes it, exactly as in an LRU cache), \(\gamma\) is a decay factor just below 1, and \(I(m)\) is an importance score the LLM assigns once at write time (“on a scale of 1 to 10, how consequential is this?”). Park et al. use \(\gamma = 0.99\), so recency weight halves after \(\ln 0.5 / \ln 0.99 \approx 69\) hours — roughly three days — and they set all three \(\alpha = 1\) after min-max normalising each component to \([0,1]\). The normalisation matters: raw cosine scores for a sentence embedder cluster in a narrow band (often 0.3–0.8), so without rescaling the relevance term would be nearly constant and the ranker would collapse into pure recency.

def recall_ranked(mem: AgentMemory, query: str, top_k: int = 5,
                  gamma: float = 0.99, now: float | None = None,
                  weights: tuple[float, float, float] = (1.0, 1.0, 1.0),
                  n_candidates: int = 50) -> list[tuple[str, float]]:
    """
    Generative-Agents-style retrieval: recency + importance + relevance.
    Expects entries written with metadata['last_access'] (epoch seconds) and
    metadata['importance'] in [0, 1]; both default sensibly if absent.

    Two-stage: cheap ANN/dot-product recall of a wide candidate set, then an
    exact re-rank of those candidates. This is the standard retrieve-then-rerank
    shape -- you never score the whole store with the expensive function.
    """
    now = now if now is not None else time.time()
    cands = mem.vectors.search(query, top_k=n_candidates, min_score=-1.0)
    if not cands:
        return []

    rel = np.array([s for _, s in cands], dtype=np.float32)
    hours = np.array(
        [(now - float(e.metadata.get("last_access", now))) / 3600.0
         for e, _ in cands], dtype=np.float32)
    rec = np.power(gamma, hours, dtype=np.float32)
    imp = np.array([float(e.metadata.get("importance", 0.5))
                    for e, _ in cands], dtype=np.float32)

    def minmax(x: np.ndarray) -> np.ndarray:
        span = float(x.max() - x.min())
        # A flat component carries no ranking signal; make it neutral.
        return np.full_like(x, 0.5) if span < 1e-9 else (x - x.min()) / span

    w_rec, w_imp, w_rel = weights
    score = w_rec * minmax(rec) + w_imp * minmax(imp) + w_rel * minmax(rel)

    order = np.argsort(-score)[:top_k]
    for i in order:                       # refresh recency on access
        entry = cands[int(i)][0]
        entry.metadata["last_access"] = now
    mem.vectors._save()
    return [(cands[int(i)][0].text, float(score[i])) for i in order]

The three weights are the knobs worth tuning per application. A coding agent working inside one repository wants \(\alpha_{\text{rec}}\) high (the codebase changes under it); a personal assistant recalling stable preferences wants \(\alpha_{\text{rec}}\) near zero, because “the user prefers British English” does not become less true after three days.

8.5.11 The Library Layer: LangGraph, mem0, and Letta¶

You now know exactly what a memory system does, because you built one. In production you rarely hand-roll it — and the libraries below are worth reading precisely because each one is a recognisable instance of the architecture in this chapter.

8.5.11.1 LangGraph: checkpointers (short-term) and stores (long-term)¶

LangGraph splits memory along exactly the axis of section 8.5.1. A checkpointer persists the graph state after every step, scoped to a thread_id — that is short-term, in-context memory made durable, so a crashed or resumed session replays from the last checkpoint. A store (BaseStore) is a namespaced key-value store with optional vector indexing that lives across threads — that is long-term memory. Both are injected into your node functions by the runtime.

# pip install langgraph langgraph-checkpoint-sqlite sentence-transformers
from langgraph.checkpoint.memory import InMemorySaver   # swap: SqliteSaver, PostgresSaver
from langgraph.store.memory import InMemoryStore        # swap: PostgresStore
from langgraph.store.base import BaseStore
from langgraph.graph import StateGraph, MessagesState, START
from langchain_core.runnables import RunnableConfig

def embed(texts: list[str]) -> list[list[float]]:
    """Any callable list[str] -> list[list[float]] works; reuse our embedder."""
    return _get_embedder().encode(texts, normalize_embeddings=True).tolist()

# Long-term: cross-thread, semantically searchable (our VectorMemory).
store = InMemoryStore(index={"embed": embed, "dims": 384})
# Short-term: per-thread state, snapshotted after every node (our EpisodicLog).
checkpointer = InMemorySaver()

def agent_node(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    ns = ("memories", user_id)          # namespace tuple == the scope of a memory
    last = state["messages"][-1].content

    hits = store.search(ns, query=last, limit=3)             # READ  (vector recall)
    recalled = "\n".join(f"- {i.value['text']}" for i in hits)
    # ... here you would call the model with `recalled` prepended to the system
    #     prompt; we echo it so the example runs without an API key ...
    store.put(ns, f"mem-{len(state['messages'])}", {"text": last})   # WRITE

    return {"messages": [{"role": "assistant",
                          "content": f"recalled:\n{recalled or '(nothing yet)'}"}]}

g = StateGraph(MessagesState)
g.add_node("agent", agent_node)
g.add_edge(START, "agent")
app = g.compile(checkpointer=checkpointer, store=store)

cfg = {"configurable": {"thread_id": "thread-1", "user_id": "alice"}}
app.invoke({"messages": [{"role": "user", "content": "I prefer British English"}]}, cfg)
# A *new* thread_id starts a fresh conversation but still sees alice's store.
app.invoke({"messages": [{"role": "user", "content": "Which spelling do I use?"}]},
           {"configurable": {"thread_id": "thread-2", "user_id": "alice"}})

The namespace tuple is the piece most people underuse: ("memories", user_id) versus ("memories", user_id, "preferences") gives you the episodic/semantic separation of section 8.5.2 for free, because search takes a namespace prefix. Swapping InMemorySaver for SqliteSaver.from_conn_string("checkpoints.sqlite") and InMemoryStore for PostgresStore is the only change needed to go from a demo to a durable multi-user service.

8.5.11.2 mem0 and Letta: managed memory layers¶

mem0 takes the opposite stance: instead of you deciding what to write, you hand it raw messages and an LLM decides. Its extraction pass proposes candidate facts, and a second pass reconciles them against what is already stored by emitting one of ADD / UPDATE / DELETE / NOOP — an automated version of the “upsert on new evidence” rule in our table, including the contradiction handling that a naive append-only vector store lacks.

# pip install mem0ai   (needs an LLM + embedder configured; see the repo README)
from mem0 import Memory

m = Memory()
m.add([{"role": "user", "content": "I prefer British English and TypeScript"}],
      user_id="alice")                      # LLM extracts facts, then ADD/UPDATE/DELETE
print(m.search("which language does the user write in?", user_id="alice"))
print(m.get_all(user_id="alice"))

Letta (formerly MemGPT) implements the OS-paging metaphor literally: the agent has a small in-context core memory of editable blocks plus paged-out archival and recall storage, and it edits its own core memory through tools (core_memory_append, core_memory_replace) rather than through harness code. That self-editing design is powerful with a frontier model and unusable at 100M parameters — which is exactly the trade-off flagged in section 8.5.6.

Choosing: hand-roll (as in 8.5.7) when you want auditable files and no dependencies; LangGraph when memory must be consistent with agent control flow and durable execution; mem0 or Letta when you want fact reconciliation and self-editing without writing the reflect loop yourself.

8.5.11.3 Does it actually work? Evaluating memory¶

Memory systems fail silently: the agent simply does not mention the thing it should have remembered, and no exception is raised. Two public benchmarks are the standard checks. LoCoMo (Maharana et al., 2024) tests question answering over very long multi-session conversations, probing single-hop, multi-hop, and temporal reasoning across sessions. LongMemEval (Wu et al., ICLR 2025) is the sharper diagnostic: roughly 500 questions across five abilities — information extraction, multi-session reasoning, temporal reasoning, knowledge updates (does the system honour a correction?), and abstention (does it say “I don’t know” instead of confabulating?) — and it reports a substantial accuracy drop for commercial assistants as history grows. Run these before trusting a memory design, and build the same five categories into your own regression suite; the knowledge-update and abstention categories are the ones hand-rolled append-only stores fail most often. For the general evaluation machinery, see Agent Evaluation & Benchmarks.

Summary¶

Memory is the difference between a stateless chatbot and a persistent agent. We have covered the full stack:

• Short-term vs long-term memory — the context window is the working memory; external stores are the long-term memory.
• Episodic vs semantic — timestamped events vs stable facts; do not mix them indiscriminately.
• Store types — vector stores for fuzzy similarity, key-value for deterministic retrieval, graphs for relational knowledge, file systems for human-readable persistence.
• Write / read / reflect — three operations that every memory-capable agent must implement.
• Compaction — you must proactively manage context size or the window fills and information is lost.
• Session persistence — bootstrap read, in-session writes, graceful shutdown, and periodic reflection.
• Ranked recall — recency, importance, and relevance combined, not cosine similarity alone.
• The library layer — LangGraph checkpointers/stores, mem0’s LLM-managed fact reconciliation, Letta’s self-editing core memory — and LoCoMo/LongMemEval to check that any of it works.

The code in this chapter gives you a concrete starting point. In the next chapter we look at how the Model Context Protocol (MCP) standardises the interface through which agents access both memory and tools. For retrieval-heavy applications, the detailed treatment of ANN indexes and hybrid search in Vector Databases & Approximate Nearest Neighbor Search and Advanced RAG: GraphRAG, Agentic RAG & Long-Context vs RAG will be essential.

Further Reading¶

• Park et al., “Generative Agents: Interactive Simulacra of Human Behavior,” 2023 — introduced the episodic + reflection + retrieval architecture for persistent agent behaviour; the landmark paper on LLM memory systems.
• Zep — the Memory Layer for AI (getzep.com) — production system implementing temporal knowledge graphs and semantic extraction; useful reference architecture.
• MemGPT: Towards LLMs as Operating Systems, Packer et al., 2023 — framed memory management as an OS paging problem; introduced the virtual-context-manager pattern for infinite-length conversations.
• LangGraph (LangChain) — open-source agent framework with first-class support for persistent state stores across agent steps; good reference implementation of session persistence.
• Cognee (topoteretes/cognee on GitHub) — open-source memory layer with knowledge-graph extraction from unstructured text; shows a practical graph-memory pipeline.
• Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks, Lewis et al., 2020 — the foundational RAG paper; motivates why external retrieval beats storing everything in weights.

Exercises¶

1. For each of the following memory items, decide whether it belongs in the episodic store or the semantic store, and give a one-line justification. Then explain, using the argument in section 8.5.2, why storing (a) and © together in a single undifferentiated vector store degrades retrieval quality.

• (a) “2025-11-04 14:12: the user asked me to refactor the auth module and I found a bug in the JWT refresh endpoint.”
• (b) “The production database engine is PostgreSQL 15.”
• © “2025-11-05 09:30: reran the integration suite; 3 tests still failing on token refresh.”
• (d) “The user prefers British English spelling.”

2. You call mem.recall(query, top_k=3, min_score=0.3). The query embeds (after L2 normalisation) to \(q = [0.6,\ 0.8,\ 0.0]\). The vector store holds four already-normalised memory vectors:

• \(m_1 = [0.6,\ 0.8,\ 0.0]\)
• \(m_2 = [0.0,\ 1.0,\ 0.0]\)
• \(m_3 = [1.0,\ 0.0,\ 0.0]\)
• \(m_4 = [0.8,\ -0.6,\ 0.0]\)

Compute each cosine similarity by hand, then state exactly which entries recall returns and in what order. (Recall from the code that search sorts by descending score and breaks as soon as a score falls below min_score.)

[(m1_text, 1.00), (m2_text, 0.80), (m3_text, 0.60)]

3. You are sizing a memory store with a 1024-dimensional float32 embedding model and a target of \(N = 250{,}000\) memories, each averaging 80 tokens of text.

• (a) Compute the raw vector storage in bytes and MB/GB.
• (b) Compute the total text token count and the UTF-8 text storage at 4 bytes/token.
• © If the injection budget is 1500 tokens per call, what is the maximum number of 80-token entries you could inject, and how does that compare to the chapter’s recommended top_k?
• (d) At $2.00 per million input tokens, what does injecting top_k = 5 entries (80 tokens each) cost per call, and over 50 000 calls?

4. The “writing everything verbatim” pitfall recommends fix ©: run a deduplication pass that merges near-identical memories. Implement a deduplicate(self, threshold: float = 0.95) -> int method on VectorMemory that removes near-duplicate entries (keeping the first member of each duplicate group) and returns the number removed. Exploit the fact that the stored vectors are already L2-normalised. Explain why that normalisation makes the check a single dot product.

def deduplicate(self, threshold: float = 0.95) -> int:
    """
    Remove near-duplicate entries. Keeps the first member of each
    near-duplicate group; deletes the rest. Returns the number removed.
    Relies on stored vectors being L2-normalised (cosine == dot product).
    """
    if len(self._entries) < 2:
        return 0

    kept_entries: list[MemoryEntry] = []
    kept_vecs: list[np.ndarray] = []
    removed = 0

    for i, entry in enumerate(self._entries):
        vec = self._vectors[i]
        # cosine == dot product because both operands are unit vectors
        is_dup = any(float(vec @ kv) >= threshold for kv in kept_vecs)
        if is_dup:
            removed += 1
        else:
            kept_entries.append(entry)
            kept_vecs.append(vec)

    if removed:
        self._entries = kept_entries
        self._vectors = (
            np.vstack(kept_vecs) if kept_vecs
            else np.empty((0, self.dim), dtype=np.float32)
        )
        self._save()
    return removed

5. Implement the structured compaction trigger described in the practitioner tip and section 8.5.5.3. Write a function compact_if_needed(mem, transcript_tokens, llm_call, ctx_window=128_000, threshold=0.75) that: (a) does nothing and returns None while the transcript is below threshold of the context window; (b) otherwise asks the LLM to serialise current state into the structured JSON schema from 8.5.5.3, parses it robustly, persists the fields to the semantic store, and returns the parsed dict. Why is serialising into this schema more reliable than a free-text summary?

import re, json as _json

def compact_if_needed(mem, transcript_tokens, llm_call,
                      ctx_window: int = 128_000,
                      threshold: float = 0.75):
    """
    Trigger a structured compaction pass once the in-context transcript
    crosses `threshold` of the context window. Persists the compacted
    state to the semantic store and returns the parsed dict (or None
    if compaction was not needed).
    """
    if transcript_tokens < threshold * ctx_window:
        return None  # still room; do not compact yet

    prompt = f"""Serialise the current session state into EXACTLY this JSON schema:
{{
  "task_state": "one-line status",
  "decisions_made": ["..."],
  "open_questions": ["..."],
  "user_preferences": {{"key": "value"}},
  "next_steps": ["..."]
}}
Fill every field from the conversation so far. Output only the JSON object.

JSON output:"""

    raw = llm_call(prompt)

    # Robust parse: grab the first {...} block.
    match = re.search(r"\{.*\}", raw, re.DOTALL)
    if not match:
        return None
    state = _json.loads(match.group())

    # Persist each field to the semantic store under canonical keys.
    mem.remember("compaction.task_state", state.get("task_state", ""))
    mem.remember("compaction.decisions_made", state.get("decisions_made", []))
    mem.remember("compaction.open_questions", state.get("open_questions", []))
    mem.remember("compaction.next_steps", state.get("next_steps", []))
    for k, v in (state.get("user_preferences") or {}).items():
        mem.remember(f"user.{k}", v)

    return state