企业官网建设流程全解析

LLM Council，也就是多模型评审团，正在从实验玩法变成内容生产团队的真实选择。我实际跑过一个由 9 个模型组成的小型评审团，用来每周生成一份财经资讯通讯：研究员负责整理素材，写稿模型负责成文，事实核查和结构评审分别提出修改意见，最后由汇总模型确认终稿。这套流程听起来像是把多个 LLM 组合在一起“开会”，但跑起来之后真正需要花精力处理的不是某个模型写得不好，而是整条流水线哪些环节会断，以及断了之后如何快速恢复。这篇文章会围绕这个问题展开：先拆解 9 个模型的职责分工，再给出一个可以运行的最小实现，然后重点分析我踩过的 5 类故障，最后提供一套从日志倒推根因的排查链路和生产环境加固方案。

本文适合已经会调用单个模型 API，准备尝试多模型协作或多 Agent 流水线的开发者。不会涉及投资建议，只讨论多模型生成内容的工程问题。

1. 先理解 LLM Council 的结构：它不是单模型调用，而是一条内容流水线

1.1 评审团模式解决的是单模型的质量不稳定性

单个 LLM 生成内容时，质量并不稳定。同一个提示词在多轮调用里可能会产出完全不同的结构，也可能在事实细节上含糊其辞。更麻烦的是，单个模型很难发现自己的错误，因为它生成文本时的注意力分布和表达偏好是固定的。

LLM Council 的核心思路是：让多个模型承担不同角色，先独立产出，再交叉评审，最后合并意见。这样做的价值不是让模型数量变多，而是让不同模型的输出差异成为质量缓冲。比如写稿模型和事实核查模型来自不同厂商，训练数据不同，对同一个财经新闻的重复表述也会不同，交叉检查时更容易暴露明显错误。

“评审团”不是让 9 个模型同时回复同一个问题，而是把内容生产拆成多个阶段，每个阶段有明确的输入、输出和评判标准。这样系统才具备可追踪性，出问题时能定位到具体环节。

1.2 9 个模型如何分工：角色、输入与输出

我设计的 9 个角色如下。这里的“9 个模型”可以是 9 个不同厂商或不同版本的模型实例，也可以是同一个模型按不同提示词和温度创建的 9 个角色实例。关键区别在于角色定义，而不是模型本身的名称。

这里有一个容易误解的地方：不要把“角色”理解成模型内置能力。提示词决定了角色，模型只是按提示词输出。同一个模型既可以是 writer，也可以是 compliance_reviewer，只是它的输出风格和约束会完全不同。

1.3 从选题到终稿，整条链路要拆成十个可追踪的 stage

我建议把整条链路拆成 10 个 stage，每个 stage 都可以独立记录日志、保存快照、单独重试：

1. 选题确定：人工或模型生成候选主题。
2. 素材采集：调用搜索 API、RSS 或数据库。
3. researcher 整理素材包。
4. writer 生成初稿。
5. fact_checker 事实核查并返回 JSON。
6. structure、style、compliance 三个评审并行执行。
7. aggregator 合并评审意见并生成修订稿。
8. finalizer 生成最终排版和标题。
9. validator 校验输出格式和风险标记。
10. 人工复核后发布。

把 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_format=None): 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.txt

requirements.txt 里至少包含：

litellm>=1.40.0 jsonschema>=4.21.0 pyyaml>=6.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.7

3.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_workers=3) 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] tokens=1540 latency=2.8s status=json_ok [stage:writer] tokens=2230 latency=6.1s status=ok [stage:fact_checker] tokens=890 latency=3.2s status=json_ok [stage:structure_reviewer] tokens=610 latency=2.7s status=ok [stage:style_reviewer] tokens=720 latency=3.0s status=ok [stage:compliance_reviewer] tokens=300 latency=1.8s status=ok [stage:aggregator] tokens=980 latency=4.5s status=ok [stage:finalizer] tokens=1150 latency=4.9s status=ok 已写入 output/2025-w01.md

如果看到某个 stage 长时间的status=retryable_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_retries=2): 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 *= 2

5. 故障排查链路：从一条日志倒推到根因

5.1 先定位失败发生在哪一层

遇到流水线报错时，不要直接修改提示词。先用分层的方式确定故障位置。

排查顺序建议：先看异常来自哪一层，再看该层有没有原始输出，最后再判断是提示词问题还是代码问题。不要一上来就换模型，那样会失去复现依据。

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 高频错误与处理方式对照表

下面这张表可以直接作为排错手册使用。

这一节的重点不是背错误码，而是养成“先看原始输出，再看解析逻辑，最后看提示词”的排查习惯。大多数问题都能通过这一条路径找到答案。

6. 生产环境加固方案

6.1 输出校验、自动重试与格式修复

生产环境不能允许“JSON 解析失败”直接中断整期生成。在解析函数之上加一层重试逻辑：

def call_with_retry(role, prompt, max_retries=2): 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 = f"data/checkpoints/{stage_id}.json" with open(path, "w") as f: json.dump({"stage_id": stage_id, "result": result}, f, ensure_ascii=False)

恢复时先检查 checkpoint：

import os def load_or_run(stage_id, func): path = f"data/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 result

checkpoint 的另一个用途是审计。每一期内容由哪些模型、哪些素材生成，都可以在事后核对，对内容生产系统的可解释性帮助很大。

6.3 降级策略：从 9 个模型缩到 3 个甚至 1 个

预算、接口波动或上游故障时，系统要能自动降级。降级不是越少越好，而是按照质量损失从小到大选择：

关键点：降级条件不能靠人临时判断，而要在调度代码里通过异常次数和预算阈值自动触发。例如连续 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 和降级逻辑做扎实，再逐步扩模型数量；每多一个模型，就多