← AI 智能体探索 · ← 上一篇：结构化辩论协议

辩论的评分与共识

在L2 结构化辩论协议中，我们给裁判装了一把「尺子」——四个维度（逻辑、证据、回应、诚实），每条论据独立打分。这比 L1 的「自由发挥」可靠得多。

但它引入了一个新问题：如果裁判本身就不可靠怎么办？

更具体地说：

• 同一个辩论，两位裁判给出的分数差距很大——你信谁的？
• 裁判 A 对所有论据打分都偏高（平均 8 分），裁判 B 对同样论据打分都偏低（平均 5 分）——你怎么比较他们的排名？
• 三位裁判中有两位说正方更好，一位说反方更好——这算「共识」还是「分歧」？
• 如果评委意见根本不可调和——你是强行选一个，还是承认「这件事我们无法确定」？

这些问题有一个共同的名字：评分者间信度。在人类评估领域（临床诊断、学术评审、司法判决），这是一个被研究了半个多世纪的老问题——但对于 AI Agent 辩论系统，它才刚刚开始被认真对待。

本文要做的是：把人类评估领域成熟的方法论，移植到多 Agent 辩论系统的裁判层。

为什么一位裁判不够

别误会——L2 的 StructuredJudge 在单裁判场景下工作得很好。问题是单点故障。无论你给裁判多详细的评分标准，它都是一个 LLM——它有自己的知识盲区、偏好和随机性。

三个根本原因

这三个问题不是 LLM 特有的。人类评委也有完全一样的问题——这就是为什么学术期刊用 2-4 位审稿人、法庭用陪审团、竞技体育用多裁判并去掉最高最低分。

评分校准

在讨论多裁判之前，先解决一个基础问题：如何让不同裁判的分数可比较？

为什么需要校准

假设你对同一条论据用两位裁判：

裁判 A 的分数始终低 2-3 分。但两位裁判对论据的排序完全一致：PRO-3 > PRO-2 > PRO-1。这说明：

原始分数的绝对值不重要——重要的是相对排序和标准化后的差距。

校准方法一：Z-Score 归一化

对每位裁判的所有分数做标准化：

z_score = (原始分数 - 该裁判的平均分) / 该裁判的标准差

标准化后，所有裁判的分数分布都变成了均值为 0、标准差为 1 的分布。这时你可以直接比较：-1.5 不管来自哪个裁判，都表示「显著低于该裁判的平均水平」。

优点：消除了个体打分习惯差异。
缺点：如果裁判只评了很少的论据（比如 3 条），均值估计不准确。

校准方法二：Min-Max 归一化

把分数压缩到 [0, 1] 区间：

normalized = (原始分数 - 该裁判最低分) / (该裁判最高分 - 该裁判最低分)

优点：简单直观，没有分布假设。
缺点：极端值会严重影响归一化结果——一条 10 分和一条 1 分会让中间的分数挤在一起。

什么时候用哪种

评委间信度：你的裁判们到底多一致

校准之后，下一个问题是：这些裁判到底有多一致？

如果你有三位裁判对同一批论据打分，你需要一个数字来量化他们的一致性。这就是评委间信度要回答的问题。

最常用的两个指标：

对于我们的辩论场景，两个都要用：Alpha 用于数值评分（逻辑/证据/回应/诚实），Kappa 用于分类判定（论据是成立/被驳倒/不确定）。

Alpha 的解读：

多裁判专家组设计

一位裁判不够，你需要一个裁判组。但裁判组不是简单地把同一个裁判跑三次——你需要差异化的裁判角色。

四种裁判角色

根据决策场景，我们定义四种互补的裁判角色：

不同的评分偏重意味着：技术裁判会给逻辑严密的论据更高权重，而商业裁判会给有具体 ROI 数据的论据更高权重。这不是偏袒——这是有目标的差异化。

裁判组的编排流程

多裁判不是并行跑然后取平均——它有严格的执行顺序：

1. 独立评分阶段：每位裁判独立阅读完整辩论记录，独立给出评分和判定。裁判之间不互相通信——这是保证独立性的关键。
2. 校准阶段：对所有裁判的原始分数进行归一化（Z-Score 或 Min-Max），消除个人打分习惯差异。
3. 加权阶段：根据每位裁判的领域相关度和历史准确率，对校准后的分数赋予不同权重。
4. 共识计算阶段：计算 Krippendorff's Alpha 和 Fleiss' Kappa，量化裁判间的一致程度。
5. 分歧决策阶段：如果 Alpha ≥ 0.67，综合加权分数做结论。如果 Alpha < 0.50，标记为「不可调和分歧」，触发人工介入。

加权投票机制

不是所有裁判都等权。我们引入两层权重：

第一层：领域相关度权重

不同裁判对不同类型的辩论有不同的发言权：

这些权重不是拍脑袋——它们应该根据辩论关键词自动匹配。例如辩题中含「架构」「技术栈」「性能」等词时，自动提高技术裁判的权重。

第二层：历史准确率权重

这是更高级的机制——追踪每位裁判在历史辩论中的表现，动态调整权重。

怎么判断裁判的准确性？一个可操作的方法：

• 对于有已知正确答案的校准辩论（如已发生的历史事件、已有明确结果的决策），用真实结果对比裁判的判定。
• 对于没有已知答案的新辩论，用裁判组内的共识偏离度——长期与其他裁判高度不一致的裁判，可能判断力偏弱。
• 每完成 N 场辩论，重新计算每位裁判的准确率权重。

共识度量：Krippendorff's Alpha 和 Fleiss' Kappa

这是本文最核心的技术部分。我们来详细拆解这两个指标的数学原理和代码实现。

Krippendorff's Alpha 的工作原理

Alpha 的核心思想：

α = 1 - (观察到的分歧 / 期望的随机分歧)

α = 1 - (D_o / D_e)

其中:
  D_o = 评委之间实际观察到的分歧（加权平方差之和）
  D_e = 如果评委随机打分，期望的分歧

Alpha = 1.0 表示完全一致（观察到的分歧为 0）。Alpha = 0 表示一致性不比随机好。Alpha 可以为负——表示系统性分歧（评委之间存在反向关系）。

对于定距数据（我们的 1-10 评分），分歧用平方差度量：两位评委给同一条论据的评分相差 N 分，就贡献 N² 的分歧权重。相差 1 分是小分歧，相差 5 分是大分歧。

Fleiss' Kappa 的工作原理

Kappa 用于分类数据。在我们的场景中，它度量的是多位裁判对「论据最终状态」（UPHELD / PARTIALLY_UPHELD / REFUTED / UNCERTAIN）分类的一致性。

κ = (P_o - P_e) / (1 - P_e)

其中:
  P_o = 观察到的评委之间的一致性比例
  P_e = 如果评委随机分类，期望的一致性比例

与 Alpha 不同，Kappa 只关心「是否分类相同」，不关心「分歧的程度」。两位裁判把同一论据判为 UPHELD 和 PARTIALLY_UPHELD，在 Kappa 看来和判为 UPHELD 和 REFUTED 是一样的——都是「不一致」。

这和实际决策场景吻合：对大多数决策者来说，你更关心的是法官们的分类判断是否一致——到底这个论据算不算「成立」。

何时标记「不可调和分歧」

这是多裁判系统最关键的功能：敢于说「我不知道」。

触发不可调和分歧的条件：

1. Alpha < 0.50 且 Kappa < 0.40：裁判在数值评分和分类判定上同时存在严重分歧。
2. 特定论据的评分方差极大：同一条论据，四位裁判分别打出了 2、4、7、9——说明该论据本身就是高度主观的。
3. 加权投票后双方得分接近（差距 < 5% 加权总分）：即便经过校准和加权，双方仍然难分高下。

当触发不可调和分歧时，系统不应强行输出一个「赢家」。它应该：

• 明确告知决策者「当前裁判组无法达成可靠共识」。
• 列出分歧最大的具体论据，供人工审查。
• 建议获取更多信息（如补充数据、外部验证、人类专家意见）。

系统的诚实比系统的自信更重要。一个敢于说「我们的裁判意见严重分歧，这个问题需要你亲自判断」的系统，比一个强行取均值输出「正方胜 5.67 分 vs 5.32 分」的系统，要可靠得多。

代码实现

下面的代码扩展了 L2 的 debate_protocol.py。它新增了四个核心组件：

1. 多裁判面板（MultiJudgePanel）：管理多个差异化裁判，独立评分后综合结果。
2. 分数校准器（ScoreCalibrator）：Z-Score 和 Min-Max 归一化。
3. 加权投票器（WeightedVoter）：领域相关度 + 历史准确率双重加权。
4. 共识计算器（ConsensusCalculator）：Krippendorff's Alpha + Fleiss' Kappa + 分歧检测。

保存为 debate_consensus.py，与 L1 的 debate.py 和 L2 的 debate_protocol.py 放在同一目录下。

"""
辩论共识系统 — 多裁判面板 + 评分校准 + 加权投票 + 共识度量
扩展自 L2 的 debate_protocol.py，引入多裁判编排和共识计算。

依赖: pip install openai numpy
"""
import os
import json
import math
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from enum import Enum

import numpy as np
from openai import OpenAI

# ──────────────────────────────────────────────
# 从 L2 导入核心数据结构（如果你把 L2 代码放在同目录）
# ──────────────────────────────────────────────
# 如果 debate_protocol.py 在同一目录，可以直接 import：
# from debate_protocol import (
#     Argument, CrossExamResponse, ScoringResult,
#     StructuredDebateAgent, StructuredJudge,
#     run_structured_debate
# )
#
# 以下重新定义核心类型，保持本文代码自包含。
# ──────────────────────────────────────────────

client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.example.com/v1"
)


@dataclass
class Argument:
    """结构化论据（复用 L2 定义）"""
    id: str
    claim: str
    reasoning: str
    evidence: str

    def to_dict(self) -> dict:
        return {
            "id": self.id, "claim": self.claim,
            "reasoning": self.reasoning, "evidence": self.evidence
        }

    def to_text(self) -> str:
        return (
            f"[{self.id}] 断言: {self.claim}\n"
            f"推理: {self.reasoning}\n"
            f"证据: {self.evidence}"
        )


# ──────────────────────────────────────────────
# 1. 裁判画像 — 定义差异化角色
# ──────────────────────────────────────────────
class ExpertiseDomain(Enum):
    TECHNICAL = "technical"
    BUSINESS = "business"
    RISK = "risk"
    GENERAL = "general"


@dataclass
class JudgeProfile:
    """定义一位裁判的专业领域和评分偏好"""
    name: str                            # 裁判名称
    domain: ExpertiseDomain              # 专业领域
    # 自定义四个维度的评分权重
    dimension_weights: dict = field(default_factory=lambda: {
        "logic": 0.30, "evidence": 0.30,
        "responsiveness": 0.25, "honesty": 0.15
    })
    # 历史准确率追踪
    historical_accuracy: float = 1.0     # 初始默认 1.0（未校准）
    calibrations_completed: int = 0
    total_correct: int = 0

    def get_system_prompt(self, topic: str) -> str:
        """根据角色生成差异化的系统提示词"""
        base = (
            f"你是一位严格公正的辩论裁判。你的专长领域是"
            f"【{self._domain_cn()}】。\n"
            f"辩题: 「{topic}」\n\n"
        )
        domain_instructions = {
            ExpertiseDomain.TECHNICAL: (
                f"你特别关注技术实现的可行性和架构设计的合理性。"
                f"对于技术细节和实现路径的论证，你会进行深入审查。"
                f"你不接受没有具体实现路径的空洞技术承诺。"
            ),
            ExpertiseDomain.BUSINESS: (
                f"你特别关注商业合理性和成本效益分析。"
                f"对于 ROI 数据、市场论证和资源投入产出比，"
                f"你会进行严格的审查。"
                f"你不接受没有量化支撑的商业论断。"
            ),
            ExpertiseDomain.RISK: (
                f"你特别关注风险和边界条件。你会主动寻找论证中"
                f"被忽略的风险因素、假设条件、失败模式。"
                f"你关注的不是「理想情况下的最优解」，"
                f"而是「最坏情况下是否可承受」。"
            ),
            ExpertiseDomain.GENERAL: (
                f"你进行全面的综合评估，平衡技术、商业和风险因素。"
                f"你关注辩论的整体信息质量和论证完整性。"
            ),
        }
        base += domain_instructions[self.domain]
        base += (
            "\n\n### 评分规则\n"
            "对每一条开场论据从以下四个维度打分（1-10 整数）：\n"
            f"1. logic_score: 推理链是否自洽？"
            f"（你对该维度的权重: {self.dimension_weights['logic']}）\n"
            f"2. evidence_score: 证据是否具体可验证？"
            f"（权重: {self.dimension_weights['evidence']}）\n"
            f"3. responsiveness_score: 回应质询的质量？"
            f"（权重: {self.dimension_weights['responsiveness']}）\n"
            f"4. honesty_score: 诚实度？"
            f"（权重: {self.dimension_weights['honesty']}）\n\n"
            "对每条论据，也给出最终状态判定: "
            "UPHELD | PARTIALLY_UPHELD | REFUTED | UNCERTAIN\n\n"
            "### 输出格式\n"
           

... [OUTPUT TRUNCATED - 4968 chars omitted out of 54968 total] ...

topic_type, cls.TOPIC_WEIGHTS["default"]
        )
        domain_weight = weights.get(profile.domain, 0.20)

        # 融合历史准确率（如果有校准数据）
        accuracy_weight = profile.historical_accuracy
        if profile.calibrations_completed == 0:
            accuracy_weight = 1.0  # 无历史数据，准确率权重为 1（不调整）

        # 最终权重 = 0.7 × 领域相关度 + 0.3 × 历史准确率
        return 0.7 * domain_weight + 0.3 * accuracy_weight

    @classmethod
    def compute_weighted_scores(
        cls,
        judges: list[JudgeProfile],
        calibrated_scores: dict[str, dict[str, float]],
        topic: str
    ) -> dict[str, float]:
        """
        对每位裁判的校准分数进行加权求和。

        参数:
            judges: 裁判画像列表
            calibrated_scores: {裁判名: {论据ID: 校准后分数}}
            topic: 辩论主题

        返回:
            {"pro": 正方加权总分, "con": 反方加权总分,
             "pro_details": {...}, "con_details": {...}}
        """
        judge_name_to_profile = {j.name: j for j in judges}

        pro_weighted = 0.0
        con_weighted = 0.0
        total_weight = 0.0
        pro_details = defaultdict(float)
        con_details = defaultdict(float)

        for judge_name, arg_scores in calibrated_scores.items():
            profile = judge_name_to_profile.get(judge_name)
            if not profile:
                continue

            weight = cls.get_domain_weight(profile, topic)

            for arg_id, score in arg_scores.items():
                weighted = score * weight
                total_weight += weight
                if arg_id.startswith("PRO"):
                    pro_weighted += weighted
                    pro_details[judge_name] += weighted
                elif arg_id.startswith("CON"):
                    con_weighted += weighted
                    con_details[judge_name] += weighted

        # 归一化
        if total_weight > 0:
            pro_weighted /= total_weight
            con_weighted /= total_weight

        return {
            "pro": round(pro_weighted, 3),
            "con": round(con_weighted, 3),
            "pro_details": dict(pro_details),
            "con_details": dict(con_details),
            "topic_type": WeightedVoter.detect_topic_type(topic),
        }


# ──────────────────────────────────────────────
# 4. 共识计算器
# ──────────────────────────────────────────────
class ConsensusCalculator:
    """计算 Krippendorff's Alpha 和 Fleiss' Kappa"""

    @staticmethod
    def krippendorff_alpha(
        reliability_data: list[list[float]],
        metric: str = "interval"
    ) -> float:
        """
        计算 Krippendorff's Alpha。

        参数:
            reliability_data: 形状为 (n_judges, n_items) 的矩阵。
                每一行是一位裁判对所有论据的评分。
                NaN 表示缺失值。
            metric: 距离度量类型。
                "interval" - 平方差（适用于 1-10 定距评分）
                "nominal"  - 相等性（适用于分类数据）

        返回:
            Alpha 值。1.0 = 完全一致，0 = 随机水平一致，可能为负。
        """
        data = np.array(reliability_data, dtype=float)
        n_raters, n_items = data.shape

        if n_raters < 2 or n_items < 2:
            return 1.0

        # 距离函数
        if metric == "nominal":
            def delta(a, b):
                return 0.0 if a == b else 1.0
        else:  # interval
            def delta(a, b):
                return (a - b) ** 2

        # 计算观察到的分歧 D_o
        D_o = 0.0
        n_pairs = 0
        for i in range(n_items):
            for r1 in range(n_raters):
                if np.isnan(data[r1, i]):
                    continue
                for r2 in range(r1 + 1, n_raters):
                    if np.isnan(data[r2, i]):
                        continue
                    D_o += delta(data[r1, i], data[r2, i])
                    n_pairs += 1

        if n_pairs == 0:
            return 1.0

        D_o /= n_pairs

        # 计算期望分歧 D_e（对分布中所有值对求距离均值）
        all_values = []
        for i in range(n_items):
            for r in range(n_raters):
                if not np.isnan(data[r, i]):
                    all_values.append(data[r, i])

        if len(all_values) < 2:
            return 1.0

        D_e = 0.0
        n_val_pairs = 0
        for v1 in all_values:
            for v2 in all_values:
                D_e += delta(v1, v2)
                n_val_pairs += 1

        if n_val_pairs == 0 or D_e == 0:
            return 1.0 if D_o == 0 else 0.0

        D_e /= n_val_pairs
        alpha = 1.0 - (D_o / D_e)
        return round(alpha, 4)

    @staticmethod
    def fleiss_kappa(
        classifications: list[list[str]]
    ) -> float:
        """
        计算 Fleiss' Kappa（用于分类数据）。

        参数:
            classifications: 形状为 (n_items, n_raters) 的矩阵。
                每一行是一条论据，每一列是一位裁判的分类判定。
                例如: [["UPHELD", "UPHELD", "PARTIALLY_UPHELD"], ...]

        返回:
            Kappa 值。
        """
        # 转置为 (n_raters, n_items) 方便读取
        data = np.array(classifications, dtype=str).T
        n_raters, n_items = data.shape

        if n_raters < 2 or n_items < 2:
            return 1.0

        # 收集所有类别
        all_categories = sorted(set(
            c for row in classifications for c in row
        ))
        n_categories = len(all_categories)
        if n_categories < 2:
            return 1.0

        cat_to_idx = {c: i for i, c in enumerate(all_categories)}

        # 每个论据中各类别的评分者数量
        count_matrix = np.zeros((n_items, n_categories))
        for i in range(n_items):
            for r in range(n_raters):
                j = cat_to_idx[data[r, i]]
                count_matrix[i, j] += 1

        # P_i: 论据 i 上裁判之间的一致性
        P_i = np.zeros(n_items)
        for i in range(n_items):
            total = 0.0
            for j in range(n_categories):
                total += count_matrix[i, j] * (count_matrix[i, j] - 1)
            if n_raters > 1:
                P_i[i] = total / (n_raters * (n_raters - 1))

        # P_bar: 所有论据的平均一致性
        P_bar = np.mean(P_i)

        # p_j: 每个类别的总体比例
        p_j = np.sum(count_matrix, axis=0) / (n_items * n_raters)

        # P_e: 期望一致性
        P_e = np.sum(p_j ** 2)

        if P_e >= 1.0:
            return 1.0 if P_bar >= 1.0 else 0.0

        kappa = (P_bar - P_e) / (1.0 - P_e)
        return round(kappa, 4)

    @staticmethod
    def detect_irreconcilable(
        alpha: float,
        kappa: float,
        per_arg_variances: dict[str, float],
        weighted_scores: dict,
        alpha_threshold: float = 0.50,
        kappa_threshold: float = 0.40,
        variance_threshold: float = 3.0,
        score_gap_threshold: float = 0.05
    ) -> dict:
        """
        检测是否存在不可调和的分歧。

        返回:
            {
                "irreconcilable": bool,
                "reasons": [原因列表],
                "high_variance_args": [高分歧论据ID],
                "recommendation": str
            }
        """
        reasons = []
        high_variance_args = [
            arg_id for arg_id, var in per_arg_variances.items()
            if var >= variance_threshold
        ]

        score_gap = abs(
            weighted_scores.get("pro", 0) -
            weighted_scores.get("con", 0)
        )

        irreconcilable = False

        if alpha < alpha_threshold and kappa < kappa_threshold:
            irreconcilable = True
            reasons.append(
                f"Alpha ({alpha}) 和 Kappa ({kappa}) 同时低于阈值，"
                f"裁判在数值评分和分类判定上均存在严重分歧"
            )
        elif alpha < alpha_threshold:
            reasons.append(
                f"Alpha ({alpha}) 低于阈值 ({alpha_threshold})，"
                f"裁判评分一致性不足"
            )
        elif kappa < kappa_threshold:
            reasons.append(
                f"Kappa ({kappa}) 低于阈值 ({kappa_threshold})，"
                f"裁判分类判定一致性不足"
            )

        if high_variance_args:
            reasons.append(
                f"以下论据的裁判间评分方差过大: "
                f"{', '.join(high_variance_args)}"
            )
            irreconcilable = True

        if score_gap < score_gap_threshold:
            reasons.append(
                f"加权后双方得分差距 ({score_gap:.3f}) 小于阈值 "
                f"({score_gap_threshold})，双方实力过于接近"
            )
            irreconcilable = True

        if not irreconcilable:
            recommendation = (
                "裁判组达成可接受的共识。可基于加权评分做决策，"
                "但建议人工审查高分歧论据。"
            )
        else:
            recommendation = (
                "裁判组存在不可调和的分歧。建议: (1) 人工审查高分歧论据；"
                "(2) 补充更多数据或外部信息；(3) 邀请人类专家介入。"
                "在分歧解决之前，不应基于此辩论结果做关键决策。"
            )

        return {
            "irreconcilable": irreconcilable,
            "reasons": reasons,
            "high_variance_args": high_variance_args,
            "alpha": alpha,
            "kappa": kappa,
            "score_gap": round(score_gap, 4),
            "recommendation": recommendation,
        }


# ──────────────────────────────────────────────
# 5. 多裁判面板 — 编排所有裁判
# ──────────────────────────────────────────────
@dataclass
class PanelResult:
    """多裁判面板的输出"""
    # 每位裁判的原始评分
    raw_scores: dict[str, list[dict]] = field(default_factory=dict)
    # 校准后的分数 {裁判名: {论据ID: 分数}}
    calibrated_scores: dict[str, dict[str, float]] = field(
        default_factory=dict
    )
    # 加权投票结果
    weighted_result: dict = field(default_factory=dict)
    # 共识指标
    alpha: float = 0.0
    kappa: float = 0.0
    # 分歧检测
    divergence: dict = field(default_factory=dict)
    # 论据级别的统计
    per_arg_stats: dict[str, dict] = field(default_factory=dict)


class MultiJudgePanel:
    """
    多裁判面板 — 管理多位差异化裁判，
    独立评分 → 校准 → 加权投票 → 共识计算 → 分歧检测。
    """

    def __init__(self, judges: list[JudgeProfile]):
        """
        参数:
            judges: 裁判画像列表，至少需要 2 位裁判
        """
        if len(judges) < 2:
            raise ValueError("多裁判面板至少需要 2 位裁判")
        self.judges = judges
        self.calibrator = ScoreCalibrator()
        self.voter = WeightedVoter()
        self.consensus = ConsensusCalculator()

    def _single_judge_evaluate(
        self, profile: JudgeProfile,
        pro_args: list[Argument],
        con_args: list[Argument],
        pro_cross_text: str,
        con_cross_text: str,
        pro_closing: str,
        con_closing: str,
        topic: str
    ) -> dict:
        """
        调用 LLM 让单一位裁判评估辩论。
        返回解析后的 JSON 评分。
        """
        user_prompt = (
            f"## 辩论主题\n{topic}\n\n"
            f"## 正方开场论据\n" +
            "\n\n".join(a.to_text() for a in pro_args) +
            f"\n\n## 反方开场论据\n" +
            "\n\n".join(a.to_text() for a in con_args) +
            f"\n\n## 正方质询\n{pro_cross_text}\n\n"
            f"## 反方质询\n{con_cross_text}\n\n"
            f"## 正方总结\n{pro_closing}\n\n"
            f"## 反方总结\n{con_closing}\n\n"
            f"请按系统提示中的 JSON 格式输出你的评估。"
        )

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system",
                 "content": profile.get_system_prompt(topic)},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.2,
            max_tokens=3000
        )

        reply = response.choices[0].message.content

        # 清理可能的 markdown 代码块
        import re
        cleaned = re.sub(r'```(?:json)?\s*', '', reply).strip()
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError:
            return {
                "error": "JSON 解析失败",
                "raw_response": reply,
                "scores": [],
                "overall": {"pro_total_raw": 0, "con_total_raw": 0}
            }

    def evaluate(
        self,
        topic: str,
        pro_args: list[Argument],
        con_args: list[Argument],
        pro_cross_text: str,
        con_cross_text: str,
        pro_closing: str,
        con_closing: str
    ) -> PanelResult:
        """
        执行完整的多裁判评估流程。
        """
        result = PanelResult()

        # ── 阶段 1: 独立评分 ──
        print(f"\n{'=' * 60}")
        print(f"👥 多裁判面板 — {len(self.judges)} 位裁判独立评估")
        print(f"{'=' * 60}")

        for judge in self.judges:
            print(f"\n  📋 {judge.name} ({judge._domain_cn()}) 评估中...")
            evaluation = self._single_judge_evaluate(
                judge, pro_args, con_args,
                pro_cross_text, con_cross_text,
                pro_closing, con_closing, topic
            )
            result.raw_scores[judge.name] = evaluation.get("scores", [])

        # ── 阶段 2: 分数校准 ──
        print(f"\n{'─' * 60}")
        print(f"📐 阶段 2: 分数校准 (Z-Score 归一化)")
        print(f"{'─' * 60}")

        # 提取每位裁判对每条论据的加权平均分
        raw_per_judge = {}
        for judge in self.judges:
            scores = result.raw_scores.get(judge.name, [])
            if not scores:
                continue
            dw = judge.dimension_weights
            raw_per_judge[judge.name] = []
            for s in scores:
                weighted = (
                    s.get("logic_score", 5) * dw["logic"] +
                    s.get("evidence_score", 5) * dw["evidence"] +
                    s.get("responsiveness_score", 5) * dw["responsiveness"] +
                    s.get("honesty_score", 5) * dw["honesty"]
                )
                raw_per_judge[judge.name].append(weighted)

        # 收集所有论据 ID（按顺序）
        all_arg_ids = []
        for judge in self.judges:
            for s in result.raw_scores.get(judge.name, []):
                aid = s.get("argument_id", "")
                if aid and aid not in all_arg_ids:
                    all_arg_ids.append(aid)

        # 对每位裁判的分数做 Z-Score 归一化
        calibrated = self.calibrator.calibrate_all(raw_per_judge)

        # 映射到论据 ID
        result.calibrated_scores = {}
        for judge in self.judges:
            jname = judge.name
            if jname not in calibrated or jname not in raw_per_judge:
                continue
            raw_list = raw_per_judge[jname]
            cal_list = calibrated[jname]
            j_scores = result.raw_scores.get(jname, [])
            result.calibrated_scores[jname] = {}
            for i, s in enumerate(j_scores):
                aid = s.get("argument_id", f"UNKNOWN-{i}")
                result.calibrated_scores[jname][aid] = (
                    cal_list[i] if i < len(cal_list) else 0.0
                )

        # ── 阶段 3: 加权投票 ──
        print(f"\n{'─' * 60}")
        print(f"⚖️  阶段 3: 加权投票")
        print(f"{'─' * 60}")

        result.weighted_result = self.voter.compute_weighted_scores(
            self.judges, result.calibrated_scores, topic
        )
        print(f"  辩论类别: {result.weighted_result['topic_type']}")
        for j in self.judges:
            w = self.voter.get_domain_weight(j, topic)
            print(f"  {j.name}: 最终权重 = {w:.3f}")

        # ── 阶段 4: 共识计算 ──
        print(f"\n{'─' * 60}")
        print(f"🤝 阶段 4: 共识度量")
        print(f"{'─' * 60}")

        # 构建 Krippendorff's Alpha 数据矩阵 (n_judges, n_items)
        alpha_data = []
        for judge in self.judges:
            jname = judge.name
            row = []
            for aid in all_arg_ids:
                row.append(
                    result.calibrated_scores.get(jname, {}).get(aid, np.nan)
                )
            alpha_data.append(row)

        result.alpha = self.consensus.krippendorff_alpha(
            alpha_data, metric="interval"
        )

        # 构建 Fleiss' Kappa 分类数据 (n_items, n_raters)
        kappa_data = []
        for i, aid in enumerate(all_arg_ids):
            standings = []
            for judge in self.judges:
                j_scores = result.raw_scores.get(judge.name, [])
                if i < len(j_scores):
                    standings.append(j_scores[i].get("standing", "UNCERTAIN"))
                else:
                    standings.append("UNCERTAIN")
            kappa_data.append(standings)

        result.kappa = self.consensus.fleiss_kappa(kappa_data)

        print(f"  Krippendorff's Alpha: {result.alpha}")
        print(f"  Fleiss' Kappa:       {result.kappa}")

        # ── 阶段 5: 分歧检测 ──
        print(f"\n{'─' * 60}")
        print(f"🔍 阶段 5: 分歧检测")
        print(f"{'─' * 60}")

        # 计算每条论据的评分方差（跨裁判）
        per_arg_var = {}
        for i, aid in enumerate(all_arg_ids):
            vals = []
            for judge in self.judges:
                jname = judge.name
                val = result.calibrated_scores.get(jname, {}).get(aid)
                if val is not None and not (
                    isinstance(val, float) and math.isnan(val)
                ):
                    vals.append(val)
            if len(vals) >= 2:
                per_arg_var[aid] = float(np.var(vals, ddof=1))
            else:
                per_arg_var[aid] = 0.0

        result.divergence = self.consensus.detect_irreconcilable(
            alpha=result.alpha,
            kappa=result.kappa,
            per_arg_variances=per_arg_var,
            weighted_scores=result.weighted_result,
        )

        print(f"  不可调和分歧: {'⚠️ 是' if result.divergence['irreconcilable'] else '✅ 否'}")
        for reason in result.divergence.get("reasons", []):
            print(f"    - {reason}")

        # ── 阶段 6: 论据级别统计 ──
        result.per_arg_stats = {}
        for i, aid in enumerate(all_arg_ids):
            standings = []
            for judge in self.judges:
                j_scores = result.raw_scores.get(judge.name, [])
                if i < len(j_scores):
                    standings.append(
                        j_scores[i].get("standing", "UNCERTAIN")
                    )
            counter = Counter(standings)
            result.per_arg_stats[aid] = {
                "variance": per_arg_var.get(aid, 0.0),
                "standings": dict(counter),
                "majority": counter.most_common(1)[0][0]
                if counter else "UNCERTAIN",
            }

        return result

    def print_report(self, result: PanelResult, topic: str):
        """打印人类可读的综合报告"""
        print(f"\n{'=' * 60}")
        print(f"📊 多裁判综合报告")
        print(f"{'=' * 60}")
        print(f"\n辩论主题: {topic}")
        print(f"裁判人数: {len(self.judges)}")
        print(f"辩论类别: {result.weighted_result.get('topic_type', 'N/A')}")

        print(f"\n── 共识指标 ──")
        a_label = (
            "✅ 高度一致" if result.alpha >= 0.80
            else "⚠️ 需关注" if result.alpha < 0.67
            else "📌 中等一致"
        )
        k_label = (
            "✅ 高度一致" if result.kappa >= 0.80
            else "⚠️ 需关注" if result.kappa < 0.67
            else "📌 中等一致"
        )
        print(f"  Krippendorff's Alpha: {result.alpha} ({a_label})")
        print(f"  Fleiss' Kappa:       {result.kappa} ({k_label})")

        print(f"\n── 加权评分 ──")
        print(f"  正方: {result.weighted_result.get('pro', 'N/A')}")
        print(f"  反方: {result.weighted_result.get('con', 'N/A')}")
        gap = abs(
            result.weighted_result.get("pro", 0) -
            result.weighted_result.get("con", 0)
        )
        print(f"  差距: {gap:.3f}")

        print(f"\n── 分歧状态 ──")
        print(f"  不可调和: {'⚠️ 是' if result.divergence.get('irreconcilable') else '✅ 否'}")
        for reason in result.divergence.get("reasons", []):
            print(f"    - {reason}")
        print(f"  建议: {result.divergence.get('recommendation', 'N/A')}")


# ──────────────────────────────────────────────
# 6. 使用示例
# ──────────────────────────────────────────────
def run_consensus_debate(topic: str) -> PanelResult:
    """
    运行带多裁判共识计算的完整辩论。

    这个函数假设你已经通过 L2 的 debate_protocol.py
    获取了辩论记录。这里用模拟数据演示裁判面板流程。
    """

    # ── 创建裁判组 ──
    judges = [
        JudgeProfile(
            name="技术裁判",
            domain=ExpertiseDomain.TECHNICAL,
            dimension_weights={
                "logic": 0.40, "evidence": 0.35,
                "responsiveness": 0.15, "honesty": 0.10
            }
        ),
        JudgeProfile(
            name="商业裁判",
            domain=ExpertiseDomain.BUSINESS,
            dimension_weights={
                "logic": 0.20, "evidence": 0.40,
                "responsiveness": 0.20, "honesty": 0.20
            }
        ),
        JudgeProfile(
            name="风险裁判",
            domain=ExpertiseDomain.RISK,
            dimension_weights={
                "logic": 0.15, "evidence": 0.20,
                "responsiveness": 0.35, "honesty": 0.30
            }
        ),
        JudgeProfile(
            name="综合裁判",
            domain=ExpertiseDomain.GENERAL,
            dimension_weights={
                "logic": 0.30, "evidence": 0.30,
                "responsiveness": 0.25, "honesty": 0.15
            }
        ),
    ]

    # ── 创建裁判面板 ──
    panel = MultiJudgePanel(judges)

    # ── 准备辩论数据（模拟，实际使用中从 L2 的
    #     run_structured_debate() 获取） ──
    pro_args = [
        Argument("PRO-1", "独立部署缩短发布周期",
                 "微服务允许独立构建/测试/部署，避免单体全量部署的瓶颈",
                 "基准测试: 单体 3.8h vs 微服务 0.7h"),
        Argument("PRO-2", "团队技术栈灵活性提高",
                 "每个服务可独立选择最合适的技术栈",
                 "某创业公司案例: 核心用 Go + 分析用 Python"),
        Argument("PRO-3", "故障隔离降低系统风险",
                 "单个服务故障不影响其他服务正常运行",
                 "AWS 可用区实践: 故障半径从全集群降至单服务"),
    ]

    con_args = [
        Argument("CON-1", "运维复杂度显著增加",
                 "微服务引入分布式系统固有复杂性: "
                 "网络延迟、服务发现、分布式事务",
                 "研究表明运维成本增加 40-60%"),
        Argument("CON-2", "团队认知负荷过高",
                 "10 人团队维护 8+ 个服务，"
                 "每个开发者需要理解多个服务的交互",
                 "小型团队调研: 超过 5 个服务后开发效率下降"),
        Argument("CON-3", "初期开发速度下降",
                 "微服务需要额外的基础设施搭建和 DevOps 投入",
                 "初创公司通常在 6-12 个月后才看到 ROI"),
    ]

    pro_cross_text = (
        "对 CON-1: 挑战 — 运维成本增加的数字是否包含"
        "现代容器编排工具的自动化能力？\n"
        "对 CON-2: 部分承认 — 确实有认知负荷，但可通过"
        "统一 API 网关和文档规范缓解\n"
        "对 CON-3: 承认 — 初期速度确实会下降，但长期收益"
        "值得投入"
    )

    con_cross_text = (
        "对 PRO-1: 挑战 — 基准测试条件过于理想化，"
        "未包含网络延迟和 CI/CD 流水线时间\n"
        "对 PRO-2: 挑战 — 技术栈多样性在小型团队中"
        "反而增加招聘和维护难度\n"
        "对 PRO-3: 反驳 — 故障隔离是有代价的，"
        "分布式系统本身引入新的故障模式"
    )

    pro_closing = (
        "我们承认微服务在运维复杂度和初期开发速度上的不足。"
        "但核心主张不变：对于需要长期发展且预期快速增长的"
        "创业公司，微服务的独立部署能力和故障隔离优势"
        "在长远来看胜出。"
    )

    con_closing = (
        "正方未能有效回应运维成本和团队认知负荷的核心挑战。"
        "对于 10 人以下团队，微服务引入的复杂度不匹配团队规模。"
        "建议先用模块化单体架构，在团队和业务增长到"
        "必要规模后再拆分。"
    )

    # ── 运行多裁判评估 ──
    result = panel.evaluate(
        topic=topic,
        pro_args=pro_args,
        con_args=con_args,
        pro_cross_text=pro_cross_text,
        con_cross_text=con_cross_text,
        pro_closing=pro_closing,
        con_closing=con_closing,
    )

    # ── 打印报告 ──
    panel.print_report(result, topic)

    return result


# ──────────────────────────────────────────────
# 7. 辅助函数: 校准面板（用于追踪历史准确率）
# ──────────────────────────────────────────────
def update_judge_accuracy(
    profile: JudgeProfile,
    ground_truth: str,  # "PRO" | "CON" | "TIE"
    judge_vote: str     # "PRO" | "CON" | "TIE"
):
    """
    根据已知正确答案更新裁判的历史准确率。
    仅用于有 ground-truth 的校准辩论。
    """
    profile.calibrations_completed += 1
    if judge_vote == ground_truth:
        profile.total_correct += 1
    profile.historical_accuracy = (
        profile.total_correct / profile.calibrations_completed
    )


# ──────────────────────────────────────────────
# 8. 无需 LLM 的纯统计测试（快速验证算法正确性）
# ──────────────────────────────────────────────
def test_consensus_without_llm():
    """不调用 LLM，用模拟数据验证共识算法。"""
    import numpy as np  # noqa: F811
    print("=" * 60)
    print("🧪 纯统计测试 — 验证共识算法（无需 LLM）")
    print("=" * 60)

    # 模拟：4 位裁判 × 6 条论据
    mock_scores = [
        [7.5, 8.0, 6.5, 4.0, 3.5, 5.0],  # 裁判 1
        [8.0, 8.5, 7.0, 3.5, 3.0, 4.5],  # 裁判 2
        [6.0, 7.0, 5.5, 5.0, 4.5, 6.0],  # 裁判 3 (分歧较大)
        [np.nan, 8.0, 6.0, 4.0, np.nan, 5.0],  # 裁判 4 (含缺失值)
    ]

    calc = ConsensusCalculator()
    alpha = calc.krippendorff_alpha(mock_scores)
    print(f"\nKrippendorff's Alpha (模拟数据): {alpha}")
    print(f"预期: 0.70-0.90 之间（存在一定分歧）")

    # 模拟分类数据
    mock_classifications = [
        ["UPHELD", "UPHELD", "PARTIALLY_UPHELD",
         "REFUTED", "REFUTED", "PARTIALLY_UPHELD"],
        ["UPHELD", "UPHELD", "UPHELD",
         "REFUTED", "REFUTED", "REFUTED"],
        ["UPHELD", "PARTIALLY_UPHELD", "PARTIALLY_UPHELD",
         "PARTIALLY_UPHELD", "REFUTED", "UNCERTAIN"],
        ["UPHELD", "UPHELD", "PARTIALLY_UPHELD",
         "REFUTED", "REFUTED", "REFUTED"],
    ]

    # Fleiss' Kappa 需要 (n_items, n_raters) 格式
    # 当前数据是 (n_raters, n_items)，需要转置
    kappa_data = list(zip(*mock_classifications))
    kappa_data = [list(row) for row in kappa_data]
    kappa = calc.fleiss_kappa(kappa_data)
    print(f"\nFleiss' Kappa (模拟数据): {kappa}")
    print(f"预期: 0.60-0.90 之间（大部分一致，小部分分歧）")

    # 测试校准
    cal = ScoreCalibrator()
    raw = {
        "裁判A": [5.0, 6.0, 7.0, 4.0, 3.0, 5.0],
        "裁判B": [8.0, 9.0, 10.0, 7.0, 6.0, 8.0],
    }
    calibrated = cal.calibrate_all(raw, method="zscore")
    print(f"\nZ-Score 校准:")
    for name, scores in calibrated.items():
        print(f"  {name}: {[round(s, 3) for s in scores]}")
    print(f"  预期: 两裁判的 Z-Score 分布应几乎相同")


if __name__ == "__main__":
    # 先运行纯统计测试（无需 LLM，无需 API）
    test_consensus_without_llm()

    print(f"\n{'=' * 60}")
    print(f"💡 要运行完整的多裁判辩论评估，请调用 run_consensus_debate()")
    print(f"   需要配置有效的 API 凭证（your-api-key + api.example.com）")
    print(f"{'=' * 60}")

    # 如果需要运行完整评估，取消以下注释：
    # result = run_consensus_debate(
    #     topic="小型创业公司（10人以下）"
    #           "是否应该从第一天就采用微服务架构？"
    # )
    # with open("/tmp/consensus_debate_result.json", "w") as f:
    #     json.dump(result, f, ensure_ascii=False, indent=2, default=str)

代码结构解析

相比 L2 的 debate_protocol.py，L3 新增了以下核心组件：

使用流程：从辩论到决策

把 L1、L2、L3 的代码串起来看：

1. L1 (debate.py)：两个 Agent 对抗辩论，单一裁判给出自由文本结论。
   适用场景：快速探索、头脑风暴。
2. L2 (debate_protocol.py)：3 轮结构化协议（开场→质询→总结），单一裁判提供多维度评分 + 论据追溯表。
   适用场景：在确定性问题上的深度辩论。
3. L3 (debate_consensus.py)：多裁判专家面板，评分校准 + 加权投票 + 共识度量 + 分歧检测。
   适用场景：关键决策——需要多人验证结论可靠性的场景。

你可以渐进式采用：先用 L1 试试看，如果发现辩论深度不够，升级到 L2；如果发现单裁判结论不可靠，升级到 L3。

局限性与未来方向

1. LLM 裁判的自我一致性：我们假设同一位裁判对同一场辩论跑两次会给出相近的分数——但这个假设不一定成立。即使 temperature=0.2，LLM 仍有一定随机性。生产环境中应该对每位裁判跑 2-3 次，用平均分作为该裁判的最终评分。
2. Alpha 和 Kappa 的小样本问题：如果一场辩论只有 6 条论据且只有 3 位裁判，Alpha 和 Kappa 的估计会有较大方差。在论据少于 10 条或裁判少于 3 位时，这些指标仅供参考，不应作为唯一决策依据。
3. 裁判可能集体犯错：如果所有裁判共享相同的知识盲区（因为它们的训练数据相似），即便 Alpha 很高，结论也可能是错的。使用不同厂商的模型作为裁判（GPT-4o + Claude + Gemini）可以在一定程度上缓解这一问题。
4. 领域相关度权重需要人工校准：本文用的关键词匹配来确定辩论类型比较粗糙。在严肃应用中，应该由人工标注辩论类别，或训练一个分类器来识别。

关键收获

1. 单裁判 = 单点故障：无论裁判系统多精细，一个裁判的观点总是不完整的。多裁判专家组是保障结论可靠性的基础设施。
2. 校准让分数可比较：Z-Score 归一化消除了裁判之间的「打分习惯」差异，让你能真正比较不同裁判对同一论据的评价。
3. 加权投票反映专业性：不同领域的问题应该由不同专业背景的裁判评估——领域相关度和历史准确率提供了合理的权重分配机制。
4. Alpha 和 Kappa 量化共识：你不再需要凭感觉判断「裁判们意见一致吗」——有两个精确的数字告诉你。
5. 不可调和分歧是信号，不是失败：当裁判们无法达成共识时，系统不应该强行输出答案。诚实地告诉决策者「这个问题的争议很大，需要更多信息」比制造虚假共识更有价值。

📎 系列说明：本文是多 Agent 辩论系列的第 3 篇。建议按顺序阅读：L1：对抗协作入门 → L2：结构化辩论协议 → 本文 L3。下一篇 L4 探讨生产环境部署和实际应用。

📖 下一篇：多 Agent 辩论系统的生产部署 — 实际应用场景、系统架构、性能优化