LangGraph Error Recovery: Handling Node Failures Gracefully
Why your LangGraph agent crashes at 2am and nobody notices until morning
You shipped a LangGraph agent last week. It calls a tool, hits an API, writes to a database, and loops back for another round of reasoning. In testing, it worked every time. In production, it fell over on a Tuesday night because a downstream API returned a 503, and your graph had no idea what to do with that. The whole run died, the user got nothing, and the only trace was a stack trace buried in your logs.
This is the part of building agents that nobody puts in the demo video. Happy-path graphs are easy. The moment you add real tools — external APIs, databases, file systems, other LLMs — you inherit their failure modes too. A node in LangGraph is just a Python function wired into a graph, and Python functions raise exceptions. If you don't plan for that, your entire state machine grinds to a halt on the first flaky network call.
The good news is that LangGraph gives you real primitives for handling this: node-level try/except patterns, conditional edges that route based on error state, retry policies you can attach directly to nodes, checkpointing so you don't lose progress, and fallback nodes that degrade gracefully instead of exploding. This article walks through all of it, with working code, so you can build agents that survive contact with the real world instead of node failures taking down the whole run.
The default failure mode: one bad node kills the whole graph
Before fixing anything, it helps to see the actual failure. Here's a minimal graph with a node that calls an external service and sometimes fails.
from typing import TypedDict
from langgraph.graph import StateGraph, END
import random
class AgentState(TypedDict):
query: str
result: str
def call_flaky_api(state: AgentState) -> AgentState:
# Simulates a real API call that fails ~30% of the time
if random.random() < 0.3:
raise ConnectionError("upstream API timed out")
return {"query": state["query"], "result": "data fetched successfully"}
graph = StateGraph(AgentState)
graph.add_node("fetch", call_flaky_api)
graph.set_entry_point("fetch")
graph.add_edge("fetch", END)
app = graph.compile()
result = app.invoke({"query": "get user profile", "result": ""})Run this enough times and roughly one in three invocations throws an unhandled ConnectionError straight out of app.invoke(). There's no retry, no fallback, no partial result — just a crash. If this graph is behind a web endpoint, your users see a 500 error. If it's part of a longer multi-step agent, every bit of state accumulated before this node is thrown away.
This is the baseline every team starts from, and it's exactly why error recovery has to be designed in, not bolted on after an incident.
Layer 1: catch errors inside the node itself
The simplest and often most effective fix is to stop letting exceptions escape the node at all. Instead, catch them and encode the failure into the state itself, so downstream logic can react to it.
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
query: str
result: Optional[str]
error: Optional[str]
retry_count: int
def call_flaky_api(state: AgentState) -> AgentState:
try:
# Real call would go here
response = fetch_from_upstream(state["query"])
return {
"query": state["query"],
"result": response,
"error": None,
"retry_count": state.get("retry_count", 0),
}
except ConnectionError as e:
return {
"query": state["query"],
"result": None,
"error": f"connection_error: {e}",
"retry_count": state.get("retry_count", 0) + 1,
}
except ValueError as e:
return {
"query": state["query"],
"result": None,
"error": f"bad_input: {e}",
"retry_count": state.get("retry_count", 0),
}
def fetch_from_upstream(query: str) -> str:
# placeholder for real network call
return "data fetched successfully"Notice the two different exception types are handled differently. A ConnectionError is transient — worth retrying. A ValueError from bad input is not transient — retrying won't fix malformed data. This distinction matters enormously once you start building retry logic, because retrying a permanent failure just wastes time and API budget.
The key idea here is that the node's contract changes. Instead of "returns a result or throws," it becomes "always returns a state update, and the state tells you whether it succeeded." That single change makes everything downstream in the graph programmable instead of catastrophic.
Layer 2: conditional edges that route around failure
Once your node reports errors through state instead of exceptions, you can use LangGraph's conditional edges to decide what happens next: retry, fall back to a simpler path, or terminate gracefully with a useful message.
def route_after_fetch(state: AgentState) -> str:
if state["error"] is None:
return "process_result"
if "connection_error" in state["error"] and state["retry_count"] < 3:
return "fetch" # retry the same node
if "connection_error" in state["error"]:
return "fallback_cache" # give up retrying, use cached data
return "handle_permanent_error" # bad_input, don't retry
graph = StateGraph(AgentState)
graph.add_node("fetch", call_flaky_api)
graph.add_node("process_result", process_result)
graph.add_node("fallback_cache", use_cached_data)
graph.add_node("handle_permanent_error", handle_permanent_error)
graph.set_entry_point("fetch")
graph.add_conditional_edges(
"fetch",
route_after_fetch,
{
"fetch": "fetch",
"process_result": "process_result",
"fallback_cache": "fallback_cache",
"handle_permanent_error": "handle_permanent_error",
},
)
graph.add_edge("process_result", END)
graph.add_edge("fallback_cache", END)
graph.add_edge("handle_permanent_error", END)
app = graph.compile()This is where LangGraph's graph structure actually earns its keep over a plain linear chain. The routing function route_after_fetch reads the state and picks the next node by name, so you get retry loops, fallback branches, and terminal error handling all expressed as edges instead of nested try/except blocks. When someone new joins the team and looks at this graph, the failure handling is visible in the graph topology, not hidden inside a giant function.
One thing to watch: a self-loop like "fetch": "fetch" needs a hard upper bound on iterations, or a transient error that never clears will spin forever. The retry_count check inside route_after_fetch is what prevents that here — always cap retries at the routing level, not just inside the node.
Layer 3: built-in retry policies
Writing your own retry loop with conditional edges works, but LangGraph also ships a RetryPolicy you can attach directly to a node, which handles exponential backoff and exception filtering without you writing routing logic for the common case.
from langgraph.graph import StateGraph, END
from langgraph.pregel import RetryPolicy
def call_flaky_api(state: AgentState) -> AgentState:
response = fetch_from_upstream(state["query"])
return {"query": state["query"], "result": response, "error": None, "retry_count": 0}
graph = StateGraph(AgentState)
graph.add_node(
"fetch",
call_flaky_api,
retry=RetryPolicy(
max_attempts=4,
initial_interval=0.5,
backoff_factor=2.0,
retry_on=(ConnectionError, TimeoutError),
),
)
graph.set_entry_point("fetch")
graph.add_edge("fetch", END)
app = graph.compile()With retry_on scoped to ConnectionError and TimeoutError, a ValueError from bad input still propagates immediately instead of being retried four times for nothing. This is the cleanest option when a node's failure mode is genuinely "transient network blip" and you don't need custom branching logic — let the framework handle the backoff math instead of hand-rolling it.
You can combine both layers: use RetryPolicy for the "just try again with backoff" cases, and reserve conditional-edge routing for the cases where a retry isn't the right answer at all — like falling back to a cache, a smaller model, or a degraded response.
graph.add_node(
"call_llm",
call_llm_node,
retry=RetryPolicy(max_attempts=3, retry_on=(TimeoutError,)),
)
graph.add_conditional_edges(
"call_llm",
lambda state: "fallback_model" if state.get("error") else "continue",
{"fallback_model": "fallback_model", "continue": "continue"},
)This pattern shows up constantly in production agents: retry the flaky thing a few times automatically, and if it's still failing after that, switch strategy entirely rather than retrying forever.
Layer 4: checkpointing so failures don't erase progress
Retry policies and conditional routing solve the "what happens right after a failure" problem. Checkpointing solves a different one: what happens if the whole process crashes, or you need to resume a long-running graph from where it left off instead of from scratch.
LangGraph's checkpointer persists the state of the graph after every step, so a crashed run can be resumed rather than restarted.
from langgraph.checkpoint.memory import MemorySaver
# For production, use a persistent backend instead of MemorySaver, e.g.:
# from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
try:
result = app.invoke({"query": "generate report", "result": None, "error": None, "retry_count": 0}, config=config)
except Exception as exc:
print(f"Run interrupted: {exc}")
# State up to the last successful node is already saved under thread_id
# A later call with the same config resumes rather than restartingThe thread_id is the key here. If your agent has already completed five nodes and crashes on the sixth, resuming with the same thread_id picks the graph back up from the last checkpoint instead of re-running the five nodes that already succeeded — including any side effects you'd rather not repeat, like sending an email or charging a card. For anything long-running (multi-step research agents, document processing pipelines, agents that call paid APIs), skipping checkpointing means every crash is a full restart at full cost.
In production you'll want a durable checkpointer backed by Postgres, SQLite, or Redis rather than MemorySaver, since in-memory state disappears the moment the process restarts — which is precisely when you need it most.
Layer 5: fallback nodes and graceful degradation
Sometimes the right response to a failure isn't a retry at all — it's doing something less ideal but still useful. This is the difference between an agent that says "error: connection refused" and one that says "I couldn't reach the live pricing API, so here's the price from this morning's cache, plus a note that it may be stale."
class OrderState(TypedDict):
order_id: str
price: Optional[float]
price_source: Optional[str]
error: Optional[str]
def get_live_price(state: OrderState) -> OrderState:
try:
price = call_pricing_service(state["order_id"])
return {**state, "price": price, "price_source": "live", "error": None}
except Exception as e:
return {**state, "price": None, "price_source": None, "error": str(e)}
def get_cached_price(state: OrderState) -> OrderState:
price = lookup_cached_price(state["order_id"])
return {**state, "price": price, "price_source": "cache_fallback", "error": None}
def route_pricing(state: OrderState) -> str:
return "cached_price" if state["error"] else "finalize"
graph = StateGraph(OrderState)
graph.add_node("live_price", get_live_price)
graph.add_node("cached_price", get_cached_price)
graph.add_node("finalize", finalize_order)
graph.set_entry_point("live_price")
graph.add_conditional_edges(
"live_price", route_pricing, {"cached_price": "cached_price", "finalize": "finalize"}
)
graph.add_edge("cached_price", "finalize")
graph.add_edge("finalize", END)
app = graph.compile()The price_source field is doing important work here. It's not just a fallback for the sake of avoiding a crash — it's an honest fallback, because the state records which path was taken. Whatever consumes this graph's output (a UI, a downstream node, a log line) can distinguish "live price" from "cached price" and behave accordingly, like showing a small disclaimer to the user. Silent degradation that looks identical to the happy path is worse than an explicit failure, because it erodes trust the moment someone notices the numbers were stale and nobody told them.
Layer 6: a global error boundary around the whole graph
Node-level handling covers most cases, but you also want a top-level safety net for anything that slips through — a bug in your routing function, an unexpected exception type, a checkpointer failure. Wrap the invocation itself.
import logging
logger = logging.getLogger("agent_runtime")
def run_agent_safely(app, initial_state: dict, config: dict) -> dict:
try:
return app.invoke(initial_state, config=config)
except Exception as exc:
logger.exception("Unhandled graph failure for thread %s", config.get("configurable", {}).get("thread_id"))
return {
**initial_state,
"error": f"unrecoverable_error: {exc}",
"result": None,
}
result = run_agent_safely(
app,
{"query": "generate report", "result": None, "error": None, "retry_count": 0},
{"configurable": {"thread_id": "user-session-42"}},
)
if result.get("error"):
# Surface a clean message to the user instead of a stack trace
print("Something went wrong processing your request. Our team has been notified.")
else:
print(result["result"])This boundary is what stands between "an unhandled Python exception reaches your API layer and returns a 500" and "the user sees a clean, predictable message while your logs get a full stack trace for debugging." It costs almost nothing to add and it should wrap every graph invocation you ship to production, no exceptions — the node-level handling above reduces how often you hit this boundary, but it doesn't make it unnecessary.
Streaming runs and partial failure
If you're streaming intermediate results with app.stream() instead of a single invoke(), error handling needs a small adjustment, because a failure can happen mid-stream after some nodes have already yielded output the user has seen.
def run_streaming_with_recovery(app, initial_state: dict, config: dict):
partial_results = []
try:
for chunk in app.stream(initial_state, config=config):
partial_results.append(chunk)
yield chunk
except Exception as exc:
logger.exception("Stream interrupted after %d steps", len(partial_results))
yield {
"error": {
"message": "The process was interrupted before completing.",
"completed_steps": len(partial_results),
}
}Because the graph is checkpointed (see Layer 4), the steps that already streamed successfully are also saved, so a resumed run with the same thread_id won't redo work the user already saw complete. Streaming makes failures more visible to users in real time, which is usually a good thing — silent, buffered failures are much harder to debug and much more jarring when they surface all at once at the end.
Testing your error recovery on purpose
None of this is worth much if you never actually exercise the failure paths. Write tests that deliberately inject failures into nodes and assert the graph routes correctly, rather than only testing the happy path.
import pytest
from unittest.mock import patch
def test_graph_falls_back_on_connection_error():
with patch("your_module.fetch_from_upstream", side_effect=ConnectionError("boom")):
result = app.invoke(
{"query": "test", "result": None, "error": None, "retry_count": 0},
config={"configurable": {"thread_id": "test-1"}},
)
assert result["price_source"] == "cache_fallback"
assert result["error"] is None # fallback cleared the error
def test_graph_does_not_retry_permanent_errors():
with patch("your_module.fetch_from_upstream", side_effect=ValueError("bad query")):
result = app.invoke(
{"query": "", "result": None, "error": None, "retry_count": 0},
config={"configurable": {"thread_id": "test-2"}},
)
assert "bad_input" in result["error"]
assert result["retry_count"] == 0 # confirms no retry loop was triggeredTreat these tests as first-class, not an afterthought bolted on after an incident. The whole point of building explicit error-recovery paths is that they're testable — unlike a bare, unhandled exception, which you can only really "test" by waiting for it to happen in production.
Putting the layers together
None of these six layers replace each other — they stack. A well-built LangGraph agent typically looks like this: nodes catch their own exceptions and encode failure into state; a RetryPolicy handles transient errors with backoff; conditional edges route around anything that isn't transient; a checkpointer means crashes don't erase completed work; fallback nodes degrade gracefully and label themselves honestly when they do; and a top-level boundary catches anything that still slips through, so users never see a raw stack trace.
Start small if you're retrofitting an existing graph — pick your highest-traffic node, wrap it in a try/except, add a RetryPolicy for its transient failure modes, and add one conditional edge for its permanent ones. Then move to the next node. You don't need all six layers on day one, but you do need to know which layer is missing before your agent hits production traffic, not after the first incident report.
If you want to go deeper on this — building multi-agent LangGraph systems with proper state management, checkpointing across long-running workflows, and production-grade error handling patterns beyond what fits in one article — that's exactly what we cover hands-on in the LangGraph Tutorial course here on teachyou.ai, with real graphs you build and break on purpose so you know how they behave before your users find out for you.
BootcampA 30-day guided bootcamp: build, harden and ship a production autonomous agent from scratch.
AI AgentsUnderstand how AI agents really work: the loop, the tools, the memory, and why most agent projects fail.