A multi-agent infinite loop will burn tokens and freeze your pipeline faster than any single bad prompt. In orchestration code, agents calling agents without termination guards turn a trivial task into a runaway process that saturates your rate limits. This guide gives concrete steps to instrument, bound, and break those cycles before they reach production.
Step 1: Define explicit termination conditions
Every agent loop needs a declared stop. Too many frameworks treat “the model says STOP” as sufficient. It isn’t. Models hallucinate continuation or emit malformed terminators.
Write a state machine, not a while True. Enumerate the terminal states up front.
from enum import Enum, auto
class AgentState(Enum):
IDLE = auto()
PLANNING = auto()
EXECUTING = auto()
AWAITING_TOOL = auto()
DONE = auto()
FAILED = auto()
def transition(current: AgentState, event: str) -> AgentState:
if current == AgentState.PLANNING and event == "plan_ready":
return AgentState.EXECUTING
if current == AgentState.EXECUTING and event == "task_complete":
return AgentState.DONE
if current == AgentState.EXECUTING and event == "error":
return AgentState.FAILED
# no transition back to PLANNING from EXECUTING without explicit reset
raise ValueError(f"Illegal transition {current} -> {event}")The key: forbid implicit re-entry to the planning state. If an agent finishes execution, it either terminates or fails. It does not silently spawn a new plan.
Step 2: Implement a global step counter and token budget
A multi-agent infinite loop often hides as slow growth: each iteration adds a small tool call. Cap total steps across the whole orchestration, not per agent.
class OrchestrationBudget:
def __init__(self, max_steps: int, max_tokens: int):
self.steps = 0
self.tokens = 0
self.max_steps = max_steps
self.max_tokens = max_tokens
def charge_step(self, n=1):
self.steps += n
if self.steps > self.max_steps:
raise RuntimeError("Step budget exceeded")
def charge_tokens(self, used: int):
self.tokens += used
if self.tokens > self.max_tokens:
raise RuntimeError("Token budget exceeded")Wire this into your agent runner. Every LLM call and every tool invocation calls charge_step. Every token returned from the API calls charge_tokens. When the budget throws, catch it at the top level and emit a structured failure.
budget = OrchestrationBudget(max_steps=50, max_tokens=200_000)
try:
for task in queue:
run_agent(task, budget)
except RuntimeError as e:
logger.error("orchestration halted", reason=str(e))
emit_alert("multi-agent loop bounded", str(e))Set max_steps relative to expected depth. For a three-agent pipeline with two round trips, 50 is generous. If you see loops, lower it.
Step 3: Add cycle detection on the agent call graph
Agents calling each other create directed edges. A cycle in that graph is a guaranteed multi-agent infinite loop unless broken externally.
Maintain a call stack per root request. Before dispatching to a child agent, check if that agent is already in the active path.
from typing import Set, Dict, List
class CallGraph:
def __init__(self):
self.edges: Dict[str, List[str]] = {}
self.active: Set[str] = set()
def enter(self, agent_id: str, parent: str | None):
if agent_id in self.active:
raise RuntimeError(f"Cycle detected: {agent_id} already active")
self.active.add(agent_id)
if parent:
self.edges.setdefault(parent, []).append(agent_id)
def exit(self, agent_id: str):
self.active.discard(agent_id)Use it as a context manager around each agent invocation:
from contextlib import contextmanager
@contextmanager
def agent_span(graph: CallGraph, agent_id: str, parent: str | None):
graph.enter(agent_id, parent)
try:
yield
finally:
graph.exit(agent_id)If agent A calls B, B calls C, and C calls A, the third enter raises. You get a stack trace, not a hang.
Step 4: Use hard timeouts and deadlock breakers
Even with counters, a single agent can block on a network call. Set a per-agent timeout and a wall-clock deadline for the whole tree.
import signal
class TimeoutError(Exception):
pass
def timeout_handler(signum, frame):
raise TimeoutError("agent exceeded wall clock")
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(30) # 30 seconds max for this agent
try:
result = blocking_llm_call()
finally:
signal.alarm(0)For distributed systems, use an external deadline propagated via headers. If a child misses the deadline, the parent cancels via asyncio.Task.cancel() or a message queue TTL.
A deadlock breaker: if no agent has made progress (zero state transitions) for N seconds, kill the root. Progress means a step charge or token charge. Track last_progress_ts.
import time
class ProgressWatcher:
def __init__(self, stale_sec: int):
self.last = time.monotonic()
self.stale_sec = stale_sec
def tick(self):
self.last = time.monotonic()
def check(self):
if time.monotonic() - self.last > self.stale_sec:
raise RuntimeError("No progress: possible deadlock")Step 5: Route through a resilient gateway to avoid retry storms
A frequent cause of looping is a provider returning 429 or 500, and the agent code catching the error and retrying with the same parameters forever. Instead of hand-rolling retries, send traffic through a gateway that fails fast or falls back.
n4n.ai provides automatic fallback when a provider is rate-limited or degraded, so your orchestration sees a single successful response or a clean error instead of a retry spiral. Honor its cache-control hints to skip redundant generations.
import openai
client = openai.OpenAI(
base_url="https://api.n4n.ai/v1",
api_key="YOUR_KEY"
)
# single call, gateway handles fallback
resp = client.chat.completions.create(
model="anthropic/claude-3.5-sonnet",
messages=[{"role": "user", "content": "summarize"}],
extra_headers={"x-n4n-cache": "read"} # forward cache hint
)If you must self-host retries, use exponential backoff with jitter and a max attempt count wired to the budget from Step 2.
import random, time
def backed_off_call(fn, max_attempts=3):
for i in range(max_attempts):
try:
return fn()
except TransientError:
time.sleep((2 ** i) + random.random())
raise RuntimeError("retries exhausted")Step 6: Emit observability signals and alert on loop patterns
You cannot fix a multi-agent infinite loop you cannot see. Log every state transition with a correlation id and the current step count.
{
"ts": "2025-04-12T10:22:01Z",
"root_id": "req_123",
"agent": "planner",
"state": "EXECUTING",
"step": 14,
"tokens": 4200
}Ship these to a log store and build a simple detector: if step count increments without a DONE state for more than K entries, page on-call.
from collections import defaultdict
step_counts = defaultdict(int)
def on_log(entry):
if entry["state"] not in ("DONE", "FAILED"):
step_counts[entry["root_id"]] = entry["step"]
if entry["step"] > 40:
alert(f"possible loop on {entry['root_id']}")A dashboard showing steps-per-root over time makes loops obvious: a flat line at the cap means your breaker works; a climbing line means you missed a path.
Verify success
Deploy the budget and cycle detector in a staging run with a deliberately broken agent that loops. Confirm the process raises RuntimeError within the step limit and the call graph logs a cycle. Then run a normal workload for 24 hours and check that no root_id exceeds 80% of max_steps. If alerts fire on real traffic, tune the constants—do not remove the guards.
That is the whole defense: explicit terminals, global budgets, cycle detection, timeouts, resilient routing, and telemetry. Implement them as a library and wrap every agent you ship.