LangGraph Production Agent SOP: Timeouts, Retries, and Fault Tolerance

The hardest part of building a production AI Agent is not prompt engineering--it is resilience. An LLM API call hangs and kills the entire chain. A tool throws an exception with no fallback. A process restart wipes all intermediate state. A human review step has no way to pause. These failures stay hidden in demos and explode in production. LangGraph 1.x turns these into configurable engineering parameters: per-node timeouts, retry policies, error handlers, global state persistence via checkpointer, and interrupt-based human approval breakpoints.

This SOP walks through building a production-grade Agent with LangGraph end to end: defining typed state, attaching a checkpointer for durability, configuring TimeoutPolicy to prevent hangs, using RetryPolicy for automatic retries, wiring error_handler for fallback, and implementing human-in-the-loop with interrupt. Each step includes real Python code and a parameter reference table, followed by a complete runnable Agent example and pitfall log. All APIs below are verified against LangGraph 1.2.

1. Why LangGraph: Durable State Is the Foundation

The fundamental difference between a production Agent and a demo Agent is one concept: durable state. A regular Agent script crashes mid-run and loses everything--you start over. LangGraph's StateGraph persists execution state to a checkpointer after every node, so a crashed process can resume from the last checkpoint, and long-running tasks can span processes and machines.

According to an independent 2026 comparison (2000 runs, 5 tasks x 100 iterations), LangGraph showed the best latency and token cost predictability--each LLM call is a discrete, known quantity, unlike frameworks that bundle multi-step calls into a black box. Among LangGraph, CrewAI, and AutoGen (AG2), LangGraph was rated "most production-ready" because it offers durable execution, fine-grained error handling, and enterprise observability as a cohesive package.

LangGraph v0.4 (April 2026) further improved state persistence and human-in-the-loop checkpoints; by 1.x these features are stable. The following six steps build a complete production Agent.

2. Core: StateGraph and Typed State

The center of LangGraph is the StateGraph--a directed graph where each node reads and writes a shared typed state. First, define the state:

from typing import Annotated
from typing_extensions import TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # reducer: append, not overwrite
    query: str
    research: str
    draft: str
    approved: bool

Annotated[list, operator.add] tells LangGraph: when multiple nodes write to messages, merge by appending (operator.add) rather than last-write-wins. This is the key to typed state--each field's merge strategy is explicitly declared, preventing state races.

After defining the state, build the graph skeleton:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)
# Nodes added below
builder.add_edge(START, "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review")

This defines the execution order: research -> draft -> review. Now fill in the production features.

3. Durable State: Checkpointer

The checkpointer is what makes the graph "crash-resumable." Pass a checkpointer to compile, and LangGraph automatically saves a state snapshot (checkpoint) after every node executes:

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

InMemorySaver stores state in memory--suitable for development and testing. Production needs persistent storage: LangGraph supports SqliteSaver and PostgresSaver (install separately via langgraph-checkpoint-sqlite / langgraph-checkpoint-postgres), so state survives process restarts and can be shared across instances.

When invoking the graph, pass a thread_id--the unique identifier for a state thread. Multiple calls with the same thread_id share state:

import uuid

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
result = graph.invoke({"query": "Compare LangGraph and CrewAI", "messages": []}, config)

4. Per-Node Timeout: Preventing Hangs

LLM API calls hanging is the most common production failure. LangGraph 1.x adds a timeout parameter to add_node, letting you set a hard limit per node.

Simplest usage--pass a number for a wall-clock cap in seconds:

async def draft_node(state: AgentState) -> AgentState:
    # Call LLM to generate response
    ...

builder.add_node("draft", draft_node, timeout=30)  # 30-second hard limit

For finer control, use TimeoutPolicy to set both a wall-clock timeout and an idle timeout:

from langgraph.types import TimeoutPolicy

builder.add_node("research", research_node, timeout=TimeoutPolicy(
    run_timeout=60,    # wall-clock hard cap: node runs at most 60 seconds total
    idle_timeout=15,   # idle timeout: 15 seconds with no progress = timeout; progress signals reset the timer
    refresh_on="auto",  # automatically listens for callback events to refresh idle timer (default)
))

The difference between run_timeout and idle_timeout: run_timeout is a hard ceiling regardless of activity; idle_timeout resets whenever the node emits a progress signal (e.g., receiving a streaming chunk), making it ideal for long tasks that are "usually progressing but occasionally stall." When exceeded, LangGraph raises NodeTimeoutError, and the retry policy (if any) decides whether to retry.

Critical limitation: timeouts only work for async nodes. Synchronous nodes using time.sleep or CPU-bound work block the GIL and cannot be safely cancelled in-process. Use async def for all production nodes.

5. Retry Strategy: RetryPolicy

Transient API failures (rate limits, gateway 5xx) are routine--they should not crash the whole chain. The retry_policy parameter on add_node accepts a RetryPolicy:

from langgraph.types import RetryPolicy

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(
        max_attempts=3,           # max 3 attempts (including the first)
        initial_interval=1.0,     # wait 1 second before first retry
        backoff_factor=2.0,       # double interval each retry: 1s -> 2s -> 4s
        max_interval=128.0,      # cap retry interval at 128 seconds
        jitter=True,              # add random jitter to prevent retry storms
    ),
)

The retry_on parameter controls which exceptions trigger a retry. LangGraph's default policy default_retry_on behaves as follows:

• Retries: ConnectionError, HTTP 5xx responses (httpx.HTTPStatusError / requests.HTTPError with status 500-599)
• Does NOT retry: ValueError, TypeError, KeyError (LookupError), OSError, and other programming logic errors

This default is sensible--5xx is a transient server-side failure worth retrying; ValueError is a bug in your code that retrying will not fix. To customize, pass a callable:

from langgraph.errors import NodeTimeoutError

def retry_on_timeout_or_api_error(exc: Exception) -> bool:
    """Only retry on timeouts and API errors; let everything else crash."""
    if isinstance(exc, NodeTimeoutError):
        return True
    return False

builder.add_node("research", research_node,
    retry_policy=RetryPolicy(max_attempts=3, retry_on=retry_on_timeout_or_api_error),
)

6. Error Handler: Fallback with error_handler

What happens when retries are exhausted and the node still fails? The error_handler parameter lets you specify a fallback node. When the original node raises an exception, execution routes to the handler instead of crashing the graph:

async def research_fallback(state: AgentState) -> AgentState:
    """Fallback logic after research node exhausts retries."""
    return {
        "research": "External data source unavailable, using cached summary.",
        "messages": [{"role": "system", "content": "research degraded to cache mode"}],
    }

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(max_attempts=3),
    error_handler=research_fallback,  # 3 retries all failed -> route here
)

error_handler receives a node function (callable). Its return value is written back to state, and the graph continues executing. This is cleaner than wrapping the node function in try/except--error handling logic is separated from business logic, and the fallback path is visible in the graph structure.

7. Structured Output and Self-Correction: Conditional Edges

Production Agents need structured output--not free-form LLM text, but responses constrained to a schema. Typed state is that schema: each node returns a partial AgentState dictionary with declared field types.

To add a self-correction loop (rewrite when output quality is insufficient), use add_conditional_edges for dynamic routing:

def should_approve(state: AgentState) -> str:
    """Review node: pass -> end, fail -> back to draft for rewrite."""
    return END if state["approved"] else "draft"

builder.add_conditional_edges("review", should_approve)

The path function returns the next node name (or END), and add_conditional_edges routes accordingly. This creates a draft -> review -> (rejected) -> draft self-correction loop that only terminates when review passes.

8. Human-in-the-Loop: Interrupt Checkpoints

Some operations (sending emails, executing transactions, deleting data) require human confirmation. LangGraph's interrupt function pauses graph execution within a node, then resumes with Command(resume=...) after human input:

from langgraph.types import interrupt, Command

def review_node(state: AgentState) -> AgentState:
    # Pause graph execution, send draft to human reviewer
    answer = interrupt({
        "type": "approval_request",
        "draft": state["draft"],
        "message": "Review the response above. Reply yes to approve, no to rewrite.",
    })
    # interrupt() raises GraphInterrupt on first call, pausing execution
    # After human resumes with Command(resume=...), answer receives the input
    return {"approved": answer == "yes"}

Resuming execution:

# First invoke triggers interrupt, graph pauses
for chunk in graph.stream({"query": "...", "messages": []}, config):
    print(chunk)
# Output: {'__interrupt__': (...)}

# Human reviews and resumes
for chunk in graph.stream(Command(resume="yes"), config):
    print(chunk)
# approved=True -> graph continues to END

Critical prerequisite: interrupt requires a checkpointer. When the graph pauses, state must be persisted to survive; on resume, it is read back. A graph compiled without a checkpointer cannot use interrupt.

9. Complete Production Agent Code

Combining all six features into one runnable Agent:

import uuid
import operator
from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command, RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeTimeoutError

# ---- 1. Typed State ----
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    query: str
    research: str
    draft: str
    approved: bool

# ---- 2. Nodes (all async, required for timeout) ----
async def research_node(state: AgentState) -> AgentState:
    """Call search API to gather information."""
    # Replace with your actual search tool call
    result = f"Search results for '{state['query']}'..."
    return {"research": result, "messages": [{"role": "tool", "content": result}]}

async def draft_node(state: AgentState) -> AgentState:
    """Call LLM to generate a response draft."""
    # Replace with your actual LLM call
    draft = f"Draft response based on '{state['research']}'..."
    return {"draft": draft, "messages": [{"role": "assistant", "content": draft}]}

def review_node(state: AgentState) -> AgentState:
    """Human review breakpoint: pause for approval."""
    answer = interrupt({"draft": state["draft"], "msg": "Approve (yes) / Reject (no)"})
    return {"approved": answer == "yes"}

# ---- 3. Error Handler (fallback node) ----
async def research_fallback(state: AgentState) -> AgentState:
    return {"research": "Data source unavailable, using cache."}

# ---- 4. Custom retry predicate ----
def retry_on_timeout(exc: Exception) -> bool:
    return isinstance(exc, NodeTimeoutError)

# ---- 5. Build Graph ----
builder = StateGraph(AgentState)

builder.add_node("research", research_node,
    timeout=TimeoutPolicy(run_timeout=60, idle_timeout=15),
    retry_policy=RetryPolicy(max_attempts=3, retry_on=retry_on_timeout),
    error_handler=research_fallback,
)
builder.add_node("draft", draft_node, timeout=30)
builder.add_node("review", review_node)

builder.add_edge(START, "research")
builder.add_edge("research", "draft")
builder.add_edge("draft", "review")

def should_approve(state: AgentState) -> str:
    return END if state["approved"] else "draft"

builder.add_conditional_edges("review", should_approve)

# ---- 6. Compile (with checkpointer) ----
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# ---- 7. Run ----
config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# First invoke: runs until review node triggers interrupt, then pauses
for chunk in graph.stream(
    {"query": "LangGraph production best practices", "messages": []},
    config,
):
    print(chunk)

# Resume after human approval
for chunk in graph.stream(Command(resume="yes"), config):
    print(chunk)

Run this and you have a complete production Agent skeleton: durable state + timeout + retry + error handler + human-in-the-loop.

10. Parameter Reference Table

11. Pitfall Log

Pitfall 1: Timeouts only work for async nodes. Passing add_node(timeout=30) to a synchronous def node does not error but silently does nothing--sync functions block the GIL and cannot be safely cancelled in-process. Use async def for all production nodes, or timeouts are meaningless.

Pitfall 2: Using interrupt without a checkpointer. interrupt() raises GraphInterrupt and persists state for resumption. A graph compiled without a checkpointer will error when interrupt is called. Always pass checkpointer=InMemorySaver() (or a production saver) to compile.

Pitfall 3: Retrying exceptions that should not be retried. default_retry_on returns True for unrecognized exceptions. If your node raises a business logic error (e.g., "insufficient user balance"), it will be pointlessly retried 3 times, wasting time and tokens. Use a custom retry_on function to control precisely.

Pitfall 4: InMemorySaver loses state in production. InMemorySaver stores state in process memory--a process restart wipes everything. Production must use SqliteSaver or PostgresSaver (pip install langgraph-checkpoint-sqlite / langgraph-checkpoint-postgres) to persist state to disk or database.

Pitfall 5: Confusing idle_timeout with run_timeout. run_timeout=60 means 60 seconds total, period. idle_timeout=15 means 15 seconds of no progress, but progress signals reset the timer. For streaming LLM output, use idle_timeout--as long as chunks arrive, there is progress, and the timer resets, preventing false kills.

Pitfall 6: Calling failure-prone logic inside error_handler. The fallback node itself can fail (e.g., the cache service is also down). Keep error_handler simple--return static fallback values or local cached data. Do not make external API calls in the fallback path.

12. FAQ

Q1: How to choose between LangGraph, CrewAI, and AutoGen? According to an independent 2026 comparison (2000 runs), LangGraph had the best latency and token cost predictability and was rated "most production-ready." CrewAI is faster to prototype with; AutoGen (AG2) excels at multi-Agent dialogue. For durable state, fine-grained error control, and long-running production Agents, choose LangGraph. For rapid idea validation, CrewAI.

Q2: What is the difference between passing a number and a TimeoutPolicy to timeout? A number (timeout=30) is equivalent to TimeoutPolicy(run_timeout=30)--a wall-clock cap that is never refreshed. TimeoutPolicy(run_timeout=60, idle_timeout=15) sets both a hard ceiling and an idle timeout that resets on progress signals, ideal for streaming long tasks.

Q3: How do retry_policy and error_handler relate? retry_policy triggers first when a node raises an exception, retrying up to max_attempts times. Only after all retries are exhausted does error_handler activate. The handler returns a fallback result, and the graph continues without interruption. They form a layered "retry first, then fall back" strategy.

Q4: How do I resume after an interrupt? Use Command(resume="human input value") with the same thread_id to call graph.stream(Command(resume=...), config) again. The graph re-executes from the start of the review node, and the interrupt() function returns the value you passed in resume.

Q5: Which checkpointer should I use in production? Use InMemorySaver for development (in-memory, lost on restart), SqliteSaver for single-machine production (SQLite file), and PostgresSaver for multi-instance deployments (PostgreSQL, shared state across instances). The latter two require installing their respective checkpoint packages.

References

• LangGraph Docs - StateGraph and add_node
• LangGraph Docs - Human-in-the-loop
• LangSmith Blog - Fault Tolerance in LangGraph: Retries, Timeouts, and Error Handlers (2026-06-04, Quanzheng Long + Sydney Runkle)
• LangGraph Docs - Persistence (Checkpointer)
• LangGraph GitHub Repository - langchain-ai/langgraph