Building Multi-Step Agents with LangGraph Checkpoints

What Are LangGraph Checkpoints?

LangGraph checkpoints are persisted snapshots of an agent's state at the end of each node execution. They capture the full conversation history, intermediate tool calls, and any accumulated state variables so that a multi-step agent workflow can be paused, inspected, resumed, or even rewound to a previous point. Think of them as save points in a video game — at any moment you can stop, examine exactly where you are, and decide how to proceed from there.

Under the hood, LangGraph stores checkpoints in a configurable backend — an in-memory dictionary for quick prototyping, a SQLite database for local persistence, or a production-grade solution like Postgres with LangGraph Cloud. Each checkpoint is keyed by a combination of thread_id (identifying the conversation) and a monotonically increasing checkpoint ID that creates a linked list of states over time.

Why Checkpoints Matter for Multi-Step Agents

Without checkpointing, a multi-step agent is a black box — it runs from start to finish and you either get the final output or an error. This is fundamentally limiting for real-world applications. Checkpoints unlock several critical capabilities:

Human-in-the-Loop Approval. Before executing an irreversible action (like sending an email or making a payment), the agent can pause and wait for human confirmation. The checkpoint preserves the exact state so the human can review tool arguments, approve or reject them, and the agent resumes with that feedback.

Fault Tolerance. Long-running agent workflows can span dozens of tool calls over many minutes. If the process crashes at step 17, checkpoints let you resume from step 17 rather than starting over from scratch — saving compute, API costs, and user patience.

Debugging and Observability. Every state transition is recorded. Developers can step through the exact sequence of decisions the agent made, inspect intermediate reasoning, and pinpoint where things went wrong.

Time Travel. You can rewind the agent to a previous checkpoint and explore alternative branches — what if the agent had chosen a different tool at step 3? This is invaluable for testing and optimization.

How Checkpoints Work Internally

When you compile a LangGraph graph with a checkpointer, the framework wraps each node execution with a checkpointing mechanism. Here's the flow:

Before a node runs, the current state is read from the most recent checkpoint (or initialized if this is the first step).
The node executes its logic — calling an LLM, running a tool, or applying a reducer.
After the node completes, the updated state is serialized and written to the checkpointer backend with a new checkpoint ID.
The checkpointer returns the updated state, and the graph proceeds to the next node or pauses if an interrupt is set.

The key insight: checkpoints are state-level snapshots, not code-level. They don't capture Python call stacks or local variables — they capture the structured state object flowing through the graph. This makes them lightweight and serializable.

Setting Up Your Environment

Install the required packages. LangGraph requires langgraph and langchain-core. For LLM access, install your preferred provider:

pip install langgraph langchain-core langchain-openai
# Optional: for persistent storage beyond in-memory
pip install langgraph-checkpoint-sqlite

Import the essential components:

from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI

Defining a Stateful Agent State

Every LangGraph agent begins with a state definition. For a multi-step agent, the state typically includes a running list of messages and any accumulated data. Here's a typed state for a customer support agent that handles order lookups and refunds:

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Appends new messages
    user_id: str                              # Customer identifier
    order_id: str | None                      # Retrieved order
    refund_status: str | None                 # Tracks refund progress
    requires_approval: bool                   # Flag for human-in-the-loop

The Annotated[list, operator.add] pattern tells LangGraph that new messages should be appended to the existing list rather than overwriting it — this is the reducer pattern that makes incremental state updates work correctly.

Building the Agent Nodes

Each node in the graph performs a specific job. Let's build a router node, a tool execution node, and a human approval node:

def route_intent(state: AgentState):
    """Analyzes the user's latest message and decides next action."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    
    # Bind tools the agent can call
    tools = [
        lookup_order_by_id,
        process_refund,
        escalate_to_human
    ]
    llm_with_tools = llm.bind_tools(tools)
    
    response = llm_with_tools.invoke(state["messages"])
    
    # Check if the LLM wants to call a tool
    if response.tool_calls:
        tool_name = response.tool_calls[0]["name"]
        return {
            "messages": [response],
            "next_action": tool_name
        }
    
    # No tool call — conversation is complete or needs clarification
    return {
        "messages": [response],
        "next_action": "end"
    }


def execute_tool(state: AgentState):
    """Executes the tool call from the previous node."""
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]
    
    if tool_name == "lookup_order_by_id":
        result = lookup_order_by_id(**tool_args)
        # Flag for approval if sensitive operation follows
        return {
            "messages": [ToolMessage(content=str(result), tool_call_id=tool_call["id"])],
            "order_id": result.get("order_id"),
            "requires_approval": True
        }
    elif tool_name == "process_refund":
        result = process_refund(**tool_args)
        return {
            "messages": [ToolMessage(content=str(result), tool_call_id=tool_call["id"])],
            "refund_status": result.get("status")
        }
    
    return {"messages": [ToolMessage(content="Tool executed", tool_call_id=tool_call["id"])]}


def human_approval_node(state: AgentState):
    """
    This node represents a pause point.
    The graph will interrupt here and wait for external input.
    """
    # The state at this checkpoint contains everything for review
    approval_payload = {
        "order_id": state.get("order_id"),
        "proposed_action": "process_refund",
        "messages_for_review": state["messages"][-3:]
    }
    # In a real system, this would trigger a notification
    # The graph pauses; resumption happens via invoke with updated state
    return {"requires_approval": False}

Now define the actual tool functions that interact with your backend systems:

def lookup_order_by_id(order_id: str):
    """Simulates database lookup for an order."""
    # In production: query your order database
    mock_orders = {
        "ORD-12345": {"order_id": "ORD-12345", "status": "delivered", "amount": 89.99},
        "ORD-67890": {"order_id": "ORD-67890", "status": "pending", "amount": 149.00}
    }
    return mock_orders.get(order_id, {"error": "Order not found"})


def process_refund(order_id: str, amount: float, reason: str):
    """Simulates processing a refund. Requires human approval first."""
    # In production: call your payment processor
    return {
        "status": "refund_processed",
        "transaction_id": "RFND-98765",
        "order_id": order_id,
        "amount": amount
    }


def escalate_to_human(summary: str):
    """Escalates to a human agent for complex cases."""
    return {"status": "escalated", "ticket_id": "TICK-45678"}

Composing the Graph with Checkpointing

Now wire everything together into a StateGraph. This is where checkpoints come alive — you attach a checkpointer during compilation:

def build_agent_graph():
    """Constructs the multi-step agent graph with checkpointing."""
    workflow = StateGraph(AgentState)
    
    # Add nodes
    workflow.add_node("router", route_intent)
    workflow.add_node("tool_executor", execute_tool)
    workflow.add_node("human_approval", human_approval_node)
    
    # Set entry point
    workflow.set_entry_point("router")
    
    # Define conditional edges from router
    workflow.add_conditional_edges(
        "router",
        lambda state: state.get("next_action", "end"),
        {
            "lookup_order_by_id": "tool_executor",
            "process_refund": "tool_executor",
            "escalate_to_human": "tool_executor",
            "end": END
        }
    )
    
    # After tool execution, check if approval is needed
    workflow.add_conditional_edges(
        "tool_executor",
        lambda state: "human_approval" if state.get("requires_approval") else "router",
        {
            "human_approval": "human_approval",
            "router": "router"
        }
    )
    
    # After human approval, go back to router for next step
    workflow.add_edge("human_approval", "router")
    
    return workflow

Compile the graph with a checkpointer. For development, MemorySaver is perfect. For production, switch to SqliteSaver or a cloud-backed checkpointer:

# Development: ephemeral in-memory checkpoints
checkpointer = MemorySaver()
agent_graph = build_agent_graph().compile(checkpointer=checkpointer)

# Production: persistent checkpoints with SQLite
# checkpointer = SqliteSaver.from_conn_string("agent_checkpoints.db")
# agent_graph = build_agent_graph().compile(checkpointer=checkpointer)

Running the Agent with Threads

Every invocation requires a thread_id in the config. This thread ID ties all checkpoints together into a single conversation lineage. Without it, LangGraph cannot know which conversation to resume:

def run_agent_with_thread(thread_id: str, user_input: str):
    """Invoke the agent with a persistent thread."""
    config = {"configurable": {"thread_id": thread_id}}
    
    result = agent_graph.invoke(
        {"messages": [HumanMessage(content=user_input)]},
        config=config
    )
    
    return result


# First interaction
thread_id = "user-session-abc123"
state_after_step1 = run_agent_with_thread(
    thread_id, 
    "I need to check my order ORD-12345 and possibly get a refund"
)

print(f"Messages: {len(state_after_step1['messages'])}")
print(f"Order ID: {state_after_step1.get('order_id')}")
print(f"Requires Approval: {state_after_step1.get('requires_approval')}")

Human-in-the-Loop: Interrupting and Resuming

The true power of checkpoints shines with interrupt points. When the graph hits the human_approval node, it pauses. You can inspect the checkpoint, make a decision, and resume with updated state. Here's how to set up explicit interrupts:

def build_agent_with_interrupt():
    """Builds a graph that explicitly interrupts before sensitive operations."""
    workflow = StateGraph(AgentState)
    
    workflow.add_node("router", route_intent)
    workflow.add_node("tool_executor", execute_tool)
    workflow.add_node("human_approval", human_approval_node)
    
    workflow.set_entry_point("router")
    
    workflow.add_conditional_edges(
        "router",
        lambda state: state.get("next_action", "end"),
        {
            "lookup_order_by_id": "tool_executor",
            "process_refund": "tool_executor",
            "end": END
        }
    )
    
    # After tool execution, go to approval if flagged
    workflow.add_conditional_edges(
        "tool_executor",
        lambda state: "human_approval" if state.get("requires_approval") else "router",
        {"human_approval": "human_approval", "router": "router"}
    )
    
    workflow.add_edge("human_approval", "router")
    
    # Compile with interrupt BEFORE the human_approval node
    checkpointer = MemorySaver()
    return workflow.compile(
        checkpointer=checkpointer,
        interrupt_before=["human_approval"]
    )

Now when the agent reaches the approval node, it stops and returns the current state. You can examine it, then resume with a decision:

interrupt_graph = build_agent_with_interrupt()
thread_id = "session-def456"
config = {"configurable": {"thread_id": thread_id}}

# First invocation — runs until the interrupt point
partial_result = interrupt_graph.invoke(
    {"messages": [HumanMessage(content="Look up order ORD-12345 and process a refund")]},
    config=config
)

# The graph paused! Inspect the checkpoint
print("Agent paused for approval.")
print(f"Current state keys: {list(partial_result.keys())}")
print(f"Pending action review:")
for msg in partial_result.get("messages", [])[-3:]:
    print(f"  - {msg.content[:100] if hasattr(msg, 'content') else str(msg)[:100]}")

# Human decides: approve the refund with a note
# Resume by invoking with None input (or additional state updates)
final_result = interrupt_graph.invoke(
    None,  # No new human message — just resume
    config=config
)

print(f"Final refund status: {final_result.get('refund_status')}")

Inspecting and Navigating Checkpoint History

LangGraph provides utilities to list checkpoints and rewind to previous states. This is invaluable for debugging and for building "undo" features:

def explore_checkpoint_history(thread_id: str):
    """List all checkpoints for a thread and demonstrate rewinding."""
    config = {"configurable": {"thread_id": thread_id}}
    
    # Get all checkpoints for this thread
    checkpoints = list(agent_graph.list_checkpoints(config=config))
    
    print(f"Found {len(checkpoints)} checkpoints for thread '{thread_id}':")
    for checkpoint in checkpoints:
        cp_id = checkpoint.checkpoint_id[:8]  # Truncate for display
        step_count = len(checkpoint.state.get("messages", []))
        print(f"  Checkpoint {cp_id}... - {step_count} messages in state")
    
    return checkpoints


def rewind_to_checkpoint(thread_id: str, checkpoint_id: str):
    """Resume execution from a specific historical checkpoint."""
    config = {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_id": checkpoint_id
        }
    }
    
    # Invoke from that checkpoint — the agent picks up from there
    result = agent_graph.invoke(
        {"messages": [HumanMessage(content="Actually, let's try a different approach")]},
        config=config
    )
    
    return result


# Explore checkpoints after running the agent
checkpoints = explore_checkpoint_history("user-session-abc123")

# Rewind to an earlier checkpoint and take a different path
if checkpoints:
    earliest_checkpoint_id = checkpoints[-1].checkpoint_id  # Oldest first
    rewind_result = rewind_to_checkpoint("user-session-abc123", earliest_checkpoint_id)
    print(f"Rewound and took new path. Messages: {len(rewind_result['messages'])}")

Persistent Checkpointing with SQLite

For applications that need to survive server restarts, swap MemorySaver for SqliteSaver. The API is identical — only the backend changes:

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

def build_persistent_agent():
    """Builds an agent with SQLite-backed checkpointing."""
    # Create a connection — in production, use a proper connection pool
    conn = sqlite3.connect("agent_sessions.db", check_same_thread=False)
    checkpointer = SqliteSaver(conn)
    
    workflow = build_agent_graph()
    graph = workflow.compile(checkpointer=checkpointer)
    
    return graph, conn


# Usage
persistent_graph, db_connection = build_persistent_agent()

# Session persists across script invocations
config = {"configurable": {"thread_id": "persistent-session-001"}}

result = persistent_graph.invoke(
    {"messages": [HumanMessage(content="What's the status of order ORD-67890?")]},
    config=config
)

# Even if you restart the Python process and reconnect to the same DB,
# the checkpoints are still there
db_connection.close()

For production deployments, LangGraph Cloud offers a fully managed checkpointing service with Postgres, automatic scaling, and built-in monitoring — but the local SQLite approach works well for single-server deployments and development.

Complete Multi-Step Agent Example

Here's a complete, runnable example that ties everything together — a customer support agent that looks up orders, requests human approval for refunds, and handles the full lifecycle:

from typing import TypedDict, Annotated, Literal
import operator
import sqlite3
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI


# --- State Definition ---
class SupportState(TypedDict):
    messages: Annotated[list, operator.add]
    user_id: str
    order_id: str | None
    refund_status: str | None
    requires_approval: bool
    next_action: str | None


# --- Tool Functions ---
def lookup_order(order_id: str) -> dict:
    orders = {
        "ORD-A001": {"id": "ORD-A001", "status": "delivered", "amount": 89.99, "customer": "user_42"},
        "ORD-A002": {"id": "ORD-A002", "status": "pending", "amount": 149.00, "customer": "user_42"},
    }
    return orders.get(order_id, {"error": "Order not found"})


def issue_refund(order_id: str, amount: float, reason: str) -> dict:
    # Simulated refund processing
    return {
        "status": "refund_approved",
        "transaction_id": f"RFND-{order_id}",
        "amount_refunded": amount,
        "reason": reason
    }


# --- Node Functions ---
def router_node(state: SupportState):
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    tools = [lookup_order, issue_refund]
    llm_with_tools = llm.bind_tools(tools)
    
    response = llm_with_tools.invoke(state["messages"])
    
    if response.tool_calls:
        return {
            "messages": [response],
            "next_action": response.tool_calls[0]["name"]
        }
    return {"messages": [response], "next_action": "end"}


def tool_node(state: SupportState):
    last_msg = state["messages"][-1]
    tool_call = last_msg.tool_calls[0]
    
    if tool_call["name"] == "lookup_order":
        result = lookup_order(**tool_call["args"])
        return {
            "messages": [ToolMessage(content=str(result), tool_call_id=tool_call["id"])],
            "order_id": result.get("id"),
            "requires_approval": True if "error" not in result else False
        }
    elif tool_call["name"] == "issue_refund":
        result = issue_refund(**tool_call["args"])
        return {
            "messages": [ToolMessage(content=str(result), tool_call_id=tool_call["id"])],
            "refund_status": result.get("status"),
            "requires_approval": False
        }
    
    return {"messages": [ToolMessage(content="Done", tool_call_id=tool_call["id"])]}


def approval_node(state: SupportState):
    # This node is the pause point for human review
    # In production: send notification, wait for webhook
    return {
        "requires_approval": False,
        "next_action": "router"
    }


# --- Graph Construction ---
def create_support_agent(checkpointer):
    workflow = StateGraph(SupportState)
    
    workflow.add_node("router", router_node)
    workflow.add_node("tools", tool_node)
    workflow.add_node("approval", approval_node)
    
    workflow.set_entry_point("router")
    
    workflow.add_conditional_edges(
        "router",
        lambda s: s.get("next_action", "end"),
        {
            "lookup_order": "tools",
            "issue_refund": "tools",
            "end": END
        }
    )
    
    workflow.add_conditional_edges(
        "tools",
        lambda s: "approval" if s.get("requires_approval") else "router",
        {"approval": "approval", "router": "router"}
    )
    
    workflow.add_edge("approval", "router")
    
    return workflow.compile(
        checkpointer=checkpointer,
        interrupt_before=["approval"]
    )


# --- Initialize ---
conn = sqlite3.connect("support_sessions.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
agent = create_support_agent(checkpointer)


# --- Run the Agent ---
thread_id = "customer-user42-session1"
config = {"configurable": {"thread_id": thread_id}}

# Step 1: User asks about an order
state1 = agent.invoke(
    {"messages": [HumanMessage(content="Hi, I'm user_42. Can you look up my order ORD-A001?")]},
    config=config
)

print("--- After Step 1 ---")
for msg in state1["messages"]:
    if hasattr(msg, "content"):
        print(f"  [{msg.__class__.__name__}] {msg.content[:120]}")

# The agent pauses at the approval node because requires_approval is True
print(f"\nAgent paused! Checkpoint created.")
print(f"Order ID found: {state1.get('order_id')}")

# Step 2: Human reviews and resumes (simulated)
# In a real app, the human would see the order details and approve the refund
state2 = agent.invoke(
    None,  # Resume without new human input
    config=config
)

print("\n--- After Human Approval & Resume ---")
print(f"Refund status: {state2.get('refund_status')}")
print(f"Total messages: {len(state2['messages'])}")

# Step 3: User asks for another action in the same thread
state3 = agent.invoke(
    {"messages": [HumanMessage(content="Great, now please issue a refund for that order. The reason is 'product defective'.")]},
    config=config
)

print("\n--- After Step 3 (Refund Request) ---")
print(f"Final refund status: {state3.get('refund_status')}")

# Clean up
conn.close()

Best Practices for Checkpoint-Driven Agents

Design State Carefully. Your state TypedDict should contain everything needed to reconstruct context at any checkpoint. Include message history, extracted entities, flags like requires_approval, and any domain-specific data the agent accumulates. Avoid storing ephemeral references (like open file handles) — checkpoints must be serializable.
Use Meaningful Thread IDs. Thread IDs should map to real-world identifiers: user sessions, order numbers, or conversation UUIDs. This makes it trivial to look up a specific conversation's checkpoint history and resume it from any system.
Place Interrupts Strategically. Set interrupt_before on nodes that gate irreversible actions — refunds, emails, data deletions. The agent should do all its reasoning and data gathering first, then pause right before the destructive operation. This minimizes the human's cognitive load while maximizing safety.
Keep Nodes Idempotent Where Possible. When resuming from a checkpoint, the same node may execute again. Design tool functions to be idempotent — for example, use transaction IDs to prevent double-processing refunds, or check preconditions before acting.
Test Time Travel Flows. Write integration tests that run the agent, rewind to a previous checkpoint, inject a modified state, and verify the agent takes the expected alternate path. This catches edge cases where stale state from a checkpoint could cause unexpected behavior.
Monitor Checkpoint Growth. Each checkpoint stores the full state. For long-running conversations with hundreds of steps, the accumulated state can grow large. Consider truncating message history (keeping the last N messages plus a summary) or compressing older state in a production system.
Use the Right Checkpointer for Your Scale. MemorySaver is for notebooks and tests only — it loses everything on restart. SqliteSaver works for single-process production apps. For distributed systems with multiple servers, use LangGraph Cloud's Postgres-backed checkpointer or implement a custom one with your organization's primary data store.
Handle Checkpoint Errors Gracefully. Checkpointer backends can fail (database connection lost, disk full). Wrap your invoke calls in try/except blocks and implement retry logic. If a checkpoint write fails, the agent should surface a clear error rather than silently losing state.

Advanced: Custom Checkpoint Serialization

By default, LangGraph serializes state using standard JSON/pickle mechanisms. If your state contains non-serializable objects (custom classes, database connections), you can implement custom serialization. Here's a pattern using a wrapper checkpointer:

from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint

class CustomSerializationCheckpointer(BaseCheckpointSaver):
    """Wraps a base checkpointer with custom serialization logic."""
    
    def __init__(self, base_checkpointer):
        self.base = base_checkpointer
    
    def _serialize_state(self, state: dict) -> dict:
        # Convert non-serializable objects before persisting
        serialized = {}
        for key, value in state.items():
            if hasattr(value, 'to_dict'):
                serialized[key] = value.to_dict()
            else:
                serialized[key] = value
        return serialized
    
    def _deserialize_state(self, state: dict) -> dict:
        # Reconstruct objects when loading
        deserialized = {}
        for key, value in state.items():
            if isinstance(value, dict) and key == "database_row":
                deserialized[key] = DatabaseRow.from_dict(value)
            else:
                deserialized[key] = value
        return deserialized
    
    def put(self, checkpoint: Checkpoint, **kwargs):
        checkpoint.state = self._serialize_state(checkpoint.state)
        return self.base.put(checkpoint, **kwargs)
    
    def get(self, checkpoint_id: str, **kwargs):
        checkpoint = self.base.get(checkpoint_id, **kwargs)
        if checkpoint:
            checkpoint.state = self._deserialize_state(checkpoint.state)
        return checkpoint
    
    def list(self, **kwargs):
        for checkpoint in self.base.list(**kwargs):
            checkpoint.state = self._deserialize_state(checkpoint.state)
            yield checkpoint

Conclusion

LangGraph checkpoints transform multi-step agents from linear, opaque pipelines into interactive, debuggable, and resilient workflows. By persisting state at every node boundary, they enable human-in-the-loop approval patterns, fault-tolerant resumption from mid-execution failures, and powerful time-travel debugging capabilities. The API is elegantly simple — attach a checkpointer at compilation time, provide a thread ID on each invocation, and you gain all these capabilities with minimal code changes. Whether you're building customer support agents that handle refunds, data pipelines that require sign-off at key stages, or conversational agents that need to survive server restarts, mastering LangGraph checkpoints is an essential skill for building production-grade LLM applications that go beyond simple request-response patterns.