LangGraph checkpoints enable human approval workflows

Your LangGraph agent works in testing but crashes mid-approval with cryptic checkpoint serialization errors, losing user context and forcing restarts. LangGraph’s checkpoint system reduces agent recovery time from full restart to sub-100ms state restoration [VERIFY: 2026-10-12]. This guide shows working developers how to implement LangGraph checkpoints for human approval workflows with proper error handling, state migration, and graceful degradation when approval services fail.

What are LangGraph checkpoints and why do they enable human approval?

LangGraph checkpoints are persistent snapshots of graph state that enable interruption and resumption. They serialize the complete state, including messages, custom data, and current node, allowing workflows to pause indefinitely for human input and resume exactly where they left off.

According to the official LangGraph persistence documentation, the checkpoint system captures the full state dictionary at every node transition. This mechanism transforms a state machine from a fire-and-forget execution model into a resumable workflow engine. When you invoke a graph with a thread_id, the checkpointer stores state before and after each node executes. If execution halts, you can reconstruct the entire graph context by loading that snapshot.

The key enabling feature for human approval is the ability to persist state across process boundaries. When a checkpoint survives in external storage, the approval workflow can outlive the initial request. This matters because human decisions often happen outside your application’s runtime. An approval might arrive via email link, a mobile app, or an internal dashboard hours after the original agent execution started.

Popular checkpoint persistence backends include:

• MemorySaver: Fast, zero-dependency, data loss on process restart
• SqliteSaver: Lightweight disk persistence for development
• PostgreSQLSaver: Durable, queryable state for production workloads
• RedisSaver: Fast persistent option with TTL support

How checkpoint persistence works under the hood

Each checkpoint contains three critical pieces of data: the current state values, the configuration (including the thread_id), and the next node to execute. When a graph runs with a checkpointer attached, LangGraph writes a checkpoint after each node completes. To resume, you invoke the graph again with the same thread_id, and the checkpointer loads the last saved snapshot.

The checkpoint lifecycle follows this pattern:

1. Graph receives initial input with a thread_id
2. State initializes or loads from existing checkpoint
3. Each node executes and writes a new checkpoint
4. interrupt() halts execution and writes a special “pending” checkpoint
5. Resume call with Command input updates state and continues from the interrupt point

Because checkpoints capture the full execution graph, they support complex topologies including cycles, conditional edges, and nested subgraphs. Each checkpoint tracks not just the state values but the precise position within the graph topology.

Serializers: memory vs. SQLite vs. Redis vs. PostgreSQL

The checkpointer backend determines durability and performance characteristics. Memory checkpointers store state in a Python dictionary and lose all data when the process terminates. SQLite checkpointers write to a local file, offering persistence without external dependencies but limited scalability. PostgreSQL checkpointers provide production-grade durability with transactional semantics, making them suitable for multi-instance deployments. Redis checkpointers offer high-speed persistence with built-in TTL for automatic cleanup.

The trade-offs are clear. Development environments can use memory or SQLite checkpointers for fast iteration. Production workloads handling sensitive agent state should opt for PostgreSQL when audit trails matter or Redis when low latency is critical. The langgraph-checkpoint-postgres package is the recommended production persistence layer for most deployments [VERIFY: Is langgraph-checkpoint-postgres the recommended production persistence as of 2026-10-12?].

How do you implement basic human approval with LangGraph interrupt?

Use interrupt() inside a node to halt execution and surface a question to the user. The checkpoint captures state automatically. Resume by calling ainvoke() with user input, which restores state and continues execution.

The interrupt mechanism works by raising a special exception that the LangGraph runtime catches. Instead of propagating the error, the runtime writes a checkpoint with a “pending” status and returns control to the caller. The caller can then gather user input and resume execution by passing a Command object with the response.

This pattern differs from conditional edges. The interrupt() function pauses execution synchronously and waits for immediate user input within the same session. Conditional edges route to different nodes based on state. Use conditional edges with external approval services that poll or use callbacks for asynchronous approval workflows.

The LangGraph checkpoint API defines interrupt() as the primary mechanism for human-in-the-loop flows. The function accepts a string value that becomes available to the resuming caller, allowing the graph to communicate what input it needs.

Single-turn approval with interrupt()

A basic approval node uses interrupt() to pause and wait for confirmation. The node returns a value based on the user’s response after resumption.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt

class AgentState(TypedDict):
    messages: list
    action: str
    approved: bool

def approval_node(state: AgentState) -> dict:
    user_response = interrupt(f"Approve action: {state['action']}?")
    approved = user_response.lower() in ["yes", "approve", "true"]
    return {"approved": approved}

def execution_node(state: AgentState) -> dict:
    if state["approved"]:
        action_result = f"Executed: {state['action']}"
    else:
        action_result = "Action cancelled by user"
    return {"messages": state["messages"] + [action_result]}

def build_approval_graph():
    graph = StateGraph(AgentState)
    graph.add_node("approval", approval_node)
    graph.add_node("execution", execution_node)
    graph.set_entry_point("approval")
    graph.add_edge("approval", "execution")
    graph.add_edge("execution", END)
    return graph.compile(checkpointer=MemorySaver())

# Usage
app = build_approval_graph()
config = {"configurable": {"thread_id": "approval-demo-001"}}

# First invocation - interrupts at approval node
result1 = app.invoke({"messages": [], "action": "delete_database"}, config)
# Returns with interrupt

# Resume with approval
from langgraph.types import Command
result2 = app.invoke(Command(resume="approve"), config)
print(result2["messages"][-1])  # "Executed: delete_database"

The graph above demonstrates the fundamental pattern. The first invoke() stops at the approval node and returns without reaching execution. The checkpoint captures that the graph is mid-execution at the approval node. The second invoke() with Command(resume=...) provides the user input, which becomes the return value of interrupt().

Multi-turn conversation handling

Real approval workflows often require clarifying questions. Multiple interrupt() calls within the same node create a conversation flow.

def clarifying_approval_node(state: AgentState) -> dict:
    initial_response = interrupt(f"Action '{state['action']}' requires approval. Proceed?")
    
    if initial_response.lower() == "details":
        details = f"Action will affect 47 records in production database."
        confirm = interrupt(f"{details}\nApprove now?")
        approved = confirm.lower() in ["yes", "approve"]
    else:
        approved = initial_response.lower() in ["yes", "approve"]
    
    return {"approved": approved}

# Multi-turn usage
app = build_approval_graph()
config = {"configurable": {"thread_id": "multi-turn-001"}}

# First turn: user asks for details
app.invoke({"messages": [], "action": "batch_update"}, config)
app.invoke(Command(resume="details"), config)

# Second turn: user approves after seeing details
result = app.invoke(Command(resume="approve"), config)

Each interrupt() call writes a checkpoint automatically. No special handling is required to maintain conversation context.

Validating and sanitizing user responses

Never trust user input directly. Validate responses before using them in state updates.

from pydantic import BaseModel, ValidationError

class ApprovalResponse(BaseModel):
    approved: bool
    reason: str | None = None

def validated_approval_node(state: AgentState) -> dict:
    raw_response = interrupt(f"Approve action '{state['action']}'? (yes/no with reason)")
    
    # Sanitize and validate
    parts = raw_response.strip().split(maxsplit=1)
    approval_str = parts[0].lower()
    reason = parts[1] if len(parts) > 1 else None
    
    try:
        validated = ApprovalResponse(
            approved=approval_str in ["yes", "approve", "y"],
            reason=reason
        )
    except ValidationError as e:
        return {"messages": state["messages"] + [f"Invalid response: {e}"]}
    
    return {
        "approved": validated.approved,
        "messages": state["messages"] + [f"User decision: {validated.approved}, reason: {validated.reason}"]
    }

This approach uses Pydantic models to enforce type safety, addressing serialization constraints documented in Pydantic v2 docs.

What are the three architectural patterns for human approval in LangGraph?

Pattern 1: In-graph interrupt() for synchronous UIs. Pattern 2: Conditional edges with external approval service for async workflows. Pattern 3: Hybrid with task queues for enterprise scale. Each trades latency for reliability and complexity.

The right pattern depends on your UX requirements, infrastructure, and fault tolerance needs. Synchronous interrupt() offers the lowest latency and requires persistent WebSocket or SSE connections. It does not survive browser refresh without session affinity. Async conditional edge polling survives disconnections with higher latency and more complex client implementation. External task queues scale to enterprise workloads but require separate infrastructure like Celery or BullMQ, adding operational complexity.

Pattern 1: Synchronous interrupt()

Use this pattern for chat interfaces where the user remains engaged during the approval process. The implementation is straightforward, as shown in the previous section.

Best suited for:

• Real-time chatbots
• Interactive CLI tools
• WebSocket-based applications with sticky sessions

The primary limitation is session affinity. If your load balancer routes the resume request to a different process, that process lacks the in-memory checkpoint. Use sticky sessions or a shared checkpoint backend to resolve this.

Pattern 2: Async conditional edge polling

For email approvals or dashboard review flows, synchronous interrupts block resources unnecessarily. Conditional edges with a polling mechanism handle these use cases efficiently.

import asyncio
from datetime import datetime

class ApprovalService:
    def __init__(self):
        self.pending_approvals = {}
    
    async def request_approval(self, action_id: str, action_details: str) -> str:
        self.pending_approvals[action_id] = {
            "status": "pending",
            "details": action_details,
            "created_at": datetime.utcnow()
        }
        return action_id
    
    async def check_status(self, action_id: str) -> str:
        return self.pending_approvals.get(action_id, {}).get("status", "unknown")
    
    async def approve(self, action_id: str, approved: bool):
        if action_id in self.pending_approvals:
            self.pending_approvals[action_id]["status"] = "approved" if approved else "rejected"

approval_service = ApprovalService()

async def request_approval_node(state: AgentState) -> dict:
    action_id = f"action_{datetime.utcnow().timestamp()}"
    await approval_service.request_approval(action_id, state["action"])
    return {"approval_id": action_id, "approval_status": "pending"}

async def wait_for_approval_node(state: AgentState) -> dict:
    action_id = state["approval_id"]
    
    # Poll with backoff
    for attempt in range(60):  # 60 iterations = ~30 minutes with backoff
        status = await approval_service.check_status(action_id)
        if status in ["approved", "rejected"]:
            return {"approval_status": status}
        await asyncio.sleep(min(30, 2 ** attempt))
    
    return {"approval_status": "timeout"}

def route_approval(state: AgentState) -> str:
    return "execute" if state["approval_status"] == "approved" else "cancel"

def build_async_approval_graph():
    graph = StateGraph(AgentState)
    graph.add_node("request", request_approval_node)
    graph.add_node("wait", wait_for_approval_node)
    graph.add_node("execute", execution_node)
    graph.add_node("cancel", lambda s: {"messages": s["messages"] + ["Cancelled"]})
    
    graph.set_entry_point("request")
    graph.add_edge("request", "wait")
    graph.add_conditional_edges("wait", route_approval, {"execute": "execute", "cancel": "cancel"})
    graph.add_edge("execute", END)
    graph.add_edge("cancel", END)
    
    return graph.compile(checkpointer=MemorySaver())

This pattern decouples the approval request from the decision. An external system (email link, dashboard button) calls approval_service.approve(), and the graph polls for state changes.

Pattern 3: External task queue with checkpoint hooks

Enterprise systems often require approval workflows that integrate with existing task management systems. This pattern uses a hybrid approach with a task queue for approval routing and checkpoints for state persistence.

The architecture separates concerns:

• LangGraph handles the agent logic
• An external task queue (Celery, Temporal) manages approval routing
• Checkpoint hooks sync state between systems

from celery import Celery
from langgraph.checkpoint.postgres import PostgresSaver

celery_app = Celery("approvals", broker="redis://localhost:6379")
checkpointer = PostgresSaver(connection_string="postgresql://user:pass@localhost/langgraph")

@celery_app.task
def create_approval_task(thread_id: str, action: str, callback_url: str):
    # Fetch current state from checkpoint
    config = {"configurable": {"thread_id": thread_id}}
    checkpoint = checkpointer.get(config)
    
    # Create task in external system (Jira, ServiceNow, etc.)
    task_id = external_task_system.create(
        title=f"Approval required: {action}",
        metadata={"thread_id": thread_id, "checkpoint_id": checkpoint.id}
    )
    
    return task_id

@celery_app.task
def process_approval_decision(thread_id: str, approved: bool):
    config = {"configurable": {"thread_id": thread_id}}
    
    # Resume graph execution with decision
    from langgraph.types import Command
    app.invoke(Command(resume="approve" if approved else "reject"), config)

Decision matrix: which pattern for which scenario

Pattern	Latency	Survives Disconnect	Infrastructure	Use Case
Sync interrupt()	<100ms	No	Minimal	Real-time chat, CLI
Async polling	1-60s	Yes	Moderate	Email approval, dashboards
Task queue	Variable	Yes	Complex	Enterprise workflows

For comprehensive context on state machine design, see our guide on ai workflow state machine: 5 design patterns for 2026.

How do you handle checkpoint serialization failures?

Serialization fails when state contains non-serializable objects like file handles or unpicklable custom types. Solutions: implement custom JSON encoders, use Pydantic models for all state, or configure pickle with allowed classes for trusted environments.

LangGraph checkpointers serialize state using JSON or pickle depending on configuration. JSON serialization fails on Python objects that lack a direct JSON representation. Pickle handles most Python objects but fails on certain types including file handles, database connections, and objects with circular references.

The most common error appears as:

TypeError: Object of type datetime is not JSON serializable

This occurs when state contains datetime objects, Decimal instances, or other Python types without JSON equivalents.

Common serialization errors and fixes

Error	Cause	Fix
TypeError: Object of type datetime is not JSON serializable	State contains datetime or Decimal objects	Use Pydantic models with model_dump(mode='json')
TypeError: Object of type set is not JSON serializable	Python sets in state	Convert to lists before returning from nodes
RecursionError: maximum recursion depth exceeded	Circular reference in state	Redesign state structure to avoid cycles
pickle.PicklingError: Can't pickle <class 'function'>	Functions or lambdas in state	Store function names as strings, resolve at runtime

Implementing custom checkpoint serializers

For complex state types, implement custom serialization logic.

import json
from datetime import datetime
from typing import Any

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj: Any) -> Any:
        if isinstance(obj, datetime):
            return {"__datetime__": obj.isoformat()}
        if isinstance(obj, set):
            return {"__set__": list(obj)}
        if hasattr(obj, "model_dump"):
            return obj.model_dump(mode="json")
        return super().default(obj)

def custom_json_encoder(obj: Any) -> str:
    return json.dumps(obj, cls=CustomJSONEncoder)

# Use with Pydantic v2 models
from pydantic import BaseModel

class SafeAgentState(BaseModel):
    messages: list[str]
    action: str
    timestamp: datetime
    tags: set[str]
    
    class Config:
        # Pydantic v2 uses model_config
        json_encoders = {datetime: lambda v: v.isoformat()}

Always validate checkpoint round-trip integrity in tests:

def test_checkpoint_roundtrip():
    state = SafeAgentState(
        messages=["test"],
        action="delete",
        timestamp=datetime.utcnow(),
        tags={"urgent", "manual"}
    )
    
    # Serialize
    serialized = state.model_dump_json()
    
    # Deserialize
    restored = SafeAgentState.model_validate_json(serialized)
    
    assert restored.action == state.action
    assert restored.timestamp.year == state.timestamp.year
    assert restored.tags == state.tags

How do you build production-ready human approval with error recovery?

Production requires idempotent approval handling, checkpoint expiration policies, and graceful degradation when approval services timeout. Implement circuit breakers to skip approval on service failure, with audit logging of all human decisions.

A robust approval system handles failure modes gracefully. The circuit breaker pattern prevents cascading failures when the approval service becomes unavailable. Proper idempotency ensures duplicate requests do not cause side effects. Auditing provides compliance trails for regulated industries.

Idempotency and duplicate approval prevention

Webhook-based approval systems often send duplicate requests on retry. Each resume call must be idempotent.

import hashlib

def generate_approval_token(thread_id: str, action: str) -> str:
    content = f"{thread_id}:{action}"
    return hashlib.sha256(content.encode()).hexdigest()[:16]

class IdempotentApprovalHandler:
    def __init__(self):
        self.processed_tokens = set()
    
    def handle_approval(self, thread_id: str, action