A Four-Layer Memory Design for AI Agents With Provenance
Short and long-term memory split, four layers (buffer, episodic, semantic, procedural), storage choices, async writes, and defenses against memory poisoning.
This article is published in English.
Picture an agent that fails at step fourteen of a long workflow. The model isn't to blame. The agent simply no longer knows what it has already done. The failure follows a familiar pattern. Tool calls are clean and prompts are solid, but once the conversation crosses some invisible length, the agent starts to loop. It repeats questions that were answered, fetches data it already holds, and contradicts itself between step six and step twelve as if the first five steps never occurred.
Swapping in a different frontier model rarely changes this. The bottleneck is the infrastructure around the model, and more precisely the lack of a real memory architecture. Pasting the last N messages into the prompt works for a chatbot. It falls apart when an agent must run a twenty-step workflow and remember at step seventeen what it decided at step three.
A pattern that addresses this is a four-layer memory stack, with the right storage behind each layer. What follows covers what each layer does, how teams at GitHub, Anthropic and Berkeley describe their versions, and Python sketches for every piece.
A context window is not memory
A context window is working memory: RAM, not a hard drive. Every token in it has a cost. Stuffing a whole conversation history into a 128K window sounds safe until you run a tool-heavy agent. A thirty-step workflow with tool calls, observations and reasoning traces eats through 100K tokens quickly, and when the window fills the oldest content is dropped from the front, which is usually where the user's original constraints and task framing sit.
A subtler problem is worse. A study from Stanford and UC Berkeley researchers, published in a 2024 ACL journal under the title "Lost in the Middle: How Language Models Use Long Contexts", found that LLM performance follows a U-shaped curve across the window. Models do well with information at the start and end of the context and clearly worse with information buried in the middle. The effect appeared across several architectures, model sizes and context lengths, with windows tested from 4K to 32K tokens. A 2025 follow-up from MIT offered a mathematical explanation, arguing that causal attention masking and rotary position encoding both reduce attention weight on tokens far from either end of the sequence.
For agents the consequence is clear. Even if everything fits, the model won't attend to it evenly. A critical constraint at the 30K mark of a 60K-token context has a real chance of being effectively invisible.
The model forgets because the system feeds it more than it can reason over reliably, and nothing controls what enters the window.
The starting split: short-term and long-term memory
Before the four-layer taxonomy, there's a simpler division that most engineers carry in their heads. Memory in AI systems falls into short-term and long-term, and seeing what each can't do motivates the finer architecture.
Short-term memory is scoped to a session and exists only for one interaction. It holds what the agent needs right now: the state of the current task, the step it's on, the last tool output and the reasoning trace under construction. In practice that's the context window, which is highly dynamic, constantly updated, costly per token and wiped when the session ends.
LangChain's documentation puts it well, describing short-term memory as storing conversation history, tool outputs and temporary reasoning state. What it leaves unsaid is that this is also the first thing to give way at scale. In a workflow of twenty agentic steps, intermediate outputs, reflections, search hits and tool replies pile up and compete for the same bounded space. It resembles trying to hold a complicated plan in your head while someone keeps interrupting with new facts.
Short-term memory has two practical components:
- Task state is the working record of progress: the current step, pending subtasks, tool execution status and coordination between agents. A travel-planning agent keeps the destination, budget limits, flights explored and hotels shortlisted so it doesn't redo work from step two when it reaches step eight.
- Intermediate outputs are reasoning artifacts in flight: tool results, generated summaries, SQL outputs, search snippets and scratchpad notes. Anthropic's engineering team has described a pattern across production systems in which agents that keep notes outside the context window and fetch them later do better on long tasks than agents that try to hold everything in context. Note-taking is itself a short-term memory strategy.
Long-term memory outlives sessions. It lets an agent recall that a user likes concise answers, that last week's deployment failed because of a migration problem, or that the database schema changed in January. LangChain defines it as memory that persists across threads and sessions, and in practical terms that data needs a database, not a context window.
It has two basic kinds:
- User preferences are personalization signals that build up over time, including writing style, tone, technical depth, tooling choices and formatting habits. A writing assistant that remembers you always want markdown and a high technical level becomes useful in a way no stateless system can match, however good its model.
- Historical decisions are the log of attempts and their results. Picture a DevOps agent that recalls a deployment approach that went badly last time, or the cloud vendor the team favors: it sidesteps costly repeats, and its behavior stays consistent from one session to the next.
Many teams stumble in the same way. They implement short-term memory by stuffing the context window and long-term memory by building one vector store. Both are wrong, or at best incomplete. The first fails at scale for the reasons above. The second breaks down since similarity search is a poor way to fetch session logs, preferences or behavior rules, and every kind of long-term memory wants a different backend. The four-layer design is a sharper version of the short/long split that maps each memory type to the storage it needs.
Four memory types, one taxonomy
Teams building production AI systems have independently arrived at the same taxonomy. The names vary, but the structure doesn't.
The MemGPT project, introduced by Charles Packer and colleagues at UC Berkeley in 2023, popularized the idea of treating an LLM's context window like RAM, which is limited and valuable. An operating system stretches memory through paging and disk storage, and agent frameworks can similarly simulate long-term memory by moving older context into larger storage and recalling it when needed. That idea later became the basis of Letta.
LangChain's LangMem team, the ReMem framework from Google DeepMind and Mem0's production architecture all describe variants of four types:
- Working memory is whatever is in context right now: current conversation turns, retrieved documents and tool outputs. It's fast, bounded and expensive per token.
- Episodic memory stores records of past interactions, such as what the agent did for a given user last Tuesday and how it turned out. It's chronological and session-scoped, usually held in a fast key-value store or SQL database and retrieved by recency or semantic similarity.
- Semantic memory keeps general facts, domain knowledge and things learned along the way. It's the RAG layer, pulled in by embedding similarity whenever a task calls for domain knowledge or an earlier finding.
- Procedural memory captures how the agent should behave: conventions, rules, constraints and learned patterns. Most teams get this one wrong. The naive version is a JSON file of rules, while the production version, according to LangMem's engineers, is the system prompt itself, an evolving set of instructions updated through a feedback loop and not a separate database.
These types mirror cognitive science, which explains why teams converge on them. They correspond to the episodic, semantic and procedural memory systems of human neuroscience, and working memory as context window is more than a metaphor. It's the closest functional analogue available.
Choosing storage: SQL, vector or graph
Which database should hold memory? Before writing code, the honest answer is that it varies with the memory type at hand.
Vector databases (Weaviate, Pinecone, Qdrant, Chroma) fit semantic memory, because they make similarity search over embeddings quick at scale, and most now offer hybrid retrieval that combines BM25 with vectors and clearly beats pure vector search on text. They lack temporal reasoning and relational queries, and two conflicting facts with near-identical embeddings can mislead you. Keep episodic records and structured audit trails out of them.
SQL databases such as PostgreSQL and SQLite suit episodic memory. Queries are structured, by user ID, session, task type or timestamp, write volume is low, and ACID guarantees matter when an agent logs what it did and when. PostgreSQL with the pgvector extension offers relational queries and vector search in a single system. Below tens of millions of records that's the pragmatic pick, with familiar operations, cheap hosting and SQL for debugging.
Graph databases such as Neo4j belong in a memory system only when episodic or semantic data has complex relational structure that needs multi-hop queries, such as who reported to whom three months ago when a document was written. Simple agent task records don't qualify. Plenty of teams reach for a graph and find they needed SQL.
The pattern that dominates production deployments is relational storage (SQL or PostgreSQL) for episodes and user profiles, a vector store for semantic lookups, and in-process state for working context, with PostgreSQL plus pgvector as the hybrid for teams that want less operational overhead. Choose the storage tier by what the memory type must do, and don't vectorize everything.
Layer 1: buffer memory with summarization
The most basic form is a deque capped at a fixed number of messages:
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
This holds up for brief exchanges. Over longer ones, older messages simply drop off the end and the agent loses its footing. A better approach, which LangMem labels the "hot path" method, condenses the old messages into a summary before discarding them:
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem separates hot-path summarization, which runs live and adds latency for the user, from background summarization that runs after the session. For user-facing applications background is almost always better: summaries improve because the whole conversation can be analyzed, and nobody waits. For autonomous agents running unattended, the hot path is acceptable since no user is waiting anyway.
Layer 2: episodic memory that checks its own citations
Here many teams fall short: they build retrieval but never record what the agent actually did.
GitHub Copilot's engineering blog describes a pattern worth copying. Each memory entry carries citations, meaning specific locations such as file paths, URLs, tool names and line numbers that let you check at retrieval time whether the stored fact still holds. GitHub calls this just-in-time validation. When the agent loads a memory at the start of a task, it checks that the cited source still exists and still supports the claim. If not, the memory is corrected or discarded, so the pool heals itself with no offline curation.
GitHub reported a measurable effect: the coding agent with memory showed a 7% rise in pull-request merge rate, 90% against 83%.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
The 28-day expiry follows GitHub's production choice. Old episodes aren't deleted but become ineligible for retrieval, and a LIMIT query over a date-filtered index is fast and needs no background curation job.
Layer 3: semantic memory using Weaviate's hybrid search
This is the RAG layer. Most guides cover setup and stop, while the real production problems lie in retrieval quality and provenance.
A frequent first error is making semantic memory read-only. A long-lived agent needs it to be read-write: when the agent finds a fact during a task, validates it and decides it's worth keeping, it should be written to semantic memory for future sessions. The second mistake is relying on cosine similarity alone. Weaviate's hybrid search blends BM25 keyword matching with vector similarity and tends to outperform either alone on agent-memory retrieval.
Third, and this is what opens the door to poisoning, entries often carry no provenance. Each one needs a source, a confidence score and a session identifier. Lacking those, trust-weighted retrieval is impossible, stale or doubtful entries can't be downranked, and tracing a wrong retrieval becomes guesswork.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Two details in retrieve_hybrid are missing from many implementations. One is that Weaviate's hybrid endpoint receives both the raw query text (used for BM25) and its embedding (used for vector search). Exact keyword hits come from the first and semantic closeness from the second, and the pairing dependably outperforms either alone. The other is the min_confidence filter, which keeps low-trust memories out of the context regardless of how close they are semantically. In effect it's a security control, not just a relevance tweak.
The update_confidence method supplies the feedback path. If the agent's own reasoning contradicts a retrieved memory during a task, lower its confidence. If a citation check confirms it, raise it. The pool should learn from use.
Layer 4: procedural memory and prompt evolution
This layer gets overlooked most. The naive design is a JSON file of rules that the agent reads, follows and extends after failures. It works but hits a ceiling, because rules in a separate file compete for context space and need explicit retrieval.
LangMem's engineers at LangChain published an idea worth borrowing: in production, procedural memory lives as the system prompt itself. It isn't stored beside the prompt or retrieved into it. It is the prompt. Whenever a failure teaches the agent a new rule of conduct, the prompt itself is rewritten, a practice they call prompt optimization as procedural memory. Learned behaviors become natural-language instructions the model reads at the start of every interaction.
The simpler JSON-file version is where most teams should begin:
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale tackles forgetting at the procedural level. A rule created after one unusual failure and never triggered again should fade, whereas a rule that fires regularly stays no matter how old it is. It's a blunt tool, but forgetting is simplest to implement in exactly this layer.
Once an agent runs persistently, the natural next step is to promote rules that are both trusted and frequently used into the system prompt itself. LangMem automates that with a prompt-optimization loop, and hand curation is perfectly fine for smaller setups.
Wiring the layers together
The next listing shows how the four layers fit together in production. One constraint is non-negotiable: writing to memory may never hold up the response.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
The order of context in build_context is deliberate. Research on positional attention bias supports the idea that models weigh the start and end of the window more heavily. Behavioral rules go at the top, since they're constraints that should shape everything after. The current conversation goes at the bottom because it's what the model most needs to answer next. Semantic facts and past episodes sit between, offering reference without crowding the active task.
The line self._pool.submit(self._persist_worker, result) is the most important engineering choice in the class. Every write call returns at once, so the LLM response is already streaming to the user before any database write begins.
Three decisions that shape the architecture
With four layers in place, three architectural decisions affect everything else, and many teams get at least one wrong.
Hot-path or background writes. The LangMem team spent considerable effort on this. Hot-path writes happen during the conversation, when the agent spots a fact worth storing and issues a write inline. Background writes happen after the session ends and analyze the full conversation for what deserves to be kept. The hot path adds latency, because every semantic.store() placed in the response loop is a database commit that the user sits through. Background writing improves recall quality, since themes and decisions across the whole arc can be examined and richer summaries stored. For user-facing production apps, background is nearly always right. The thread pool in persist() gets you the bulk of the gain while avoiding a fully asynchronous pipeline.
Retrieval or summarization. These solve different problems, and teams often conflate them. Retrieval means finding the relevant passage in a large corpus. The query is specific and the corpus large and fairly stable, like documentation, product catalogs or code. It fails when the query is complex, the relevant information is scattered over many passages, or the corpus changes faster than the embedding index. Summarization means compressing a conversation or document to fit the context. It works when input is sequential, as in conversation history, and some loss is tolerable, and it fails when the discarded detail turned out to matter. During 2025 the industry began replacing both terms with "context engineering": the question is no longer retrieval versus summarization but which mix of stored knowledge, retrieved passages and compressed history gives the best context for this particular inference call. The memory manager's build_context method is context engineering, since it decides which of the four stores to read, how much to take from each, and the order for putting the pieces together.
Provenance from day one. You can't retrofit provenance into a memory system. Starting with the very first write, each entry should hold its source, confidence score, session ID and creation time. Without them there's no trust-weighted retrieval, no filtering by recency and no way to audit a bad retrieval. Teams that skip provenance in the first sprint end up spending weeks adding it to a database that already holds thousands of entries without it.
Memory poisoning: attack patterns and defenses
This is the section that memory-design guides tend to omit, and it belongs in every one.
A 2025 arXiv paper introduced the MINJA attack framework, among the most thorough public analyses of memory poisoning in LLM agents. Its authors report that through ordinary user interactions, with no elevated privileges, attackers can reach injection success rates above 95% using a three-step pattern. A bridging step ties an innocent, legitimate subject to the topic the attacker is aiming at. An indication step supplies subtle cues that make the harmful memory look trustworthy. A progressive shortening step compresses the injection into a minimal footprint that survives memory extraction.
What distinguishes memory poisoning from ordinary prompt injection is that poisoned memories cross session boundaries and compound. One successful injection in the first session can influence every later session, indefinitely.
A separate 2025 paper on Bayesian trust defenses outlined a five-layer design that covers the attack surface:
- Moderation of inputs at write time. Ahead of each memory write, a combined trust score is computed from several independent signals: source provenance, semantic analysis for instruction patterns and anomaly detection against the user's history. No single signal is enough, and MINJA injections are built to pass single-signal checks.
- Instruction stripping before writes. If a candidate memory reads like a command ("always", "never", "you must", "do not"), clean it or throw it away instead of saving it. Genuine factual memories never need to give orders to the agent.
- Provenance tags on every entry. That means source, session ID, trust score at ingestion and creation timestamp, which makes the next two layers possible.
- Retrieval that weighs trust. The
min_confidencefilter on the Weaviate side of Layer 3 is an example. Low-provenance memories stay out of context whatever their similarity. In a fuller implementation the confidence field is updated from validation results, so memories that check out gain confidence and contradicted ones lose it. - Behavioral monitoring. Establish a baseline of normal behavior and alert on deviations, such as unusual tool-call sequences, unexpected memory write patterns, or outputs that diverge from historical distributions for similar inputs.
For most teams the sensible order is provenance tagging on every write first, then instruction detection at write time, then trust-weighted retrieval. Behavioral monitoring comes last because it needs a behavioral baseline, and building one takes weeks of production data.
One more point: research using real-world memory systems with pre-existing legitimate entries found that realistic conditions reduce attack effectiveness considerably compared with lab setups, since poisoned entries compete with legitimate memories for retrieval slots. That's no reason to skip defenses. It's a reason to keep legitimate entries high-quality and high-confidence, because they form the first line of defense.
What to build first
Not all four layers are needed immediately. For the majority of AI applications two of them cover about 80% of memory failures: Layer 1 (buffer plus summarization) and Layer 3 (Weaviate with hybrid search, the confidence filter and provenance). Between them they handle the two most frequent problems, an agent that forgets partway through a task and one that has no domain knowledge. Build both, compare task completion before and after, and only then judge what's absent.
Add episodic memory when you catch yourself asking why the agent forgot what it achieved last session, since that question is the signal. Don't add it earlier. Add procedural memory when the same class of mistake recurs across tasks and prompt engineering doesn't cure it, because a behavioral pattern needs behavioral memory.
PostgreSQL with pgvector is the sensible default for most teams. One database instance with CREATE EXTENSION IF NOT EXISTS vector provides ACID compliance, flexible SQL, vector similarity search and hybrid scoring, and one estimate puts hosting at about $45 a month for 10K memories. A standalone vector database only becomes worthwhile when semantic memory grows to the point where PostgreSQL's HNSW index is the thing slowing you down, something most teams will never experience.
Make every memory write asynchronous. Responding to the user shouldn't hinge on a database commit, so hand the work to a thread pool and move on.
And skip in-memory vector stores in production. A deployment without persistent storage loses all accumulated knowledge at the first server restart, a mistake few people make twice.