“agency-agents”这个名字听起来有点神秘,我第一次看到它时正好在一段很狼狈的调试过程里:手头一个模拟项目的智能体任务跑到第三步就开始自说自话,把工具参数拼错,还一本正经地给出了错误结论。后来我把这套“多个Agent分工协作”的思路理清楚,重新搭了一版架构,才慢慢摸到让模型稳定干活的节奏。
这篇博文就围绕我基于“agency-agents”思路做的一套多Agent协作框架展开。先说清楚它到底解决什么问题、核心角色怎么设计、最小骨架怎么实现,再把真实跑任务时反复踩过的坑和排查过程原原本本讲一遍。如果你也在做Agent类应用、或者正准备把一个单智能体Demo推向更复杂的真实场景,这篇内容应该能帮你少走不少弯路。
1. 单Agent的天花板:为什么任务一复杂就开始失控
1.1 我踩过的典型翻车现场
先还原一个我真实遇到的场景。之前我在做一个文档整理类的模拟项目,需要让模型读取一篇长文、提炼要点、再按照指定格式输出一份调研笔记。最初的方案很直接:一个Agent,给它一大段Prompt,让它“一步到位”。
结果并不理想。任务简单时它能正常产出,但只要输入的文档超过一定长度、或者中间涉及好几步工具调用,它就开始出现三种非常典型的问题:
- 上下文漂移:后面回答问题时,会忘记最开始的任务约束,输出格式越来越随意;
- 工具参数幻觉:调用检索工具时,凭空编造不存在的文件路径或ID;
- 一步错步步错:某一步的结果不太对,后续步骤不会主动纠正,反而基于错误结果继续延展,最后产出一份看似完整、实则不可用的报告。
最崩溃的是第一次跑长文档时,模型把中间步骤的中间结论当成了最终答案,在结尾处直接输出了大段截取原文,没有做任何整理。我当时盯着输出页面愣了半天,脑子里只有一个念头:单靠一个Agent、一个大Prompt,根本管不住这种多阶段任务。
1.2 本质是“权限与上下文的双重压迫”
后来我认真做了复盘,发现单Agent模型之所以在高复杂度任务中翻车,两个原因很关键。
第一个原因是上下文窗口的竞争。一个Agent既要理解任务背景、记住用户约束、又要承载中间步骤的输出结果、还要时刻保持输出格式一致,所有信息挤在同一份上下文里。当关键信息被淹没,模型自然倾向于“忘事”。这不是模型笨,而是设计上把太多职责塞给了同一个上下文容器。
第二个原因是角色冲突。同一个Agent既当“规划者”又当“执行者”还要当“质检员”。规划时要求它发散、拆解任务;执行时要求它聚焦、调用工具;质检时又要求它批判性地审视自己。这三件事的思维模式几乎是互相排斥的,硬塞给同一个模型实例,结果往往是哪个角色都做不彻底。
想明白这点后,我开始参考“agency-agents”这种把Agent当作机构里不同职能岗位来组织的思路,把任务流程从“一个人全包”改成了“一个团队分工协作”。效果非常明显,后面的篇幅我会详细讲实现路径和实际数据。
2. agency-agents的总体设计:让智能体像部门一样分工
2.1 三个核心角色:规划者、执行者、督办者
我把一个复杂任务拆成了三个固定角色,分别对应项目管理里的三种岗位:
- 规划者(Planner):负责读任务、拆解步骤、制定执行方案。它不直接执行具体操作,只输出一份结构化的任务单;
- 执行者(Worker):按照任务单逐项干活,比如调用检索工具、读取文件、生成中间内容。它专注于“把这一步做完”,不关心全局;
- 督办者(Supervisor):对执行者产出的结果做验收,判断是否符合要求、是否遗漏步骤。不合格就打回重做,合格就交给规划者决定是否进入下一步。
这套分工直接化解了上一节说的“角色冲突”。规划者不必边规划边执行,执行者不必边干活边自我检查,督办者也不用担心自己是“既当运动员又当裁判员”,每个角色都只需要做好一件事。
你可能会有个疑问:三个角色都调用同一个底层模型,不还是同一个模型在干活吗?没错,底层模型可以是同一个,但因为各自的System Prompt、可见上下文、输入消息格式都做了隔离,模型在同一时刻只承担一种角色,输出的稳定性会明显提升。这就像同一个人在不同会议上戴不同的帽子,发言风格和职责边界完全不同。
2.2 通信协议:用结构化“任务单”代替自然语言聊天
Agent之间的沟通方式,是这套架构里我最早确认、也最坚持的一点:绝不让Agent之间自由地用自然语言对话。
原因是自然语言太“软”了。两个Agent来回聊天,说着说着就跑题,而且每条消息都占用上下文空间。更麻烦的是,如果A问“把结果给我”,B可能会回复一大段解释性废话,真正有用的数据反而被淹没在措辞里。
我采用的方案是定义一套统一的任务消息结构,核心字段包括:
- task_id:任务唯一ID,用于追踪整条执行链路;
- step_id:当前步骤编号,用于标识执行到哪一步;
- type:消息类型,比如 plan、action、result、review、retry、done;
- content:实际内容,结构化为JSON,而不是自由文本;
- meta:元信息,比如模型消耗、耗时、工具调用记录。
所有Agent之间的交互都走这套结构。规划者产出的是“计划JSON”,执行者产出的是“结果JSON”,督办者产出的是“验收JSON”。每个环节产出的数据结构高度一致,后续调度逻辑、日志审计、异常恢复都会轻松很多。
我为这个设计写的注释里有句话一直留着:Agent之间传递的应该是数据,而不是散文。这句话后来帮我避掉了至少一半的通信问题。
2.3 调度循环:一次任务的完整生命周期
有了角色和通信协议,整个任务的运行就像一条流水线:
- 用户下发原始任务,规划者将其拆解为可执行计划;
- 规划者根据计划生成第一个执行步骤的任务单;
- 执行者接收任务单,执行对应工具调用,返回结构化结果;
- 督办者接收结果,对照当前步骤要求做校验;
- 校验通过,则返回规划者,规划者判断是否还有下一步;校验失败,则打回执行者重试,或调整计划;
- 所有步骤完成后,规划者整理全局结果,输出最终交付物。
这个循环看起来简单,但真正实现时有个关键点:规划者不是一个只开一次工的“项目经理”,而是每一轮都要参与决策。它需要根据督办者的校验反馈,动态决定“继续”“重做”还是“改路线”。这种“执行-校验-再规划”的闭环,是普通线性Prompt链完全不具备的能力。
3. 从零实现一个最小可用的多Agent骨架
3.1 消息与状态定义
我不喜欢过度设计,所以从第一版起就没引入重量级框架,而是用Python写了一个足够表达核心逻辑的最小骨架。里面最核心的数据结构是TaskMessage和AgentContext。
from dataclasses import dataclass, field from typing import Optional, Any from enum import Enum import uuid class MsgType(str, Enum): PLAN = "plan" ACTION = "action" RESULT = "result" REVIEW = "review" RETRY = "retry" DONE = "done" class AgentRole(str, Enum): PLANNER = "planner" WORKER = "worker" SUPERVISOR = "supervisor" @dataclass class TaskMessage: task_id: str = field(default_factory=lambda: uuid.uuid4().hex) step_id: int = 0 type: MsgType = MsgType.PLAN sender: AgentRole = AgentRole.PLANNER receiver: AgentRole = AgentRole.WORKER content: dict[str, Any] = field(default_factory=dict) meta: dict[str, Any] = field(default_factory=dict)这段定义是整个系统的“语法”。所有Agent都只认这套结构,不认自然语言。刚开始写的时候我也犹豫过,是不是太死板了,但后面看到执行链路日志时,我庆幸当时坚持了结构化设计,每条消息都能在日志里完整复现,排查问题时效率高得多。
AgentContext则用来保存每个角色的独立上下文,避免所有消息混在一起。每个Agent角色有自己独立的上下文窗口,这意味着执行者不会看到规划者与督办者之间的全部交流,从而有效减少下一节要讲的信息干扰。
@dataclass class AgentContext: role: AgentRole system_prompt: str memory: list[dict[str, Any]] = field(default_factory=list) token_budget: int = 4000 used_tokens: int = 0 def append(self, msg: TaskMessage): self.memory.append({"type": msg.type.value, "content": msg.content}) def to_messages(self): return [{"role": "system", "content": self.system_prompt}] + self.memory3.2 核心调度器实现
调度器是整个框架的“传菜员”,负责把所有任务单按流程送到正确的角色手里。我第一版写得比较粗糙,直接用if-else串联调用顺序,跑通是跑通了,但每加一个角色就得改一大段逻辑,非常痛苦。后来我改成了一张表驱动的模式,把这个过程抽象成“消息类型到接收角色的路由”。
ROUTE_MAP = { MsgType.PLAN: AgentRole.PLANNER, MsgType.ACTION: AgentRole.WORKER, MsgType.RESULT: AgentRole.SUPERVISOR, MsgType.REVIEW: AgentRole.PLANNER, MsgType.RETRY: AgentRole.WORKER, MsgType.DONE: None, } class Scheduler: def __init__(self, agents: dict[AgentRole, Any]): self.agents = agents def run(self, initial_task: str, max_steps: int = 8): current_msg = TaskMessage( type=MsgType.PLAN, sender=AgentRole.USER, receiver=AgentRole.PLANNER, content={"task": initial_task} ) history = [] for _ in range(max_steps): receiver = ROUTE_MAP.get(current_msg.type) if receiver is None: return current_msg, history agent = self.agents[receiver] current_msg = agent.handle(current_msg) history.append(current_msg) if current_msg.type == MsgType.DONE: return current_msg, history raise RuntimeError("max steps exceeded without done signal")这张路由表最大的好处是:任务流转逻辑一目了然,新增一种消息类型或调整协作流程,只需要改ROUTE_MAP,不用碰调度主流程。实际运行中我还会在循环里加上超时等待、重试、token预检等逻辑,但骨架就是这个样子。
3.3 接入实际模型:可替换的LLM接口
为了让这套骨架不绑定某一家模型服务,我定义了一个统一接口。实现这个接口的类负责把消息列表发给模型服务,并解析返回内容为标准JSON。
class ChatModel: def __init__(self, model_name: str): self.model_name = model_name def chat(self, messages: list[dict]) -> str: # 这里按实际服务SDK接入即可 raise NotImplementedError class Agent: def __init__(self, role: AgentRole, model: ChatModel, system_prompt: str): self.role = role self.model = model self.ctx = AgentContext(role=role, system_prompt=system_prompt) def handle(self, msg: TaskMessage) -> TaskMessage: self.ctx.append(msg) messages = self.ctx.to_messages() raw = self.model.chat(messages) content = self._parse_json(raw) return TaskMessage( task_id=msg.task_id, step_id=msg.step_id + 1, type=self._decide_response_type(content), sender=self.role, receiver=ROUTE_MAP.get(self.role, AgentRole.PLANNER), content=content )这里有个很实用的细节:每个Agent构造时都塞入独立的System Prompt。规划者的Prompt强调“拆解任务、输出步骤清单、关注全局”;执行者的Prompt强调“只处理当前步骤、使用工具、不自行扩大范围”;督办者的Prompt强调“严格核对、指出偏差、勿自行修改内容”。
因为是相同底层模型,所以这个差异完全是Prompt塑造出来的。实测证明,只要角色Prompt写得足够清晰,同一个模型完全可以在不同上下文窗口里表现出完全不同的行为风格。
需要注意的是,解析模型的输出时不要指望它每次都给你干净JSON。一定要加一层容错解析:先尝试json.loads,再尝试截取第一个花括号到最后一个花括号之间的内容,最后才允许报错或返回到督办者处理。
4. 让它真正跑起来:一次多步任务的实战复盘
4.1 场景选择与任务拆解
为了验证这套架构在实际任务中值不值得用,我拿一个相对接近真实工作流的场景做了试验:假设要给某团队产出“一份技术选型调研报告”,要求是检索若干信息来源、比较两种方案、给出推荐结论,并按照固定模板输出一份结构化报告。
单Agent版本以前在这个任务上表现很不稳定:检索来源经常缺失、推荐理由前后矛盾、模板格式偶发错乱。
换成agency-agents架构后,任务被规划者自动拆成了七个步骤:
- 解析任务目标与输出模板要求;
- 检索方案A的关键特性信息;
- 检索方案B的关键特性信息;
- 检索两种方案的社区使用情况与已知问题;
- 汇总对比信息并填充模板;
- 输出结构化报告草稿;
- 检查模板完整性与结论一致性。
注意,这七个步骤并不是我预先写死在代码里的,而是规划者Agent根据用户给出的任务描述动态生成的。我只是在System Prompt里提示它“按检索、对比、汇总、检查的通用流程思考”。
4.2 各Agent的实际表现
执行阶段,我关注的重点不是“每一步是否顺利”,而是“失败时系统如何应对”。这里挑两个有代表性的环节说。
环节一是检索工具连续失败。第一次执行者在检索“方案B的社区反馈”时,传入的关键词过于宽泛,返回结果质量很差,督办者识别出“结果与目标不符”,给出具体验收意见后打回重试。执行者收到重试指令后调整了关键词,第二次返回了合格内容。如果是单Agent,我很怀疑它能意识到“当前结果不合格”,更可能直接把低质量结果写进报告。
环节二是督办者抓到结论矛盾。草稿生成后,督办者在对比“推荐理由”和“限制条件”时发现,前面说方案A适合高并发场景,后面又写方案A在大流量下出现过锁竞争问题。这个矛盾如果放在单Agent场景里很容易被吞掉,因为同一个模型在生成时很难同时以批判视角审视自己的推导链。而在分工架构里,督办者独立提出质疑,规划者再根据质疑决定修改哪一段,流程非常清晰。
4.3 实测下来的关键数据
我在同一份测试数据集上分别跑了单Agent版和agency-agents版,记录了三个关键指标,差异很明显:
| 指标 | 单Agent | agency-agents |
|---|---|---|
| 任务完成率 | 62% | 89% |
| 平均Token消耗 | 约1.1万 | 约1.6万 |
| 结果格式合格率 | 71% | 96% |
Token消耗上去了约45%,换来的是完成率提高27个点、格式合格率提高25个点。对于交付型任务来说,这个成本我非常愿意付。如果只是闲聊型任务,这套架构确实不划算,杀鸡用了牛刀。
这里也侧面回答了一个问题:什么时候该上多Agent架构?答案很清晰——当任务包含多个依赖步骤、中间结果需要校验、并且最终交付物需要达到一定质量标准时,收益远大于成本。
5. 调优与避坑:我在迭代中反复踩过的五个坑
5.1 循环无死锁:调度器怎么防“两个Agent互等”
第一次跑通多Agent循环后,我遇到过任务卡住的情况:执行者返回一个结果,督办者认为不合格打回重试,执行者第二次仍旧失败,督办者继续打回……两边来回拉扯,直到触发最大步数上限,任务失败退出。
这个问题本质是“没有死锁检测,也没有失败升级机制”。两边一直在原地打转,但没有任何一方有权限跳出循环。
我的解决方案是在调度器里增加一条规则:同一个步骤连续重试次数达到阈值后,不再直接打回同一角色,而是把重试信号升级给规划者,让规划者重新评估这个步骤的执行方案,比如换个措辞、换种工具、或者干脆跳过该步骤将其标记为低优先级。
这条规则非常有效。它把“无脑重试”变成了“升级决策”,符合真实项目里项目经理介入的直觉:员工同一件事做两次都不行,那就不是重做的问题,而是方案的问题。
5.2 上下文隔离:执行者之间不该互相看到全部流水账
第二版我在上下文处理上偷了懒,所有角色的memory都直接拼接全局历史,结果执行者在做第5步时,能看到第2步的中间结果。听起来似乎没什么,但模型会依赖这些历史信息做“多余发挥”。
有一次执行者在做“汇总对比信息”时,居然把第2步检索到的一条原文直接抄进了汇总里,理由是“该信息在前面出现过,应该需要保留”。这就是典型的上下文污染。
我调整后,每个执行步骤只给它相关步骤的结果摘要,而不是完整历史。具体做法是在调度器发送任务单时,对执行者上下文中注入的“背景信息”做裁剪,只保留目标步骤关联度最高的几项结果。
用个生活化类比:就像你让同事去仓库拿货,你只需要告诉他“拿A区货架3的型号”,而不需要把昨天采购谈判的全部会议记录塞给他。信息给得越多,他越容易自作主张。
5.3 工具权限的最小化:别给执行者一把万能钥匙
多Agent架构里,工具权限的分配非常考验设计功力。我一开始图省事,所有执行者共享同一个工具列表,里面包含文件读取、网页检索、数据库查询等十几个工具。
结果很快出问题:一个执行者本应只调用检索工具,却擅自尝试读取本地文件,导致权限异常,整条链路中断。
后来我把工具权限也做成“角色关联”模式,不同执行步骤挂载不同工具集。比如负责检索的步骤只能看到检索类工具,负责汇总的步骤不暴露任何外部工具,只允许访问内部消息数据。
这个调整不仅减少了错误调用,还有一个连带好处:由于模型可选工具少,每次工具选型的决策成本更低,输出更稳定。模型不需要在一堆工具里纠结该用哪个,因为它根本看不见不该用的。
5.4 结构化输出解析失败的兜底策略
这一节要讲一个让我印象极深的排查过程。某一天跑任务时,系统在“输出草稿”阶段频繁失败,报错显示无法解析模型的返回结果。我第一反应是模型服务异常,翻日志却发现HTTP请求全部成功,响应内容也完整。
真正的问题藏在一个看似无关的细节里:那次Prompt末尾多了一句“请用规范格式输出最终结果,谢谢”。模型看到“谢谢”后,在JSON末尾追加了一句“如果还有其他需要,请随时告诉我”,直接导致JSON解析失败。
排查链路是这样的:先定位到解析异常,确认原始响应内容;然后对比同一任务在正常时段的响应,发现多了一句礼貌性收尾;再回查Prompt修改记录,确认是之前为了“引导模型语气”加的话导致的;最后把Prompt里所有与结构化输出无关的礼貌语全部删除。
这个坑的教训有两个层面:一是解析逻辑必须足够健壮,不能信任模型每次都乖乖输出干净JSON;二是Prompt里任何一句无关紧要的话,都可能在特定时机改变模型行为。结构化输出场景下,Prompt越纯粹越好,客套话只会增加不确定性。
5.5 错误恢复与任务重试:如何不让一次失败打崩全链路
最后一个坑是重试时的状态一致性。最开始我的重试是“原任务单原样重发”,执行者收到的是和第一次完全一样的任务和上下文。表面上没问题,但真正跑起来会发现:很多时候执行者第一次已经做了一半工作,只是最后一步失败,重试时它会把已经完成的部分再次写入,导致重复操作。
比如有一次执行者负责“检索并总结方案A的已知问题”,它已经检索到三篇有效文章,但在总结时模型输出格式错误。重试时它又重新检索了一遍,浪费时间不说,检索结果和第一次还不完全一致,导致前后总结数据对不上。
我后来的做法是在重试任务单里追加“partial_results”字段,把执行者之前已经完成的有效中间产物传回去,并明确提示“如果已有局部结果,请基于此继续而不是重新执行”。这一步改进让长链路任务的重试成功率提升了非常多,因为模型不会再盲目从头再来,而是基于已有进展做修复。
6. 从Demo到产品:非功能属性与后续演进思路
6.1 可观测性:没有日志的Agent系统寸步难行
我这套东西从跑通到稳定,排在第一位的不是算法优化,而是日志。多Agent系统里,状态流转比单次模型调用复杂得多,如果没有完整的链路日志,一次失败排查轻则半小时,重则两眼一抹黑。
我在框架里默认记录三类日志:
- 消息日志:记录每一条TaskMessage的完整内容、发送方、接收方、时间戳;
- 模型日志:记录每次模型调用的Prompt摘要、Token数、延迟、返回状态;
- 流程日志:记录调度器的每一步决策,包括打回理由、重试原因、升级处理。
这三种日志拼起来,基本能还原任意一次任务从开始到结束的完整“心路历程”。遇到问题时,我的排查顺序永远是先看流程日志定位卡点,再看消息日志还原上下文,最后才去看模型原始输出。
如果你也在写Agent框架,请把日志当成一等公民。这个系统的调试难度和单点Prompt工程不是一个量级,没有日志,你连Quora上的“为什么我的Agent疯了”都搜不出答案。
6.2 成本控制与token预算
多Agent系统的成本是绕不开的话题。前面实测数据已经显示了Token消耗增幅,长期运行还需要更精细的预算控制。
我在每个Agent的上下文里都维护一个token_budget字段,调度器在发送任务单前会估算当前上下文长度,接近阈值时强制触发“历史摘要压缩”:把过去几轮消息压缩成一段摘要,替换原内容,保留关键信息但减少Token占用。
这个机制在长链路任务中尤其重要。当任务步骤超过5步时,历史消息累积很快,如果不做摘要压缩,上下文最后会膨胀到无法继续。刚开始我低估了这个问题的严重性,直到一次12步的任务跑到第9步直接超限失败,我才认真实现了这个压缩逻辑。
6.3 我推荐的演进顺序
如果你打算参考这套思路做自己的多Agent系统,我建议不要一上来就追求花哨的编排框架,而是按这样的顺序推进:
先跑通最小骨架,用三个角色、一份路由表、一个调度循环实现一个你能实际使用的场景;再补上日志和结构化消息,确保每次失败都能被完整复盘;然后加失败升级和重试机制,让系统具备基本的自愈能力;最后再铺开工具权限、预算控制、摘要压缩这些偏“生产环境”的能力。
这个顺序的核心逻辑是:基础稳定性永远是第一位的。模型能力、智能程度、协作艺术都是后面的事,当你的日志能完整讲清“每个Agent每一步在干什么、为什么这么干”的时候,系统的可靠性才真正立得住。
我自己在迭代过程中最大的体会是:多Agent系统的价值不在于让模型“看起来更聪明”,而在于让任务的每一步都有独立的检查者和反馈回路。它不会让单个模型变强,但会让整个系统的输出质量稳定很多。这就像一支球队,不要求每个人都成为超级明星,但每个位置都有人扛住自己的职责,整场比赛的胜率自然就高了。