GenAI 2026 · Module 10 · Notebook 11

Memory Architecture
for Production Chatbots

Every notebook so far answers one question, forgets everything, and dies.
A real product has to remember, protect, control cost, and survive failures.
This deck builds each capability from first principles.

7

Production requirements every chatbot must solve

4

Distinct memory strategies with different tradeoffs

1

Unified API — create_agent for all of it

∞

Composable middlewares — add as many as you need

langchainlanggraphcreate_agentInMemorySaverSqliteSaverInMemoryStoreinterrupt()

The Problem

Every call starts from zero

A bare create_agent with no memory config is 100% stateless.
Two consecutive calls share absolutely nothing — just like every raw LLM API call you've ever made.

# Turn 1 — works fine
r1 = bare_agent.invoke({"messages": "Hi, my name is Noordeen."})
# → "Hi Noordeen! How can I help?"

# Turn 2 — blank slate
r2 = bare_agent.invoke({"messages": "What's my name?"})
# → "I'm sorry, I don't know your name."  ← no shared state at all

WITHOUT MEMORY

Each .invoke() is an isolated API call. The model sees only what you pass in that single call. No history, no context, no continuity.

✗ Forgets your name instantly

✗ Can't refer back to anything

✗ Every turn is turn 1

WITH MEMORY

A checkpointer= snapshots state after every turn. The same thread_id on the next call resumes exactly where it left off.

✓ Full conversation history

✓ Cross-turn references work

✓ Multiple users, zero interference

Real-world consequence: Amazon Alexa, Google Assistant, any customer-support bot — none of them would be usable without persistent memory. "I just told you my order number!" is the #1 user complaint in stateless chatbot deployments.

Memory Type 1

Short-Term Memory
One conversation, remembered

What: The conversation history for a single chat session.
Where: Lives in a checkpointer, scoped to a thread_id.
Why: Every follow-up question ("What about the second option?") needs context from earlier in the same chat.

from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
  model=MODEL, tools=[],
  checkpointer=InMemorySaver()  # ← this line
)
# Same thread_id = same conversation
config_a = {"configurable": {"thread_id": "chat-a"}}
agent.invoke({"messages": "My name is Noordeen."}, config_a)
agent.invoke({"messages": "What's my name?"}, config_a)
# → "Your name is Noordeen."  ✓

# Different thread_id = completely separate chat
config_b = {"configurable": {"thread_id": "chat-b"}}
agent.invoke({"messages": "What's my name?"}, config_b)
# → "I don't know your name."   ← fresh slate

HOW IT WORKS

1

User sends turn 1

Checkpointer snapshots state, keyed by thread_id

2

User sends turn 2 (same thread_id)

Checkpointer loads snapshot → model sees full history

3

Different user (new thread_id)

Zero state shared — completely independent

⚠ InMemorySaver limitation: lives only in RAM. Dies the moment the Python process exits. Fine for a notebook, not for production. Section 3 fixes this.

🎯 Check your understanding

Two users are chatting with the same agent object simultaneously. User A's thread_id is "alice-session", User B's is "bob-session". User B types "What did User A just say?" — what happens?

Memory Type 1b

Persistent Memory
Survives process restarts

What: Short-term memory backed by a database instead of RAM.
Where: SqliteSaver for local apps, PostgresSaver/MongoDBSaver for scale.
Why: A user closes their laptop and comes back tomorrow. Their conversation must still be there.

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# Process #1 — agent stores a fact
conn = sqlite3.connect("chat.db", check_same_thread=False)
agent_v1 = create_agent(model=MODEL, tools=[], checkpointer=SqliteSaver(conn))
agent_v1.invoke({"messages": "My favorite language is Python."}, config)
del agent_v1  # ← simulate process restart

# Process #2 — brand new object, same SQLite file
conn2 = sqlite3.connect("chat.db", check_same_thread=False)
agent_v2 = create_agent(model=MODEL, tools=[], checkpointer=SqliteSaver(conn2))
r = agent_v2.invoke({"messages": "What's my favorite language?"}, config)
# → "Your favorite language is Python."  ✓ survived the restart

Dev / Notebook

InMemorySaver

RAM only. Zero setup. Dies on process exit. Use this in notebooks and tests.

RAMNo persistence

Small Apps

SqliteSaver

Local SQLite file. Survives restarts. One pip install. Perfect for a local product or prototype.

File on diskSurvives restart

Production Scale

PostgresSaver

Real database. Handles concurrent users, millions of conversations. Same code, different class name.

Postgres / MongoScale

Key insight: Swapping from InMemorySaver to SqliteSaver to PostgresSaver is a one-word change — only the class name changes. The rest of your code is identical. This is the entire design point of the checkpointer abstraction.

Cost Control

Token Budget
Don't let cost grow forever

The problem: Every stored message gets re-sent to the model on every turn. A 200-turn conversation sends 200 messages per call. Cost and latency grow linearly — the #1 production cost bug in chat systems.

TOKENS SENT PER TURN (unbounded history)

Turn 1

~50

50 tok

Turn 10

~500

500 tok

Turn 50

~2500

2.5K tok

Turn 200

~10,000

10K tok

OPTION A — SUMMARIZATION

SummarizationMiddleware(
  model=MODEL,
  trigger=("tokens", 500),  # fires when >500 tokens
  keep=("messages", 4),    # keep last 4 verbatim
)
# Old turns → LLM-generated summary
# Latest N turns → kept verbatim
# Old facts survive inside the summary ✓

Cost: One extra LLM call when it fires. Benefit: Old facts survive — the summary carries them.

OPTION B — TRIMMING

@before_model
def keep_last_n(state, runtime):
  N = 6
  if len(state["messages"]) <= N:
    return None
  return {"messages": [
    RemoveMessage(id=REMOVE_ALL_MESSAGES),
    *state["messages"][-N:]
  ]}
# Old turns simply dropped — gone forever
# Zero extra LLM calls — cheap ✓
# But: "My name is Noordeen" from turn 1 vanishes ✗

Cost: Zero extra calls. Risk: Facts in dropped messages are permanently gone.

Strategy	Extra LLM calls?	Old facts survive?	Use when
Summarization	Yes (when triggered)	Yes — compressed	Facts must survive; cost is secondary
Trimming	No	No — dropped	Conversation is episodic; old turns are disposable

Memory Type 2

Long-Term Memory
Remembers you across sessions

What: Facts about a user that persist across entirely separate conversations, days apart.
Where: An InMemoryStore (or persistent backend), keyed by user_id, searched semantically.
Why: "Alexa, set my usual alarm" only works if Alexa remembers what "usual" means from a conversation 6 months ago.

# 1. Create a vector-indexed Store
store = InMemoryStore(index={
  "embed": init_embeddings("openai:text-embedding-3-small"),
  "dims": 1536
})

# 2. Inject memories via @dynamic_prompt middleware
@dynamic_prompt
def personalized_prompt(request) -> str:
  user_id = request.runtime.context.user_id
  memories = request.runtime.store.search(
    ("memories", user_id),
    query="user preferences", limit=3
  )
  facts = "\n".join(m.value["text"] for m in memories)
  return "You are helpful." + (f"\nFacts:\n{facts}" if facts else "")

# 3. Seed a fact from "yesterday's session"
store.put(("memories", "user-42"), "fact-1", {
  "text": "The user's favorite language is Rust."
})

# 4. Brand-new thread — never seen this conversation before
r = memory_agent.invoke(
  {"messages": "What's my favorite language?"},
  {"configurable": {"thread_id": "new-chat"}},
  context=Context(user_id="user-42"),
)
# → "Your favorite language is Rust."  ✓
# Came from the Store, not the checkpointer!

TWO MEMORY SYSTEMS

	Checkpointer	Store
Keyed by	thread_id	user_id (or anything)
Scope	One conversation	Cross-session
Survives new chat?	No	Yes
Use case	"What did we just say?"	"Who is this user?"

Search is semantic: When you call store.search(query="user preferences"), it does vector similarity search — the same embeddings from notebook 04. A stored fact "loves Rust" matches a query "programming language preference" even though the words differ.

The store holds facts about user-42. Fire a query — it does vector similarity, so it matches on meaning, not exact words. Watch which facts light up.

query: (none yet)

→ embedded to a vector, compared against every stored fact by cosine similarity.

🦀 Favorite language is Rust—

🍕 Loves margherita pizza—

💼 Works as a data engineer—

⚙️ Prefers static typing—

🎯 Check your understanding

A user chats with your bot on Monday (thread "mon-chat"), seeds a fact about their job title. On Friday they open a brand-new chat (thread "fri-chat"). Where does the job title fact come from?

UX

Streaming
Don't make users stare at a blank screen

What: Yield tokens as they are generated, one by one, instead of waiting for the full response.
Where: .stream(..., stream_mode="messages") — same agent, same middleware, different call.
Why: Perceived latency drops dramatically. GPT-4o takes ~2–4 seconds for a long answer. Without streaming, users see nothing for that whole time.

import time

for chunk, metadata in memory_agent.stream(
  {"messages": "Why does streaming matter?"},
  {"configurable": {"thread_id": "stream-demo"}},
  context=Context(user_id="user-42"),
  stream_mode="messages",   # ← one tuple per token
):
  if chunk.content:
    print(chunk.content, end="", flush=True)
    time.sleep(0.05)  # 50ms per token (demo only)
print()              # remove sleep in production

Each (chunk, metadata) tuple carries one token fragment. chunk.content is the text; metadata has node name, run IDs, etc. The sleep makes it visually obvious in a notebook — remove it in production.

BEFORE STREAMING (blocking)

User types → blank screen → blank screen → blank screen → full answer appears. 3-4 seconds of nothing.

Note on astream_events: There's also astream_events(version="v3") for event-level streaming (tool calls, node transitions). It's marked experimental and may throw NotImplementedError. Use stream_mode="messages" by default.

Safety

Guardrails
Stop sensitive data and unsafe content

Two separate concerns — both implemented as middleware in the same list, running before the model ever receives the message.

PII FILTERING

PIIMiddleware("email",
  strategy="redact",
  apply_to_input=True),
PIIMiddleware("credit_card",
  strategy="mask",
  apply_to_input=True),

# Input:  "My email is john@ex.com and card 4111..."
# Model sees: "My email is [REDACTED_EMAIL]
#              and card **** **** **** 1111"
# Real email/card never left this process ✓

Deterministic, no LLM involved. PIIMiddleware runs pure regex patterns — fast, cheap, 100% predictable. No model call to redact data.

CONTENT MODERATION

@wrap_model_call
def moderate_input(request, handler):
  last = request.messages[-1].content
  result = client.moderations.create(
    model="omni-moderation-latest",
    input=last
  )
  if result.results[0].flagged:
    return AIMessage(content="I can't help with that.")
  return handler(request)  # pass through if safe

# "recipe for banana bread" → passes through ✓
# "how to build a bomb"    → blocked ✗

@wrap_model_call intercepts the call before it reaches the chat model. If flagged, it returns a safe response directly — the expensive model call never happens.

① What the user typed My email is john.doe@example.com and my card is 4111 1111 1111 1111.

↓ PIIMiddleware (regex, no LLM) ↓

② What the model actually receives My email is john.doe@example.com and my card is 4111 1111 1111 1111.

🎯 Check your understanding

A user types "My SSN is 123-45-6789, help me fill a form." PIIMiddleware has strategy="redact" for SSN. What does the model actually receive?

Safety

Human-in-the-Loop
Pause before anything risky

What: LangGraph's interrupt() pauses graph execution inside a tool, before the risky operation, and waits for an explicit human decision.
Where: Inside the tool function itself — not a middleware, but the checkpointer makes it durable.
Why: Sending an email, issuing a refund, deleting an account — these cannot be undone. No LLM should do them unilaterally.

Phase 1 — Pause

Agent decides → tool pauses

Model calls send_email. Tool hits interrupt(). Graph snapshots state and returns. Result has __interrupt__ key with approval payload.

Phase 2 — Resume

Human decides → execution continues

Command(resume="approve") or "reject". Same thread_id. Execution resumes at the exact interrupt() call site inside the tool.

from langgraph.types import Command, interrupt

def send_email(recipient: str, subject: str, body: str) -> str:
  # interrupt() is placed BEFORE the risky operation
  decision = interrupt({
    "question": "Approve sending this email?",
    "to": recipient, "subject": subject, "body": body,
  })
  # Execution RESUMES HERE after Command(resume=...) is called
  if decision == "approve":
    return f"Email sent to {recipient}"
  return f"Rejected — not sent."

# Phase 1 — triggers interrupt()
config = {"configurable": {"thread_id": "approval-demo"}}
result = hitl_agent.invoke(
  {"messages": "Send email to bob@example.com..."}, config)
# result["__interrupt__"] has the approval payload

# Phase 2 — human approves (same config!)
approved = hitl_agent.invoke(Command(resume="approve"), config)
# → "Email sent to bob@example.com"

WHY INSIDE THE TOOL?

interrupt() placement is intentional: before the dangerous operation. The gate is part of the tool contract, not bolted on as an afterthought.

SAME thread_id = SAME PAUSE

The checkpointer holds the paused state. As long as you pass the same thread_id, the graph resumes from exactly where interrupt() stopped — even hours later.

REJECT PATH

Command(resume="reject") → tool returns the rejection message → model receives it and apologises. The risky operation never ran.

Reliability

Resilience
Model providers fail — handle it gracefully

Three composable middlewares for three distinct failure modes. Stack them in the same list.

1

ModelRetryMiddleware

transient errors: rate limits, 5xx, timeouts → exponential backoff

↓

2

ModelFallbackMiddleware("gpt-4o-mini")

primary model fully down → cascade to backup silently

↓

3

ToolRetryMiddleware(tools=["search","fetch_url"])

flaky tools only — never retries send_email or delete

# Primary model is deliberately broken — fallback fires
resilient_agent = create_agent(
  model="openai:gpt-does-not-exist-9999",  # ← nonexistent
  tools=[],
  middleware=[
    ModelRetryMiddleware(max_retries=2, backoff_factor=2.0),
    ModelFallbackMiddleware("openai:gpt-4o-mini"),
  ],
)
r = resilient_agent.invoke({"messages": "Say: fallback-worked"})
# → "fallback-worked"  ✓
# User saw zero errors. The fallback fired silently.

# ToolRetryMiddleware — only named tools, not everything
ToolRetryMiddleware(
  max_retries=2,
  tools=["search", "fetch_url"],  # NEVER include send_email!
  retry_on=(TimeoutError,),
)

Why three separate middlewares? Retry, fallback, and tool retry are different failure modes with different policies. Retrying a failed HTTP call 3 times is fine. Retrying send_email 3 times because of a timeout would send 3 emails. Keep them separate so you can tune them independently.

Composition

One Production Agent
All 7 capabilities, one create_agent call

Every middleware goes in the same list. Order matters — run the cheap, deterministic checks first, then personalization, then cost control, then resilience.

production_agent = create_agent(
  model="openai:gpt-does-not-exist-9999",  # broken — fallback saves it
  tools=[send_email],
  context_schema=Context,
  checkpointer=SqliteSaver(conn),# short-term + durablestore=store,# long-term, cross-session
  middleware=[
    # ① Deterministic, cheap — run firstPIIMiddleware("email", strategy="redact", apply_to_input=True),
    moderate_input,         # @wrap_model_call# ② Personalization from long-term Store
    personalized_prompt,    # @dynamic_prompt# ③ Cost controlSummarizationMiddleware(
      model=MODEL, trigger=("tokens", 2000), keep=("messages", 10)),

    # ④ Resilience — last, wraps the actual model callModelRetryMiddleware(max_retries=2),
    ModelFallbackMiddleware("openai:gpt-4o-mini"),

    # HITL: interrupt() lives inside send_email already — no middleware needed
  ],
)


  MIDDLEWARE EXECUTION ORDER (per turn)
①
PIIMiddleware + moderate_input
deterministic, no LLM cost
↓
②
personalized_prompt
Store lookup → inject user facts into system prompt
↓
③
SummarizationMiddleware
token check → summarize if needed
↓
④
ModelRetryMiddleware + ModelFallbackMiddleware
wraps the actual model call
↓
⑤
LLM responds
checkpointer snapshots state
Adding a new capability: Write a new middleware function or tool, add it to the list — zero changes to everything else. This is the entire design point.

Summary

What you can now build

Seven production requirements, seven solutions, all composed through a single create_agent call.

Requirement	Solution	Old (deprecated)
Remember this conversation	`checkpointer=` + `thread_id`	ConversationBufferMemory
Survive restarts	`SqliteSaver` → `PostgresSaver`	custom persistence
Cost won't grow forever	`SummarizationMiddleware` or trimming	ConversationSummaryMemory
Remember user across sessions	`store=` + `@dynamic_prompt`	global dict / fine-tune
Stream tokens to UI	`.stream(..., stream_mode="messages")`	—
No PII/unsafe content leaks	`PIIMiddleware` + `@wrap_model_call`	—
Human approval on risky tools	`interrupt()` + `Command(resume=...)`	—
Provider outages handled	`ModelRetryMiddleware` + `ModelFallbackMiddleware`	—

WHAT'S NEXT

This chatbot can hold a conversation, remember its user, protect itself, and recover from failure — but it still answers from what the model already knows.

Wire the retrieval pipeline from notebooks 05–08 into a tool this agent can call, and you have Conversational RAG: a chatbot that remembers and knows your documents.

That's the capstone.

🎯 Final question

You're building a customer support bot. A user's conversation gets long — 300 turns. Which memory strategy keeps the user's account number (mentioned in turn 1) available at turn 300?

Memory Architecturefor Production Chatbots

Every call starts from zero

Short-Term MemoryOne conversation, remembered

Persistent MemorySurvives process restarts

Token BudgetDon't let cost grow forever

Long-Term MemoryRemembers you across sessions

StreamingDon't make users stare at a blank screen

GuardrailsStop sensitive data and unsafe content

Human-in-the-LoopPause before anything risky

ResilienceModel providers fail — handle it gracefully

One Production AgentAll 7 capabilities, one create_agent call

What you can now build

Memory Architecture
for Production Chatbots

Short-Term Memory
One conversation, remembered

Persistent Memory
Survives process restarts

Token Budget
Don't let cost grow forever

Long-Term Memory
Remembers you across sessions

Streaming
Don't make users stare at a blank screen

Guardrails
Stop sensitive data and unsafe content

Human-in-the-Loop
Pause before anything risky

Resilience
Model providers fail — handle it gracefully

One Production Agent
All 7 capabilities, one create_agent call