Introduction to Error Recovery Patterns with LangGraph
Building reliable LLM applications means accepting a hard truth: things will fail. API calls time out, models return malformed JSON, tools throw exceptions, and rate limits bite when you least expect them. LangGraph, the graph-based orchestration framework from LangChain, gives you powerful primitives to handle these failures gracefully — but only if you know how to wire them up correctly.
This guide walks through the most important error recovery patterns available in LangGraph, from simple retries to sophisticated fallback graphs and human-in-the-loop recovery. Every pattern includes runnable code you can adapt to your own agents.
What Is LangGraph?
LangGraph is a library for building stateful, multi-actor applications with LLMs. It models your workflow as a directed graph where each node is a function or agent, and edges define control flow. Because the graph is explicit and state is centralized, you have fine-grained control over what happens when a node fails — something that is much harder in linear chains.
Why Error Recovery Matters
- Reliability: Production agents run for minutes or hours. A single unhandled error can waste an entire run.
- Cost: LLM calls are expensive. Losing intermediate work to a transient failure wastes tokens and money.
- User trust: An agent that silently dies or returns garbage erodes confidence faster than one that reports and recovers.
- Compliance: Many workflows require audit trails of what failed and how it was handled.
Pattern 1: Retry with Exponential Backoff
The simplest recovery pattern is retrying a failed operation. LangGraph nodes are just Python functions, so you can wrap risky operations with a retry decorator. This is ideal for transient failures like network blips or rate-limit errors.
import time
import random
from functools import wraps
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
def retry_with_backoff(max_attempts=3, base_delay=1.0):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempt = 0
while attempt < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempt += 1
if attempt >= max_attempts:
raise
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
print(f"Attempt {attempt} failed: {e}. Retrying in {delay:.1f}s")
time.sleep(delay)
return wrapper
return decorator
class State(TypedDict):
query: str
result: str
@retry_with_backoff(max_attempts=4, base_delay=0.5)
def call_model(state: State) -> dict:
# Simulate a flaky external API
if random.random() < 0.7:
raise ConnectionError("Transient API failure")
return {"result": f"Answer for: {state['query']}"}
builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
graph = builder.compile()
print(graph.invoke({"query": "What is LangGraph?"}))
The decorator keeps the node function clean while encapsulating the retry logic. For production use, prefer a battle-tested library like tenacity instead of a hand-rolled decorator.
Pattern 2: Conditional Routing on Error
Retries alone are not enough when the failure is deterministic — for example, a model returning invalid JSON. In those cases you want to route to a different node that can repair the output or try a different strategy. LangGraph's conditional edges make this natural.
The trick is to capture errors into the state instead of raising them, then branch based on whether an error field is populated.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
raw_output: str
parsed: Optional[dict]
error: Optional[str]
retries: int
def call_model(state: State) -> dict:
# Simulate a model that sometimes returns bad JSON
if state.get("retries", 0) == 0:
return {"raw_output": "{bad json missing closing brace"}
return {"raw_output": '{"answer": "42"}'}
def parse_output(state: State) -> dict:
import json
try:
parsed = json.loads(state["raw_output"])
return {"parsed": parsed, "error": None}
except json.JSONDecodeError as e:
return {"parsed": None, "error": str(e), "retries": state.get("retries", 0) + 1}
def repair(state: State) -> dict:
print(f"Repairing after error: {state['error']}")
# In a real app, you might ask the model to fix the JSON
return {"raw_output": '{"answer": "repaired"}'}
def should_retry_or_finish(state: State) -> str:
if state.get("parsed") is not None:
return "done"
if state.get("retries", 0) >= 3:
return "give_up"
return "repair"
builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_node("parse_output", parse_output)
builder.add_node("repair", repair)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "parse_output")
builder.add_conditional_edges(
"parse_output",
should_retry_or_finish,
{"done": END, "repair": "repair", "give_up": END},
)
builder.add_edge("repair", "call_model")
graph = builder.compile()
result = graph.invoke({"query": "meaning of life", "retries": 0})
print(result)
This pattern turns errors into data. The graph never crashes; it just routes to a recovery path. The retries counter prevents infinite loops.
Pattern 3: Fallback Models and Tools
Sometimes the right recovery is to try a different model entirely. A cheap, fast model might fail on a hard reasoning task; you then fall back to a more capable (and expensive) model. You can model this as two nodes with a conditional edge, or as a single node that tries models in sequence.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
prompt: str
answer: Optional[str]
model_used: Optional[str]
def try_primary(state: State) -> dict:
try:
# Simulate primary model call that fails on complex prompts
if len(state["prompt"]) > 20:
raise RuntimeError("Primary model: context too complex")
return {"answer": "primary answer", "model_used": "gpt-4o-mini"}
except Exception as e:
print(f"Primary failed: {e}")
return {"answer": None, "model_used": None}
def try_fallback(state: State) -> dict:
print("Falling back to stronger model")
return {"answer": "fallback answer", "model_used": "gpt-4o"}
def route(state: State) -> str:
return "done" if state.get("answer") else "fallback"
builder = StateGraph(State)
builder.add_node("primary", try_primary)
builder.add_node("fallback", try_fallback)
builder.add_edge(START, "primary")
builder.add_conditional_edges("primary", route, {"done": END, "fallback": "fallback"})
builder.add_edge("fallback", END)
graph = builder.compile()
print(graph.invoke({"prompt": "Explain quantum entanglement in detail"}))
The same pattern works for tools: if a primary search API fails, route to a backup provider. Keeping each tool behind its own node makes fallbacks composable.
Pattern 4: Checkpointing for Resumable Workflows
Long-running agents accumulate valuable intermediate state. If the process crashes, you do not want to start over. LangGraph supports checkpointing via a checkpointer, which persists state after every node execution. You can resume from the last checkpoint after a failure.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
tasks: list[str]
completed: list[str]
def step_one(state: State) -> dict:
completed = state.get("completed", []) + ["task_1"]
return {"completed": completed}
def step_two(state: State) -> dict:
# Imagine this node crashes halfway through on the first run
completed = state.get("completed", []) + ["task_2"]
return {"completed": completed}
def step_three(state: State) -> dict:
completed = state.get("completed", []) + ["task_3"]
return {"completed": completed}
builder = StateGraph(State)
builder.add_node("step_one", step_one)
builder.add_node("step_two", step_two)
builder.add_node("step_three", step_three)
builder.add_edge(START, "step_one")
builder.add_edge("step_one", "step_two")
builder.add_edge("step_two", "step_three")
builder.add_edge("step_three", END)
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "run-001"}}
# First invocation runs all steps
result = graph.invoke({"tasks": ["task_1", "task_2", "task_3"], "completed": []}, config)
print("First run:", result)
# Later, you can inspect or resume from the checkpoint
state = graph.get_state(config)
print("Checkpointed completed:", state.values["completed"])
For production, swap MemorySaver for SqliteSaver or PostgresSaver so state survives process restarts. The thread_id lets you isolate concurrent runs and resume any one of them independently.
Pattern 5: Human-in-the-Loop Recovery
Some errors need a human judgment call — for example, an agent that hits an ambiguous tool result or a policy violation. LangGraph supports pausing execution with an interrupt, asking a human for input, and resuming.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
class State(TypedDict):
draft: str
approved: bool
human_feedback: Optional[str]
def generate_draft(state: State) -> dict:
return {"draft": "Dear customer, we regret to inform you..."}
def human_review(state: State) -> dict:
# Execution pauses here. In a real app, the UI collects the response.
feedback = interrupt({"draft": state["draft"], "prompt": "Approve or edit?"})
return {"human_feedback": feedback, "approved": feedback.get("approved", False)}
def finalize(state: State) -> dict:
if state["approved"]:
return {"draft": state["draft"]}
# Incorporate feedback and regenerate
return {"draft": f"Revised based on feedback: {state['human_feedback']}"}
builder = StateGraph(State)
builder.add_node("generate_draft", generate_draft)
builder.add_node("human_review", human_review)
builder.add_node("finalize", finalize)
builder.add_edge(START, "generate_draft")
builder.add_edge("generate_draft", "human_review")
builder.add_edge("human_review", "finalize")
builder.add_edge("finalize", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "review-001"}}
# First invoke pauses at the interrupt
result = graph.invoke({"draft": "", "approved": False}, config)
print("Paused, waiting for human input")
# Simulate a human responding
result = graph.invoke(
Command(resume={"approved": False, "comment": "Make it warmer"}),
config,
)
print("Final draft:", result["draft"])
This pattern is invaluable for high-stakes workflows like email sending, payments, or code deployment. The graph handles the plumbing; you just provide a UI that reads the interrupted state and submits a resume command.
Pattern 6: Circuit Breakers for External Dependencies
When an external service is down, retrying endlessly makes things worse. A circuit breaker stops calling the service after a threshold of failures, giving it time to recover. You can implement this as a stateful node that tracks failure counts.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
query: str
result: str
failures: int
circuit_open: bool
CIRCUIT_THRESHOLD = 3
def call_external_api(state: State) -> dict:
# Simulated failing service
return {"result": "", "failures": state.get("failures", 0) + 1}
def circuit_check(state: State) -> dict:
failures = state.get("failures", 0)
if failures >= CIRCUIT_THRESHOLD:
print(f"Circuit open after {failures} failures")
return {"circuit_open": True, "result": "Service unavailable, using cached response"}
return {"circuit_open": False}
def route(state: State) -> str:
return "fallback" if state.get("circuit_open") else "call"
builder = StateGraph(State)
builder.add_node("circuit_check", circuit_check)
builder.add_node("call", call_external_api)
builder.add_node("fallback", lambda s: {"result": "cached_response"})
builder.add_edge(START, "circuit_check")
builder.add_conditional_edges("circuit_check", route, {"call": "call", "fallback": "fallback"})
builder.add_edge("call", END)
builder.add_edge("fallback", END)
graph = builder.compile()
for i in range(5):
result = graph.invoke({"query": "test", "failures": i, "circuit_open": False})
print(f"Run {i}: {result['result']}")
In a real system, you would also track a cooldown period and reset the failure count after a successful call. Storing circuit state in a shared store or the checkpointer keeps it consistent across invocations.
Best Practices
- Make errors part of your state. Instead of raising exceptions from nodes, catch them and store structured error objects. This makes branching and debugging far easier.
- Always bound your retries. Unbounded retries create infinite loops. Use a counter in state or a max-attempts decorator.
- Use checkpointers in production. MemorySaver is great for development, but you need persistent storage to survive restarts.
- Log at node boundaries. Because LangGraph calls nodes in a defined order, logging entry and exit of each node gives you a natural audit trail.
- Separate transient from deterministic failures. Retry network errors; do not retry invalid inputs. Route deterministic failures to repair or fallback nodes.
- Test failure paths explicitly. It is easy to verify the happy path. Inject failures in unit tests to confirm your recovery edges fire correctly.
- Keep recovery nodes small. A recovery node should do one thing: repair, fall back, or escalate. Mixing concerns makes graphs hard to reason about.
- Prefer tenacity for retry logic. It handles jitter, stop conditions, and retry-only-on-specific-exceptions better than a custom decorator.
Conclusion
Error recovery is not an afterthought in LangGraph — it is a first-class concern baked into the graph model. By treating errors as data, routing conditionally, checkpointing state, and pausing for human input, you can build agents that degrade gracefully instead of crashing spectacularly. Start with simple retries for transient failures, add conditional routing for deterministic ones, layer in checkpointing for long runs, and reserve human-in-the-loop interrupts for the decisions that truly need judgment. The result is an application that stays useful even when the underlying LLMs, APIs, and tools do not cooperate — which, in production, is the only scenario worth designing for.