langgraph-checkpoints（语言图检查点）

LangGraph Checkpointing

Persist workflow state for recovery and debugging.

Checkpointer Options

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

# Development: In-memory
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Production: SQLite
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = workflow.compile(checkpointer=checkpointer)

# Production: PostgreSQL
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = workflow.compile(checkpointer=checkpointer)

Using Thread IDs

# Start new workflow
config = {"configurable": {"thread_id": "analysis-123"}}
result = app.invoke(initial_state, config=config)

# Resume interrupted workflow
config = {"configurable": {"thread_id": "analysis-123"}}
result = app.invoke(None, config=config)  # Resumes from checkpoint

PostgreSQL Setup

def create_checkpointer():
    """Create PostgreSQL checkpointer for production."""
    return PostgresSaver.from_conn_string(
        settings.DATABASE_URL,
        save_every=1  # Save after each node
    )

# Compile with checkpointing
app = workflow.compile(
    checkpointer=create_checkpointer(),
    interrupt_before=["quality_gate"]  # Manual review point
)

Inspecting Checkpoints

# Get all checkpoints for a workflow
checkpoints = app.get_state_history(config)

for checkpoint in checkpoints:
    print(f"Step: {checkpoint.metadata['step']}")
    print(f"Node: {checkpoint.metadata['source']}")
    print(f"State: {checkpoint.values}")

# Get current state
current = app.get_state(config)
print(current.values)

Resuming After Crash

import logging

async def run_with_recovery(workflow_id: str, initial_state: dict):
    """Run workflow with automatic recovery."""
    config = {"configurable": {"thread_id": workflow_id}}

    try:
        # Try to resume existing workflow
        state = app.get_state(config)
        if state.values:
            logging.info(f"Resuming workflow {workflow_id}")
            return app.invoke(None, config=config)
    except Exception:
        pass  # No existing checkpoint

    # Start fresh
    logging.info(f"Starting new workflow {workflow_id}")
    return app.invoke(initial_state, config=config)

Step-by-Step Debugging

# Execute one node at a time
for step in app.stream(initial_state, config):
    print(f"After {step['node']}: {step['state']}")
    input("Press Enter to continue...")

# Rollback to previous checkpoint
history = list(app.get_state_history(config))
previous_state = history[1]  # One step back
app.update_state(config, previous_state.values)

Store vs Checkpointer (2026 Best Practice)

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

# Checkpointer = SHORT-TERM memory (thread-scoped)
# - Conversation history within a session
# - Workflow state for resume/recovery
# - Scoped to thread_id

checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)

# Store = LONG-TERM memory (cross-thread)
# - User preferences across sessions
# - Learned facts about users
# - Shared across ALL threads for a user

store = PostgresStore.from_conn_string(DATABASE_URL)

# Compile with BOTH for full memory support
app = workflow.compile(
    checkpointer=checkpointer,  # Thread-scoped state
    store=store                  # Cross-thread memory
)

Using Store for Cross-Thread Memory

from langgraph.store.base import BaseStore

async def agent_with_memory(state: AgentState, *, store: BaseStore):
    """Agent that remembers across conversations."""
    user_id = state["user_id"]

    # Read cross-thread memory (user preferences)
    memories = await store.aget(namespace=("users", user_id), key="preferences")

    # Use memories in agent logic
    if memories and memories.value.get("prefers_concise"):
        state["system_prompt"] += "\nBe concise in responses."

    # Update cross-thread memory (learned facts)
    await store.aput(
        namespace=("users", user_id),
        key="last_topic",
        value={"topic": state["current_topic"], "timestamp": datetime.now().isoformat()}
    )

    return state

# Register node with store access
workflow.add_node("agent", agent_with_memory)

Memory Architecture

┌─────────────────────────────────────────────────────────────┐
│                    User: alice                               │
├─────────────────────────────────────────────────────────────┤
│  Thread 1 (chat-001)    │  Thread 2 (chat-002)              │
│  ┌─────────────────┐    │  ┌─────────────────┐              │
│  │ Checkpointer    │    │  │ Checkpointer    │              │
│  │ - msg history   │    │  │ - msg history   │              │
│  │ - workflow pos  │    │  │ - workflow pos  │              │
│  └─────────────────┘    │  └─────────────────┘              │
├─────────────────────────────────────────────────────────────┤
│                     Store (cross-thread)                     │
│  namespace=("users", "alice")                                │
│  - preferences: {prefers_concise: true}                     │
│  - last_topic: {topic: "langgraph", timestamp: "..."}       │
└─────────────────────────────────────────────────────────────┘

Graph Migrations (2026 Feature)

LangGraph handles topology changes automatically:

# Safe changes (handled automatically):
# - Adding new nodes
# - Removing nodes
# - Renaming nodes
# - Adding state keys
# - Removing state keys

# Works for both active and completed threads
# Limitation: Cannot remove node if thread is interrupted at that node

Checkpoint Cleanup Strategies

from datetime import datetime, timedelta

# Option 1: TTL-based cleanup (configure at DB level)
# CREATE INDEX idx_checkpoints_created ON checkpoints(created_at);
# DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days';

# Option 2: Manual cleanup
async def cleanup_old_checkpoints(db, days: int = 30):
    """Remove checkpoints older than N days."""
    cutoff = datetime.now() - timedelta(days=days)
    await db.execute(
        "DELETE FROM langgraph_checkpoints WHERE created_at < $1",
        cutoff
    )

# Option 3: Per-thread cleanup
async def cleanup_thread(db, thread_id: str, keep_latest: int = 10):
    """Keep only latest N checkpoints per thread."""
    await db.execute("""
        DELETE FROM langgraph_checkpoints
        WHERE thread_id = $1
        AND id NOT IN (
            SELECT id FROM langgraph_checkpoints
            WHERE thread_id = $1
            ORDER BY created_at DESC
            LIMIT $2
        )
    """, thread_id, keep_latest)

Key Decisions

Common Mistakes

• No checkpointer in production (lose progress)
• Random thread IDs (can't resume)
• Not handling missing checkpoints
• Using only checkpointer for user preferences (lost across threads)
• Not using namespaces in Store (data collisions)
• Not cleaning up old checkpoints (database bloat)
• Removing nodes while threads are interrupted at them

Evaluations

See references/evaluations.md for test cases.

Related Skills

• langgraph-state - State schemas that persist well with checkpointing
• langgraph-human-in-loop - Interrupt patterns that leverage checkpoints
• langgraph-supervisor - Checkpoint supervisor progress for fault tolerance
• langgraph-streaming - Stream checkpoint updates to clients
• langgraph-functional - Functional API with automatic checkpointing
• database-schema-designer - PostgreSQL checkpoint table setup

Capability Details

checkpoint-saving

Keywords: save checkpoint, checkpoint, persist state, save state Solves:

• Save workflow state at key points
• Implement checkpoint strategies
• Handle checkpoint serialization

checkpoint-loading

Keywords: load checkpoint, restore, resume, recovery Solves:

• Resume workflows from checkpoints
• Implement state recovery
• Handle checkpoint versioning

memory-backends

Keywords: memory backend, MemorySaver, SqliteSaver, PostgresSaver Solves:

• Configure checkpoint storage backends
• Choose between memory/SQLite/Postgres
• Implement custom checkpoint storage

async-checkpoints

Keywords: async checkpoint, AsyncSqliteSaver, async persistence Solves:

• Implement async checkpoint operations
• Handle concurrent checkpoint access
• Optimize checkpoint performance

conversation-history

Keywords: conversation, history, message history, thread Solves:

• Persist conversation history
• Implement thread-based checkpoints
• Manage conversation state