LangGraph in Production: Five Patterns That Keep Agents Safe

An agent that works in a demo has had one good run. An agent in production gets restarted mid-task, hits rate limits, and waits two days for a manager to approve something. Your customer still expects it to finish the job correctly.

LangGraph is the framework we build on for that kind of work. It makes you model the agent as an explicit state machine: named steps, defined transitions, saved state. The framework does not make the agent safe on its own, though. You have to build safety in, and these five patterns are how we do it.

The business case is simple. Each pattern removes one way an agent fails a customer, wastes staff time, or takes an action nobody approved. The failure modes these patterns contain are set out for buyers in the hub post; this one is for the engineers who build the controls. Those failures are engineering gaps, not model limits.

Pattern 1: Persistent checkpointing, with Postgres as the default

The biggest gap between demo and production is resumability. A demo runs to completion in one process. A production agent may run for ten minutes or ten days. Along the way it can be interrupted by a deploy, a container scaling down, or a human who says stop.

LangGraph's checkpointer saves the full graph state at every step. When the process comes back, the graph resumes from the last saved step. Without it, a restart throws away the customer's work and someone redoes it by hand.

Postgres is the right default checkpointer for production. It is durable, it handles many concurrent conversations, and most teams already run and back it up. LangChain ships this as the langgraph-checkpoint-postgres package, which needs a .setup() call to create its tables.

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

DB_URI = os.environ["CHECKPOINT_DB_URI"]

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    await checkpointer.setup()  # creates checkpoint tables on first run
    graph = builder.compile(checkpointer=checkpointer)

    # Each conversation or case gets its own thread_id
    config = {"configurable": {"thread_id": "case-48213"}}
    result = await graph.ainvoke({"user_input": message}, config=config)

Use the SQLite or in-memory savers for local development and tests only. They do not survive a serverless timeout or a second server.

Two kinds of memory, one database

Checkpoints hold execution state: where the workflow is, which tools ran, what came back, what happens next. Agents usually also need long-term memory: customer preferences, past cases, facts that outlive one workflow.

Teams often split these across a vector database, a Redis cache and an application database. That is three bills and three ways to drift out of sync. LangGraph ships a PostgresStore for long-term memory, with optional semantic search. Running checkpoints and memory in one Postgres instance keeps one backup policy, one access-control model and one thing to monitor.

Why "just send the whole history" fails

The shortcut many pilots take is resending the entire conversation to the model on every step. Cost grows faster than most budgets assume. If history grows by g tokens per step over n steps, total input tokens per session are roughly:

g × n × (n + 1) / 2

At 750 tokens per step over 20 steps, the session ends at 15,000 tokens of history. The model has read 157,500 input tokens to get there. Multiply by your per-token price and monthly session count to see your own number. Long prompts also degrade instruction-following, so the agent is more likely to skip an early constraint. Checkpointed state lets each step send only what it needs, and prompt caching makes the part you must resend cheap.

Pattern 2: Approval gates with interrupt()

The dangerous agent is the one that takes an irreversible action unchecked. Sending a customer email, issuing a refund and updating a CRM record all belong behind a human approval gate.

Calling interrupt() inside a node pauses the graph and saves its state through the checkpointer. The agent then uses no compute while it waits. Your API sends the payload to a reviewer, and Command(resume=...) continues the run with their decision. LangChain documents four built-in reviewer responses to an interrupt: approve, edit, reject, or respond directly.

from langgraph.types import interrupt, Command

def human_approval(state: AgentState) -> dict:
    action = state["pending_action"]
    decision = interrupt({
        "type": "approval_required",
        "action": action,
        "risk": state.get("risk_level", "medium"),
    })
    if decision["approved"]:
        return {"approved": True}
    return {"approved": False, "rejection_reason": decision.get("reason", "")}

# API layer, called when the reviewer clicks approve or reject
async def handle_decision(thread_id: str, approved: bool, reason: str = ""):
    config = {"configurable": {"thread_id": thread_id}}
    return await graph.ainvoke(
        Command(resume={"approved": approved, "reason": reason}),
        config=config,
    )

Use one pause mechanism, not two. If you also compile with interrupt_before=["human_approval"], the graph stops once before the node and again inside it. Reviewers then see a phantom pause they have to clear.

Gate any action that is irreversible, costs money per call, is visible outside the company, or rests on inferred intent. Deciding where the gate belongs and what the reviewer sees is a design job in its own right. If the gated action reaches customers and the rules differ by product or region, the gate is a compliance control, not a UX feature. That is the problem Verel Coordinator handles for inbound work: rule-based routing and a compliance check before anything reaches a customer.

Pattern 3: Bounded retries, and errors as state

Every external call fails sometimes: model APIs return rate limits, internal services time out. Without retries, one blip kills the workflow. Unlimited retries are worse. They pile load onto a struggling service and burn your rate limit.

The answer is a retry budget: a fixed number of attempts, with growing delays and random jitter. LangGraph 1.x has this built in. You do not need a hand-rolled loop.

from langgraph.types import RetryPolicy

TRANSIENT = (httpx.TimeoutException, RateLimitError)

builder.add_node(
    "execute_tool",
    execute_tool,
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_interval=1.0,
        backoff_factor=2.0,
        max_interval=30.0,
        jitter=True,
        retry_on=TRANSIENT,
    ),
)

Split failures into two kinds. Transient errors are raised so the policy retries them. If the budget runs out, the run stops at its last checkpoint and can be resumed once the service recovers. Business errors, such as "customer not found" or "invalid amount," will not fix themselves on retry. Catch those and write them into state, so the graph can route them deliberately. A retry policy that catches too much is one of the silent failures that live at the API layer. It masks a permanently malformed request as a transient blip.

async def execute_tool(state: AgentState) -> dict:
    try:
        result = await crm.update(state["pending_action"])  # transient errors propagate
        return {"tool_result": result, "error": None}
    except CRMValidationError as e:
        return {"tool_result": None, "error": str(e)}

def route_after_tool(state: AgentState) -> str:
    return "error" if state.get("error") else "success"

Retries also mean a node can run more than once. Any tool that writes to another system needs an idempotency key, usually derived from the thread ID and step. Otherwise a retry can send the same email twice.

Pattern 4: Typed state, validated where it matters

Many subtle production bugs are state bugs. One node writes customer_id as an integer and the next expects a string. A field the last turn set never gets cleared. Demos do not catch these because demo data is clean.

Define state as a TypedDict with explicit types. Then validate the data that actually enters from outside: model output and tool responses.

from typing import Literal, NotRequired, TypedDict
from pydantic import BaseModel

class PlannedAction(BaseModel):
    tool: Literal["crm_update", "send_email", "refund"]
    args: dict
    description: str

class AgentState(TypedDict):
    session_id: str
    user_input: str
    pending_action: NotRequired[dict | None]
    risk_level: NotRequired[Literal["low", "medium", "high"]]
    approved: NotRequired[bool | None]
    error: NotRequired[str | None]

async def plan_action(state: AgentState) -> dict:
    raw = await planner_llm.ainvoke(state["user_input"])
    action = PlannedAction.model_validate_json(raw.content)  # fails loudly here
    return {"pending_action": action.model_dump(), "approved": None}

You can make the graph state a Pydantic model instead. Know the limit: LangGraph validates the input to the graph, not the partial updates each node returns. A node can still write a bad value unnoticed. Validating at the boundary where untrusted data arrives catches far more.

Pattern 5: Tracing and time travel

A production agent makes many model calls per request, with a routing decision at each step. Without traces, a complaint like "it sometimes gives wrong answers" cannot be diagnosed. You end up guessing at prompt changes.

LangSmith integrates directly with LangGraph and shows the path taken, each node's inputs and outputs, tokens and latency. Langfuse is a good self-hostable alternative and connects through a callback handler. Either way, configure it through the environment and tag every run. LangChain's pricing page lists the LangSmith Developer plan as free for one seat with 5,000 base traces a month, and Plus at $39 per seat.

# Environment, never hardcoded:
# LANGSMITH_TRACING=true
# LANGSMITH_API_KEY=...
# LANGSMITH_PROJECT=support-agent-prod

config = {
    "configurable": {"thread_id": thread_id},
    "metadata": {"tenant": tenant_id, "workflow": "refund_request"},
    "tags": ["production", f"release-{RELEASE}"],
}

Track three numbers from day one:

1. ▸Error rate by node. This tells you where to harden first.
2. ▸Approval rate. Too high and reviewers start rubber-stamping. Too low and the agent may be acting unchecked.
3. ▸Slow-path latency (p95) by graph route. This shows which branches make customers wait.

Traces are also the raw material for step-level evaluation on those traces. A judge grades one tool call against one question, rather than the whole run.

Checkpoints also give you time travel. graph.get_state_history(config) lists every saved step for a thread. If the agent passed a malformed argument at step seven, an operator can correct state with graph.update_state(...) and resume from that checkpoint. Steps one to six are not redone, and the customer does not start over. For the wider set of things that go wrong between demo and launch, see 12 POC-to-production failure modes.

The skeleton, assembled

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)

builder.add_node("understand_intent", understand_intent)
builder.add_node("retrieve_context", retrieve_context)
builder.add_node("plan_action", plan_action)              # Pattern 4
builder.add_node("human_approval", human_approval)        # Pattern 2
builder.add_node("execute_tool", execute_tool,
                 retry_policy=TOOL_RETRY)                 # Pattern 3
builder.add_node("synthesize_response", synthesize_response)
builder.add_node("error_handler", error_handler)

builder.add_edge(START, "understand_intent")
builder.add_edge("understand_intent", "retrieve_context")
builder.add_edge("retrieve_context", "plan_action")

builder.add_conditional_edges("plan_action", route_by_risk, {
    "needs_approval": "human_approval",
    "safe_to_execute": "execute_tool",
})

builder.add_conditional_edges("human_approval",
    lambda s: "execute" if s["approved"] else "rejected", {
    "execute": "execute_tool",
    "rejected": "synthesize_response",
})

builder.add_conditional_edges("execute_tool", route_after_tool, {
    "success": "synthesize_response",
    "error": "error_handler",
})

builder.add_edge("synthesize_response", END)
builder.add_edge("error_handler", END)

# Pattern 1. No interrupt_before: the pause lives inside human_approval.
graph = builder.compile(checkpointer=checkpointer)

Pre-launch gate

Use this table as a go/no-go check. Each row is a failure a customer would notice.

Frequently asked questions

How does LangGraph compare to CrewAI and AutoGen for production? CrewAI and AutoGen let you get a prototype running with less code. LangGraph is more verbose. In return, you get explicit control over state, persistence, pauses and error routes. Production needs exactly those things. The three families disagree on what an agent run even is, and how LangGraph compares with CrewAI and AutoGen comes down to that disagreement.

Does checkpointing slow the agent down? Each step adds one database write. The time it takes depends on state size and how close Postgres is to the agent. For business workflows where each model call takes seconds, the write is rarely what users notice. Keep state lean by storing document IDs rather than whole documents. LangGraph's durability setting also controls when checkpoints are written, if you need to tune it.

Can one agent serve many users at once? Yes. Give each conversation or case its own thread_id, and run threads in parallel against a shared Postgres checkpointer. The one rule is never to run two invocations on the same thread at the same time. Load-test with realistic concurrency before launch, because that is where demo-grade setups usually break first.

This week: run the first row of the gate

Take the first row of the pre-launch table and run it on your current agent, in staging if you have one. Start a run with a real thread_id, kill the process halfway through, and bring it back up. If the same thread resumes at the right step, you have Pattern 1. If it starts over or loses the case, fix that first. It costs less than any model change you are considering.