企业官网建设流程全解析

MiroFish 这个项目，我是在一次彻底失败的 Agent 任务里撞上的。那天凌晨两点，我盯着屏幕上已经跑了四十分钟、却还在原地打转的执行日志，狠下心把原来那套方案整个推翻。任务说复杂也不复杂：把一份三十多页的调研材料拆成事实、口径、缺口三块，再交叉验证里面的数字是否自洽。我最初的方案朴素得有点自信过头——写一个足够长的 System Prompt，把所有规则塞进去，让一个模型从头干到尾。结果是上下文越滚越长，前面读到的数字到后面已经被它自己“记错”了两次，而且我根本不知道错在哪一步。

后来我把这套流程改成了多智能体协作：拆解的、取数的、验算的、写结论的，各管一段，彼此通过消息交互。MiroFish 就是在这个阶段进入我视野的，它把“鱼群协同”当成核心设计隐喻——没有一条鱼掌握全局，但整群鱼能完成迁徙、围捕、避险这类复杂动作。落到工程上，就是一组职责单一的 Agent 通过消息总线协作，靠局部信息和简单规则涌现出整体行为的编排框架。这篇东西写给两类人：一类是被单体 Agent 的上下文和稳定性折磨过、想上多智能体的；另一类是已经在用多智能体、但被成本、并发和排查难度劝退的。我会把这套东西的结构、调度、上线前的三道关，以及我自己踩过的坑一条条摊开讲。

1. 单体 Agent 撞墙之后，MiroFish 要解决的真问题

1.1 上下文膨胀、错误累积、无法定位：单体的三个死结

先说清楚为什么非得拆。单体 Agent 在任务规模小的时候非常香——一个 Prompt、一次调用、一个结果，调试起来毫无负担。但只要任务跨过某个临界点，三件事会几乎同时发生。第一是上下文膨胀：你要让它记住原始材料、中间结论、格式要求、历史轮次，窗口被塞满之后，模型对早期内容的注意力会明显衰减，表现就是“前面说过的它记不住”。第二是错误累积：一步算错，后面所有推理都建立在错误前提上，而且没有任何一道校验关卡能把它拦住。第三是定位困难：输出错了，你只能对着一整段对话回放，猜是哪一句带偏的，这个过程的成本高到让人放弃优化。

我做过一个粗略的对比。同一个“材料拆解+数字校验”的任务，单体方案在材料篇幅超过两万字之后，事实抽取的准确率会从九成出头掉到七成上下，而错误分布是没有规律的，有时错在取数，有时错在口径判断，有时压根是格式没对齐。这个数据不是模型能力问题，是架构问题——你把“记忆、推理、校验、格式化”四件事压在一段上下文里，它们必然互相干扰。

MiroFish 的思路是把这四件事拆开。负责抽取的 Agent 只看到材料片段，任务就是把这一段里出现的数字和口径抄下来；负责验算的 Agent 只看到被抽出来的数字表，任务是检查单位、口径、内部一致性；负责汇总的 Agent 只看到验算通过的结论列表，任务是组织语言。每一环的上下文都很短，每一步的输出都很窄，错在哪一步一目了然。

1.2 鱼群协同不是万能药：什么场景千万别上多智能体

这里必须先泼一盆冷水。多智能体不是升级，是换赛道，它用“架构复杂度”换“任务上限”。如果你的任务是单轮问答、文本改写、简单分类、固定模板生成，上多智能体纯属自找麻烦：延迟翻三倍，成本翻两倍，出问题的排查难度上一个量级，而效果提升几乎为零。我见过太多团队在单轮任务上硬堆五个 Agent，最后收益没看到，运维成本先上去了。

真正值得上的场景有这么几个特征：任务可以自然分解成职责不同的子任务；子任务之间存在需要传递和校验的中间结果；单次执行需要的上下文超出舒适窗口；或者任务本身带有“验证—修正”的循环属性。反过来，如果子任务之间没有实质的数据依赖，那它其实是个批处理问题，用不着 Agent 编排，上个并发队列就够了。判断标准很简单：把任务手工拆成流程图之后，如果节点之间要来回传话、要互相确认，那就是多智能体的活儿；如果是单向流水线，那就老老实实写代码。

1.3 它和工作流引擎的区别：没有写死的 DAG

另一个常被搞混的点是 MiroFish 这类框架和传统工作流引擎（比如那种拖拽画 DAG 的编排工具）的差别。工作流的执行路径是你在设计期定死的：A 完了走 B，B 失败走 C，条件分支写死在节点上。多智能体编排把一部分决策权挪到了运行时——下一步调谁、需不需要再补一轮验证、这个结果够不够格进入汇总，是由 Agent 根据当前状态判断的。

这个差别带来的好处是灵活，代价是不可预测。工作流出问题，你看着图就能定位；多智能体出问题，你得看 Trace 才能还原路径。所以我后来给自己定了一条规矩：能用固定流程解决的环节，一律退回固定流程，只在真正需要动态判断的地方留 Agent。一个典型的混合结构是：外层是写死的流水线（加载、分片、归档），中间嵌一层动态的 Agent 协作区（拆解、验证、补漏），出口再回到写死的格式化与落库。这样既拿到了灵活性，又把不可控面积压到最小。

2. 把“鱼群”翻译成代码：MiroFish 的四层结构

2.1 角色层：职责越窄，输出越稳

鱼群里每条鱼只做三件事：跟住邻居、避开障碍、别掉队。映射到 Agent 设计上，就是每个角色的职责必须窄到能用一句话说清。我给每个 Agent 写配置时有个硬性要求：如果它的 System Prompt 里出现“同时”“并且”“顺便”这类并列词超过两次，就说明该拆了。一个负责“从段落里抽取带单位的数值”的 Agent，比一个负责“分析这段材料的所有要点”的 Agent 输出稳定得多，原因很直接——任务边界清晰，模型的自由发挥空间小，幻觉产生的入口就少。

角色划分还有一个容易忽略的收益：可以给不同角色配不同量级的模型。抽取、分类、格式校验这类任务，小模型完全够用；只有最终汇总、冲突裁决这种需要全局判断的环节才值得上大模型。我在一个项目里把七个角色的模型配成“五小两大”，整体成本降了六成多，效果几乎没变。这个优化只有在角色拆得够细的前提下才做得出来。

2.2 感知层：每个 Agent 只拿该拿的那一块

这一层是多智能体能不能跑得动的关键。很多人第一次做多智能体，习惯把“共享记忆”当成一个大池子，所有 Agent 都能读全量历史。这是个灾难：上下文膨胀的问题原封不动地搬了进来，还多了一层。正确做法是给每个 Agent 定义明确的输入契约——它只能看到哪几个字段、这些字段从哪来、格式长什么样。

我通常把上下文分成三区：任务区（这个 Agent 要干什么，短而固定）、输入区（上游传过来的结构化数据，有长度上限）、事实区（共享状态里已经确认过的结论，只读）。三区之外的东西一律不注入。这样做的另一个好处是 Prompt 可以复用——同一个角色的 Prompt 模板，配上不同的输入区内容，就能处理成千上万条数据，不用为每条单独调。

2.3 通信层：消息总线与共享状态的分工

Agent 之间怎么说话，决定了系统会不会乱。我的做法是把通信拆成两条通道。消息通道传的是“请求和响应”，点对点、有生命周期、用完即弃，比如拆解 Agent 把任务单发给抽取 Agent。共享状态区存的是“已经确认的事实”，有明确的写入权限，只有通过校验的结果才能进去。这两条通道混在一起，是后面上下文污染类故障的根源。

共享状态区的写入必须带来源标记。每条事实记四个字段：内容、由谁写入、依据是什么、什么时间。这样一旦发现某条结论有问题，可以顺着来源一路回查到原始的模型输出。我吃过一次亏：某个 Agent 把“推测值”当“确认值”写进了共享区，后面三个 Agent 全都在这个错值上继续推理，最后结论整体偏了百分之十五，而排查花了整整一天。

2.4 收敛层：怎么判断“这群鱼已经到岸了”

多智能体最容易失控的地方就是没有明确的终止条件。两个 Agent 互相确认、互相补充，能友好地聊上一百轮，每一轮都在花钱。所以收敛条件必须是显式的、可计算的三类之一：目标达成（所需字段全部填满且通过校验）、预算耗尽（轮次或 Token 触顶）、增益停滞（连续两轮没有新增有效信息）。

提示：增益停滞这个条件一定要有。它是防止死循环的最后一道闸，而且实现成本极低——记录每轮的“新增事实数”，连续为零就熔断。

3. 跑通第一条鱼群链路：环境与最小骨架

3.1 环境里最容易被忽略的三件事

先说环境。第一件是 Python 版本和异步运行时。多智能体天然是并发的，同步调用会把整条链路串成一串糖葫芦，延迟直接叠加。我的环境固定在 3.11 及以上，用asyncio做事件循环，所有 Agent 调用都是异步的。注意不要混用同步的 HTTP 客户端，哪怕只有一个环节用了同步库，整个事件循环都会被它卡住。

第二件是密钥和时间管理。多智能体意味着多个并发请求，很容易撞上接口的速率限制。我会在环境变量里预留并发上限和重试次数的配置项，而不是写死在代码里，方便随时调。第三件是日志前置——很多人等程序出问题了才加日志，但多智能体的问题往往不可复现，事后加日志已经晚了。从第一天起就把每次 Agent 调用的输入、输出、耗时、Token 数落盘，后面省的时间不止十倍。

# 建议的目录结构 mirofish-demo/ ├── config/ │ ├── agents.yaml # 角色定义与模型配置 │ └── runtime.yaml # 并发、重试、预算上限 ├── agents/ │ ├── extractor.py # 抽取类角色 │ ├── verifier.py # 校验类角色 │ └── summarizer.py # 汇总类角色 ├── core/ │ ├── bus.py # 消息通道 │ ├── state.py # 共享状态区 │ └── orchestrator.py # 编排主循环 ├── traces/ # 每次运行的调用记录 └── main.py

3.2 写第一个 Agent：System Prompt 的三段式结构

角色定义落在agents.yaml里，我习惯用三段式写 System Prompt：身份 + 输入契约 + 输出契约。身份一句话，输入契约说明会收到什么字段，输出契约规定必须返回什么结构的 JSON。三段式的好处是模板化，改一个角色不用重写整段 Prompt。

# config/agents.yaml extractor: model: small-fast temperature: 0.1 system: | 你是一个数值抽取器。 输入: 一个文本片段 text。 输出: 严格返回 JSON，形如 {"items":[{"value":数字,"unit":字符串,"context":"原文短句"}]}。 规则: 只抽取显式出现的数值，不做任何换算，不推测缺失值。找不到则返回空数组。 max_output_tokens: 800 verifier: model: small-fast temperature: 0.0 system: | 你是一个口径校验器。 输入: 一组 items，每个含 value、unit、context。 输出: 严格返回 JSON，形如 {"issues":[{"index":下标,"type":"单位冲突|口径不明|数值异常","reason":"简述"}]}。 规则: 只报告问题，不修改原值。没有问题时返回空数组。

两个细节值得单独拎出来说。temperature 一定要压低，抽取和校验这类任务不需要创造性，0 到 0.1 之间最稳。输出契约里必须写明“找不到时返回什么”，否则模型会倾向于编一个看起来合理的结果塞进去，这是幻觉最隐蔽的入口。

3.3 编排主循环：下发、回收、失败重试

编排层干的活儿说穿了就是四步循环：挑出待处理的单元、并发下发、回收结果、更新共享状态。下面这段骨架阉掉了业务细节，但结构是完整的、可以直接照着改。

# core/orchestrator.py import asyncio from core.bus import Bus from core.state import SharedState class Orchestrator: def __init__(self, agents, cfg): self.agents = agents self.cfg = cfg self.bus = Bus() self.state = SharedState() self.max_rounds = cfg["max_rounds"] self.concurrency = cfg["concurrency"] self.budget = cfg["token_budget"] async def run(self, units): sem = asyncio.Semaphore(self.concurrency) async def handle(unit): async with sem: # 并发闸门 for attempt in range(self.cfg["retries"]): res = await self.agents["extractor"].call(unit, self.state.facts()) if res.ok: return res await asyncio.sleep(0.5 * (2 ** attempt)) # 指数退避 return res # 重试耗尽，返回最后一次结果并标记 round_no = 0 while round_no < self.max_rounds: round_no += 1 pending = self.state.pending_units() if not pending: break results = await asyncio.gather(*[handle(u) for u in pending]) added = self.state.merge(results) # 只有校验通过的事实才会被写入 if added == 0: break # 增益停滞，熔断 if self.state.tokens_used > self.budget: break return self.state.snapshot()

这段代码里有三个点，是我从踩坑里换来的。并发闸门用 Semaphore 而不是无限 gather，不然任务一多直接把接口打爆。重试用指数退避，因为多智能体场景下失败大多是限流，立刻重试只会雪上加霜。每轮结束后检查“本轮新增事实数”，为零就退出，这是 2.4 节说的收敛条件在代码上的落地。

4. 任务分片与调度：十几个 Agent 同时跑而不打架

4.1 三种切分方式，选错了后面全是坑

分片方式决定了整条链路的形状，选错的话后面的优化都是白费。我把常见的分法归成三类：按数据切（把长材料切成若干段，每段独立处理）、按能力切（抽取、校验、汇总各是不同角色）、按阶段切（先粗筛、再细验、最后归并）。实际项目里通常是三者混用。

按数据切最容易出问题的地方是跨界信息。一段材料被切成十块，某个数字的分母在第三块、分子在第七块，抽取 Agent 各自为战，谁也发现不了。我的处理办法是加一个“对齐角色”，它只看所有片段的抽取结果，任务是找“同一事物的不同表述”和“明显不成对的数据”。这个角色成本很低，但能捞回不少漏。

4.2 并发度与限流：一个参数错了整条链路雪崩

并发度这个参数，我调过很多次，经验是先压到很低，再慢慢往上加。原因在于多智能体的并发不是简单的一次调用，而是“扇出”——一个上游 Agent 的输出可能触发十几个下游调用。如果上游并发设成 10，下游每个再扇出 5，瞬时压力就是 50。这个时候接口返回的不是错误，而是变慢，而变慢会让超时开始出现，超时触发重试，重试又加重压力，雪崩就是这么来的。

我的做法是分两层限流。全局设一个总并发上限，卡住整条链路的峰值；每个角色再设一个独立上限，防止某个重角色吃光配额。另外一定要监控排队等待时长这个指标，它比错误率更早预警——错误率还是零的时候，等待时长可能已经在翻倍了。

4.3 竞态、重复劳动与幂等

并发的另一个麻烦是重复劳动。两个抽取 Agent 可能被分到内容重叠的片段，各自抽出一模一样的事实，写进共享状态时就变成了两条。表面上只是浪费，实际上会让后续的统计类角色重复计数，结论直接错。解决办法是给每条事实算一个指纹——把归一化后的内容做哈希，写入前查一遍指纹表，重复的直接丢弃。

竞态则更多出现在状态更新上。多个协程同时读共享状态、各自修改、再写回去，后写的会覆盖先写的。我在共享状态的所有写操作上都加了异步锁，只保护写入这一段，读操作不加锁（读的是不可变快照），这样性能损失很小，但一致性有了保障。这类问题在小规模测试里几乎不会出现，只有并发上去之后才会零星冒出来，所以别指望测试环境能帮你发现。

5. 上线前的三道关：成本、延迟与稳定性

5.1 成本是乘法的，先算账再动手

单体 Agent 的成本是加法，多智能体的成本是乘法。一次任务可能有 1 次拆解、12 次抽取、3 次校验、1 次汇总，如果每次调用平均消耗三千 Token，单次的账就是五万 Token 量级。我见过团队在本地跑得很开心，一上量发现账单直接超预算十倍。

控制成本我一般从三处下手。模型分级，前面说过，能小模型的地方绝不放大模型，这一项通常能省一半以上。输出截断，给每个角色的max_output_tokens设一个上限，尤其是抽取类角色，很多时候模型会自作主张写一段长篇解释，纯粹浪费。输入去重，同一份材料的不同片段里，背景段落往往重复出现，注入前做一次段落级去重，能省下不少输入 Token。这三项加起来，我在一个项目里把单次任务的成本从约 0.7 元压到了 0.12 元，效果评分没有下降。

5.2 缓存与复用：哪些中间结果值得存

不是所有中间结果都值得缓存，判断标准是“是否会被重复请求”。我通常缓存三类东西：原文片段的结构化版本（同一段材料被多个角色读取时复用）、工具调用结果（外部数据源查询的返回）、校验通过的结论集（后续追问或增量任务可以直接用）。

缓存键要设计得足够细，我一般是“角色 + 输入内容的哈希 + 模型版本”三者拼起来。这里有个坑：模型版本变了，缓存必须失效。我们有一次在切换模型之后忘了清缓存，结果新旧结果混在一起，出现了一批风格不一致的输出，排查了很久才想到是缓存的问题。

5.3 失败域隔离与降级

多智能体系统的失败是局部的，能不能扛住就看隔离做得好不好。我的原则是任何一个 Agent 挂掉，都不能让整条链路挂掉。

每条策略背后的逻辑都一样：宁可少给一点，不可给错。缺了可以标出来让下游知道，错了会污染整条链路。我在输出层加了一个“完整度”字段，取值是 0 到 1，表示必填字段的填充比例，下游业务根据这个值决定是直接用还是转人工，这个设计比强行凑满字段靠谱得多。

6. 联调期的五类典型故障与完整排查链路

6.1 死循环：两个 Agent 互相“确认”了一百轮

现象：日志里两个角色来回发消息，格式都对、内容都在推进，但轮次不断上涨，任务不结束。假设：收敛条件失效，或者终止条件的判定依赖的字段始终没被更新。验证：先看每轮的新增事实数，如果一直是零，说明状态根本没变；再看这两个角色的退出前置条件，是不是互相等待对方先满足。根因：校验角色要求“所有字段都有来源标注”，而抽取角色认为“来源标注由对方补”，两边互相等。修复：把来源标注的责任明确划给抽取角色，校验角色只做检查不做补全。验证：重新跑一遍，观察轮次从一百多降到三轮。

6.2 上下文污染：猜测被当成事实写进共享区

现象：结论中的某个数字稳定地偏离实际值，但每一段抽取结果单独看都没问题。假设：某处存在未经校验的写入。验证：顺着共享状态里那条错误事实的来源字段往回查，找到写入它的角色和当时的原始输出。根因：抽取角色在原文没有明确数值时，出于“补全”的本能填了一个估算值，而这个值绕过了校验直接进了共享区。修复：在输出契约里明确“找不到返回 null”，同时在共享区写入前加一道 schema 校验，空值不允许写入事实区，只能写入待确认区。验证：构造一批缺值样本，确认不再有估算值进入事实区。

6.3 长输出下的结构化解析崩溃

现象：短文本下一切正常，处理长材料时零星出现解析失败。假设：输出被截断，或者模型在长上下文下忘了格式要求。验证：把解析失败的那次原始输出完整打出来看，通常是尾部被max_output_tokens砍断，JSON 缺了收尾括号。根因：抽取数量超预期，输出长度估算不足。修复：把输入分片调小，单个片段限制在 800 字以内；同时给解析层加一次自动修复，尝试补齐缺失的括号和引号。验证：用同一批长材料重跑，解析失败率从百分之三降到千分之二以内。

6.4 超时、僵尸任务与资源泄漏

现象：任务跑了很久不结束，内存占用缓慢爬升不回落。假设：有协程被遗忘，没有设置总超时。验证：打印当前活跃任务数，对比预期的并发数。根因：asyncio.gather里某个协程抛异常被吞掉，导致主循环一直在等一个永远不会完成的 Future。修复：给每个调用加独立超时，用asyncio.wait_for包一层；gather 加return_exceptions=True，异常单独处理并计入失败。验证：注入一个必定超时的假任务，确认整体能在预期时间内退出且内存正常释放。

6.5 结果“看起来对”：离线评估集的必要性

这一类不算故障，但比故障更危险。多智能体的输出往往读起来很顺、逻辑自洽，很容易让人产生“成了”的错觉。我后来强制自己做两件事：固定一个五十到一百条的小评估集，每次改动前后都跑一遍，看关键字段的准确率；人工抽查二十条，重点看有没有那种“格式完美、内容错误”的样本。这两件事加起来大概占用半天时间，但拦下过好几次差点上线的回归问题。评估集不需要很大，关键在于固定不变，能形成前后对比。

7. 可观测性：没有 Trace 的多智能体就是黑盒

7.1 一次 Agent 调用必须落库的字段

我给自己定过一个清单，每次调用必须记下来，少一个字段都不行。

这套字段落下来之后，排查效率的提升非常明显。以前出问题只能靠复现，现在可以直接按 run_id 把整条链路拉出来，哪一步慢、哪一步贵、哪一步在重试，一眼就能看到。

7.2 用一张看板回答“钱花在哪、时间耗在哪”

数据落盘之后，我会做两张最简单的聚合表。成本表按角色分组，看每个角色的 Token 消耗占比——通常情况下，抽取角色因为调用次数多，占比会达到一半以上，优化它收益最大。延迟表把耗时拆成排队等待和执行两部分，如果排队占了大头，那就说明并发度设小了或者下游在限流；如果执行占大头，那就该看看输入是不是太长了。

有一次我盯着延迟表看了半天，发现某个校验角色的执行耗时是同类角色的四倍，查下去才发现它的输入区没有做去重，每次都被塞进了全量事实，输入长度是必要长度的六倍。这种问题不看聚合数据是根本发现不了的。

7.3 回放与回归：让每次改动都可追溯

最后一件我认为必须做的事是回放能力。把一次完整运行的输入快照存下来，之后任何时候都能用它重跑一遍，对比输出差异。这样调整 Prompt、换模型、改并发，都不用担心“改好了这个、改坏了那个”。我现在的习惯是每次提交前跑一遍固定的三条回放样例，十分钟出结果，比上线后发现问题再回滚便宜太多。

回放的时候要注意把随机性压住：temperature 设为 0，并且尽量固定模型版本。回放的目的不是验证“绝对确定性”，而是验证“行为没有意外变化”。只要输出在合理范围内，就算通过；如果出现了新的事实、新的字段、新的失败类型，那就得停下来看看。

8. 我个人在这套东西上的一点体会

用到现在，如果只让我留一句话给准备上多智能体的人，那就是：先把单体做扎实，再拆。拆分的收益来自“职责隔离”，而不是来自“Agent 数量”。我见过太多项目，把本该由一段代码完成的事情包装成一个 Agent，结果链路变长、成本翻倍、排查变难，最后还要花时间拆回去。判断标准其实很朴素——这件事需要模型来判断吗？不需要，那就写成函数。

另一个体会是，收敛条件和预算上限从一开始就要写进配置，别等出问题再加。它们是这套系统的安全带，平时感觉不到存在，但真出事的时候只有它们能救你。至于成本控制，别指望事后优化，模型分级和输出限制这两项在第一天做和第十天做，代价完全不一样。