Skip to content

Persistence & Checkpoints

1. Why this matters

A stateless graph forgets everything between .invoke() calls. You can't build a chatbot with that — the user says "What about tomorrow?" and the bot has no idea what they're following up on.

Checkpointing makes state durable per thread, which unlocks:

  • Persistent conversations keyed by thread_id.
  • Crash recovery — process dies, restart, resume.
  • Time-travel debugging — list all checkpoints, replay from any one.
  • Human-in-the-loop — pause, ask a human, resume.
  • Branching alternate runs — fork from a past checkpoint to try a different path.

2. Mental model

Think of a checkpointer as the graph's save game mechanism:

flowchart LR
    I[Initial state] --> N1[Node 1 runs]
    N1 -->|"snapshot saved (thread t1)"| CP1[Checkpoint 1]
    N1 --> N2[Node 2 runs]
    N2 -->|snapshot| CP2[Checkpoint 2]
    N2 --> END([Final])
    END -->|next invoke same thread_id| LOAD[Load latest checkpoint]
    LOAD --> N3[Continue from there]

Every invocation with the same thread_id: 1. Loads the latest checkpoint as the starting state. 2. Merges your .invoke(...) input on top. 3. Runs, saving checkpoints along the way.

3. Architecture / Flow

flowchart TD
    A[graph.invoke<br/>thread_id=t1] --> B{Latest checkpoint<br/>for t1?}
    B -->|exists| C[Load state from checkpoint]
    B -->|none| D[Start fresh from input]
    C --> E[Run nodes]
    D --> E
    E --> F[Save snapshot after each super-step]
    F --> G[Final state returned]
    F --> H[Checkpointer store<br/>InMemory / SQLite / Postgres]

4. Core concepts

  • Checkpointer — pluggable backend that persists snapshots.
  • MemorySaver — RAM, ephemeral. Dev only.
  • SqliteSaver — single-file SQLite. Local apps, single-machine.
  • PostgresSaver — production multi-process. From langgraph-checkpoint-postgres.
  • Thread — a logical session, identified by thread_id. State is keyed by thread.
  • Checkpoint — one snapshot in the thread's history. Each super-step produces one.
  • config={"configurable": {"thread_id": ...}} — pass on every invoke to scope to a thread.
  • graph.get_state(config) — current state snapshot for a thread.
  • graph.get_state_history(config) — generator over every checkpoint, newest first.
  • graph.update_state(config, values, as_node=...) — surgically modify state mid-run (essential for HITL — see Chapter 10).

5. Code — minimal working example

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

llm = ChatOpenAI(model="gpt-4o-mini")

def chat(state: ChatState):
    return {"messages": [llm.invoke(state["messages"])]}

b = StateGraph(ChatState)
b.add_node("chat", chat)
b.add_edge(START, "chat")
b.add_edge("chat", END)

graph = b.compile(checkpointer=MemorySaver())   # ← the only change to enable memory

cfg = {"configurable": {"thread_id": "alice-1"}}

# Turn 1
graph.invoke({"messages": [HumanMessage("Hi, I'm Alice.")]}, config=cfg)
# Turn 2 — same thread, remembers Alice
out = graph.invoke({"messages": [HumanMessage("What's my name?")]}, config=cfg)
print(out["messages"][-1].content)   # "Your name is Alice."

6. Code — real-world pattern

Production-ready setup with SQLite (persists across process restarts):

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("chat_history.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

graph = builder.compile(checkpointer=checkpointer)

# Different users, fully isolated histories
alice_cfg = {"configurable": {"thread_id": "user_alice"}}
bob_cfg   = {"configurable": {"thread_id": "user_bob"}}

Postgres for multi-process / horizontal scaling:

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pwd@host:5432/dbname?sslmode=require"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()       # create tables (run once)
    graph = builder.compile(checkpointer=checkpointer)
    # … use graph …

Inspecting state for a thread:

# Current snapshot
snap = graph.get_state(cfg)
print(snap.values)              # full state dict
print(snap.next)                # which node(s) would run next
print(snap.config)              # includes checkpoint_id

# Full history (newest first)
for s in graph.get_state_history(cfg):
    print(s.created_at, "→", s.values.get("messages", [])[-1].content[:60])

Time-travel: replay from an earlier checkpoint:

history = list(graph.get_state_history(cfg))
earlier = history[3]                       # 4 turns back

# Resume from that checkpoint by passing its config
graph.invoke(None, config=earlier.config)
# Or branch: edit and resume
graph.update_state(earlier.config, {"messages": [HumanMessage("Try this instead.")]})

Clear a thread's history (e.g., user clicks "Clear chat"):

# SqliteSaver — delete by thread
conn.execute("DELETE FROM checkpoints WHERE thread_id = ?", ("user_alice",))
conn.commit()

7. Common pitfalls

  • ❗ Compiling without a checkpointer when you need memory. Then your chatbot forgets after every .invoke().
  • ❗ Forgetting thread_id in config. Without it, every call effectively starts a fresh "default" thread — state still saves, but isolation breaks.
  • ❗ Using MemorySaver in production. Dies with the process. Always use SQLite/Postgres for anything that needs to survive a restart.
  • ❗ Cross-user state leakage. Always derive thread_id from the authenticated user/session. Never trust a client-supplied thread_id.
  • ❗ No TTL / cleanup. Threads grow forever. Add periodic cleanup of old thread_ids.
  • ❗ Putting large blobs in state. Each super-step writes a full snapshot. Heavy state → slow saves. Keep state lean.
  • ❗ Confusing update_state parameters. as_node="some_node" tells the engine which node's reducer rules apply to the update — important when reducers are field-specific.

8. When to use vs not use

Use a checkpointer when Skip when
Building a chatbot / agent with memory Pure one-shot transformation
Long-running workflows you may want to resume All processing fits in a single fast .invoke()
Human-in-the-loop is needed No pause/approve step
You want time-travel debugging —
Multi-turn or multi-user app —
Checkpointer When
MemorySaver Dev / tests only
SqliteSaver Single-machine local app, side projects
PostgresSaver Production, multi-process, horizontal scale
Custom (Redis, DynamoDB) Subclass BaseCheckpointSaver

9. Cheatsheet

# Checkpointers
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# Compile with one
graph = builder.compile(checkpointer=MemorySaver())

# Always pass thread_id
cfg = {"configurable": {"thread_id": "user_42"}}
graph.invoke({"messages": [...]}, config=cfg)

# Inspect
state = graph.get_state(cfg)            # latest snapshot
state.values                             # full state dict
state.next                               # tuple of next node names (empty if at END)
state.config                             # incl. checkpoint_id

# History (newest first)
for snap in graph.get_state_history(cfg):
    ...

# Time travel — pass an old config
graph.invoke(None, config=old_snap.config)

# Edit state surgically
graph.update_state(cfg, {"messages": [HumanMessage("override")]},
                   as_node="chat")      # apply that node's reducer

# Async variants exist for all of the above: ainvoke, astream, aget_state, ...

10. Q&A — recall test

  • Q: What enables a chatbot to remember previous turns? A: Compiling with a checkpointer + invoking with a consistent thread_id per user.

  • Q: What does the engine do when you invoke an already-active thread? A: Loads the latest checkpoint, merges your new input on top, runs, saves new checkpoints.

  • Q: Difference between MemorySaver and SqliteSaver? A: MemorySaver is RAM-only — state vanishes when the process restarts. SqliteSaver writes to a file — survives restarts.

  • Q: How do you replay from an older state? A: Grab the snapshot from graph.get_state_history(cfg), invoke with its config (which includes the checkpoint_id).

  • Q: Why scope thread_id to the authenticated user? A: Otherwise users can see/overwrite each other's history. Always derive thread_id from a server-trusted identity.

  • Q: Production checkpointer choice? A: PostgresSaver. It survives restarts, scales horizontally, supports concurrent processes.

Practice

What does this print?

Expected: True

# Checkpoints are saved after EACH node, indexed by thread_id
config = {"configurable": {"thread_id": "user-123"}}
print("thread_id" in config["configurable"])

Use a unique thread_id per conversation (not a constant)

Expected: True

thread_id = "default"        # bug: all users share this thread; conversations cross-contaminate
per_user = False
print(not per_user)

Quiz — Quick check

What you remember

Q1. What does a checkpointer do?

  • Saves graph state after each step, indexed by thread_id
  • Checks for errors
  • Validates inputs
  • Logs metrics

Why: Checkpoints enable resume-after-crash, time travel (replay from any step), and human-in-the-loop (pause now, continue later). Each thread_id is a separate conversation.

Q2. What's the right checkpointer for production?

  • In-memory
  • PostgresSaver (or another persistent store) — survives restarts, scales horizontally
  • File-based
  • No checkpointer

Why: In-memory is fine for development. Production needs persistent storage so conversations survive deploys, restarts, and multi-instance scaling.

Q3. Why does each user need a unique thread_id?

  • So their conversation state is isolated from other users
  • Required by Postgres
  • Performance
  • Optional

Why: Without unique IDs, all users share one state, leaking conversations between sessions. Always derive thread_id from the authenticated user (or anonymous session cookie).

Common doubts

How big can a checkpoint get?

Bounded by your state size. Each step writes the full state (or a delta, depending on the checkpointer). Keep state small — store IDs, results summaries, not raw documents or LLM context. For huge data, store externally and reference by ID in state.

Can I resume a conversation from days ago?

Yes — that's exactly what persistent checkpointers enable. Reload the graph with the same thread_id and call .invoke(None, config=...) to continue from the last checkpoint. Cross-day, cross-process, cross-deploy.

How do I handle checkpoint schema migrations?

Same as any database. If you change the state schema, old checkpoints might break. Strategies: (1) add new fields with defaults, (2) version your state schema, (3) write a migration script. Treat checkpoint data like production database data.