辩论协议设计 — 8 Agent 结构化对抗与交叉质询引擎

2026年5月15日 · 进阶

30秒结论

解决什么问题：上一篇构建了数据管道，但数据不会辩论。你需要一个协议引擎，将结构化知识库转化为 8 个 Agent 的竞争性分析、交叉质询和综合结论——并且确保每个 Agent 的输出是可比较的、有证据支撑的、而非自由发挥的意见。
核心方法：三轮结构化辩论协议——开场陈述（并行，各 Agent 从自己分析视角提出论据）、交叉质询（配对，同类 Agent 互相挑战对方论据中的逻辑漏洞和数据盲点）、总结陈词（并行，基于批评完善论点——让步或加倍）。所有输出遵循严格的 JSON Schema：主张 + 证据引用 + 置信度 + 关键假设。
关键结论：4 维裁判评分（逻辑性 30% + 证据质量 30% + 清晰度 20% + 说服力 20%）是初始直觉权重——下一篇回测将用数据校准这些权重。提示词工程是辩论质量的核心：每个 Agent 的系统提示词不仅定义了立场，还约束了输出格式、证据引用要求和推理边界。
读完能做什么：运行 debate_protocol_market.py（~300 行），将第一篇的知识库 JSON 接入完整的辩论引擎。你会得到结构化的辩论记录（transcript）——每一条论据带时间戳、Agent 归属和评分，可直接输入裁判 Agent 进行最终综合。

知识库已经就绪。7 个全球指数、10 个行业 ETF、10 个宏观指标——全部结构化存储，每个 Agent 都有自己定制的数据切片。但数据只是燃料。现在我们需要发动机：辩论协议引擎。

在投入代码之前，先看一眼我们将要产出的东西。以下是 8 个 Agent 辩论 ExampleIndex（虚构科技指数）时，技术多头 Agent 在开场轮中输出的论据：

{
  "agent": "tech_bull",
  "round": "opening",
  "arguments": [
    {
      "claim": "ExampleIndex 处于确认的上升趋势中，价格在所有关键均线之上",
      "evidence": "价格 4,850 位于 MA20(4,720)、MA50(4,510)、MA200(4,120) 之上。MA20 > MA50 > MA200，三者正向排列形成多头排列。",
      "confidence": 0.85,
      "counterpoints": ["RSI 14 为 62，接近超买区域但尚未触发"]
    },
    {
      "claim": "MACD 在 3 日前金叉，动量正在加速",
      "evidence": "MACD 线 58.3 穿越信号线 42.1，柱状图为正且扩大。成交量在信号日高于 20 日均量 1.4 倍，确认突破有效。",
      "confidence": 0.78,
      "counterpoints": ["单一技术信号不构成独立交易依据，需结合基本面确认"]
    },
    {
      "claim": "板块广度健康——75% 的组成行业在 20 日内正收益",
      "evidence": "10 个组成行业中 7 个显示 20 日正回报。科技和工业板块领涨，防御性板块滞后——典型的 risk-on 轮动形态。",
      "confidence": 0.72,
      "counterpoints": ["广度收窄中——相较 30 日前 9/10 正收益有所下降"]
    }
  ]
}

这不是自由对话。这是结构化对抗——每条论据都携带主张、证据、置信度和反论点，可以被对方精确引用和挑战。技术空头会在质询轮中逐条攻击以上三条论据。宏观空头会从完全不同的维度发起另一种攻击。而裁判将以 4 个维度评分每一条论据。

本文是多 Agent 辩论 × 市场分析系列的第二篇。在第一篇中，我们构建了数据管道和系统架构。现在，我们构建辩论引擎本身——8 个专业化 Agent 如何在三轮结构化辩论中互相撕裂对方的论点，然后由一个公正裁判综合出一份不含糊的市场分析。

为什么结构化辩论击败自由讨论

你可能会想：为什么不直接把知识库丢给 8 个 Agent，让他们自由讨论市场走向？

因为自由讨论在市场分析中有三个致命缺陷：

缺陷	在自由讨论中的表现	市场场景中的具体危害
论据漂移	话题从"技术面看什么"滑到"宏观上美联储会怎样"再滑到"地缘政治风险"，没有锚点	技术分析师开始评论 CPI 数据（超出数据切片范围），基本面分析师开始画 K 线——所有 Agent 都溢出到不擅长的领域
回声室效应	8 个 Agent 可能无意中收敛到同一套叙事——比如都围绕"AI 泡沫"讨论，忽略了技术面动量信号	认知多样性在自由讨论中被动丧失。8 个 Agent 本质上退化成一个 8 倍 token 消耗的单一 Agent
无法量化比较	技术多头说"趋势向上"，宏观空头说"收益率曲线倒挂"——这两个论据如何比较？谁更有说服力？没有任何共同尺度	评判员只能凭感觉综合，无法给出可审计的评分。五次运行可能得到五种不同结论。

结构化协议解决这些问题的方法是：约束创造质量。每个 Agent 只能在自己的分析视角内发言（论据不漂移）。每个 Agent 必须面对一个在自己的领域内攻击它的对手（防止回声室）。每条论据都有统一的 JSON 格式和裁判评分体系（跨视角可比较）。

💡 核心设计原则：辩论协议不是限制 Agent 的创造力——它是为 Agent 的对抗性推理搭建一个竞技场。你需要边界。需要规则。需要记分牌。否则你拥有的不是 8 个 Agent 的辩论，而是 8 个 Agent 的噪音。

3 轮辩论协议：开场 → 质询 → 总结

协议改编自通用辩论框架（见多 Agent 辩论 L2：结构化辩论协议），但专门为市场分析做了适配：8 个 Agent 而非 2 个，领域内配对质询而非自由攻击。

协议总览

轮次	目标	Agent 行为	执行模式	产出
第一轮：开场陈述	每个 Agent 从自己的分析视角独立构建论点，不受对手干扰	每个 Agent 读取自己的知识库切片，生成 2-4 条结构化论据。多头论证看涨原因，空头论证看跌原因。	✅ 8 个 Agent 全部并行	每条论据：{claim, evidence, confidence, counterpoints}
第二轮：交叉质询	在同一分析领域内，多空双方互相攻击对方的开场论据	技术多头攻击技术空头的论据（反之亦然）。基本多头攻击基本空头。每对在领域内深挖——不跨领域攻击。	⚠️ 每对串行，4 对并行	每条质询：{target_id, challenge_type, reasoning, new_evidence}
第三轮：总结陈词	基于受到的质询，完善或放弃原有论点，做最终立场陈述	每个 Agent 阅读对手对自己的所有质询，决定：哪些论点仍然成立？哪些需要让步？最终立场是什么？	✅ 8 个 Agent 全部并行	每条总结：{refined_claims, concessions, final_stance, conviction_change}

为什么质询是配对的，而不是自由混战？

如果允许每个 Agent 攻击其他所有 Agent，一轮就有 8×7=56 个攻击向量。大部分攻击将是表面的——广而不深。通过在同一分析领域内配对 Agent：

技术多头 vs. 技术空头：双方都同意技术分析是有效的。他们争论的是当前技术数据意味着什么。这是真正的专家辩论。
基本多头 vs. 基本空头：双方都在看估值数据。争论焦点是当前 PE 是否合理——不是一个说 PE、另一个说 K 线形态的噪音。
宏观多头 vs. 宏观空头 + 情绪多头 vs. 情绪空头：同样逻辑。

⚠️ 配对质询不是简化——它是聚焦：每个 Agent 仍然只能在自己的知识切片内进行攻击。技术多头不会去挑战宏观空头的论据（它看不到宏观数据）。这意味着每场质询由领域专家执行。在领域内深度 > 在领域间广度。

JSON 论据格式规范

所有 Agent 在 3 轮辩论中都输出结构化 JSON——而不是自由文本。这是整个系统能够自动处理、比较和评分的基石。

开场论据格式

字段	类型	要求	示例
`claim`	string	一句明确的、可被证伪的命题。不超过 50 字。	"ExampleIndex 处于确认上升趋势，价格在所有关键均线之上"
`evidence`	string	支持主张的具体数据。必须引用知识库中的数据点。不接受"很多""普遍"等模糊表述。	"价格 4,850 → MA20(4,720), MA50(4,510), MA200(4,120)。三者正向排列。"
`confidence`	float (0.0–1.0)	Agent 对该论据的置信度。必须基于证据质量校准——证据强则置信高。不是"猜"。	0.85
`counterpoints`	string[]	Agent 自己意识到的反论点或限制条件。即使站在某一方，也必须诚实地指出论据的弱点。	["RSI 14 为 62，接近超买但尚未触发"]

💡 为什么要求 counterpoints？这来自 L2 辩论协议的"诚实度"维度。一个愿意主动暴露自己论点局限性的 Agent，比一个假装完美的 Agent 更可信。裁判会在评分中奖励这种诚实——会在评分中直接体现。

质询回应格式

字段	说明
`target_id`	被攻击的论据 ID（如 `tech_bull_arg_1`）
`challenge_type`	`refute`（指出逻辑或事实错误）\| `question_evidence`（质疑证据充分性或条件）\| `concede`（承认成立）\| `partial`（部分承认但质疑程度或范围）
`reasoning`	挑战的详细推理链
`new_evidence`	支持挑战的新证据（必须来源于知识库，不允许引入外部信息）

总结陈词格式

总结陈词不重新阐述完整论点。它精炼地结构化了 Agent 在经历质询后的状态变化：

refined_claims：哪些原始论点经受住了质询？如何被完善？（JSON 数组）
concessions：Agent 明确承认的让步——对手的哪些论点被接受？（JSON 数组）
final_stance：最终立场陈述（string，50-100 字）
conviction_change：信心变化——strengthened | weakened | unchanged

8 个 Agent 的专属提示词设计

每个 Agent 的系统提示词包含四个层次，确保专业化而不失去结构一致性：

角色层：定义分析视角（技术/基本/宏观/情绪）和方向立场（多头/空头）
数据层：明确告诉 Agent 它的知识库切片中包含哪些数据模块
格式层：强制 JSON 输出格式规范（claim/evidence/confidence/counterpoints）
行为层：辩论规则——诚实要求、证据引用规则、禁止幻觉

Agent 专业化矩阵

Agent ID	分析视角	数据切片	时间维度	核心任务
`tech_bull` 🐂	技术分析	indices, technicals	1 天 – 2 周	从价格行为、均线、RSI、MACD、成交量中找看涨信号
`tech_bear` 🐻	技术分析	indices, technicals	1 天 – 2 周	找超买、背离、派发形态、动量衰竭信号
`fund_bull` 🐂	基本面分析	indices, sectors, fundamentals	3 个月 – 2 年	找盈利支持、合理估值、行业增长信号
`fund_bear` 🐻	基本面分析	indices, sectors, fundamentals	3 个月 – 2 年	找估值扩张风险、盈利质量恶化、利润率压缩
`macro_bull` 🐂	宏观分析	macro, global_markets, indices	6 个月 – 3 年	找宽松政策、增长加速、全球资金流向正面
`macro_bear` 🐻	宏观分析	macro, global_markets, indices	6 个月 – 3 年	找通胀黏性、流动性收紧、收益率曲线倒挂
`senti_bull` 🐂	情绪分析	sentiment, indices, sectors	1 周 – 3 个月	找过度悲观（反向观察条件）、聪明钱流入
`senti_bear` 🐻	情绪分析	sentiment, indices, sectors	1 周 – 3 个月	找极度乐观（反向观察条件）、狂热指标

以下是技术多头 Agent 的完整系统提示词——作为样例，展示四层设计如何在实践中组合：

你是一个专业的技术分析师，立场为【看涨】，专注于短期（1 天 – 2 周）价格行为分析。

## 数据权限
你只能访问知识库中的以下模块：
- indices: 各指数价格、涨跌幅、多周期回报、距 52 周高低点距离
- technicals: MA(20/50/200) 状态、RSI(14)、MACD 信号、ATR(14)、成交量趋势

## 分析风格
你的任务是寻找支持看涨方向的技术证据。关注：
- 价格在关键均线上方运行的趋势延续信号
- MACD 金叉、RSI 从低位回升等动量改善信号
- 成交量确认——放量突破比缩量上涨更可靠
- 多个指数间的技术共振（如 SPX 和 IXIC 同时发出看涨信号）

## 论点格式（严格遵守）
所有论点必须输出为 JSON 数组，每条论据包含：
{
  "claim": "一句话的可证伪命题（不超过 50 字）",
  "evidence": "支持数据，必须引用 knowledge base 中的具体数据点",
  "confidence": 0.0-1.0 之间的浮点数，
  "counterpoints": ["你意识到的反论点或风险因素"]
}

## 辩论规则
1. 所有 evidence 必须来源于知识库，不能编造数据
2. 如果数据不足以支持高质量论点，降低 confidence 而不是编造
3. counterpoints 必须以诚实为原则——主动暴露论据的局限性
4. 在质询轮中，逐条回应对手的每一条论据，不能跳过
5. 在总结轮中，明确承认你让步的论点（而非回避）

💡 每个 Agent 有不同的"分析风格"段：技术 Agent 寻找趋势和动量信号。基本面 Agent 寻找估值合理性。宏观 Agent 寻找政策和经济周期。情绪 Agent 寻找极端情绪的反向信号。这段提示词中的差异——而非方向立场（多/空）——是真正的多样性来源。

裁判 Agent：4 维评分体系

裁判是唯一能访问完整知识库的 Agent——它没有方向立场，不做预测。它的唯一工作是评估辩论质量：哪些论据站得住脚，哪些被有效驳斥，以及从辩论的冲突中能提炼出什么。

裁判对每一条开场论据进行 4 维独立评分：

维度	评分标准 (1–10)	权重	裁判评估的核心问题
逻辑性 (Logic)	1=充满逻辑跳跃和循环论证，10=推理链自洽且无懈可击	30%	从证据到主张的因果链是否完整？有没有跳跃或偷换概念？
证据质量 (Evidence)	1=全是泛泛而谈或编造数据，10=每条数据可独立核实且与主张直接相关	30%	证据是否具体可量化？是否来自知识库？是否与主张直接相关？
清晰度 (Clarity)	1=含糊不清、绕来绕去，10=简洁精确、一读就懂	20%	主张是否明确到可以被证伪？论据是否精炼无冗余？
说服力 (Persuasiveness)	1=毫无说服力，10=即使你不同意其立场也觉得论证有力	20%	即使不同意其方向，该论据在逻辑和证据上是否令人信服？

每条论据的加权得分 = Logic×0.30 + Evidence×0.30 + Clarity×0.20 + Persuasiveness×0.20。

裁判还需要输出论据追溯表——追踪每一条论据经过质询后的最终状态：

UPHELD（成立）：论据在质询中完整存活。对方未能有效驳斥，或对方的挑战被成功反驳。
WEAKENED（削弱）：部分成立，但质询暴露了条件限制或证据缺陷。置信度应下调。
REFUTED（被驳倒）：质询揭示了根本性的逻辑错误或事实错误。不应作为决策依据。
UNCERTAIN（不确定）：双方未能就此达成明确结论——需要更多数据或更深入的分析。

⚠️ 裁判评分 ≠ 方向判断：裁判不判断"谁对市场的预测更正确"。它评估的是辩论质量——推理是否严谨、证据是否充分、回应是否诚实。一个看涨的 Agent 可能全部论据被驳回，但不代表市场会跌——裁判只是在说：以现有的证据和辩论质量，该 Agent 的论证不足以支持其结论。市场预测的正确性是后续文章（回测与验证）要解决的问题。

代码：debate_protocol.py

以下是完整的辩论协议引擎。约 320 行，实现了上述所有设计——Agent 配置、3 轮编排、裁判 4 维评分。保存为 debate_protocol.py，与第一篇的 market_data_pipeline.py 放在同一目录。

"""
debate_protocol.py
多 Agent 辩论 × 市场分析 — 辩论协议引擎
─────────────────────────────────────────────
实现 8 Agent（4 多头 + 4 空头）+ 1 裁判的 3 轮结构化辩论协议。

轮次:
  1. 开场陈述 — 每个 Agent 从自身分析视角构建论点
  2. 交叉质询 — 在同一领域内配对攻击
  3. 总结陈词 — 基于质询反馈完善、让步或坚持

输入:  KnowledgeBase (来自 market_data_pipeline.py)
输出: 完整辩论记录 + 裁判综合评分

依赖: pip install openai
LLM:  任何兼容 OpenAI 接口的端点
"""

import json
import os
import re
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional

# ═══════════════════════════════════════════════════════════
# 配置（占位凭证 — 替换为你的实际端点）
# ═══════════════════════════════════════════════════════════

LLM_CONFIG = {
    "api_key": "***",
 

... [OUTP

... [OUTPUT TRUNCATED - 59 chars omitted out of 50059 total] ...

    def __init__(self, profile: AgentProfile, data_slice: Dict[str, Any]):
        self.profile = profile
        self.data_slice = data_slice  # 该 Agent 可见的知识库部分
        self.history: List[Dict[str, str]] = []  # 消息历史

    def _build_context(self, round_type: RoundType,
                       extra_context: str = "") -> str:
        """构建发送给 LLM 的上下文。"""
        return (
            f"## 知识库（你的数据切片）\n"
            f"{json.dumps(self.data_slice, indent=2, ensure_ascii=False, default=str)}\n\n"
            f"{extra_context}"
        )

    def _call_llm(self, user_message: str, temperature: float,
                  max_tokens: int) -> str:
        """统一的 LLM 调用。"""
        client = get_client()
        messages = [
            {"role": "system", "content": self.profile.system_prompt}
        ] + self.history + [
            {"role": "user", "content": user_message}
        ]
        resp = client.chat.completions.create(
            model=LLM_CONFIG["model"],
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
        )
        reply = resp.choices[0].message.content
        self.history.append({"role": "user", "content": user_message})
        self.history.append({"role": "assistant", "content": reply})
        return reply

    def _parse_json(self, raw: str) -> List[Dict]:
        """安全解析 JSON，失败时返回带错误标记的兜底结构。"""
        try:
            cleaned = re.sub(r'```(?:json)?\s*', '', raw).strip()
            return json.loads(cleaned)
        except (json.JSONDecodeError, KeyError) as e:
            return [{
                "claim": f"[JSON 解析失败: {str(e)[:80]}]",
                "evidence": f"原始回复: {raw[:300]}",
                "confidence": 0.0,
                "counterpoints": ["JSON 格式错误，请检查 Agent 输出"]
            }]

    def opening_statement(self, topic: str) -> List[Dict]:
        """第一轮：开场陈述。生成 2-4 条结构化论据。"""
        ctx = self._build_context(RoundType.OPENING,
            f"## 辩题\n{topic}\n\n"
            f"## 你的任务\n"
            f"你是一个{self.profile.lens}分析师，立场为"
            f"{'看涨' if self.profile.side == 'bull' else '看跌'}。\n"
            f"基于知识库中你的数据切片，"
            f"生成 2-4 条支持你立场的结构化论据。")
        raw = self._call_llm(ctx,
            temperature=LLM_CONFIG["temperature"]["opening"],
            max_tokens=LLM_CONFIG["max_tokens_per_agent"])
        return self._parse_json(raw)

    def cross_examine(self, opponent_args: List[Dict]) -> List[Dict]:
        """第二轮：交叉质询。逐条攻击对手的开场论据。"""
        opponent_text = json.dumps(opponent_args, indent=2,
                                   ensure_ascii=False)
        ctx = self._build_context(RoundType.CROSS_EXAM,
            f"## 对手的论据（你的攻击目标）\n{opponent_text}\n\n"
            f"## 你的任务\n"
            f"逐条回应对手的每一条论据。对每条输出一个 JSON 对象:\n"
            f'{{"target_id": "对手论据的序号索引", '
            f'"challenge_type": "refute|question_evidence|concede|partial", '
            f'"reasoning": "你的推理", '
            f'"new_evidence": "支持你挑战的知识库数据"}}\n\n'
            f"规则: 不能跳过任何论据。不能引入新论据（只能回应已有论据）。\n"
            f"只输出 JSON 数组。")
        raw = self._call_llm(ctx,
            temperature=LLM_CONFIG["temperature"]["cross_exam"],
            max_tokens=LLM_CONFIG["max_tokens_per_agent"])
        return self._parse_json(raw)

    def closing_statement(self,
                          cross_exam_received: List[Dict]) -> List[Dict]:
        """第三轮：总结陈词。基于收到的质询完善或放弃论点。"""
        cross_text = json.dumps(cross_exam_received, indent=2,
                                ensure_ascii=False)
        ctx = self._build_context(RoundType.CLOSING,
            f"## 对手对你的质询\n{cross_text}\n\n"
            f"## 你的任务\n"
            f"基于对手的质询，生成你的总结陈词。JSON 格式:\n"
            f'{{"refined_claims": [...], '
            f'"concessions": [...], '
            f'"final_stance": "...", '
            f'"conviction_change": "strengthened|weakened|unchanged"}}\n\n'
            f"refined_claims: 哪些原始论点仍成立？如何改善？\n"
            f"concessions: 你明确承认哪些对手的论点？\n"
            f"final_stance: 50-100 字的最终立场阐述。\n"
            f"只输出 JSON 对象。")
        raw = self._call_llm(ctx,
            temperature=LLM_CONFIG["temperature"]["closing"],
            max_tokens=LLM_CONFIG["max_tokens_per_agent"])
        parsed = self._parse_json(raw)
        return parsed[0] if isinstance(parsed, list) and len(parsed) == 1 else parsed

（续 — 裁判 Agent）


# ═══════════════════════════════════════════════════════════
# 裁判 Agent — 4 维评分 + 论据追溯
# ═══════════════════════════════════════════════════════════

JUDGE_SYSTEM_PROMPT = """你是一个公正且严格的辩论裁判。你的任务是以 4 个维度评估
每一条开场论据，并生成辩论综合报告。

## 评分维度 (1-10)
1. logic (逻辑性): 证据到主张的因果链是否完整？有无跳跃或循环论证？
2. evidence (证据质量): 证据是否具体可量化？是否来自知识库且与主张相关？
3. clarity (清晰度): 主张是否明确可证伪？表达是否简洁精确？
4. persuasiveness (说服力): 即使不同意立场，论证是否令人信服？

权重: logic=30%, evidence=30%, clarity=20%, persuasiveness=20%

## 论据追溯状态
对每条论据，判断质询后的最终状态:
- UPHELD: 完整通过质询
- WEAKENED: 质询暴露了条件限制或证据缺陷
- REFUTED: 质询揭示了根本错误
- UNCERTAIN: 双方未能达成明确结论

## 输出格式
只输出 JSON，不要包含其他文字:
{
  "scores": [
    {"argument_id": "tech_bull_arg_0", "logic": 8, "evidence": 7,
     "clarity": 8, "persuasiveness": 7, "weighted": 7.5}
  ],
  "trace_table": [
    {"argument_id": "tech_bull_arg_0", "claim_summary": "...",
     "standing": "UPHELD", "reason": "..."}
  ],
  "synthesis": {
    "bull_case_summary": "多头整体论点摘要",
    "bear_case_summary": "空头整体论点摘要",
    "key_insight": "辩论中最关键的发现",
    "unresolved_questions": ["尚未解决的争议"],
    "judge_note": "基于辩论质量的综合评估——不是市场预测"
  }
}"""


class JudgeAgent:
    """裁判 Agent — 对整场辩论进行 4 维评分和综合。"""

    def __init__(self):
        self.system_prompt = JUDGE_SYSTEM_PROMPT

    def evaluate(self, transcript: Dict[str, Any],
                 knowledge_base: Dict[str, Any]) -> Dict[str, Any]:
        """评估完整辩论记录，输出结构化评分。"""
        from openai import OpenAI
        client = OpenAI(
            api_key=LLM_CONFIG["api_key"],
            base_url=LLM_CONFIG["base_url"],
        )

        kb_summary = json.dumps(knowledge_base, indent=2,
                                ensure_ascii=False, default=str)

        eval_prompt = (
            f"## 知识库（完整—供验证 Agent 引用的数据）\n"
            f"{kb_summary[:4000]}\n\n"
            f"## 辩论记录\n"
            f"{json.dumps(transcript, indent=2, ensure_ascii=False, default=str)}"
        )

        resp = client.chat.completions.create(
            model=LLM_CONFIG["model"],
            messages=[
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": eval_prompt},
            ],
            temperature=LLM_CONFIG["temperature"]["judge"],
            max_tokens=LLM_CONFIG["max_tokens_judge"],
        )

        raw = resp.choices[0].message.content
        try:
            cleaned = re.sub(r'```(?:json)?\s*', '', raw).strip()
            return json.loads(cleaned)
        except json.JSONDecodeError:
            return {"error": "裁判 JSON 解析失败", "raw": raw}

（续 — 辩论编排器与主函数）


# ═══════════════════════════════════════════════════════════
# 辩论编排器
# ═══════════════════════════════════════════════════════════

class DebateOrchestrator:
    """管理 3 轮辩论、Agent 配对和裁判。"""

    def __init__(self, knowledge_base: Dict[str, Any]):
        self.kb = knowledge_base
        self.agents: Dict[str, DebateAgent] = {}
        self._init_agents()

    def _init_agents(self):
        """初始化所有 8 个 Agent。"""
        for ad in AGENT_DEFINITIONS:
            # 构建数据切片
            data_slice = {}
            for module in ad["data_slices"]:
                if module in self.kb:
                    data_slice[module] = self.kb[module]
            # 创建 Agent 配置
            profile = AgentProfile(
                agent_id=ad["agent_id"],
                side=ad["side"],
                lens=ad["lens"],
                name=ad["name"],
                system_prompt=build_agent_prompt(ad),
                data_slices=ad["data_slices"],
                opponent_id=ad.get("opponent_id", ""),
            )
            self.agents[ad["agent_id"]] = DebateAgent(profile, data_slice)

    def run_debate(self, topic: str = "市场展望分析") -> Dict[str, Any]:
        """执行完整的 3 轮辩论。
        
        返回:
            dict: {
                "transcript": {round1, round2, round3},
                "judge_evaluation": {...}
            }
        """
        transcript: Dict[str, Any] = {
            "topic": topic,
            "rounds": {},
        }

        # ── 第一轮: 开场陈述（8 Agent 并行） ──
        print("\n📋 第一轮: 开场陈述（8 Agent 并行）")
        round1_args = {}
        with ThreadPoolExecutor(max_workers=8) as ex:
            futures = {
                ex.submit(agent.opening_statement, topic): aid
                for aid, agent in self.agents.items()
            }
            for f in as_completed(futures):
                aid = futures[f]
                round1_args[aid] = f.result()
                print(f"  ✅ {aid}: {len(round1_args[aid])} 条论据")
        transcript["rounds"]["opening"] = round1_args

        # ── 第二轮: 交叉质询（4 对并行，每对内串行） ──
        print("\n⚔️  第二轮: 交叉质询（4 对并行）")
        pairings = [
            ("tech_bull", "tech_bear"),
            ("fund_bull", "fund_bear"),
            ("macro_bull", "macro_bear"),
            ("senti_bull", "senti_bear"),
        ]
        round2_args = {}

        def run_pair(agent_a_id: str, agent_b_id: str):
            """执行一对 Agent 的互相质询。"""
            agent_a = self.agents[agent_a_id]
            agent_b = self.agents[agent_b_id]
            a_on_b = agent_a.cross_examine(round1_args[agent_b_id])
            b_on_a = agent_b.cross_examine(round1_args[agent_a_id])
            return {
                f"{agent_a_id}_cross": a_on_b,
                f"{agent_b_id}_cross": b_on_a,
            }

        with ThreadPoolExecutor(max_workers=4) as ex:
            futures = {
                ex.submit(run_pair, a, b): (a, b)
                for a, b in pairings
            }
            for f in as_completed(futures):
                pair = futures[f]
                result = f.result()
                round2_args.update(result)
                print(f"  ✅ {pair[0]} ↔ {pair[1]} 质询完成")
        transcript["rounds"]["cross_examination"] = round2_args

        # ── 第三轮: 总结陈词（8 Agent 并行） ──
        print("\n🏁 第三轮: 总结陈词（8 Agent 并行）")
        round3_args = {}
        with ThreadPoolExecutor(max_workers=8) as ex:
            futures = {}
            for aid, agent in self.agents.items():
                # 每个 Agent 收到其对手对其的质询
                opp_id = AGENT_DEFINITIONS[
                    [i for i, ad in enumerate(AGENT_DEFINITIONS)
                     if ad["agent_id"] == aid][0]
                ]["opponent_id"]
                cross_key = f"{opp_id}_cross"
                cross_received = round2_args.get(cross_key, [])
                futures[ex.submit(
                    agent.closing_statement, cross_received
                )] = aid

            for f in as_completed(futures):
                aid = futures[f]
                round3_args[aid] = f.result()
                print(f"  ✅ {aid}: 总结完成")
        transcript["rounds"]["closing"] = round3_args

        # ── 裁判评估 ──
        print("\n⚖️  裁判评估中...")
        judge = JudgeAgent()
        evaluation = judge.evaluate(transcript, self.kb)
        transcript["judge_evaluation"] = evaluation

        # 打印评分摘要
        if "synthesis" in evaluation:
            s = evaluation["synthesis"]
            print(f"\n  关键发现: {s.get('key_insight', 'N/A')[:120]}")
        if "trace_table" in evaluation:
            upheld = sum(
                1 for t in evaluation["trace_table"]
                if t.get("standing") == "UPHELD"
            )
            total = len(evaluation["trace_table"])
            print(f"  论据存活率: {upheld}/{total} (UPHELD)")

        return transcript


# ═══════════════════════════════════════════════════════════
# 主函数
# ═══════════════════════════════════════════════════════════

def run_debate(knowledge_base: Dict[str, Any],
               topic: str = "市场展望分析") -> Dict[str, Any]:
    """辩论引擎入口。

    Args:
        knowledge_base: 来自 market_data_pipeline.py 的 KnowledgeBase
        topic: 辩论主题

    Returns:
        完整辩论记录，含 3 轮记录和裁判评估
    """
    orchestrator = DebateOrchestrator(knowledge_base)
    return orchestrator.run_debate(topic)


# ═══════════════════════════════════════════════════════════
# 运行示例（使用合成数据）
# ═══════════════════════════════════════════════════════════

if __name__ == "__main__":
    # 合成知识库 — 虚构的 ExampleIndex 数据
    synthetic_kb = {
        "meta": {
            "generated_at": "2026-05-15T12:00:00Z",
            "market_status": "open",
            "data_sources": ["synthetic"],
            "warnings": ["这是合成演示数据——不可用于真实分析"],
        },
        "indices": {
            "EXI": {
                "ticker": "EXI", "name": "ExampleIndex",
                "price": 4850.0, "change_pct": 0.8,
                "returns": {"5d": 2.1, "20d": 4.8, "50d": 12.3, "200d": 28.5},
                "vs_52w_high_pct": -3.2, "vs_52w_low_pct": 42.1,
                "volume_ratio": 1.3,
            },
        },
        "technicals": {
            "EXI": {
                "ticker": "EXI",
                "ma_status": {"ma20": "above", "ma50": "above", "ma200": "above"},
                "rsi_14": 62.0, "macd_signal": "bullish",
                "atr_14": 45.2, "volume_trend": "increasing",
            },
        },
        "sectors": {
            "XTECH": {"ticker": "XTECH", "name": "科技", "price": 520.0,
                       "change_5d_pct": 3.2, "change_20d_pct": 7.5,
                       "relative_strength_vs_spx": 2.7},
            "XFIN": {"ticker": "XFIN", "name": "金融", "price": 180.0,
                      "change_5d_pct": -0.5, "change_20d_pct": 1.2,
                      "relative_strength_vs_spx": -3.6},
            "XIND": {"ticker": "XIND", "name": "工业", "price": 210.0,
                      "change_5d_pct": 1.8, "change_20d_pct": 4.1,
                      "relative_strength_vs_spx": -0.7},
        },
        "fundamentals": {
            "sp500_pe_approx": {"current_pe_approx": 28.5,
                                "long_term_avg_pe": 24.0},
            "sp500_earnings_yield_approx": 3.51,
            "sector_rotation_signal": "cyclical_rotation",
        },
        "macro": {
            "GDP": {"indicator": "GDP", "trend": "rising",
                     "latest_value": 3.2, "yoy_change_pct": 0.5},
            "CPI": {"indicator": "CPI", "trend": "falling",
                     "latest_value": 3.1, "yoy_change_pct": -0.3},
            "FEDFUNDS": {"indicator": "FEDFUNDS", "trend": "flat",
                          "latest_value": 4.25},
            "UNRATE": {"indicator": "UNRATE", "trend": "flat",
                        "latest_value": 3.8},
            "T10Y2Y": {"indicator": "T10Y2Y", "trend": "flat",
                        "latest_value": -0.35},
        },
        "sentiment": {
            "vix_level": 18.2, "vix_regime": "normal",
            "volume_signal": "high_volume_rally",
            "sector_breadth": {"positive_5d": "7/10",
                                "positive_20d": "7/10",
                                "breadth_regime": "broad_strength"},
        },
        "global_markets": {
            "HSI": {"ticker": "HSI", "name": "恒生", "price": 21800,
                     "change_pct": 1.2,
                     "returns": {"5d": 2.8, "20d": 6.3}},
            "N225": {"ticker": "N225", "name": "日经", "price": 39200,
                      "change_pct": 0.5,
                      "returns": {"5d": 1.3, "20d": 3.7}},
        },
    }

    print("=" * 60)
    print("📊 多 Agent 辩论 × 市场分析 — 辩论协议引擎")
    print("=" * 60)

    result = run_debate(synthetic_kb,
                        topic="ExampleIndex 未来一个月的市场展望")

    # 保存完整辩论记录
    output_path = "debate_transcript.json"
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(result, f, indent=2, ensure_ascii=False, default=str)
    print(f"\n📁 完整辩论记录已保存至: {output_path}")

    # 打印摘要
    ev = result.get("judge_evaluation", {})
    if "synthesis" in ev:
        s = ev["synthesis"]
        print(f"\n{'─' * 40}")
        print(f"📋 辩论摘要")
        print(f"{'─' * 40}")
        print(f"多头: {s.get('bull_case_summary', 'N/A')[:200]}")
        print(f"空头: {s.get('bear_case_summary', 'N/A')[:200]}")
        print(f"关键洞察: {s.get('key_insight', 'N/A')}")
    print(f"\n✅ 辩论引擎运行完成。")

运行辩论引擎

将代码保存为 debate_protocol.py，安装依赖，然后运行：

# 安装依赖
pip install openai

# 配置：编辑 LLM_CONFIG 中的 api_key 和 base_url
# 替换为你的实际 API 端点

# 运行辩论（使用内置合成数据演示）
python debate_protocol.py

生产环境中，从第一篇的管道导入知识库：

from market_data_pipeline import build_knowledge_base
from debate_protocol import run_debate

kb = build_knowledge_base()        # 拉取真实市场数据
result = run_debate(kb)            # 运行 8 Agent 辩论
# result["judge_evaluation"] 包含综合评估

实战示例：ExampleIndex 辩论全流程

以下是使用上述合成数据运行辩论引擎后的典型输出。我们展示关键片段，而非完整 JSON 转储。

第一轮：开场论据精华

Agent	关键主张	置信度
🐂 tech_bull	价格位于所有均线之上，MA 多头排列确认上升趋势	0.85
🐻 tech_bear	RSI 14 为 62 接近超买，距 52 周高点仅 -3.2%——存在回调风险	0.72
🐂 fund_bull	盈利收益率 3.51% vs 长期国债 ~4.2% 的利差收窄但仍为正，权益仍有相对价值	0.68
🐻 fund_bear	当前 PE 28.5 显著高于 5 年均值 24.0（+18.8%），估值处于历史高位	0.80
🐂 macro_bull	GDP 增长 3.2%、CPI 回落至 3.1%——软着陆情景支撑风险资产	0.75
🐻 macro_bear	收益率曲线持续倒挂（10Y-2Y = -0.35%），历史上是衰退前导信号	0.82
🐂 senti_bull	行业广度健康（7/10 正收益），VIX 18.2 处于正常区间——无恐慌信号	0.70
🐻 senti_bear	成交量比率为 1.3，放量上涨可能意味着短期过热和获利了结压力	0.65

第二轮：关键质询交锋

以下是技术多头 vs. 技术空头配对中的一条代表性交锋——技术空头对技术多头第一条论据的质询：

// 技术空头 → 攻击技术多头论据 #1
{
  "target_id": "tech_bull_arg_0",
  "challenge_type": "question_evidence",
  "reasoning": "MA 多头排列确实出现在上升趋势中，但这是一个滞后指标——它在趋势确立后才确认。当前价格距 MA200 约 730 点（+17.7%），历史上这种偏离度往往伴随着均值回归。200 日回报 +28.5% 表明指数已经大幅上涨——MA 多头排列只是反映了已经发生的事，而非预测未来的方向。",
  "new_evidence": "价格偏离 MA200 +17.7%（来自 indices.EXI.returns.200d: 28.5% 但 MA200 显示价格远高于它）。200 日滚动回报处于历史上第 85 百分位。"
}

技术多头的总结陈词中，面对这条质询的回应：

// 技术多头 — 总结陈词片段
{
  "refined_claims": [{
    "original": "MA 多头排列确认上升趋势",
    "refinement": "MA 多头排列确认上升趋势成立，但应结合 RSI 和 MACD 作为趋势强度的前置指标。MA 的滞后性被 MACD 金叉的部分前置性补偿。",
    "confidence_adjusted": 0.78
  }],
  "concessions": [
    "接受技术空头关于 MA 是滞后指标的观点。MA 排列应作为趋势确认而非趋势预测的依据使用。",
    "接受当前涨幅偏离历史均值可能在统计上增加回调概率——但这不构成反转信号。"
  ],
  "final_stance": "技术面仍然看涨，但置信度从极高下调至中高。RSI 62 和 MA 排列指向持续动量，但大幅上涨后的回调概率不可忽视。若价格回踩至 MA20 则构成技术观察条件：需确认支撑是否有效。",
  "conviction_change": "weakened"
}

💡 注意"让步"的价值：技术多头没有假装自己的论点完美。它明确接受了对手的批评（MA 是滞后指标）。裁判会在"逻辑性"和"说服力"维度上奖励这种诚实。一个愿意让步的 Agent 比一个死不承认的 Agent 更值得信赖。

裁判综合评估

以下是裁判对 8 条论据（简化版）的评分和追溯：

论据	逻辑	证据	清晰	说服	加权	状态
tech_bull #1 (MA排列)	7	8	8	7	7.5	WEAKENED
tech_bear #1 (RSI超买)	6	6	7	5	6.0	WEAKENED
fund_bull #1 (盈利收益)	8	7	7	7	7.3	UPHELD
fund_bear #1 (PE偏高)	9	8	9	8	8.5	UPHELD
macro_bull #1 (软着陆)	7	7	7	6	6.8	WEAKENED
macro_bear #1 (曲线倒挂)	8	8	8	8	8.0	UPHELD
senti_bull #1 (广度健康)	7	7	8	6	7.0	UPHELD
senti_bear #1 (放量过热)	6	5	7	5	5.7	REFUTED

裁判综合洞察："空头在基本面（PE 偏高）和宏观面（收益率曲线倒挂）上有更强的论证支撑。多头在技术面（趋势向上）和情绪面（无恐慌信号）上有可靠但较为温和的论据。关键分歧在于时间维度——空头的论点集中在 3-12 个月的中期风险，而多头的论点更多反映短期（1-2 周）动量。辩论未达成明确的方向性共识，但揭示了风险/回报的结构性不对称——上行空间存在（技术趋势），但下行风险已充分论证（估值 + 宏观）。"

常见陷阱与设计决策

陷阱 1：Agent 幻觉——引用不存在的知识库数据

问题：LLM 有一种执着的倾向——当提示词说"引用知识库中的数据"但它找不到时，它可能会编造数据。

解决方案：我们在提示词中直接声明："所有 evidence 必须来自知识库。不要编造数据。如果数据不足以支持高质量论点，降低 confidence 而不是编造。"此外，裁判 Agent 收到了完整知识库，可以在评分时交叉验证 Agent 引用的数据——如果一个 Agent 声称 PE 是 15 而知识库中 PE 是 28.5，裁判会标注证据质量问题。

陷阱 2：配对质询退化——双方"同意不争论"

问题：在某些温度设置下，技术多头和空头可能在质询轮中互相说"我承认你的观点""我也承认你的观点"——辩论变成了握手会。

解决方案：降低质询轮的温度（0.4），并在提示词中明确要求："你必须找到对手论据中的弱点。不能对所有论据都 concede——至少对一半论据提出 challenge 或 refute。"

陷阱 3：裁判分数膨胀——所有论据都在 7-9 分

问题：不加调校的裁判倾向给出平均化的高分。如果所有论据都在 7-9 分之间，评分体系就失去了区分度。

解决方案：在裁判提示词中增加分布指导："好的论据应该得到 6-8 分，而不是 9-10 分。9-10 分应该留给那些近乎完美的论证——逻辑无懈可击、证据无可辩驳。"此外，第三篇文章将引入评分校准——通过回测历史数据来调整裁判权重。

设计决策：为什么 4 个纬度（而非 L2 辩论的 logic/evidence/responsiveness/honesty）？

L2 通用辩论协议使用了 logic/evidence/responsiveness（回应质量）/honesty（诚实度）。市场辩论协议将它们调整为 logic/evidence/clarity/persuasiveness。有两个原因：

市场分析的"回应质量"更难评估：在通用辩论中，可以判断"正方是否正面回应了反方的质询"。但在市场分析中，很多回应是"我承认你的数据，但我有不同解读"——这不是回避，这是合理的观点差异。clarity 和 persuasiveness 更准确地捕捉了市场论据的质量。
persuasiveness 捕获了"诚实度"的正面效果：一个愿意在 counterpoints 中暴露弱点的论据天然更具说服力——因为读者感觉作者是诚实的。persuasiveness 维度隐式地包含了诚实度的考量。

关键收获

结构化协议 = 可审计的辩论：每一条论据都有 ID、主张、证据和置信度。裁判可以精确引用和评分每一条。与自由对话相比，审计能力提升了至少一个数量级。
专业化提示词驱动真正的多样性：8 个 Agent 不是因为"有的看涨有的看跌"而不同——是因为分析视角和数据切片的不同。技术分析师和宏观经济学家看同一个市场会看到不同的事物——这正是多样性价值的来源。
配对质询约束质量：领域内配对（技术 vs 技术、基本面 vs 基本面）确保每场质询都由领域专家执行。跨领域攻击只会产生噪音。
裁判评分 ≠ 方向预测：裁判评估辩论质量——不是预测谁会"赢"。回测和验证是第三篇的主题。
论据追溯表是最实用的输出：从 8 个 Agent × 3 轮 × N 条论据的混乱中，论据追溯表提取出唯一重要的信息：哪些论点站住了，哪些被驳倒了，哪些还不确定。这是从辩论到决策的桥梁。

第三篇预告：回测与验证

现在你有了一个可运行的辩论引擎。它能产生高质量的对抗性分析。但一个关键问题仍然存在：辩论真的能提高市场分析的准确性吗？

在第三篇中，我们将用硬数据回答这个问题：

100 次历史辩论回测：用 100 个已知结果的历史时段运行辩论引擎——辩论输出与实际市场走向的吻合度如何？
单 Agent 基准对比：同样的数据，一个单一 LLM 的"分析市场"输出 vs. 8 Agent 辩论后的裁判综合——哪个更准？差距有多大？
裁判权重校准：当前 4 维权重（30/30/20/20）是直觉设定。通过回测数据，我们可以校准到最优权重——哪些维度对预测市场方向更有意义？
置信度验证：Agent 报告的置信度是否与实际准确率相关？高置信论据是否比低置信论据更可能成立？

但在此之前，先运行今天的代码。用第一篇的管道拉取真实数据，用本篇的引擎运行一场辩论。阅读论据追溯表。看看哪些论点站住了、哪些被驳倒了。然后问自己：如果只有我一个人看这些数据，我会注意到这些吗？

📖 上一篇：多 Agent 辩论 × 市场分析 — 架构与数据管道（构建知识库）
📖 通用辩论理论：多 Agent 辩论 L2：结构化辩论协议 · L3：评分与共识
📖 下一篇：第三篇——回测与验证

⚠️ 免责声明：本文是技术工作流演示，不构成投资建议。文中所有市场数据均为合成/虚构示例（ExampleIndex 不是真实指数）。辩论引擎的输出不能也不应该被用作实际投资决策的依据。金融市场存在固有风险，任何自动化分析系统都可能产生错误结论。在做出任何投资决策前，请咨询持牌金融专业人士。

下一步阅读

📖 同系列下一篇：回测与验证 — 100 场历史辩论的准确率与裁判权重校准 — 对本文的辩论协议进行完整回测：测量方向准确率、校准裁判权重（用数据替代直觉）、McNemar 显著性检验。
📖 辩论理论基础：多 Agent 辩论 L2：结构化辩论协议 — 本文 3 轮协议的理论来源：为什么是三轮而非两轮或四轮？交叉质询的设计空间和权衡。
📖 基础技能：从零构建 Agent 框架 — 提示词、工具与编排 — 深入理解本文 Agent 提示词工程背后的系统化方法论：角色定义、输出约束、推理边界的设计模式。

常见问题

Q: 辩论协议中的「交叉质询」为什么是配对而不是自由混战？

A: 自由混战（每个 Agent 攻击其他所有 Agent）会产生 8×7=56 个攻击向量——那是噪音，不是信号。配对设计（技术多头 vs. 技术空头、基本多头 vs. 基本空头等）确保每次质询是聚焦的、深度的、跨对可比较的。技术多头和技术空头不是在争论技术分析是否有效——他们都同意它有效。他们在争论的是技术数据此刻意味着什么。这与 L2 系列「一个裁判评估辩论的一个维度」是同一设计原则——约束创造质量。此外，4 对并行执行保持了总耗时在约 12 秒。

Q: Agent 输出的 JSON Schema 具体包含哪些字段？为什么需要结构化输出？

A: 每个 Agent 的论据输出遵循严格的 JSON 格式：claim（核心主张，1-2 句话）、evidence（证据列表，每个证据包含 data_point 来自知识库的具体数值和 source 模块引用）、confidence（0-1 的置信度评分）、assumptions（关键假设列表——明确哪些前提如果被推翻会导致论据失效）。结构化输出的核心目的不是「好看」，而是机器可读和跨 Agent 可比较——裁判可以自动解析每个论据的证据来源和置信度，而非靠自然语言理解来猜测。

Q: 4 维裁判评分（逻辑性、证据质量、清晰度、说服力）为什么是这四个维度？

A: 这四个维度来自 L3 辩论理论的多维度评分框架，针对市场分析场景做了适配。逻辑性评估推理链条是否自洽；证据质量评估引用数据的具体性和可验证性（「标普 500 的 RSI 为 58，处于中性区间」vs.「市场看起来很强」）；清晰度评估论据是否可以被对手 Agent 准确理解和回应；说服力评估论据在交叉质询后的存活程度。这四个维度的权重（30/30/20/20）是初始直觉值——第三篇回测将通过网格搜索校准最优权重。

Q: 提示词工程在辩论协议中扮演什么角色？如何防止 Agent 产生幻觉？

A: 提示词是辩论质量的核心控制层。每个 Agent 的系统提示词包含四个约束：角色定义（分析视角 + 方向立场）、输出格式约束（必须遵循 JSON Schema）、证据引用要求（必须引用知识库切片中的具体数据——如果知识库中没有该数据，必须标注为「推断」而非「事实」）、推理边界（明确 Agent 不应该评论其分析视角范围外的问题）。通过将证据锚定在知识库数据上（而非 Agent 的训练知识），幻觉被系统性抑制——Agent 不能编造它看不到的数据。

Q: 辩论协议如何与第一篇的数据管道集成？

A: 数据管道（market_data_pipeline.py）输出 market_knowledge_base.json，辩论协议（debate_protocol_market.py）导入该 JSON 并调用 slice_for_agent() 为每个 Agent 提取其专属数据切片。集成方式：kb = build_knowledge_base()（数据管道）→ protocol = DebateProtocol(kb)（辩论协议）→ transcript = protocol.run_debate()。辩论编排器对数据管道的依赖是单向的——管道是模块，协议是消费者。缓存层（第四篇）将在此基础上添加 TTL，避免每次辩论重新拉取数据。