Mini-project 5: build a memory system

You have read two chapters of theory about memory. Now you build one.

By the end of this chapter you will have a working system that does all six things:

Extracts candidate memories from a conversation.
Stores them with embeddings and full provenance.
Consolidates — including detecting and resolving a genuine contradiction.
Retrieves with scope filtering and blended relevance/recency/importance scoring.
Injects the results into the next turn’s context with confidence annotations.
Forgets — honoring a source revocation without nuking everything downstream.

It has no required dependencies and runs offline with a deterministic mock embedder, so every line in this chapter executes on your machine right now. Then we rebuild the storage layer on ChromaDB and show the mem0 equivalent, so you can go either way.

The whole thing is about 300 lines. That is not because memory is easy — it is because the hard parts are decisions, not code, and you have already made them in Chapters 4 and 5.

Part 1: embeddings you can run without an API key

Every memory system needs to turn text into a vector. For this project we use a hash embedder: deterministic, offline, and honest about being crude.

DIM = 256

def hash_embed(text: str, dim: int = DIM) -> list[float]:
    """Deterministic bag-of-words hash embedding. No network, no API key.

    Words and character trigrams are hashed into buckets, so texts sharing
    vocabulary -- or just word stems, thanks to the trigrams -- land near each
    other. It is good enough to demonstrate the mechanics and bad enough that
    you should replace it: it has no idea that "lunch" relates to "vegetarian".
    The signature matches any embedding provider's, so swapping it is one line.
    """
    def bucket(s: str) -> int:
        return int(hashlib.blake2b(s.encode(), digest_size=8).hexdigest(), 16) % dim

    vec = [0.0] * dim
    for tok in re.findall(r"[a-z0-9]+", text.lower()):
        if tok in STOPWORDS:
            continue
        vec[bucket(tok)] += 1.0
        padded = f"^{tok}$"
        for i in range(len(padded) - 2):          # trigrams give crude stemming
            vec[bucket(padded[i:i + 3])] += 0.35
    norm = math.sqrt(sum(v * v for v in vec)) or 1.0
    return [v / norm for v in vec]

def cosine(a: list[float], b: list[float]) -> float:
    return sum(x * y for x, y in zip(a, b))          # both are unit vectors

The trigrams are there so “diet” and “dietary” land near each other, which a pure bag-of-words model would not manage. What it cannot do is connect “lunch” to “vegetarian” — that requires actual semantics. You will see that limitation in the output, and it is exactly the right reason to swap in a real embedder.

To swap: replace hash_embed with a call to client.embeddings.create(...) (OpenAI) or client.models.embed_content(...) (Gemini) and normalize the result. One function, same signature.

Part 2: the data model, where provenance lives

The record is where most of the design happens, so read this carefully.

SourceType = Literal["bootstrapped", "user_explicit", "user_implicit", "tool_output"]

# Base trust by source type. Bootstrapped data comes from a system of record;
# tool output is brittle and stale by the time you read it back.
TRUST: dict[str, float] = {
    "bootstrapped":  0.95,
    "user_explicit": 0.90,
    "user_implicit": 0.65,
    "tool_output":   0.30,
}

@dataclass
class Source:
    """One thing a memory was derived from. Memories can have several."""
    source_type: SourceType
    source_id: str                     # session id, CRM record id, form id
    excerpt: str = ""                  # the span it came from, for auditing
    at: float = field(default_factory=time.time)

@dataclass
class MemoryRecord:
    content: str
    user_id: str
    app_name: str = "default"
    scope: Literal["user", "session", "app"] = "user"
    session_id: Optional[str] = None
    topic: str = "general"
    importance: float = 0.5
    half_life_days: float = 180.0
    confidence: float = 0.6
    sources: list[Source] = field(default_factory=list)
    memory_id: str = field(default_factory=lambda: "mem_" + uuid.uuid4().hex[:8])
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)
    invalidated_at: Optional[float] = None
    superseded_by: Optional[str] = None   # lineage: what replaced this
    history: list[str] = field(default_factory=list)   # audit trail of edits
    embedding: list[float] = field(default_factory=list)

    def active(self):    return self.invalidated_at is None
    def age_days(self, now=None):  ...       # (now - updated_at) / 86400
    def decayed_confidence(self, now=None):  # confidence halves every half_life
        return self.confidence * 0.5 ** (self.age_days(now) / self.half_life_days)

Five decisions in there worth defending.

sources is a list, not a field. A memory can be derived from several conversations, and consolidation makes that the normal case rather than an edge case. A single source_id string would be a lie the first time you merged anything.

invalidated_at and superseded_by, not DELETE. When a memory is contradicted, it does not disappear. It gets tombstoned, and it records what replaced it. That chain is what lets you answer “why does the system believe this” six months from now, and it costs one nullable column.

half_life_days lives on the record. A name should decay over a decade; a current-issue memory should decay in a month. A single global decay constant is wrong for both.

confidence separate from TRUST. Trust is a property of the source type. Confidence is a property of this specific memory, starting from source trust and moving with corroboration and age.

Tool output sits at 0.30. As Chapter 5 argued: memories from tool output are brittle and go stale. Encoding that as a low trust score means the consolidator will refuse to let a tool result overwrite something the user actually said.

Part 3: storage, where isolation is enforced

class MemoryStore:
    def __init__(self, path: str = ":memory:") -> None:
        self.db = sqlite3.connect(path, check_same_thread=False)
        self.db.execute("""CREATE TABLE IF NOT EXISTS memories (
            memory_id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
            app_name TEXT NOT NULL, scope TEXT NOT NULL, blob TEXT NOT NULL)""")
        self.db.execute("CREATE INDEX IF NOT EXISTS ix_owner "
                        "ON memories(user_id, app_name, scope)")
        self.db.commit()

    def put(self, m: MemoryRecord) -> None:
        """Embed if needed, then UPSERT the record as a JSON blob."""
        ...

    def scan(self, user_id: str, app_name: str = "default",
             scopes: Iterable[str] = ("user", "app"),
             include_invalid: bool = False) -> list[MemoryRecord]:
        """Scope filtering happens HERE, in code, in the WHERE clause.

        Never in a prompt. A rule the model can talk itself out of is not
        a rule, and this one is the difference between per-user isolation
        and a data leak.
        """
        scopes = tuple(scopes)
        q = ("SELECT blob FROM memories WHERE app_name=? AND scope IN "
             f"({','.join('?' * len(scopes))}) AND (user_id=? OR scope='app')")
        rows = self.db.execute(q, (app_name, *scopes, user_id)).fetchall()
        out = [self._hydrate(r[0]) for r in rows]
        return out if include_invalid else [m for m in out if m.active()]

The indexed columns — user_id, app_name, scope — are promoted out of the JSON blob deliberately. Everything you filter on in a WHERE clause needs to be a real column. Everything else can live in the blob, and putting it there means you can evolve the record shape without a migration every sprint.

scan is the security boundary of the entire system, which is why the docstring is longer than the code. The predicate (user_id = ? OR scope = 'app') is the whole isolation model: you see your own memories plus the deliberately global ones, and nothing else.

Linear cosine scan over the results is fine, and will be fine for longer than you expect — a per-user memory store rarely exceeds a few thousand records, and you have already filtered to one user before you compute a single distance.

Part 4: extraction

In production, extraction is one structured-output LLM call with the Pydantic schema from Chapter 5. For an offline, deterministic demo, we use rules — same interface, no API key:

@dataclass
class Candidate:
    content: str
    topic: str
    confidence: float
    excerpt: str
    importance: float = 0.5
    half_life_days: float = 180.0

RULES = [
    # (regex, topic, template, confidence, half_life_days)
    (r"\bi'?m (?:a )?(vegetarian|vegan|pescatarian)\b", "diet",
     "The user follows a {0} diet.", 0.9, 365),
    (r"\bi(?:'ve| have)? (?:started|gone back to) eating (fish|meat|dairy)\b",
     "diet", "The user eats {0} again.", 0.9, 365),
    (r"\bi (?:prefer|like|want) (?:the |a |an )?(window|aisle|middle) seat\b",
     "travel", "The user prefers {0} seating on flights.", 0.85, 365),
    (r"\bcall me ([A-Z]\w+)\b", "identity", "The user goes by {0}.", 0.95, 3650),
    (r"\b(?:export|upload|sync) (?:keeps )?(?:failing|timed out|times out)\b",
     "issue", "The user reported a failing export/upload.", 0.7, 30),
]

def rule_extract(messages: list[dict]) -> list[Candidate]:
    """Fire every rule against every USER message; emit one Candidate per hit."""
    ...

Notice that only role == "user" messages are considered. Extracting facts about the user from the assistant’s messages is a real bug that produces memories about things the agent asserted rather than things the user said, and it compounds — the agent remembers its own guesses as facts.

excerpt captures the source span so provenance is auditable. When someone asks “why does the system think I’m vegetarian,” you can show them the sentence.

To go live, replace rule_extract with an LLM call using this prompt and a Pydantic schema:

EXTRACTION_PROMPT = """\
Extract durable facts about the user from the conversation below.

Rules:
- One fact per memory. Never combine two facts into one sentence.
- Write in the third person, self-contained: a reader with no access to this
  conversation must understand it fully.
- Record what the user SAID or REPORTED, not what is objectively true.
- Skip pleasantries, transient details, and anything already established.
- Return an empty list if there is nothing durable. Do not invent memories.

CONVERSATION:
{conversation}
"""

Part 5: consolidation, and the bug you will write

This is the part that matters, and it contains the single most common mistake in homegrown memory systems.

Here is the mistake, stated plainly.

You cannot find contradictions with similarity search.

“The user follows a vegetarian diet” and “The user eats fish again” are a direct contradiction. They share essentially no vocabulary. No embedding model with a sane similarity threshold is going to pair them. So if your consolidation step is “find the top-k similar memories and ask the LLM about those,” contradictions slip through, both memories stay in the store, and the model gets to pick one at random every turn.

I wrote this bug while building this chapter. The first run produced CREATE where it should have produced UPDATE.

The fix is to union the semantic neighbours with a topic sweep:

def _similar(self, text, user_id, app_name, topic=""):
    """Candidate memories that this new information might affect.

    Similarity search alone is NOT enough here, and this is the single
    most common bug in homegrown memory systems. "The user follows a
    vegetarian diet" and "The user eats fish again" share almost no
    vocabulary, so no embedding search with a sane threshold will pair
    them -- and they are a direct contradiction. Union the semantic
    neighbours with everything on the same topic.
    """
    e = hash_embed(text)
    existing = self.store.scan(user_id, app_name)
    scored = sorted(((cosine(e, m.embedding), m) for m in existing),
                    key=lambda p: -p[0])
    out = [m for s, m in scored if s >= self.candidate_threshold]
    seen = {m.memory_id for m in out}
    for m in existing:                      # topic sweep
        if topic and m.topic == topic and m.memory_id not in seen:
            out.append(m)
    return out

This is also the argument for putting a topic field on every memory at extraction time. It looks like metadata you will never use, and then it turns out to be the only thing that makes consolidation work.

Now the decision function. In production this is an LLM call — show it the candidate and the existing memories, get back operations. Offline, the same policy in code:

@dataclass
class Operation:
    action: Literal["CREATE", "UPDATE", "NOOP"]
    target_id: Optional[str]
    content: str
    reason: str

# Topics where a user can only hold one value at a time. Two memories on an
# exclusive topic are a contradiction, not two facts.
EXCLUSIVE_TOPICS = {"diet", "travel", "identity"}

def offline_decide(cand, similar, source) -> Operation:
    for m in similar:
        if m.content.strip().lower() == cand.content.strip().lower():
            return Operation("NOOP", m.memory_id, cand.content,
                             "identical to an existing memory")
        if m.topic == cand.topic and m.topic in EXCLUSIVE_TOPICS:
            if TRUST[source.source_type] >= min(TRUST[s.source_type]
                                                for s in m.sources):
                return Operation("UPDATE", m.memory_id, cand.content,
                    f"contradicts {m.memory_id} on exclusive topic "
                    f"'{m.topic}'; newer information from a "
                    f"{source.source_type} source wins")
            return Operation("NOOP", m.memory_id, cand.content,
                             "lower-trust source than the existing memory")
    return Operation("CREATE", None, cand.content, "novel topic for this user")

Two policies are encoded there, and both are choices you should make consciously.

Exclusive topics. A user has one diet and one seat preference at a time. Two memories on an exclusive topic are a contradiction to resolve, not two facts to keep. Topics like issue are not exclusive — a user can have many open problems.

Trust gating. New information only wins if its source is at least as trustworthy as the weakest source behind the existing memory. A tool result does not get to overwrite something the user stated. This is the provenance hierarchy from Chapter 5, in four lines.

And the reason string gets persisted into history, which is what makes the store debuggable rather than mysterious.

Applying an operation is where lineage is recorded:

def _apply(self, op, cand, src, user_id, app_name):
    if op.action == "NOOP":
        if op.target_id:                     # corroboration raises confidence
            old = self.store.get(op.target_id)
            old.sources.append(src)
            old.confidence = min(1.0, old.confidence + 0.05)
            old.history.append(f"{_ts()} corroborated by {src.source_type}")
            self.store.put(old)
        return

    new = MemoryRecord(
        content=op.content, user_id=user_id, app_name=app_name,
        topic=cand.topic, importance=cand.importance,
        half_life_days=cand.half_life_days,
        confidence=min(cand.confidence, TRUST[src.source_type]),
        sources=[src])

    if op.action == "UPDATE" and op.target_id:
        old = self.store.get(op.target_id)
        old.invalidated_at = time.time()
        old.superseded_by = new.memory_id
        old.history.append(f"{_ts()} superseded by {new.memory_id}: {op.reason}")
        self.store.put(old)
        # Lineage: the new memory inherits the old one's sources.
        new.sources = old.sources + [src]
        new.history.append(f"{_ts()} created from {old.memory_id}: {op.reason}")
    self.store.put(new)

The inherited sources on the last-but-one line are the important bit. The replacement memory knows about every conversation that contributed to its predecessor. Without that, the erasure feature in Part 8 cannot work — you would revoke a source and the memory derived from it would look innocent.

Also note: confidence=min(cand.confidence, TRUST[src.source_type]). A memory can never be more confident than its source is trustworthy. The extractor’s optimism is capped by provenance.

Part 6: retrieval and injection

def retrieve(self, query, *, user_id, app_name="default",
             scopes=("user", "app"), k=5, w=(0.6, 0.25, 0.15)):
    now = time.time()
    qe = hash_embed(query)
    wr, wrec, wimp = w
    scored = []
    for m in self.store.scan(user_id, app_name, scopes):
        relevance = cosine(qe, m.embedding)
        recency = 0.5 ** (m.age_days(now) / m.half_life_days)
        scored.append((wr * relevance + wrec * recency + wimp * m.importance, m))
    scored.sort(key=lambda p: -p[0])
    return scored[:k]

def inject(self, system: str, retrieved) -> str:
    lines = []
    for score, m in retrieved:
        conf = m.decayed_confidence()
        band = "high" if conf >= 0.75 else "medium" if conf >= 0.5 else "low"
        src = m.sources[-1].source_type if m.sources else "unknown"
        lines.append(f"- [confidence: {band}, source: {src}] {m.content}")
    return (f"{system}\n\n<MEMORIES>\n"
            "Information you know about this user from previous "
            "conversations. Treat low-confidence items as weak priors.\n"
            + "\n".join(lines) + "\n</MEMORIES>")

retrieve calls scan, which filters by owner, so isolation is enforced before a single distance is computed. That ordering is deliberate — you cannot accidentally rank a memory you were not allowed to see.

inject bands the confidence rather than printing a float. Models handle “high / medium / low” more consistently than 0.6842, and the extra precision was never real. Note it uses decayed_confidence(), not raw confidence: an old memory presents as less certain automatically.

Part 7: run it

SESSION 1  (source: user_implicit, session s1)
    CREATE               'The user follows a vegetarian diet.'
           reason: novel topic for this user
    CREATE               'The user goes by Ada.'
           reason: novel topic for this user
    CREATE               'The user prefers aisle seating on flights.'
           reason: novel topic for this user

SESSION 2  (the same fact, said differently)
    NOOP   -> mem_8326ef86 'The user follows a vegetarian diet.'
           reason: identical to an existing memory

Session 2 produced no new memory, and quietly raised the confidence on the existing one from 0.65 to 0.70. That is deduplication and corroboration in one step.

Now the contradiction:

SESSION 3  (six months later -- a genuine contradiction)
    UPDATE -> mem_8326ef86 'The user eats fish again.'
           reason: contradicts mem_8326ef86 on exclusive topic 'diet'; newer
                   information from a user_explicit source wins

The store afterwards:

STORE STATE -- ACTIVE:
  mem_7152aaca  [diet    ] conf=0.90  The user eats fish again.
      sources: [('user_implicit', 's1'), ('user_implicit', 's2'), ('user_explicit', 's3')]
  mem_752ac84c  [identity] conf=0.65  The user goes by Ada.
      sources: [('user_implicit', 's1')]
  mem_471c90a5  [travel  ] conf=0.65  The user prefers aisle seating on flights.
      sources: [('user_implicit', 's1')]

INVALIDATED (kept for lineage, never retrieved):
  mem_8326ef86  The user follows a vegetarian diet.
      2026-08-06 20:11:27 corroborated by user_implicit
      2026-08-06 20:11:27 superseded by mem_7152aaca: contradicts mem_8326ef86 on
          exclusive topic 'diet'; newer information from a user_explicit source wins

Read that carefully, because it is the whole point of the chapter.

There is exactly one active diet memory, and it is the current one. The superseded memory is retained, tombstoned, with a written explanation of why it was replaced and a pointer to its replacement. The new memory inherited all three sources, so it knows it descends from sessions s1 and s2 as well as s3. And its confidence is 0.90 rather than 0.65, because the user stated it explicitly this time.

A memory system without consolidation would have both "follows a vegetarian diet" and "eats fish again" active, retrieval would return both, and the agent would recommend the tuna about half the time.

Retrieval:

query: 'any dietary restrictions I should know about?'
  0.473  The user eats fish again.
  0.388  The user prefers aisle seating on flights.
  0.370  The user goes by Ada.

query: 'what seat should I book on the flight?'
  0.444  The user prefers aisle seating on flights.
  0.392  The user eats fish again.
  0.370  The user goes by Ada.

query: 'what should we order for lunch?'
  0.370  The user goes by Ada.
  0.370  The user eats fish again.
  0.349  The user prefers aisle seating on flights.

The first two rank correctly. The third does not, and that is the honest limit of a hash embedder: nothing in it knows that “lunch” is about food. A real embedding model puts the diet memory first on that query without any other change to the code. This is a useful thing to have seen — when your retrieval is bad, the embedder is one of the two suspects, and generation quality is the other.

Isolation, and the assembled context:

isolation check -- a different user sees nothing:
  []

CONTEXT FOR THE NEXT TURN
You are a travel concierge for Solaris Trips.

<MEMORIES>
Information you know about this user from previous conversations. Treat
low-confidence items as weak priors.
- [confidence: medium, source: user_implicit] The user prefers aisle seating on flights.
- [confidence: high, source: user_explicit] The user eats fish again.
- [confidence: medium, source: user_implicit] The user goes by Ada.
</MEMORIES>

That block is what actually goes to the model. Three memories, annotated with how much to trust each one, in a delimited section that will not be confused for dialogue.

Part 8: forgetting a source

A user revokes access to something. The naive implementation deletes every memory that source ever touched, which destroys memories that had three other perfectly valid sources.

def forget_source(self, source_id, user_id, app_name="default") -> list[str]:
    """Right-to-erasure: drop memories that depend ONLY on this source,
    and flag for regeneration those that merely touched it."""
    removed, regenerate = [], []
    for m in self.store.scan(user_id, app_name, include_invalid=True):
        ids = {s.source_id for s in m.sources}
        if source_id not in ids:
            continue
        if ids == {source_id}:
            m.invalidated_at = time.time()
            m.history.append(f"{_ts()} erased: sole source {source_id} revoked")
            self.store.put(m)
            removed.append(m.memory_id)
        else:
            regenerate.append(m.memory_id)
    return removed + [f"{r} (needs regeneration)" for r in regenerate]

Running it:

RIGHT TO ERASURE: revoking source s1
   mem_752ac84c
   mem_471c90a5
   mem_8326ef86 (needs regeneration)
   mem_7152aaca (needs regeneration)

remaining active:
   mem_7152aaca  The user eats fish again.

Two memories derived solely from session s1 were erased outright. Two memories that blended s1 with other sources were flagged for regeneration rather than deleted, because deleting them would throw away information from s2 and s3 that the user never revoked.

Regeneration — re-running extraction and consolidation over only the remaining valid sources — is the expensive, correct completion of this. It is left as the exercise, and it is a good one, because it forces you to keep enough source material around to do it.

This whole feature is only possible because every memory carries a source list. That is the case for provenance, made concrete: it is not bookkeeping, it is the thing that makes a legal obligation implementable.

Part 9: the same system on a real stack

You would not ship the hand-rolled version. Here are both realistic paths.

ChromaDB — you keep the pipeline, it owns the storage

Chroma gives you the vector index and metadata filtering; extraction and consolidation stay yours. This runs offline too, using our hash embedder as the embedding function:

import chromadb
from chromadb.utils import embedding_functions

class HashEF(embedding_functions.EmbeddingFunction):
    """Offline embedding function so this runs without an API key.
    In production: embedding_functions.OpenAIEmbeddingFunction(...) or
    GoogleGenerativeAiEmbeddingFunction(...)."""
    def __init__(self): pass
    def __call__(self, input): return [hash_embed(t) for t in input]
    def name(self): return "hash_embed"

client = chromadb.EphemeralClient()      # PersistentClient(path=...) to persist
col = client.get_or_create_collection("memories", embedding_function=HashEF())

col.upsert(
    ids=["mem_1", "mem_2", "mem_3"],
    documents=["The user follows a vegetarian diet.",
               "The user prefers aisle seating on flights.",
               "The user goes by Ada."],
    metadatas=[{"user_id": "u_ada", "scope": "user", "topic": "diet",
                "source_type": "user_implicit", "source_id": "s1",
                "confidence": 0.65, "active": True},
               # ... one metadata dict per memory
               ],
)

# Consolidation: the contradiction arrives. Invalidate, do not delete.
col.update(ids=["mem_1"], metadatas=[{..., "active": False,
                                      "superseded_by": "mem_4"}])
col.upsert(ids=["mem_4"], documents=["The user eats fish again."],
           metadatas=[{"user_id": "u_ada", "scope": "user", "topic": "diet",
                       "source_type": "user_explicit", "source_id": "s3",
                       "confidence": 0.90, "active": True}])

# Scoped retrieval: isolation is a metadata filter, enforced by the database.
res = col.query(
    query_texts=["any dietary restrictions I should know about?"],
    n_results=3,
    where={"$and": [{"user_id": "u_ada"}, {"active": True}]},
)

Real output, on chromadb 1.5.9:

  dist=1.655  conf=0.9   The user eats fish again.
  dist=1.789  conf=0.65  The user prefers aisle seating on flights.
  dist=2.000  conf=0.65  The user goes by Ada.

  candidate sweep for consolidation (same topic, any similarity):
    active=False  The user follows a vegetarian diet.
    active=True   The user eats fish again.

Three notes for the transition.

Chroma returns distances, not similarities — lower is better, and the direction flip is an easy bug to write. Your provenance fields become metadata, and Chroma’s where filter is what enforces both isolation and the active/invalidated split. And the topic sweep is a col.get(where={"topic": "diet"}) — the same fix from Part 5, expressed as a metadata query instead of a scan.

What Chroma does not give you is extraction or consolidation. Those are still yours. Chroma is the storage layer, and everything interesting in this chapter lives above it.

mem0 — it owns the whole pipeline

mem0 is a memory manager in the Chapter 4 sense: it does extraction and consolidation for you, using an LLM you configure.

from mem0 import Memory

m = Memory.from_config({
    "vector_store": {"provider": "qdrant", "config": {"path": "./qdrant"}},
    "llm":      {"provider": "openai", "config": {"model": "gpt-4.1-mini"}},
    "embedder": {"provider": "openai", "config": {"model": "text-embedding-3-small"}},
})

m.add([{"role": "user", "content": "I'm a vegetarian, keep that in mind."}],
      user_id="u_ada")

# Six months later. mem0 runs its own consolidation: it retrieves similar
# memories, asks an LLM what to do, and applies ADD / UPDATE / DELETE.
m.add([{"role": "user", "content": "I've started eating fish again."}],
      user_id="u_ada")

print(m.search("dietary restrictions", user_id="u_ada", limit=5))
print(m.get_all(user_id="u_ada"))

That is the whole thing. add runs extraction and consolidation; search runs retrieval; user_id is the scope. mem0 requires at least one of user_id, agent_id, or run_id on every operation — the isolation model is not optional, which is the right default. It also keeps a history database so you can inspect what changed and when, which is its version of the lineage you built in Part 5.

Which should you use?

Use mem0 (or Vertex AI Memory Bank, or Zep) if memory is a feature of your product rather than the product. Consolidation is genuinely hard, they have already tuned the prompts, and you have better things to build.

Use Chroma plus your own pipeline when your definition of “meaningful” is unusual enough that a general-purpose extractor will get it wrong, when you need provenance semantics the managed service does not model, or when the data cannot leave your infrastructure.

Either way, having built the hand-rolled version, you now know what the service is doing — which is what you need the first time it does something you did not expect.

Exercises

Swap the embedder. Replace hash_embed with a real embedding API. Re-run the retrieval demo and watch the “lunch” query start ranking the diet memory first.
Swap the extractor. Replace rule_extract with a structured-output LLM call using EXTRACTION_PROMPT and a Pydantic schema. Notice how much more it finds, and how much more noise you now have to consolidate.
Implement regeneration. Complete forget_source so that flagged memories are rebuilt from remaining valid sources rather than left in place.
Add decay-based pruning. A background pass that invalidates memories whose decayed_confidence() has fallen below a floor. Watch what disappears after a simulated year.
Add session-scoped memories. Generate a session-scope memory at the end of a conversation and use it in place of the transcript on the next turn — Chapter 3’s compaction and this chapter’s memory, joined up.
Break it. Write a conversation designed to poison the store: a user asserting facts about another user, or trying to get something written at application scope. Then fix what it broke.

What you should be able to do now

Build a memory pipeline end to end — extraction, storage with embeddings, consolidation, scoped retrieval, and context injection — and run it offline.
Explain why similarity search alone cannot detect contradictions, and implement the topic sweep that fixes it.
Record provenance on every memory such that a contradiction resolution is auditable and a source revocation is implementable.
Enforce scope isolation in the query layer rather than the prompt layer, and demonstrate that another user retrieves nothing.
Annotate injected memories with decayed confidence and source type, and explain how that changes model behavior.
Port the same system to ChromaDB, and say precisely which parts a managed memory manager like mem0 would take over and which parts would still be yours.

Production Agent Engineering — A Practical Guide