我最近在做一版多Agent协作框架仓库名就叫agency-agents。这个名字起初是起着玩的因为感觉它就像一家虚拟公司几个角色分工明确、有事上报、有活分组。真正动手之后发现名字起得好不好不重要关键是要回答一个很现实的问题——单Agent明明能跑通的事为什么非要拆成多个Agent如果你已经用单个Agent做过一些自动化任务又明显感觉到任务一复杂就开始失控那么这篇文章应该能给你提供一个从架构到落地、从踩坑到收敛的完整参考。很多人入门Agent开发时都是从单Agent开始的给一个系统提示词挂上几个工具让它在循环里调用工具、观察结果、再推理。这种模式应对单点任务基本没问题比如查一下天气生成一段文案调用接口抓数据。但当你把任务复杂度往上抬比如分析这个代码仓库的问题、修复其中三个Bug、补上对应单元测试、最后更新README单Agent就开始难受了。第一个问题是上下文膨胀。Agent的每一步都要把当前状态塞进对话历史任务越长历史越臃肿。我实测过一个包含意图理解、仓库扫描、多文件修改的长任务跑到第三四轮的时候早期的关键指令已经被大量中间输出稀释模型开始出现记不清最初要求的症状比如改了A文件却忘了B文件的关联依赖。第二个问题是错误传播不可逆。单Agent没有换人重来的机制第一步理解错了后面所有工作都建立在错误的假设上而且很难回溯到出错点。你想让模型重新想一下它往往只是在错误的上下文里再生成一遍错误答案。第三个问题是无法并行。一个任务里如果既有资料调研、又有代码改动、还有文档撰写单Agent只能串行处理。串行意味着慢也意味着任何一个环节的阻塞都会拖垮整条链路。所以就有了把职能拆开的直觉让规划Agent负责拆解需求研究员Agent只做资料收集编码Agent专注改代码审查Agent负责找毛病。听起来合理但真把多个Agent塞进一个系统时问题并没有消失而是转移了——谁先跑结果怎么交接中间消息要不要互相看某个Agent卡住了怎么办这些问题叠在一起我就搭了一个最小实验台这就是agency-agents项目的最初形态。1. 为什么需要一个Agent机构而不是继续用单个Agent1.1 单Agent看起来够用其实是视觉错觉先说一个容易误判的点如果你只在Demo里跑单Agent它的表现往往相当不错。原因很简单单Agent任务通常由人肉把复杂问题拆成了小问题再逐个交给模型处理。但真实业务不是这样的用户丢过来的是一个完整诉求中间没有人为拆解。把这种完整诉求直接丢给单Agent时它需要同时扮演分析师、执行者、质检员三个角色而这三个角色对提示词和上下文的需求是互相冲突的。分析师需要看到全局执行者需要聚焦局部质检员需要怀疑一切。让同一个模型在同一个上下文窗口里不停切换视角结果是每个视角都做不深。我见过最典型的一个例子是让单Agent写一个带登录验证的页面它一分钟生成完代码看起来没问题可一旦追问如果验证服务不可用登录超时怎么办它就开始含糊其辞。不是模型不会而是它的上下文里塞满了生成代码时的细节已经没有余地去仔细考虑异常路径了。这个现象背后的原因是认知预算问题。大模型的注意力是有限的上下文越长平均分配在每个token上的注意力就越稀薄。单Agent通过延长上下文来应对复杂任务等于在同一个桌面上堆越来越多文件最后谁都会找不到自己要的那张纸。1.2 多Agent系统真正要处理的核心矛盾拆开之后你会发现多Agent系统本质不是把任务发给N个模型而是在多个有独立上下文的执行单元之间设计一套可控制的信息流和状态流。如果把全部消息都放到一个共享列表里那就等于没有拆只是把同一个上下文切成几段喂给几个模型最终效果可能还不如单Agent。所以我在项目里给自己定了三条设计原则后来看几乎所有的坑都是违背了这三条才踩的隔离优先每个Agent的长期记忆和短期观察默认互相隔离信息传递必须显式发生。显式交接A要把结果交给B必须通过编排层登记不允许A直接写B的上下文。可终止性任何Agent执行都不能无限循环必须有步数上限、时间上限和成本上限。这三条看着简单实际落地时牵扯出一堆细节。比如隔离优先意味着每个Agent要维护自己的消息队列不能所有消息一锅炖显式交接意味着要设计一套标准消息格式而不是让Agent之间用自然语言随意对话可终止性更是要同时管住循环次数、Token消耗和真实时间。2. 核心抽象把Agent当成有分工的最小部门2.1 角色、指令、工具与预算四位一体我在代码里把Agent定义成这样一个结构from dataclasses import dataclass, field dataclass class ToolSpec: name: str handler: callable timeout: float 20.0 error_retry: int 2 dataclass class AgentSpec: name: str # 角色名如 planner / researcher / coder instructions: str # 系统提示词定义职责边界 tools: list[ToolSpec] # 允许使用的工具白名单 budget_steps: int 30 # 最大推理-行动轮数 budget_time: float 300.0 # 最大执行秒数 model_name: str default # 模型路由字段这个结构的设计逻辑可以类比一家公司里的岗位name是职位instructions是岗位说明书tools是员工能碰的设备清单两个budget是考勤和预算。岗位说明书决定了Agent遇到什么情况该怎么想工具白名单决定了Agent能做什么不能做什么而预算则保证任何一个员工都不会无限加班。其中容易被忽略的是工具白名单。我在调试中遇到过一种很尴尬的情况研究员Agent只是想去搜索资料结果因为工具列表里挂着执行代码权限它竟然尝试写脚本去爬数据。不是模型故意越权而是提示词稍微模糊一点模型就会选择看起来更高效的工具。所以工具权限一定要在系统层面写死而不是寄希望于提示词约束。权限这件事能落到代码就不要落到提示词。2.2 记忆三层分离会话、角色与工具观测刚开始偷懒所有Agent共用一个全局message list结果很快就发现互相污染。后来我把记忆拆成三层每一层的读写规则都不同会话级共享黑板存放任务的总体目标、当前进度、成品归档。所有Agent可读只有编排器能改。角色级个人台账每个Agent单独一份记录它在本次任务里的思考过程、中间结论和待办。仅该Agent可读写。工具级观测日志工具每次调用的原始输入输出统一落盘不进入对话上下文需要时再按检索方式取用。这套分层的直接效果是编码Agent不会看到研究员Agent搜索时产生的几十条原始网页摘要它只会在共享黑板里看到研究员提交的一份资料结论干净且结构化。这一步对节省Token和减少上下文干扰起了决定性作用。2.3 消息协议与交互边界Agent之间不直接互相调用。我把它们之间的交互抽象成四种消息类型TaskMessage上级下发的任务包含目标、约束、期望输出格式。ResultMessage任务完成后的结果回传。AskMessage需要其他Agent补充信息时发出的请求。NotifyMessage无需回执的广播比如我改完了A文件请注意。这样设计的原因有两个。第一任何两个Agent的耦合都通过标准消息完成换人换模型、换角色不用改代码。第二消息都是可序列化的数据结构方便后续落日志、做追踪、做回放这对生产环境太重要了。2.4 预算控制与软超时预算控制我单独说一下。就算设定了budget_steps30一个卡在自我反思循环里的Agent也可能在这30步里消耗大量Token。所以我还会在每个Agent执行前做一个成本估算根据当前对话长度、目标模型单价、预估步数算出大致费用如果超出任务预算直接在编排层拒绝下发并提示用户拆分任务。软超时的设计也很有意思Agent超过budget_time的80%时编排器会向它发一条NotifyMessage提醒请尽快收敛输出而不是立刻杀掉它。实测中这个方法很有效很多Agent在被提醒后会主动收尾。真正超时才硬性终止防止整个链路僵死。3. 从零搭一个能跑的多Agent编排骨架3.1 技术栈选择与理由这一版我选了Python加asyncio用一个内存队列当消息通道。选择理由很简单第一Python生态里和模型接口、工具封装相关的库最全不是自己造轮子就是调现成的第二asyncio的轻量任务调度非常适合模拟多个Agent分时复用模型的场景——底层模型其实是串行调用的但异步框架让并发管理变得干净第三消息队列先落在内存里能少引入一个中间件就少引入一个后面真要上生产再换成更重的消息中间件接口不变。3.2 核心代码结构我拆出三个文件agent_core.py放Agent类orchestrator.py放编排器message_bus.py放消息队列与事件分发。Agent类的核心逻辑大概长这样class Agent: def __init__(self, spec: AgentSpec, model_client, message_bus: MessageBus): self.spec spec self.model model_client self.bus message_bus self.role_log: list[dict] [] self.bus.subscribe(self.spec.name, self.handle_message) async def handle_message(self, msg): for event in await self.run_task(msg): self.bus.publish(event) async def run_task(self, msg): 把一条任务消息转换成一系列动作事件 step_count 0 while step_count self.spec.budget_steps: step_count 1 response await self.model.call( systemself.spec.instructions, messagesself.role_log [{role: user, content: msg.content}] ) self.role_log.append({role: assistant, content: response.content}) action self.extract_action(response) if action.is_final(): yield Event(task_result, agentself.spec.name, payloadaction.output) return if action.type tool_call: yield Event(tool_call, agentself.spec.name, payloadaction)编排器的核心是事件循环——它不断从队列里取事件、更新黑板、判断下一步该触发哪个Agent。这里有个关键点编排器不关心Agent内部怎么推理它只关心Agent对外发出的消息和当前状态是否满足终止条件。这种拆分让Agent可以换出换入编排逻辑几乎不用动。编排器的基础循环可以写成这样async def orchestrate(board, bus, agent_map): while not board.is_done(): event await bus.fetch_next(timeout10) if event is None: continue if event.type task_result: board.apply(event.payload) # 把结果写入共享黑板 next_agent decide_next(board) # 根据黑板状态决定下一步 if next_agent: task build_task_from_board(board, next_agent) bus.publish(TaskMessage(tonext_agent, contenttask)) check_budget(board) await maybe_snapshot(board)真正的decide_next会比这复杂本质上是一个小型状态机根据任务清单是否生成文案是否完成代码是否通过审查这些标志位决定下一步触发谁。但核心思想就是决策逻辑集中在编排层Agent本身只响应消息。3.3 一个最小场景的完整流程我用一个具体场景来验证骨架某开发者提交了一个模拟项目X的需求希望生成一个H5落地页包含需求分析、文案撰写和代码生成三个环节。流程是这样的编排器往黑板写入任务目标下发TaskMessage给planner。planner生成三阶段任务清单把任务拆件发布到黑板并通过TaskMessage派发给copywriter和coder。copywriter阅读黑板上需求输出文案初稿发ResultMessage给orchestrator同时NotifyMessage通知coder文案结构已定。coder拿到文案结构和页面规范生成HTML/CSS代码附自测说明发ResultMessage。编排器汇总结果写黑板触发reviewer做最终审查。reviewer返回审查意见如果有问题编排器把问题作为新任务再发给coder最多循环两轮。黑板里的done标志置为true编排器退出循环返回一份汇总报告。整套跑完我在日志里能看到一条完整的事件链。这里最让我满意的是每个Agent的日志都只包含自己相关的消息复盘时一眼能看出是谁在哪个环节出了问题而不是在几十页的混合对话里翻找。3.4 日志与可观测的最简做法调试阶段我用了最朴素的做法——JSON Lines日志每个事件一行{seq: 12, ts: 2025-01-01T12:00:00Z, type: task_start, agent: coder, task_id: t-003, input_len: 1240}别小看这个格式它让我在排查某Agent为什么胡说八道时能精确回放到它当时实际收到的输入。**很多看似是Agent智商问题的事故最后查出来都是输入里混进了不该有的信息。**这一条经验至少帮我省了两天的调试时间。4. 协作模式三种常见方案与取舍4.1 直接链路调用最直觉但最不好扩展最直观的做法是让Agent之间直接互相调用A解析完需求后在自己的代码里直接调用B的接口。好处是实现快、链路清晰但它有三个硬伤协调责任分散每一次Agent间的调用都需要发起方处理错误和超时代码里到处是重试逻辑改一个环节牵动全身。链路深不可测A调用B、B调用C、C又回头调用A复杂度呈指数增长几乎不可能做全局终止。复用性差新Agent要接入必须改老Agent的代码违背了隔离原则。4.2 中央编排加消息队列最可控但编排器容易越长越肥第二种方案是引入一个中央编排器负责调度Agent之间不直连所有通信经过消息队列。中央编排器决定谁在什么时候收到什么消息Agent只对消息做出响应。这个模式的优点很明显可控性强任何一个Agent的身份和权限都在编排层可见可追踪消息全部经过总线天然有审计日志。缺点也明显编排器的逻辑会越来越重尤其当业务流程复杂时判断下一步该触发谁的状态机写起来很累还容易出现编排器比业务还复杂的尴尬。4.3 共享黑板简单但同步契约得自己写第三种是共享黑板模式所有Agent共享一份状态写完自己那块就走。这个模式实现最轻多个Agent之间的依赖通过状态间接发生。但它有个很麻烦的问题Agent之间没有稳定的先后契约你无法保证A写完某块数据时B正好在读。你需要在黑板里手动维护同步状态比如requirement_statusready才能让协作有序这就又是在写一种状态机。4.4 我的选择以黑板为主干以队列为分支经过几轮对比我最终选了混合模式简单说就是共享黑板保存所有Agent都需要的结论态消息队列承载需要实时响应的交互态。任务的总体目标和最终产物都在黑板上任何Agent随时可以查而Agent之间的实时询问、提醒、任务下发走消息队列。我用一张表把这个取舍讲清楚方便你做决定协作模式优点主要代价最适用场景直接链路调用简单、直观链路深、难终止、耦合重临时脚本、一次性任务中央编排消息队列可控、可追踪编排器复杂、易过度设计流程长、角色多、强审计共享黑板轻量、状态清晰同步契约需自建角色少、任务线性黑板队列混合兼顾状态与响应要同时写两套逻辑中型多Agent系统这个混合模式不是最优的但它在我这个规模里是投入产出比最高的。如果你只有两三个Agent建议直接用黑板模式别把消息队列扯进来如果你有几十个Agent且业务流程明确老老实实用中央编排加消息队列。5. 联调与实战中踩过的坑5.1 上下文污染某Agent把别人的消息当成了自己的任务第一次联调时我犯过一个典型错误。我把所有消息都丢进全局列表每个Agent读这个列表然后自己过滤看起来和自己有关的消息。结果研究员Agent开始回复编码Agent的任务消息因为它觉得代码改动可能影响文档结论。排查过程是这样的先看输出发现内容牛头不对马嘴再看日志发现该Agent的输入里有一条来自另一个Agent的TaskMessage然后查过滤逻辑发现它只是按关键词匹配相关消息而关键词设得太宽。根因清楚了——过滤责任不该放在Agent头上该由消息路由层解决。修复方式就是上面说的消息总线按to字段精确投递Agent只收发给自己的消息其他一律不进场。这个改动看着小实际上是把尽量不让不可控的东西进上下文这条原则落地了。5.2 循环调用A说改、B就改A又说还要改另一个让我头疼的问题是循环。编码Agent改了一版代码审查Agent说函数命名风格不统一编码Agent改了命名审查Agent又指出没处理边界情况编码Agent再改审查Agent又开始提别的……如果不干预这个循环可以跑到预算上限白白烧掉大量Token。我的排查路径是先量化问题——在日志里按Agent分类统计消息数量发现审查Agent和编码Agent的消息占了总量的80%然后给消息加上相似度指纹把连续多条高度相似的消息标记出来最后在编排层加了收敛规则——如果同一对Agent之间往复消息超过3轮就强制进入人工决策节点让模型优先汇总分歧而不是继续发散。收敛规则还配合了提示词注入当检测到循环时系统会给双方追加一句请优先采纳对方建议并给出最终版本不要引入新问题。实测效果很好大多数循环在第二轮就被掐掉了。5.3 静默失败某个环节挂了编排器毫无反应还有一次研究员Agent调用的资料接口超时工具层捕获异常后返回了retrieval_failed字符串研究员Agent把这个字符串当成了没有资料继续往下走最终交了一份内容空泛的规划。整个流程没有任何报错但结果完全不可用。这是典型的静默失败。修复办法我做了两层第一层工具层出现异常时必须返回结构化错误码例如{error: timeout, detail: ...}Agent看到错误码可以尝试重试但绝不能假装什么都没发生继续生成结论第二层编排器对每个Agent的开始和结束做心跳登记如果某Agent在收到任务后超过阈值还没任何输出按疑似卡死处理触发重试或转人工。这一条现在是我所有编排系统的标配。5.4 Token与成本失控问题出在一句提示词上最后提一个成本控制层面的事。为了调死循环问题我加过放宽预算看一下能不能自动收敛的功能结果某次任务一个Agent跑了接近两万Token其中一多半是用来生成让自己冷静下来的废话。排查下来发现问题出在系统提示词里那句请一步一步思考——在无约束场景下这句话会被模型理解成尽量长地输出推理过程。从那以后我对提示词的看法就变了规划的细节要放在代码里提示词只留必要的行为约束。比如不再说仔细思考而是说在5个要点内完成分析每个要点不超过30个字。把模糊的鼓励改成明确的输出规格Token消耗立刻下来一个量级。6. 往生产环境走哪些改造值得优先做6.1 追踪把每一步交给可观测系统本地调试用JSON Lines就够了上生产还得补两层链路追踪和状态快照。链路追踪就是给每个任务分配trace_id把Agent消息、工具调用、模型响应全部串起来状态快照则是定期把黑板内容落盘一旦进程崩溃可以从最近的快照恢复而不是从头重跑。别看这些名词高大上落地其实就是几行代码的事给事件都带上trace_id字段写一个每N个事件存一次黑板副本的函数。但这两件事能救命的场景很多比如你同时跑100个任务某个任务中途发疯只能靠追踪把当时的完整现场捞出来。6.2 降级与熔断别让一个环节拖垮全部多Agent系统还有一个独有的大坑链路中某一环耗时过长或频繁出错会逐步吞噬整个系统的资源和预算。我的处理方案是在编排器里维护一个健康度指标对每个Agent计算最近几次任务的失败率和平均耗时。健康度低于阈值后该Agent的任务会自动降级——比如研究员Agent挂了就把检索结果置空让规划Agent基于经验继续推进而不是让整个任务挂在检索上。6.3 模型路由强模型和轻量模型错开最后一个有效的优化是模型路由。不是所有Agent都需要最强的模型reviewer的找问题能力和researcher的归纳总结能力对模型要求高但类似formatter这种纯格式整理的工作用一个轻量模型就能跑。我在AgentSpec里加了一个model_name字段按角色配置不同模型整个任务成本降了差不多一半。这个做法背后是一个朴素的道理多Agent系统的成本不是各个Agent成本的简单相加而是和每个Agent的出活质量与调用次数相乘的。把高成本Agent用在刀刃上把低成本Agent用来干体力活整体收益非常明显。最后再补一个小的体会。如果你正在考虑上多Agent架构先别急着把架构堆大。可以先从一个简单的共享黑板加两三个角色跑起来观察消息量、上下文和成本再决定要不要引入消息队列和复杂调度。Agent系统的复杂度曲线比大多数人预想的要陡跑通一个环节容易跑通一条完整链路且能稳定复现才是真正考验编排设计的地方。我也是在这个项目从单Agent演化到多Agent再一步步踩坑收敛的过程中才真正理解了治理这两个字的分量。