intermediate⏱45 minLesson 3 of 5

Persistence, Checkpointing and Threads

Explore MemorySaver, per-thread checkpointing, replaying from checkpoints, branching from past states, and PostgresSaver for production.

Persistence, Checkpointing and Threads

One of LangGraph's most powerful features is persistence. Every step of a graph execution can be saved as a checkpoint, enabling replay, rollback, and branching from any prior state.

Mermaid: Checkpoint Lifecycle

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Every invoke() call with a thread_id triggers a checkpoint save after each node. Checkpoints are stored sequentially within a thread, forming an append-only timeline.

MemorySaver

MemorySaver is the simplest checkpointing backend. It stores checkpoints in memory and is ideal for prototyping.

python

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

# Create a checkpoint saver
memory = MemorySaver()

# Pass it when compiling
app = builder.compile(checkpointer=memory)

# Each invocation needs a thread config
config = {"configurable": {"thread_id": "session-1"}}
result = app.invoke({"messages": ["Hello"]}, config)

⚠️Warning

MemorySaver is ephemeral — all checkpoints are lost when the Python process ends. Use PostgresSaver or a custom saver for production workloads.

Comparison: Checkpoint Backends

Feature	MemorySaver	PostgresSaver	Custom Saver
Persistence	In-memory	PostgreSQL	User-defined
Production-ready	No	Yes	Depends on impl.
Thread isolation	Yes	Yes	Yes
Replay support	Yes	Yes	Must implement
Branching support	Yes	Yes	Must implement
Setup complexity	None	Requires DB schema	High
Cross-process survival	No	Yes	Depends on impl.
Scalability	Single process	Multi-process	User-defined
Cost	Free	Storage costs	Variable

Mermaid: Thread State Diagram

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Each thread transitions through active running states and checkpointed pause points. Branches create entirely new thread lineages from a parent checkpoint.

Checkpointing States Per Thread

A thread is a conversation or execution session identified by a thread_id. LangGraph stores a new checkpoint after every node execution within a thread.

python

# Same graph, same thread — state accumulates
app.invoke({"messages": ["Turn 1"]}, {"configurable": {"thread_id": "t1"}})
app.invoke({"messages": ["Turn 2"]}, {"configurable": {"thread_id": "t1"}})

# Different thread — isolated state
app.invoke({"messages": ["Thread 2 start"]}, {"configurable": {"thread_id": "t2"}})

State is namespaced per thread. Checkpoints form a timeline within each thread.

📌Important

Thread isolation is critical in production. Each user session gets its own thread_id — never share a thread_id across users. Use a unique identifier like user_id:conversation_id as the thread ID to guarantee isolation.

Replaying from Checkpoints

You can replay execution from a specific checkpoint by providing a checkpoint_id.

python

# Get the parent checkpoint ID from the last run
parent_id = result["__run"]["checkpoint_id"]

# Replay from that checkpoint
replayed = app.invoke(
    {"messages": ["New message"]},
    {"configurable": {"thread_id": "t1", "checkpoint_id": parent_id}}
)

Replay does not re-execute nodes before the checkpoint — it resumes from that exact state.

💡Tip

Replay is invaluable for testing and debugging. You can replay a specific checkpoint with modified input to see how the graph would behave with different data at that exact state.

Checkpoint Replay with thread_id

python

import uuid

def replay_thread(app, thread_id: str, checkpoint_id: str, new_input: dict):
    """Utility function to replay a checkpoint."""
    config = {
        "configurable": {
            "thread_id": thread_id,
            "checkpoint_id": checkpoint_id,
        }
    }
    return app.invoke(new_input, config)

# Usage
result = replay_thread(
    app,
    thread_id="user-123",
    checkpoint_id="1ef345ab...",
    new_input={"messages": ["Corrected query"]}
)

Branching from Past States

You can fork a thread at any checkpoint, creating a branch that diverges from the original timeline.

python

# Fork from an earlier checkpoint
fork_config = {
    "configurable": {
        "thread_id": "t1-branch-1",
        "checkpoint_id": parent_id
    }
}
fork_result = app.invoke({"messages": ["Branch message"]}, fork_config)

The branch starts with the state of the parent checkpoint and proceeds independently. This is useful for "what-if" analysis or human corrections.

Branch Creation Example

python

def create_branch(app, original_thread: str, checkpoint_id: str, branch_suffix: str, new_input: dict):
    """Create a branch from a checkpoint and execute it."""
    branch_thread = f"{original_thread}-{branch_suffix}"
    config = {
        "configurable": {
            "thread_id": branch_thread,
            "checkpoint_id": checkpoint_id,
        }
    }
    return app.invoke(new_input, config)

# Compare two branches from the same checkpoint
branch_a = create_branch(app, "session-1", ckpt_id, "rollback", {"messages": ["Try A"]})
branch_b = create_branch(app, "session-1", ckpt_id, "experiment", {"messages": ["Try B"]})

# Analyze which branch produced better results

💡Tip

Branching enables A/B testing of agent decisions. Run multiple branches from the same checkpoint with different prompts, parameters, or routes, then compare outcomes to optimize your agent.

Checkpoint Storage Costs

⚠️Warning

Every node execution creates a full state checkpoint. If your state is large (e.g., embedding vectors, full conversation histories), checkpoint storage can grow quickly. Consider:

Using PostgresSaver with table partitioning by thread_id
Implementing a retention policy that prunes old checkpoints
Keeping state schemas lean — only store what downstream nodes need
Using custom savers with S3/GCS lifecycle policies

PostgresSaver for Production

PostgresSaver persists checkpoints to a PostgreSQL database, surviving restarts.

python

from langgraph.checkpoint import PostgresSaver
import asyncpg

# Connect to PostgreSQL
conn = await asyncpg.connect("postgresql://user:pass@localhost/langgraph")
saver = PostgresSaver(conn)

# Compile with the production saver
app = builder.compile(checkpointer=saver)

# State survives process restarts
result = await app.ainvoke({"messages": ["Hello"]}, {"configurable": {"thread_id": "prod-1"}})

bash

# Schema setup (run once)
pip install langgraph-checkpoint-postgres
python -c "from langgraph.checkpoint import PostgresSaver; PostgresSaver.create_tables('postgresql://user:pass@localhost/langgraph')"

Production PostgresSaver Setup

python

import asyncpg
from langgraph.checkpoint import PostgresSaver
from contextlib import asynccontextmanager

@asynccontextmanager
async def get_graph_app():
    """Production-ready graph with Postgres persistence."""
    conn = await asyncpg.connect(
        user="app_user",
        password="app_password",
        host="postgres.example.com",
        port=5432,
        database="langgraph_prod",
        # Connection pooling is recommended for production
        min_size=5,
        max_size=20,
    )
    try:
        saver = PostgresSaver(conn)
        app = builder.compile(checkpointer=saver)
        yield app
    finally:
        await conn.close()

# Usage
async with get_graph_app() as app:
    result = await app.ainvoke(
        {"messages": ["Process order 12345"]},
        {"configurable": {"thread_id": "order:12345:user:678"}}
    )

Mermaid: Checkpoint Timeline

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Each checkpoint is a snapshot of the full state. Branches fork from a parent checkpoint and create their own timeline.

Mermaid: Thread Isolation Visualization

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Thread isolation ensures that User 1's conversation never leaks into User 2's state. Each thread has its own independent checkpoint chain.

Practice Question

Which saver is appropriate for production use?

Practice Question

What identifies a unique execution session in LangGraph?

Practice Question

What happens when you replay from a checkpoint?

Practice Question

What is a branch in LangGraph checkpointing?

Practice Question

Which of the following is NOT a feature of MemorySaver?

Practice Question

Scenario: You need to debug why an agent gave a wrong answer two turns ago. You have the checkpoint IDs. What should you do?

✅Success

Key Takeaways

MemorySaver stores checkpoints in-process memory for prototyping.
Each thread (thread_id) has an independent state and checkpoint timeline.
Replaying from a checkpoint resumes execution without re-running prior steps.
Branching forks a timeline from any past checkpoint for experimentation.
PostgresSaver provides production-grade persistence with full replay and branching.
Checkpoints are stored after every node execution by default.
Custom savers can implement any backend (Redis, S3, etc.).
Thread isolation is essential for multi-tenant production systems.
Use unique thread IDs per user session to prevent state leakage.
Monitor checkpoint storage costs with large state schemas.

Progress60%

Nodes, Edges and Conditional Flow

Human-in-the-Loop, Breakpoints and Dynamic Control

Course Content

1. LangGraph Fundamentals and State Graphs 2. Nodes, Edges and Conditional Flow 3. Persistence, Checkpointing and Threads 4. Human-in-the-Loop, Breakpoints and Dynamic Control 5. Multi-Agent Orchestration and Subgraphs