Persist LangGraph agent state correctly with MemorySaver and PostgresSaver —
thread_id discipline, JSON-serializable state rules, time-travel, schema
migration. Use when adding chat memory, migrating from ConversationBufferMemory,
or time-traveling an agent state to debug an incident.
Trigger with "langgraph checkpointer", "MemorySaver", "PostgresSaver",
"thread_id", "langgraph time travel", "langgraph state persistence".
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Persist LangGraph agent state correctly with MemorySaver and PostgresSaver —
thread_id discipline, JSON-serializable state rules, time-travel, schema
migration. Use when adding chat memory, migrating from ConversationBufferMemory,
or time-traveling an agent state to debug an incident.
Trigger with "langgraph checkpointer", "MemorySaver", "PostgresSaver",
"thread_id", "langgraph time travel", "langgraph state persistence".
Designed for Claude Code, also compatible with Codex
LangGraph Checkpointing (Python)
Overview
A chat agent that "keeps introducing itself" is almost always P16. The caller
invokes graph.invoke(state) without passing config={"configurable": {"thread_id": ...}} — LangGraph's checkpointer silently spawns a fresh state per
call. No error, no warning, no log line. The user sees it; the code does not.
That is one of five separate checkpointing pitfalls this skill covers:
P16 — missing thread_id silently resets memory
P17 — interrupt_before raises TypeError when state holds non-JSON values
(datetime, Decimal, custom classes) — and it raises at the interrupt
boundary, not when the bad value was first assigned, so the traceback points
at the wrong line
P20 — PostgresSaver does not auto-migrate checkpoint schema; upgrading
langgraph silently reads old checkpoints as empty state
P40 — ConversationBufferMemory and the rest of legacy chat memory were
removed in LangChain 1.0; checkpointers are the replacement
P51 — Deep Agent virtual-FS state in state["files"] grows unboundedly
and eventually makes checkpoint writes a latency hotspot
This skill walks through picking a checkpointer by environment, enforcing
thread_id at the application boundary, constraining state to JSON-safe
primitives, Postgres setup + migration, and time-travel for incident debugging.
Pinned to langgraph >= 1.0, < 2.0, langgraph-checkpoint-postgres >= 1.0, < 2.0. Pain-catalog anchors: P16, P17, P18, P20, P22, P40, P51.
For Postgres: pip install langgraph-checkpoint-postgres and a Postgres 13+
instance
For async Postgres: the same package plus asyncpg
A thread_id strategy — typically a UUID4 string per conversation; see
thread-id-discipline.md
Instructions
Step 1 — Pick a checkpointer by environment
Env
Checkpointer
Import
Dev, tests, notebooks
MemorySaver
langgraph.checkpoint.memory
Single-host CLI / desktop
SqliteSaver
langgraph.checkpoint.sqlite
Staging, prod (sync)
PostgresSaver
langgraph.checkpoint.postgres
Staging, prod (async / FastAPI)
AsyncPostgresSaver
langgraph.checkpoint.postgres.aio
MemorySaver is in-process only. State vanishes on restart. Every worker has
its own (P22 analog for LangGraph). Use it anywhere state loss is acceptable;
never in a multi-worker web backend.
PostgresSaver and its async sibling require setup() on every startup and
after every langgraph upgrade (see Step 5). Checkpoint storage overhead is
typically 1-10 KB per step of serialized state; plan your DB size
accordingly — a 2,000-turn conversation with 3 KB average state fits in
~6 MB per thread.
See checkpointer-comparison.md for the
full matrix including latency, concurrency, and the FastAPI lifespan pattern.
Step 2 — Require thread_id at every invocation
This is the fail-loud middleware that prevents P16:
from typing importAnydefrequire_thread_id(config: dict[str, Any]) -> dict[str, Any]:
"""Raise if thread_id is missing. Fails loud so P16 surfaces in tests,
not in user-visible conversation logs."""
configurable = (config or {}).get("configurable", {})
thread_id = configurable.get("thread_id")
ifnot thread_id:
raise ValueError(
"thread_id missing from config['configurable']. ""Every graph invocation must carry a thread_id."
)
ifnotisinstance(thread_id, str):
raise TypeError(
f"thread_id must be str (UUID), got {type(thread_id).__name__}"
)
return config
For web endpoints, extract to a FastAPI dependency (Header(...) with no
default — forces 422 on missing). For multi-tenant apps, scope the thread id by composing tenant + user +
conversation ids into a single colon-delimited string (example:
"acme:alice:conv-1"). See
thread-id-discipline.md for UUID
generation, rotation, and the integration test that proves tenants do not share
state.
Step 3 — Keep state JSON-serializable (TypedDict, primitives only)
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage
classAgentState(TypedDict):
# Messages are safe — LangGraph registers a custom serializer.
messages: Annotated[list[AnyMessage], add_messages]
# Primitives only below. NO datetime, NO Decimal, NO custom classes.
user_id: str
turn_count: int
last_action_at: str# ISO string, not datetime.datetime
pending_approval: bool
metadata: dict[str, str] # dict keys must be str
plan: list[dict[str, str]] # list of primitive dicts
The rule: state fields must be JSON-safe primitives or recursive structures
of them (str, int, float, bool, None, list, dict[str, ...]).
json.dumps(state) must succeed. If it raises, the checkpointer raises —
often at a HITL interrupt many steps later (P17), which is why the traceback
never points at the line that introduced the bad value.
For non-primitive inputs, coerce at node output boundaries with a helper:
from datetime import datetime
from decimal import Decimal
defrecord_purchase(state: AgentState) -> dict:
now = datetime.utcnow()
price = Decimal("19.99")
return {
"last_action_at": now.isoformat(),
"metadata": {**state["metadata"], "price": str(price)},
}
Step 4 — Compile the graph with a checkpointer (Postgres, sync)
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph
import os
DB_URI = os.environ["DATABASE_URL"]
defbuild_graph() -> StateGraph:
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("human_approval", human_approval_node)
builder.set_entry_point("agent")
builder.add_edge("agent", "human_approval")
builder.set_finish_point("human_approval")
return builder
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # Idempotent. Creates checkpoint tables if missing.
graph = build_graph().compile(
checkpointer=checkpointer,
interrupt_before=["human_approval"],
)
config = {"configurable": {"thread_id": "user-123"}}
require_thread_id(config)
result = graph.invoke({"messages": [HumanMessage("hi")]}, config=config)
For async, mirror the pattern with AsyncPostgresSaver.from_conn_string(...)
inside a FastAPI @asynccontextmanager lifespan; every call site uses
await graph.ainvoke(...).
Step 5 — Run setup() on startup AND after every langgraph upgrade
P20 is the quiet one: you pip install --upgrade langgraph, tests pass, CI
goes green, you deploy. Existing threads come back empty. No DB error.
# Put this in your deploy script / migration runbook:with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
# Sanity check: read one known thread and assert it's not empty.
snap = checkpointer.get({"configurable": {"thread_id": "canary-thread"}})
assert snap isnotNone, "Canary thread lost after schema migration"
Run this in staging first, with a canary thread whose state you pre-populated
from an older langgraph version. If the assertion holds, promote. If not,
the migration path is: dump checkpoints, upgrade, restore via a migration
script. Do not promote to production on a version bump without this check.
Step 6 — Time-travel for incident debugging
Every checkpoint is keyed by (thread_id, checkpoint_id) and reachable via
graph.get_state_history(config):
config = {"configurable": {"thread_id": bad_thread_id}}
history = list(graph.get_state_history(config))
for i, snap inenumerate(history):
print(i, snap.metadata.get("step"), snap.next, list(snap.values.keys()))
# Resume from a specific prior checkpoint — None as input = "use this state":
past = history[3] # history is newest-first
result = graph.invoke(None, config=past.config) # past.config pins checkpoint_id
The original branch is preserved — update_state creates a new checkpoint
alongside the old one. Useful for forensics and for A/B comparing an old vs
new model on the exact same state.
See time-travel-and-replay.md for the
full incident playbook, branching for A/B eval, and checkpoint pruning SQL.
Output
A checkpointer selected by environment, not copy-pasted from a tutorial
A require_thread_id middleware that fails loud on missing thread_id, and
at least one integration test that asserts two tenants do not share state
An AgentState TypedDict constrained to JSON-safe primitives, plus a
to_state helper at node output boundaries for datetime / Decimal /
Pydantic / enums
PostgresSaver.setup() wired into startup and every deploy that bumps
langgraph
An incident runbook that uses get_state_history + update_state to
time-travel an agent state, with prior branches preserved
Error Handling
Error
Cause
Fix
Agent forgets every turn, no error logged
Missing thread_id (P16)
Add require_thread_id middleware at app boundary; FastAPI Header(...) with no default
TypeError: Object of type datetime is not JSON serializable at a HITL pause
Non-JSON value in state (P17)
Coerce at node output: dt.isoformat(), str(decimal), model.model_dump(mode="json")
Upgrading langgraph returns empty state for existing threads
Checkpoint schema drift (P20)
PostgresSaver.setup() after every upgrade; canary-thread assertion in staging before prod
ImportError: cannot import name 'ConversationBufferMemory'
Legacy memory removed in 1.0 (P40)
Migrate to MemorySaver/PostgresSaver + thread_id per conversation
Checkpoint writes take 500+ ms on Deep Agent runs
Unbounded state["files"] growth (P51)
Add cleanup node that prunes state["files"] older than N steps
Use await graph.ainvoke(...) everywhere, or switch to sync PostgresSaver
Examples
Minimal chat agent with MemorySaver
Dev-mode multi-turn chat that survives within a single process. Good for local
iteration and pytest fixtures; state vanishes on restart.
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic
checkpointer = MemorySaver()
agent = create_react_agent(
model=ChatAnthropic(model="claude-sonnet-4-6"),
tools=[...],
checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "dev-session-1"}}
agent.invoke({"messages": [HumanMessage("hi")]}, config=config)
agent.invoke({"messages": [HumanMessage("remember my name is alice")]}, config=config)
agent.invoke({"messages": [HumanMessage("what's my name?")]}, config=config) # alice
Multi-tenant Postgres + FastAPI lifespan
Async production shape. One AsyncPostgresSaver pool, thread-id extracted from
a required header, tenants isolated by a composite key.
See checkpointer-comparison.md for the
complete FastAPI lifespan pattern and pool sizing advice (max_size needs to
exceed concurrent graph count, and the LangGraph pool should be separate from
the application's primary DB pool).
Migrating from ConversationBufferMemory (P40)
Legacy 0.x pattern replaced end-to-end:
# OLD — raises ImportError on 1.0:# from langchain.memory import ConversationBufferMemory# memory = ConversationBufferMemory()# chain = LLMChain(llm=llm, memory=memory)# NEW:from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URI) as cp:
cp.setup()
agent = create_react_agent(model=llm, tools=tools, checkpointer=cp)
config = {"configurable": {"thread_id": user_session_id}}
agent.invoke({"messages": [HumanMessage(user_input)]}, config=config)
The thread_id is now the "session key" that ConversationBufferMemory used
to carry implicitly.
Incident time-travel
Production returned a wrong answer for thread_id=bad-thread at 14:03.
Walk history newest-first, inspect the prior state, patch and replay:
config = {"configurable": {"thread_id": "bad-thread"}}
history = list(graph.get_state_history(config))
for i, snap inenumerate(history):
print(i, snap.metadata.get("step"), snap.next)
# Assume step 7 wrote the bad output; step 6 is the input.
prior = history[-7] # or index by metadata["step"]print("input to bad node:", prior.values)
graph.update_state(prior.config, {"retry_count": 0}, as_node="validator")
graph.invoke(None, config=prior.config) # new branch with correct result
Full playbook (finding thread_id in logs, inspecting writes metadata,
pruning history) in
time-travel-and-replay.md.
