LangGraph debugging infinite loops starts with understanding that cycles are a feature, not a bug — until they never terminate. The framework’s graph-based execution model makes it trivial to express recursive workflows, but the same flexibility lets agents chase their own tails indefinitely. This guide walks through a repeatable process to catch runaway loops before they hit production, instrument your graphs for visibility, and add guardrails that preserve legitimate recursion while killing the pathological cases.
Step 1: Reproduce the loop in a minimal graph
Strip your workflow down to the smallest subgraph that still exhibits the problem. Remove external API calls, database writes, and non-deterministic components. Replace them with stubs that return fixed outputs or raise controlled exceptions. A minimal reproduction isolates the control-flow logic from environmental noise and gives you a test case that runs in seconds.
# minimal_loop.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
class State(TypedDict):
counter: int
route: Literal["continue", "exit"]
def increment(state: State) -> State:
return {"counter": state["counter"] + 1, "route": "continue"}
def should_continue(state: State) -> Literal["increment", "exit"]:
# Bug: condition never becomes false
return "increment" if state["counter"] < 10 else "exit"
builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_conditional_edges("increment", should_continue)
builder.set_entry_point("increment")
builder.add_edge("exit", END)
graph = builder.compile()
result = graph.invoke({"counter": 0, "route": "continue"})
print(result)Run this and confirm it hangs or hits the recursion limit. If it terminates correctly, your bug lives in a component you removed — add them back one at a time until the loop returns.
Verify success: The script either terminates with counter == 10 (fixed) or runs until Python raises RecursionError / LangGraph’s internal step limit (still broken).
Step 2: Enable step-level tracing
LangGraph’s debug flag and callback system surface every node transition. Wrap your graph invocation with a tracer that logs state deltas, edge decisions, and timestamps. This turns “it loops forever” into “node router chose edge retry 847 times with identical state.”
# tracer.py
import json
import time
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
from typing import Any, Dict
class StepTracer:
def __init__(self, max_steps: int = 100):
self.steps = []
self.max_steps = max_steps
self.start_time = time.time()
def on_step(self, step: int, node: str, state: Dict[str, Any], next_node: str | None):
elapsed = time.time() - self.start_time
entry = {
"step": step,
"node": node,
"next": next_node,
"elapsed_ms": round(elapsed * 1000, 2),
"state_keys": list(state.keys()),
"state_snapshot": {k: v for k, v in state.items() if k in ("counter", "route", "attempt")},
}
self.steps.append(entry)
print(json.dumps(entry))
if step >= self.max_steps:
raise RuntimeError(f"Step limit {self.max_steps} exceeded")
def run_with_trace(graph, initial_state: dict, max_steps: int = 50):
tracer = StepTracer(max_steps)
config = {"configurable": {"thread_id": "debug-session"}, "callbacks": [tracer]}
return graph.invoke(initial_state, config=config)Attach this to your minimal reproduction from Step 1. The JSON output lets you grep for repeated state signatures — the hallmark of a true infinite loop versus a long-but-finite computation.
Verify success: You can see the exact node sequence and state values at each step. Identical (node, state_hash) pairs appearing more than once confirm a cycle.
Step 3: Add a hard step budget per thread
LangGraph’s recursion_limit (default 25) catches runaway recursion in a single thread, but multi-agent graphs often spawn nested invocations that each get their own budget. Set a global step ceiling at the application layer and enforce it in a wrapper node that wraps the entire subgraph.
# budget.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.base import BaseCheckpointSaver
from typing import TypedDict, Callable, Any
import uuid
class BudgetState(TypedDict):
budget: int
payload: dict
def budget_wrapper(node_fn: Callable, max_steps: int = 100):
def wrapped(state: BudgetState) -> BudgetState:
if state["budget"] <= 0:
raise RuntimeError("Step budget exhausted")
state["budget"] -= 1
result = node_fn(state["payload"])
return {"budget": state["budget"], "payload": result}
return wrapped
def with_budget(graph, max_steps: int = 100):
"""Wrap a compiled graph with a step budget."""
class BudgetGraph:
def __init__(self, inner, budget):
self.inner = inner
self.budget = budget
def invoke(self, state: dict, config: dict | None = None):
budget_state = {"budget": self.budget, "payload": state}
# This is simplified; real implementation uses a parent graph
# that calls the inner graph node-by-node via .stream()
return self._run_with_budget(budget_state, config)
def _run_with_budget(self, budget_state, config):
# Use stream() to intercept each step
for chunk in self.inner.stream(budget_state["payload"], config=config, stream_mode="values"):
budget_state["budget"] -= 1
budget_state["payload"] = chunk
if budget_state["budget"] <= 0:
raise RuntimeError("Step budget exhausted")
return budget_state["payload"]
return BudgetGraph(graph, max_steps)Apply this wrapper to any subgraph that an agent can invoke repeatedly. The budget decrements on every streamed chunk, so even internal LangGraph steps count toward the limit.
Verify success: Invoke the wrapped graph with a known-looping input. It should raise RuntimeError("Step budget exhausted") within your configured limit, not hang indefinitely.
Step 4: Detect semantic cycles with state hashing
Step budgets are blunt instruments. They kill legitimate long-running workflows. For precision, compute a hash of the semantically relevant state fields at each node entry. If the same (node, hash) pair appears twice, you’ve entered a semantic cycle — the agent is revisiting a decision point with identical information.
# cycle_detector.py
import hashlib
import json
from typing import Any, Dict, Set
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver
class CycleDetector:
def __init__(self, state_keys: list[str], max_cycles: int = 1):
self.state_keys = state_keys
self.max_cycles = max_cycles
self.seen: Dict[tuple, int] = {}
def _hash_state(self, state: Dict[str, Any]) -> str:
relevant = {k: state[k] for k in self.state_keys if k in state}
serialized = json.dumps(relevant, sort_keys=True, default=str)
return hashlib.sha256(serialized.encode()).hexdigest()[:16]
def check(self, node: str, state: Dict[str, Any]) -> bool:
"""Returns True if cycle detected, False otherwise."""
h = self._hash_state(state)
key = (node, h)
count = self.seen.get(key, 0)
if count >= self.max_cycles:
return True
self.seen[key] = count + 1
return False
def make_cycle_aware_node(detector: CycleDetector, node_fn, node_name: str):
def wrapped(state):
if detector.check(node_name, state):
raise RuntimeError(f"Semantic cycle detected at {node_name}")
return node_fn(state)
return wrappedIntegrate this by wrapping each node in your graph. Choose state_keys carefully — include fields that drive routing decisions (route, intent, attempt_count) but exclude monotonically increasing fields like timestamps or log lengths.
# Integration example
detector = CycleDetector(state_keys=["route", "intent", "attempt_count"], max_cycles=1)
builder = StateGraph(State)
builder.add_node("router", make_cycle_aware_node(detector, router_fn, "router"))
builder.add_node("tool_a", make_cycle_aware_node(detector, tool_a_fn, "tool_a"))
# ...Verify success: The detector raises RuntimeError on the second visit to a node with identical decision-relevant state. Legitimate loops (e.g., retry with incrementing attempt_count) pass through because the hash changes.
Step 5: Add explicit loop-breaking edges
Sometimes the cleanest fix is structural. If an agent can legitimately retry but must eventually escalate or fail, model that as an explicit edge with a counter. LangGraph’s conditional edges make this straightforward.
# structured_retry.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
class AgentState(TypedDict):
task: str
attempt: int
max_attempts: int
result: str | None
error: str | None
def execute_task(state: AgentState) -> AgentState:
# Simulate flaky operation
if state["attempt"] < 2:
return {"attempt": state["attempt"] + 1, "error": "transient failure"}
return {"result": "success", "error": None}
def route_after_execute(state: AgentState) -> Literal["retry", "escalate", "done"]:
if state["error"] is None:
return "done"
if state["attempt"] >= state["max_attempts"]:
return "escalate"
return "retry"
def escalate(state: AgentState) -> AgentState:
return {"result": "escalated to human", "error": "max attempts reached"}
builder = StateGraph(AgentState)
builder.add_node("execute", execute_task)
builder.add_node("escalate", escalate)
builder.add_conditional_edges("execute", route_after_execute, {
"retry": "execute",
"escalate": "escalate",
"done": END,
})
builder.add_edge("escalate", END)
builder.set_entry_point("execute")
graph = builder.compile()
result = graph.invoke({"task": "flaky-api-call", "attempt": 0, "max_attempts": 3})
print(result)This pattern replaces implicit recursion with explicit, bounded iteration. The attempt counter is part of the state, so the cycle detector from Step 4 won’t false-positive on it.
Verify success: Run with max_attempts=3. The graph executes execute → retry → execute → retry → execute → done and terminates. Increase max_attempts and confirm it scales linearly, not infinitely.
Step 6: Write regression tests that fail on loops
Add a test that runs your graph with a loop-inducing input and asserts termination within a step budget. Use pytest-timeout as a backstop.
# test_no_infinite_loops.py
import pytest
from langgraph.graph import StateGraph
from budget import with_budget # from Step 3
@pytest.mark.timeout(5) # hard wall at 5 seconds
def test_router_does_not_loop_forever():
graph = build_production_graph() # your real graph factory
budgeted = with_budget(graph, max_steps=50)
# Input known to trigger the old loop
initial_state = {"query": "recursive trap", "context": []}
result = budgeted.invoke(initial_state)
# If we get here without exception, the loop is fixed
assert "final_answer" in result or "escalated" in result
def test_semantic_cycle_detector_catches_repeated_state():
from cycle_detector import CycleDetector
detector = CycleDetector(state_keys=["route"], max_cycles=1)
state = {"route": "retry", "data": "x"}
assert detector.check("router", state) is False # first visit
assert detector.check("router", state) is True # second visit = cycleRun these in CI. A loop regression now fails the build instead of burning compute in production.
Verify success: pytest test_no_infinite_loops.py passes. Introduce a deliberate loop in your graph code and confirm the test fails.
Step 7: Monitor production with structured metrics
Instrumentation doesn’t stop at debug logs. Emit metrics for every graph invocation: step count, unique nodes visited, cycle detector triggers, budget exhaustion events. Ship these to your observability stack (Prometheus, Datadog, CloudWatch) and alert on anomalies.
# metrics.py
from prometheus_client import Counter, Histogram
import time
GRAPH_INVOCATIONS = Counter("langgraph_invocations_total", "Total graph runs", ["graph_name", "outcome"])
GRAPH_STEPS = Histogram("langgraph_steps", "Steps per invocation", ["graph_name"])
CYCLE_DETECTED = Counter("langgraph_cycles_detected_total", "Semantic cycles caught", ["graph_name", "node"])
BUDGET_EXHAUSTED = Counter("langgraph_budget_exhausted_total", "Step budget exhausted", ["graph_name"])
def instrumented_invoke(graph, state: dict, graph_name: str, config: dict | None = None):
start = time.time()
steps = 0
try:
for _ in graph.stream(state, config=config, stream_mode="values"):
steps += 1
GRAPH_INVOCATIONS.labels(graph_name=graph_name, outcome="success").inc()
return _
except RuntimeError as e:
if "cycle" in str(e).lower():
CYCLE_DETECTED.labels(graph_name=graph_name, node="unknown").inc()
GRAPH_INVOCATIONS.labels(graph_name=graph_name, outcome="cycle").inc()
elif "budget" in str(e).lower():
BUDGET_EXHAUSTED.labels(graph_name=graph_name).inc()
GRAPH_INVOCATIONS.labels(graph_name=graph_name, outcome="budget").inc()
else:
GRAPH_INVOCATIONS.labels(graph_name=graph_name, outcome="error").inc()
raise
finally:
GRAPH_STEPS.labels(graph_name=graph_name).observe(steps)Dashboards showing steps_per_invocation trending upward or cycles_detected spiking catch regressions before users notice latency spikes.
Verify success: Deploy to staging, run load test. Metrics appear in your monitoring system. Inject a loop via feature flag and confirm alerts fire.
Step 8: Use checkpointing to inspect stuck threads
When a production thread hits the step budget or cycle detector, its checkpoint contains the full state at the moment of failure. LangGraph’s MemorySaver or PostgresSaver lets you retrieve and inspect that state offline.
# inspect_stuck.py
from langgraph.checkpoint.postgres import PostgresSaver
import json
DB_URI = "postgresql://user:pass@localhost/langgraph"
def dump_stuck_threads(limit: int = 10):
saver = PostgresSaver.from_conn_string(DB_URI)
checkpoints = saver.list(None, limit=limit) # None = all threads
for cp in checkpoints:
print(f"Thread: {cp.config['configurable']['thread_id']}")
print(f" Step: {cp.metadata.get('step', '?')}")
print(f" State: {json.dumps(cp.values, default=str, indent=2)}")
print("---")
if __name__ == "__main__":
dump_stuck_threads(20)Run this after an alert fires. The checkpoint reveals exactly which node was executing, what the state looked like, and why the router made the decision it did. This turns “something looped” into a concrete fix.
Verify success: After a budget-exhaustion event in staging, run the script. You can reconstruct the loop path from the checkpoint sequence.
Putting it together
The complete defense-in-depth stack:
- Minimal reproduction — isolates the control-flow bug
- Step tracing — makes the loop visible
- Step budget — guarantees termination
- Semantic cycle detection — distinguishes bugs from legitimate retries
- Structured retry edges — replaces implicit recursion with explicit bounds
- Regression tests — prevents reintroduction
- Production metrics — catches regressions that slip through
- Checkpoint inspection — accelerates root-cause analysis
Each layer catches what the previous one misses. The budget is your safety net; the cycle detector is your precision tool; the structured edges are your design fix; the tests and metrics are your ongoing guarantee.
If you’re running multiple models behind a gateway like n4n.ai, the same principles apply — wrap the gateway call in a node with its own budget and cycle detection, since provider retries can create their own loops.
Start with Steps 1 and 2 on your next stuck graph. You’ll have a fix and a test before the day ends.