多Agent辩论裁决六层栈：从群体盲从到共识收敛

def measure_causal_influence(agent_outputs: list[dict]) -> dict:
    """量化每个Agent对最终输出的因果贡献"""
    # Counterfactual：移除Agent_i的输出，观察结果变化
    influences = {}
    full_output = aggregate_all(agent_outputs)

    for i, agent in enumerate(agent_outputs):
        # 反事实基线：排除Agent_i
        counterfactual = aggregate_all(
            [o for j, o in enumerate(agent_outputs) if j != i]
        )
        # 因果影响 = 有/无Agent_i时的输出差异
        influences[f"agent_{i}"] = semantic_distance(
            full_output, counterfactual
        )

    return influences

graph TB
    subgraph L6["第六层：贝叶斯最优聚合"]
        B6[高阶投票<br/>Bayesian-optimal Aggregation]
    end
    subgraph L5["第五层：动态权重共识"]
        B5[DWC-MAD<br/>历史准确率 × 实时置信度]
    end
    subgraph L4["第四层：结构化对抗"]
        B4[Catfish/鲶鱼Agent<br/>Free-MAD反从众框架]
    end
    subgraph L3["第三层：可验证奖励"]
        B3[Verifiable Reward<br/>审议重启 + 噪声丢弃]
    end
    subgraph L2["第二层：拜占庭容错"]
        B2[WBFT 加权拜占庭<br/>信任权重 × 恶意检测]
    end
    subgraph L1["第一层：潜在空间通信"]
        B1[Latent State Transfer<br/>隐向量传递 + 状态扩散]
    end

    L6 --> L5 --> L4 --> L3 --> L2 --> L1

class LatentCommunicationLayer:
    """Agent间通过潜在空间通信而非文本"""

    def encode_state(self, reasoning_trace, confidence_map, 
                     contradiction_log) -> torch.Tensor:
        """将推理状态编码为高维向量"""
        # 推理轨迹 → 向量
        trace_emb = self.trace_encoder(reasoning_trace)
        # 置信度图谱 → 向量
        conf_emb = self.confidence_encoder(confidence_map)
        # 矛盾记录 → 向量
        contra_emb = self.contradiction_encoder(contradiction_log)

        return torch.cat([trace_emb, conf_emb, contra_emb], dim=-1)

    def decode_state(self, latent_vector: torch.Tensor) -> dict:
        """接收方解码为结构化推理状态"""
        return {
            "reasoning": self.trace_decoder(latent_vector),
            "confidence": self.confidence_decoder(latent_vector),
            "red_flags": self.contradiction_decoder(latent_vector)
        }

class WBFTConsensus:
    """加权拜占庭容错共识引擎"""

    def __init__(self, agents: list, initial_trust: dict):
        self.trust_scores = initial_trust  # {"gpt-4o": 0.9, "claude": 0.88, ...}
        self.history = PerformanceTracker()

    def reach_consensus(self, proposals: dict) -> dict:
        """加权投票达成共识"""
        weighted_votes = {}
        for agent_id, proposal in proposals.items():
            weight = self.trust_scores[agent_id]
            weighted_votes[agent_id] = {
                "proposal": proposal,
                "weight": weight,
                "confidence": proposal.get("confidence", 0.5)
            }

        # 加权聚合
        result = self.weighted_aggregate(weighted_votes)

        # 更新信任评分
        self.update_trust(proposals, result)

        return result

    def update_trust(self, proposals, ground_truth):
        """根据与最终结果的一致性更新信任"""
        for agent_id, proposal in proposals.items():
            alignment = self.compute_alignment(
                proposal, ground_truth
            )
            # 指数移动平均更新
            old = self.trust_scores[agent_id]
            self.trust_scores[agent_id] = (
                0.8 * old + 0.2 * alignment
            )

class DeliberationRestart:
    """审议重启机制"""

    def should_restart(self, reasoning_trace: list) -> bool:
        """检测是否需要重启审议"""
        signals = {
            "circular": self.detect_circular_reasoning(reasoning_trace),
            "stagnation": self.detect_confidence_stagnation(reasoning_trace),
            "lazy_partner": self.detect_lazy_agent(reasoning_trace),
        }
        return any(signals.values())

    def restart(self, agent_state: dict) -> dict:
        """执行审议重启"""
        return {
            # 保留：高层目标
            "goal": agent_state["goal"],
            # 保留：已验证的事实
            "verified_facts": agent_state["verified_facts"],
            # 丢弃：噪声推理历史
            "reasoning_trace": [],
            # 保留：因果影响度量
            "influence_map": agent_state["influence_map"],
        }

# 鲶鱼Agent的投票示例
$ nexus-agents vote \
    --proposal "Add Redis caching to the API gateway" \
    --strategy supermajority

Architect (Claude):    APPROVE  (91%)
Security (Gemini):     APPROVE  (84%)
DevEx (Codex):         APPROVE  (79%)
PM (Claude):           APPROVE  (88%)
Catfish (Gemini):      REJECT   (72%)
  "Redis宕机怎么办？无回退策略。"

Result: APPROVED (4-1, supermajority met)
# 鲶鱼虽然被否决，但其反馈被纳入改进

class DWC_MAD:
    """动态加权共识框架"""

    def compute_agent_weight(self, agent_id: str, 
                              current_response: dict) -> float:
        """计算Agent在当前轮次的动态权重"""
        # 维度1：实时置信度
        realtime_conf = self.quantify_confidence(
            current_response["text"],
            reasoning_depth=current_response["steps"],
            hedging_frequency=current_response["hedging_words"]
        )

        # 维度2：历史准确率
        longitudinal_acc = self.history.get_accuracy(agent_id)

        # 动态权重 = 加权组合
        weight = (
            0.4 * realtime_conf + 
            0.6 * longitudinal_acc
        )

        # 惩罚：近期表现下滑的Agent降权
        recent_trend = self.history.get_trend(agent_id, window=5)
        if recent_trend < 0:  # 趋势下降
            weight *= (1 + recent_trend)  # 负数，所以是降权

        return max(weight, 0.1)  # 保底权重，不完全剥夺话语权

    def weighted_debate_round(self, agents_responses: list) -> dict:
        """执行一轮加权辩论"""
        weights = {
            r["agent_id"]: self.compute_agent_weight(r["agent_id"], r)
            for r in agents_responses
        }

        # 加权聚合答案
        answer_scores = defaultdict(float)
        for response in agents_responses:
            w = weights[response["agent_id"]]
            answer_scores[response["answer"]] += w

        return max(answer_scores, key=answer_scores.get)

from langgraph.graph import StateGraph, END

def debate_round(state):
    """一轮辩论：每个Agent给出观点+置信度"""
    responses = {}
    for agent in state["agents"]:
        resp = agent.invoke({
            "topic": state["topic"],
            "previous_arguments": state["arguments"],
            "my_role": agent.role
        })
        responses[agent.id] = resp
    return {"arguments": responses}

def consensus_check(state):
    """检查是否达成加权共识"""
    weights = compute_dwc_weights(state["arguments"])
    weighted_result = weighted_aggregate(state["arguments"], weights)

    if weighted_result["confidence"] > 0.85:
        return "consensus_reached"
    elif state["round"] >= state["max_rounds"]:
        return "catfish_intervention"  # 触发鲶鱼Agent
    else:
        return "continue_debate"

graph = StateGraph(DebateState)
graph.add_node("debate", debate_round)
graph.add_node("catfish", catfish_agent)
graph.add_conditional_edges("debate", consensus_check, {
    "continue_debate": "debate",
    "catfish_intervention": "catfish",
    "consensus_reached": END
})