LangGraph Module 5: Persistence, Checkpointers, and Threads
Introduction
A compiled graph forgets everything between invocations unless you give it a checkpointer. A checkpointer saves the state after every step, grouped into threads. That gives you conversation memory, the ability to pause and resume, and a history you can inspect and replay. This module shows how.
Prerequisites
- Module 4 completed, with the agent builder available
Step-by-Step Guide
Step 1: Compile with a checkpointer
Pass a checkpointer when compiling. InMemorySaver is ideal for learning and tests, but its data disappears when the process stops.
from langgraph.checkpoint.memory import InMemorySaver
agent = builder.compile(checkpointer=InMemorySaver())Step 2: Use a thread ID
With a checkpointer, every invocation needs a thread_id in its configuration. Calls that share a thread_id continue the same conversation, and a new thread_id starts fresh.
config = {"configurable": {"thread_id": "thread-1"}}
agent.invoke(
{"messages": [{"role": "user", "content": "Hi, my name is Bob."}]}, config
)
reply = agent.invoke(
{"messages": [{"role": "user", "content": "What is my name?"}]}, config
)
print(reply["messages"][-1].text)Step 3: Inspect the saved state
get_state returns a snapshot of the latest state. Its values field holds the state, and its next field lists the nodes still to run, which is empty when the graph has finished. get_state_history lists past snapshots, newest first.
snapshot = agent.get_state(config)
print(len(snapshot.values["messages"]), "messages saved")
print("Next nodes:", snapshot.next)
for past in agent.get_state_history(config):
print(past.metadata.get("step"), past.created_at)Step 4: Replay and edit history
Each snapshot has a checkpoint ID. Invoking the graph with a configuration that includes an earlier checkpoint_id replays from that point, which is useful for debugging and for trying a different path. update_state writes a new checkpoint with your changes, and it applies reducers, so list keys accumulate.
Step 5: Choose a durable checkpointer for production
SQLite and Postgres checkpointers come in separate packages (for SQLite, pip install langgraph-checkpoint-sqlite). They keep state across restarts, and Postgres suits multi-server deployments. Keep thread IDs under 255 characters for Postgres. When you run under the LangGraph development server or platform, persistence is provided for you, so you compile without a checkpointer there.
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver(sqlite3.connect("agent.db", check_same_thread=False))
agent = builder.compile(checkpointer=checkpointer)FAQs
- What is the difference between a thread and a checkpoint?
- A thread is a whole conversation or run, identified by thread_id. A checkpoint is one saved snapshot within that thread, taken after each step.
- Does persistence give me long-term memory across threads?
- No. A checkpointer is scoped to a thread. For facts that should be shared across conversations, such as user preferences, use LangGraph's store feature or your own database.
Summary
Compile with a checkpointer and pass a thread_id to get memory, history, replay, and resumability. Use InMemorySaver to learn and a database-backed saver in production.
Comments
Log in or create a free account to join the discussion.