企业官网建设流程全解析

这次我们来看一个很有意思的项目：I run a 9-model LLM council to write a financial newsletter. What breaks。

这个项目的核心不是“单模型写稿”，而是把 9 个 LLM 组织成一个小型“内容委员会”，每个模型承担不同职责：有的负责收集市场信息，有的负责宏观分析，有的专门写多头逻辑，有的专门写空头逻辑，最后由一个模型汇总成正式的金融简报。同时项目作者真正关心的问题是：这套多模型协作流程，到底会在哪个环节断掉。

如果你正在做多 Agent 协作、LLM Pipeline、金融资讯自动生成、或者想把多个模型接进同一套工作流，这篇文章建议直接收藏。接下来我会从架构设计、环境准备、调用方式、批量任务、资源占用和故障排查几个维度，把这类项目的完整玩法拆开讲清楚。

先说几个读者最关心的判断：

• 项目价值：适合研究多模型分工、RAG、结构化输出和内容审核流水线，不是单纯“调 API 生成一篇文章”。
• 门槛：用云端 API 时基本不挑显卡；本地推理则需要按所选模型的实际大小准备显存和内存。
• 最容易翻车的地方：上下文长度、JSON 输出格式不稳定、多模型意见趋同、级联失败、成本失控。
• 本文会演示：9 模型流程设计、部署方式、功能验证、API 调用示例、批量任务脚本、常见故障排查。

1. 核心能力速览

2. 适用场景与使用边界

2.1 适合谁

• LLM 工程实践者：不满足于单模型调用，想理解多模型协作时的数据流、错误传播和降级策略。
• 金融内容生产团队：需要每天生成结构稳定的市场简报，对格式、多空逻辑、风险提示有明确要求。
• 研究 RAG 和多 Agent 协作的开发者：这个项目本质上是“多角色多 Agent”的内容生产流水线。
• 对成本敏感的个人开发者：可以用小模型做前置筛选，用大模型做最终总结，形成金字塔式调度。

2.2 能解决什么问题

单个模型写金融简报，最常见的问题是：信息视野有限、立场单一、输出格式不稳定。9 模型委员会把流程拆成多个环节：信息收集、筛选、分析、辩论、汇总、审核。每个环节由专门模型负责，输出的结构化和稳定性会比“单模型一口气生成”好很多。

2.3 不适合什么场景

• 追求极低延迟：9 个模型串行或半并行调用，单次延迟会明显高于单个模型。
• 完全自动驾驶：金融内容涉及合规和责任问题，不能没有人工复核。
• 预算极有限：如果全部使用大模型 API，每日多次运行的成本会累积。
• 对可解释性要求极高的场景：LLM 的分析过程仍是黑盒，需要审计日志兜底。

2.4 合规与安全边界

金融信息有强监管属性。生产环境中不要直接对外发布未经人工审核的内容。涉及到公司财务、个股推荐、投资建议等敏感内容，必须：

• 在文末声明“不构成投资建议”。
• 保留模型输入、输出、审核记录等可追溯日志。
• 对新闻源做版权和授权检查。
• 涉及用户私有数据时，先做脱敏和权限控制。

3. 环境准备与前置条件

3.1 系统与软件要求

3.2 Python 依赖清单

下面是通用依赖示例，实际版本请以项目requirements.txt或官方文档为准。

pip install openai pydantic python-dotenv requests pandas feedparser

依赖说明：

• openai：调用 OpenAI 兼容接口。
• pydantic：做输出结构校验。
• python-dotenv：管理 API Key 和模型配置。
• requests：调用其他 HTTP 服务或自建接口。
• pandas：处理结构化数据，比如新闻源表格。
• feedparser：可选的 RSS 解析，用于批量收集财经新闻。

3.3 环境变量配置

在项目根目录创建.env文件：

# .env 示例 OPENAI_API_KEY=sk-your-key-here OPENAI_BASE_URL=https://api.openai.com/v1 COUNCIL_MODEL_ANALYST=gpt-4o-mini COUNCIL_MODEL_EDITOR=gpt-4o LOCAL_MODEL_URL=http://127.0.0.1:8000/v1 REQUEST_TIMEOUT=120 MAX_RETRIES=3

说明：供应商、模型名和 Endpoint 需要按照你实际可用的服务填写，上面只是模板。如果同时使用本地模型和云端模型，可以像上面那样分别配置。

4. 安装部署与启动方式

4.1 项目目录结构

一个典型的 9 模型委员会项目，代码结构可以这样组织：

llm-council/ ├── .env ├── requirements.txt ├── config.yaml ├── src/ │ ├── orchestrator.py │ ├── models_registry.py │ ├── collectors/ │ │ ├── rss_collector.py │ │ └── news_api_collector.py │ ├── analysts/ │ │ ├── macro_analyst.py │ │ ├── sector_analyst.py │ │ └── risk_analyst.py │ ├── writers/ │ │ ├── bull_writer.py │ │ ├── bear_writer.py │ │ └── editor.py │ └── guards/ │ └── compliance_checker.py ├── data/ │ ├── raw/ │ ├── processed/ │ └── outputs/ └── scripts/ └── run_batch.py

这种分目录管理的核心目的是：每个模型模块独立，方便替换、测试和排查。

4.2 模型注册表设计

9 个模型不应写死在业务代码里，而是通过一个注册表统一管理。

# src/models_registry.py 示例 from dataclasses import dataclass @dataclass class ModelRole: role: str model_name: str temperature: float max_tokens: int base_url: str = "https://api.openai.com/v1" MODEL_REGISTRY = [ ModelRole("collector", "gpt-4o-mini", 0.1, 1000), ModelRole("filter", "gpt-4o-mini", 0.1, 1000), ModelRole("macro", "gpt-4o", 0.3, 2000), ModelRole("sector", "gpt-4o", 0.3, 2000), ModelRole("bull", "gpt-4o", 0.4, 1500), ModelRole("bear", "gpt-4o", 0.4, 1500), ModelRole("arbiter", "gpt-4o", 0.2, 1500), ModelRole("editor", "gpt-4o", 0.2, 2500), ModelRole("compliance", "gpt-4o-mini", 0.0, 1000), ]

这里的角色、模型名、温度都可以按需求调整，重点是每个角色有独立的 temperature 和 token 上限，避免“所有模型生成风格一模一样”的问题。

4.3 服务启动方式

4.3.1 单次命令行运行

python src/orchestrator.py --config config.yaml --date 2025-05-20

命令行参数含义：

• --config：指定配置文件。
• --date：指定生成哪一天的简报。
• --dry-run：只打印每个模型的输入输出，不真正写入文件。

4.3.2 校验配置

python src/orchestrator.py --check-config

这个命令用于检查模型注册表、API Key、目录是否存在，适合部署前快速验证。

4.3.3 启动 API 编排服务

如果你想对外提供“生成简报”的接口，可以加一层 FastAPI：

uvicorn src.api:app --host 127.0.0.1 --port 8000

注意：如果本机 8000 端口被其他服务占用，换成 8010 或 9000 都可以。

5. 9 模型流程设计：谁负责什么

这是整个项目最有价值的部分。与其把九个模型“一起扔进去”生成一篇文章，不如把工作拆开，让每个模型只做一件事。

5.1 环节一：信息采集与筛选

这一步通常用小模型即可。核心目标是减少后续大模型处理的信息量，控制 token 成本。

5.2 环节二：市场分析

建议这 3 个模型并行调用，再把结果汇总给后面的辩论模块。

5.3 环节三：多空辩论

多空辩论是关键设计。如果不设置这个环节，简报很容易变成“单一立场的长篇大论”；设置以后，最终文章会有明确的多空对照。

5.4 环节四：汇总与合规

合规环节的优先级最高：如果 compliance 返回失败，草稿不应进入发布流程。

5.5 流程控制逻辑

整个 pipeline 可以用一个简单的编排器串联：

# src/orchestrator.py 核心逻辑示例 from concurrent.futures import ThreadPoolExecutor def run_council(context: dict) -> dict: # 阶段1：采集和筛选 raw_news = collector.run(context["news_sources"]) filtered_news = news_filter.run(raw_news) # 阶段2：并行分析 with ThreadPoolExecutor(max_workers=3) as executor: macro_future = executor.submit(macro_analyst.run, filtered_news) sector_future = executor.submit(sector_analyst.run, filtered_news) risk_future = executor.submit(risk_analyst.run, filtered_news) macro_result = macro_future.result() sector_result = sector_future.result() risk_result = risk_future.result() # 阶段3：多空辩论 bull_case = bull_writer.run(macro_result, sector_result) bear_case = bear_writer.run(macro_result, sector_result) arbiter_result = arbiter.run(bull_case, bear_case) # 阶段4：编辑和合规 draft = editor.run(arbiter_result, risk_result) compliance_result = compliance_checker.run(draft) if not compliance_result["pass"]: return {"status": "rejected", "reason": compliance_result["reason"]} return {"status": "ok", "newsletter": draft}

从这个流程可以看到，一个环节出错，整条链都会受影响。这也是原作者问“What breaks”的核心原因。

6. 功能测试与效果验证

6.1 测试目标

不要一上来就跑完整流程，要把每个环节拆出来单独测。

6.2 单环节测试：输出 JSON 是否合法

很多流程断在“模型输出无法解析”。建议强制模型返回 JSON，并做 schema 校验。

import json from pydantic import BaseModel class FilterResult(BaseModel): kept_ids: list[str] dropped_ids: list[str] reason: str # 调用模型后 raw_output = model_response.choices[0].message.content try: data = json.loads(raw_output) result = FilterResult(**data) print("JSON valid, kept:", result.kept_ids) except Exception as e: print("Invalid JSON or schema mismatch:", e)

6.3 辩论对抗测试

这是多模型流程特有的测试。输入同一份宏观数据，连续运行 3 次 bull 和 bear，观察：

• 多空论据是否重复。
• 立场的差异性。
• 是否产生了明显错误的数据引用。

判断标准：多空论据重叠度不应过高，否则说明模型在“复读”而不是“对抗”。

6.4 合规拦截测试

准备一份包含明显风险表述的草稿，例如“该股票未来必然上涨”，交给 compliance checker，正常应该返回不通过。

如果合规环节没有拦截，说明 prompt 里的审核规则太弱，需要补强。

6.5 端到端测试

python src/orchestrator.py --config config.yaml --date 2025-05-20 --output-dir ./data/outputs

成功后检查输出目录：

data/outputs/ ├── 2025-05-20/ │ ├── newsletter.md │ ├── audit_log.json │ ├── raw_news.json │ └── compliance_report.json

audit_log.json建议包含：每个模型的输入摘要、输出摘要、耗时、token 消耗、成功/失败状态。

这段日志在出问题时会帮你快速定位断点。

7. 接口 API 调用与批量任务

7.1 单次 API 调用模板

不管底层用哪个供应商，都建议统一封装成 OpenAI 兼容接口。请求代码示例如下：

from openai import OpenAI import os client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL"), ) def call_model(role: str, system_prompt: str, user_content: str) -> str: """ 统一模型调用入口，实际参数需要按项目模型配置调整。 """ response = client.chat.completions.create( model=os.getenv(f"COUNCIL_MODEL_{role.upper()}", "gpt-4o-mini"), messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_content}, ], temperature=0.2, max_tokens=1024, ) return response.choices[0].message.content

有几点建议：

• 调用时设置timeout，避免某个模型卡住拖垮整个流程。
• 对结果做字符串长度检查，防止空输出进入下一环节。
• 在日志里记录每轮调用的模型名和 token 数。

7.2 curl 调用示例

如果你把编排服务包装成 HTTP 接口，可以用 curl 测试：

curl -X POST http://127.0.0.1:8000/api/newsletter \ -H "Content-Type: application/json" \ -d '{"date": "2025-05-20", "source": ["reuters", "bloomberg"], "mode": "quick"}'

响应示例：

{ "status": "ok", "newsletter": "## 市场综述\n...", "audit": { "total_tokens": 41230, "elapsed_seconds": 86.4, "models_ok": 9, "models_failed": 0 } }

7.3 批量任务设计

批量生成多天简报时，常见做法是按日期遍历，每次独立运行。

# scripts/run_batch.py 示例 import subprocess import time dates = ["2025-05-18", "2025-05-19", "2025-05-20"] for date in dates: print(f"=== {date} ===") result = subprocess.run( ["python", "src/orchestrator.py", "--date", date], capture_output=True, text=True, ) if result.returncode != 0: print(f"FAILED: {date}, stderr: {result.stderr[-500:]}") else: print(f"OK: {date}") time.sleep(2)

批量任务要保留日志，失败时不要静默跳过。建议把失败日期写入failed_dates.txt，后续重试。

8. 资源占用与性能观察

8.1 用 API 时的资源占用

纯 API 模式下，本地资源占用很低，不需要特别强的 GPU。主要成本是：

• Token 消耗：9 个模型多次调用，单日简报通常消耗明显高于单次文章生成。
• 接口延迟：多个串行调用会放大延迟。
• 并发限制：供应商的并发配额是瓶颈。

8.2 用本地模型时的显存估算

如果想完全本地运行，需要根据模型参数量估算显存：

• 7B 模型，使用 INT4 量化大约需要 6GB 左右显存，具体以实际版本为准。
• 13B 模型，INT4 量化大约需要 10GB 以上。
• 70B 模型，通常需要多卡或 CPU 内存池。

这不是精确数字，只是给一个判断思路。更准确的方法是启动模型后执行：

nvidia-smi

观察推理进程的显存占用，再根据你机器实际剩余显存做调整。

8.3 延迟观察

对于 9 模型流程，建议记录每个环节的耗时。一个常见现象是：过滤和采集阶段很快，宏观分析和多空辩论阶段最慢。

可以从两个方向优化：

• 把小模型和大模型分开：筛选用小模型，分析用大模型。
• 把无依赖环节并行：比如 macro、sector、risk 三个分析模型用ThreadPoolExecutor并行执行。

8.4 ComfyUI 与 LLM 是否必须同一台电脑

这是一个经常被问到的问题。如果你同时跑 ComfyUI 做图像生成，又跑 LLM 文本分析，两者不一定需要同一台电脑：

• 如果本地显存足够大，可以在一台机器上同时跑两个服务，但要注意显存竞争。
• 如果显存紧张，可以分开部署：一台 GPU 跑 ComfyUI，另一台或远程服务器跑 LLM，通过 HTTP API 互相调用。

实际操作中，判断标准是“单卡剩余显存能不能塞下模型”，而不是“必须在一台机器上”。

8.5 如何降低资源占用

• 使用小模型做前置过滤，减少进入大模型的信息量。
• 控制每轮输入的新闻条数，而不是把原始数据全部塞进 prompt。
• 开启日志压缩，只保留关键字段，不要把所有中间结果都落盘。
• 批量任务设置重试上限，避免卡住的进程持续占内存。

9. 常见问题与排查：这个方案会断在哪里

这是整个项目最有价值的部分。下面按故障概率从高到低列出。

10. 最佳实践：让 9 个模型更稳定地协作

10.1 每步都做结构校验

不要让模型输出“自由文本”再指望下一步能解析。每一步都要求输出 JSON，并用pydantic做校验。校验失败就重试或进入降级流程。

10.2 分层使用模型

不需要 9 个模型都是大模型。合理的分层策略是：

• 采集、过滤、合规检查：小模型即可。
• 宏观分析、多空辩论：大模型。
• 最终编辑：最强的模型。

10.3 保留审计日志

每个模型输出的原始内容、token 数、耗时、重试次数都要记录。没有日志，多模型流程出问题后几乎无法排查。

10.4 设置人工复核门

在合规检查通过之后，加一个“人工复核”状态。哪怕只是快速扫一眼标题和风险提示，也能避免很多责任问题。

10.5 批量任务加失败重试

批量任务不要一次跑到底。按日期分成独立任务，失败的任务写入单独的队列，全部完成后统一重试。

10.6 做模型降级预案

某个模型不可用时，不要直接让整个流程崩溃。可以降级为：

• 用规则模板代替 collector。
• 用另一个同能力模型替代。
• 跳过非关键分析环节，但必须在简报中标注“未完成分析”。

11. 总结与下一步

这个 9 模型委员会项目最值得尝试的点，不是“用 9 个模型生成了篇文章”，而是让你看到多模型协作的完整故障面。它能正常跑起来，靠的不是单个模型多聪明，而是每一步的结构约束、格式校验和错误处理。

如果你准备自己动手复现，建议按这个顺序推进：

1. 先从 3 个角色起步：collector、editor、compliance。
2. 跑通后再扩展成 5 个、9 个。
3. 每一步都要保留 JSON 校验和日志。
4. 第一次全流程运行前，先确认 API Key、模型名和输入数据日期是正确的。
5. 跑完看audit_log.json，哪一环耗时最长、哪一环出错最多，再针对性优化。

最容易踩的坑是：9 个模型全部串行调用，中间没有任何校验，某一环输出格式突变，后面所有环节全部连锁失败。

后续可以扩展的方向也很多：接入实时行情数据、增加图表生成环节、把结果自动发布到内部系统、用向量库做信息检索增强、对多空观点做历史胜率回测。

如果你对多 Agent 编排、结构化输出、批量内容生产感兴趣，这个项目是很好的练手模板。先把单日流程跑稳，再考虑批量、并发和上线发布。