多 Agent 协作系统设计:从拓扑结构到一套能跑起来的编排器

做了几个 Agent 项目之后,我越来越确信一件事:单 Agent 能解决的问题,比大多数人想象的要少。一个挂满工具的大模型,看起来无所不能,但只要任务链路一长,它就开始翻车——上下文塞爆、工具调用顺序错乱、前面的结论到后面自己忘了。这不是模型不够聪明,而是我们把一个本该拆给一个团队做的事,硬塞给了一个人。多 Agent 系统的本质,就是把一个全能选手换成一支有分工的团队。这篇文章不谈虚的,我想把多 Agent 协作里真正决定成败的几个设计点讲清楚,最后给一套不依赖任何重型框架、能直接复制运行的编排器代码。一、为什么要拆成多个 Agent先说清楚什么时候不该上多 Agent。如果你的任务能用一次 prompt 几个工具搞定,那就别折腾,多 Agent 只会增加延迟和成本。真正需要拆分的信号通常是这几个:任务本身有清晰的阶段,比如先调研,再写初稿,再审校,每个阶段需要的能力和提示词差异很大;单次上下文装不下,需要有人专门负责筛选信息、压缩中间结果;需要不同的人格或权限,比如一个负责生成、一个负责挑刺,让它们互相制衡比让同一个模型既当运动员又当裁判更可靠。最后这一点我体会最深。让同一个 Agent 自己写完自己审,它几乎总是给自己打高分。把审校单独拆出来、用不同的系统提示词,质量立刻就上去了。二、四种协作拓扑,先想清楚结构再写代码多 Agent 系统设计,80% 的功夫在拓扑选择上。选错了结构,后面写再多代码都是在补窟窿。常见的就四种:Show Image1. 编排器–执行器(Orchestrator-Worker) 一个中心 Agent 负责规划和调度,把子任务分发给专职执行器,再汇总结果。这是生产环境里最稳、最可控的结构。调度逻辑集中在一处,出了问题好排查。我自己的项目里九成都是这套——我们做企业级 Agent 平台(Bizfocus ADP)时,默认的编排骨架就是它。2. 流水线(Pipeline) 固定的处理链路,上一个 Agent 的输出是下一个的输入。适合阶段明确、不需要动态决策的场景,比如清洗→抽取→结构化。简单可靠,但不灵活。3. 对等协作(Peer / Group Chat) 多个 Agent 在一个共享会话里自由发言,像开会一样。表达力最强,但也最容易失控——它们可能陷入互相恭维的死循环,或者跑偏。没有强终止机制别轻易上这个。4. 层级(Hierarchical) 编排器之下还有子编排器,适合超大型任务。但层级一深,调试成本指数级上升,我的建议是能两层解决就别上三层。选型经验: 任务流程固定就用流水线;需要根据中间结果动态决策就用编排器–执行器;只有当任务高度开放、确实需要头脑风暴时才考虑对等协作。三、Agent 之间怎么说话:共享状态优于直接传话Agent 协作绕不开通信。我见过不少实现把上一个 Agent 的完整输出原封不动塞给下一个,结果上下文越滚越大,到链路末端早就爆了。更干净的做法是黑板模式(Blackboard):所有 Agent 不直接对话,而是读写一块共享状态。每个 Agent 只取自己需要的字段,只写自己负责的产出。好处有三个:上下文可控,谁也不用背整条历史;产出有结构,后续好做校验和回溯;调试时把黑板打印出来,整个协作过程一目了然。这套思路下面的代码里会直接体现。四、上代码:一套能直接跑的协作编排器下面这套实现刻意不依赖 LangGraph、AutoGen 之类的框架,就用标准库,目的是把协作的骨架暴露出来——很多人用了框架,却说不清里面到底发生了什么。场景:输入一个主题,产出一篇结构清晰的短文。拆成四个角色:规划 → 检索 → 写作 → 审校,审校不通过就打回重写,带最大轮次保护。为了让你不配 API Key 也能直接跑通流程,我加了一个 mock 兜底,配了OPENAI_API_KEY就走真实模型。pythonimport os import json from dataclasses import dataclass, field from typing import Optional # ---------- LLM 调用层:有 Key 走真实模型,没 Key 走 mock ---------- def call_llm(system: str, user: str) - str: api_key os.getenv(OPENAI_API_KEY) if not api_key: return _mock_llm(system, user) from openai import OpenAI # pip install openai client OpenAI(api_keyapi_key) resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: system}, {role: user, content: user}, ], temperature0.3, ) return resp.choices[0].message.content.strip() def _mock_llm(system: str, user: str) - str: # 离线占位,仅用于跑通流程;配置 OPENAI_API_KEY 后自动切换真实模型 role system.split(。)[0] if 审校 in role: return json.dumps({pass: True, comment: 结构清晰,通过}, ensure_asciiFalse) return f[{role} 的模拟产出] 基于:{user[:30]}... # ---------- 黑板:Agent 之间唯一的共享状态 ---------- dataclass class Blackboard: task: str artifacts: dict field(default_factorydict) def context(self, *keys) - str: 只取需要的字段,避免上下文膨胀 picked {k: self.artifacts.get(k, ) for k in keys} return json.dumps(picked, ensure_asciiFalse, indent2) # ---------- Agent 基类:角色 系统提示词 ---------- dataclass class Agent: name: str role_prompt: str def run(self, instruction: str) - str: return call_llm(self.role_prompt, instruction) # ---------- 编排器:调度 审校循环 终止保护 ---------- class Orchestrator: def __init__(self, max_revisions: int 2): self.max_revisions max_revisions self.planner Agent(规划, 你是规划 Agent。把主题拆成 3 个写作要点,逐行输出。) self.searcher Agent(检索, 你是检索 Agent。针对要点给出关键事实,简洁罗列。) self.writer Agent(写作, 你是写作 Agent。根据要点和事实写一篇 300 字短文。) self.reviewer Agent(审校, 你是审校 Agent。判断短文是否合格,只返回 JSON: {pass: bool, comment: str}) def run(self, topic: str) - Blackboard: board Blackboard(tasktopic) # 1. 规划 board.artifacts[outline] self.planner.run(f主题:{topic}) # 2. 检索 board.artifacts[facts] self.searcher.run( f要点:\n{board.context(outline)}) # 3. 写作 审校循环(关键:有上限的反馈闭环) feedback for attempt in range(self.max_revisions 1): draft self.writer.run( f主题:{topic}\n素材:\n{board.context(outline, facts)}\n f上一轮审校意见(若有):{feedback}) board.artifacts[draft] draft review json.loads(self.reviewer.run(f短文:\n{draft})) if review[pass]: board.artifacts[status] f第 {attempt1} 轮通过 break feedback review[comment] board.artifacts[status] f第 {attempt1} 轮被打回:{feedback} else: board.artifacts[status] 达到最大重写次数,强制交付 return board if __name__ __main__: board Orchestrator(max_revisions2).run(多 Agent 系统的工程价值) print(json.dumps(board.artifacts, ensure_asciiFalse, indent2))跑一遍你会看到:四个角色各司其职,黑板里清晰记录了每一步产出,审校不通过会带着意见打回去重写,而且永远不会无限循环。这套骨架我在实际项目里改改提示词、换换工具就能直接复用。五、真正决定生产可用性的几个工程点demo 跑通容易,上生产难。下面几条是我踩过坑之后总结的,比拓扑选择更影响最终成败:1. 终止条件是第一优先级。 多 Agent 最大的事故来源不是答错,而是停不下来——两个 Agent 互相打回、规划器无限拆任务。上面代码里那个max_revisions和for...else不是装饰,是保命的。任何反馈闭环都必须有硬上限。2. 成本会失控。 一次单 Agent 调用,在多 Agent 里可能变成十几次。审校循环、上下文重复传递,token 烧得飞快。务必在编排器层做全局计数和熔断,别等账单出来才发现——这类全局 token 计量和熔断,其实更适合沉到平台层统一做,我们在 Bizfocus ADP 上就是把它作为编排引擎的默认能力,业务侧不用每个项目重复造一遍。3. 把黑板做成可观测的。 出问题时,你需要的不是模型日志,而是第几步、哪个 Agent、输入输出是什么。结构化的共享状态天然就是最好的 trace,这也是我坚持黑板模式的重要原因。我们在 Bizfocus ADP 里把每个 Agent 每一步的产出都落成结构化记录,排查一条链路通常几分钟就能定位到出错的那一环。4. 错误隔离。 单个执行器失败,不该拖垮整条链路。给每个 Agent 调用包上重试和降级,失败时编排器要能决定是跳过、重试还是中止。5. 不要过度拆分。 这是最反直觉的一条。我见过有人把任务拆成八九个 Agent,结果延迟翻倍、协调成本暴涨、效果还不如三个 Agent。拆分是有成本的,每多一个 Agent 就多一层通信和失败面。先用最少的角色跑通,再按瓶颈逐步拆。写在最后多 Agent 协作不是把模型堆起来就行,它本质上是一个分布式系统设计问题:你要考虑通信、状态、终止、容错和成本,跟当年设计微服务的那套思维高度同源。我的建议很实在:从编排器–执行器结构起步,用黑板做共享状态,把终止条件和成本熔断当成一等公民,然后用最少的 Agent 跑通你的第一个版本。 把上面那套骨架复制下去改一改,你就有了一个可控、可观测、能迭代的起点。模型还在快速变强,但如何把多个智能体组织成一个可靠系统这件事,工程含量只会越来越高。这恰恰是我们做工程的人,长期的价值所在。