这些年“Agents”这个火热各种框架层出不穷但每次看到“agency-agents”这种命名我还是会忍不住多琢磨几秒。它不像“agent-demo”那么直白也不是某个模型厂商的官方仓库名而更像是一个描述“一群智能体如何组织起来干活”的集合概念。我刚接触这个方向的时候也以为它只是一个简单的学习项目直到自己动手搭了一遍才发现“agent”本身不是难点“agency”也就是多个智能体之间的协作、权责分配、消息流转才是真正烧脑的部分。这篇博文就围绕我在一个模拟项目X里如何把一个“一堆智能体”的混沌想法逐步落地成一个可运行、可扩展的“agency-agents”协作框架。我会把设计思路、架构选择、踩过的坑、还有关键的代码骨架都写出来尽量做到既能让刚入门的朋友看懂也能让已经在做多智能体应用的同学找到点共鸣。1. 破题从“agent”到“agency-agents”到底是多了什么1.1 拆解标题里的三层含义先说结论我认为“agency-agents”这个组合词至少表达了三层意思第一层它是一个智能体的集合。单个agent能处理的问题有限比如让它写一段代码、做一次检索、整理一份摘要这些都算是单点能力。而“agency-agents”天然暗示着“一批”智能体同时存在彼此之间有关系、有分工。第二层它强调“代理机构”式的组织感。在真实世界里一家agency有策划、有执行、有审核、有对接客户的岗位。映射到系统里就是不同的agent扮演不同职能角色而不是一堆同样的agent各干各的。这种角色化设计是多智能体系统和“并行调用多个模型”最本质的区别。第三层它指代“可以协作的框架”。既然是复数又带有组织含义那实现上就必须有通信机制、任务分发机制、状态同步机制。哪怕只是一个很简化的实现也得有这些骨架否则顶多算一个“多进程脚本”。我当时拿到“agency-agents”这个题目时第一反应是需求方到底想要一个能用的产品原型还是一个用来验证多智能体协作概念的技术Demo后来通过和某导师沟通确认了定位是后者——一个可扩展的、代码清晰的参考框架目的是验证“多角色智能体协作解决复杂任务”的可行性。1.2 为什么单Agent模式不能满足需求很多人刚接触大模型应用时会习惯性地把任务丢给一个超长的Prompt指望一个Agent全流程搞定。我也走过这条路实际结果通常是Prompt写得越长模型越容易丢失早期指令。尤其是超过上下文窗口的临界点之后前面的“角色设定”和“步骤要求”会被后续内容稀释甚至是遗忘。任务一旦出现异常分支单个Agent只能靠着“if-else”式的自我推理去兜底可它往往既要做决策又要做执行还得自查错误脑袋里塞了太多事效果自然下降。单Agent模式下所有的工具调用、中间结果、上下文记忆都耦合在一起出了问题很难定位是模型推理错了还是工具返回异常还是上下文被污染排查成本极高。而“agency-agents”模式就好比把一家公司的流程搬到系统里策划Agent只负责拆解目标和制定计划执行Agent只负责调用工具和生成内容审核Agent只负责挑错。各司其职上下文被隔离问题域被切分。实测下来整体成功率比单Agent硬扛要稳定得多。2. 方案选型为什么我选择“规划-执行-审核”三级协作2.1 常见的多Agent协作模式对比多智能体协作的实现方式其实有很多种我之前大概梳理过三类主流模式各有各的适用场景模式核心思想优点缺点管道模式智能体按阶段串行处理前一个的输出是后一个的输入流程清晰、容易调试链路长时延迟高且中间环节出问题会中断整体流程编排模式有一个“规划者”负责任务分解和调度执行者听从分配职责清晰、扩展性好规划者一旦判断失误整个任务方向就会跑偏协商模式多个智能体地位平等通过讨论达成共识适合开放式问题实现复杂容易陷入无休止的来回对话成本不可控这次的项目X采用的是“编排模式”的变体叫“规划-执行-审核”。选择它不是因为它最先进而是因为它在可控性和效果之间最平衡。项目要求是用尽量简单的方式验证多智能体的价值那就必须让责任边界足够清晰每一个环节出的问题都能被单独观测和修正。2.2 固定流程编排的取舍在项目初期我也考虑过要不要上更“智能”的动态编排也就是让模型自己决定下一步该调用哪个Agent。后来果断放弃了原因有三点都是实操层面的动态编排意味着每一个Agent返回的格式都可能是半结构化的解析成本高错误率高。模型生成的JSON字段经常缺胳膊少腿处理这些“格式垃圾”的时间比写业务逻辑还长。协商模式的token消耗非常大。一次任务来回二十轮对话成本翻好几倍而项目X只是验证概念不需要无限探索。动态编排的调试体验极其痛苦。今天能跑通的链路明天换一个模型版本可能就走不通了完全不可控。所以最后采用了一个看似“笨”但非常可靠的方法固定流程 角色隔离 节点级重试。也就是流程的骨架是固定的但每个节点内部允许根据输出质量自动触发一次重试或修复。这种方式既保留了多Agent协作的优势又把实现复杂度控制在了合理范围内。3. 核心架构消息总线、任务状态机和角色定义3.1 消息总线Agent之间怎么说话在“agency-agents”框架里Agent之间不能直接互相调用方法否则就变成强耦合了。我参考了消息队列的思路实现了一个轻量级的内存消息总线。所有的Agent实例只和总线打交道不关心消息是谁发的也不关心最终谁会消费。消息总线的数据结构非常简单但非常关键。我定义了这样几个字段dataclass class AgentMessage: message_id: str sender: str receiver: str msg_type: str # task_request / task_result / review_report / error payload: dict timestamp: float retry_count: int 0别小看这个结构。“receiver”字段决定了消息是点对点发送还是广播“msg_type”字段则方便消费者快速路由。之前我见过很多项目把所有信息都塞进payload然后用一个大的type字段区分结果后面根本没法维护。这样拆开之后每个Agent收到消息时只需要匹配自己关心的msg_type其他一律忽略逻辑非常清爽。3.2 任务状态机一个任务从生到死的完整轨迹多智能体系统中最容易“失控”的地方就是任务状态。A任务被B、C、D三个Agent同时处理时谁先完成谁在等待谁报错了这些必须一目了然。所以我在框架里建了一个任务状态机流转路径是pending - queued - dispatching - processing - reviewing - completed \- failed - retry状态机的实现并不复杂核心是围绕任务ID维护状态字典并且每次状态变化时都会触发一次回调把当前状态推送给相关的Agent。有一次我在测试中发现执行Agent明明已经往总线上发了task_result但规划Agent那边迟迟没有反应。排查了好一会儿才发现是任务状态还挂在“processing”上没有正确更新为“reviewing”。当时就在状态机里加了一个强制校验只有上游状态合法才允许切换到下一个状态。从那以后这类问题基本绝迹了。3.3 角色定义三个核心Agent的职责边界项目X里我一共定义了三种角色正好对应前面提到的三级协作规划AgentPlanner负责任务的拆解、步骤编排、以及最终结果的汇总。它不直接调用底层工具只做“脑力劳动”。执行AgentExecutor负责任务的具体执行包括调用搜索工具、文档解析工具、代码生成工具等。它会读取规划Agent下发的子任务执行并生成结果。审核AgentReviewer负责对执行结果进行评估和纠错。如果发现质量问题打回重做如果通过则上报给规划Agent进行汇总。这三者的关系有点像现实项目里的甲方、干活的和监理。规划者不干活只管方向干活的不决策只管实现监理不讨好任何人只负责挑毛病。角色边界一旦清晰系统行为的可预期性就会大幅提升。4. 实操过程从零搭建一个最小可用的agency-agents框架4.1 环境准备与依赖选型先说环境我的建议是不要在这上面花太多时间纠结版本选一套自己最熟的组合就行。项目X用的是Python 3.10版本模型接口走OpenAI兼容的SDK消息总线从零手写没有引入Celery这类重型任务队列。工具调用方面直接把几个常见的工具封装成了函数注册表方便后续扩展。依赖只需要三个包pydantic用于定义数据结构openai用于调用模型接口python-dotenv用于管理环境变量。如果只跑Demo其实连pydantic都可以不用但为了后续维护方便加上它会舒服很多。4.2 定义消息模型和工具注册表整个框架的代码量其实不大最复杂的部分反而是消息模型的定义。我用Pydantic定义了所有Agent之间通信的结构同时在工具层做了一层注册机制。这样执行Agent就不需要写死“能干什么”而是运行时查注册表判断自己有没有能力处理某个子任务。class ToolRegistry: _tools {} classmethod def register(cls, name): def decorator(func): cls._tools[name] func return func return decorator classmethod def get(cls, name): return cls._tools.get(name) classmethod def has(cls, name): return name in cls._tools之后我用ToolRegistry.register(web_search)这样的方式把一个个真实工具函数挂载进去。这种注册表模式的优点在于扩展性极好想加一个工具不需要改执行Agent的内部逻辑只需要新写一个函数并注册即可。这个设计我在很多实际项目中反复使用成本低效果却非常显著。4.3 核心循环消息路由与任务分发系统的核心引擎其实就是一个持续运行的消息循环。它不断从消息总线里取出消息根据消息的receiver字段找到对应的Agent然后调用Agent的handle方法。听起来很简单但真正实现时有两个细节容易踩坑。第一个细节是消息的幂等性。因为消息循环可能会因为异常而重复读取一条消息如果处理逻辑不是幂等的就会产生重复执行浪费token还污染结果。我的解决方案是给每条消息带上message_id并在已处理集合里留痕重复消息直接丢弃。第二个细节是超时处理。每个Agent处理消息时如果等待模型响应时间过长不能无限等下去。我引入了asyncio.wait_for机制超时后会触发自定义的“AgentBusyError”异常让规划Agent重新调度相邻节点。关键代码大致如下async def dispatch_message(message: AgentMessage): agent agent_registry[message.receiver] try: result await asyncio.wait_for( agent.handle(message), timeout30.0 ) return result except asyncio.TimeoutError: logger.error(fAgent {message.receiver} timeout) raise AgentBusyError(message.receiver, message)4.4 规划Agent如何把需求拆成可执行的子任务列表规划Agent是整个框架里“最像人”的一个环节它负责把一个总任务拆分成多个子任务并为每个子任务指定执行Agent和期望输出格式。系统提示词中必须明确约束它只输出结构化的JSON数组不要输出任何解释性文字否则后面解析起来会非常痛苦。我给规划Agent定义的任务拆分数据结构如下class SubTask(BaseModel): task_id: str description: str agent_role: str requires_tool: list[str] output_format: str每个字段都有它的意义。agent_role决定由哪个角色的Agent执行requires_tool是一个工具列表执行Agent会先检查自己注册表里有没有这些工具如果缺工具就直接返回错误而不是硬着头皮瞎编。实测下来这种“工具预检”机制能挡住至少半数的无效任务。因为模型有时候会展开想象让执行Agent去做它根本没能力做的事与其等它失败不如一开始就拒绝。4.5 执行Agent工具调用与结果标准化执行Agent的工作相对机械但极为重要接收子任务解析需要调用的工具执行工具然后把结果整理成标准格式返回。这里最大的坑在于工具的返回结果五花八门有些是纯文本有些是JSON有些是渲染后的HTML。如果这些结果直接塞回大模型上下文不仅浪费token还可能干扰模型输出格式。所以我在执行Agent里强制加了一个“标准化输出”步骤。不管工具返回什么内容最终往外发消息时payload里的content字段必须是规范化的文本通常不超过1500字。超出部分用截断和摘要处理。这个过程听起来像是在“损失信息”但实际上是为了保住消息流的有序性。毕竟执行Agent的下游是审核Agent它需要的是结构清晰、逻辑自洽的结果而不是一堆原始垃圾。标准化处理的核心逻辑其实就是调用一次轻量级模型做摘要把工具原始返回结果压缩成“核心事实”和“关键引用”两个部分。实测下来这种压缩能让后续所有Agent的推理准确率显著提升因为上下文里的信噪比提高了。4.6 审核Agent质量关卡如何设计审核Agent的设计是整个系统里最容易被低估的环节。很多人觉得审核Agent就是拿另一个模型生成一句“看起来不错”那就大错特错了。真正的审核Agent必须能提出具体的修改建议并且判断结果是否达到“放行”标准。我设计的审核维度有四个完整性结果是否覆盖了任务描述中的所有要求。准确性事实、数据、引用是否可靠是否有一本正经地胡说八道。格式合规性输出结构是否符合output_format的要求。安全与尺度是否包含越界、敏感或低质量的内容。每次审核完成后审核Agent必须返回pass或fix结果。如果返回fix必须附带具体的原因说明和修改建议。规划Agent在收到fix结果后会决定是让原执行Agent带着修改建议重做一次还是换一个执行Agent重做一次。这个“重做策略”在我们的测试中新能解决大约30%的错误案例。5. 验证场景与能力边界跑通了但也看到了天花板5.1 一个完整的任务流转示例为了验证框架可用性项目X里构造了一个“跨领域信息整理”的任务具体内容是调研某行业近半年的发展趋势整理出关键事件列表并针对其中一个事件进行深度分析。这类任务在单Agent模式下往往表现得很一般因为既要检索、又要梳理、还要做分析工作量和上下文复杂度都摆在那里。任务下发后系统经历了一条完整的链路规划Agent拆解出四个子任务趋势概览、关键事件提取、特定事件深度分析、最终汇总。执行Agent依次处理四个子任务调用搜索和文本解析工具每种结果都经过标准化压缩。审核Agent对每个子任务结果做了详细检查其中“特定事件分析”被判定为不达标原因是缺少了来源引用且结论过于情绪化。规划Agent根据修改建议让执行Agent重新跑了那个子任务补充了引用调整了口径二次审核通过。规划Agent汇总所有结果输出最终报告。整个流程大约耗时三分钟调用模型次数在8次左右。如果让单Agent干这件事我和同行比较过大部分情况下会在一千多个token之后开始遗漏早期要求内容变得飘忽不定。而多Agent模式虽然花掉的调用次数更多但每一段结果基本都在预期轨道上。5.2 能力边界多Agent也不是银弹在实际使用中我对这套框架的边界有了清晰认识写在下面方便大家参考规划Agent的任务拆分能力决定了上限。如果规划Agent本身就理解错了需求后面所有Agent再努力也没用。所以对需求模糊、表达不清的任务系统表现会显著下降。工具能力决定了执行Agent的物理上限。我之前试过不注册工具让执行Agent纯靠模型“编造”结果审核Agent居然还能给出“看起来合理”的评价。这说明审核链路再强也挡不住事实数据源的缺失。消息丢失和状态不一致的问题在高并发时会放大。单任务系统里内存消息总线足够用但一旦同时跑几十个任务消息顺序、状态同步都开始变得脆弱。我后来加入了简单的按任务ID分桶的队列才缓解了这个问题。成本控制是绕不开的话题。多Agent协作一定会比单Agent调用更费token。每个Agent各带一套上下文、对话历史累计起来非常惊人。项目X里我强制要求每条消息在进入模型之前必须做一次轻量压缩效果显著但也会引入信息损耗。6. 踩坑记录与排错技巧那些文档里不会写的坑6.1 模型返回JSON格式不稳定怎么治这应该是做Agent应用的人都会遇到的第一个头疼问题。用模型生成JSON结果经常多一个换行、少一个引号甚至带着md代码块标记一并返回。我的处理方式分两步走。第一步在后处理时允许用正则容忍最外层被Markdown代码块包裹的情况。第二步无论模型接口的响应格式写的多标准都必须在解析层做重试兜底。import re, json def extract_json(text): text text.strip() code_block_match re.search(r(?:json)?\s*([\s\S]*?), text) if code_block_match: text code_block_match.group(1).strip() # 尝试去掉首尾多余字符 start text.find({) end text.rfind(}) if start ! -1 and end ! -1: return json.loads(text[start:end 1]) raise ValueError(JSON parse failed)这个方法算不上完美但能解决80%的格式问题。剩下20%的情况我在框架里做了重试机制如果解析失败就把原始响应丢回给模型提示它“上一次输出格式不符合要求请重新输出JSON”。这里有个技巧重试提示里最好附带上上次输出中的具体错误原因而不是笼统地说“格式错误”这样才能给模型足够的信息去自我更正。6.2 上下文污染多Agent之间信息泄露如何隔离在设计初期我犯过一个错误把多个Agent的对话历史放在同一个session里管理结果导致模型的角色设定互相混淆。比如执行Agent在回答里突然带上了审核Agent的口吻一看就是系统提示词被冲撞了。解决办法是每个Agent完全独立的对话session不共享任何上下文。所有跨Agent的信息传递都走消息总线用结构化的payload承载。刚开始我担心这样会显得很“隔”但实测发现只有彻底隔离每个Agent才能稳定扮演自己的角色。共享上下文这个想法看起来很高效实际是灾难。6.3 调试工具如何追踪一条消息的完整生命周期多Agent系统调试时的痛苦源于“黑盒”。模型输出不可控消息流转又交错复杂出了问题很难定位。我的做法是在消息总线入口处加了一个完整的日志记录器每条消息从入队到被消费的每一步都记录时间戳和状态。调试时直接搜索message_id就能看到这消息的完整生命轨迹。这个日志记录器看起来很简单但在排查问题时帮了大忙。有一次系统运行到一半突然挂起我通过日志发现是消息总线的消费队列里卡了一条receiver为“executor”的消息但是对应的执行Agent实例因为之前的异常已经退出了没有人消费它而且没有超时清理机制。加了“孤儿消息扫描”后定期清理超过10分钟未消费的消息这个问题就解决了。6.4 重试风暴与熔断机制多Agent系统还有一个隐蔽问题就是“重试风暴”。当一个执行Agent连续报错时规划Agent可能机械地反复给它下发相同任务导致错误无限放大。我当时在审核Agent里加了一个阈值计数同一个子任务被重试超过3次就会触发告警并直接标记为失败让规划Agent放弃该节点。这个方法源于现实运维里的熔断思想。在流程类系统里无限重试远比一次失败更危险因为每次重试都在消耗真金白银的模型调用和宝贵的时间。设置重试上限其实也是在倒逼规划Agent注意任务拆分的合理性而不是盲目地把“锅”甩给下游。7. 后续可以横向扩展的方向项目X跑通之后我在原有框架的基础上又做了几个方向的延伸尝试其中有些已经验证可用有些仍在打磨中。引入记忆模块目前每个Agent的任务上下文都是即时生成的任务结束后就清空了。如果要让系统具备持续学习能力就得给每个Agent配一个长期记忆存储把历史任务的经验教训沉淀下来。RAG增强工具链执行Agent目前依赖的外部工具比较基础如果接入一个成熟的知识库检索组件让它能访问领域内的文档库执行能力会有一个质的提升。可视化编排界面现在整个流程是代码配置出来的不够直观。后续如果做成配置面板用拖拽方式定义Agent角色和消息路由就能让非技术背景的人也能用上这套框架。动态流程控制虽然现在固定流程很稳但某些业务场景确实需要动态分支我目前在实验用审核Agent的结果自动引导下一步调用路径实现一种“半固定流程”的模式。就我个人的体会而言做“agency-agents”这类项目最大的收获不是代码本身而是通过亲手搭建真正体会到“智能体协作”和“单点调API”之间的鸿沟。角色边界、消息协议、状态管理、失败重试这些东西看起来都是工程上的老生常谈但一旦放到Agent这个新物种身上每一个老问题都会换个马甲重新出现。希望这篇分享能帮到正在折腾同样方向的朋友少走几步弯路。 
长河君 · 2026/10/10 17:26:28 · 阅读 1.2k 从Agent到Agency-Agents：多智能体协作框架的设计与实践
这些年“Agents”这个火热各种框架层出不穷但每次看到“agency-agents”这种命名我还是会忍不住多琢磨几秒。它不像“agent-demo”那么直白也不是某个模型厂商的官方仓库名而更像是一个描述“一群智能体如何组织起来干活”的集合概念。我刚接触这个方向的时候也以为它只是一个简单的学习项目直到自己动手搭了一遍才发现“agent”本身不是难点“agency”也就是多个智能体之间的协作、权责分配、消息流转才是真正烧脑的部分。这篇博文就围绕我在一个模拟项目X里如何把一个“一堆智能体”的混沌想法逐步落地成一个可运行、可扩展的“agency-agents”协作框架。我会把设计思路、架构选择、踩过的坑、还有关键的代码骨架都写出来尽量做到既能让刚入门的朋友看懂也能让已经在做多智能体应用的同学找到点共鸣。1. 破题从“agent”到“agency-agents”到底是多了什么1.1 拆解标题里的三层含义先说结论我认为“agency-agents”这个组合词至少表达了三层意思第一层它是一个智能体的集合。单个agent能处理的问题有限比如让它写一段代码、做一次检索、整理一份摘要这些都算是单点能力。而“agency-agents”天然暗示着“一批”智能体同时存在彼此之间有关系、有分工。第二层它强调“代理机构”式的组织感。在真实世界里一家agency有策划、有执行、有审核、有对接客户的岗位。映射到系统里就是不同的agent扮演不同职能角色而不是一堆同样的agent各干各的。这种角色化设计是多智能体系统和“并行调用多个模型”最本质的区别。第三层它指代“可以协作的框架”。既然是复数又带有组织含义那实现上就必须有通信机制、任务分发机制、状态同步机制。哪怕只是一个很简化的实现也得有这些骨架否则顶多算一个“多进程脚本”。我当时拿到“agency-agents”这个题目时第一反应是需求方到底想要一个能用的产品原型还是一个用来验证多智能体协作概念的技术Demo后来通过和某导师沟通确认了定位是后者——一个可扩展的、代码清晰的参考框架目的是验证“多角色智能体协作解决复杂任务”的可行性。1.2 为什么单Agent模式不能满足需求很多人刚接触大模型应用时会习惯性地把任务丢给一个超长的Prompt指望一个Agent全流程搞定。我也走过这条路实际结果通常是Prompt写得越长模型越容易丢失早期指令。尤其是超过上下文窗口的临界点之后前面的“角色设定”和“步骤要求”会被后续内容稀释甚至是遗忘。任务一旦出现异常分支单个Agent只能靠着“if-else”式的自我推理去兜底可它往往既要做决策又要做执行还得自查错误脑袋里塞了太多事效果自然下降。单Agent模式下所有的工具调用、中间结果、上下文记忆都耦合在一起出了问题很难定位是模型推理错了还是工具返回异常还是上下文被污染排查成本极高。而“agency-agents”模式就好比把一家公司的流程搬到系统里策划Agent只负责拆解目标和制定计划执行Agent只负责调用工具和生成内容审核Agent只负责挑错。各司其职上下文被隔离问题域被切分。实测下来整体成功率比单Agent硬扛要稳定得多。2. 方案选型为什么我选择“规划-执行-审核”三级协作2.1 常见的多Agent协作模式对比多智能体协作的实现方式其实有很多种我之前大概梳理过三类主流模式各有各的适用场景模式核心思想优点缺点管道模式智能体按阶段串行处理前一个的输出是后一个的输入流程清晰、容易调试链路长时延迟高且中间环节出问题会中断整体流程编排模式有一个“规划者”负责任务分解和调度执行者听从分配职责清晰、扩展性好规划者一旦判断失误整个任务方向就会跑偏协商模式多个智能体地位平等通过讨论达成共识适合开放式问题实现复杂容易陷入无休止的来回对话成本不可控这次的项目X采用的是“编排模式”的变体叫“规划-执行-审核”。选择它不是因为它最先进而是因为它在可控性和效果之间最平衡。项目要求是用尽量简单的方式验证多智能体的价值那就必须让责任边界足够清晰每一个环节出的问题都能被单独观测和修正。2.2 固定流程编排的取舍在项目初期我也考虑过要不要上更“智能”的动态编排也就是让模型自己决定下一步该调用哪个Agent。后来果断放弃了原因有三点都是实操层面的动态编排意味着每一个Agent返回的格式都可能是半结构化的解析成本高错误率高。模型生成的JSON字段经常缺胳膊少腿处理这些“格式垃圾”的时间比写业务逻辑还长。协商模式的token消耗非常大。一次任务来回二十轮对话成本翻好几倍而项目X只是验证概念不需要无限探索。动态编排的调试体验极其痛苦。今天能跑通的链路明天换一个模型版本可能就走不通了完全不可控。所以最后采用了一个看似“笨”但非常可靠的方法固定流程 角色隔离 节点级重试。也就是流程的骨架是固定的但每个节点内部允许根据输出质量自动触发一次重试或修复。这种方式既保留了多Agent协作的优势又把实现复杂度控制在了合理范围内。3. 核心架构消息总线、任务状态机和角色定义3.1 消息总线Agent之间怎么说话在“agency-agents”框架里Agent之间不能直接互相调用方法否则就变成强耦合了。我参考了消息队列的思路实现了一个轻量级的内存消息总线。所有的Agent实例只和总线打交道不关心消息是谁发的也不关心最终谁会消费。消息总线的数据结构非常简单但非常关键。我定义了这样几个字段dataclass class AgentMessage: message_id: str sender: str receiver: str msg_type: str # task_request / task_result / review_report / error payload: dict timestamp: float retry_count: int 0别小看这个结构。“receiver”字段决定了消息是点对点发送还是广播“msg_type”字段则方便消费者快速路由。之前我见过很多项目把所有信息都塞进payload然后用一个大的type字段区分结果后面根本没法维护。这样拆开之后每个Agent收到消息时只需要匹配自己关心的msg_type其他一律忽略逻辑非常清爽。3.2 任务状态机一个任务从生到死的完整轨迹多智能体系统中最容易“失控”的地方就是任务状态。A任务被B、C、D三个Agent同时处理时谁先完成谁在等待谁报错了这些必须一目了然。所以我在框架里建了一个任务状态机流转路径是pending - queued - dispatching - processing - reviewing - completed \- failed - retry状态机的实现并不复杂核心是围绕任务ID维护状态字典并且每次状态变化时都会触发一次回调把当前状态推送给相关的Agent。有一次我在测试中发现执行Agent明明已经往总线上发了task_result但规划Agent那边迟迟没有反应。排查了好一会儿才发现是任务状态还挂在“processing”上没有正确更新为“reviewing”。当时就在状态机里加了一个强制校验只有上游状态合法才允许切换到下一个状态。从那以后这类问题基本绝迹了。3.3 角色定义三个核心Agent的职责边界项目X里我一共定义了三种角色正好对应前面提到的三级协作规划AgentPlanner负责任务的拆解、步骤编排、以及最终结果的汇总。它不直接调用底层工具只做“脑力劳动”。执行AgentExecutor负责任务的具体执行包括调用搜索工具、文档解析工具、代码生成工具等。它会读取规划Agent下发的子任务执行并生成结果。审核AgentReviewer负责对执行结果进行评估和纠错。如果发现质量问题打回重做如果通过则上报给规划Agent进行汇总。这三者的关系有点像现实项目里的甲方、干活的和监理。规划者不干活只管方向干活的不决策只管实现监理不讨好任何人只负责挑毛病。角色边界一旦清晰系统行为的可预期性就会大幅提升。4. 实操过程从零搭建一个最小可用的agency-agents框架4.1 环境准备与依赖选型先说环境我的建议是不要在这上面花太多时间纠结版本选一套自己最熟的组合就行。项目X用的是Python 3.10版本模型接口走OpenAI兼容的SDK消息总线从零手写没有引入Celery这类重型任务队列。工具调用方面直接把几个常见的工具封装成了函数注册表方便后续扩展。依赖只需要三个包pydantic用于定义数据结构openai用于调用模型接口python-dotenv用于管理环境变量。如果只跑Demo其实连pydantic都可以不用但为了后续维护方便加上它会舒服很多。4.2 定义消息模型和工具注册表整个框架的代码量其实不大最复杂的部分反而是消息模型的定义。我用Pydantic定义了所有Agent之间通信的结构同时在工具层做了一层注册机制。这样执行Agent就不需要写死“能干什么”而是运行时查注册表判断自己有没有能力处理某个子任务。class ToolRegistry: _tools {} classmethod def register(cls, name): def decorator(func): cls._tools[name] func return func return decorator classmethod def get(cls, name): return cls._tools.get(name) classmethod def has(cls, name): return name in cls._tools之后我用ToolRegistry.register(web_search)这样的方式把一个个真实工具函数挂载进去。这种注册表模式的优点在于扩展性极好想加一个工具不需要改执行Agent的内部逻辑只需要新写一个函数并注册即可。这个设计我在很多实际项目中反复使用成本低效果却非常显著。4.3 核心循环消息路由与任务分发系统的核心引擎其实就是一个持续运行的消息循环。它不断从消息总线里取出消息根据消息的receiver字段找到对应的Agent然后调用Agent的handle方法。听起来很简单但真正实现时有两个细节容易踩坑。第一个细节是消息的幂等性。因为消息循环可能会因为异常而重复读取一条消息如果处理逻辑不是幂等的就会产生重复执行浪费token还污染结果。我的解决方案是给每条消息带上message_id并在已处理集合里留痕重复消息直接丢弃。第二个细节是超时处理。每个Agent处理消息时如果等待模型响应时间过长不能无限等下去。我引入了asyncio.wait_for机制超时后会触发自定义的“AgentBusyError”异常让规划Agent重新调度相邻节点。关键代码大致如下async def dispatch_message(message: AgentMessage): agent agent_registry[message.receiver] try: result await asyncio.wait_for( agent.handle(message), timeout30.0 ) return result except asyncio.TimeoutError: logger.error(fAgent {message.receiver} timeout) raise AgentBusyError(message.receiver, message)4.4 规划Agent如何把需求拆成可执行的子任务列表规划Agent是整个框架里“最像人”的一个环节它负责把一个总任务拆分成多个子任务并为每个子任务指定执行Agent和期望输出格式。系统提示词中必须明确约束它只输出结构化的JSON数组不要输出任何解释性文字否则后面解析起来会非常痛苦。我给规划Agent定义的任务拆分数据结构如下class SubTask(BaseModel): task_id: str description: str agent_role: str requires_tool: list[str] output_format: str每个字段都有它的意义。agent_role决定由哪个角色的Agent执行requires_tool是一个工具列表执行Agent会先检查自己注册表里有没有这些工具如果缺工具就直接返回错误而不是硬着头皮瞎编。实测下来这种“工具预检”机制能挡住至少半数的无效任务。因为模型有时候会展开想象让执行Agent去做它根本没能力做的事与其等它失败不如一开始就拒绝。4.5 执行Agent工具调用与结果标准化执行Agent的工作相对机械但极为重要接收子任务解析需要调用的工具执行工具然后把结果整理成标准格式返回。这里最大的坑在于工具的返回结果五花八门有些是纯文本有些是JSON有些是渲染后的HTML。如果这些结果直接塞回大模型上下文不仅浪费token还可能干扰模型输出格式。所以我在执行Agent里强制加了一个“标准化输出”步骤。不管工具返回什么内容最终往外发消息时payload里的content字段必须是规范化的文本通常不超过1500字。超出部分用截断和摘要处理。这个过程听起来像是在“损失信息”但实际上是为了保住消息流的有序性。毕竟执行Agent的下游是审核Agent它需要的是结构清晰、逻辑自洽的结果而不是一堆原始垃圾。标准化处理的核心逻辑其实就是调用一次轻量级模型做摘要把工具原始返回结果压缩成“核心事实”和“关键引用”两个部分。实测下来这种压缩能让后续所有Agent的推理准确率显著提升因为上下文里的信噪比提高了。4.6 审核Agent质量关卡如何设计审核Agent的设计是整个系统里最容易被低估的环节。很多人觉得审核Agent就是拿另一个模型生成一句“看起来不错”那就大错特错了。真正的审核Agent必须能提出具体的修改建议并且判断结果是否达到“放行”标准。我设计的审核维度有四个完整性结果是否覆盖了任务描述中的所有要求。准确性事实、数据、引用是否可靠是否有一本正经地胡说八道。格式合规性输出结构是否符合output_format的要求。安全与尺度是否包含越界、敏感或低质量的内容。每次审核完成后审核Agent必须返回pass或fix结果。如果返回fix必须附带具体的原因说明和修改建议。规划Agent在收到fix结果后会决定是让原执行Agent带着修改建议重做一次还是换一个执行Agent重做一次。这个“重做策略”在我们的测试中新能解决大约30%的错误案例。5. 验证场景与能力边界跑通了但也看到了天花板5.1 一个完整的任务流转示例为了验证框架可用性项目X里构造了一个“跨领域信息整理”的任务具体内容是调研某行业近半年的发展趋势整理出关键事件列表并针对其中一个事件进行深度分析。这类任务在单Agent模式下往往表现得很一般因为既要检索、又要梳理、还要做分析工作量和上下文复杂度都摆在那里。任务下发后系统经历了一条完整的链路规划Agent拆解出四个子任务趋势概览、关键事件提取、特定事件深度分析、最终汇总。执行Agent依次处理四个子任务调用搜索和文本解析工具每种结果都经过标准化压缩。审核Agent对每个子任务结果做了详细检查其中“特定事件分析”被判定为不达标原因是缺少了来源引用且结论过于情绪化。规划Agent根据修改建议让执行Agent重新跑了那个子任务补充了引用调整了口径二次审核通过。规划Agent汇总所有结果输出最终报告。整个流程大约耗时三分钟调用模型次数在8次左右。如果让单Agent干这件事我和同行比较过大部分情况下会在一千多个token之后开始遗漏早期要求内容变得飘忽不定。而多Agent模式虽然花掉的调用次数更多但每一段结果基本都在预期轨道上。5.2 能力边界多Agent也不是银弹在实际使用中我对这套框架的边界有了清晰认识写在下面方便大家参考规划Agent的任务拆分能力决定了上限。如果规划Agent本身就理解错了需求后面所有Agent再努力也没用。所以对需求模糊、表达不清的任务系统表现会显著下降。工具能力决定了执行Agent的物理上限。我之前试过不注册工具让执行Agent纯靠模型“编造”结果审核Agent居然还能给出“看起来合理”的评价。这说明审核链路再强也挡不住事实数据源的缺失。消息丢失和状态不一致的问题在高并发时会放大。单任务系统里内存消息总线足够用但一旦同时跑几十个任务消息顺序、状态同步都开始变得脆弱。我后来加入了简单的按任务ID分桶的队列才缓解了这个问题。成本控制是绕不开的话题。多Agent协作一定会比单Agent调用更费token。每个Agent各带一套上下文、对话历史累计起来非常惊人。项目X里我强制要求每条消息在进入模型之前必须做一次轻量压缩效果显著但也会引入信息损耗。6. 踩坑记录与排错技巧那些文档里不会写的坑6.1 模型返回JSON格式不稳定怎么治这应该是做Agent应用的人都会遇到的第一个头疼问题。用模型生成JSON结果经常多一个换行、少一个引号甚至带着md代码块标记一并返回。我的处理方式分两步走。第一步在后处理时允许用正则容忍最外层被Markdown代码块包裹的情况。第二步无论模型接口的响应格式写的多标准都必须在解析层做重试兜底。import re, json def extract_json(text): text text.strip() code_block_match re.search(r(?:json)?\s*([\s\S]*?), text) if code_block_match: text code_block_match.group(1).strip() # 尝试去掉首尾多余字符 start text.find({) end text.rfind(}) if start ! -1 and end ! -1: return json.loads(text[start:end 1]) raise ValueError(JSON parse failed)这个方法算不上完美但能解决80%的格式问题。剩下20%的情况我在框架里做了重试机制如果解析失败就把原始响应丢回给模型提示它“上一次输出格式不符合要求请重新输出JSON”。这里有个技巧重试提示里最好附带上上次输出中的具体错误原因而不是笼统地说“格式错误”这样才能给模型足够的信息去自我更正。6.2 上下文污染多Agent之间信息泄露如何隔离在设计初期我犯过一个错误把多个Agent的对话历史放在同一个session里管理结果导致模型的角色设定互相混淆。比如执行Agent在回答里突然带上了审核Agent的口吻一看就是系统提示词被冲撞了。解决办法是每个Agent完全独立的对话session不共享任何上下文。所有跨Agent的信息传递都走消息总线用结构化的payload承载。刚开始我担心这样会显得很“隔”但实测发现只有彻底隔离每个Agent才能稳定扮演自己的角色。共享上下文这个想法看起来很高效实际是灾难。6.3 调试工具如何追踪一条消息的完整生命周期多Agent系统调试时的痛苦源于“黑盒”。模型输出不可控消息流转又交错复杂出了问题很难定位。我的做法是在消息总线入口处加了一个完整的日志记录器每条消息从入队到被消费的每一步都记录时间戳和状态。调试时直接搜索message_id就能看到这消息的完整生命轨迹。这个日志记录器看起来很简单但在排查问题时帮了大忙。有一次系统运行到一半突然挂起我通过日志发现是消息总线的消费队列里卡了一条receiver为“executor”的消息但是对应的执行Agent实例因为之前的异常已经退出了没有人消费它而且没有超时清理机制。加了“孤儿消息扫描”后定期清理超过10分钟未消费的消息这个问题就解决了。6.4 重试风暴与熔断机制多Agent系统还有一个隐蔽问题就是“重试风暴”。当一个执行Agent连续报错时规划Agent可能机械地反复给它下发相同任务导致错误无限放大。我当时在审核Agent里加了一个阈值计数同一个子任务被重试超过3次就会触发告警并直接标记为失败让规划Agent放弃该节点。这个方法源于现实运维里的熔断思想。在流程类系统里无限重试远比一次失败更危险因为每次重试都在消耗真金白银的模型调用和宝贵的时间。设置重试上限其实也是在倒逼规划Agent注意任务拆分的合理性而不是盲目地把“锅”甩给下游。7. 后续可以横向扩展的方向项目X跑通之后我在原有框架的基础上又做了几个方向的延伸尝试其中有些已经验证可用有些仍在打磨中。引入记忆模块目前每个Agent的任务上下文都是即时生成的任务结束后就清空了。如果要让系统具备持续学习能力就得给每个Agent配一个长期记忆存储把历史任务的经验教训沉淀下来。RAG增强工具链执行Agent目前依赖的外部工具比较基础如果接入一个成熟的知识库检索组件让它能访问领域内的文档库执行能力会有一个质的提升。可视化编排界面现在整个流程是代码配置出来的不够直观。后续如果做成配置面板用拖拽方式定义Agent角色和消息路由就能让非技术背景的人也能用上这套框架。动态流程控制虽然现在固定流程很稳但某些业务场景确实需要动态分支我目前在实验用审核Agent的结果自动引导下一步调用路径实现一种“半固定流程”的模式。就我个人的体会而言做“agency-agents”这类项目最大的收获不是代码本身而是通过亲手搭建真正体会到“智能体协作”和“单点调API”之间的鸿沟。角色边界、消息协议、状态管理、失败重试这些东西看起来都是工程上的老生常谈但一旦放到Agent这个新物种身上每一个老问题都会换个马甲重新出现。希望这篇分享能帮到正在折腾同样方向的朋友少走几步弯路。