# How Do You Add Episodic and Semantic Memory to an AI Agent? A Working Python Build With SQLite, Local Embeddings and Tests

> Add episodic and semantic memory to an AI agent with two SQLite tables and two code paths: episodes record what happened and when, facts hold what is true now and a new value closes the old one, a write path extracts both after each session, and a read path ranks them into the prompt. Full Python with local embeddings, nine passing tests, no API key.

_Gaurav Dadhich · 2026-10-28_

\# How Do You Add Episodic and Semantic Memory to an AI Agent? A Working Python Build With SQLite, Local Embeddings and Tests \*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\](https://github.com/qdrant/fastembed) running \[BAAI/bge-small-en-v1.5\](https://huggingface.co/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\](https://www.maximem.ai/blog/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\](https://www.maximem.ai/blog/short-term-vs-long-term-memory-llm) 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\](https://arxiv.org/abs/2309.02427) 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\](https://docs.langchain.com/oss/python/concepts/memory) 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\](https://openai.github.io/openai-agents-python/sessions/) 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. \`\`\`bash 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. \`\`\`python """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\](https://github.com/asg017/sqlite-vec) or Postgres with \[pgvector\](https://github.com/pgvector/pgvector); \[our comparison of what a vector database does and does not give agent memory\](https://www.maximem.ai/blog/can-a-vector-database-be-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: \`\`\`python 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: \`\`\`python """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": "", "outcome": "", "importance": <0.0 to 1.0>}}, "facts": \[{{"attribute": "", "value": "", "statement": "", "multi": }}\]}} 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 \`\`\`python # ---------- 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\](https://arxiv.org/abs/2304.03442) 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: \`\`\`text 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 \`\`\`python """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: \`\`\`text 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 \`\`\`python 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\](https://www.maximem.ai/blog/how-to-test-agent-memory) 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\](https://aws.amazon.com/blogs/machine-learning/build-agents-to-learn-from-experiences-using-amazon-bedrock-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\](https://www.maximem.ai/blog/how-to-give-an-llm-long-term-memory) 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. \`\`\`python 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)\](https://arxiv.org/abs/2309.02427); \[Park et al., Generative Agents (arXiv 2304.03442)\](https://arxiv.org/abs/2304.03442); \[LangChain, memory concepts\](https://docs.langchain.com/oss/python/concepts/memory); \[OpenAI Agents SDK, sessions\](https://openai.github.io/openai-agents-python/sessions/); \[AWS, AgentCore episodic memory\](https://aws.amazon.com/blogs/machine-learning/build-agents-to-learn-from-experiences-using-amazon-bedrock-agentcore-episodic-memory/); \[fastembed\](https://github.com/qdrant/fastembed); \[BAAI/bge-small-en-v1.5\](https://huggingface.co/BAAI/bge-small-en-v1.5); \[sqlite-vec\](https://github.com/asg017/sqlite-vec); \[pgvector\](https://github.com/pgvector/pgvector); \[Maximem Synap documentation\](https://docs.maximem.ai).\*

---

Source: [https://www.maximem.ai/blog/episodic-and-semantic-memory-ai-agents](https://www.maximem.ai/blog/episodic-and-semantic-memory-ai-agents)