Rolling your own memory store
gbrain and managed memory products are worth knowing, but building one small memory store yourself is what makes every managed product's marketing page legible afterward. This is a minimal, real design: Postgres plus pgvector, one write path, one retrieval path, one forgetting policy.
Schema: separate facts from events
-- Semantic memory: distilled, durable facts about an entity
create table semantic_memory (
id uuid primary key default gen_random_uuid(),
subject text not null, -- e.g. 'user:42' or 'client:northwind'
fact text not null, -- 'prefers email over calls'
embedding vector(1536),
confidence float default 1.0,
source_event_id uuid, -- provenance: which episode wrote this
created_at timestamptz default now(),
updated_at timestamptz default now()
);
-- Episodic memory: what happened, when, raw
create table episodic_memory (
id uuid primary key default gen_random_uuid(),
subject text not null,
summary text not null, -- compressed, not the full raw transcript
embedding vector(1536),
occurred_at timestamptz not null,
metadata jsonb default '{}'
);
create index on semantic_memory using ivfflat (embedding vector_cosine_ops);
create index on episodic_memory using ivfflat (embedding vector_cosine_ops);The write path: a gate, not a firehose
Not every turn deserves to become a permanent memory. A write gate decides what is worth keeping, typically an LLM call asked "does this turn contain a durable fact, preference, or decision worth remembering - yes or no, and if yes, in one sentence, what." This keeps the store small, relevant, and cheap to search, instead of a raw log of every message ever sent.
async def maybe_write_memory(subject: str, turn: str) -> None:
gate = await llm_judge(
prompt=f"Does this contain a durable fact/preference/decision worth "
f"remembering long-term? Reply JSON: "
f'{{"worth_keeping": bool, "fact": str | null}}\n\nTurn: {turn}'
)
if not gate["worth_keeping"]:
return
embedding = embed(gate["fact"])
await db.execute(
"insert into semantic_memory (subject, fact, embedding) values ($1, $2, $3)",
subject, gate["fact"], embedding,
)The retrieval path: hybrid, not similarity-only
Pure cosine similarity retrieves what is *similar*, not what is *relevant right now*. Combine similarity with recency and explicit subject filtering. The same lesson RAG teaches for documents applies to memory: retrieval quality is the ceiling on what the agent can act on.
async def recall(subject: str, query: str, k: int = 5) -> list[dict]:
query_embedding = embed(query)
rows = await db.fetch(
"""
select fact, updated_at,
1 - (embedding <=> $2) as similarity
from semantic_memory
where subject = $1
order by (1 - (embedding <=> $2)) * 0.7
+ (extract(epoch from now() - updated_at) / -86400.0) * 0.3 desc
limit $3
""",
subject, query_embedding, k,
)
return [dict(r) for r in rows]Forgetting: a policy, not an accident
- Decay - lower a fact's effective weight over time unless it is re-confirmed by a later write.
- Explicit contradiction - a new fact that conflicts with an old one should supersede it, not sit alongside it silently (track `source_event_id` so you can explain why a fact changed).
- Deletion on request - if a user asks you to forget something, the store needs a real delete path, not just "stop retrieving it."
Critical
Memory is an attack surface
A poisoned write persists across every future session until something overwrites or deletes it. Apply the same untrusted-content discipline from Week 3's security lesson to anything that reaches the write gate.
Watch out
Common mistakes
- Writing every raw turn to memory and turning retrieval into a search-engine problem you never solve.
- Retrieving by similarity alone, so week-old irrelevant chatter outranks yesterday's decision.
- No contradiction handling, so the store accumulates stale facts nobody resolves.
- No deletion path, so "forget this" is a lie the system tells the user.