Imagine you've deployed a customer support chatbot for a mid-sized financial services company. The first few turns of every conversation work beautifully — the LLM is responsive, contextually accurate, and impressively fluent. Then around turn eight, something goes wrong. The customer references a transaction they mentioned three exchanges ago, and the bot responds as if hearing it for the first time. Or worse: you're running a multi-session workflow where a customer returns two days later to continue an insurance claim, and the model has no recollection of any prior interaction. The conversation starts from zero. Trust evaporates.
This is the central problem of LLM memory architecture: language models are stateless by design. Every API call is a clean slate. The "memory" that makes a conversation feel coherent is an illusion you, the engineer, have to construct and maintain. Getting this right in a toy demo is trivial. Getting it right in a production enterprise system — one handling thousands of concurrent users, variable conversation lengths, sensitive regulatory data, latency constraints, and session persistence across days or weeks — requires genuine architectural thinking.
By the end of this lesson, you'll have the deep understanding and practical tooling to design and implement enterprise-grade memory systems for LLM applications. You'll know the trade-offs between every major approach, when to use each one, and how to avoid the failure modes that sink most production deployments.
What you'll learn:
This lesson assumes you're comfortable with:
Before building anything, you need to internalize what's actually happening when a model "remembers" something. A large language model has no persistent state between API calls. When you call gpt-4o or claude-3-5-sonnet, you're passing in a sequence of tokens and receiving a completion. The model's attention mechanism processes the entire input simultaneously — there's no "reading" happening linearly, no gradual accumulation. Everything the model knows about your conversation must be present in the input tokens of each call.
The context window is your physical budget. GPT-4o has a 128K token context window. Claude 3.5 Sonnet offers 200K. Gemini 1.5 Pro can reach 1M tokens in some configurations. These numbers sound enormous, but they evaporate quickly in enterprise applications:
Add it up across a long technical support session and you'll find yourself bumping against ceilings sooner than expected. And cost compounds: at GPT-4o pricing (~$5 per million input tokens as of mid-2024), a 100K-token conversation processed at every turn costs roughly $0.50 per conversation in input tokens alone. Scale that to 50,000 support conversations per month and you're looking at $25,000 monthly just in input token costs — before responses.
This is why memory architecture matters. It's not an academic exercise. Every architectural decision you make directly affects cost, latency, quality, and scalability.
The simplest approach: store every message in a list and pass the complete history to the model on every call. This is what most tutorials show you, and it's correct for short, single-session interactions.
from openai import OpenAI
from dataclasses import dataclass, field
from typing import Literal
import tiktoken
MessageRole = Literal["system", "user", "assistant", "tool"]
@dataclass
class Message:
role: MessageRole
content: str
@dataclass
class ConversationBuffer:
system_prompt: str
messages: list[Message] = field(default_factory=list)
model: str = "gpt-4o"
def __post_init__(self):
self.encoder = tiktoken.encoding_for_model(self.model)
def add_message(self, role: MessageRole, content: str):
self.messages.append(Message(role=role, content=content))
def count_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
def total_tokens(self) -> int:
total = self.count_tokens(self.system_prompt)
for msg in self.messages:
total += self.count_tokens(msg.content) + 4 # per-message overhead
return total
def to_api_format(self) -> list[dict]:
result = [{"role": "system", "content": self.system_prompt}]
result.extend({"role": m.role, "content": m.content} for m in self.messages)
return result
def chat(self, client: OpenAI, user_input: str) -> str:
self.add_message("user", user_input)
response = client.chat.completions.create(
model=self.model,
messages=self.to_api_format()
)
assistant_message = response.choices[0].message.content
self.add_message("assistant", assistant_message)
return assistant_message
This works until it doesn't. The failure mode is silent: once total_tokens() exceeds the model's context limit, the API call will either error with a context length exceeded message or, with some providers, silently truncate from the beginning of the conversation — which is often the worst possible outcome, because you lose the setup context that gives the whole conversation meaning.
Add a guard:
def chat(self, client: OpenAI, user_input: str, max_context_tokens: int = 120000) -> str:
self.add_message("user", user_input)
projected_tokens = self.total_tokens()
if projected_tokens > max_context_tokens:
raise ValueError(
f"Context limit exceeded: {projected_tokens} tokens "
f"(limit: {max_context_tokens}). Consider using a "
f"summarization buffer or sliding window."
)
response = client.chat.completions.create(
model=self.model,
messages=self.to_api_format()
)
assistant_message = response.choices[0].message.content
self.add_message("assistant", assistant_message)
return assistant_message
When to use full history: Single-session interactions under ~20 turns, high-stakes conversations where every detail matters (medical intake, legal document review), debugging and development environments where you want maximum fidelity. Never use this pattern as your primary approach for persistent, long-running enterprise workflows.
The sliding window approach is conceptually simple: keep only the N most recent messages in context. When the buffer fills up, drop the oldest messages. It's cheap to implement and guarantees you'll never hit a context limit.
The problem is that naive sliding window implementations destroy coherence in ways that are hard to detect. Consider this support conversation:
If your window only retained turns 10–15, the model has no idea the customer mentioned their account type in turn 1, no access to the account number from turn 7, and can't answer the question coherently. The conversation collapses.
Here's a smarter sliding window that preserves the system-relevant opening:
@dataclass
class SlidingWindowBuffer:
system_prompt: str
max_tokens: int = 8000
model: str = "gpt-4o"
pinned_messages: list[Message] = field(default_factory=list)
sliding_messages: list[Message] = field(default_factory=list)
def __post_init__(self):
self.encoder = tiktoken.encoding_for_model(self.model)
def count_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
def pin_message(self, role: MessageRole, content: str):
"""Pin a message so it's never evicted from context."""
self.pinned_messages.append(Message(role=role, content=content))
def add_message(self, role: MessageRole, content: str):
self.sliding_messages.append(Message(role=role, content=content))
self._trim_to_budget()
def _calculate_fixed_tokens(self) -> int:
fixed = self.count_tokens(self.system_prompt) + 10
for msg in self.pinned_messages:
fixed += self.count_tokens(msg.content) + 4
return fixed
def _trim_to_budget(self):
fixed_tokens = self._calculate_fixed_tokens()
available = self.max_tokens - fixed_tokens
# Walk from the end, keeping messages that fit
kept = []
running_tokens = 0
for msg in reversed(self.sliding_messages):
msg_tokens = self.count_tokens(msg.content) + 4
if running_tokens + msg_tokens <= available:
kept.insert(0, msg)
running_tokens += msg_tokens
else:
break
self.sliding_messages = kept
def to_api_format(self) -> list[dict]:
result = [{"role": "system", "content": self.system_prompt}]
result.extend({"role": m.role, "content": m.content} for m in self.pinned_messages)
result.extend({"role": m.role, "content": m.content} for m in self.sliding_messages)
return result
The pin_message capability lets you preserve critical context — the customer's name, account type, and the stated purpose of the conversation — regardless of window size. In practice, you'd extract these facts programmatically at the start of a session and pin them.
When to use sliding window: Low-stakes, high-volume customer service bots where recency is more important than full coherence. Live assistants where the "current topic" is what matters. Never use this as your sole memory strategy for complex workflows where early-conversation details are referenced late.
Summarization buffers represent a significant step up in architectural sophistication. Instead of discarding old messages outright, you compress them using the LLM itself, maintaining a rolling summary of everything that's happened while keeping recent exchanges in full fidelity.
The core idea: maintain two zones of memory. The "hot zone" contains the last N messages in full detail. The "cold zone" is a compact LLM-generated summary of everything before the hot zone. When the hot zone fills up, you summarize its oldest messages and merge that summary into the cold zone, then clear those messages from the hot zone.
import asyncio
from openai import AsyncOpenAI
SUMMARIZATION_PROMPT = """You are maintaining a running summary of a conversation between a customer support agent and a customer.
Given the existing summary and new conversation segments, produce an updated summary that:
1. Preserves all specific details: names, account numbers, dates, product names, error codes, amounts
2. Captures the current state of any open issues or pending actions
3. Notes any commitments made by either party
4. Is written in third person, past tense
5. Stays under 400 words
Existing summary:
{existing_summary}
New conversation to incorporate:
{new_conversation}
Updated summary:"""
@dataclass
class SummarizationBuffer:
system_prompt: str
summarization_model: str = "gpt-4o-mini" # Cheaper model for summarization
inference_model: str = "gpt-4o"
hot_zone_max_tokens: int = 6000
summary: str = ""
hot_zone: list[Message] = field(default_factory=list)
def __post_init__(self):
self.encoder = tiktoken.encoding_for_model(self.inference_model)
self.client = AsyncOpenAI()
def count_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
def hot_zone_tokens(self) -> int:
return sum(self.count_tokens(m.content) + 4 for m in self.hot_zone)
async def _compress_oldest_messages(self, n_messages: int = 4):
"""Summarize the oldest n_messages from the hot zone into the summary."""
messages_to_compress = self.hot_zone[:n_messages]
conversation_text = "\n".join(
f"{m.role.upper()}: {m.content}"
for m in messages_to_compress
)
prompt = SUMMARIZATION_PROMPT.format(
existing_summary=self.summary or "No prior summary.",
new_conversation=conversation_text
)
response = await self.client.chat.completions.create(
model=self.summarization_model,
messages=[{"role": "user", "content": prompt}],
max_tokens=600
)
self.summary = response.choices[0].message.content
self.hot_zone = self.hot_zone[n_messages:]
async def add_message(self, role: MessageRole, content: str):
self.hot_zone.append(Message(role=role, content=content))
if self.hot_zone_tokens() > self.hot_zone_max_tokens:
# Compress in batches of 4 (2 exchanges) to preserve dialogue rhythm
await self._compress_oldest_messages(n_messages=4)
def to_api_format(self) -> list[dict]:
messages = [{"role": "system", "content": self.system_prompt}]
if self.summary:
messages.append({
"role": "system",
"content": f"[Conversation history summary]\n{self.summary}"
})
messages.extend(
{"role": m.role, "content": m.content}
for m in self.hot_zone
)
return messages
async def chat(self, user_input: str) -> str:
await self.add_message("user", user_input)
response = await self.client.chat.completions.create(
model=self.inference_model,
messages=self.to_api_format()
)
assistant_message = response.choices[0].message.content
await self.add_message("assistant", assistant_message)
return assistant_message
Notice the architectural choice to use a cheaper, faster model (gpt-4o-mini) for summarization. Summarization is a well-structured task that doesn't require the same reasoning power as your primary inference. This is not just a cost optimization — it also reduces latency for the compression step, which matters when compression happens mid-conversation.
The summarization prompt deserves careful attention. Generic summarization destroys the specific details — account numbers, error codes, promised callback times — that enterprise conversations depend on. Your prompt must explicitly instruct the model to preserve these structured facts. In production, you'll want to tune this prompt for your specific domain and validate it against real conversation samples.
Warning: Summarization introduces an irreversibility problem. Information compressed into the summary is now filtered through the model's interpretation. If the summarization model misses or mischaracterizes a detail, that error propagates forward and cannot be corrected without access to the original transcript. Always store the raw full conversation history separately, even if you're serving the model a summarized version.
In a synchronous chat loop, triggering a summarization call mid-conversation adds perceptible latency. A production pattern to mitigate this: run summarization proactively, slightly before you actually need it.
async def add_message(self, role: MessageRole, content: str):
self.hot_zone.append(Message(role=role, content=content))
# Trigger compression at 80% of limit, not 100%
# This lets compression run while user is reading the response
if self.hot_zone_tokens() > self.hot_zone_max_tokens * 0.8:
# Fire and forget — don't await here if this is called during response processing
asyncio.create_task(self._compress_oldest_messages(n_messages=4))
This creates a background task that compresses while the user is reading the model's response, hiding the latency completely.
The three patterns above are all variations on the same core approach: inject context into the prompt through linear history. Vector-based retrieval memory is architecturally different. Instead of maintaining a contiguous conversation history, you store every turn of every conversation as an embedding in a vector store, then retrieve only the most semantically relevant memories for each new query.
This approach scales horizontally in ways the others don't. It handles multi-session memory (a customer returning after a week) and cross-session memory (aggregating patterns from a user's entire history) naturally, because the retrieval mechanism is semantic rather than temporal.
from openai import AsyncOpenAI
import asyncpg
import numpy as np
from datetime import datetime, timezone
import json
class VectorMemoryStore:
"""
Enterprise vector memory using PostgreSQL + pgvector.
In production, you'd use a managed service like Pinecone,
Weaviate, or Qdrant, but pgvector keeps the data in your
existing infrastructure, which matters for compliance.
"""
def __init__(self, connection_string: str, embedding_model: str = "text-embedding-3-small"):
self.connection_string = connection_string
self.embedding_model = embedding_model
self.client = AsyncOpenAI()
async def initialize(self):
self.pool = await asyncpg.create_pool(self.connection_string)
async with self.pool.acquire() as conn:
await conn.execute("""
CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE IF NOT EXISTS conversation_memories (
id BIGSERIAL PRIMARY KEY,
user_id TEXT NOT NULL,
session_id TEXT NOT NULL,
turn_index INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
embedding vector(1536),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_memories_user
ON conversation_memories(user_id);
CREATE INDEX IF NOT EXISTS idx_memories_session
ON conversation_memories(session_id);
CREATE INDEX IF NOT EXISTS idx_memories_embedding
ON conversation_memories USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
""")
async def embed_text(self, text: str) -> list[float]:
response = await self.client.embeddings.create(
model=self.embedding_model,
input=text
)
return response.data[0].embedding
async def store_turn(
self,
user_id: str,
session_id: str,
turn_index: int,
role: str,
content: str,
metadata: dict = None
):
embedding = await self.embed_text(content)
async with self.pool.acquire() as conn:
await conn.execute("""
INSERT INTO conversation_memories
(user_id, session_id, turn_index, role, content, embedding, metadata)
VALUES ($1, $2, $3, $4, $5, $6::vector, $7)
""",
user_id, session_id, turn_index, role, content,
str(embedding), json.dumps(metadata or {}))
async def retrieve_relevant(
self,
user_id: str,
query: str,
k: int = 5,
exclude_session_id: str = None,
recency_weight: float = 0.2
) -> list[dict]:
"""
Retrieve k most relevant memories for a user, weighted by
semantic similarity and recency.
"""
query_embedding = await self.embed_text(query)
async with self.pool.acquire() as conn:
# Using a combined score: cosine similarity + recency decay
rows = await conn.fetch("""
SELECT
content,
role,
session_id,
turn_index,
created_at,
metadata,
1 - (embedding <=> $1::vector) as similarity,
EXTRACT(EPOCH FROM (NOW() - created_at)) / 86400.0 as days_ago
FROM conversation_memories
WHERE user_id = $2
AND ($3::text IS NULL OR session_id != $3)
ORDER BY
(1 - (embedding <=> $1::vector)) * (1 - $4) +
(1.0 / (1.0 + EXTRACT(EPOCH FROM (NOW() - created_at)) / 86400.0)) * $4
DESC
LIMIT $5
""", str(query_embedding), user_id, exclude_session_id,
recency_weight, k)
return [dict(row) for row in rows]
async def get_session_history(
self,
session_id: str,
max_turns: int = 20
) -> list[dict]:
async with self.pool.acquire() as conn:
rows = await conn.fetch("""
SELECT role, content, turn_index, created_at
FROM conversation_memories
WHERE session_id = $1
ORDER BY turn_index ASC
LIMIT $2
""", session_id, max_turns)
return [dict(row) for row in rows]
Now you can build a memory-augmented conversation manager that draws from both the current session's history and relevant memories from prior sessions:
class MultiSessionMemoryManager:
def __init__(self, memory_store: VectorMemoryStore, system_prompt: str):
self.memory_store = memory_store
self.system_prompt = system_prompt
self.client = AsyncOpenAI()
async def build_context(
self,
user_id: str,
session_id: str,
current_query: str,
hot_window_turns: int = 8
) -> list[dict]:
# 1. Get recent turns from current session
recent_turns = await self.memory_store.get_session_history(
session_id, max_turns=hot_window_turns
)
# 2. Retrieve semantically relevant memories from prior sessions
prior_memories = await self.memory_store.retrieve_relevant(
user_id=user_id,
query=current_query,
k=4,
exclude_session_id=session_id,
recency_weight=0.25
)
# 3. Assemble context
messages = [{"role": "system", "content": self.system_prompt}]
if prior_memories:
memory_block = self._format_retrieved_memories(prior_memories)
messages.append({
"role": "system",
"content": f"[Relevant context from prior interactions]\n{memory_block}"
})
# Add current session turns
for turn in recent_turns:
messages.append({"role": turn["role"], "content": turn["content"]})
return messages
def _format_retrieved_memories(self, memories: list[dict]) -> str:
lines = []
for mem in memories:
days_ago = mem.get("days_ago", 0)
time_label = f"{int(days_ago)} days ago" if days_ago >= 1 else "earlier today"
lines.append(f"[{time_label}, {mem['role']}]: {mem['content']}")
return "\n".join(lines)
async def chat(
self,
user_id: str,
session_id: str,
user_input: str,
turn_index: int
) -> str:
# Store user turn
await self.memory_store.store_turn(
user_id=user_id,
session_id=session_id,
turn_index=turn_index,
role="user",
content=user_input
)
# Build context with retrieval
messages = await self.build_context(user_id, session_id, user_input)
# Add current user message (it's already the last turn in session history,
# but we need it as the final message to trigger response)
messages.append({"role": "user", "content": user_input})
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=messages
)
assistant_message = response.choices[0].message.content
# Store assistant turn
await self.memory_store.store_turn(
user_id=user_id,
session_id=session_id,
turn_index=turn_index + 1,
role="assistant",
content=assistant_message
)
return assistant_message
Important design consideration: Retrieve relevant context from prior sessions (excluded by
exclude_session_id), not from the current one. The current session's recent history is already being injected through the hot window. Mixing retrieval results from the current session with its chronological history creates confusing duplicate context and can cause the model to respond as though it has seen information twice.
The vector memory system above handles retrieval, but enterprise applications need more than retrieval — they need structured state. Consider what a claims processing AI needs to track:
This is structured state, not conversational history, and it needs a different storage model:
from pydantic import BaseModel
from enum import Enum
import json
class ClaimStatus(str, Enum):
INITIATED = "initiated"
PENDING_DOCS = "pending_documentation"
UNDER_REVIEW = "under_review"
APPROVED = "approved"
DENIED = "denied"
CLOSED = "closed"
class ClaimState(BaseModel):
claim_id: str
user_id: str
status: ClaimStatus = ClaimStatus.INITIATED
policy_number: str | None = None
incident_date: str | None = None
incident_description: str | None = None
documented_damages: list[str] = []
pending_documents: list[str] = []
scheduled_appointments: list[dict] = []
session_count: int = 0
total_turns: int = 0
last_updated: datetime | None = None
def to_context_injection(self) -> str:
"""Format state as a concise context block for injection into prompts."""
lines = [
f"Claim ID: {self.claim_id}",
f"Status: {self.status.value}",
f"Policy: {self.policy_number or 'not yet provided'}",
f"Incident date: {self.incident_date or 'not yet provided'}",
]
if self.documented_damages:
lines.append(f"Documented damages: {', '.join(self.documented_damages)}")
if self.pending_documents:
lines.append(f"PENDING: Customer still needs to submit: {', '.join(self.pending_documents)}")
if self.scheduled_appointments:
for appt in self.scheduled_appointments:
lines.append(f"Scheduled: {appt.get('type')} on {appt.get('date')}")
return "\n".join(lines)
class StateManager:
"""Persist and hydrate structured state from a database."""
def __init__(self, db_pool: asyncpg.Pool):
self.pool = db_pool
async def save_state(self, state: ClaimState):
async with self.pool.acquire() as conn:
await conn.execute("""
INSERT INTO claim_states (claim_id, user_id, state_json, updated_at)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (claim_id) DO UPDATE
SET state_json = $3, updated_at = NOW()
""", state.claim_id, state.user_id, state.model_dump_json())
async def load_state(self, claim_id: str) -> ClaimState | None:
async with self.pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT state_json FROM claim_states WHERE claim_id = $1",
claim_id
)
if row:
return ClaimState.model_validate_json(row["state_json"])
return None
async def update_state_from_llm_extraction(
self,
state: ClaimState,
conversation_turn: str,
extraction_client: AsyncOpenAI
) -> ClaimState:
"""
Use an LLM to extract state updates from the latest conversation turn.
This is the entity extraction step that keeps structured state current.
"""
extraction_prompt = f"""
Analyze this conversation turn and extract any new information that updates the claim state.
Current state: {state.to_context_injection()}
New turn: {conversation_turn}
Respond ONLY with a JSON object containing fields to update.
Only include fields that have new information. Valid fields:
policy_number, incident_date, incident_description,
documented_damages (array), pending_documents (array), status
If nothing new was revealed, respond with: {{}}
"""
response = await extraction_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": extraction_prompt}],
response_format={"type": "json_object"}
)
updates = json.loads(response.choices[0].message.content)
if updates:
updated_data = state.model_dump()
updated_data.update(updates)
updated_data["last_updated"] = datetime.now(timezone.utc)
state = ClaimState(**updated_data)
await self.save_state(state)
return state
The pattern here is separating concerns: conversational memory (what was said) from structured state (what we know). The LLM handles natural language; Pydantic and your database handle the structured representation. You use a lightweight extraction LLM call to bridge between them.
Production enterprise systems rarely use a single memory pattern. The right architecture is a layered stack:
┌─────────────────────────────────────────────┐
│ Layer 4: Structured State (Pydantic/DB) │ ← What we know (facts, status)
├─────────────────────────────────────────────┤
│ Layer 3: Cross-Session Retrieval (Vector) │ ← What happened before
├─────────────────────────────────────────────┤
│ Layer 2: Summarization Buffer │ ← Compressed current session
├─────────────────────────────────────────────┤
│ Layer 1: Hot Window (Full History) │ ← Last N exchanges verbatim
└─────────────────────────────────────────────┘
When building the context for each LLM call, you inject these layers in priority order (least to most recent), with structured state always closest to the system prompt and the hot window immediately preceding the user's message. This layout leverages research on LLM attention patterns: models attend more strongly to the beginning and end of context, so your most critical structured information goes at the top, and your most recent conversation at the bottom.
You need to measure three things in a production memory system: latency, cost per conversation, and coherence quality.
Latency breakdown for a typical turn:
| Operation | P50 | P95 |
|---|---|---|
| Embedding (text-embedding-3-small) | 80ms | 200ms |
| Vector retrieval (pgvector, 100K records) | 15ms | 45ms |
| Summarization call (gpt-4o-mini) | 800ms | 2.1s |
| Primary inference (gpt-4o) | 1.2s | 4.8s |
| State extraction (gpt-4o-mini) | 600ms | 1.8s |
The summarization and state extraction calls are the expensive components. Key optimization: run them asynchronously, decoupled from the user-facing response path. The user gets the primary inference response immediately; summarization and state extraction happen in the background before the next turn.
async def chat_with_background_processing(
self,
user_id: str,
session_id: str,
user_input: str,
turn_index: int
) -> str:
# Critical path: only what's needed to generate the response
messages = await self.build_context(user_id, session_id, user_input)
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=messages
)
assistant_message = response.choices[0].message.content
# Background tasks: storage, summarization, state extraction
# These don't block the response
asyncio.gather(
self.memory_store.store_turn(user_id, session_id, turn_index, "user", user_input),
self.memory_store.store_turn(user_id, session_id, turn_index + 1, "assistant", assistant_message),
self._maybe_summarize(session_id),
self._extract_state_updates(user_id, user_input + "\n" + assistant_message),
return_exceptions=True # Don't let background failures crash the response
)
return assistant_message
Cost modeling: For a 30-turn customer support session using the layered architecture:
Total: ~$0.75 per 30-turn session. Compare to a naive full-history approach at the same turn count: by turn 30, you're sending 15,000+ tokens per call, for a total input cost of roughly $2.25, three times higher.
Enterprise memory systems handle sensitive data — customer PII, financial records, health information. Three concerns dominate:
Encryption at rest and in transit. Your vector store and state database must encrypt at rest. Embeddings are non-trivially reversible: recent research demonstrates that with enough effort, semantic content can be partially reconstructed from embeddings. Don't treat embeddings as inherently anonymized.
Memory isolation. Your retrieval system must enforce strict user-level isolation. A query for user A must never return embeddings from user B's conversations. Implement this at the database query level (the WHERE user_id = $2 clause in our examples), not at the application level alone. Defense in depth: if your application layer has a bug, the database layer should still prevent cross-user contamination.
Right to erasure (GDPR, CCPA compliance). Users may request deletion of all stored data. Your memory architecture must support complete deletion by user ID across all storage layers: the vector store, the summarization store, the state database, and any caches. Build this deletion capability explicitly before you go to production, not as an afterthought.
async def delete_user_data(self, user_id: str):
"""GDPR right-to-erasure implementation."""
async with self.pool.acquire() as conn:
async with conn.transaction():
# Delete conversation memories and embeddings
deleted_memories = await conn.fetchval(
"DELETE FROM conversation_memories WHERE user_id = $1 RETURNING COUNT(*)",
user_id
)
# Delete structured state
deleted_states = await conn.fetchval(
"DELETE FROM claim_states WHERE user_id = $1 RETURNING COUNT(*)",
user_id
)
# Audit log the deletion (keep the audit log, delete the content)
await conn.execute("""
INSERT INTO deletion_audit_log (user_id, deleted_at, records_deleted)
VALUES ($1, NOW(), $2)
""", user_id, (deleted_memories or 0) + (deleted_states or 0))
Retention policies. Not all memory should be retained indefinitely. Raw conversation transcripts may need to be deleted after 90 days by policy, while structured claim state needs to be kept for seven years for regulatory reasons. Build TTL awareness into your storage schema from day one.
Build a multi-turn research assistant with layered memory for a financial analyst workflow. The assistant should:
aiosqlite) so conversations survive application restartsStep 1: Set up the database schema. Create tables for conversation_messages (session_id, turn_index, role, content, timestamp) and analyst_state (session_id, research_question, companies JSON array, key_findings JSON array, open_questions JSON array, updated_at).
Step 2: Implement a ResearchSessionBuffer class that combines a SummarizationBuffer (from the lesson) with your structured state. The state should be updated via an LLM extraction call after each turn.
Step 3: Write the system prompt. It should define the analyst assistant's behavior and include a placeholder {state_block} that you fill with the current structured state before each API call.
Step 4: Test coherence across a simulated 20-turn session. Start with a research question about comparing two bank stocks (pick any two: JPM vs. BAC, for example). Introduce specific financial data points in early turns (P/E ratios, dividend yields, recent earnings). In turn 15, ask the assistant to summarize what it knows about the earlier turns' data. Verify that the structured state correctly captured the data points, and that the summary buffer preserved the context even though those early turns have been compressed.
Stretch goal: Add a session restore feature. Save the complete state to disk, restart your application, load the state, and verify the assistant correctly resumes the conversation with full context.
Mistake 1: Letting the full conversation accumulate in the database but serving summarized context to the model, without realizing the gap.
You'll encounter this when you're debugging a "why did the model forget X?" issue and discover that X was in turn 4, which was summarized, and the summarization dropped it. Solution: always log what context is actually being sent to the model. Build a debug_context() method that outputs the exact messages array before every API call during development.
Mistake 2: Summarizing too aggressively. Compressing 20 turns into a 300-word summary loses too much. The model's comprehension of nuanced prior context degrades sharply. Rule of thumb: aim for a compression ratio of no more than 5:1. If 20 turns generate 6,000 tokens of conversation, your summary should be 1,000–1,200 tokens, not 300.
Mistake 3: Not accounting for parallel sessions. Enterprise applications have many users, many sessions. If you're storing conversation state in memory (a Python dict keyed by session ID) rather than a database, you'll hit a race condition when horizontally scaling across multiple app instances. Every stateful component must be backed by an external store (Redis, PostgreSQL) from day one.
Mistake 4: Conflating the session ID with the user ID. A user can have multiple sessions. Your retrieval and state management systems need both identifiers. A common bug: using session_id as the primary key for structured state, which means returning users lose their persistent state when they start a new session.
Mistake 5: Using the same model for summarization and inference without prompting it differently. If you use GPT-4o for both tasks with a generic prompt, you'll get verbose, LLM-flavored summaries that waste tokens on hedging language and explanatory prose. Summarization prompts need to be explicitly terse and structured. Test your summarization prompt in isolation with real conversation samples before integrating it into the pipeline.
Troubleshooting context coherence issues: When the model "forgets" something that should be in context, instrument your pipeline with a context audit:
def audit_context(messages: list[dict], target_term: str) -> bool:
"""Check whether a specific term appears anywhere in the context."""
full_context = " ".join(m["content"] for m in messages)
found = target_term.lower() in full_context.lower()
if not found:
print(f"AUDIT: '{target_term}' not found in {len(messages)} messages "
f"({sum(len(m['content']) for m in messages)} chars)")
return found
Call this before every API call during debugging. Nine times out of ten, a "the model forgot X" complaint will reveal that X was never in the context at all — it was evicted, never summarized, or stored under a different session ID.
You've covered the full stack of LLM memory architecture, from the physical constraints of token context windows through four distinct memory patterns and into the complexities of enterprise-grade persistent, multi-session, multi-user systems.
The key architectural principles to carry forward:
Where to go from here:
The difference between an LLM application that feels like a toy and one that operates reliably at enterprise scale often comes down entirely to memory architecture. You now have the foundation to build the latter.
Learning Path: Intro to AI & Prompt Engineering