Building Multi-Step Agents with LangGraph Checkpoints

Understanding Checkpoints in LangGraph

Checkpoints are the persistence mechanism inside LangGraph that capture the full state of your agent graph after each node execution. They form the backbone of reliable, multi-step agents, enabling you to pause execution, resume from where you left off, inject human feedback, and even branch into alternative execution paths. In this tutorial you’ll learn what checkpoints are, why they matter for agentic applications, how to implement them step by step, and best practices for production use.

What Is a Checkpoint?

In LangGraph, a checkpoint is a snapshot of your graph’s state at a specific node. It contains:

• The current values of your state dictionary (or Pydantic model)
• Metadata such as the node that just completed and the next pending writes
• A unique identifier that allows you to retrieve, replay, or branch from that point

Checkpoints are created automatically when you attach a Checkpointer to your StateGraph. LangGraph provides several built-in checkpointers, from in‑memory storage for development to SQLite and Postgres adapters for production.

Why Checkpoints Matter for Multi‑Step Agents

Without checkpoints, an agent runs from start to finish in a single uninterrupted shot. That works for simple chains but breaks down when you need:

• Human‑in‑the‑loop – pause before a tool call or final answer to get approval
• Long‑running workflows – survive server restarts without losing progress
• Debugging and auditing – inspect exactly what the agent thought at step N
• Time‑travel and branching – rewind to a previous checkpoint and try a different path

Checkpoints turn a fragile, fire‑and‑forget agent into a robust, steerable system that can be interrupted, resumed, and debugged with ease.

Setting Up Your First Checkpoint‑Enabled Agent

Let’s build a minimal multi‑step agent that decides whether to call a tool or finish. We’ll attach an in‑memory checkpointer and then run the agent with a thread ID so state persists across invocations.

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver

# 1. Define the state schema
class AgentState(TypedDict):
    messages: list
    next_step: Optional[str]  # 'tool_node' or 'END'

# 2. Node: chatbot logic (simplified)
def chatbot(state: AgentState) -> dict:
    last_msg = state["messages"][-1] if state["messages"] else ""
    # Decide next step based on last message content
    if "weather" in last_msg.lower():
        return {"next_step": "tool_node"}
    return {"next_step": "END"}

# 3. Node: simulate a tool call
def weather_tool(state: AgentState) -> dict:
    state["messages"].append("Tool: Weather is 72°F and sunny.")
    return {"next_step": "chatbot"}

# 4. Build the graph
graph = StateGraph(AgentState)
graph.add_node("chatbot", chatbot)
graph.add_node("tool_node", weather_tool)
graph.add_edge(START, "chatbot")
graph.add_conditional_edges(
    "chatbot",
    lambda s: s["next_step"],
    {
        "tool_node": "tool_node",
        "END": END,
    },
)
graph.add_edge("tool_node", "chatbot")

# 5. Compile with a checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

The MemorySaver stores checkpoints in process memory – perfect for prototyping. For persistence across restarts, swap it with SqliteSaver or PostgresSaver later.

Running the Agent with a Thread Identifier

Every invocation must provide a thread_id inside the configurable dictionary. This ID acts as a session key: all checkpoints for a conversation are grouped under it.

# Initial state for a new conversation
initial_state = {"messages": ["User: What's the weather in Paris?"], "next_step": None}
config = {"configurable": {"thread_id": "user-123-session"}}

# First call – runs until it hits the tool node or END
result = app.invoke(initial_state, config)
print(result["messages"])
# Output: ['User: What's the weather in Paris?', 'Tool: Weather is 72°F and sunny.']

After this run, a checkpoint is saved after the chatbot node and again after tool_node. You can inspect the latest checkpoint:

latest_checkpoint = app.get_state(config)
print(latest_checkpoint.values["messages"])
# Shows the full message list including the tool response

Interrupting and Resuming Execution

The real power of checkpoints comes from interrupts. You can pause the graph before a critical node (like a tool execution) to ask for human confirmation or to inject extra context, then resume gracefully.

Adding an Interrupt Before the Tool

Replace the direct edge to tool_node with an interrupt node that calls interrupt(). LangGraph will pause and wait for a resume value.

from langgraph.types import interrupt, Command

# New node that interrupts before the tool
def human_approval(state: AgentState) -> dict:
    # This call pauses execution and waits for a resume value
    approved = interrupt("Approve weather tool call?")
    if not approved:
        return {"next_step": "END"}
    return {"next_step": "tool_node"}

# Update the graph
graph = StateGraph(AgentState)
graph.add_node("chatbot", chatbot)
graph.add_node("human_approval", human_approval)
graph.add_node("tool_node", weather_tool)
graph.add_edge(START, "chatbot")
graph.add_conditional_edges(
    "chatbot",
    lambda s: s["next_step"],
    {
        "tool_node": "human_approval",   # route to approval node
        "END": END,
    },
)
graph.add_edge("human_approval", "tool_node")
graph.add_edge("human_approval", END)   # fallback if not approved
graph.add_edge("tool_node", "chatbot")

# Compile again with checkpointer
app = graph.compile(checkpointer=MemorySaver())

Invoking and Resuming the Interrupted Graph

When you call invoke now, the graph will pause at human_approval. You get an Interrupt exception (or handle it via stream). Then you resume by passing a Command with the resume value.

config = {"configurable": {"thread_id": "user-456"}}
initial_state = {"messages": ["User: Weather in London?"], "next_step": None}

# Start the run – it will interrupt
try:
    app.invoke(initial_state, config)
except Interrupt:
    print("Graph interrupted – waiting for approval...")

# Now resume with the human decision
resume_value = True  # human says "go ahead"
app.invoke(Command(resume=resume_value), config)

# Check final state
final_state = app.get_state(config)
print(final_state.values["messages"])
# Includes the tool response

This pattern gives you a full human‑in‑the‑loop workflow. The checkpoint preserves the exact state before the tool call, so resuming continues exactly where it stopped – no state loss, no re‑execution of previous nodes.

Inspecting Checkpoint History and Time‑Travel

LangGraph stores a history of checkpoints for each thread. You can list them, fetch a specific past checkpoint, and even replay from that point – enabling powerful debugging and branching.

# Get all checkpoints for the thread (oldest first)
history = list(app.get_state_history(config))
for snapshot in history:
    print(f"Checkpoint at node '{snapshot.metadata['source']}' -> messages: {snapshot.values['messages']}")

# Replay from a specific checkpoint
old_checkpoint = history[1]  # e.g., right after the chatbot node
replay_config = old_checkpoint.config
replayed_state = app.invoke(None, replay_config)  # None state = resume from checkpoint

The ability to rewind and branch is invaluable when you want to explore “what‑if” scenarios or recover from a mistaken tool call without restarting the entire conversation.

Best Practices for Production Checkpoints

• Use persistent checkpointer implementations – SqliteSaver for single‑server apps, PostgresSaver or MongoSaver for distributed systems. In‑memory savers disappear on restart.
• Always supply a thread_id – it groups checkpoints per conversation and prevents cross‑session state pollution. Use deterministic IDs (user ID + session ID).
• Limit checkpoint history – pass max_history to the checkpointer constructor (e.g., MemorySaver(max_history=50)) to avoid unbounded memory growth.
• Serialize state carefully – if your state contains custom objects, implement __getstate__ / __setstate__ or use Pydantic models with model_dump and model_validate. Avoid storing non‑serializable objects like open file handles or raw LLM clients.
• Checkpoint after “risky” nodes – insert interrupts before tool calls, API writes, or final answers. This gives you a safety net to review and correct before irreversible actions.
• Use get_state() for debugging – after a failed run, inspect the last checkpoint to see the exact state that led to the error. Combine with logging to capture metadata like node name and timestamp.
• Combine with threads for parallel conversations – each thread ID is independent; you can safely run hundreds of concurrent agent sessions by using different thread_id values.

Conclusion

LangGraph checkpoints turn a linear agent execution into a controllable, observable, and resumable process. By attaching a checkpointer, providing a thread ID, and strategically placing interrupts, you can build multi‑step agents that gracefully handle human approvals, survive restarts, and offer time‑travel debugging. Start with MemorySaver for quick iteration, then move to SqliteSaver or PostgresSaver for production. With these patterns, your agents become not just smarter, but also far more reliable and maintainable.