LLM Council也就是多模型评审团正在从实验玩法变成内容生产团队的真实选择。我实际跑过一个由 9 个模型组成的小型评审团用来每周生成一份财经资讯通讯研究员负责整理素材写稿模型负责成文事实核查和结构评审分别提出修改意见最后由汇总模型确认终稿。这套流程听起来像是把多个 LLM 组合在一起“开会”但跑起来之后真正需要花精力处理的不是某个模型写得不好而是整条流水线哪些环节会断以及断了之后如何快速恢复。这篇文章会围绕这个问题展开先拆解 9 个模型的职责分工再给出一个可以运行的最小实现然后重点分析我踩过的 5 类故障最后提供一套从日志倒推根因的排查链路和生产环境加固方案。本文适合已经会调用单个模型 API准备尝试多模型协作或多 Agent 流水线的开发者。不会涉及投资建议只讨论多模型生成内容的工程问题。1. 先理解 LLM Council 的结构它不是单模型调用而是一条内容流水线1.1 评审团模式解决的是单模型的质量不稳定性单个 LLM 生成内容时质量并不稳定。同一个提示词在多轮调用里可能会产出完全不同的结构也可能在事实细节上含糊其辞。更麻烦的是单个模型很难发现自己的错误因为它生成文本时的注意力分布和表达偏好是固定的。LLM Council 的核心思路是让多个模型承担不同角色先独立产出再交叉评审最后合并意见。这样做的价值不是让模型数量变多而是让不同模型的输出差异成为质量缓冲。比如写稿模型和事实核查模型来自不同厂商训练数据不同对同一个财经新闻的重复表述也会不同交叉检查时更容易暴露明显错误。“评审团”不是让 9 个模型同时回复同一个问题而是把内容生产拆成多个阶段每个阶段有明确的输入、输出和评判标准。这样系统才具备可追踪性出问题时能定位到具体环节。1.2 9 个模型如何分工角色、输入与输出我设计的 9 个角色如下。这里的“9 个模型”可以是 9 个不同厂商或不同版本的模型实例也可以是同一个模型按不同提示词和温度创建的 9 个角色实例。关键区别在于角色定义而不是模型本身的名称。角色实际作用主要输入主要输出推荐温度researcher阅读新闻源整理事实素材原始新闻、RSS 摘要100 到 300 字/条的结构化素材0.2writer根据素材写初稿素材包 选题Markdown 初稿0.7fact_checker抽取断言并判断可信度初稿 来源清单JSONclaims 数组0.1structure_reviewer检查段落顺序和章节逻辑初稿修改建议 JSON0.3style_reviewer检查表达是否清晰易懂初稿修改建议 JSON0.4compliance_reviewer过滤夸大、误导或违规表述初稿风险标记 JSON0.0aggregator合并各评审意见并生成修订稿初稿 所有评审意见修订版全文0.3finalizer输出最终排版文本和标题修订版全文终稿 Markdown0.5validator校验格式、字数和风险标记终稿JSON是否通过 问题列表0.0这里有一个容易误解的地方不要把“角色”理解成模型内置能力。提示词决定了角色模型只是按提示词输出。同一个模型既可以是 writer也可以是 compliance_reviewer只是它的输出风格和约束会完全不同。1.3 从选题到终稿整条链路要拆成十个可追踪的 stage我建议把整条链路拆成 10 个 stage每个 stage 都可以独立记录日志、保存快照、单独重试选题确定人工或模型生成候选主题。素材采集调用搜索 API、RSS 或数据库。researcher 整理素材包。writer 生成初稿。fact_checker 事实核查并返回 JSON。structure、style、compliance 三个评审并行执行。aggregator 合并评审意见并生成修订稿。finalizer 生成最终排版和标题。validator 校验输出格式和风险标记。人工复核后发布。把 stage 拆细有两个直接好处。第一失败范围被限制在单个 stage不用整期重跑第二每个 stage 的输入输出可以保存成快照方便复现和回归测试。2. 环境准备与依赖选型模型先不谈协议和数据结构要先定2.1 用统一客户端封装多模型接入先不要急着写业务代码第一步是解决多模型调用协议差异。不同模型平台的路径、鉴权方式、参数命名不完全一致有的叫max_tokens有的叫max_completion_tokens有的支持 JSON Mode有的只能用提示词约束输出格式。推荐使用litellm这类统一调用层或者自己封装一个薄客户端。下面是一个最小封装示例# pipeline/client.py import time from litellm import completion DEFAULT_TIMEOUT 45 MODEL_CONFIG { researcher: {model: openai/gpt-4o-mini, temperature: 0.2, max_tokens: 1200}, writer: {model: anthropic/claude-3-5-sonnet, temperature: 0.7, max_tokens: 3000}, reviewer: {model: openai/gpt-4o, temperature: 0.2, max_tokens: 1500}, } def call_model(role: str, prompt: str, response_formatNone): cfg MODEL_CONFIG[role] kwargs { model: cfg[model], messages: [{role: user, content: prompt}], temperature: cfg[temperature], max_tokens: cfg[max_tokens], timeout: DEFAULT_TIMEOUT, } if response_format: kwargs[response_format] response_format start time.time() resp completion(**kwargs) latency_ms int((time.time() - start) * 1000) content resp.choices[0].message.content usage resp.get(usage, {}) return content, usage, latency_ms这个封装解决了三个问题角色配置集中管理、超时统一处理、耗时和 token 用量在调用入口就能拿到。不要把各种模型的 API Key 直接写进代码建议通过环境变量注入避免误提交到仓库。注意不要把多个模型的配置散落在业务代码里。角色、模型名、温度、max_tokens 都应该集中在配置文件中这样后续降级和调参只需要改一处。2.2 每个 stage 的数据结构必须能校验多模型协作最怕的是“每个模型都在输出自由文本下游只能靠猜”。在进入实现之前需要先定义每个角色输出的 JSON Schema。以 researcher 输出为例{ $schema: http://json-schema.org/draft-07/schema#, type: object, required: [facts], properties: { facts: { type: array, items: { type: object, required: [title, summary, source], properties: { title: {type: string}, summary: {type: string}, source: {type: string} } } } } }除了 JSON Schema建议在代码里定义统一的 stage 返回结构# pipeline/stage.py from dataclasses import dataclass dataclass class StageResult: stage_id: str model: str status: str # success | retryable_error | fatal_error content: str raw_output: str usage: dict request_id: str error: str 为什么要保存raw_output因为模型返回的文本可能被截断、可能带 Markdown 代码块、可能直接触发了安全策略返回空内容。只保存解析后的 JSON 会丢失排查线索保存原始输出才能定位问题。2.3 动手之前先算清成本预算和限流上限9 个模型的成本不是简单相加还要考虑重试和评审放大效应。一期财经通讯可能包含素材采集、初稿、事实核查、3 个评审、汇总、终稿、校验至少 8 到 12 次模型调用。如果中间某个 stage 重试 2 次调用次数会明显上升。建议在配置里设置三项约束每个模型请求的max_tokens防止意外长输出。每个 stage 的max_retries防止极端情况下无限重试。每期总成本上限在调度代码里累计 usage超过阈值就终止。例如BUDGET_PER_ISSUE 3.0 # 单位美元每次调用结束都累加成本超过预算后直接进入降级路径。不要等到账单出来再拍脑袋。3. 最小可运行实现用 Python 跑通一期财经通讯3.1 项目结构与配置文件我用一个简化的项目结构演示。完整生产项目还需要测试、日志、监控和部署配置这里先保证最小可运行。council-newsletter/ ├── config/ │ ├── models.yaml │ └── roles.yaml ├── pipeline/ │ ├── __init__.py │ ├── client.py │ ├── stage.py │ ├── validate.py │ └── run.py ├── data/ │ ├── checkpoints/ │ └── topic.json ├── output/ └── requirements.txtrequirements.txt 里至少包含litellm1.40.0 jsonschema4.21.0 pyyaml6.0.0模型配置示例# config/models.yaml models: researcher: openai/gpt-4o-mini writer: anthropic/claude-3-5-sonnet reviewer: openai/gpt-4o角色配置示例# config/roles.yaml roles: researcher: prompt: 你是财经新闻研究员。请围绕主题整理 5 条事实要点。只输出 JSON。 temperature: 0.2 writer: prompt: 你是财经通讯主笔。请基于素材写作语言客观引用素材编号。输出 Markdown。 temperature: 0.73.2 核心调度代码并行评审与顺序汇总最简调度逻辑分 4 步整理素材、写初稿、事实核查、并行评审、汇总终稿。示例代码如下# pipeline/run.py import json from concurrent.futures import ThreadPoolExecutor from pipeline.client import call_model from pipeline.validate import parse_json_strict def build_material_prompt(facts: list[dict]) - str: lines [] for i, f in enumerate(facts, 1): lines.append(f[{i}] {f[title]}\n{f[summary]}\n来源: {f[source]}) return 请使用以下编号素材写作引用时标注编号。\n \n.join(lines) def run_issue(topic: str) - str: raw_material, _, _ call_model(researcher, f围绕 {topic} 整理 5 条事实要点只输出 JSON) facts parse_json_strict(raw_material, researcher) material_prompt build_material_prompt(facts.get(facts, [])) draft, _, _ call_model(writer, material_prompt) raw_fact_check, _, _ call_model(reviewer, f核查初稿中的事实断言输出 JSON。\n{draft}) fact_check parse_json_strict(raw_fact_check, fact_checker) with ThreadPoolExecutor(max_workers3) as pool: reviews list(pool.map( lambda role: call_model(role, draft)[0], [structure_reviewer, style_reviewer, compliance_reviewer] )) merged_reviews \n\n.join(reviews) aggregated, _, _ call_model( aggregator, f合并下面各评审意见生成修订版全文。\n草案\n{draft}\n评审意见\n{merged_reviews} ) final, _, _ call_model(finalizer, f基于修订版生成最终排版。\n{aggregated}) return final这一段说明了两个关键点。第一researcher、writer、aggregator、finalizer 等阶段是顺序依赖的必须同步执行而 structure、style、compliance 三个评审角色相互独立可以并行缩短整体耗时。第二所有需要继续传递给下游的内容都应尽量结构化writer 生成的草稿可以保留 Markdown因为它最终是给读者看的内容但 fact_checker、reviewer 的输出必须解析成 JSON 才能参与决策。注意并行调用 3 个评审模型时必须考虑上游 API 的限流。不要直接并行 9 个请求。3.3 运行方式与正常输出命令行入口可以这样写python -m pipeline.run --topic 2025 年人工智能芯片市场观察 --out output/2025-w01.md正常执行的输出示意[stage:researcher] tokens1540 latency2.8s statusjson_ok [stage:writer] tokens2230 latency6.1s statusok [stage:fact_checker] tokens890 latency3.2s statusjson_ok [stage:structure_reviewer] tokens610 latency2.7s statusok [stage:style_reviewer] tokens720 latency3.0s statusok [stage:compliance_reviewer] tokens300 latency1.8s statusok [stage:aggregator] tokens980 latency4.5s statusok [stage:finalizer] tokens1150 latency4.9s statusok 已写入 output/2025-w01.md如果看到某个 stage 长时间的statusretryable_error或直接抛异常就说明需要进入第 5 节的排查链路。不要只验证“程序能跑完”就结束还要确认每个 stage 的 JSON 字段、耗时和 token 消耗是否符合预期。4. 九个模型评审团里最常坏掉的五个环节4.1 上下文过长导致引用“失忆”现象writer 在写作时把较早新闻里的数据安到了后面新闻的主体上例如把 A 公司的营收写在 B 公司的段落里。原因把所有原始新闻全文拼进一个 prompt超过模型的有效注意力范围后前部信息容易被稀释。另一个常见情况是长文本被框架截断模型只“看到”了末尾几段。解决办法不要直接把原始新闻丢给 writer。先让 researcher 把每条新闻压缩到 200 到 300 字并用编号显式组织def build_material_prompt(facts): lines [] for i, f in enumerate(facts, 1): lines.append(f[{i}] {f[title]}\n{f[summary]}\n来源: {f[source]}) return 请使用以下编号素材写作引用时标注编号。\n \n.join(lines)然后要求 writer 在关键信息来源前标注编号。这样即便模型引用错误也能在事实核查阶段快速定位到素材编号。4.2 结构化输出解析失败JSON 不是理所当然的现象researcher 明明要求“只输出 JSON”返回内容却是好的这是整理后的结果 json { facts: [...] }希望对你有所帮助json.loads 会直接抛 JSONDecodeError整条流水线中断。 原因模型不会自动遵守“只输出 JSON”这种口头约束尤其当底层模型不支持 JSON Mode 时。 解决办法解析前先剥离 Markdown 代码块并保留原始输出 python # pipeline/validate.py import json import re def parse_json_strict(raw: str, stage_id: str) - dict: text raw.strip() if text.startswith(): text re.sub(r^(?:json)?\s*, , text) text re.sub(r\s*$, , text) try: data json.loads(text) except json.JSONDecodeError as exc: raise ValueError(f{stage_id}: invalid json, tail{text[-120:]!r}) from exc if not isinstance(data, dict): raise ValueError(f{stage_id}: json is not object: {type(data)}) return data更稳妥的方式是优先使用支持 JSON Mode 的模型并把response_format参数传给 API。但不能完全依赖它因为某些情况下模型仍可能输出空内容或整段安全提示。4.3 评审模型互相附和评审机制失效现象3 个评审模型输出的意见几乎一致全部都是“内容完整、结构清晰、无需修改”。但人工复核时发现初稿有明显的逻辑跳跃和数据缺失。原因评审模型和写稿模型共享了过多上下文或者评审提示词没有强制要求发现具体问题。默认情况下模型倾向于给出温和的正面反馈。解决办法把评审模型的输入限制在最小必要范围并且在提示词中强制输出“必须改进项”。示例你负责结构评审。请逐段检查初稿的信息顺序。 必须给出以下 JSON {must_improve: [至少一个具体问题], suggestions: [可执行的修改建议], overall_score: 0-10} 不要输出“内容完整、结构清晰”这类无信息量的评价。如果多个评审模型仍给出雷同意见可以刻意让 writer 和 reviewer 使用不同厂商的模型或者拉开温度差距。同质化是评审团模式最隐蔽的失效方式因为流程看似跑完了实际质量并没有提升。4.4 外部数据源变动拖垮整条流水线现象素材采集阶段连续失败或者搜索 API 返回的字段名变化导致 researcher 拿不到有效输入。原因外部数据源不受我们控制。RSS 结构升级、搜索接口调整、单条新闻字段缺失都会直接破坏下游 LLM 阶段的输入格式。解决办法把外部依赖与 LLM 调用彻底解耦。外部采集失败时不阻塞 LLM 流程可以先用本地缓存数据生成内容同时标记“本次内容基于缓存资料”。对外部源统一加超时、重试和字段默认值例如def safe_get(data, key, default): return data.get(key, default) if isinstance(data, dict) else default另外每次采集成功后保存一份本地快照。重跑同一期时优先从快照取数既能保证可复现也能减少对外部服务的依赖。我建议把这一条写进上线检查清单避免上线后第一个周末就被外部源变更打断。4.5 成本失控与限流模型越多越难控制现象某期内容实际消耗是预估的 8 倍或者某个模型接口连续返回429 Too Many Requests。原因常见原因有三个。一是某个模型输出过长max_tokens设置没有生效二是重试逻辑没有退避失败后立刻反复请求三是并行调用多个评审模型时触发了上游并发限制。解决办法在调度入口累加每次调用的 usage超过预算直接终止重试时使用指数退避并发数控制在 3 到 5 以内。不要为了追求速度把 9 个模型同时打出去。import time def call_with_backoff(role, prompt, max_retries2): delay 1 for attempt in range(max_retries 1): try: return call_model(role, prompt) except Exception as exc: if attempt max_retries: raise time.sleep(delay) delay * 25. 故障排查链路从一条日志倒推到根因5.1 先定位失败发生在哪一层遇到流水线报错时不要直接修改提示词。先用分层的方式确定故障位置。故障层典型现象首选检查方向网络 / API 层401、429、ConnectionErrorHTTP 状态码、请求耗时、API Key 状态外部数据源层素材为空、字段缺失采集模块日志、上游返回样例LLM 调用层空回复、finish_reason 异常原始输出头尾、finish_reason 字段JSON 解析层JSONDecodeError、字段缺失原始输出前后 120 字符下游写入层文件未生成、格式错误validator 返回的问题列表排查顺序建议先看异常来自哪一层再看该层有没有原始输出最后再判断是提示词问题还是代码问题。不要一上来就换模型那样会失去复现依据。5.2 给每个 stage 打上追踪 ID多模型流水线一旦超过 5 个 stage日志就会乱。每次请求都生成一个request_id并在一整条调用链里透传import uuid request_id uuid.uuid4().hex[:12]每个 stage 的记录建议包含这些字段{ ts: 2025-06-01T10:00:00Z, stage_id: fact_checker, request_id: req_1234abcd, model: openai/gpt-4o-mini, latency_ms: 3400, prompt_tokens: 1200, completion_tokens: 300, status: json_parse_error, raw_tail: ... }记录raw_tail而不是整个原始输出是为了在排查 JSON 解析问题时看到输出末尾是否有附加内容。如果整条日志过大也方便后续接日志系统。5.3 高频错误与处理方式对照表下面这张表可以直接作为排错手册使用。错误现象可能原因检查方式处理建议JSONDecodeError模型输出带 Markdown 代码块或追加语气词查看 raw_output 头尾剥离代码块后重试或改用 JSON Modecontent 为空命中内容安全策略或 max_tokens 设置过小查看 finish_reason降低温度重试或更换模型429 Too Many Requests并发过高或触达账户额度查看 RateLimit 响应头指数退避降低并发素材字段缺失外部数据源结构变化单测采集模块更新解析逻辑并保存快照评审意见全是正面评审提示词缺少强制改进要求抽查评审输出调整提示词强制输出 must_improve偶发超时上游服务波动查看 latency 分布增加超时时间加入降级路径这一节的重点不是背错误码而是养成“先看原始输出再看解析逻辑最后看提示词”的排查习惯。大多数问题都能通过这一条路径找到答案。6. 生产环境加固方案6.1 输出校验、自动重试与格式修复生产环境不能允许“JSON 解析失败”直接中断整期生成。在解析函数之上加一层重试逻辑def call_with_retry(role, prompt, max_retries2): for attempt in range(max_retries 1): raw, usage, latency call_model(role, prompt) try: return parse_json_strict(raw, role) except ValueError: if attempt max_retries: raise prompt prompt \n注意只输出 JSON不要使用 Markdown 代码块。重试时不是简单重复而是附加更强的格式约束同时适当降低温度。这样第二次成功的概率会明显提升。如果重试后仍失败该 stage 标记为fatal_error进入降级流程而不是一直死循环。6.2 checkpoint 快照与断点恢复每一个 stage 完成后把输入、输出、模型名、token 用量保存为 JSON 文件。这样某个 stage 失败时不必整期重跑。def save_checkpoint(stage_id: str, result: dict): path fdata/checkpoints/{stage_id}.json with open(path, w) as f: json.dump({stage_id: stage_id, result: result}, f, ensure_asciiFalse)恢复时先检查 checkpointimport os def load_or_run(stage_id, func): path fdata/checkpoints/{stage_id}.json if os.path.exists(path): with open(path) as f: return json.load(f)[result] result func() save_checkpoint(stage_id, result) return resultcheckpoint 的另一个用途是审计。每一期内容由哪些模型、哪些素材生成都可以在事后核对对内容生产系统的可解释性帮助很大。6.3 降级策略从 9 个模型缩到 3 个甚至 1 个预算、接口波动或上游故障时系统要能自动降级。降级不是越少越好而是按照质量损失从小到大选择降级级别模型组合适用条件质量影响L0 完整9 个模型全部参与预算充足、服务稳定质量最高L1 精简researcher writer fact_checker finalizer3 个评审接口不可用缺少结构/风格/合规评审L2 最小writer finalizer多个模型不可用质量明显下降需要人工复核L3 止损单模型直接输出全部下游不可用只做规则校验和格式检查关键点降级条件不能靠人临时判断而要在调度代码里通过异常次数和预算阈值自动触发。例如连续 3 次模型调用失败就跳过评审阶段直接进入 aggregated 逻辑。6.4 用回归测试盯住质量变化模型版本会更新提示词会调整数据源也会变化这些都可能让输出质量悄悄劣化。建议准备固定测试集包含 5 个左右有代表性的财经选题每次调整后统批量重跑并记录各模型版本和输出结果。回归检查清单可以是每期是否成功生成 Markdown 文件。每个 stage 是否返回合法 JSON。事实断言是否能在素材包中找到对应编号。是否有明显重复段落。人工打分是否低于历史基线。有了这些指标才能判断一次模型升级或提示词改动到底是在进步还是退化。不要只凭一两篇样本文本做判断。7. 最佳实践与扩展方向7.1 上线前检查清单我整理了一份可以直接使用的检查清单适合在接入生产环境前逐项核对所有外部 API 是否配置了超时和重试。所有 API Key 是否通过环境变量或密钥管理平台注入。每个角色的提示词是否有版本号。每个 stage 是否记录 token、耗时、模型名称和请求 ID。是否保存模型原始输出而不仅仅是解析后的结果。是否有 JSON Schema 校验和自动重试。是否有 checkpoint 和断点恢复。是否有降级路径降级触发条件是什么。是否有人工复核步骤负责人是谁。是否有固定测试集和回归记录。是否设置了单期成本上限。是否记录模型版本或 API 版本方便回滚判断。7.2 值得继续做的扩展方向首先是引入检索增强。当前设计里 researcher 依赖人工选择素材或固定数据源后续可以接入向量检索把事实证据召回做得更可控。其次是异步化。把长流程从同步请求改成异步任务队列前端只需要提交任务并轮询结果这样可以支持多期内容并行生成也方便重试。第三是接入人工反馈。每期内容发布后把人工修改和退回记录保存下来再通过偏好数据微调或调整提示词奖励方向形成质量闭环。成本优化方面可以尝试用开源模型替换商业模型中承担简单评审工作的部分例如 compliance 校验完全可以用规则加小模型完成不一定需要大模型。还有一个方向是可视化。做一个简单的管理台展示每个模型的响应时间、token 成本、失败率和输出样例对定位模型异常会有很大帮助。注意不要把多模型编排和 ComfyUI 这类本地图形节点工作流混为一谈。LLM Council 的模型通常通过 API 调用分散在不同服务端真正的难点是网络、鉴权、并发和成本控制而不是本地节点连线。7.3 如果从零开始建议先跑通最小版本不要直接搭 9 个模型。建议先用 3 个模型跑通全链路writer 写稿、fact_checker 核查、validator 校验。先记录 2 到 3 期内容的质量作为基线再逐步加入结构评审、风格评审和合规评审。每增加一个角色都要同时补上两样东西一条失败路径处理逻辑和一条降级策略。这样模型数量增加的同时系统的稳定性不会倒退。最后回到题目里的问题what breaks。真实运行之后断掉的地方几乎都不在模型生成环节而发生在模型调用之前和之后上下文如何组织、JSON 如何解析、外部素材如何获取、成本如何限制、失败之后如何恢复。9 个模型解决的是内容多样性和交叉验证问题但它本身并不会让流水线更稳定稳定性要靠工程手段补上。建议先把日志、校验、checkpoint 和降级逻辑做扎实再逐步扩模型数量每多一个模型就多