AI agents as state graphs in LangGraph: checkpoints, interrupts and replay

The simplest agent is a loop: send the conversation to the model, execute the tool calls it requests, append the results, repeat until it stops requesting tools. It works for demos and for agents with one or two tools. Then the agent has to survive a process restart, or wait for a person, or someone has to debug a run days after it happened.

The alternative is explicit control flow. The agent becomes a state graph: nodes are steps, edges are named transitions, the state is a typed structure, and the runtime persists it after every step. The model makes its decision inside a node, and the graph picks the next step. The examples use LangGraph. Any orchestrator with the same primitives would do.

What the loop lacks

Start with retries. A failed tool call inside the loop gets retried by the model, by your code, or not at all, and the only global safeguard is a step counter. You can't say "try retrieval up to three times, never retry the payment call".

Intermediate state isn't persisted, because a loop keeps it in local variables. If the process dies at step 7, it's gone, and restarting from scratch repeats steps 1 to 6 together with whatever side effects they had. For the same reason there's no way to wait for a person: approval means stopping mid-run, sometimes for days, and continuing in a different process that has nothing to pick up from.

And replay. A loop leaves a log at best, and to debug a bad run you need the exact state from before the wrong decision, plus the ability to run from there again.

The loop's stop condition can fail too. I tested a ready-made agent harness with a local model: in non-interactive mode, one answer without a tool call was enough to end the loop, so the model stopped halfway through the task. The harness passed the resumed run 26,600 tokens of history, and the model still claimed it couldn't see any of it. That was a model error. With an instruction to work until the end and not ask questions, the fix finally went through, but it took 19 minutes to change one character. The tool calls themselves were correct in every run.

The graph

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import RetryPolicy


class AgentState(TypedDict):
    task: str
    plan: list[str]
    context: Annotated[list[str], operator.add]  # appended, not overwritten
    result: str | None
    needs_context: bool
    done: bool
    reflections: int


def route_after_plan(state: AgentState) -> str:
    return "retrieve" if state["needs_context"] else "act"


def route_after_act(state: AgentState) -> str:
    if state["done"] or state["reflections"] >= 2:
        return END
    return "reflect"


builder = StateGraph(AgentState)
builder.add_node("plan", planner)
builder.add_node("retrieve", retriever, retry_policy=RetryPolicy(max_attempts=3))
builder.add_node("act", tool_runner)
builder.add_node("reflect", critic)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", route_after_plan, ["retrieve", "act"])
builder.add_edge("retrieve", "act")
builder.add_conditional_edges("act", route_after_act, ["reflect", END])
builder.add_edge("reflect", "plan")

planner, retriever, tool_runner and critic are ordinary functions that take the state and return a partial update. The routing functions are plain Python and can be unit-tested without a model.

reflections is a business rule kept in the state: at most two critique rounds. recursion_limit in the run config is a separate, technical cap on the total number of steps. Go over it and you get GraphRecursionError instead of a run that never ends. The default limit in langgraph 1.2 is 10007 steps, so set your own explicitly, as in the example below. The retry policy applies to the retrieve node only, so it never repeats a tool with side effects. The search isn't retried on every error either. Unless you set your own retry_on, RetryPolicy doesn't retry ValueError, RuntimeError, OSError and a few other built-in errors, or 4xx responses raised by httpx. It does retry the built-in ConnectionError, httpx 5xx and any other exception, including your own classes. With requests the default gets it backwards: an HTTPError is retried whatever the status, while requests.ConnectionError and Timeout subclass OSError and are not retried. If you call HTTP through requests, pass your own retry_on.

Checkpoints in Postgres

A checkpointer stores the state after each step (a "super-step" in LangGraph terms), keyed by a thread ID. The Postgres implementation is in the langgraph-checkpoint-postgres package and uses psycopg 3.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://agent:secret@localhost:5432/agents"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates the checkpoint tables; run once, like a migration
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "loan-app-20931"}, "recursion_limit": 30}
    graph.invoke({"task": "...", "context": [], "reflections": 0}, config)

After a crash, graph.invoke(None, config) with the same thread ID continues from the last completed step. graph.get_state(config) returns the current state and the next node; graph.get_state_history(config) returns all previous snapshots.

Each checkpoint can restore the whole state. Large strings in the state (full retrieved documents, base64 files) are written into every checkpoint row, and a list extended by a reducer is written out in full on every change. Keep references in the state and the payloads somewhere else. Checkpoints also pile up per thread indefinitely unless you add a retention job. The package has delete_thread() and prune() for that, and prune() with the keep_latest strategy keeps only the latest checkpoint. ShallowPostgresSaver has been deprecated since version 2.0.20 of the package. In a regulated environment the retention job has to match your data retention policy anyway.

Waiting for a human

interrupt() stops the graph inside a node, saves the state and returns the payload to the caller. The run is resumed later, from any process, with Command(resume=...).

from langgraph.types import Command, interrupt


def approve_refund(state: AgentState) -> dict:
    # Assumes `amount` and `approved` are declared in `AgentState`.
    answer = interrupt({"question": "Approve refund?", "amount": state["amount"]})
    return {"approved": answer == "yes"}


# Hours later, e.g. in the HTTP handler of an approval screen:
graph.invoke(Command(resume="yes"), {"configurable": {"thread_id": "loan-app-20931"}, "recursion_limit": 30})

On resume, the node runs again from its first line, and interrupt() returns the resume value instead of stopping. So any code before interrupt() runs twice. Keep that part free of side effects, or move the side effect to its own node after the approval.

Also check how long the runtime is willing to wait. One agent orchestrator had a hard-coded 15-minute limit on the pause for human approval. After that the whole run ended with an error and none of the next steps started. Approvals often come hours later, so the pause has to be saved with the run state.

What a checkpoint does not protect

The checkpoint is written after a node finishes. If a node sends an email, charges a card or streams output to the user and the process dies before the checkpoint is written, resuming runs the node again. Per node, the guarantee is at-least-once.

Put one external side effect in a node, so the unit that may repeat is small and you know what it is. Derive an idempotency key from the thread ID, the node name and the step number (langgraph_step in the config metadata), because in a graph with a loop the same node runs more than once per thread, and a rerun after a crash gets the same step number. Pass the key to the external system if it accepts one (Stripe, for example, takes an Idempotency-Key header). In API v1, Stripe may remove a key once it is 24 hours old, so a rerun after a pause of several days can go through as a new request. On resume the node streams again, so the client should treat streamed output as a preview and replace what it already displayed with the value the node returned into the checkpointed state, instead of appending a second copy.

from langchain_core.runnables import RunnableConfig


def send_confirmation(state: AgentState, config: RunnableConfig) -> dict:
    # Assumes `customer_email` and `confirmation_sent` are declared in `AgentState`.
    thread_id = config["configurable"]["thread_id"]
    step = config["metadata"]["langgraph_step"]
    # `mailer` is your client; the provider deduplicates by this key.
    mailer.send(
        to=state["customer_email"],
        body=state["result"],
        idempotency_key=f"{thread_id}:send_confirmation:{step}",
    )
    return {"confirmation_sent": True}

I split work for my coding agents the same way: packages of about 250–300 lines of diff. Whole phases (550–930 lines) came back with corrections roughly one time in three.

Replay for debugging and regression

Every snapshot from get_state_history has its own config with a checkpoint ID. Invoking the graph with None and that config runs again from that point and creates a new branch of history; the original run stays intact. If nodes receive their tools through a factory (build_graph(tools)), you can take the state from a failed production run, build the graph with recorded or mocked tools, and check whether a changed prompt or routing rule leads to a different decision. Production state may contain personal data. The replay environment needs the same access controls as production.

Observability

Log one structured event per node execution. A wrapper at registration time covers every node without touching its code:

import inspect
import logging
import time

from langgraph.errors import GraphBubbleUp

log = logging.getLogger("agent")


def observed(name, fn):
    # A node may take the state alone or (state, config).
    takes_config = len(inspect.signature(fn).parameters) > 1

    # No functools.wraps: LangGraph would then read fn's signature and, for a
    # state-only node, would not pass config to the wrapper.
    def wrapper(state, config):
        started = time.perf_counter()
        error = None
        try:
            return fn(state, config) if takes_config else fn(state)
        except GraphBubbleUp:  # interrupt() pauses the graph by raising it
            raise
        except Exception as exc:
            error = type(exc).__name__
            raise
        finally:
            log.info("agent_step", extra={
                "thread_id": config["configurable"]["thread_id"],
                "node": name,
                "latency_ms": round((time.perf_counter() - started) * 1000),
                "error": error,
            })
    return wrapper


# Instead of builder.add_node("act", tool_runner); a node name can be registered only once.
builder.add_node("act", observed("act", tool_runner))

Token counts belong in the same event. In LangChain chat models they are available on the response message as usage_metadata; return them from the node or log them where the model is called. With thread ID, node, latency, tokens and error in one record, the cost of a run, a slow tool or a node that started failing after a deploy is one query away.

One step run by Codex got the wrong model name, the API rejected the request in 2 seconds, and the orchestrator running it marked the step as done. Tests and lint passed afterwards, on code that hadn't changed at all.

When a graph is overkill

A single model call with structured output, or a fixed sequence of steps without branching, is better served by a plain function. A short synchronous task that nobody approves and nobody will replay doesn't justify a checkpoint table and an extra database write on every step. The graph starts paying for itself when a run can take long enough to get interrupted, when someone approves something halfway through, or when a failed run has to be reproduced later.