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 完了走 BB 失败走 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 触顶、增益停滞连续两轮没有新增有效信息。收敛类型判定条件触发后的动作目标达成必填字段校验全通过进入汇总正常退出预算耗尽轮次或 token 超阈值输出当前最优结果并标记不完整增益停滞连续 2 轮无新事实写入终止循环转人工或降级输出提示增益停滞这个条件一定要有。它是防止死循环的最后一道闸而且实现成本极低——记录每轮的“新增事实数”连续为零就熔断。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.py3.2 写第一个 AgentSystem 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 挂掉都不能让整条链路挂掉。失败类型表现处理策略单次调用超时某个片段无返回重试两次仍失败则标记该片段为缺失格式解析失败返回非 JSON走一次带约束的重问仍失败则丢弃该条校验不通过字段缺失或冲突回退给上游补抽最多补一轮整体预算耗尽触顶退出输出阶段结果明确标注完整度每条策略背后的逻辑都一样宁可少给一点不可给错。缺了可以标出来让下游知道错了会污染整条链路。我在输出层加了一个“完整度”字段取值是 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_exceptionsTrue异常单独处理并计入失败。验证注入一个必定超时的假任务确认整体能在预期时间内退出且内存正常释放。6.5 结果“看起来对”离线评估集的必要性这一类不算故障但比故障更危险。多智能体的输出往往读起来很顺、逻辑自洽很容易让人产生“成了”的错觉。我后来强制自己做两件事固定一个五十到一百条的小评估集每次改动前后都跑一遍看关键字段的准确率人工抽查二十条重点看有没有那种“格式完美、内容错误”的样本。这两件事加起来大概占用半天时间但拦下过好几次差点上线的回归问题。评估集不需要很大关键在于固定不变能形成前后对比。7. 可观测性没有 Trace 的多智能体就是黑盒7.1 一次 Agent 调用必须落库的字段我给自己定过一个清单每次调用必须记下来少一个字段都不行。字段用途run_id / round_no串联整条链路定位轮次agent_name / model知道是谁、用什么模型算的输入摘要哈希 长度判断输入是否重复估算规模原始输出截断存储出问题时能还原现场prompt_tokens / completion_tokens成本归因耗时排队 执行延迟归因状态成功/解析失败/超时失败率统计重试次数判断链路健康度这套字段落下来之后排查效率的提升非常明显。以前出问题只能靠复现现在可以直接按 run_id 把整条链路拉出来哪一步慢、哪一步贵、哪一步在重试一眼就能看到。7.2 用一张看板回答“钱花在哪、时间耗在哪”数据落盘之后我会做两张最简单的聚合表。成本表按角色分组看每个角色的 Token 消耗占比——通常情况下抽取角色因为调用次数多占比会达到一半以上优化它收益最大。延迟表把耗时拆成排队等待和执行两部分如果排队占了大头那就说明并发度设小了或者下游在限流如果执行占大头那就该看看输入是不是太长了。有一次我盯着延迟表看了半天发现某个校验角色的执行耗时是同类角色的四倍查下去才发现它的输入区没有做去重每次都被塞进了全量事实输入长度是必要长度的六倍。这种问题不看聚合数据是根本发现不了的。7.3 回放与回归让每次改动都可追溯最后一件我认为必须做的事是回放能力。把一次完整运行的输入快照存下来之后任何时候都能用它重跑一遍对比输出差异。这样调整 Prompt、换模型、改并发都不用担心“改好了这个、改坏了那个”。我现在的习惯是每次提交前跑一遍固定的三条回放样例十分钟出结果比上线后发现问题再回滚便宜太多。回放的时候要注意把随机性压住temperature 设为 0并且尽量固定模型版本。回放的目的不是验证“绝对确定性”而是验证“行为没有意外变化”。只要输出在合理范围内就算通过如果出现了新的事实、新的字段、新的失败类型那就得停下来看看。8. 我个人在这套东西上的一点体会用到现在如果只让我留一句话给准备上多智能体的人那就是先把单体做扎实再拆。拆分的收益来自“职责隔离”而不是来自“Agent 数量”。我见过太多项目把本该由一段代码完成的事情包装成一个 Agent结果链路变长、成本翻倍、排查变难最后还要花时间拆回去。判断标准其实很朴素——这件事需要模型来判断吗不需要那就写成函数。另一个体会是收敛条件和预算上限从一开始就要写进配置别等出问题再加。它们是这套系统的安全带平时感觉不到存在但真出事的时候只有它们能救你。至于成本控制别指望事后优化模型分级和输出限制这两项在第一天做和第十天做代价完全不一样。