Designing Memory Architecture for Stateful AI Agents: Short-Term Context and Long-Term Persistence
Short-term memory answers what just happened in this conversation. Long-term memory answers what the system knows about this user.
·
A language model holds nothing between calls. Each request arrives with no memory of the previous one, and the model responds from the prompt in front of it and nothing more. For a one-shot question, that is fine. For an agent a person returns to across days, threads, and separate tasks, it becomes the defining design constraint. The behavior users expect from an agent, recalling what they said last week, resuming a task that stalled, applying their stated preferences without being asked again, is precisely the behavior a stateless model cannot produce on its own. A large language model (LLM) does not forget your context because it chose to. It never had it.
The reflex fix is to paste the history back in. Keep a buffer of prior turns and prepend it to every prompt, so the agent looks like it remembers. That illusion survives a short demo and breaks in production. Building stateful AI agents that hold continuity across sessions takes a memory architecture, not a bigger buffer.
Why Stateless Models Break Multi-Session Agents
The conversation-buffer approach fails in three measurable ways, and they compound. The first is cost. Every prior turn you replay is billed again on the next call, so a buffer that grows linearly turns a fifty-turn session into fifty replays of the same opening tokens. The second is latency. Larger prompts take longer to transmit and to process, and that overhead lands on every turn, not only the long ones. The third is the one that surprises teams: recall does not hold up as the prompt grows. Research on long-context behavior, notably Liu and colleagues in 2023, found that models attend most reliably to information at the start and end of a prompt and least reliably to whatever sits in the middle. Bury a critical instruction in turn nine of a forty-turn transcript and the model may act as though it is not there.
There is a harder limit underneath all of this. Every model has a fixed context window, commonly between 8,000 and 128,000 tokens depending on the model. Cross it, and the provider either rejects the request with a context-length error or silently truncates the oldest turns, which means the agent loses exactly the early context the buffer was meant to preserve. None of this touches the cross-session problem. A buffer lives inside one session. Open a new thread tomorrow, and the buffer is empty again. The preferences and facts the user established yesterday are gone, because nothing extracted them into a place that outlives the session.
So the standard approach does not scale, and the reason is that it conflates two different jobs: holding the immediate conversation and remembering across conversations. Those are separate problems with separate lifetimes, and they need separate machinery.
Measuring Context Pressure in a Live Agent
Before designing anything, instrument what you already have. The failure modes above stay invisible until you measure token consumption per turn against the model's window, and most teams find the ceiling only when a request errors out in front of a customer. A small instrument that tracks tokens as the session grows turns a production incident into a logged warning.
# context_budget.py
# Instruments an agent's session log so you can see context pressure build
# up turn by turn, instead of discovering it when a request silently truncates
# or the provider returns a context-length error mid-conversation.
from dataclasses import dataclass, field
# In production, count with the model's real tokenizer (for example tiktoken's
# cl100k_base or the o200k_base encoding). We wrap it so a missing encoding or
# offline build degrades to a conservative heuristic instead of crashing the
# request path, which is what you actually want in a hot loop.
def count_tokens(text: str) -> int:
try:
import tiktoken
return len(tiktoken.get_encoding("o200k_base").encode(text))
except Exception:
# ~4 characters per token is a safe upper-bound heuristic for English.
return (len(text) + 3) // 4
@dataclass
class SessionLog:
budget: int # hard ceiling for prompt tokens
events: list = field(default_factory=list)
def add(self, role: str, text: str) -> None:
self.events.append((role, text, count_tokens(text)))
def used(self) -> int:
return sum(t for _, _, t in self.events)
def pressure(self) -> float:
# Fraction of the budget consumed. Alert teams above ~0.75 so there is
# headroom for the model's own output tokens, not just the prompt.
return self.used() / self.budget
log = SessionLog(budget=2000)
for turn in range(1, 13):
log.add("user", "Walk me through the failure we saw in the nightly run. " * turn)
log.add("assistant", "Here is the breakdown of the root cause and the fix. " * turn)
if log.pressure() > 0.75:
print(f"turn {turn:>2}: used={log.used():>5} tokens "
f"pressure={log.pressure():.0%} ACTION: evict or summarize")
else:
print(f"turn {turn:>2}: used={log.used():>5} tokens pressure={log.pressure():.0%}")The threshold matters. Alerting at seventy-five percent of the window rather than at one hundred leaves headroom for the model's own output tokens, which are drawn from the same budget. Count tokens with the model's actual tokenizer, since the tiktoken library exposes the o200k_base and cl100k_base encodings used by current models, because a whitespace estimate can be off by a wide margin on code, punctuation, or non-English text. The heuristic fallback in the wrapper exists so the request path degrades to an estimate rather than throwing when an encoding is unavailable. Once you can watch pressure climb, the case for a second memory layer stops being theoretical.
A Two-Layer Memory Architecture for Stateful AI Agents
The architecture that holds up splits memory by lifetime. Short-term memory is the raw event log of the current session: every user turn and agent response, appended cheaply, scoped to one interaction, archived or discarded when the session ends. Long-term memory is the distilled residue, the facts, preferences, and summaries worth keeping, extracted from raw events and persisted across every future session. Short-term memory answers what just happened in this conversation. Long-term memory answers what the system knows about this user. Keep both in one store and you will either bloat the prompt with raw transcript or lose durable facts in the churn.
Long-term memory is shared infrastructure, often across many tenants, so the first structural decision is addressing. Records are keyed by a hierarchical namespace rather than a flat identifier, and the hierarchy itself enforces isolation. A retrieval scoped to one tenant cannot resolve a key that belongs to another, because the tenant identifier is a structural prefix of every key rather than a filter applied after the fetch.
# namespaces.py
# Memory records are addressed by a hierarchical namespace, not a flat key.
# The hierarchy is what enforces multi-tenant isolation: a retrieval scoped to
# one tenant can never resolve a key that belongs to another, because the
# tenant id is a structural prefix of every key rather than a filter applied
# after the fetch.
import re
# Components are validated so a caller cannot smuggle a separator into an id and
# escape its own subtree. This is the single most common isolation bug in
# home-grown memory stores: an unescaped "/" in a user id reads as a new level.
_SAFE = re.compile(r"^[A-Za-z0-9_-]+$")
def make_namespace(tenant: str, actor: str, scope: str) -> str:
for part in (tenant, actor, scope):
if not _SAFE.match(part):
raise ValueError(f"unsafe namespace component: {part!r}")
return f"/{tenant}/{actor}/{scope}"
class MemoryStore:
def __init__(self):
self._records: dict[str, list[str]] = {}
def put(self, namespace: str, value: str) -> None:
self._records.setdefault(namespace, []).append(value)
def query(self, namespace_prefix: str) -> list[str]:
# Only keys under the caller's prefix are visible. The prefix check is
# the isolation boundary, so it runs before any ranking or scoring.
hits = []
for key, values in self._records.items():
if key == namespace_prefix or key.startswith(namespace_prefix + "/"):
hits.extend(values)
return hits
store = MemoryStore()
store.put(make_namespace("tenant-a", "user-42", "preferences"), "prefers metric units")
store.put(make_namespace("tenant-a", "user-42", "facts"), "deploys to two regions")
store.put(make_namespace("tenant-b", "user-42", "preferences"), "prefers imperial units")
print("tenant-a/user-42 sees:", store.query("/tenant-a/user-42"))
print("tenant-b/user-42 sees:", store.query("/tenant-b/user-42"))
# A caller trying to break out of its subtree by embedding a separator is
# rejected at construction time, before any record is written or read.
try:
make_namespace("tenant-a", "user-42/../tenant-b", "preferences")
except ValueError as exc:
print("isolation guard:", exc)The validation is not decoration. The most common isolation defect in a home-grown memory store is an unescaped separator in a user-supplied identifier, where a value shaped like a path fragment reads as a new namespace level and lets a caller climb out of its own subtree. Rejecting unsafe components at construction time closes that hole before a single record is written or read.
Raw events are cheap to append and nearly useless to search. Turning them into retrievable knowledge is the job of an extraction pipeline, and that pipeline belongs off the request path. Distillation calls a model and costs hundreds of milliseconds, so running it inline would tax every turn with work the user never sees. An asynchronous worker reads batches of raw events and writes back typed long-term records.
# extraction_worker.py
# Raw session events are cheap to append and useless to search. The extraction
# worker runs off the request path, reads batches of raw events and distills
# them into durable long-term records: facts, preferences and rolling
# summaries. Keeping this asynchronous is the point. Distillation calls a model
# and takes hundreds of milliseconds, so doing it inline would tax every turn.
import queue
import threading
import time
raw_events: "queue.Queue[dict]" = queue.Queue()
long_term: list[dict] = []
def distill(batch: list[dict]) -> list[dict]:
# PLACEHOLDER for the model call. In production this is a structured-output
# request to an extraction model that returns typed records. The rule-based
# stub below keeps the worker deterministic and runnable for tests; swap it
# for the real call and keep the same return contract.
records = []
for ev in batch:
text = ev["text"].lower()
if "prefer" in text:
records.append({"type": "preference", "value": ev["text"], "src": ev["id"]})
elif any(k in text for k in ("region", "deploy", "cluster", "schema")):
records.append({"type": "fact", "value": ev["text"], "src": ev["id"]})
return records
def worker(batch_size: int = 4, idle_flush: float = 0.2) -> None:
batch: list[dict] = []
last = time.monotonic()
while True:
try:
ev = raw_events.get(timeout=idle_flush)
if ev is None: # poison pill: drain then stop
if batch:
long_term.extend(distill(batch))
break
batch.append(ev)
except queue.Empty:
pass
# Flush on a full batch or after an idle gap, so a quiet session still
# gets its memory persisted instead of waiting for more traffic.
if len(batch) >= batch_size or (batch and time.monotonic() - last > idle_flush):
long_term.extend(distill(batch))
batch, last = [], time.monotonic()
t = threading.Thread(target=worker, daemon=True)
t.start()
for i, msg in enumerate([
"I prefer metric units in reports",
"we deploy to two regions",
"the order schema changed last week",
"thanks, that helps",
"I prefer dark themed dashboards",
]):
raw_events.put({"id": f"e{i}", "text": msg})
raw_events.put(None)
t.join(timeout=2)
for rec in long_term:
print(rec)Two details in that worker earn their place. The batch flushes on a size trigger or an idle timeout, so a quiet session still gets its memory persisted instead of waiting for traffic that never arrives. The drain-on-shutdown path matters even more: an earlier version that broke on the poison pill without flushing the partial batch silently dropped the last few events whenever the worker stopped, the kind of lost-write bug that surfaces as missing memories weeks later. The model call is stubbed here with deterministic rules so the worker stays testable. The contract is the point. Swap the stub for a structured-output extraction call, and the rest holds.
With both layers in place, a single agent turn reads from each, assembles a prompt within the token budget, and persists the new event for later extraction.
# agent_turn.py
# One agent turn that reads from both memory layers and respects a token
# budget. Long-term records come first because they are dense and stable;
# recent raw turns fill whatever budget remains. This ordering matters: models
# attend less reliably to the middle of a long prompt, so the highest-value
# memory belongs near the edges, not buried in the transcript.
def count_tokens(text: str) -> int:
return (len(text) + 3) // 4 # heuristic; production swaps in the real tokenizer
def assemble_context(long_term: list[str], recent: list[str], budget: int) -> list[str]:
context, used = [], 0
# Long-term memory is admitted first, up to half the budget, so a long
# transcript can never crowd out durable facts the agent depends on.
for rec in long_term:
cost = count_tokens(rec)
if used + cost > budget // 2:
break
context.append(rec)
used += cost
# Recent turns fill the remainder, newest first, dropping the oldest.
for turn in reversed(recent):
cost = count_tokens(turn)
if used + cost > budget:
break
context.append(turn)
used += cost
return context
def run_turn(user_msg, long_term, recent, budget=400):
# Reserve room for the user message before admitting any memory, otherwise
# the assembled prompt overshoots the budget once the message is appended.
reserve = count_tokens(f"user: {user_msg}") + 1
context = assemble_context(long_term, recent, budget - reserve)
prompt = "\n".join(context + [f"user: {user_msg}"])
# PLACEHOLDER for the model call. The returned context window is what the
# agent actually reasons over; assert it before spending tokens on inference.
assert count_tokens(prompt) <= budget, "context exceeds budget"
recent.append(f"user: {user_msg}") # short-term log keeps growing
return prompt
long_term = ["fact: deploys to two regions", "preference: prefers metric units"]
recent = [f"user: earlier message {i}" for i in range(20)]
prompt = run_turn("why did the canary roll back?", long_term, recent, budget=120)
print(prompt)
print("---")
print("prompt tokens:", count_tokens(prompt), "of budget 120")Ordering is a deliberate choice. Long-term records go in first and are capped at half the budget so a long transcript can never crowd out the durable facts the agent depends on; then recent turns fill the remainder, newest first. Because models attend least reliably to the middle of a prompt, the highest-value memory belongs near the edges. Reserving room for the user message before admitting any memory keeps the assembled prompt from overshooting once the message is appended, a bug the budget assertion catches the moment it slips.
Trade-Offs in Extraction Timing, Retrieval and Consistency
Every decision in this design buys one property at the cost of another, and pretending otherwise is how teams get surprised later. Asynchronous extraction keeps turn latency flat, but it introduces an eventual-consistency window: for the few seconds between an event and its distillation, the long-term layer does not yet know what the user just said. For most assistants, that lag is invisible. For a workflow where the user states a fact and immediately expects the agent to act on it across a session boundary, you either shorten the window or read through to the short-term log for the current session. Synchronous extraction removes the lag and adds those hundreds of milliseconds to every turn instead. Pick the trade deliberately.
Retrieval carries its own tension. Score candidates purely on semantic similarity and a stale record can win, because a preference the user set months ago and has since reversed still matches the query embedding closely. Blend a recency decay into the score and freshness gets a vote.
# ranking.py
# Retrieval that scores only on semantic similarity will happily surface a
# year-old preference the user has since reversed. Blending a recency decay
# into the score fixes that, but the decay rate is a real tuning knob: too
# aggressive and the agent forgets stable facts, too gentle and stale memory
# wins. This compares pure similarity against a recency-weighted score on the
# same candidate set so the trade-off is visible.
import numpy as np
# query vector and candidate vectors stand in for embeddings; in production
# these come from an embedding model and a vector index, but the ranking math
# is identical regardless of where the vectors originate.
query = np.array([0.9, 0.1, 0.0])
candidates = [
{"text": "prefers metric units", "vec": np.array([0.88, 0.12, 0.0]), "age_days": 210},
{"text": "prefers imperial units", "vec": np.array([0.86, 0.14, 0.0]), "age_days": 3},
{"text": "works on payments", "vec": np.array([0.20, 0.80, 0.1]), "age_days": 12},
]
def cosine(a, b):
return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))
def rank(decay_per_day: float):
scored = []
for c in candidates:
sim = cosine(query, c["vec"])
# Exponential recency decay. A fresher record with slightly lower
# similarity can now outrank a near-duplicate that is months old.
recency = np.exp(-decay_per_day * c["age_days"])
blended = sim if decay_per_day == 0 else 0.6 * sim + 0.4 * recency
scored.append((round(float(blended), 3), c["text"]))
return sorted(scored, reverse=True)
print("pure semantic: ", rank(decay_per_day=0.0)[0])
print("recency-weighted 0.01:", rank(decay_per_day=0.01)[0])
print("recency-weighted 0.05:", rank(decay_per_day=0.05)[0])The decay rate is a real knob, not a default to ignore. Too aggressive and the agent forgets stable facts that should persist for months; too gentle and outdated memory keeps surfacing. Notice in the output how a record that is three days old overtakes a months-old record once recency is weighted at all, even though the older record scores marginally higher on similarity alone. The right constant depends on how fast your domain's facts actually change, which is something only the people running the system can calibrate against real traffic.
There is also a quieter cost to the whole approach. Extraction and embedding are not free, since every event you choose to distill and index consumes compute and storage, so indexing everything is rarely the right call. Selective extraction, keeping facts, preferences, and periodic summaries rather than raw chatter, keeps the long-term store small enough to search quickly and cheap enough to justify. Teams that skip the two-layer split do not avoid this cost. They pay it as a degraded user experience: repetitive agents that re-ask for the same information, prompts that balloon until they error, and memory that evaporates the moment a session closes. Separating short-term context from long-term persistence is what turns a stateless model into an agent that behaves as though it remembers, and it stays an ordinary software engineering problem once you stop expecting the model to solve it for you.
Opinions expressed by DZone contributors are their own.
Comments