How to checkpoint LLM agent progress with LangGraph

LangGraph treats agents as state machines, but a process restart or unhandled exception throws away all in-flight progress. langgraph checkpointing solves this by serializing graph state to a durable backend after every step, so you can resume a stalled run without replaying expensive LLM calls. This guide builds a runnable Postgres-backed agent that survives crashes and shows exactly where to hook in.

Step 1: Install dependencies and pick a backend

For anything beyond a laptop prototype, use a real database. The official Postgres checkpointer is battle-tested and supports the exact semantics LangGraph needs: per-thread, per-step state with atomic writes. SQLite works for local dev via MemorySaver or SqliteSaver, but it won’t handle concurrent workers or network restarts.

pip install "langgraph" "langgraph-checkpoint-postgres" "psycopg[binary]" "langchain-openai"

If you already run LangChain, pin langgraph>=0.2.0 to get the stable checkpointer API. Avoid rolling your own pickle-to-Redis layer; the checkpoint contract includes blob storage and parent pointers that are easy to get wrong.

Step 2: Provision the Postgres checkpointer

Create a database and a connection. PostgresSaver expects a psycopg connection (or pool). Call setup() once to create the checkpoints and checkpoint_blobs tables. This call is idempotent.

import psycopg
from langgraph.checkpoint.postgres import PostgresSaver

conn = psycopg.connect(
    "postgres://user:password@localhost:5432/agentdb",
    autocommit=True,
)
checkpointer = PostgresSaver(conn)
checkpointer.setup()  # creates tables if missing

Keep the connection alive for the process lifetime, or use psycopg_pool.ConnectionPool in production. The checkpointer writes a row per state transition keyed by thread_id and checkpoint_id. Each row stores the full state snapshot plus metadata, so you get time-travel debugging for free.

Step 3: Define agent state and graph nodes

LangGraph state is a TypedDict (or Pydantic model). For a chat-style agent, store messages. We’ll add a node that calls an LLM. To avoid vendor lock and get automatic fallback across providers, point ChatOpenAI at n4n.ai’s OpenAI-compatible endpoint—one URL covers 240+ models and forwards cache-control hints, so a rate-limited provider doesn’t kill your run.

from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

llm = ChatOpenAI(
    model="openai/gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",
    api_key="your-n4n-key",
    temperature=0,
)

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

The add_messages reducer appends new messages; checkpoints store the full list each step. Do not store large binary blobs in state—Postgres rows aren’t meant for 10 MB payloads. Keep state lean: IDs, text, and small dicts.

Step 4: Compile the graph with the checkpointer

Wire a single-node graph (extend with tools later) and attach the checkpointer at compile time. This is the only integration point for langgraph checkpointing—no manual save calls inside nodes.

builder = StateGraph(AgentState)
builder.add_node("agent", call_model)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

graph = builder.compile(checkpointer=checkpointer)

If you forget the checkpointer argument, state lives only in memory and vanishes on exit. The compiled graph is thread-safe as long as the underlying checkpointer is.

Step 5: Run a thread and generate checkpoints

Every invocation needs a thread_id in the config. This ID is your resume handle. Use a business key (ticket ID, user ID + task) rather than a random UUID you can’t map later.

config = {"configurable": {"thread_id": "support-ticket-882"}}

initial = {"messages": [("user", "Summarize the last 3 orders for customer 123.")]}
result = graph.invoke(initial, config)
print(result["messages"][-1].content)

After this call, Postgres holds a checkpoint for the START, the agent node, and END. Inspect the current state:

state = graph.get_state(config)
print(state.values["messages"])

List the full history to see each step:

for checkpoint in graph.get_state_history(config):
    print(checkpoint.id, checkpoint.next)

That history is what makes crash recovery and rollback possible.

Step 6: Crash, then resume from the last checkpoint

Real systems die mid-run. Add a second node that writes to a CRM, and simulate a failure.

def write_to_crm(state: AgentState):
    raise RuntimeError("CRM down")

builder.add_node("crm", write_to_crm)
builder.add_edge("agent", "crm")
builder.add_edge("crm", END)
graph = builder.compile(checkpointer=checkpointer)

First run fails at crm. The agent checkpoint is safe. Fix the CRM code, restart the process (re-create checkpointer and graph), and re-invoke with the same thread_id:

# new process, same checkpointer setup as Step 2
graph.invoke(None, config)  # None input merges with existing state

LangGraph loads the last checkpoint, skips the agent node, and runs crm. No duplicate LLM spend. You can also jump to an earlier point using a specific checkpoint ID:

old = list(graph.get_state_history(config))[-2]
graph.invoke(None, {"configurable": {"thread_id": "support-ticket-882", "checkpoint_id": old.id}})

That’s langgraph checkpointing used for time-travel, not just crash recovery.

Step 7: Verify success

Verification is concrete:

1. Query the database:

psql agentdb -c "SELECT thread_id, checkpoint_id FROM checkpoints;"

You should see rows for support-ticket-882.

2. In Python, assert state round-trips:

state = graph.get_state(config)
assert any("Summarize" in m.content for m in state.values["messages"])

3. Force a resume: truncate the crm node logic to a no-op, run graph.invoke(None, config), and confirm the agent message appears only once in the message list (no replay). If the message count increments on resume, your reducer or checkpoint config is wrong.

4. Kill the process with kill -9 between the agent and CRM nodes (add a time.sleep in write_to_crm). After restart, the resume must complete without calling the LLM again. Log llm.invoke calls to prove it.

Step 8: Production caveats

• Concurrency: Use a connection pool. PostgresSaver from a single sync connection blocks under parallel threads. psycopg_pool.ConnectionPool with open=True is the right call.
• Retention: Checkpoints accumulate. Add a cron to delete old thread_ids after 30 days unless you need audit trails. The tables have no built-in TTL.
• Secrets: Never put API keys or tokens in state. LangGraph serializes the whole state dict; use a separate secrets store referenced by ID.
• Human-in-the-loop: Use graph.interrupt() to pause; the checkpoint captures the pause point. Resume with graph.invoke(None, config) after approval. The interrupt does not consume the LLM call again.
• Model routing: When you rely on a gateway like n4n.ai, set the model per request and let the gateway handle provider degradation. Your checkpoint logic stays identical regardless of which backend served the token.

langgraph checkpointing turns fragile scripts into recoverable services. The pattern is unchanged whether you run one agent or ten thousand; the thread_id is the unit of resumability, and Postgres is the cheapest insurance you can buy.