多Agent可诊断性六栈：因果追踪到涌现度量

graph TB
    subgraph "多Agent可诊断性六栈"
        L6["L6 治理闭环层<br/>Agent Card · EU AI Act · 策略自适应"]
        L5["L5 分级裁决层<br/>TAO · HITL · 人机分级审查"]
        L4["L4 协作度量层<br/>MultiAgentBench · SwarmBench · 协作质量量化"]
        L3["L3 涌现检测层<br/>MAEBE · 群体偏差识别 · 行为漂移"]
        L2["L2 因果归因层<br/>AgentTrace · 因果图 · 根因定位"]
        L1["L1 执行追踪层<br/>LangSmith · Langfuse · Arize Phoenix · Span"]
    end

    L1 -->|"调用链数据"| L2
    L2 -->|"因果断言"| L3
    L3 -->|"涌现指标"| L4
    L4 -->|"协作评分"| L5
    L5 -->|"裁决策略"| L6
    L6 -->|"治理规则"| L1

    style L1 fill:#4A90D9,color:#fff
    style L2 fill:#E67E22,color:#fff
    style L3 fill:#E74C3C,color:#fff
    style L4 fill:#9B59B6,color:#fff
    style L5 fill:#2ECC71,color:#fff
    style L6 fill:#34495E,color:#fff

import langfuse
from langfuse.decorators import observe

@observe()
def orchestrator(task: str):
    # 每个Agent调用自动生成嵌套trace
    research = research_agent(task)
    draft = writer_agent(research)
    review = critic_agent(draft)
    # langfuse自动记录完整调用链
    return review

# 自定义多Agent评分函数
def collaboration_score(trace):
    """评估Agent间协作效率"""
    agent_spans = [s for s in trace.spans if s.name.endswith("_agent")]
    # 信息传递效率：输出利用率
    utilization = sum(s.output_tokens / s.input_tokens for s in agent_spans) / len(agent_spans)
    # 冗余度：重复信息占比
    redundancy = compute_overlap([s.output for s in agent_spans])
    return {"utilization": utilization, "redundancy": redundancy}

Step 1: 执行日志采集
  → 记录每个Agent的输入、输出、内部推理链

Step 2: 因果图构建
  → 节点 = Agent间传递的消息/数据
  → 边 = "Agent X的输出影响了Agent Y的输入"
  → 权重 = 影响强度（基于attention/prompt贡献度分析）

Step 3: 反事实推理
  → "如果把Agent A的第3次输出替换为正确值，最终结果会变吗？"
  → 通过干预分析计算每个节点的因果效应

Step 4: 根因排序
  → 按因果效应从大到小排列
  → 输出："根因是Agent A的第3次调用，影响了下游3个Agent"

class AgentTrace:
    def build_causal_graph(self, execution_log):
        """从执行日志构建因果图"""
        graph = CausalGraph()
        for step in execution_log.steps:
            # 每个Agent调用是一个节点
            node = graph.add_node(
                agent=step.agent_name,
                input_data=step.input,
                output_data=step.output,
                internal_trace=step.reasoning_chain
            )
            # 连接因果关系：谁喂给了谁
            for dep in step.dependencies:
                graph.add_edge(dep, node, weight=dep.influence_score)
        return graph

    def root_cause_analysis(self, graph, failure_node):
        """反事实推理定位根因"""
        candidates = graph.ancestors(failure_node)  # 所有上游节点
        causal_effects = {}
        for candidate in candidates:
            # 反事实：如果这个节点输出正确，结果会变吗？
            effect = graph.counterfactual(
                intervene=candidate,
                correct_output=candidate.expected_output,
                target=failure_node
            )
            causal_effects[candidate] = effect
        return sorted(causal_effects.items(), key=lambda x: -x[1])

graph LR
    subgraph "MAEBE涌现检测流程"
        A["单Agent安全基线<br/>已知安全测试通过"] --> B["多Agent部署<br/>3-10个Agent协作"]
        B --> C["交互日志采集<br/>每轮对话·决策·工具调用"]
        C --> D{"涌现行为分析"}
        D -->|"集体合理化"| E["错误强化分数<br/>>阈值=预警"]
        D -->|"群体极化"| F["观点偏移量<br/>Δ>阈值=预警"]
        D -->|"责任分散"| G["停止信号强度<br/><阈值=预警"]
        D -->|"涌现策略"| H["策略分类器<br/>有利/有害/中性"]
    end

    style E fill:#E74C3C,color:#fff
    style F fill:#E74C3C,color:#fff
    style G fill:#E74C3C,color:#fff
    style H fill:#F39C12,color:#fff

# SwarmBench评估示例（概念性代码）
class SwarmEvaluation:
    def evaluate(self, agents, task):
        results = {
            "task_completion": 0.0,   # 任务完成率
            "coordination_overhead": 0.0,  # 协调开销
            "robustness": 0.0,        # 鲁棒性（去掉一个Agent后的退化）
            "scalability": 0.0,       # 可扩展性
            "emergence_score": 0.0     # 涌现度（超出预期的协作行为）
        }

        # 基线：每个Agent单独完成
        solo_scores = [agent.solve(task) for agent in agents]
        # 群体：协作完成
        swarm_score = self.orchestrate_swarm(agents, task)

        # 涌现度 = 群体表现 / 最优个体表现
        results["emergence_score"] = swarm_score / max(solo_scores)

        # 鲁棒性：随机移除一个Agent
        reduced = random_remove_one(agents)
        degraded_score = self.orchestrate_swarm(reduced, task)
        results["robustness"] = degraded_score / swarm_score

        return results

graph TB
    subgraph "TAO 分级裁决金字塔"
        T1["Tier 1: 快速检查<br/>8B小模型<br/>过滤95%正常输出<br/>延迟: <100ms"]
        T2["Tier 2: 深度审查<br/>中等模型<br/>过滤80%虚警<br/>延迟: 1-3s"]
        T3["Tier 3: 人类裁决<br/>领域专家<br/>处理1%极端案例<br/>延迟: 分钟级"]
    end

    AGENT["多Agent系统输出<br/>100%"] --> T1
    T1 -->|"可疑: 5%"| T2
    T1 -->|"正常: 95%"| PASS["✓ 通过"]
    T2 -->|"确认问题: 1%"| T3
    T2 -->|"虚警: 4%"| PASS
    T3 -->|"最终决策"| DECISION["裁决结果<br/>修正/拒绝/放行"]

    style T1 fill:#3498DB,color:#fff
    style T2 fill:#E67E22,color:#fff
    style T3 fill:#E74C3C,color:#fff
    style PASS fill:#2ECC71,color:#fff
    style DECISION fill:#9B59B6,color:#fff

class TAOFramework:
    def __init__(self):
        self.tier1 = SmallModelChecker(model="llama-3-8b")
        self.tier2 = DeepReviewer(model="claude-sonnet")
        self.tier3 = HumanExpertPanel()

    async def oversee(self, agent_output, context):
        # Tier 1: 快速检查
        t1_result = await self.tier1.check(agent_output, context)
        if t1_result.confidence > 0.95:
            return Decision.PASS  # 95%直接通过

        # Tier 2: 深度审查
        t2_result = await self.tier2.review(
            agent_output, context,
            t1_flags=t1_result.flags
        )
        if t2_result.action == "correct":
            return t2_result.correction  # 自动纠正
        elif t2_result.action == "pass":
            return Decision.PASS  # 虚警

        # Tier 3: 人类裁决（仅1%走到这里）
        human_decision = await self.tier3.escalate(
            agent_output, context,
            t1_analysis=t1_result,
            t2_analysis=t2_result
        )
        return human_decision

graph LR
    subgraph "治理闭环"
        D["诊断事件<br/>L1-L5的发现"] --> A["归因分析<br/>哪个Agent/模式"]
        A --> P["策略生成<br/>新约束/权限调整"]
        P --> E["执行下发<br/>Agent Card更新"]
        E --> M["效果监控<br/>策略是否有效"]
        M -->|"无效/副作用"| D
    end

    AUDIT["审计溯源链<br/>不可篡改日志"] --- D
    AUDIT --- P
    AUDIT --- E

    style D fill:#3498DB,color:#fff
    style A fill:#E67E22,color:#fff
    style P fill:#9B59B6,color:#fff
    style E fill:#2ECC71,color:#fff
    style M fill:#E74C3C,color:#fff
    style AUDIT fill:#34495E,color:#fff