Agent hallucination cascades start when one flawed output from a sub-agent gets consumed as fact by another, propagating errors faster than any human reviewer can catch. In multi-agent systems, a single misformatted JSON or confident but wrong summary can silently corrupt an entire pipeline. The fix is not more guardrails on the final answer—it’s instrumentation that flags the cascade at the second hop.
Step 1: Establish a correlated trace context across agent calls
You cannot detect a cascade if you cannot tell which agent output fed which subsequent prompt. Assign a single trace_id per user request and a span_id per agent invocation, with parent_span_id linking them. Use OpenTelemetry or a minimal homegrown context propagator.
import uuid
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class Span:
trace_id: str
span_id: str
parent_span_id: Optional[str]
agent_name: str
meta: dict = field(default_factory=dict)
def new_span(agent_name: str, parent: Optional[Span] = None) -> Span:
return Span(
trace_id=parent.trace_id if parent else uuid.uuid4().hex,
span_id=uuid.uuid4().hex,
parent_span_id=parent.span_id if parent else None,
agent_name=agent_name,
)Thread the Span through your agent executor. Every LLM call logs its span. This gives you the directed graph you need later.
Propagating span via contextvars
In async or threaded orchestrators, pass the span implicitly using contextvars so you don’t litter every function signature.
import contextvars
current_span = contextvars.ContextVar("span")
def agent_call(agent_name, fn, *args):
parent = current_span.get(None)
span = new_span(agent_name, parent)
token = current_span.set(span)
try:
return fn(*args, span)
finally:
current_span.reset(token)Now every agent can read current_span.get() to attach its output to the right node.
Step 2: Capture inputs and outputs as structured records
Tracing metadata alone is not enough. You need the actual prompt and completion text, plus model identity and token counts, to reconstruct what each agent believed to be true. Write each interaction to an append-only store.
{
"trace_id": "a1b2c3",
"span_id": "d4e5",
"parent_span_id": "f6g7",
"agent": "extractor",
"model": "gpt-4o-mini",
"prompt": "Extract entities from: ...",
"completion": "{\"entities\": [\"Acme Corp\"]}",
"usage": {"prompt_tokens": 120, "completion_tokens": 15},
"ts": "2024-05-01T12:00:00Z"
}A simple SQLite table or JSONL file works for most teams. The key is that the record is queryable by trace_id and parent_span_id so you can walk the chain. If you use a gateway that provides per-token usage metering, store that usage field directly; for example, n4n.ai returns per-token usage on each response, which drops cleanly into the usage field and helps attribute cost to specific cascade paths.
Step 3: Define lightweight consistency assertions between adjacent agents
Before spinning up expensive critic models, encode the cheap checks you already know. If agent A produces a list of IDs that agent B must use, assert subset membership. If A returns a JSON schema, validate B’s prompt inclusion. Numeric ranges, enum values, and regex patterns are all fair game.
def assert_entity_passthrough(parent_completion: str, child_prompt: str) -> list[str]:
import json
try:
ids = set(json.loads(parent_completion).get("entities", []))
except json.JSONDecodeError:
return ["parent_output_not_json"]
missing = [e for e in ids if e not in child_prompt]
return missing
# Example usage in orchestrator
missing = assert_entity_passthrough(parent_span.completion, child_span.prompt)
if missing:
log_failure(child_span, f"missing entities: {missing}")These assertions run synchronously and cost nothing but CPU. They catch the most common agent hallucination cascades: dropped context, mutated identifiers, or silently ignored instructions. Add a severity field so a missing critical ID weighs more than a missing optional field.
Step 4: Score factual grounding with a separate critic model
Some hallucinations are semantic, not structural. Use a critic agent that takes (parent_output, child_output) and returns a confidence score for factual alignment. Route this through a small, fast model to keep latency low.
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
def critic_score(parent: str, child: str) -> float:
resp = client.chat.completions.create(
model="mistralai/mixtral-8x7b-instruct",
messages=[
{"role": "system", "content": "Rate 0-1 how much child contradicts parent facts."},
{"role": "user", "content": f"PARENT:\n{parent}\nCHILD:\n{child}"}
],
temperature=0,
)
return float(resp.choices[0].message.content.strip())If you route through a single OpenAI-compatible gateway such as n4n.ai, which fronts 240+ models with automatic fallback, you can swap the critic model without client changes when a provider is degraded. The point is to treat the critic as a measurable edge in your trace graph, not a one-off sanity check.
Step 5: Detect cascade patterns with graph traversal
Once you have spans, assertions, and critic scores, build a per-trace DAG. Edges carry a risk weight: 0 for clean, >0 for assertion failures or critic score above threshold. A cascade is a path where cumulative risk exceeds a bound, or where three consecutive edges each show nonzero risk.
import networkx as nx
def build_trace_graph(records: list[dict]) -> nx.DiGraph:
g = nx.DiGraph()
for r in records:
g.add_node(r["span_id"], agent=r["agent"])
if r["parent_span_id"]:
risk = r.get("assertion_failures", 0) + (1 if r.get("critic_score", 0) > 0.5 else 0)
g.add_edge(r["parent_span_id"], r["span_id"], risk=risk)
return g
def find_cascades(g: nx.DiGraph, threshold: int = 2) -> list[list[str]]:
cascades = []
for path in nx.all_simple_paths(g, source=min(g.nodes)):
total = sum(g[u][v]["risk"] for u, v in zip(path, path[1:]))
if total >= threshold:
cascades.append(path)
return cascadesRun this after each trace completes, or streamingly as spans close. Tune threshold against historical traces: too low and you alert on noise, too high and real agent hallucination cascades slip through.
Step 6: Alert and short-circuit before compounding
Detection is useless if the pipeline keeps running. When find_cascades returns a non-empty list, raise a CascadeDetected exception in the orchestrator and skip remaining agents.
class CascadeDetected(Exception):
pass
def orchestrate(trace_records):
g = build_trace_graph(trace_records)
if find_cascades(g):
raise CascadeDetected("halt: agent hallucination cascades detected")
# otherwise continue to next agentWire this into your task queue. The partial trace is still saved for post-mortem. The user gets a structured error instead of a confidently wrong final answer. Push the cascade path to your alerting channel (Slack, PagerDuty) with the trace_id so on-call can inspect the exact graph.
Step 7: Verify the detection works with injected faults
A monitoring system you have not tested is a liability. Write a pytest that constructs a fake trace where agent B drops an entity from agent A, then confirm the assertion and cascade detector fire.
def test_cascade_detection():
recs = [
{"span_id": "s1", "parent_span_id": None, "agent": "extractor",
"completion": '{"entities": ["Acme"]}', "assertion_failures": 0},
{"span_id": "s2", "parent_span_id": "s1", "agent": "writer",
"prompt": "Write about nothing", "assertion_failures": 1, "critic_score": 0.8},
]
g = build_trace_graph(recs)
assert find_cascades(g, threshold=1) == [["s1", "s2"]]Run this in CI. Additionally, replay a sample of production traces with one agent’s output randomly mutated; you should see alert rate climb. That confirms the detector is live, not silently passing. For load verification, inject faults in 5% of canary requests and confirm p95 detection latency stays under 50ms (the graph step is cheap for traces with <100 spans).
Verification checklist
After deploying these steps, confirm success by:
- Issuing a request with a known bad sub-agent output (via unit test or staged canary) and observing the
CascadeDetectederror in logs. - Checking that the trace store contains the full DAG with
riskedges visualized. - Ensuring no downstream agent executed after the halt (instrument a counter on agent invocations per trace).
- Measuring that critic model calls add less than 200ms p95 latency to the trace when routed to a small model.
- Running the pytest suite in CI on every change to the assertion or graph logic.
Agent hallucination cascades are not rare edge cases in complex pipelines; they are the default failure mode when agents trust each other uncritically. Correlate, assert, criticize, graph, and halt. Do that and you turn a silent compounding error into a loud, debuggable event.