When multi-agent systems fail, traditional debugging breaks down -learn systematic approaches to trace errors through agent handoffs, tool calls, and state management.
TL;DR
Jump to Why multi-agent debugging is different · Jump to The five layers · Jump to Tracing strategies · Jump to Common failure patterns · Jump to Debugging tools
At 2 AM last Tuesday, our partnership qualification agent silently stopped booking meetings. No errors logged. No alerts fired. Just... silence. Five hours of lost opportunities before anyone noticed.
The culprit? Agent B enriched lead data but returned an empty industry field. Agent C, expecting that field, scored every lead as 0. Agent D, seeing low scores, discarded everyone. None of the agents technically "failed" -they just produced garbage.
Welcome to multi-agent debugging, where traditional approaches fall apart and you need entirely new mental models to identify what went wrong.
"Debugging distributed systems taught me humility. Debugging multi-agent systems taught me that humility was wildly insufficient." – Sam Lambert, former GitHub VP Engineering (conference talk, 2024)
Traditional debugging assumes linear execution: function A calls function B calls function C. Stack traces show exactly where things broke. Multi-agent systems shatter this assumption.
Consider a simple three-agent workflow:
Agent A (research) → Agent B (analysis) → Agent C (action)
Agent C fails. Was it because:
Traditional stack traces stop at "Agent C threw an exception." They don't show you the chain of decisions that led there.
LLMs introduce randomness. The same inputs can produce different outputs. This makes reproduction difficult:
| Attempt | Agent A output | Agent B output | Agent C output | Result |
|---|---|---|---|---|
| 1 | "SaaS company" | Score: 85 | Book meeting | ✓ Success |
| 2 | "Software company" | Score: 72 | Add to nurture | ⚠ Different |
| 3 | "Technology firm" | Score: 68 | Add to nurture | ⚠ Different |
| 4 | "Tech business" | Score: 65 | Discard | ✗ Failure |
Same input lead, four different outcomes. How do you debug that?
Multi-agent systems maintain state across agents:
State corruption anywhere breaks everything downstream. But pinpointing where corruption occurred requires meticulous tracing.
According to a 2024 survey of 200+ engineering teams building multi-agent systems, 78% reported that debugging was their #1 operational challenge, ahead of cost optimization (62%) and performance tuning (54%) (AI Engineering Survey, 2024).
Effective multi-agent debugging requires thinking in layers. Problems occur at different abstraction levels, and you need tools for each.
What it is: The high-level routing between agents -which agent ran when, what triggered handoffs, what the intended workflow was.
Common failures:
How to debug:
import logging
from datetime import datetime
class OrchestrationTracer:
"""Track agent invocations and handoffs."""
def __init__(self, workflow_id: str):
self.workflow_id = workflow_id
self.events = []
def log_invocation(self, agent_name: str, input_data: dict):
"""Record agent start."""
event = {
"workflow_id": self.workflow_id,
"timestamp": datetime.utcnow().isoformat(),
"event_type": "agent_invoked",
"agent": agent_name,
"input_preview": str(input_data)[:200]
}
self.events.append(event)
logging.info(f"[Orchestration] Invoking {agent_name}", extra=event)
def log_completion(self, agent_name: str, output_data: dict, duration_ms: int):
"""Record agent completion."""
event = {
"workflow_id": self.workflow_id,
"timestamp": datetime.utcnow().isoformat(),
"event_type": "agent_completed",
"agent": agent_name,
"duration_ms": duration_ms,
"output_preview": str(output_data)[:200]
}
self.events.append(event)
logging.info(f"[Orchestration] {agent_name} completed in {duration_ms}ms", extra=event)
def log_handoff(self, from_agent: str, to_agent: str, reason: str):
"""Record agent handoff."""
event = {
"workflow_id": self.workflow_id,
"timestamp": datetime.utcnow().isoformat(),
"event_type": "handoff",
"from_agent": from_agent,
"to_agent": to_agent,
"reason": reason
}
self.events.append(event)
logging.info(f"[Orchestration] Handoff: {from_agent} → {to_agent}", extra=event)
def get_timeline(self):
"""Generate visual timeline of workflow."""
return "\n".join([
f"{e['timestamp']} | {e['event_type']:20} | {e.get('agent', 'N/A'):20}"
for e in self.events
])
Example timeline from a failed workflow:
2024-12-28T14:23:01Z | agent_invoked | research_agent
2024-12-28T14:23:03Z | agent_completed | research_agent
2024-12-28T14:23:03Z | handoff | research → analysis
2024-12-28T14:23:04Z | agent_invoked | analysis_agent
2024-12-28T14:23:09Z | agent_completed | analysis_agent
2024-12-28T14:23:09Z | handoff | analysis → action
[NO FURTHER EVENTS]
Diagnosis: Action agent was never invoked. Orchestrator bug or handoff failure.
What it is: The internal state of each agent -what it knows, what context it has, what decisions it's considering.
Common failures:
How to debug:
Instrument agents to log their decision-making process:
class DebuggableAgent:
"""Agent with built-in state inspection."""
def __init__(self, name: str):
self.name = name
self.state = {}
def run(self, input_data: dict):
# Capture input state
self.log_state("input_received", input_data)
# Process
try:
# Extract fields agent needs
required_fields = ["email", "company", "industry"]
for field in required_fields:
if field not in input_data:
self.log_state("missing_field", {"field": field})
raise ValueError(f"Missing required field: {field}")
# Make decision
decision = self.make_decision(input_data)
self.log_state("decision_made", {"decision": decision, "reasoning": self.reasoning})
return decision
except Exception as e:
self.log_state("error", {"exception": str(e)})
raise
def log_state(self, event: str, data: dict):
"""Log internal state changes."""
state_snapshot = {
"agent": self.name,
"event": event,
"timestamp": datetime.utcnow().isoformat(),
"data": data,
"full_state": self.state.copy()
}
logging.debug(f"[{self.name}] {event}", extra=state_snapshot)
When debugging, grep logs for a specific agent:
grep "analysis_agent" logs/agent.log | jq
You'll see exactly what that agent received, what it thought, and what it output.
What it is: External API calls, database queries, file operations that agents perform.
Common failures:
How to debug:
Wrap every tool call with instrumentation:
from functools import wraps
import time
def traced_tool(func):
"""Decorator to trace tool executions."""
@wraps(func)
def wrapper(*args, **kwargs):
tool_name = func.__name__
start_time = time.time()
# Log invocation
logging.info(f"[Tool] Calling {tool_name}", extra={
"tool": tool_name,
"args": str(args)[:100],
"kwargs": str(kwargs)[:100]
})
try:
result = func(*args, **kwargs)
duration_ms = (time.time() - start_time) * 1000
# Log success
logging.info(f"[Tool] {tool_name} succeeded in {duration_ms:.0f}ms", extra={
"tool": tool_name,
"duration_ms": duration_ms,
"result_preview": str(result)[:200]
})
return result
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
# Log failure
logging.error(f"[Tool] {tool_name} failed after {duration_ms:.0f}ms", extra={
"tool": tool_name,
"duration_ms": duration_ms,
"error": str(e),
"error_type": type(e).__name__
})
raise
return wrapper
# Use on every tool
@traced_tool
def enrich_lead(email: str):
"""Fetch lead data from Clearbit."""
response = requests.get(
f"https://person-stream.clearbit.com/v2/combined/find?email={email}",
auth=(CLEARBIT_API_KEY, ''),
timeout=10 # Explicit timeout
)
response.raise_for_status()
return response.json()
@traced_tool
def update_crm(contact_id: str, data: dict):
"""Update HubSpot contact."""
response = requests.patch(
f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
headers={"Authorization": f"Bearer {HUBSPOT_API_KEY}"},
json={"properties": data}
)
response.raise_for_status()
return response.json()
Now when a tool fails, you see:
{
"timestamp": "2024-12-28T14:23:07Z",
"level": "ERROR",
"message": "[Tool] enrich_lead failed after 10042ms",
"tool": "enrich_lead",
"duration_ms": 10042,
"error": "ReadTimeout: HTTPConnectionPool(host='person-stream.clearbit.com', port=443): Read timed out.",
"error_type": "ReadTimeout"
}
Diagnosis: Clearbit API is slow. Add retry logic or increase timeout.
What it is: The data passed between agents -format, completeness, correctness.
Common failures:
{"score": 85}, Agent B expects {"lead_score": 85})industry)"85", Agent B expects int 85)How to debug:
Use schema validation at every handoff:
from pydantic import BaseModel, ValidationError
from typing import Optional
class ResearchOutput(BaseModel):
"""Schema for research agent output."""
company: str
industry: str
employee_count: Optional[int]
funding_stage: Optional[str]
website: str
confidence: float
class AnalysisInput(BaseModel):
"""Schema for analysis agent input."""
company: str
industry: str # Required!
employee_count: Optional[int]
funding_stage: Optional[str]
def handoff_with_validation(from_agent: str, to_agent: str, data: dict, schema: BaseModel):
"""Validate data during handoff."""
try:
validated_data = schema(**data)
logging.info(f"[Handoff] {from_agent} → {to_agent}: validation passed")
return validated_data.dict()
except ValidationError as e:
logging.error(f"[Handoff] {from_agent} → {to_agent}: validation failed", extra={
"from_agent": from_agent,
"to_agent": to_agent,
"validation_errors": e.errors(),
"invalid_data": data
})
raise ValueError(f"Invalid handoff data: {e}")
# Use in orchestrator
research_output = research_agent.run(input_data)
validated_output = handoff_with_validation(
"research_agent",
"analysis_agent",
research_output,
AnalysisInput
)
analysis_result = analysis_agent.run(validated_output)
When a handoff fails, you get explicit errors:
{
"from_agent": "research_agent",
"to_agent": "analysis_agent",
"validation_errors": [
{
"loc": ["industry"],
"msg": "field required",
"type": "value_error.missing"
}
],
"invalid_data": {
"company": "Acme Corp",
"employee_count": 150,
"website": "acme.com",
"confidence": 0.92
}
}
Diagnosis: Research agent didn't return industry field. Bug in research agent or upstream data source.
What it is: The reasoning process of LLM-based agents -what prompts they received, what they generated, why they made specific decisions.
Common failures:
How to debug:
Log full prompts and completions:
def call_llm_with_logging(prompt: str, model: str = "gpt-4-turbo-preview"):
"""Call LLM with full request/response logging."""
logging.info("[LLM] Sending request", extra={
"model": model,
"prompt_length": len(prompt),
"prompt_preview": prompt[:500]
})
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}]
)
logging.info("[LLM] Received response", extra={
"model": model,
"completion_length": len(response.choices[0].message.content),
"completion": response.choices[0].message.content,
"tokens_used": response.usage.total_tokens
})
return response.choices[0].message.content
For debugging, review the LLM's actual reasoning:
[LLM Request]
Model: gpt-4-turbo-preview
Prompt: "Classify this lead's industry: Company name: Acme Corp, Description: We build software"
[LLM Response]
Completion: "Based on the description 'We build software', this company operates in the Technology industry, specifically Software Development."
If classification was wrong, you can see why -maybe the description was too vague, or the prompt didn't provide enough context.
The gold standard for multi-agent debugging is distributed tracing -propagating a unique trace ID through every agent, tool call, and LLM request in a workflow.
import uuid
from contextvars import ContextVar
# Thread-safe storage of current trace ID
current_trace_id = ContextVar("trace_id", default=None)
class TraceContext:
"""Manage trace ID across async operations."""
def __init__(self, trace_id: str = None):
self.trace_id = trace_id or str(uuid.uuid4())
def __enter__(self):
self.token = current_trace_id.set(self.trace_id)
return self
def __exit__(self, *args):
current_trace_id.reset(self.token)
def get_trace_id() -> str:
"""Get current trace ID."""
trace_id = current_trace_id.get()
if not trace_id:
# If no trace context, create one
trace_id = str(uuid.uuid4())
current_trace_id.set(trace_id)
return trace_id
# Update all logging to include trace ID
def traced_log(level: str, message: str, **kwargs):
"""Log with automatic trace ID inclusion."""
trace_id = get_trace_id()
extra = {"trace_id": trace_id, **kwargs}
if level == "info":
logging.info(message, extra=extra)
elif level == "error":
logging.error(message, extra=extra)
# ... etc
# Use in orchestrator
def process_workflow(input_data: dict):
with TraceContext() as trace:
traced_log("info", "Starting workflow", input_preview=str(input_data)[:100])
# All agents and tools inherit this trace ID
result_a = agent_a.run(input_data)
result_b = agent_b.run(result_a)
result_c = agent_c.run(result_b)
traced_log("info", "Workflow completed")
return result_c
Now every log entry has the same trace_id, allowing you to reconstruct the entire workflow:
grep "trace_id: abc-123" logs/agent.log | jq
Returns chronological sequence of all events for that workflow.
Convert traces to visual timelines using tools like Jaeger or Honeycomb:
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
# Setup OpenTelemetry
tracer_provider = TracerProvider()
jaeger_exporter = JaegerExporter(
agent_host_name="localhost",
agent_port=6831,
)
tracer_provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)
# Instrument workflow
def traced_workflow(input_data: dict):
with tracer.start_as_current_span("workflow") as workflow_span:
workflow_span.set_attribute("input_size", len(str(input_data)))
with tracer.start_as_current_span("agent_a"):
result_a = agent_a.run(input_data)
with tracer.start_as_current_span("agent_b"):
result_b = agent_b.run(result_a)
with tracer.start_as_current_span("agent_c"):
result_c = agent_c.run(result_b)
return result_c
Jaeger UI shows:
Workflow (total: 3.2s)
├─ agent_a (1.1s)
│ ├─ tool: enrich_lead (0.9s)
│ └─ llm: gpt-4 (0.2s)
├─ agent_b (1.8s)
│ ├─ llm: gpt-4 (1.6s)
│ └─ tool: calculate_score (0.2s)
└─ agent_c (0.3s)
└─ tool: book_meeting (0.3s)
Instantly see where time was spent and where failures occurred.
After debugging hundreds of multi-agent workflows, patterns emerge. Here are the most frequent.
Symptom: Agent completes successfully but produces wrong output. No errors raised.
Example: Enrichment agent returns {"industry": null}. Downstream agents treat this as valid data, leading to incorrect scoring.
Fix: Add assertions and invariant checks:
def enrich_lead(email: str):
data = call_enrichment_api(email)
# Assert critical fields are present
assert data.get("industry") is not None, f"Enrichment returned null industry for {email}"
assert data.get("company") is not None, f"Enrichment returned null company for {email}"
return data
Symptom: One slow tool call causes multiple downstream timeouts.
Example: Clearbit API takes 12 seconds. Research agent times out. Orchestrator retries. Retry also times out. Entire workflow fails.
Fix: Implement circuit breakers:
from datetime import datetime, timedelta
class CircuitBreaker:
"""Prevent cascading failures."""
def __init__(self, failure_threshold: int = 3, timeout_duration: int = 60):
self.failure_count = 0
self.failure_threshold = failure_threshold
self.timeout_duration = timeout_duration
self.opened_at = None
def call(self, func, *args, **kwargs):
# Check if circuit is open
if self.opened_at:
if datetime.now() - self.opened_at < timedelta(seconds=self.timeout_duration):
raise Exception(f"Circuit breaker open for {func.__name__}")
else:
# Try to close circuit
self.opened_at = None
self.failure_count = 0
try:
result = func(*args, **kwargs)
self.failure_count = 0 # Reset on success
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.opened_at = datetime.now()
raise
# Use with tools
clearbit_breaker = CircuitBreaker(failure_threshold=3, timeout_duration=60)
def safe_enrich_lead(email: str):
return clearbit_breaker.call(enrich_lead, email)
Symptom: Agent A runs correctly, but Agent B's previous state leaks into current execution.
Example: Analysis agent caches company data. New lead from different company arrives, but agent uses cached data from previous lead.
Fix: Enforce stateless agents:
class StatelessAgent:
"""Agent that doesn't maintain state between runs."""
def run(self, input_data: dict):
# Create fresh context for this run
context = self._create_context(input_data)
try:
result = self._process(context)
return result
finally:
# Explicitly clear context
del context
gc.collect()
def _create_context(self, input_data: dict):
"""Create isolated context for this run."""
return {
"input": input_data,
"timestamp": datetime.utcnow(),
"run_id": str(uuid.uuid4())
}
Symptom: Orchestrator starts Agent B before Agent A finishes, causing data corruption.
Example: Parallel execution starts Analysis agent whilst Research agent is still writing results to shared database.
Fix: Use explicit synchronization:
import asyncio
async def safe_parallel_execution(agents: list, input_data: dict):
"""Run agents in parallel with proper synchronization."""
# Start all agents
tasks = [agent.run_async(input_data) for agent in agents]
# Wait for ALL to complete
results = await asyncio.gather(*tasks, return_exceptions=True)
# Check for failures
failures = [r for r in results if isinstance(r, Exception)]
if failures:
raise Exception(f"Agent failures: {failures}")
return results
Centralized logging lets you query across all agents:
-- Find all workflows where agent_c failed
SELECT trace_id, timestamp, error_message
FROM logs
WHERE agent = 'agent_c' AND level = 'ERROR'
ORDER BY timestamp DESC
LIMIT 100
Visual trace timelines show exact failure points and latency breakdowns.
Purpose-built tools for agent debugging:
When a multi-agent workflow fails:
1. Identify the trace ID (if available) or timestamp range
2. Reconstruct the flow
3. Inspect each agent
4. Check tool executions
5. Validate data handoffs
6. Review LLM decisions
7. Check system resources
Week 1: Add instrumentation
Week 2: Centralize logging
Week 3: Build debugging workflows
Multi-agent debugging isn't about finding bugs faster -it's about making bugs findable at all. Traditional debugging assumes you know where to look. Distributed agent systems require systematic instrumentation, explicit state tracking, and disciplined logging. Build these into your architecture from day one, not after your first 2 AM incident.
Q: Should I log full LLM prompts and completions in production? A: Yes, but with safeguards. Redact PII (emails, phone numbers, addresses) before logging. Use sampling (log 10% of requests) if volume is high. Store logs in a secure location with access controls.
Q: How much logging is too much? A: If logs exceed 20% of your infrastructure costs or slow down agent execution noticeably (>100ms overhead), you're logging too much. Focus on decision points, handoffs, and errors rather than every variable assignment.
Q: Can you debug multi-agent systems without modifying code? A: Partially. External tools like LangSmith and Helicone can capture some data without code changes, but you'll miss internal agent state and custom business logic. Full debuggability requires instrumentation from the start.
Q: How do you handle debugging in production without impacting users? A: Use feature flags to enable verbose logging only for specific workflows or users. Implement sampling (debug 1 in 100 requests). For critical issues, replay failed workflows in a staging environment with full debugging enabled.
Further reading:
External references: