Multi-Agent Systems for Distributed Tracing Root Cause Analysis: 2026 and Beyond

import anthropic
import json
 
client = anthropic.Anthropic()
 
 
def identify_critical_path(trace: dict) -> list[dict]:
    """Find the spans that actually explain the latency/error, not every
    span in the trace — most spans in a large trace are irrelevant noise."""
    spans = trace["spans"]
    # Sort by duration contribution and error status
    critical = [s for s in spans if s.get("error") or s["duration_ms"] > trace["total_duration_ms"] * 0.15]
    return sorted(critical, key=lambda s: s["duration_ms"], reverse=True)[:8]
 
 
def dispatch_domain_agents(critical_spans: list[dict]) -> dict:
    """Route each critical span to the right specialized agent based on
    the service's domain — this routing logic is what makes the multi-agent
    approach effective rather than just splitting work arbitrarily."""
    findings = {}
    for span in critical_spans:
        domain = classify_domain(span["service_name"])    # "database" | "network" | "cache" | "application"
        if domain == "database":
            findings[span["service_name"]] = run_database_investigation_agent(span)
        elif domain == "network":
            findings[span["service_name"]] = run_network_investigation_agent(span)
        elif domain == "cache":
            findings[span["service_name"]] = run_cache_investigation_agent(span)
        else:
            findings[span["service_name"]] = run_application_investigation_agent(span)
    return findings

DATABASE_AGENT_TOOLS = [
    {
        "name": "run_explain_analyze",
        "description": "Run EXPLAIN ANALYZE on the query associated with this span",
    },
    {
        "name": "check_lock_contention",
        "description": "Check for blocking locks on the tables this query touched",
    },
    {
        "name": "check_connection_pool_saturation",
        "description": "Check if the connection pool was exhausted at the span's timestamp",
    },
]
 
 
def run_database_investigation_agent(span: dict) -> dict:
    """A tool-use loop scoped specifically to database-layer investigation —
    this agent has tools a network or cache agent would never need."""
    messages = [{
        "role": "user",
        "content": f"""Investigate why this database span was slow/errored:
Service: {span['service_name']}
Duration: {span['duration_ms']}ms
Query context: {span.get('db_statement', 'unknown')}
Timestamp: {span['timestamp']}
 
Use the available tools to investigate and report the specific root cause."""
    }]
 
    response = client.messages.create(
        model="claude-sonnet-5", max_tokens=1500,
        tools=DATABASE_AGENT_TOOLS, messages=messages
    )
    # Standard tool-use loop: execute requested tools, feed results back,
    # repeat until the agent produces a final finding
    return execute_tool_loop(response, messages, DATABASE_AGENT_TOOLS)

SYNTHESIZE_PROMPT = """Multiple specialized agents investigated different
services implicated in this trace's failure. Synthesize their findings
into a single root cause narrative.
 
Per-service findings:
{domain_findings}
 
Trace structure (which service called which, in order):
{trace_topology}
 
Determine: which finding is the actual ROOT cause, and which are downstream
symptoms of it? (e.g. a database connection pool exhaustion causing
cascading timeouts in 6 upstream services isn't 6 root causes, it's 1 root
cause with 6 symptoms — identify this pattern explicitly)"""
 
 
def synthesize_root_cause(domain_findings: dict, trace_topology: list) -> str:
    response = client.messages.create(
        model="claude-sonnet-5", max_tokens=1000,
        messages=[{
            "role": "user",
            "content": SYNTHESIZE_PROMPT.format(
                domain_findings=json.dumps(domain_findings, indent=2),
                trace_topology=trace_topology,
            )
        }]
    )
    return response.content[0].text