MAS Consensus Mechanism — 多智能体共识协议：分布式一致性与拜占庭容错

1. 解决的问题

`Skill-Multi-Agent-Debate` 解决的是"如何让多个 Agent 通过辩论收敛到更好的答案"——这是非正式共识。MAS 共识机制解决的是更严格的问题：在部分 Agent 可能失败或说谎（Byzantine 容错）的情况下，如何保证整个系统仍能达成一致且正确的决策，并有数学证明？

2. 核心算法逻辑

SkillMultiAgentDebate 解决的是"如何让多个 Agent 通过辩论收敛到更好的答案"——这是非正式共识。MAS 共识机制解决的是更严格的问题：在部分 Agent 可能失败或说谎（Byzantine 容错）的情况下，如何保证整个系统仍能达成一致且正确的决策，并有数学证明？

3. 业务应用场景

业务背景：跨境仓储有 5 个区域仓（美东、美西、欧洲、日本、东南亚）。大促前，每个仓库 Agent 基于本地数据给出备货建议，需要就"全局最优备货方案"达成一致。

业务背景：10 个 Agent 评审广告素材合规性，其中 2 个 Agent 数据被污染（数据库异常，给出错误的合规判断）。需要保证最终决策正确。

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

200-500 万

7. 代码模板

代码块数量：5 · 路径：未检测到

import random
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple


@dataclass
class AgentProposal:
    agent_id: str
    value: Any
    confidence: float = 1.0


class AegeanConsensus:
    """
    Aegean 两阶段共识协议
    适用：随机推理 Agent，概率性共识
    """

    def __init__(self, quorum_ratio: float = 0.5, max_rounds: int = 3):
        self.quorum_ratio = quorum_ratio
        self.max_rounds = max_rounds

    def run(self, proposals: List[AgentProposal]) -> Tuple[Optional[Any], bool, int]:
        """
        Returns: (consensus_value, reached_consensus, rounds_taken)
        """
        n = len(proposals)
        quorum_size = int(n * self.quorum_ratio) + 1

        for round_num in range(1, self.max_rounds + 1):
            counts: Dict[str, List[AgentProposal]] = {}
            for p in proposals:
                key = str(round(p.value, 2)) if isinstance(p.value, float) else str(p.value)
                counts.setdefault(key, []).append(p)

            for key, group in counts.items():
                if len(group) >= quorum_size:
                    vals = [p.value for p in group]
                    consensus = sum(vals) / len(vals) if isinstance(vals[0], float) else vals[0]
                    return consensus, True, round_num

            if round_num < self.max_rounds:
                proposals = self._weight_update(proposals)

        best_key = max(counts, key=lambda k: len(counts[k]))
        vals = [p.value for p in counts[best_key]]
        fallback = sum(vals) / len(vals) if isinstance(vals[0], float) else vals[0]
        return fallback, False, self.max_rounds

    def _weight_update(self, proposals: List[AgentProposal]) -> List[AgentProposal]:
        if not proposals:
            return proposals
        all_vals = [p.value for p in proposals if isinstance(p.value, (int, float))]
        if not all_vals:
            return proposals
        mean = sum(all_vals) / len(all_vals)
        updated = []
        for p in proposals:
            if isinstance(p.value, (int, float)):
                distance = abs(p.value - mean)

8. 论文来源

• 2512.20184
• 2605.09076
• 2606.01828