Long-term agent memory

1 · The window is not memory essential

A chat model is stateless between calls: it remembers nothing on its own. Everything it "knows" in a turn is whatever text you put in the context window right now — the same statelessness you first met in Ch 1 (setup). Chat apps fake memory by re-sending the whole transcript every turn. That works until the conversation outgrows the window, spans sessions, or you have thousands of users — then the transcript can't hold it all. Long-term memory is a separate store you write facts to and read facts from, so the agent can recall things from last week without re-sending last week.

2 · Short-term vs long-term memory essential

Short-term memory is the current context window: this turn's transcript, cheap to use, gone when the window fills or the session ends. Long-term memory is a durable external store (a file, a DB, a vector index) that survives across sessions and only loads what's relevant to this turn. The window is your desk; long-term memory is the filing cabinet you fetch the right folder from.

3 · Episodic vs semantic memory essential

Borrowing from cognitive science, split long-term memory two ways. Episodic memory is events: "on Tuesday the user asked to cancel order #42." Semantic memory is distilled facts: "the user prefers email over phone." Episodes are raw and timestamped; facts are stable and deduplicated. A good agent writes episodes as they happen and, over time, consolidates them into durable semantic facts.

4 · The memory loop intermediate

Every memory framework — Mem0, Letta, home-grown — is a loop. From each turn you extract what's worth keeping, write it to the long-term store; on the next turn you retrieve the memories relevant to the new message and inject them into the context. The model stays stateless; the loop supplies the state.

Note the read path is exactly retrieval (Ch 3 · RAG) — but the corpus is the user's own past, not a doc library. Memory is RAG over yourself.

5 · A tiny MemoryStore (write + retrieve) intermediate

The read path scores each stored memory against the new message and returns the best matches. The simplest scorer is bag-of-words overlap: how many words do they share? (Real systems use embedding similarity — same idea, better matching; see Ch 3.) This runs offline with the standard library only.

memory_store.pyimport re

# stopwords carry no signal; drop them so overlap ranks on meaningful words
STOP = {"the", "a", "an", "i", "is", "of", "on", "to", "how", "should",
        "this", "what", "over", "in", "user", "customer"}

def tokenize(text):
    return {w for w in re.findall(r"[a-z0-9]+", text.lower()) if w not in STOP}

class MemoryStore:
    def __init__(self):
        self.memories = []          # list of {"text":..., "toks":...}

    def write(self, text):
        self.memories.append({"text": text, "toks": tokenize(text)})

    def retrieve(self, query, k=2):
        q = tokenize(query)
        scored = []
        for m in self.memories:
            overlap = len(q & m["toks"])          # shared meaningful words
            if overlap:
                scored.append((overlap, m["text"]))
        scored.sort(key=lambda x: (-x[0], x[1]))  # best first, stable
        return [text for _, text in scored[:k]]

store = MemoryStore()
store.write("the user prefers email over phone")
store.write("the user is a premium-tier customer")
store.write("the user lives in Berlin")

print(store.retrieve("how should I contact the user by phone or email?"))
print(store.retrieve("what tier is this customer on, premium?"))

['the user prefers email over phone']
['the user is a premium-tier customer']

Only relevant memories come back — the "Berlin" fact stays out of a contact-method query. That selectivity is the whole point: you inject a handful of relevant memories, not the entire history.

6 · The write path — what & when to remember advanced

You can't store every sentence — that's just the transcript again. The write path decides what is worth keeping (durable facts and preferences, not chit-chat) and when. Production systems ask an LLM to extract facts; here is a rule-based stub that captures the shape, runs offline, and shows the recall being injected into a prompt.

fact_extract.pyimport re

def extract_facts(utterance):
    """Rule-based stub: keep durable 'I ...' statements, drop chit-chat."""
    u = utterance.strip()
    facts = []
    low = u.lower()
    if low.startswith(("hi", "hello", "thanks", "thank you", "bye")):
        return facts                      # chit-chat: remember nothing
    m = re.match(r"i (?:am|'m) (?:a |an )?(.+)", low)
    if m:
        facts.append("user is " + m.group(1).rstrip("."))
    m = re.search(r"i (?:prefer|like|want) (.+)", low)
    if m:
        facts.append("user prefers " + m.group(1).rstrip("."))
    m = re.search(r"my name is (\w+)", low)
    if m:
        facts.append("user's name is " + m.group(1).capitalize())
    return facts

turns = [
    "Hi there!",
    "My name is Dana",
    "I am a data engineer",
    "I prefer dark mode",
    "Thanks!",
]
memory = []
for t in turns:
    for f in extract_facts(t):
        if f not in memory:
            memory.append(f)

print("stored:", memory)
prompt = "System memory about the user:\n- " + "\n- ".join(memory)
print(prompt)

stored: ["user's name is Dana", 'user is data engineer', 'user prefers dark mode']
System memory about the user:
- user's name is Dana
- user is data engineer
- user prefers dark mode

7 · Forgetting & consolidation advanced

Memory that only grows becomes slow, expensive, and self-contradictory. Two maintenance jobs keep it healthy: consolidation (dedupe near-identical memories, and let newer facts supersede stale ones) and forgetting (drop memories that are old and rarely used). This is the store-side analogue of the context-budgeting you did earlier.

consolidate.pydef subject_of(m):
    """A memory's subject = who + which attribute (e.g. 'user/prefers', 'user/is')."""
    for verb in (" prefers ", " is ", " likes "):
        if verb in m:
            who = m.split(verb)[0]
            return who + "/" + verb.strip()
    return m

def consolidate(memories):
    """Dedup: a later fact about the same subject supersedes an earlier one."""
    by_subject = {}
    for i, m in enumerate(memories):
        by_subject[subject_of(m)] = (i, m)    # last write wins
    return [text for _, text in sorted(by_subject.values())]

def forget(memories_meta, now, max_age=30, min_uses=1):
    """Drop memories older than max_age days that were used fewer than min_uses times."""
    kept = []
    for m in memories_meta:
        age = now - m["last_used_day"]
        if age > max_age and m["uses"] < min_uses:
            continue                          # forgotten
        kept.append(m["text"])
    return kept

mems = ["user prefers phone", "user is premium", "user prefers email"]
print("consolidated:", consolidate(mems))

meta = [
    {"text": "likes jazz", "last_used_day": 2, "uses": 0},   # old + unused -> forget
    {"text": "is premium", "last_used_day": 40, "uses": 5},  # old but used  -> keep
]
print("kept:", forget(meta, now=45, max_age=30, min_uses=1))

consolidated: ['user is premium', 'user prefers email']
kept: ['is premium']

"user prefers phone" was superseded by the later "user prefers email"; the stale, unused "likes jazz" was forgotten while the frequently-used "is premium" survived. That is how a store stays small and current instead of accumulating contradictions.

8 · Mem0, Letta & MemGPT — self-editing memory professional

You don't have to build the loop from scratch. Mem0 is a memory layer: it runs the extract → embed → store → retrieve loop for you, with dedup and updates, behind a add()/search() API. MemGPT (the research paper) framed the LLM as an OS managing its own memory: a small in-context working set plus a larger external store, with the model deciding what to page in and out. Letta is the framework that grew from MemGPT — agents with self-editing memory: the model calls tools to write, update, and search its own memory rather than you doing it externally.

Both are the same loop from §4 with production polish (embeddings, dedup, updates). "Self-editing" means the write path in §6 becomes a tool the model calls — powerful, but you must still bound what it may store and for how long.

mem0_sdk.py# needs the SDK:  pip install mem0ai   (network + API key; not runnable offline)
from mem0 import Memory

mem = Memory()                       # backed by a vector store under the hood
mem.add("I prefer window seats", user_id="dana")   # extract + embed + store
hits = mem.search("seating preference?", user_id="dana")
# hits -> the relevant memories to inject into your prompt
# Same loop as memory_store.py, with real embeddings + dedup + updates.

9 · Tech-lead — a memory policy for the system tech-lead

A lead owns the whole memory contract, not just the code. Decide: what the system is allowed to remember (and explicitly what it must not — secrets, payment data); how long memories live and the forgetting policy; the retrieval budget (how many memories inject per turn — this is context engineering again); a privacy interface so users can view and delete their memories; and the cost (every stored memory is storage + an embedding + a retrieval on every turn). Write it down; a memory store is user data under management, not a cache you can forget about.

from collections import deque

class Window:
    def __init__(self, size=3):
        self.turns = deque(maxlen=size)   # oldest turns evicted automatically
    def add(self, turn): self.turns.append(turn)
    def recall(self): return list(self.turns)

w = Window(size=3)
for t in ["my name is Ada", "the sky is blue", "2+2=4", "what is my name?"]:
    w.add(t)
print(w.recall())
# ['the sky is blue', '2+2=4', 'what is my name?']  -- the name is GONE

def classify(item):
    # semantic = stable preference/fact; episodic = a dated event
    episodic_cues = ["yesterday", "today", "at ", "on monday", "just now"]
    if any(c in item.lower() for c in episodic_cues):
        return "episodic"
    if item.lower().startswith(("i prefer", "my", "i am", "i like")):
        return "semantic"
    return "episodic"

for i in ["I prefer dark mode", "Booked a flight today", "My name is Ada"]:
    print(f"{i!r} -> {classify(i)}")
# 'I prefer dark mode' -> semantic
# 'Booked a flight today' -> episodic
# 'My name is Ada' -> semantic

class MemoryStore:
    def __init__(self): self.mem = []
    def write(self, text):
        self.mem.append(text)
    def _score(self, query, text):
        q, t = set(query.lower().split()), set(text.lower().split())
        return len(q & t) / (len(q) or 1)      # word-overlap stand-in for cosine
    def retrieve(self, query, k=2):
        ranked = sorted(self.mem, key=lambda m: self._score(query, m),
                        reverse=True)
        return [m for m in ranked if self._score(query, m) > 0][:k]

ms = MemoryStore()
for f in ["user prefers dark mode", "user lives in Berlin",
          "user dislikes email notifications"]:
    ms.write(f)
print(ms.retrieve("what mode does the user prefer?"))
# ['user prefers dark mode']

class MemoryWriter:
    def __init__(self): self.mem = []
    def _similar(self, a, b):
        A, B = set(a.lower().split()), set(b.lower().split())
        return len(A & B) / len(A | B) if (A | B) else 0   # Jaccard
    def write(self, text):
        if len(text.split()) < 3:
            return "skipped: too trivial"
        for i, m in enumerate(self.mem):
            if self._similar(text, m) > 0.6:
                self.mem[i] = max(m, text, key=len)         # keep the fuller one
                return "merged with existing"
        self.mem.append(text); return "stored"

w = MemoryWriter()
print(w.write("ok"))                                   # skipped: too trivial
print(w.write("user prefers dark mode"))               # stored
print(w.write("the user prefers dark mode in the app"))# merged with existing
print(w.mem)

import time

class DecayingMemory:
    def __init__(self, capacity=3):
        self.capacity = capacity; self.mem = []   # list of dicts
    def write(self, text):
        self.mem.append({"text": text, "ts": time.time(), "hits": 0})
        self._consolidate()
    def retrieve(self, text):
        for m in self.mem:
            if text in m["text"]:
                m["hits"] += 1; return m["text"]
    def _score(self, m):
        age = time.time() - m["ts"]
        return m["hits"] * 10 - age            # frequent + recent scores high
    def _consolidate(self):
        while len(self.mem) > self.capacity:
            worst = min(self.mem, key=self._score)  # evict least valuable
            self.mem.remove(worst)

dm = DecayingMemory(capacity=2)
dm.write("A"); dm.write("B"); dm.retrieve("B")  # B accessed -> higher score
dm.write("C")                                    # over capacity -> evict A
print([m["text"] for m in dm.mem])               # ['B', 'C']

class AgentMemory:
    def __init__(self):
        self.semantic = []    # durable facts/preferences
        self.episodic = []    # timestamped events

    # SEAM 1 (needs embeddings/SDK): swap overlap for vector similarity
    def retrieve(self, query, k=3):
        pool = self.semantic + self.episodic
        score = lambda m: len(set(query.split()) & set(m.split()))
        return sorted(pool, key=score, reverse=True)[:k]

    def write(self, text, kind):
        (self.semantic if kind == "semantic" else self.episodic).append(text)

def agent_turn(memory, user_msg, classify, llm):
    context = memory.retrieve(user_msg)          # 1. retrieve before prompting
    prompt = f"Memories: {context}\nUser: {user_msg}"
    reply = llm(prompt)                          # 2. SEAM 2: real Anthropic call
    if "remember" in user_msg.lower():           # 3. write-after decision
        memory.write(user_msg, classify(user_msg))
    return reply

mem = AgentMemory()
mem.write("user prefers metric units", "semantic")
print(agent_turn(mem, "what units do I prefer?", lambda t: "semantic",
                 llm=lambda p: f"[echo] {p[:40]}"))

Knowledge check check yourself