Agent间冲突检测与协商机制：当多个Agent意见不一致时怎么办

{
  "agent_id": "risk_control_01",
  "decision": "REJECT",
  "confidence": 0.92,
  "reasoning": "用户近30天退款5次，触发风控规则R-07",
  "data_sources": ["order_db", "user_profile"],
  "timestamp": "2026-08-04T15:30:00Z",
  "version": "v2.1.0"
}

class ConflictDetector:
    def detect(self, agent_outputs: list[AgentOutput]) -> list[Conflict]:
        conflicts = []
        
        # 1. 决策冲突：决策结果不一致
        decisions = set(o.decision for o in agent_outputs)
        if len(decisions) > 1:
            conflicts.append(DecisionConflict(agent_outputs))
        
        # 2. 置信度冲突：某Agent置信度显著低于群体
        avg_conf = sum(o.confidence for o in agent_outputs) / len(agent_outputs)
        for o in agent_outputs:
            if o.confidence < avg_conf * 0.5:  # 低于均值50%
                conflicts.append(LowConfidenceConflict(o))
        
        # 3. 数据新鲜度冲突：基于不同版本数据
        timestamps = [o.timestamp for o in agent_outputs]
        if max(timestamps) - min(timestamps) > timedelta(minutes=5):
            conflicts.append(StaleDataConflict(agent_outputs))
            
        return conflicts

def majority_vote(agent_outputs: list[AgentOutput]) -> Decision:
    from collections import Counter
    votes = Counter(o.decision for o in agent_outputs)
    return votes.most_common(1)[0][0]

def weighted_vote(agent_outputs: list[AgentOutput]) -> Decision:
    from collections import defaultdict
    scores = defaultdict(float)
    
    for o in agent_outputs:
        # 可选：引入Agent历史准确率作为校准系数
        calibrated_weight = o.confidence * get_agent_accuracy(o.agent_id)
        scores[o.decision] += calibrated_weight
    
    return max(scores, key=scores.get)

class ArbitrationAgent:
    def arbitrate(self, conflict: Conflict) -> Decision:
        prompt = f"""
        以下Agent对同一问题给出了不同结论，请作为仲裁者做出最终决策。
        
        冲突类型：{conflict.type}
        
        Agent结论：
        {self.format_conflict_report(conflict)}
        
        仲裁规则：
        1. 风控类决策优先于体验类决策
        2. 涉及资金安全的决策必须保守
        3. 如果无法确定，选择"拒绝并上报"而非"冒险通过"
        
        请输出：最终决策、决策理由、置信度、是否需要人工复核
        """
        return self.llm.arbitrate(prompt)

conflict_resolution:
  rules:
    - condition: "any_agent_decision == 'BLOCK_TRANSACTION'"
      action: "override_to_block"
      reason: "资金安全优先原则"
    
    - condition: "risk_agent_confidence > 0.95"
      action: "adopt_risk_agent_decision"
      reason: "高风险场景信任专业Agent"
    
    - condition: "all_confidence < 0.6"
      action: "escalate_to_human"
      reason: "群体低置信度，不确定性过高"

def fallback_strategy(conflict: Conflict) -> Decision:
    if conflict.domain in ["payment", "refund", "permission"]:
        return Decision.REJECT  # Fail-Closed
    elif conflict.domain in ["recommendation", "search"]:
        return Decision.DEFAULT  # 返回兜底内容
    else:
        return Decision.ESCALATE  # 转人工

def execute_with_snapshot(agents: list[Agent], context: Context):
    # 1. 为所有Agent生成一致的数据快照
    snapshot = data_layer.snapshot(context.query_id)
    
    # 2. 所有Agent基于同一快照执行
    outputs = [agent.run(snapshot) for agent in agents]
    
    # 3. 快照版本写入日志，便于事后审计
    audit_log.record(query_id=context.query_id, snapshot_version=snapshot.version)
    
    return outputs

{
  "escalation_reason": "CONFIDENCE_TOO_LOW",
  "agent_outputs_summary": [
    {"agent": "risk", "decision": "BLOCK", "confidence": 0.45, "key_evidence": "异常登录地点"},
    {"agent": "fraud", "decision": "ALLOW", "confidence": 0.52, "key_evidence": "历史行为正常"}
  ],
  "recommended_action": "BLOCK_AND_REVIEW",
  "human_override_url": "https://admin.example.com/review/12345"
}

┌─────────────────────────────────────────┐
│           应用层 (Application)           │
│    客服 / 风控 / 推荐 / 代码审查 ...      │
├─────────────────────────────────────────┤
│         编排层 (Orchestrator)            │
│  任务分解 → Agent调度 → 结果聚合         │
├─────────────────────────────────────────┤
│         冲突治理层 (Conflict Governor)    │
│  ┌─────────┐ ┌─────────┐ ┌──────────┐  │
│  │ 检测器   │ │ 协商器   │ │ 回退器    │  │
│  │Detector │ │Mediator │ │Fallback │  │
│  └─────────┘ └─────────┘ └──────────┘  │
├─────────────────────────────────────────┤
│         Agent执行层 (Agent Pool)         │
│  Agent-A  Agent-B  Agent-C  ...         │
├─────────────────────────────────────────┤
│         数据/工具层 (Tools & Data)        │
│  API / DB / Cache / Search              │
└─────────────────────────────────────────┘