一个人干不完的活加人就能干完——这句话在软件项目里基本成立放到多智能体系统里却经常翻车。我最早接触需要同时做资料检索、数据核对、长文撰写和事实校验的任务时图省事用一个 Agent 从头跑到尾结果很典型上下文越滚越长跑到第八轮左右就开始选择性失忆前面查到的关键数据在最终稿里变成了编的。后来我把任务拆给多个 Agent 协作问题没消失只是换了形态——它们会互相附和一个说错剩下几个跟着错最后一致性极高、正确率极低。MiroFish 这类以鱼群为核心隐喻的多智能体协作框架正是在这个背景下被反复拿出来讨论的。它关心的不是怎么让单个 Agent 更聪明而是怎么让一群并不特别聪明的 Agent通过局部规则和受控的信息流动整体表现出稳定、可解释、可复现的协作行为。鱼群没有总指挥每条鱼只看得见身边几条鱼但整个鱼群能转向、能避敌、能保持队形——这套机制挪到 Agent 编排上就是多智能体、群体智能、协作编排、去中心化调度这几个关键词背后真正的工程问题。这篇内容适合三类人看正在做多 Agent 编排、被上下文和成本折磨的工程师想把一个长链路任务拆成协作流程的产品和算法同学以及单纯想知道群体智能到底是不是玄学的技术爱好者。下面我会把这类框架的骨架拆开讲清楚每个设计决策背后的取舍再给一套能直接跑起来的最小协作实现最后把我踩过的坑原样摊开。文中涉及的接口形态和参数是基于常见工程实践补全的示意性写法具体版本请以你手上仓库的实际定义为准。1. MiroFish 这类鱼群式多智能体到底在解决什么1.1 单 Agent 的三条硬天花板单 Agent 跑长任务撑不住的地方其实很集中无非三条。第一条是上下文预算的物理上限。一个需要读 15 份文档、调用 8 次外部接口、产出 3000 字报告的任务工具返回的原始数据往往就有几十万 token。你以为自己在做推理实际上大部分算力花在把已经看过的东西再看一遍上。压缩、截断、摘要都会丢信息而丢的往往正是关键那一条。第二条是错误无法自我纠正。单个 Agent 一旦在第三轮形成了一个错误假设后面所有推理都会围绕这个假设展开——这是典型的自回归陷阱它没有外部视角来质疑自己。你让它再检查一遍它检查的还是同一套逻辑。第三条是串行导致的时间成本。检索、计算、写作、校验这四个环节本来互不依赖单 Agent 只能排队做原本能并行 4 分钟的活硬生生跑成 16 分钟。1.2 鱼群隐喻的三层含义MiroFish 这个名字里的鱼群不是装饰它对应三个非常具体的工程决策。局部规则优先于全局规划。鱼群不需要一张全体转向 30 度的指令表每条鱼只需要遵守跟前面邻居保持一臂距离、不撞、朝平均方向偏一点这三条局部规则群体行为就涌现出来了。映射到 Agent 系统与其让一个中心调度器精确编排每一步不如给每个 Agent 一套简单的行为准则和有限的可见范围让整体行为从局部交互中长出来。好处是扩展性极强坏处是可预测性差——这一点后面会重点讲怎么补。群体涌现优于单体最优。鱼群里没有一条鱼是最优路径规划器但鱼群整体的避敌成功率高于任何单条鱼。对应到 Agent不要指望某个角色写得像专家而是让多个视角互相校验靠多数 异见逼近正确答案。队形随环境动态变化。遇到捕食者时鱼群会收紧平时则松散。这不是预设策略而是规则强度随环境变化的自然结果。工程上就是动态拓扑——任务简单时用星型集中调度任务复杂时切到分层结构出错了再退化成全连接广播做交叉验证。注意很多人一听去中心化就兴奋觉得可以扔掉调度器。实际工程里恰恰相反——去中心化的 Agent 行为需要更精细的中心化观测和熔断机制否则你连它为什么跑偏都不知道。1.3 它不适合谁先把边界划清楚省得你两周后骂人。它不适合延迟敏感的单轮任务。问一句话答一句话的场景上多 Agent 就是自找麻烦一次调用能搞定的事情拆成五次成本和延迟都翻倍。它也不适合强确定性流程。如果每一步的输入输出格式都固定、分支有限用工作流引擎或者普通的函数编排就好别用自然语言 Agent 去做流程图能做的事——那等于用大炮打蚊子还打不准。它真正发光的场景是任务边界模糊、需要多种视角、信息量超过单次上下文、结果需要交叉验证。比如竞品调研、长报告撰写、代码审查、多源信息汇总、方案评审。这些任务的共同点是单 Agent 做出来能看但不敢用。2. 把骨架拆开这类框架绕不开的四个部件2.1 角色定义与能力声明多 Agent 系统的第一件事是把角色写清楚。但写清楚不是写一段花哨的人设文案而是要声明三样东西能力边界、可见信息范围、输出契约。能力边界指的是这个角色能调用哪些工具、能处理哪类输入。可见信息范围决定了它能读到黑板上的哪些字段——这一点极其关键如果所有 Agent 都能看见所有信息那你就退化回了一个超长上下文的单 Agent只是把上下文切成了几段。输出契约则是要求结构化输出通常是 JSON 或者带固定标题的 Markdown方便下游解析。一份典型的角色声明长这样AGENT_SPEC { name: fact_checker, goal: 对候选结论逐条核验输出支持/反驳/存疑三态判断, tools: [web_search, doc_lookup], visible_fields: [draft.claims, evidence.pool], # 只看结论和证据池 max_steps: 6, output_schema: { claim_id: str, verdict: enum[supported, refuted, uncertain], reason: str, source: str|null } }注意visible_fields这个字段。它是控制上下文规模最有效的一把刀比任何摘要算法都管用——不给它看它就不会把无关信息带进上下文。2.2 调度器谁来决定下一步谁说话调度是这个领域最容易被低估的部分。常见的四种做法差别很大。轮询式最简单按固定顺序让每个 Agent 说一轮说完就下一轮。优点是绝对可复现缺点是浪费——有些角色这一轮完全没事干也被迫调用一次模型。信号驱动式是鱼群隐喻的正统实现Agent 只有在感知到与自己相关的状态变化时才被唤醒。工程上通常给黑板加一个版本号或者事件主题Agent 订阅感兴趣的主题。这个方案省算力但需要你仔细设计订阅关系否则会出现该醒的没醒导致流程卡死。管理者委派式最可控一个 Manager Agent 负责拆解任务、分派给合适的 Worker、收集结果。上手快但 Manager 本身会成为上下文瓶颈和单点故障。混合式是我实测最舒服的顶层用管理者做任务分解只分解不执行中间层用信号驱动做并行执行最后用固定流程做汇总和校验。既保留了可控性又把长上下文压力分散开了。2.3 共享上下文黑板模式与消息传递这两种模式经常被混着讲其实差别很本质。消息传递是我发给你、你发给他信息在 Agent 之间接力流动。问题是信息容易在传递中失真而且谁掌握了什么很难追踪。黑板模式是所有 Agent 读写同一块共享存储谁写了什么、什么时候写的、被谁读过全部有记录。MiroFish 这一类框架通常偏向黑板——因为鱼群隐喻要的正是共享环境、局部感知而不是点对点打电话。黑板的结构大致分三层原始层存工具返回的未加工数据体积大不直接进上下文提炼层存摘要和结构化中间结论小进上下文决策层存最终结论和共识状态。Agent 默认只读提炼层和决策层需要溯源时才按引用 ID 去原始层取片段。这个分层设计带来的收益很直接假设原始数据有 200 万 token提炼层可能只有 2 万 tokenAgent 的上下文压力直接降两个数量级。2.4 记忆与状态别把记忆当成第二个上下文很多人做的长期记忆其实就是把历史对话全塞回去那不叫记忆叫上下文堆积。实用的分层是这样工作记忆是当前轮次的临时状态轮次结束即清理任务记忆是本次任务的中间产物任务结束归档长期记忆是跨任务沉淀的经验比如这个数据源经常延迟更新这类问题上次的正确解法是什么通常用向量检索按需召回而不是全量注入。一个细节长期记忆写入前一定要做去重和抽象。我见过一个系统跑了三周长期记忆里存了四百多条几乎一样的用户偏好简洁回复检索时全部命中直接把上下文挤爆。抽象成一条用户偏好简洁回复命中 400 次最近一次 3 天前就够用了。3. 上下文膨胀与回声室多智能体最容易翻车的两件事3.1 上下文膨胀的算术题先算一笔账你会立刻明白为什么很多多 Agent 项目跑着跑着就烧穿了预算。假设系统里有 12 个 Agent每轮每个 Agent 输出 800 token采用最简单的全连接广播每个人的输出发给其他所有人。那么单轮里每个 Agent 新增收到的信息是 11 × 800 8800 token。跑 10 轮光是新增的对话历史就是 8800 × 10 88000 token。再加上系统提示约 600 token、角色声明约 400 token、工具返回的提炼层数据假设每轮 3000 token单次调用的输入量会逼近 88000 4000 × 10 ≈ 128000 token。更糟的是调用次数也在涨12 个 Agent × 10 轮 120 次调用。如果每次调用平均输入 10 万 token前期小、后期大总输入量就是千万级 token。而如果改用提炼层 可见字段过滤的方案每个 Agent 每轮只读推理相关的 3000 token 提炼数据加上自身角色约 1000 token单次调用输入约 4000 token120 次调用总量约 48 万 token。相差二十倍以上效果往往还更好因为噪声少了。所以优化上下文的第一优先级不是用更便宜的模型而是别让不该进上下文的东西进来。3.2 回声室为什么一群 Agent 会集体犯错这是多 Agent 最隐蔽也最致命的问题。机制是这样的多个 Agent 用同一个底座模型给它们不同的系统提示表面上角色不同实际上它们的先验偏好高度一致。第一个 Agent 提出一个看起来合理的结论第二个 Agent 在上下文里看到这个结论倾向于顺着补充而不是反驳第三个看到前两个都同意就更不会反对。三轮之后全票通过。这叫自我强化偏差也有人叫回声室。它在指标上表现得特别有欺骗性——一致性从 60% 涨到 95%正确率却从 70% 掉到 45%。如果你只看共识度这个指标会以为系统变好了。对抗手段有三条我试过都有效引入结构性异见。强制保留一个反方角色它的系统提示里明确要求你的任务是否定前述结论至少给出一个反例或一个未被考虑的假设。不要指望它自觉要用提示词硬约束。让 Agent 独立作答再聚合。在形成初步结论前禁止任何 Agent 看到别人的输出。等所有人给出独立判断后再做一次聚合。这一步会让第一轮成本增加但能把从众效应砍掉一大半。多模型混用。如果条件允许让不同角色用不同的底座模型。不同模型的自回归偏好差异明显天然带来观点多样性。这条是成本最高的但效果最实在。3.3 一条可复现的排查链路当你的系统表现不对劲比如输出很流畅但事实错误率高可以按这个顺序查不要跳步。第一步把一次完整运行的输入输出落盘。没有日志什么都查不了。记录每个 Agent 每次调用的完整输入、输出、耗时、token 数、工具调用参数。第二步看 token 曲线。如果单次调用输入在后几轮突然陡增基本可以确认是上下文膨胀去看是谁在往共享状态里写大对象。第三步看信息流向图。把谁读了谁写的画成有向图。如果出现某个 Agent 的输出被所有人高频引用而它自己几乎不读别人的输出那它就是一个权威节点错误会被放大。这类节点要么给它配上专门的校验者要么降低它的广播范围。第四步做去从众实验。把某个角色连续 5 次运行的输入里别人给它的上下文全部清空看它的判断是否会变。如果变化很大说明它不是在做独立判断只是在附和。第五步定位到具体轮次和具体角色。到这一步问题通常已经能精确定位——往往是某个角色的visible_fields配得太宽或者某个工具的返回没有做裁剪。4. 从零搭一个最小可跑的协作骨架真要把这套思路跑起来你不一定需要完整框架。下面这套等价实现用标准库加 asyncio 就能跑二十来行核心逻辑理解了它再去看任何多 Agent 框架的源码都会轻松很多。4.1 目录结构与依赖swarm/ blackboard.py # 黑板分层共享状态 scheduler.py # 信号驱动调度器 agents.py # 角色定义与执行 tools.py # 工具注册 run_logs/ # 落盘日志 main.py只依赖 Python 3.10 标准库。真实项目里把agents.py里的模型调用换成你的 SDK 即可。4.2 黑板带版本号和字段过滤的共享状态import time from dataclasses import dataclass, field dataclass class Blackboard: # 三层raw 体积大不进上下文refined 进上下文decisions 是共识结果 raw: dict field(default_factorydict) refined: dict field(default_factorydict) decisions: dict field(default_factorydict) versions: dict field(default_factorydict) # 字段 - 版本号 log: list field(default_factorylist) def _bump(self, key: str): self.versions[key] self.versions.get(key, 0) 1 self.log.append({key: key, v: self.versions[key], ts: time.time()}) def write_raw(self, key, value): self.raw[key] value self._bump(fraw.{key}) def write_refined(self, key, value): self.refined[key] value self._bump(frefined.{key}) def write_decision(self, key, value): self.decisions[key] value self._bump(fdecision.{key}) def read_visible(self, visible_fields: list[str]) - dict: 只返回该角色声明可见的字段这是控制上下文的第一道闸门 out {} for f in visible_fields: scope, _, name f.partition(.) bucket {raw: self.raw, refined: self.refined, decision: self.decisions}.get(scope) if bucket and name in bucket: out[f] bucket[name] return out def snapshot_version(self) - dict: return dict(self.versions)关键在read_visible——它强制每个角色只能拿到自己声明过的字段。这一道闸门比任何摘要策略都有效因为它从源头上切断了噪声。4.3 调度器只在相关状态变化时唤醒import asyncio class Scheduler: def __init__(self, board: Blackboard, max_rounds: int 12): self.board board self.max_rounds max_rounds self.agents [] # 每个 agent 带 watches 列表 self.seen {} # agent.name - 上次看到的版本快照 def register(self, agent): self.agents.append(agent) self.seen[agent.name] {} def _relevant_change(self, agent) - bool: cur self.board.snapshot_version() prev self.seen.get(agent.name, {}) changed any(cur.get(k) ! prev.get(k) for k in agent.watches) if changed: self.seen[agent.name] cur return changed async def run(self, task): self.board.write_refined(task, task) for r in range(self.max_rounds): awake [a for a in self.agents if self._relevant_change(a)] if not awake: break # 没有任何相关变化收敛退出 results await asyncio.gather( *[a.step(self.board) for a in awake], return_exceptionsTrue ) for a, res in zip(awake, results): if isinstance(res, Exception): self.board.write_refined(ferror.{a.name}, str(res)) return self.board.decisions三点值得注意。watches让 Agent 只订阅自己关心的字段版本避免了轮询式调度里那些无事可做也被调用一次的浪费。max_rounds是硬熔断防止 Agent 互相客套停不下来。asyncio.gather配合return_exceptionsTrue保证单个 Agent 崩了不会拖垮整轮错误被写进黑板供后续处理。4.4 角色声明与一次执行class Agent: def __init__(self, name, watches, visible_fields, step_fn): self.name name self.watches watches # 订阅的版本键 self.visible_fields visible_fields # 可见字段 self.step_fn step_fn async def step(self, board: Blackboard): ctx board.read_visible(self.visible_fields) out await self.step_fn(ctx) # 换成你的模型调用 if out.get(decision): board.write_decision(f{self.name}.result, out[decision]) if out.get(refined): board.write_refined(f{self.name}.note, out[refined]) return out把step_fn写成调用模型的函数ctx就是拼进提示词的上下文out用结构化约束解析。整个链路跑起来后你会立刻发现一个现象大部分调优时间不花在提示词上而是花在watches和visible_fields的配置上。这两个字段决定了信息流拓扑拓扑不对提示词写得再漂亮也白搭。5. 拓扑、并发与成本的三角取舍5.1 五种通信拓扑的实测对比拓扑通信复杂度上下文压力容错性适用场景全连接广播O(n²)极高高Agent 数量 ≤ 4需要密集交叉验证星型集中调度O(n)中低中心单点流程明确、角色固定的任务环形接力O(n)低中逐轮迭代打磨类任务如文稿修订分层组长-组员O(n log n)中中高角色多且可分组的复杂任务黑板信号驱动取决于订阅关系低高大部分中大型协作场景选型的判断顺序是这样的。先看角色数量超过 6 个基本要排除全连接再看任务是否需要密集交叉验证需要就至少保留一个小范围的全连接子图最后看是否有天然的分组结构有就上分层。我自己的默认配置是上层分层 组内黑板信号驱动 关键节点对做全连接校验。这个组合在大多数场景下表现稳定。5.2 并发和限流怎么定并发不是越大越好。两个约束卡着你模型端的速率限制以及你自己后端的资源。实践中我会这样设置并发上限 min(模型速率限制的 70%, Agent 数量 × 2)。留 30% 余量是给重试和突发用的不限流的话遇到瞬时高峰会出现大批 429重试风暴反而更慢。超时方面单个 Agent 步骤设 60 到 120 秒整轮设 300 秒。超时的 Agent 直接标记为失败并写进黑板不要让整个流程 hang 住。失败重试最多 2 次且第二次要用指数退避间隔分别约 2 秒和 8 秒。还有一个容易忽略的点并发写入共享状态必须加锁。asyncio里虽然不会出现多线程那种内存撕裂但读—改—写的竞态依然存在。上面骨架里我把write_*设计成幂等的单键赋值就是为了绕开这个坑。如果你的逻辑需要读当前值再追加一定要用asyncio.Lock。5.3 成本估算怎么写别用具体价格算用相对单位更实用。给每次调用记两个数输入 token 数I和输出 token 数O。总成本近似为Cost ≈ Σ(I_i × p_in O_i × p_out)其中p_in和p_out是输入输出单价。真正需要控制的是ΣI_i因为输入占了绝大多数调用成本的 80% 到 95%。优化优先级从高到低收窄visible_fields通常能砍掉 50% 到 80% 的输入。分层黑板 摘要把原始数据留在 raw 层只让提炼层进上下文。区分模型把核查、分类这类简单角色换成小模型实测能省 40% 到 60% 总成本效果差异很小。缓存稳定前缀系统提示和角色声明这部分内容每次调用都一样利用提示缓存能省下相当可观的一部分。降低轮次收敛判据做得好平均轮次能从 10 降到 6这是最直接的收益。5.4 评测集怎么建别自己骗自己多 Agent 系统的评测最容易自欺欺人拿几个看起来很顺的案例截图就说效果好。我的做法是建一个 30 到 50 条的小评测集每条包含任务描述和可自动判定的标准答案或评分规则。评分规则可以是字符串匹配、数值容差、关键点覆盖数实在不行用另一个模型做打分但要固定评分提示词。必看的四个指标任务成功率最终产物是否满足要求这是唯一不能妥协的指标。共识度Agent 之间结论的一致程度。这个指标单独看会误导必须和正确率一起看。如果共识度涨而正确率跌说明回声室出现了。平均轮次反映收敛效率。轮次突然上升通常是某个 Agent 陷入循环去看它的watches是不是漏配了。单任务 token 消耗反映成本。这个数字涨得比成功率快说明优化方向错了。评测要跑三遍取平均因为模型输出有随机性单次结果没有参考价值。6. 我在多智能体项目里踩过的具体坑6.1 把角色扮演当成了能力隔离这是我犯的第一个大错。我以为给五个 Agent 写五段不同的系统提示它们就会从五个不同角度思考。实际测试下来同一个底座模型换提示词带来的视角差异远小于我的预期。最典型的证据是我让三个 Agent 独立评估同一个方案三个都给了正面评价理由高度雷同连用词都差不多。后来我把其中一个角色的模型换掉立刻出现了不同意见而且是有价值的反对意见。结论很直白提示词能改变输出格式和关注点改不了模型的底层先验。要真正的多样性要么换模型要么在角色里硬编码必须找出 N 个问题这类强制约束。6.2 不给终止条件的调度器会无限客套早期版本我让 Agent 自由对话直到达成共识。跑起来之后发现它们会陷入一种诡异的循环A 说分析和我说的一致B 说感谢确认我补充一点C 说很好综上。解决办法有两个建议都上。一是硬熔断max_rounds必须设我一般设 12 轮。二是状态无关性检测如果连续两轮黑板版本号没有变化直接退出。后者比熔断更优雅因为它是在真正的稳态下退出的而不是被动截断。6.3 工具返回的原始数据把上下文撑爆有一个查询接口返回的是嵌套很深的 JSON单次响应大概两万 token。四个 Agent 各调一次上下文直接爆掉。修复思路是在工具层做减法而不是在 Agent 层做摘要。具体做法包装一层工具适配器把原始响应存进raw层只提取 Agent 真正需要的 5 到 8 个字段写进refined层并附上raw层的引用键。Agent 如果需要细节可以显式调用一次按引用取片段的工具。这个改动把单次工具的上下文占用从两万 token 降到约 300 token而且没有丢信息——信息还在只是不主动塞给不需要它的 Agent。6.4 并发写共享状态造成的静默覆盖有一段时间系统偶尔出现某个结论凭空消失。查了很久才发现是两个并发执行的 Agent 都在做读取当前列表、追加一条、写回的操作后写的把先写的覆盖了。asyncio单线程不会导致内存错误但读—改—写这三步之间会发生协程切换竞态照样存在。修复方式就是加asyncio.Lock或者更彻底一点把共享状态改成 append-only 的事件流所有 Agent 只追加不覆盖聚合结果在读取时现算。我后来选择了 append-only 方案因为它顺带解决了另一个问题完整的事件流本身就是最好的调试日志。6.5 没有落盘日志出问题只能靠猜这个坑最朴素也最要命。项目早期我只看终端输出出了问题时想看三天前的运行轨迹什么都没有。现在我的标准做法是每次运行生成一个独立的run_id目录里面存三样东西。完整调用记录每次请求的输入、输出、模型名、token 数、耗时黑板事件流谁在什么时刻写了哪个字段的哪个版本拓扑快照本次运行的 watches 和 visible_fields 配置。三样东西加起来不到 10MB但能在出问题时五分钟内定位到具体轮次和具体角色。最后再分享一个我用了很久的小技巧新加一个角色时先在评测集上单独跑它不接入协作流程。看它单独工作时的输出质量和格式稳定性。如果一个角色单独跑都不稳定接进多 Agent 流程只会把问题放大——你在协作层面是修复不了单点质量的。这个习惯帮我在角色层就拦掉了大部分问题省下的排查时间非常可观。后面如果你想继续往下走可以试着给黑板加一层冲突检测当两个 Agent 对同一个字段写出矛盾结论时自动触发一次专门的仲裁流程这比等到最终产物才发现分歧要划算得多。