LangGraph in Production: Patterns and Pitfalls

LangGraph tutorials are written to teach the framework, not to deploy it. They run graphs end to end in a single process against a crafted test prompt. Production systems run inside web servers, queue consumers, or serverless functions — processes that restart, deployments that roll, pods that get evicted. Four assumptions tutorials make that production immediately violates:

These are not edge cases. They are the four walls of production reality. The sections below address each one directly with patterns you can implement today.

When you compile a graph with a checkpointer, each step is persisted to the configured backend under a thread ID. If the process crashes mid run, passing the same thread ID to graph.invoke resumes from the last saved node — no replay of completed work required.

The three checkpoint backends for production:

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

DB_URI = "postgresql://user:password@host:5432/agents"

with psycopg.connect(DB_URI, autocommit=True) as conn:
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()  # creates checkpoint tables on first run

graph = workflow.compile(checkpointer=checkpointer)

# each invocation gets a thread_id; same thread_id resumes from last checkpoint
config = {"configurable": {"thread_id": "run-abc123"}}
result = graph.invoke({"messages": [...]}, config=config)

Store the thread_id with the job record in your database. On restart, pass the same ID and LangGraph skips already-completed nodes.

Define your state schema with Pydantic and enforce a maximum message history length in the validator. Unbounded message history is the direct cause of the state explosion failure mode described later. Add this to every production graph:

from pydantic import BaseModel, field_validator
from typing import Annotated
from langgraph.graph.message import add_messages

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

    @field_validator("messages")
    @classmethod
    def cap_message_history(cls, v):
        MAX_MESSAGES = 40  # tune per your context budget
        if len(v) > MAX_MESSAGES:
            # keep the system message plus the most recent window
            return [v[0]] + v[-(MAX_MESSAGES - 1):]
        return v

LLM API errors (rate limits, transient 500s, timeouts) call for exponential backoff with jitter and a hard retry cap. Tool call failures that return empty or malformed results should not be retried blindly — the agent needs to detect persistent tool failure and route to a fallback or surface the error. Node exceptions that propagate uncaught abort the run without checkpointing, losing all intermediate state.

Here is a retry-wrapped node with a hard cap and structured error handling:

import time
import random

def with_retry(fn, max_attempts: int = 3, base_delay: float = 1.0):
    """Wrap a node function with exponential backoff retry."""
    def wrapper(state):
        last_error = None
        for attempt in range(max_attempts):
            try:
                return fn(state)
            except Exception as e:
                last_error = e
                if attempt < max_attempts - 1:
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                    time.sleep(delay)
        # return structured error state instead of raising
        return {
            "error": f"Node failed after {max_attempts} attempts: {str(last_error)}",
            "failed": True,
        }
    return wrapper

@with_retry
def call_llm_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

The conditional routing function must check for the error state explicitly. A common mistake is writing a should_continue function that only checks for tool calls in the last message, without a branch for the exhausted-retry case:

def should_continue(state) -> str:
    if state.get("failed"):
        return "error_handler"
    if state.get("cost_exceeded"):
        return "error_handler"
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END

To resume after a partial failure, pass the original thread_id to the next invocation. LangGraph replays from the last successful checkpoint, skipping nodes that already completed.

Parallel edges in LangGraph run multiple nodes simultaneously and merge their outputs into a downstream node. This is the right pattern when subtasks are genuinely independent — calling multiple tools in parallel, processing multiple documents simultaneously. Do not reach for parallel edges to speed up a sequential task; graph overhead negates the gain.

The critical detail: if two parallel nodes both write to the same state key, the last writer wins and you silently lose data. Fix this with a reducer annotation — a function that merges parallel writes rather than overwriting. LangGraph’s built-in add_messages is one example. For custom fields, use Python’s operator.add or a custom merge function:

from langgraph.graph import StateGraph, START, END
from operator import add
from typing import Annotated, TypedDict

class ParallelState(TypedDict):
    # add as reducer: parallel writes append rather than overwrite
    results: Annotated[list[str], add]

def fetch_source_a(state: ParallelState) -> dict:
    return {"results": ["result from source A"]}

def fetch_source_b(state: ParallelState) -> dict:
    return {"results": ["result from source B"]}

def synthesize(state: ParallelState) -> dict:
    combined = "
".join(state["results"])
    return {"synthesis": combined}

builder = StateGraph(ParallelState)
builder.add_node("fetch_a", fetch_source_a)
builder.add_node("fetch_b", fetch_source_b)
builder.add_node("synthesize", synthesize)

builder.add_edge(START, "fetch_a")
builder.add_edge(START, "fetch_b")
builder.add_edge("fetch_a", "synthesize")
builder.add_edge("fetch_b", "synthesize")
builder.add_edge("synthesize", END)

When a parallel subtask is complex enough to have its own state schema and conditional routing, encapsulate it as a subgraph rather than expanding it inline. Subgraphs have isolated state namespaces, which eliminates the parallel write race condition entirely for that subtask's internal state.

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your_api_key
LANGCHAIN_PROJECT=your_project_name

Every graph invocation is automatically traced: node entry and exit, inputs and outputs, LLM calls with token counts, tool calls with arguments and return values. No code changes required beyond the environment variables.

Three metrics predict production failure before it happens:

Docker on a single host is the fastest path to production for low-volume internal tooling. The LangGraph CLI (langgraph up) builds and runs a Docker image from your project directory with production settings. The limit is the host: one process pool means memory-bound graphs with large state objects cap out quickly.

Kubernetes suits teams running parallel agents — many concurrent graph invocations serving user traffic or processing job queues. Deploy the LangGraph server as a Kubernetes Deployment with horizontal pod autoscaling based on queue depth or request rate. Keep the checkpoint backend (Postgres) outside the pod cluster in a managed service so state survives pod evictions. Configure readiness probes to prevent traffic routing to pods before they are warmed up.

Serverless (Cloud Run, AWS Lambda) suits event driven or batch workflows where invocations are infrequent and async. The main trade off is cold start latency: a Python LangGraph container cold starts in 2 to 8 seconds depending on dependency weight. Use minimum instances or provisioned concurrency for latency-sensitive paths.

One rule that applies to all three targets: the LangGraph graph process should be stateless. All state lives in the checkpoint backend. This makes the process horizontally scalable — any instance can pick up any run by thread ID. Architect for this from day one; retrofitting stateful graph code to a distributed checkpoint model is painful.