Checkpointing LangGraph State: Snapshots, Restores, and Forks

LangGraph checkpointing is more than crash recovery — it’s a time machine for your agents. Snapshots capture the full state at any point. Restores let you resume from any snapshot. Forks let you branch from any snapshot to explore alternatives. Together, they give you complete control over agent execution.

The Checkpoint Model

Every super-step in a LangGraph execution produces a checkpoint:

Step 0: Initial State → Checkpoint 0
    ↓
Step 1: After Agent Node → Checkpoint 1
    ↓
Step 2: After Tool Node → Checkpoint 2
    ↓
Step 3: After Synthesis → Checkpoint 3

Each checkpoint contains:

• State values — The full state dictionary
• Metadata — Timestamp, step number, parent checkpoint ID
• Versions — For schema migration support

Creating Snapshots

Automatic Snapshots

LangGraph automatically creates checkpoints at every super-step:

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver

graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Every execution creates checkpoints automatically
config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke({"messages": [("user", "Research topic")]}, config)

Manual Snapshots

Create explicit snapshots at important points:

from langgraph.checkpoint.base import create_checkpoint

class SnapshotManager:
    def __init__(self, checkpointer):
        self.checkpointer = checkpointer
    
    async def create_snapshot(
        self,
        config: dict,
        label: str = None
    ) -> str:
        """Create a named snapshot of the current state."""
        # Get current state
        state = await self.checkpointer.aget(config)
        
        # Create checkpoint with label
        checkpoint = create_checkpoint(
            state.config,
            state.metadata | {"label": label, "snapshot_time": datetime.now().isoformat()}
        )
        
        # Store the snapshot
        await self.checkpointer.aput(config, checkpoint)
        
        return checkpoint["id"]

# Usage
snapshot_mgr = SnapshotManager(checkpointer)
snapshot_id = await snapshot_mgr.create_snapshot(
    config,
    label="before_expensive_operation"
)

Restoring from Snapshots

Resume from Latest Checkpoint

# Resume execution from the last checkpoint
result = app.invoke(None, config)  # None = load from checkpoint

Resume from Specific Checkpoint

# List all checkpoints
history = list(app.get_state_history(config))

# Resume from checkpoint #2
specific_config = {
    "configurable": {
        "thread_id": "session-1",
        "checkpoint_id": history[2].config["configurable"]["checkpoint_id"]
    }
}

result = app.invoke(None, specific_config)

Restore with Modifications

# Load state, modify it, then continue
state = app.get_state(config)

# Modify the state
modified_messages = state.values["messages"] + [
    ("user", "Actually, let's try a different approach")
]

# Update state and continue
app.update_state(config, {"messages": modified_messages})
result = app.invoke(None, config)

Forking: Exploring Alternative Paths

Basic Fork

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

# Fork from an earlier checkpoint
fork_point = history[3]  # The 4th checkpoint

fork_config = {
    "configurable": {
        "thread_id": "session-1-fork",
        "checkpoint_id": fork_point.config["configurable"]["checkpoint_id"]
    }
}

# Continue from the fork with different input
app.update_state(fork_config, {
    "messages": [("user", "Try the alternative approach instead")]
})

fork_result = app.invoke(None, fork_config)

Parallel Forks

import asyncio

async def explore_alternatives(config, alternatives: list[str]):
    """Run multiple forks in parallel."""
    history = list(app.get_state_history(config))
    fork_point = history[-2]  # Fork from second-to-last checkpoint
    
    tasks = []
    for i, alt in enumerate(alternatives):
        fork_config = {
            "configurable": {
                "thread_id": f"fork-{i}",
                "checkpoint_id": fork_point.config["configurable"]["checkpoint_id"]
            }
        }
        
        app.update_state(fork_config, {
            "messages": [("user", alt)]
        })
        
        tasks.append(app.ainvoke(None, fork_config))
    
    results = await asyncio.gather(*tasks)
    return results

# Compare different approaches
results = await explore_alternatives(config, [
    "Approach A: Use vector search",
    "Approach B: Use keyword search",
    "Approach C: Combine both methods"
])

Practical Patterns

Debugging Failed Executions

def debug_failed_run(config):
    """Find where a run failed and what went wrong."""
    history = list(app.get_state_history(config))
    
    for i, checkpoint in enumerate(history):
        state = checkpoint.values
        
        # Check for errors
        if state.get("error"):
            print(f"Error at step {i}: {state['error']}")
            print(f"State before error:")
            print(json.dumps(state, indent=2, default=str))
            
            # Can fork from here to try again
            return checkpoint.config["configurable"]["checkpoint_id"]
    
    return None

# Find the failure point
failed_at = debug_failed_run(config)

# Fork from before the failure
if failed_at:
    fork_config = {
        "configurable": {
            "thread_id": "debug-fork",
            "checkpoint_id": failed_at
        }
    }
    # Modify state to fix the issue
    app.update_state(fork_config, {"retry_with_fix": True})
    result = app.invoke(None, fork_config)

A/B Testing Agent Strategies

class ABTestManager:
    def __init__(self, app):
        self.app = app
    
    async def test_strategies(
        self,
        base_config: dict,
        task: str,
        strategies: dict[str, dict]
    ) -> dict[str, dict]:
        """Test multiple strategies from the same starting point."""
        # Get the base state
        history = list(self.app.get_state_history(base_config))
        fork_point = history[0]  # Start from the beginning
        
        results = {}
        
        for strategy_name, modifications in strategies.items():
            fork_config = {
                "configurable": {
                    "thread_id": f"ab-{strategy_name}",
                    "checkpoint_id": fork_point.config["configurable"]["checkpoint_id"]
                }
            }
            
            # Apply strategy-specific modifications
            self.app.update_state(fork_config, modifications)
            
            # Run to completion
            result = await self.app.ainvoke(None, fork_config)
            
            results[strategy_name] = {
                "output": result["messages"][-1].content,
                "steps": len(result["messages"]),
                "tokens_used": result.get("token_count", 0),
            }
        
        return results

# Test different approaches
ab_test = ABTestManager(app)
results = await ab_test.test_strategies(
    base_config,
    "Research quantum computing applications",
    {
        "thorough": {"max_iterations": 20, "depth": "comprehensive"},
        "quick": {"max_iterations": 5, "depth": "brief"},
        "balanced": {"max_iterations": 10, "depth": "moderate"},
    }
)

Time-Travel Debugging

class TimeTravelDebugger:
    def __init__(self, app):
        self.app = app
    
    def inspect_trajectory(self, config):
        """Show the complete execution trajectory."""
        history = list(self.app.get_state_history(config))
        
        trajectory = []
        for i, checkpoint in enumerate(reversed(history)):
            state = checkpoint.values
            trajectory.append({
                "step": i,
                "timestamp": checkpoint.metadata.get("timestamp"),
                "messages_count": len(state.get("messages", [])),
                "last_message": str(state["messages"][-1].content)[:100] if state.get("messages") else "",
                "tool_calls": sum(
                    1 for m in state.get("messages", [])
                    if hasattr(m, "tool_calls") and m.tool_calls
                ),
            })
        
        return trajectory
    
    def compare_trajectories(self, config1, config2):
        """Compare two execution trajectories."""
        traj1 = self.inspect_trajectory(config1)
        traj2 = self.inspect_trajectory(config2)
        
        return {
            "trajectory_1": traj1,
            "trajectory_2": traj2,
            "step_difference": len(traj1) - len(traj2),
            "divergence_point": self._find_divergence(traj1, traj2),
        }
    
    def _find_divergence(self, traj1, traj2):
        """Find where two trajectories diverge."""
        for i, (s1, s2) in enumerate(zip(traj1, traj2)):
            if s1["last_message"] != s2["last_message"]:
                return i
        return min(len(traj1), len(traj2))

Production Considerations

Cleanup Old Checkpoints

class CheckpointCleanup:
    def __init__(self, checkpointer, retention_days: int = 30):
        self.checkpointer = checkpointer
        self.retention_days = retention_days
    
    async def cleanup_old_checkpoints(self):
        """Remove checkpoints older than retention period."""
        cutoff = datetime.now() - timedelta(days=self.retention_days)
        
        # For PostgreSQL checkpointer
        if hasattr(self.checkpointer, 'conn'):
            await self.checkpointer.conn.execute("""
                DELETE FROM checkpoints 
                WHERE created_at < $1
                AND NOT is_final  # Keep final checkpoints
            """, cutoff)

Schema Migration

class CheckpointMigration:
    def migrate_checkpoint(self, old_checkpoint: dict, old_version: str) -> dict:
        """Migrate a checkpoint to the current schema version."""
        if old_version == "1.0":
            # Add new fields
            old_checkpoint["metadata"]["version"] = "2.0"
            old_checkpoint["values"]["new_field"] = "default_value"
        
        return old_checkpoint

Conclusion

LangGraph checkpointing gives you snapshots for capturing state, restores for resuming execution, and forks for exploring alternatives. Together, they enable debugging, A/B testing, and time-travel workflows that would be impossible with simple state management. Use automatic checkpoints for safety, manual snapshots for important milestones, and forks for experimentation.