知识库已经就绪。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，让他们自由讨论市场走向？
因为自由讨论在市场分析中有三个致命缺陷：
结构化协议解决这些问题的方法是：约束创造质量。每个 Agent 只能在自己的分析视角内发言（论据不漂移）。每个 Agent 必须面对一个在自己的领域内攻击它的对手（防止回声室）。每条论据都有统一的 JSON 格式和裁判评分体系（跨视角可比较）。
协议改编自通用辩论框架（见多 Agent 辩论 L2：结构化辩论协议），但专门为市场分析做了适配：8 个 Agent 而非 2 个，领域内配对质询而非自由攻击。
如果允许每个 Agent 攻击其他所有 Agent，一轮就有 8×7=56 个攻击向量。大部分攻击将是表面的——广而不深。通过在同一分析领域内配对 Agent：
所有 Agent 在 3 轮辩论中都输出结构化 JSON——而不是自由文本。这是整个系统能够自动处理、比较和评分的基石。
claim
evidence
confidence
counterpoints
target_id
tech_bull_arg_1
challenge_type
refute
question_evidence
concede
partial
reasoning
new_evidence
总结陈词不重新阐述完整论点。它精炼地结构化了 Agent 在经历质询后的状态变化：
refined_claims
concessions
final_stance
conviction_change
strengthened
weakened
unchanged
每个 Agent 的系统提示词包含四个层次，确保专业化而不失去结构一致性：
tech_bull
tech_bear
fund_bull
fund_bear
macro_bull
macro_bear
senti_bull
senti_bear
以下是技术多头 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——它没有方向立场，不做预测。它的唯一工作是评估辩论质量：哪些论据站得住脚，哪些被有效驳斥，以及从辩论的冲突中能提炼出什么。
裁判对每一条开场论据进行 4 维独立评分：
每条论据的加权得分 = Logic×0.30 + Evidence×0.30 + Clarity×0.20 + Persuasiveness×0.20。
裁判还需要输出论据追溯表——追踪每一条论据经过质询后的最终状态：
以下是完整的辩论协议引擎。约 320 行，实现了上述所有设计——Agent 配置、3 轮编排、裁判 4 维评分。保存为 debate_protocol.py，与 第一篇的 market_data_pipeline.py 放在同一目录。
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"] 包含综合评估
以下是使用上述合成数据运行辩论引擎后的典型输出。我们展示关键片段，而非完整 JSON 转储。
以下是技术多头 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" }
以下是裁判对 8 条论据（简化版）的评分和追溯：
裁判综合洞察："空头在基本面（PE 偏高）和宏观面（收益率曲线倒挂）上有更强的论证支撑。多头在技术面（趋势向上）和情绪面（无恐慌信号）上有可靠但较为温和的论据。关键分歧在于时间维度——空头的论点集中在 3-12 个月的中期风险，而多头的论点更多反映短期（1-2 周）动量。辩论未达成明确的方向性共识，但揭示了风险/回报的结构性不对称——上行空间存在（技术趋势），但下行风险已充分论证（估值 + 宏观）。"
问题：LLM 有一种执着的倾向——当提示词说"引用知识库中的数据"但它找不到时，它可能会编造数据。
解决方案：我们在提示词中直接声明："所有 evidence 必须来自知识库。不要编造数据。如果数据不足以支持高质量论点，降低 confidence 而不是编造。"此外，裁判 Agent 收到了完整知识库，可以在评分时交叉验证 Agent 引用的数据——如果一个 Agent 声称 PE 是 15 而知识库中 PE 是 28.5，裁判会标注证据质量问题。
问题：在某些温度设置下，技术多头和空头可能在质询轮中互相说"我承认你的观点""我也承认你的观点"——辩论变成了握手会。
解决方案：降低质询轮的温度（0.4），并在提示词中明确要求："你必须找到对手论据中的弱点。不能对所有论据都 concede——至少对一半论据提出 challenge 或 refute。"
问题：不加调校的裁判倾向给出平均化的高分。如果所有论据都在 7-9 分之间，评分体系就失去了区分度。
解决方案：在裁判提示词中增加分布指导："好的论据应该得到 6-8 分，而不是 9-10 分。9-10 分应该留给那些近乎完美的论证——逻辑无懈可击、证据无可辩驳。"此外，第三篇文章将引入评分校准——通过回测历史数据来调整裁判权重。
L2 通用辩论协议使用了 logic/evidence/responsiveness（回应质量）/honesty（诚实度）。市场辩论协议将它们调整为 logic/evidence/clarity/persuasiveness。有两个原因：
现在你有了一个可运行的辩论引擎。它能产生高质量的对抗性分析。但一个关键问题仍然存在：辩论真的能提高市场分析的准确性吗？
在第三篇中，我们将用硬数据回答这个问题：
但在此之前，先运行今天的代码。用第一篇的管道拉取真实数据，用本篇的引擎运行一场辩论。阅读论据追溯表。看看哪些论点站住了、哪些被驳倒了。然后问自己：如果只有我一个人看这些数据，我会注意到这些吗？
📖 上一篇：多 Agent 辩论 × 市场分析 — 架构与数据管道（构建知识库） 📖 通用辩论理论：多 Agent 辩论 L2：结构化辩论协议 · L3：评分与共识 📖 下一篇：第三篇——回测与验证
⚠️ 免责声明：本文是技术工作流演示，不构成投资建议。文中所有市场数据均为合成/虚构示例（ExampleIndex 不是真实指数）。辩论引擎的输出不能也不应该被用作实际投资决策的依据。金融市场存在固有风险，任何自动化分析系统都可能产生错误结论。在做出任何投资决策前，请咨询持牌金融专业人士。
A: 自由混战（每个 Agent 攻击其他所有 Agent）会产生 8×7=56 个攻击向量——那是噪音，不是信号。配对设计（技术多头 vs. 技术空头、基本多头 vs. 基本空头等）确保每次质询是聚焦的、深度的、跨对可比较的。技术多头和技术空头不是在争论技术分析是否有效——他们都同意它有效。他们在争论的是技术数据此刻意味着什么。这与 L2 系列「一个裁判评估辩论的一个维度」是同一设计原则——约束创造质量。此外，4 对并行执行保持了总耗时在约 12 秒。
A: 每个 Agent 的论据输出遵循严格的 JSON 格式：claim（核心主张，1-2 句话）、evidence（证据列表，每个证据包含 data_point 来自知识库的具体数值和 source 模块引用）、confidence（0-1 的置信度评分）、assumptions（关键假设列表——明确哪些前提如果被推翻会导致论据失效）。结构化输出的核心目的不是「好看」，而是机器可读和跨 Agent 可比较——裁判可以自动解析每个论据的证据来源和置信度，而非靠自然语言理解来猜测。
A: 这四个维度来自 L3 辩论理论的多维度评分框架，针对市场分析场景做了适配。逻辑性评估推理链条是否自洽；证据质量评估引用数据的具体性和可验证性（「标普 500 的 RSI 为 58，处于中性区间」vs.「市场看起来很强」）；清晰度评估论据是否可以被对手 Agent 准确理解和回应；说服力评估论据在交叉质询后的存活程度。这四个维度的权重（30/30/20/20）是初始直觉值——第三篇回测将通过网格搜索校准最优权重。
A: 提示词是辩论质量的核心控制层。每个 Agent 的系统提示词包含四个约束：角色定义（分析视角 + 方向立场）、输出格式约束（必须遵循 JSON Schema）、证据引用要求（必须引用知识库切片中的具体数据——如果知识库中没有该数据，必须标注为「推断」而非「事实」）、推理边界（明确 Agent 不应该评论其分析视角范围外的问题）。通过将证据锚定在知识库数据上（而非 Agent 的训练知识），幻觉被系统性抑制——Agent 不能编造它看不到的数据。
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，避免每次辩论重新拉取数据。