之前在设计一个自动生成金融资讯早报的项目时团队最初拍板用单个 LLM 一条 prompt 出全文结果样稿质量始终不稳定有时数据解读太浅有时风险提示缺失有时文风完全不像栏目定位。后来改成由 9 个 LLM 组成模型议事会协同生产输出质量确实上来了但新的问题也接踵而至链路超时、成本翻倍、格式解析失败、数字张冠李戴……如果你也在做多模型协作、多智能体内容生产或者正在规划 LLM 工程化项目这篇文章会把 9 模型议事会的架构设计、核心代码、以及生产环境里最容易坏的环节全部拆开讲清楚。内容偏实战不绕弯子每个问题都会给出复现场景和解决思路。1. 什么是LLM 议事会为什么要用 9 个模型1.1 从单模型到多模型协作单个 LLM 的能力边界已经很强但在内容生产类任务里仍然有三个明显的短板。第一单一模型很难同时兼顾多种能力。写金融资讯早报这件事既需要快速概括新闻又需要解读行情数据还需要判断风险、把关合规、调整文风。如果全部塞到一条 prompt 里模型往往会顾此失彼文风好的时候数据容易出错数据准确的时候又缺少可读性。第二单次生成的随机性无法消除。即便使用相同的 prompt同一个模型在不同轮次生成的内容也可能有差异。对于资讯类产品这种随机性意味着每天早上拿到的稿子质量波动很大难以建立稳定的质量预期。第三缺少交叉验证。金融资讯最怕的是数字错误、结论片面、缺乏风险提示。单个模型生成完没有第二双眼睛去检查错误就只能等人工发现了。LLM 议事会的思路是把多个大模型当作一支团队来用。每个模型负责一个明确的角色像开会讨论一样先分头输出再汇总、审核、投票最终产出一篇经过多重检查的内容。这样做的好处是职责单一、提示词清晰、每个环节都可以单独验证和优化。1.2 议事会的三种典型协作方式多模型协作并不是简单地把 9 个模型都调用一遍协作方式决定了系统的复杂度。常见的模式有三种。流水线模式Pipeline。模型按顺序执行前一个模型的输出作为后一个模型的输入。这种方式逻辑清晰、每个环节职责明确缺点是总耗时会累加且错误容易向后传播。并行投票模式Voting。多个模型针对同一个问题分别给出答案再由一个聚合器选择最一致的答案。这种方式适合选择题、标题生成、结论判断能够有效降低单模型的随机性但 token 成本会成倍上升。辩论审核模式Debate。先由一个模型生成初稿另一个模型扮演反对者指出漏洞再让生成方修订。金融资讯场景里非常实用相当于让模型互相挑刺。实际项目很少只用一种模式。9 个模型的议事会通常是三种模式的组合前段用流水线完成信息加工中段用审核模式做事实核查和合规检查最后用投票模式决定终稿是否通过。1.3 为什么金融资讯场景特别适合议事会金融内容有三个特殊性强时效、强准确、强合规。新闻晚发半小时价值就大打折扣一个数字错了轻则用户投诉重则引发合规问题如果模型无意中给出了建议买入某某股票这类表述还可能越过投资建议的边界。议事会模式恰好能回应这三个特殊性。多个模型分工处理后时效性可以由并发调用改善事实核查模型可以针对数字、日期、公司名称做二次校验合规审查模型可以专门负责过滤敏感表述和风险提示。当然这里要强调一点模型议事会只是内容生产的辅助手段不能替代人工复核更不能替代专业的合规审查。2. 系统整体架构9 个模型如何分工合作2.1 九个模型的分工设计9 个模型不能随意排列需要根据内容生产的流程拆出九个明确角色。下面是我在项目里使用的一套分工方案你可以直接参考。模型编号角色核心职责输入输出M1新闻摘要员压缩原始新闻提炼关键信息原始新闻列表结构化摘要M2数据解读员解读行情数据、财务数据摘要 行情数据数据要点M3趋势分析员分析短期趋势与关联事件数据要点趋势判断M4风险提示员找出风险点与不确定性趋势判断风险清单M5初稿撰写员按栏目模板写成稿前四步结果初稿M6事实核查员核对数字、名称、时间初稿 原始新闻核查意见M7合规审查员检查敏感表述、投资建议边界修订稿合规意见M8风格编辑统一文风、压缩篇幅合规稿件终稿候选M9总编决策判断是否通过、组织投票终稿候选发布/打回这套设计的核心思路是前四个模型负责把信息加工准确中间三个模型负责把内容生成完整最后两个模型负责把质量把关严格。2.2 数据流与执行顺序实际运行时模型之间的调用关系比表格里的角色划分要复杂一些。为了便于理解可以把整体数据流抽象成下面的链路。原始新闻 → M1 新闻摘要 → M2 数据解读 → M3 趋势分析 → M4 风险提示 → M5 初稿撰写 → M6 事实核查 → M7 合规审查 → M8 风格编辑 → M9 总编决策 → 人工复核 → 发布需要注意这个链路不是必须串行。M1、M2、M4 在拿到各自输入后可以并行执行减少整体耗时。后面介绍代码时会专门演示并行调用的写法。2.3 三个关键质量控制点如果全文跑完再检查发现问题时已经浪费了 9 次模型调用。更合理的做法是在链路里设置质量控制点。第一个控制点在 M2 之后。此时应该校验 M1 的摘要是否完整有没有丢掉关键新闻数字是否清晰如果摘要质量不合格就提前重跑而不是带着错误继续往后传。第二个控制点在 M6 之后。事实核查结果出来时需要对比初稿和原始新闻统计数字错误数量和关键信息缺失数量。超过阈值就打回重写。第三个控制点在 M9 之前。终稿候选需要经过格式校验、字数校验、违禁词校验全部通过才进入总编决策。三个控制点把一条长链路切成了四段任何一段出问题都能快速定位而不是等到最终输出才发现整篇稿子废掉了。3. 环境准备与项目结构3.1 运行环境说明本文的示例代码基于 Python 环境使用 requests 库直接调用 OpenAI 兼容协议的接口。这样做的原因是 9 个模型很可能来自不同供应商而市面上大多数 LLM 服务都提供 OpenAI 兼容接口统一协议后调度代码可以只写一套。版本方面没有硬性要求Python 3.9 以上即可requests 使用常规版本。如果后续要引入官方 SDK比如 openai 包版本需要根据你的项目实际情况调整本文重点演示的是多模型调度的整体思路不绑定某个具体 SDK 的版本。建议把 API Key 统一放在环境变量或配置中心不要写死在代码里。示例中会用 os.environ 读取。3.2 项目目录结构一个能跑通的多模型议事会项目目录结构可以参考下面这样。llm_council_news/ ├── config.py # 模型配置 ├── llm_client.py # 统一 LLM 调用客户端 ├── council.py # 议事会调度逻辑 ├── validators.py # 输出校验与格式解析 ├── prompts/ │ ├── summarizer.txt # 各角色提示词 │ ├── writer.txt │ ├── fact_checker.txt │ └── ... ├── data/ │ └── daily_news.md # 原始新闻材料 └── output/ └── newsletter.md # 最终生成的早报prompts 目录和代码分离是非常重要的习惯。9 个模型意味着至少 9 套提示词如果把提示词全部写在代码里调优时每次都要改代码、重新发布效率太低。拆成独立文件后提示词的调整可以单独进行。3.3 模型接入的统一方式不同供应商的接入方式略有差异但在 OpenAI 兼容协议下统一抽象成四个参数即可api_key接口密钥base_url接口地址model模型名称timeout超时时间这里不针对某个具体模型厂商做绑定你在使用时只需要把 base_url 和 model 换成自己实际购买的资源即可。如果某些模型没有 OpenAI 兼容接口也可以先封装一层适配器把请求格式转换成目标厂商的格式但对外暴露的方法保持一致。4. 核心代码实现从配置到调度4.1 模型配置文件第一步把 9 个模型的信息集中管理。示例中的地址和模型名称都是占位符实际使用时需要替换成你自己的配置。# 文件路径config.py import os MODEL_COUNCIL { M1_summarizer: { api_key: os.environ.get(M1_API_KEY, ), base_url: os.environ.get(M1_BASE_URL, https://api.example.com/v1), model: os.environ.get(M1_MODEL_NAME, summarizer-model), temperature: 0.2, timeout: 30, }, M2_analyst: { api_key: os.environ.get(M2_API_KEY, ), base_url: os.environ.get(M2_BASE_URL, https://api.example.com/v1), model: os.environ.get(M2_MODEL_NAME, analyst-model), temperature: 0.2, timeout: 30, }, # M3 - M8 结构相同这里省略 M9_editor: { api_key: os.environ.get(M9_API_KEY, ), base_url: os.environ.get(M9_BASE_URL, https://api.example.com/v1), model: os.environ.get(M9_MODEL_NAME, editor-model), temperature: 0.1, timeout: 30, }, }把 API Key 放在环境变量里而不是写死在配置中原因很简单代码会进入版本库环境变量不会。一旦 Key 泄露到 git 历史里只能作废重建代价很高。4.2 统一 LLM 调用客户端所有模型调用都走同一个方法后续加超时、重试、日志都只需要改一处。# 文件路径llm_client.py import json import time import requests class LLMClient: 兼容 OpenAI 协议的统一客户端 def __init__(self, name: str, api_key: str, base_url: str, model: str, temperature: float 0.3, timeout: int 30): self.name name self.api_key api_key self.base_url base_url.rstrip(/) self.model model self.temperature temperature self.timeout timeout def chat(self, messages: list, max_tokens: int 2000) - str: url f{self.base_url}/chat/completions headers { Authorization: fBearer {self.api_key}, Content-Type: application/json, } payload { model: self.model, messages: messages, temperature: self.temperature, max_tokens: max_tokens, } resp requests.post(url, headersheaders, jsonpayload, timeoutself.timeout) resp.raise_for_status() data resp.json() return data[choices][0][message][content] def chat_with_retry(self, messages: list, retries: int 3, backoff_factor: float 2.0) - str: 带指数退避重试的调用 last_exc None for attempt in range(retries): try: return self.chat(messages) except Exception as exc: last_exc exc wait_time backoff_factor * (2 ** attempt) print(f[{self.name}] 第 {attempt 1} 次调用失败 f{wait_time:.1f}s 后重试错误{exc}) time.sleep(wait_time) raise RuntimeError(f[{self.name}] 重试 {retries} 次后仍然失败) from last_exc def build_messages(self, system_prompt: str, user_content: str) - list: 构造 messages避免每个调用方都重复拼接 return [ {role: system, content: system_prompt}, {role: user, content: user_content}, ]这里的重点是 chat_with_retry 方法。生产环境里模型接口随时可能抖动没有重试的调用链路就像没有保险绳的攀岩。指数退避的意思很简单第一次失败等 2 秒第二次失败等 4 秒第三次等 8 秒避免在接口还没恢复时反复把请求打进去。4.3 读取提示词文件提示词建议从文件读取方便随时调优。# 文件路径council.py from pathlib import Path PROMPT_DIR Path(__file__).parent / prompts def load_prompt(name: str) - str: path PROMPT_DIR / f{name}.txt if not path.exists(): raise FileNotFoundError(f提示词文件不存在{path}) return path.read_text(encodingutf-8)比如 M1 新闻摘要员的提示词文件内容可以是你是金融新闻摘要员。 请把用户提供的新闻材料压缩成结构化摘要要求 1. 保留公司名称、股票代码、涨跌幅、金额、日期等关键数字 2. 按“宏观 / 行业 / 个股”分类 3. 输出格式为 Markdown 列表 4. 遇到不确定的信息标注“待核实”不要猜测。提示词写得越具体输出越稳定。像保留股票代码这种约束直接决定了后续事实核查的难度。4.4 议事会调度逻辑下面这段代码是整条流水线的核心。为了兼顾完整性和并发效率前四个模型先并行执行收集结果后再进入后段。# 文件路径council.py import concurrent.futures from config import MODEL_COUNCIL from llm_client import LLMClient def build_clients() - dict: clients {} for name, cfg in MODEL_COUNCIL.items(): clients[name] LLMClient( namename, api_keycfg[api_key], base_urlcfg[base_url], modelcfg[model], temperaturecfg.get(temperature, 0.3), timeoutcfg.get(timeout, 30), ) return clients def run_news_pipeline(news_material: str, clients: dict) - dict: 前段信息加工四个模型并行执行 results {} def call_m1(): prompt load_prompt(summarizer) return clients[M1_summarizer].chat_with_retry( clients[M1_summarizer].build_messages(prompt, news_material) ) def call_m2(summary): prompt load_prompt(analyst) return clients[M2_analyst].chat_with_retry( clients[M2_analyst].build_messages(prompt, summary) ) def call_m3(analysis): prompt load_prompt(trend_analyst) return clients[M3_analyst].chat_with_retry( clients[M3_analyst].build_messages(prompt, analysis) ) def call_m4(trend): prompt load_prompt(risk_checker) return clients[M4_risk].chat_with_retry( clients[M4_risk].build_messages(prompt, trend) ) # M1 先执行因为后面三步都依赖它的输出 summary call_m1() results[summary] summary # M2、M3、M4 实际上有依赖关系这里简化成串行便于理解。 # 如果想并发可以把 M3 拆成不依赖 M2 的并行分支。 analysis call_m2(summary) results[analysis] analysis trend call_m3(analysis) results[trend] trend risk call_m4(trend) results[risk] risk return results并发部分我做了简化处理。实际项目中M2 和 M3 有依赖关系时不要盲目并行否则会拿到不完整的上下文。真正可以并行的场景是多个模型各自处理不同板块的新闻最后再合并。接下来是后段生成与审核流程。# 文件路径council.py续 def run_writing_pipeline(pipeline_results: dict, clients: dict) - str: 中段写作 审核 context \n\n.join([ f## 摘要\n{pipeline_results[summary]}, f## 数据解读\n{pipeline_results[analysis]}, f## 趋势分析\n{pipeline_results[trend]}, f## 风险提示\n{pipeline_results[risk]}, ]) # M5 初稿撰写 writer_prompt load_prompt(writer) draft clients[M5_writer].chat_with_retry( clients[M5_writer].build_messages(writer_prompt, context) ) # M6 事实核查 checker_prompt load_prompt(fact_checker) fact_check clients[M6_checker].chat_with_retry( clients[M6_checker].build_messages( checker_prompt, f原始新闻材料\n{context}\n\n初稿\n{draft}, ) ) # 简单的打回机制如果核查意见里出现“错误”关键词就提示需要人工介入 if 错误 in fact_check or 不准确 in fact_check: print(【警告】事实核查发现问题进入人工处理流程) # 生产环境可以在这里触发告警邮件或企业微信通知 return draft, fact_check这个打回机制是一个很朴素的规则判断。生产环境里更推荐把事实核查的输出定义为结构化 JSON例如 {error_count: 2, errors: [xx数据错误]}方便代码精确判断是否打回。4.5 输出校验与 JSON 解析多模型协作时模型返回的格式很难保证一致。有的模型会在 JSON 外面包一层 Markdown 代码块有的会附带解释文字。写一个健壮的解析函数非常关键。# 文件路径validators.py import json import re def extract_json(text: str) - dict: 从模型输出中提取 JSON 对象 text text.strip() # 去掉 Markdown 代码块标记 text re.sub(r^(?:json)?|$, , text, flagsre.MULTILINE).strip() # 去掉 json 前后可能的说明文字 start text.find({) end text.rfind(}) if start ! -1 and end ! -1: text text[start:end 1] return json.loads(text) def validate_required_fields(data: dict, required_keys: list) - bool: missing [key for key in required_keys if key not in data] if missing: print(f缺少必要字段{missing}) return False return True使用示例result_text json\n{error_count: 0, errors: []}\n data extract_json(result_text) print(data) # {error_count: 0, errors: []}这种解析方式虽然简单但已经能覆盖绝大多数模型输出格式问题。更严格的场景可以引入 Pydantic 做字段类型校验本文不再展开。4.6 投票与共识机制最后一个模型 M9 需要做是否通过的决策。为了让决策更稳定可以引入简单的多数投票让三个不同的模型分别对终稿候选打分取多数结果。# 文件路径validators.py续 from collections import Counter def majority_vote(values: list, min_agree: int 2): 多数投票返回票数最高的值和票数。 counter Counter(values) winner, count counter.most_common(1)[0] if count min_agree: return None, count return winner, count # 示例三个裁判模型的评分 scores [通过, 通过, 打回] winner, count majority_vote(scores, min_agree2) print(winner, count) # 通过 2投票机制的代价是额外增加模型调用但它能显著降低单模型随机性带来的误判。如果对成本敏感可以只在通过/打回这种关键决策上使用投票日常内容生成不要每个环节都投票。5. 生产环境中最容易坏的环节架构和代码都跑通之后真正的挑战才开始。下面是 9 模型议事会方案在生产环境里最容易出问题的五个方向。5.1 稳定性一个模型超时整条链路卡死最直观的问题就是超时。9 个模型串行调用时假设每个模型平均耗时 15 秒总耗时是 135 秒。如果某个模型突发故障单次请求可能要等到超时边界才返回整个链路就被拖住。更危险的是级联失败。流水线模式下M5 拿到的 context 来自前四个模型如果 M2 输出了一串无意义内容M5 会基于错误继续生成后面的 M6、M7 也在错误基础上工作等到 M9 决策时才发现整篇稿子是废稿前面 8 次调用全部白费。解决思路有三个一是给每次调用设置合理的超时时间超过就放弃或降级二是在关键环节之间加中间校验提前拦截错误三是把从新闻到初稿的链路拆成多个独立任务某一个失败时先发其他板块。5.2 成本9 个模型意味着 9 倍甚至更高的 token 消耗成本是最容易被低估的问题。原始新闻材料会被摘要、解读、分析、写作、核查等多个模型反复读取同一份材料可能被计费多次。再加上重试机制、投票机制实际 token 消耗往往比模型数 × 单个模型输出量高得多。举一个粗略的例子。假设一份原始新闻材料约 3000 token9 个模型都处理一遍仅输入就是 27000 token。如果 M1 和 M6 因为输出格式不对各重试一次还要再增加几千 token。一个月下来成本会明显超出预算。控制成本的核心手段是缓存和降级。同一个新闻材料在短时间内的摘要结果可以缓存没有新增重大事件时趋势分析不必每天重新调用投票环节可以只在必要时启用不用每次都用三个模型打分。5.3 一致性模型输出格式千奇百怪9 个模型的 prompt 水平、指令遵循能力不同输出格式很难统一。同一个请输出 JSON的指令模型 A 老老实实输出 JSON模型 B 会在 JSON 外面加 Markdown 代码块模型 C 可能输出一大段解释再加 JSON。格式不一致会导致解析层频繁报错。如果只针对某一个模型做了解析优化换一个模型又可能出问题。最稳妥的方案是在每个模型调用后统一经过 extract_json 这类清洗函数并且把输出格式规范写进提示词示例要具体不能只说请遵守格式。5.4 事实性金融数字被合理化篡改这是金融资讯场景最危险的问题。LLM 在生成时倾向于让内容看起来连贯即使原始数据里没有某个涨幅模型也可能根据上下文推算出一个合理但错误的数字。更麻烦的是多模型流水线会放大这个错误M2 解读了错误数字M5 把错误数字写进稿子M6 如果不够严谨可能会认为数字和上下文一致就直接放行。应对方案有两层。第一层在提示词里强制要求模型不得编造数字数据来源不明时必须标注并把这一条设为最高优先级。第二层在事实核查环节把原始新闻中的关键数字提取出来和初稿中的数字做程序化比对而不是只靠模型自查。5.5 时效性串行链路吃掉了新闻的鲜度资讯早报对发布时间有硬要求。如果链路设计成 9 个模型完全串行任何一个环节慢一点就会推迟整份早报的发布时间。要解决时效问题就必须在设计阶段规划好哪些步骤可以并行。新闻摘要之后可以按板块拆成两个或三个分支并行处理最后再合并。另外建议把发布截止时间作为硬约束写进调度器例如在 08:45 之前必须完成初稿否则就跳过可选环节直接使用当前内容保证至少能按时发出去。6. 常见故障与排查清单下面是 9 模型议事会项目里最常见的几类故障整理成了速查表遇到问题时可以直接对照排查。问题现象常见原因解决思路链路总耗时过长模型串行调用过多单个模型响应慢拆分并行分支设置超时设置发布截止时间某个模型频繁报错接口限流、配额不足、网络波动增加指数退避重试准备备用模型做降级输出 JSON 解析失败模型在 JSON 外添加了代码块或说明文字使用 extract_json 清洗在提示词中给格式示例初稿出现错误数字前序模型输出被污染或模型自行推断增加程序化数字校验事实核查环节返回结构化 JSON成本超预算同一材料被多次计费重试和投票放大消耗缓存中间结果按需启用投票监控每日 token 用量稿子风格不稳定同一个模型在不同温度下生成差异大降低 temperature切换模型版本时做回归对比汇总时上下文过长前序结果全部拼进新 prompt超过上下文窗口对中间结果做截断按关键信息抽取后再拼接排查顺序建议遵守从数据流下游往上游查的原则。先看最终输出是否格式错误再看事实核查是否发现问题再看初稿再往前推。避免一开始就去怀疑模型本身的生成能力大多数问题都出在调度、解析和校验层。7. 最佳实践与工程建议7.1 用超时、重试、熔断保护链路多模型服务是外部依赖必须有像对待数据库、消息队列一样的态度。每个调用设置超时超时后自动重试连续失败时触发熔断暂时不再调用该模型改走备用模型或缓存。重试不能无限重试否则接口故障时请求会全部堆积在客户端。7.2 用缓存和降级控制成本9 个模型的生产链路非常适合加缓存。最简单的做法是使用新闻标题或日期作为 key缓存 M1 的摘要结果。假设一小时内的多次尝试都命中缓存可以省掉两三次大模型的输入消耗。降级策略也要提前设计如果 M7 合规审查不可用是直接打回人工还是先输出草稿并标记未审这个问题必须在生产事故之前想清楚。7.3 用结构化输出和数据校验兜底让模型输出纯文本再解析不确定性太高。金融资讯场景建议从第一版就要求关键环节输出结构化 JSON例如事实核查结果、风险清单、投票分数。同时把原始新闻中的关键数字用正则表达式提取出来与模型输出中的数字逐一比对这是比让模型自己检查自己可靠得多的兜底方案。7.4 用人工复核划分责任边界模型议事会可以大幅提升内容生产效率但不应让它独立对外发布内容尤其是金融资讯。最合理的责任划分是模型负责初稿和建议人工负责最终审核与发布。系统要在稿件落款中保留审核人信息并保留每一次模型调用的输入输出日志这样即使出现问题也能追踪是哪一步造成的。7.5 配置与模型版本管理9 个模型的提示词、模型名称、参数配置都要纳入版本管理。推荐把提示词拆成独立文件并放进 git 仓库每次调整都带着 commit 记录。模型版本切换时要先做小批量回归测试观察输出格式、长度、风格是否变化确认无回归后再全量切换。否则供应商悄悄升级模型版本可能导致你的解析逻辑一夜之间全部失效。7.6 日志与监控多模型链路一定要记录完整的调用日志包括每个模型的输入摘要、输出摘要、耗时、token 用量、是否重试、是否成功。这些数据既能用于排查问题也能用于成本分析和模型选型。建议给每个模型加一个简单的成功率指标和平均耗时指标一旦连续多分钟低于阈值立即触发告警。8. 总结与下一步这篇文章从一个实际的 9 模型议事会项目出发梳理了多模型协作生产金融资讯的完整链路从角色分工、架构设计、环境准备到核心代码实现再到生产环境里最容易出问题的稳定性、成本、一致性、事实性和时效性风险。整个方案的核心并不是用了 9 个模型所以更厉害而是通过角色拆分、多重校验、投票决策把单模型的不确定性控制在一个可接受的范围。真正决定项目成败的往往是那些看起来不起眼的工程细节超时重试怎么写、中间结果是否缓存、数字校验是否自动化、日志是否完整。如果你准备自己搭一套类似系统建议不要一开始就上 9 个模型。先选两三个关键角色验证提示词和链路跑通后再逐步增加模型数量。先保证每天能稳定产出一份合格的早报再去追求多模型交叉验证带来的质量提升这样踩坑成本会低很多。如果本文对你有帮助可以收藏备用。后续如果遇到多模型调度相关的具体报错也欢迎在评论区留言我会根据实际项目经验补充更多排查案例。