Published 28 October 2026. Every code sample below except the Maximem Synap example, which needs an account, was run on 28 October 2026 on Python 3.13 with fastembed 0.9.1 and SQLite 3.53; the nine tests passed in 5.37 seconds with no API key.
You add episodic and semantic memory to an AI agent with two tables and two code paths. An episodes table records what happened in each conversation, when it happened and the raw turns as evidence, while a facts table holds what is currently true about the user, where a new value closes the old one instead of sitting beside it; after each session a write path extracts one episode and any durable facts, and before each model call a read path puts the current facts and the few most relevant past episodes into the prompt within a token budget. The full build below is about 180 lines of Python on SQLite with a local open-source embedding model (fastembed running BAAI/bge-small-en-v1.5), and the one LLM call it needs, extraction, is a function you pass in, so the tests run with a deterministic stand-in and cost nothing.
Two concept pieces sit behind this build. How to give an LLM long-term memory covers why memory sits outside the model and the five design decisions behind it, and short-term vs long-term memory in LLMs covers the vocabulary.
Episodic vs semantic memory in an AI agent
Episodic memory is the agent's record of specific things that happened, each tied to a time and a context: "on 25 July the user reported that order 1182 arrived damaged and asked for a refund." Semantic memory is what the agent believes is true now, with no story attached: "the user is vegetarian" or "the user lives in Berlin." The split comes from cognitive science, and the CoALA framework for language agents carries it over directly: "Semantic memory stores facts about the world ..., while episodic memory stores sequences of the agent's past behaviors."
| Episodic memory | Semantic memory | |
|---|---|---|
| Holds | Events: what happened, when, with what outcome | Facts: what is true about the user or the world now |
| Example | "25 Jul: reported order 1182 damaged, asked for a refund" | "Lives in Berlin (since 3 Sep)" |
| Answers | "Did my refund go through?", "What did we try last time?" | "Where should this ship?", "What can she eat?" |
| Written | Once per conversation or task, then left alone | Created, confirmed, superseded as facts change |
| Retrieved by | Relevance to the question, recency, importance | Mostly all current facts, ranked when there are many |
| Typical failure | Returning a similar but wrong past event | Returning a value that stopped being true |
The LangChain memory documentation uses the same three types: semantic memory for facts about a user, episodic memory for past agent actions, and procedural memory for instructions, which in practice is the system prompt. Working memory is the context window for the current call. This build covers the two long-term types, because those are the ones that have to survive the session.
Each type feeds the other. A fact almost always arrives inside an episode ("I moved to Berlin last month, please update my address" is an event that carries a fact), so the write path reads the episode and lifts the fact out, which is the small, practical version of what the memory literature calls consolidation. The episode stays as the evidence for why the agent believes the fact.
What your agent framework already does, and what it does not
Agent frameworks already persist one conversation, and it is worth being exact about how far that goes. The OpenAI Agents SDK ships a SQLiteSession, and its documentation says the runner "automatically retrieves the conversation history for the session and prepends it to the input items", so a conversation survives a process restart without any code from you. LangGraph does the equivalent with a checkpointer scoped to a thread, and adds a store with namespaces for data that should outlive a thread. For resuming one conversation, those tools are the right answer and you should use them.
What a session does not do is decide anything. It replays the transcript of one conversation; it does not notice that the user said something new about themselves, does not retire the Chicago address when Berlin arrives, and does not find the refund conversation from six weeks ago when a different session starts. LangChain's memory documentation says of a collection of memories that "the model must now delete or update existing items in the list, which can be tricky." That update logic, plus the ranking that decides what comes back, is the part you write, and it is what the code below is for.
The build runs on SQLite and a local embedding model
The build is four files: memory.py (schema, write path, read path), extractors.py (the LLM extractor and a rule-based stand-in), agent.py (a minimal agent loop) and test_memory.py. Everything is open source: fastembed is Apache-2.0, the bge-small model is MIT-licensed and produces 384-dimension vectors on the CPU through ONNX Runtime, and SQLite ships with Python.
pip install fastembed pytest # fastembed 0.9.1 needs Python 3.10 or later
python -m pytest -v test_memory.py
python agent.py demo.db
On the first run fastembed downloads the embedding model from Hugging Face; after that, everything runs offline.
The store schema: episodes and facts
memory.py appears in three parts, this schema block, the write path and the read path, and the three together are the whole file.
"""Episodic and semantic memory for an AI agent: SQLite plus a local embedding model."""
import json, math, re, sqlite3, time, uuid
import numpy as np
from fastembed import TextEmbedding
SCHEMA = """
CREATE TABLE IF NOT EXISTS episodes ( -- episodic: what happened, when
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
occurred_at REAL NOT NULL, -- unix time of the event
summary TEXT NOT NULL, -- one self-contained sentence
outcome TEXT,
importance REAL NOT NULL DEFAULT 0.5, -- 0..1, set by the extractor
transcript TEXT NOT NULL, -- raw turns as JSON (the evidence)
embedding BLOB NOT NULL
);
CREATE INDEX IF NOT EXISTS ep_user_time ON episodes(user_id, occurred_at);
CREATE TABLE IF NOT EXISTS facts ( -- semantic: what is true now
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
attribute TEXT NOT NULL, -- home_city, diet, allergy ...
value TEXT NOT NULL,
statement TEXT NOT NULL, -- "Lives in Berlin"
multi INTEGER NOT NULL DEFAULT 0, -- 1 = attribute holds many values
valid_from REAL NOT NULL,
valid_to REAL, -- NULL means current
superseded_by TEXT REFERENCES facts(id),
source_episode TEXT REFERENCES episodes(id),
confirmations INTEGER NOT NULL DEFAULT 1,
embedding BLOB NOT NULL
);
CREATE INDEX IF NOT EXISTS fact_current ON facts(user_id, attribute) WHERE valid_to IS NULL;
"""
occurred_at is the time of the event, not the time of the insert, which is what lets the read path say "25 July" and rank by age. transcript keeps the raw turns next to the summary: the summary is what gets embedded and retrieved, and the turns are the evidence when the summary is wrong or too short, which it sometimes will be.
On the facts side, valid_from, valid_to and superseded_by turn an overwrite into a history. The current view is every row where valid_to IS NULL, and the partial index keeps that lookup cheap. multi marks attributes that legitimately hold several values at once (allergies, spoken languages), so a second allergy is added rather than treated as a correction. confirmations counts restatements, which gives you a cheap confidence signal and stops a second "I am vegetarian" from creating a duplicate row. source_episode links every fact back to the conversation it came from.
Embeddings are stored as raw float32 bytes and compared in Python, which is fine for the few hundred rows a single user accumulates. Once you need nearest-neighbour search across many users' rows in one query, move the vectors into sqlite-vec or Postgres with pgvector; our comparison of what a vector database does and does not give agent memory covers that step.
The write path extracts once per session, then supersedes
Once per session, the write path hands the turns and the user's current facts to an extractor, inserts one episode, and then upserts each extracted fact:
class Embedder:
def __init__(self, model="BAAI/bge-small-en-v1.5"):
self.model = TextEmbedding(model)
def __call__(self, texts):
vecs = np.array(list(self.model.embed(texts)), dtype=np.float32)
return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)
class MemoryStore:
def __init__(self, path, extractor, embedder):
self.db = sqlite3.connect(path)
self.db.row_factory = sqlite3.Row
self.db.executescript(SCHEMA)
self.extract, self.embed = extractor, embedder
# ---------- write path ----------
def remember_session(self, user_id, session_id, turns, occurred_at=None):
now = occurred_at or time.time()
current = [dict(r) for r in self.current_facts(user_id)]
out = self.extract(turns, current) # the LLM call plugs in here
ep = out["episode"]
ep_id = uuid.uuid4().hex
with self.db:
self.db.execute(
"INSERT INTO episodes VALUES (?,?,?,?,?,?,?,?,?)",
(ep_id, user_id, session_id, now, ep["summary"], ep.get("outcome"),
ep.get("importance", 0.5), json.dumps(turns),
self.embed([ep["summary"]])[0].tobytes()))
ops = [self._upsert_fact(user_id, f, now, ep_id) for f in out["facts"]]
return ops
def _upsert_fact(self, user_id, f, now, ep_id):
rows = self.db.execute(
"SELECT id, value FROM facts WHERE user_id=? AND attribute=? AND valid_to IS NULL",
(user_id, f["attribute"])).fetchall()
same = [r for r in rows if r["value"].lower() == f["value"].lower()]
if same: # restated: reinforce, do not duplicate
self.db.execute("UPDATE facts SET confirmations = confirmations + 1 WHERE id=?",
(same[0]["id"],))
return ("confirmed", f["attribute"], f["value"])
new_id = uuid.uuid4().hex
self.db.execute(
"INSERT INTO facts (id,user_id,attribute,value,statement,multi,valid_from,"
"source_episode,embedding) VALUES (?,?,?,?,?,?,?,?,?)",
(new_id, user_id, f["attribute"], f["value"], f["statement"],
int(f.get("multi", False)), now, ep_id,
self.embed([f["statement"]])[0].tobytes()))
if rows and not f.get("multi"): # single-valued: close the old version
self.db.execute(
"UPDATE facts SET valid_to=?, superseded_by=? WHERE user_id=? AND attribute=? "
"AND valid_to IS NULL AND id<>?", (now, new_id, user_id, f["attribute"], new_id))
return ("superseded", f["attribute"], f["value"])
return ("added", f["attribute"], f["value"])
_upsert_fact makes one of three decisions. A value the user already has is a restatement, so the existing row's confirmations goes up and nothing new is written. A new value for a single-valued attribute closes the old row (valid_to set, superseded_by pointing at the replacement) inside the same transaction as the insert, so there is never a moment with two current addresses. A new value for a multi-valued attribute is simply added. Everything is keyed by user_id, which is the whole of the isolation story in a build this size.
Extraction is where the LLM call plugs in, and it is deliberately the only place. Its contract is a function that takes the turns and the current facts and returns one episode plus a list of facts:
"""Two extractors with one signature: (turns, current_facts) -> {"episode": {...}, "facts": [...]}."""
import json, re
EXTRACTION_PROMPT = """You maintain an AI agent's memory of one user.
Read the conversation and return JSON only, in this shape:
{{"episode": {{"summary": "<one self-contained sentence: what happened in this conversation>",
"outcome": "<resolved | open | failed | none>",
"importance": <0.0 to 1.0>}},
"facts": [{{"attribute": "<snake_case key>", "value": "<value>",
"statement": "<short sentence about the user>", "multi": <true|false>}}]}}
Rules: facts are durable things about the user (where they live, diet, allergies, plan,
preferences), not things that happened. Reuse an existing attribute name when the fact is
about the same thing, so a change of value replaces the old one. Set multi=true only when
the user can hold several values at once (allergies, languages). Do not invent anything.
Facts already stored: {current}
Conversation:
{transcript}"""
class LLMExtractor:
"""`complete` is any function prompt -> text: OpenAI, Anthropic, a local Ollama model."""
def __init__(self, complete):
self.complete = complete
def __call__(self, turns, current_facts):
current = [{"attribute": f["attribute"], "value": f["value"]} for f in current_facts]
transcript = "\n".join(f"{t['role']}: {t['content']}" for t in turns)
raw = self.complete(EXTRACTION_PROMPT.format(current=json.dumps(current),
transcript=transcript))
data = json.loads(raw[raw.find("{"): raw.rfind("}") + 1])
data["facts"] = [f for f in data.get("facts", [])
if f.get("attribute") and f.get("value") and f.get("statement")]
return data
class RuleExtractor:
"""Deterministic stand-in for tests and demos. Pattern-matches a few phrasings."""
PATTERNS = [
(r"\bI (?:live in|moved to|am based in) ([A-Z][a-z]+)", "home_city", "Lives in {}", False),
(r"\bI(?: am|'m) (vegetarian|vegan)\b", "diet", "Is {}", False),
(r"\bI(?: am|'m) allergic to (\w+)", "allergy", "Allergic to {}", True),
(r"\bI(?: am|'m) on the (\w+) plan", "plan", "On the {} plan", False),
]
def __call__(self, turns, current_facts):
user_text = " ".join(t["content"] for t in turns if t["role"] == "user")
facts = []
for pattern, attr, template, multi in self.PATTERNS:
for m in re.finditer(pattern, user_text):
facts.append({"attribute": attr, "value": m.group(1),
"statement": template.format(m.group(1)), "multi": multi})
return {"episode": {"summary": "User said: " + user_text[:160],
"outcome": None, "importance": 0.5},
"facts": facts}
LLMExtractor takes any complete(prompt) -> text function, so the same code runs against a hosted model or a local one. Passing the current facts into the prompt matters more than it looks: when the model can see that home_city already exists, it reuses that attribute name for "I moved to Berlin", and the supersede logic fires. Without that, you get home_city: Chicago and current_city: Berlin both marked current, which is the duplicate problem by another name.
RuleExtractor is a stand-in that pattern-matches four phrasings. It is not a memory system and is not meant to be one; it exists so the storage and ranking logic can be tested deterministically, for free, on every commit. Its episode summary is just the user's words, which an LLM would compress into one clean sentence.
Run the write path after the session, not inside the turn. LangChain's documentation lists the trade: writing in the background "eliminates latency in the primary application", at the price that a new memory is not available until the background job finishes. In agent.py below the call happens once the conversation ends; in production it belongs in a queue worker.
The read path puts current facts and ranked episodes into the prompt
# ---------- read path ----------
def current_facts(self, user_id):
return self.db.execute(
"SELECT * FROM facts WHERE user_id=? AND valid_to IS NULL ORDER BY valid_from",
(user_id,)).fetchall()
def history(self, user_id, attribute):
return self.db.execute(
"SELECT value, valid_from, valid_to FROM facts WHERE user_id=? AND attribute=? "
"ORDER BY valid_from", (user_id, attribute)).fetchall()
def build_context(self, user_id, query, max_facts=8, max_episodes=3,
half_life_days=30, budget_tokens=300, now=None):
now = now or time.time()
q = self.embed([query])[0]
facts = self.current_facts(user_id)
facts = sorted(facts, key=lambda r: -float(q @ np.frombuffer(r["embedding"], np.float32)))
eps = []
for r in self.db.execute("SELECT * FROM episodes WHERE user_id=?", (user_id,)):
sim = float(q @ np.frombuffer(r["embedding"], np.float32))
age_days = (now - r["occurred_at"]) / 86400
recency = math.exp(-math.log(2) * age_days / half_life_days)
eps.append((0.6 * sim + 0.25 * recency + 0.15 * r["importance"], r))
eps = [r for _, r in sorted(eps, key=lambda x: -x[0])[:max_episodes]]
day = lambda t: time.strftime("%Y-%m-%d", time.gmtime(t))
lines = ["What we know about this user now:"]
lines += [f"- {r['statement']} (since {day(r['valid_from'])})" for r in facts[:max_facts]]
lines += ["Relevant past conversations:"]
lines += [f"- {day(r['occurred_at'])}: {r['summary']}"
+ (f" Outcome: {r['outcome']}." if r["outcome"] else "") for r in eps]
while len(" ".join(lines)) / 4 > budget_tokens and len(lines) > 2:
lines.pop() # rough budget: ~4 characters per token
return "\n".join(lines)
def forget_user(self, user_id): # erasure request: every row, every table
with self.db:
self.db.execute("DELETE FROM facts WHERE user_id=?", (user_id,))
self.db.execute("DELETE FROM episodes WHERE user_id=?", (user_id,))
Current facts go in first, because they are short and usually all relevant; with more than eight, they are ranked by similarity to the message. Episodes are scored on three signals, an approach that comes from the Generative Agents paper by Park et al., which scores every memory on recency, importance and relevance, rates importance by asking the model for a score from 1 to 10 when the memory is created, decays recency exponentially with a factor of 0.995 per sandbox game hour since the memory was last retrieved, min-max normalises all three and weights them equally. This build uses fixed weights instead (0.6 relevance, 0.25 recency with a 30-day half-life, 0.15 importance), and the result is a short, dated block:
What we know about this user now:
- Is vegetarian (since 2026-07-25)
- Lives in Berlin (since 2026-09-03)
Relevant past conversations:
- 2026-07-25: User said: I live in Chicago and I am vegetarian. My order 1182 arrived damaged and I want a refund.
- 2026-09-03: User said: I moved to Berlin last month, please update my address.
- 2026-07-28: User said: Can you suggest a lunch place near the office?
That block is the prompt contribution, and the dates are deliberate: a model that sees "since 2026-09-03" can reason about whether a fact is fresh, and one that sees "2026-07-25" on an episode knows it is talking about something six weeks old. The token budget is a rough four characters per token and trims from the bottom, which drops the lowest-ranked episode first and the facts last.
Running the demo produced the one result I would change first. The query was "Did my refund for the damaged order go through?" at day 45, and the refund conversation (similarity 0.763, recency 0.354) scored 0.621 while the Berlin-move conversation (similarity 0.528, recency 0.891) scored 0.614. The right episode won by 0.007. Shortening the half-life from 30 to 25 days, with nothing else changed, flips the order when you ask for one episode: the Berlin conversation comes back and the refund does not, and nothing in the output says a better match was skipped. Recency weights that look sensible on a whiteboard can quietly outvote relevance, so pin the behaviour with a test like test_episodic_recall_finds_the_right_past_conversation below before you tune anything.
The agent loop: read before the model, write after the session
"""A minimal agent loop: read memory before the model, write memory after the session."""
import sys
from memory import MemoryStore, Embedder
from extractors import RuleExtractor
def call_llm(system, user):
# Replace with your model call, e.g. client.messages.create(system=system, ...)
return "(model reply)"
def run_session(store, user_id, session_id, messages, at):
turns = []
for msg in messages:
context = store.build_context(user_id, msg, now=at) # read path
reply = call_llm(system="You are a support agent.\n\n" + context, user=msg)
turns += [{"role": "user", "content": msg}, {"role": "assistant", "content": reply}]
return store.remember_session(user_id, session_id, turns, occurred_at=at) # write path
if __name__ == "__main__":
DAY, T0 = 86400, 1_785_000_000
store = MemoryStore(sys.argv[1], RuleExtractor(), Embedder())
sessions = {
"s1": (["I live in Chicago and I am vegetarian.",
"My order 1182 arrived damaged and I want a refund."], T0),
"s2": (["Can you suggest a lunch place near the office?"], T0 + 3 * DAY),
"s3": (["I moved to Berlin last month, please update my address."], T0 + 40 * DAY),
}
for sid, (msgs, at) in sessions.items():
print(sid, run_session(store, "maya", sid, msgs, at))
print()
print(store.build_context("maya", "Did my refund for the damaged order go through?",
now=T0 + 45 * DAY))
print()
print([tuple(r) for r in store.history("maya", "home_city")])
Each message reads memory before the model call; each session writes memory once it ends. Running python agent.py demo.db printed:
s1 [('added', 'home_city', 'Chicago'), ('added', 'diet', 'vegetarian')]
s2 []
s3 [('superseded', 'home_city', 'Berlin')]
What we know about this user now:
- Is vegetarian (since 2026-07-25)
- Lives in Berlin (since 2026-09-03)
Relevant past conversations:
- 2026-07-25: User said: I live in Chicago and I am vegetarian. My order 1182 arrived damaged and I want a refund.
- 2026-09-03: User said: I moved to Berlin last month, please update my address.
- 2026-07-28: User said: Can you suggest a lunch place near the office?
[('Chicago', 1785000000.0, 1788456000.0), ('Berlin', 1788456000.0, None)]
Session one added two facts, the lunch question added none, and the move to Berlin superseded Chicago, with the history row showing Chicago valid until the moment Berlin arrived. The context block above is what session four would see before replying.
How to test agent memory you built yourself
import json, pytest
from memory import MemoryStore, Embedder
from extractors import RuleExtractor, LLMExtractor
DAY = 86400
T0 = 1_785_000_000 # a fixed clock, so tests do not depend on today's date
@pytest.fixture(scope="session")
def embedder():
return Embedder()
def store(tmp_path, embedder, extractor=None):
return MemoryStore(str(tmp_path / "mem.db"), extractor or RuleExtractor(), embedder)
def chat(*user_msgs, reply="ok"):
turns = []
for m in user_msgs:
turns += [{"role": "user", "content": m}, {"role": "assistant", "content": reply}]
return turns
def test_survives_a_new_process(tmp_path, embedder):
store(tmp_path, embedder).remember_session("u1", "s1", chat("I live in Chicago"), T0)
fresh = store(tmp_path, embedder) # new connection, same file
assert [r["statement"] for r in fresh.current_facts("u1")] == ["Lives in Chicago"]
def test_new_value_supersedes_old_and_keeps_history(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("u1", "s1", chat("I live in Chicago"), T0)
ops = m.remember_session("u1", "s2", chat("I moved to Berlin last week"), T0 + 30 * DAY)
assert ops == [("superseded", "home_city", "Berlin")]
assert [r["value"] for r in m.current_facts("u1")] == ["Berlin"]
hist = m.history("u1", "home_city")
assert [h["value"] for h in hist] == ["Chicago", "Berlin"]
assert hist[0]["valid_to"] == T0 + 30 * DAY and hist[1]["valid_to"] is None
def test_restated_fact_is_confirmed_not_duplicated(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("u1", "s1", chat("I am vegetarian"), T0)
assert m.remember_session("u1", "s2", chat("I'm vegetarian, remember"), T0 + DAY) == \
[("confirmed", "diet", "vegetarian")]
assert len(m.current_facts("u1")) == 1
def test_multi_valued_attribute_accumulates(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("u1", "s1", chat("I am allergic to peanuts"), T0)
m.remember_session("u1", "s2", chat("I am allergic to shellfish too"), T0 + DAY)
assert sorted(r["value"] for r in m.current_facts("u1")) == ["peanuts", "shellfish"]
def test_episodic_recall_finds_the_right_past_conversation(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("u1", "s1", chat("My order 1182 arrived damaged and I want a refund"),
T0)
m.remember_session("u1", "s2", chat("Can you recommend a vegetarian pasta recipe?"),
T0 + 5 * DAY)
m.remember_session("u1", "s3", chat("How do I change my password?"), T0 + 6 * DAY)
ctx = m.build_context("u1", "Did my refund go through?", max_episodes=1,
now=T0 + 7 * DAY)
assert "order 1182" in ctx and "password" not in ctx
def test_users_never_see_each_other(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("alice", "s1", chat("I live in Lisbon"), T0)
m.remember_session("bob", "s2", chat("I live in Oslo"), T0)
ctx = m.build_context("bob", "where do I live?", now=T0 + DAY)
assert "Oslo" in ctx and "Lisbon" not in ctx
def test_context_respects_token_budget(tmp_path, embedder):
m = store(tmp_path, embedder)
for i in range(20):
m.remember_session("u1", f"s{i}", chat(f"Question number {i} about invoices " * 5),
T0 + i * DAY)
ctx = m.build_context("u1", "invoices", max_episodes=20, budget_tokens=120,
now=T0 + 21 * DAY)
assert len(ctx) / 4 <= 120
def test_forget_user_removes_everything(tmp_path, embedder):
m = store(tmp_path, embedder)
m.remember_session("u1", "s1", chat("I live in Paris"), T0)
m.forget_user("u1")
assert m.current_facts("u1") == []
assert m.db.execute("SELECT COUNT(*) FROM episodes WHERE user_id='u1'").fetchone()[0] == 0
def test_llm_extractor_parses_model_output(tmp_path, embedder):
canned = json.dumps({"episode": {"summary": "User asked to switch to annual billing.",
"outcome": "open", "importance": 0.7},
"facts": [{"attribute": "plan", "value": "Pro",
"statement": "On the Pro plan", "multi": False}]})
seen = {}
def fake_complete(prompt): # swap for a real model call in production
seen["prompt"] = prompt
return "Here is the JSON:\n" + canned
m = store(tmp_path, embedder, LLMExtractor(fake_complete))
m.remember_session("u1", "s1", chat("I'm on the Pro plan, can I pay yearly?"), T0)
assert "I'm on the Pro plan" in seen["prompt"]
assert [r["statement"] for r in m.current_facts("u1")] == ["On the Pro plan"]
All nine passed on 28 October 2026 in 5.37 seconds. Each test protects one behaviour that quietly breaks in hand-built memory: surviving a new process, superseding with history, confirming instead of duplicating, accumulating multi-valued facts, recalling the right episode, keeping one user's memory out of another's context, staying inside the token budget, deleting a user completely, and parsing a model's extraction output when it arrives wrapped in chatter.
Test nine is the pattern for testing the real extractor without paying for it in CI: a fake complete function returns canned JSON, and the test asserts both that the prompt carried the user's words and that the parsed fact landed in the store. Extraction quality itself needs a separate, slower evaluation against a model, with scripted multi-session histories and a known answer key; how to test agent memory before you ship lays out that one-week plan.
Episodes about the agent's own work
Everything above stores episodes about the user. The same table works for episodes about the agent itself, which is how LangChain's documentation describes episodic memory in practice: past sequences "often implemented through few-shot example prompting", so the agent learns from what worked. AWS's AgentCore episodic memory structures each episode as the goal, reasoning steps, actions, outcome and reflections, and then reflects across similar episodes to produce general lessons. In this schema that means writing the task as the summary, the result in outcome, and adding AND outcome = 'resolved' to the episode query when you want worked examples rather than history.
What this build leaves out
The build leaves out six things, each of which shows up as soon as real users arrive. Nothing resolves entities, so "my manager" and "Sarah" stay two different subjects. A poorer statement can replace a richer one, because the upsert compares values, not detail: "takes a cholesterol medication" would supersede "takes 20mg of Atorvastatin daily." There is no cross-episode consolidation pass that turns three similar conversations into one lesson, and no eviction, so old low-importance episodes stay forever. Two workers writing the same user at once can both read the old fact and both insert a new one. And changing the embedding model means re-embedding every row, because vectors from two models are not comparable.
None of these is exotic, and the evaluation set that proves a fix worked is usually more work than the fix. If the agent serves one team and the facts rarely change, this build is enough and you should own it. Once you are adding the fourth item on that list, compare the cost of continuing with a memory layer; the long-term memory guide sets out which of these operations a layer takes over.
The same memory with Maximem Synap
With Maximem Synap the agent code comes down to two calls: send each message as it happens, and ask for context before replying, which returns a formatted block ready for the prompt. Writes return immediately and extraction runs in the background, taking several seconds, so a memory written now is not retrievable a second later. Scoping is applied from the identifiers you already pass, and you do not write a schema; the memory architecture is generated from a description of what your agent does, and you approve it.
import asyncio
from maximem_synap import MaximemSynapSDK # pip install maximem-synap (Python 3.11+)
sdk = MaximemSynapSDK(api_key="your-api-key")
async def main():
await sdk.initialize()
# Monday
await sdk.conversation.record_message(
conversation_id="mon-standup",
user_id="alice",
role="user",
content="I'm migrating our auth service to OAuth2 this sprint.",
)
# Friday. Different conversation, same person.
context = await sdk.fetch(
conversation_id="fri-review",
user_id="alice",
search_query=["what is alice working on?"],
)
print(context.formatted_context)
asyncio.run(main())
context.formatted_context plays the role of build_context above: a block ready for the prompt.
Memory that works is a write path that keeps the current truth current and a read path that hands the model a small, dated slice of the past. Get both right and the agent stops asking returning customers for things they already said, which is the change they notice first.
Frequently asked questions
What is the difference between episodic and semantic memory in an AI agent?
Episodic memory records specific events with a time and context, such as a refund request on 25 July and its outcome. Semantic memory holds what is currently true, such as the user's city or diet, with no story attached. Episodes are written once and kept as evidence; facts are created, confirmed and superseded as they change.
How do you implement episodic memory in an AI agent?
Store one row per conversation or task with the time it happened, a one-sentence summary, the outcome, an importance score, the raw turns and an embedding of the summary. Write it after the session ends, and at read time rank episodes by similarity to the current message, recency and importance, then put the top few into the prompt with their dates.
How do you persist agent memory across sessions?
Keep memory in a store outside the process, keyed by a stable user ID, and read it at the start of every new session. A session store such as the OpenAI Agents SDK's SQLiteSession replays one conversation; to carry knowledge between different conversations, extract facts and episodes into their own tables and retrieve them by user.
How do you add semantic memory to an AI agent?
Extract durable statements about the user from each conversation with an LLM, store each with an attribute name, a value and a validity window, and when a new value arrives for a single-valued attribute, close the old row instead of deleting it. Put the current facts into the prompt on every turn.
Do I need a vector database for agent memory?
You do not need one to start. A single user's facts and episodes number in the hundreds, and scoring a few hundred stored embeddings against a query in Python is one matrix product. Add sqlite-vec or Postgres with pgvector when you need nearest-neighbour search across many users' rows at once.
Can I build agent memory without an OpenAI or Anthropic API key?
You can for storage, embeddings and retrieval: fastembed with bge-small-en-v1.5 runs locally on a CPU. Extraction still needs a language model in production, but the model call can be any function, including a local model, and tests can replace it with a fake that returns canned JSON.
How do you stop an agent remembering outdated facts?
Give every fact a valid_from and valid_to, read only rows where valid_to is empty, and have the extractor reuse the existing attribute name when a fact changes, by showing it the facts already stored. The old value stays in the history with a pointer to its replacement.
Sources, retrieved and run 28 October 2026: Sumers et al., Cognitive Architectures for Language Agents (arXiv 2309.02427); Park et al., Generative Agents (arXiv 2304.03442); LangChain, memory concepts; OpenAI Agents SDK, sessions; AWS, AgentCore episodic memory; fastembed; BAAI/bge-small-en-v1.5; sqlite-vec; pgvector; Maximem Synap documentation.
