AgentTrace — 因果图根因分析：0.12s 定位多 Agent 故障

1. 解决的问题

AgentTrace 的核心洞察是：多 Agent 系统（MAS）的故障传播是确定性的因果链，而非需要 LLM 进行语义推理的模糊问题。

2. 核心算法逻辑

AgentTrace 的核心洞察是：多 Agent 系统（MAS）的故障传播是确定性的因果链，而非需要 LLM 进行语义推理的模糊问题。

3. 业务应用场景

业务问题：WF-A 智能补货工作流由 3 个 Agent 串联（需求预测 Agent → 采购决策 Agent → PO 下单 Agent）。某天 PO 下单失败，人工排查需要 2-3 小时。

AgentTrace 处理： - 解析执行日志，构建 3 节点 DAG：`DemandForecastAgent → ProcurementAgent → POOrderAgent` - POOrderAgent 报错（`status=FAILED`），但其输入参数（需求量）来自 ProcurementAgent - 反向 BFS：ProcurementAgent 同样标记为 FAILED（收到了 DemandForecastAgent 的异常输出） - DemandForecastAgent 状态为 FAILED（外部数据 API 超时）→ 根因确定 - 0.12s 内完成定位，告警直接指向

量化价值：MTTR 从 2-3 小时降至 5 分钟（含修复），每次供应链故障损失节省 ¥50,000-200,000（缺货成本）

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

5-20 万

7. 代码模板

代码块数量：1 · 路径：paper2skills-code/llm_agent_engineering/agenttrace_causal_rca

# AgentTrace 因果图根因分析 — 完整可运行实现
# 论文：arXiv:2603.14688 | ICLR 2026 AI-Wild Workshop
# 复现核心：DAG 构建 + 反向 BFS 根因定位，无需 LLM，确定性算法

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional
from collections import deque
import time


# ─── 数据结构 ──────────────────────────────────────────────────────────────────

@dataclass
class ExecutionEvent:
    """单个 Agent 执行事件"""
    agent_id: str
    event_type: str          # INVOKE | RETURN | ERROR
    timestamp: float         # Unix epoch (秒)
    inputs: dict             # Agent 输入参数
    outputs: dict            # Agent 输出（含 token/ID 用于因果边推导）
    status: str              # SUCCESS | FAILED | PENDING
    trace_id: str = ""       # 分布式 trace ID（跨服务聚合用）
    error_msg: str = ""      # 错误信息（FAILED 时填写）


@dataclass
class RCAResult:
    """根因分析结果"""
    root_agent: str                  # 根因 Agent ID
    causal_path: list[str]           # 因果传播路径（根因 → 故障终点）
    confidence: float                # 置信度 [0, 1]
    diagnosis_time_ms: float         # 诊断耗时（毫秒）
    error_summary: str = ""          # 根因错误摘要


# ─── 因果 DAG ─────────────────────────────────────────────────────────────────

class CausalDAG:
    """有向无环因果图"""

    def __init__(self):
        self.nodes: dict[str, ExecutionEvent] = {}   # agent_id → event
        self.edges: dict[str, list[str]] = {}         # agent_id → [downstream agent_ids]
        self.reverse_edges: dict[str, list[str]] = {} # agent_id → [upstream agent_ids]

    def add_node(self, event: ExecutionEvent):
        self.nodes[event.agent_id] = event
        self.edges.setdefault(event.agent_id, [])
        self.reverse_edges.setdefault(event.agent_id, [])

    def add_edge(self, from_agent: str, to_agent: str):
        """添加因果边 from → to"""
        if to_agent not in self.edges.get(from_agent, []):
            self.edges[from_agent].append(to_agent)
        if from_agent not in self.reverse_edges.get(to_agent, []):
            self.reverse_edges[to_agent].append(from_agent)

    def get_failed_nodes(self) -> list[str]:
        return [aid for aid, ev in self.nodes.items() if ev.status == "FAILED"]

8. 论文来源

• 2603.14688