Checkpointing Agent State with LangGraph: Threads, Checkpointers, and Resume

Building agents that can recover from failures isn’t optional — it’s a requirement for production systems. When an agent runs a 10-step workflow and crashes at step 7, you don’t want to restart from scratch. LangGraph’s checkpointing system solves this by persisting agent state at every step, enabling resume, replay, and fork capabilities.

Why Checkpointing Matters

Agent workflows are long-running by nature. A research agent might take 5 minutes to gather sources, synthesize findings, and produce a report. A customer support agent might handle a conversation spanning multiple turns over hours. Without persistence, every interruption means starting over.

LangGraph checkpointing provides:

• Crash recovery — Resume from the last checkpoint, not from the beginning
• Human-in-the-loop — Pause execution, let a human review, then continue
• Time travel — Fork from any previous state to explore alternative paths
• Session continuity — Maintain conversation state across user interactions

Core Concepts

Threads

A thread is a unique identifier for a conversation or workflow execution. Every agent run belongs to a thread, and all checkpoints within that run share the same thread ID.

from langgraph.graph import StateGraph, MessagesState

# Thread acts as a conversation identifier
config = {"configurable": {"thread_id": "conversation-123"}}

Checkpointers

A checkpointer is the storage backend that persists state snapshots. LangGraph supports multiple backends:

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

# In-memory (development)
checkpointer = MemorySaver()

# SQLite (lightweight production)
checkpointer = SqliteSaver.from_conn_string("agent_state.db")

# PostgreSQL (scale)
checkpointer = PostgresSaver.from_conn_string("postgresql://localhost/agents")

Checkpoints

Each checkpoint captures the full graph state at a specific super-step. A checkpoint contains:

• The serialized state (messages, tool results, custom data)
• Metadata (timestamp, step number, parent checkpoint ID)
• Version information for state schema migrations

Building a Checkpointed Agent

Here’s a complete example of a research agent with checkpointing:

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # Simulated web search
    return f"Results for: {query}"

@tool
def summarize(text: str) -> str:
    """Summarize the given text."""
    return f"Summary: {text[:200]}..."

tools = [search_web, summarize]
llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def agent_node(state: MessagesState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: MessagesState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

# Build the graph with checkpointing
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

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

# Run with thread_id for persistence
config = {"configurable": {"thread_id": "research-session-1"}}
result = app.invoke({"messages": [("user", "Research LangGraph checkpointing")]}, config)

Resuming Interrupted Workflows

When an agent is interrupted (crash, human-in-the-loop pause, or explicit halt), you can resume from the last checkpoint:

# Get the latest state for a thread
latest = app.get_state(config)

# Resume execution — picks up from last checkpoint
result = app.invoke(None, config)

The None input tells LangGraph to load the previous state and continue from where it left off. The agent won’t re-execute completed steps — it picks up exactly at the interruption point.

Human-in-the-Loop with Checkpointing

The real power of checkpointing emerges with human-in-the-loop patterns. You can pause the agent before critical actions, let a human review, and then continue:

from langgraph.types import interrupt

def approval_node(state: MessagesState):
    # This will pause execution and wait for human input
    human_decision = interrupt("Review the proposed action before proceeding")
    
    if human_decision == "approve":
        return {"messages": [("ai", "Action approved by human")]}
    else:
        return {"messages": [("ai", "Action rejected, trying alternative")]}

# The agent will pause at approval_node and wait
# Human can review via get_state() and provide input via update_state()

Time Travel and Forking

LangGraph’s checkpoint history enables time travel — you can branch from any previous state to explore alternative paths:

# Get checkpoint history
history = list(app.get_state_history(config))

# Fork from an earlier checkpoint
fork_config = {
    "configurable": {
        "thread_id": "research-session-1",
        "checkpoint_id": history[2].config["configurable"]["checkpoint_id"]
    }
}

# Continue from the forked point with different inputs
app.invoke({"messages": [("user", "Try a different research angle")]}, fork_config)

This is invaluable for debugging agent behavior and testing alternative strategies without losing previous work.

Production Considerations

State Schema Evolution

As your agent evolves, its state schema changes. LangGraph handles this through checkpoint migrations:

from langgraph.checkpoint.base import create_checkpoint

# LangGraph automatically handles schema migrations
# when loading older checkpoints with newer state definitions

Storage Backend Selection

Backend	Best For	Trade-offs
MemorySaver	Development, testing	Lost on restart
SqliteSaver	Single-server production	Limited concurrency
PostgresSaver	Multi-server, high scale	Requires DB ops
RedisSubgraph	Distributed systems	Eventual consistency

Cleanup and Retention

In production, checkpoint history grows unbounded. Implement retention policies:

# PostgresS支持 checkpoint cleanup
checkpointer = PostgresSaver.from_conn_string(conn_string)

# Delete old checkpoints for a thread
checkpointer.delete_thread(thread_id="old-conversation")

Combining Checkpointing with Subgraphs

For complex systems, you can checkpoint individual subgraphs independently:

# Main graph with its own checkpointer
main_graph = StateGraph(MainState)
main_graph.add_node("orchestrator", orchestrator_node)
main_graph.add_node("researcher", researcher_app)  # Sub-app with its checkpointer
main_graph.add_node("writer", writer_app)  # Another sub-app

# Each subgraph can have its own checkpointing strategy
main_app = main_graph.compile(checkpointer=MemorySaver())

This allows different parts of your system to have different persistence and recovery characteristics.

Debugging with Checkpoint History

LangGraph’s checkpoint history is your debugging superpower:

# Inspect what happened at each step
for checkpoint in app.get_state_history(config):
    print(f"Step {checkpoint.metadata.get('step', '?')}:")
    print(f"  State keys: {list(checkpoint.values.keys())}")
    print(f"  Next: {checkpoint.next}")
    print()

This gives you full visibility into how the agent’s state evolved, which tools were called, and what decisions were made at each step.

Conclusion

Checkpointing transforms agents from fragile scripts into resilient systems. With LangGraph’s thread-based persistence, you get crash recovery, human-in-the-loop capabilities, and time travel — all with minimal code changes. Start with MemorySaver for development, graduate to SqliteSaver or PostgresSaver for production, and always design your state schemas with evolution in mind.