从单个智能体到代理机构我用一份代码把多个AI Agents组织成了能自己分工的团队最近在折腾多智能体编排手头一个编号为agency-agents的项目让我彻底改变了写AI应用的方式。简单说它是一套多代理Multi-Agent协作框架核心思想不是再写一个什么都能干的超级Agent而是把多个专职Agent组织成一个类似代理机构的团队让它们各自负责调研、执行、审查、汇总等环节像公司里的部门一样配合着把复杂任务拆掉。这套东西解决的核心痛点很明确单个Agent在面对稍微复杂的任务时容易贪多嚼不烂要么上下文一长就跑偏要么什么都想干、什么都干不精细。适合正在做AI Agent应用、想从单代理升级到多代理架构的开发者参考。下面我把整个项目的设计思路、关键实现、踩坑记录都拆开讲清楚保证你读完能照着搭一个自己的Agent机构。1. 项目定位为什么需要代理机构式的多智能体架构1.1 从单Agent到多Agent不只是多开几个实例很多朋友一开始理解的多Agent就是把同一个Agent prompt复制三份然后并发跑三个任务。这其实只是多线程调用不叫多智能体协作。真正的多Agent系统指的是多个具有不同角色定义、不同上下文窗口、不同工具权限的Agent在同一个任务流中按照某种协议进行协作。我最初做单Agent应用时遇到两个很典型的问题。第一是上下文污染一个Agent既需要处理用户输入、又要查数据库、还要生成报告所有中间过程全部堆在同一个上下文里很快就把Token窗口占满而且越到后面回答质量越差因为早期信息被稀释了。第二是职责混乱让同一个Agent既当运动员又当裁判它往往会倾向于认可自己生成的内容很难客观审查。比如让它写一段代码再自己检查Bug它经常检查不出来因为它觉得自己写得没错。这就引出了agency-agents项目的核心前提把任务拆成不同的角色不同角色由不同Agent承担每个Agent只关注自己的一亩三分地。这就像一家真实的中介机构有前台接待、有业务顾问、有风控审核、有签约专员每个人只做自己最专业的那一环。1.2 Agency模式究竟在解决什么问题我理解agency-agents这个名字里的agency不是指代理权agent的复数而是指代理机构agency的组织形态。它的核心价值有四点降低单一上下文的负载每个Agent独立维护自己的上下文只接收与自己职责相关的信息。前端Agent不需要背着一整份数据库schema去回答用户问题。提升输出质量专职Agent对单一任务的prompt调优更容易比如审查Agent只做代码审查它的评分标准和检查清单可以做到非常细致远胜于一个通用Agent顺带做审查。支持并行与流水线某些环节可以并行比如调研阶段可以同时启动市场分析Agent和技术调研Agent两个结果汇总到写作Agent这个用单Agent很难优雅实现。关键的是可观测性多Agent架构天然带来了中间产物。你能看到哪个Agent把任务转给了谁、转交时携带了什么信息出了问题可以精确定位到环节而不是对着一个大黑盒发呆。我在设计这个项目时就明确了目标做一个轻量的、不依赖重量级框架的多Agent协作骨架核心只做三件事——角色注册、任务流转、结果汇总。2. 架构设计与关键决策2.1 角色划分谁来决策、谁来执行、谁来审查在agency-agents项目里我把Agent分成了三类角色这个划分是参考了现实公司的运作逻辑协调者Coordinator相当于公司里的项目经理。它负责接收用户需求拆解任务决定派哪个Agent去做并在任务完成后向用户汇报。协调者不直接干活它只做任务调度和结果整合。执行者Worker指具体干活的Agent比如代码编写Agent、文案生成Agent、数据分析Agent。它们只关心自己拿到的那份子任务输出结构化结果交给协调者。审查者Reviewer相当于QA或风控。它拿到执行者的产出按照预设的检查规则做验证判断通过还是打回。这个角色的存在是整个系统质量的关键。这样的角色划分带来一个设计原则每个Agent的system prompt必须聚焦不能既要调度又要干活还要审查。一旦出现职能重叠整个系统的行为就变得不可预测这是我在早期版本里踩过最大的坑。2.2 协作协议任务怎么流转角色定义清楚了接下来就是任务流转机制。在多Agent系统里Agent之间的通信方式直接决定了系统的复杂度。我经历过三种方案最终固定下来一种第一种是自由对话模式所有Agent共享一个消息队列谁想说什么就说什么。这种方式看起来灵活实际上一旦Agent数量超过三个对话就会发散消息主题随时漂移最后很难收敛到任务结果。第二种是中心总线模式所有消息都经过一个MessageBus按主题路由。这个方案扩展性最好但对小型项目来说配置太重调试时到处都是消息人容易看花眼。最终采用的是协调者-执行者模式这是目前最推荐给中小型项目的方案。核心规则就一条所有任务都从协调者发出所有结果都回到协调者。执行者之间不直接通信如果一个执行者需要另一个执行者的输出由协调者转交。这个模式的好处非常明显它把系统拓扑从网状变成了星型便于追踪每条任务链的状态。代价是协调者的上下文会承载比较多的转发信息但这个代价在项目规模可控时完全值得。2.3 工具与上下文管理每个Agent的权限边界多Agent系统里还有一个容易被忽视的设计点——工具权限。不是所有Agent都需要所有工具。比如数据分析Agent需要数据库查询权限但它不应该有发送邮件的权限审查Agent可能需要代码执行沙箱但不需要访问用户聊天记录。我在项目里给每个Agent配置了一组独立的tool whitelist协调者在下发任务时把相关工具列表一并传给执行者。这个设计在初期增加了配置量但后期带来的收益很大安全隔离某个Agent的工具权限被滥用影响范围可控上下文精简Agent不需要看到所有工具的说明只加载自己用得到的工具定义省Token行为可预测审查Agent只能用审查工具它就不可能顺手去改代码上下文管理上我采用了按需注入策略。每个Agent的基础信息角色定义、输出格式常驻上下文任务相关的动态信息具体要处理的数据、上一级Agent的产出在每次任务下发时注入。任务完成后动态信息从上下文中移除这样Agent不会记得太多无关的历史。3. 实操实现如何从零搭建一套agency-agents系统3.1 技术选型与环境准备动手前先说选型。我没有选择业内那几个重量级多Agent编排框架而是基于一套成熟的大模型调用SDK直接手写编排逻辑。原因是这个项目的核心诉求是理解多Agent协作的机制而不是跑通一个框架的Hello World。自己写一遍虽然代码量多一些但对每个环节的把控感是完全不同的。环境依赖就三样Python 3.11 及以上大模型API的Python SDK用于调用LLM一个轻量配置库可选用于读取Agent配置如果你从零开始建议先建一个虚拟环境避免依赖混乱。装好SDK后验证一下API连通性再开始写代码。3.2 核心数据结构Agent、Task、Message在写具体逻辑之前先把三个核心数据结构定义清楚。我在项目里用的版本如下精简版from dataclasses import dataclass, field from typing import List, Optional, Callable, Any from enum import Enum class RoleType(Enum): COORDINATOR coordinator WORKER worker REVIEWER reviewer dataclass class Agent: name: str role: RoleType system_prompt: str tools: List[Callable] field(default_factorylist) model_config: dict field(default_factorydict) # 每个Agent维护自己的消息历史 memory: List[dict] field(default_factorylist) def run(self, task: str, additional_context: Optional[dict] None) - str: # 组装messages并调用LLM ...这个Agent类是整个项目的地基。注意我加了role字段后续所有调度逻辑都依赖这个角色类型来判断行为方式。然后是任务和消息结构dataclass class Task: id: str description: str assigned_agent: str depends_on: List[str] field(default_factorylist) # 依赖的上游任务ID result: Optional[str] None dataclass class Message: sender: str receiver: str task_id: str content: str kind: str task # task / result / reviewdepends_on字段是任务依赖关系的关键。比如写代码任务完成后审查代码任务才被触发这就是依赖通过列表形式表达可以支持多个上游任务全部完成后再启动下游任务。3.3 协调者核心逻辑任务拆解与调度协调者是系统的大脑它的核心任务是把用户的一句话拆成可以分配给不同Agent执行的子任务。我这里用了一个比较务实的方案让LLM先做一次任务规划然后按规划逐个派发。规划阶段的prompt我大致是这样设计的为了简洁缩略了部分细节你是项目协调者。你收到的用户需求需要拆解为多个子任务。 请输出JSON格式的任务规划包含每个子任务的 - id: 简短唯一标识 - description: 清晰的任务说明 - agent: 建议负责的Agent名称 - depends_on: 依赖的上游任务id列表 要求 1. 每个子任务只属于单一职责域 2. 如果任务之间没有依赖可以并行 3. 不要拆分出无意义的碎任务把规划结果解析成Task列表后协调者进入调度循环def schedule(self, tasks: List[Task]): completed {} pending tasks[:] while pending: # 找出所有上游任务已完成的待执行task ready [ t for t in pending if all(dep in completed for dep in t.depends_on) ] if not ready: # 依赖不满足且没有可推进任务 - 死锁需要报错 raise RuntimeError(task dependencies can not be satisfied) for task in ready: agent self.agents[task.assigned_agent] task.result agent.run(task.description) completed[task.id] task pending.remove(task) return completed这个调度循环看起来简单但有一个关键细节必须先判断依赖是否全部完成再决定是否执行当前任务。这个逻辑如果在并场景下稍微复杂在单线程版本里必须保证顺序正确。我最初写的时候直接把pending里的顺序当执行顺序结果遇到依赖关系就出错。3.4 审查者逻辑验证与打回执行者完成任务后结果不是直接交给用户而是先经过审查者。审查者的prompt核心是给出一套明确的验收标准。比如代码审查Agent的system prompt包含这些要点你负责审查代码质量。请检查以下维度 1. 可读性变量命名是否清晰 2. 健壮性是否有明显边界错误 3. 安全性是否有注入风险或敏感信息泄露 4. 与任务需求的匹配度是否满足任务描述 输出格式 - decision: pass 或 fails - reasons: 具体问题列表 - suggestions: 修改建议审查结果返回协调者后协调者需要做一次决策如果decision是pass则汇总结果如果是fails则把审查意见附在原任务描述后面重新派发给原执行者。def review_and_maybe_reassign(self, task: Task, reviewer: Agent): review_result reviewer.run( f请审查以下执行结果{task.result} ) if review_result.decision pass: return task, True # 打回带上审查意见重新执行 revised task.description \n[审查意见]\n review_result.suggestions task.result self.agents[task.assigned_agent].run(revised) return self.review_and_maybe_reassign(task, reviewer)这里有一个递归调用但要注意必须设置最大打回次数否则理论上可能无限循环。我设置了最多3次超过次数就强制接受当前版本并记录警告。这个强制收敛机制非常关键后面在常见问题里我会细说。3.5 完整流程演示一个模拟任务跑通为了让你直观感受整个系统怎么跑我模拟一个典型任务设计并实现一个用户注册接口包含邮箱验证功能并提供代码审查报告。这个任务在协调者的规划阶段会被拆成约三个子任务task-1: 接口设计与数据库字段定义 - Agent: 架构师Agent task-2: Python代码实现 - Agent: 编码Agent task-3: 代码审查与安全性检查 - Agent: 审查Agent 依赖关系task-2 depends_on task-1task-3 depends_on task-2调度过程是这样走的第一步task-1没有依赖立即执行。架构师Agent输出一份design doc包含接口路径、请求参数、数据库表结构和字段类型存入completed[task-1]。第二步task-2的依赖task-1完成编码Agent被唤醒。我把task-1的结果作为附加上下文注入编码Agent的消息中它基于设计文档写代码。这一步输出的代码会包含测试用例和README片段。第三步task-3的依赖task-2完成审查Agent开始工作。它对代码做静态检查可能会挑出邮箱格式校验不够严格密码字段未加密处理等问题然后返回fails。协调者把审查意见附上重新派发给编码Agent修订。这次修订完成后再次审查如果通过协调者汇总task-1的设计文档、task-2的代码和task-3的审查报告生成最终交付文本给用户。这个流程看着简单但实际跑起来时你会发现每个环节都有值得打磨的细节。比如架构师Agent输出的design doc能不能被编码Agent正确理解取决于两边的输出格式约束是否一致。如果架构师Agent输出的是markdown格式的混排文本编码Agent解析起来会很痛苦。解决方案是在prompt中强制要求结构化字段输出JSON甚至定义一份共享的schema。4. 常见问题与排查实录4.1 死循环与任务收敛问题多Agent系统最容易遇到的问题之一就是无限循环。表现是执行者和审查者来回踢皮球你改完我审、我打回你改来回十几次都不消停。第一次遇到这个现象时我以为是Agent在吵架后来发现根因在于审查意见没有约束力。如果审查Agent的反馈过于模糊比如只说代码不够好请改进执行者完全没有抓手只能漫无目的地改越改越差。解决办法两个审查意见必须结构化强制审查Agent输出具体问题点修改建议不允许泛泛而谈。我在3.4那节里提到的输出格式就是这个目的。打回次数硬上限设定最大打回次数比如3次。超过次数后可以选择强制通过或者把审查Agent最后的意见直接呈现给用户让人类决策。另外要提醒一点重试逻辑里最好加入上下文截断机制。执行者被反复打回时它的上下文会保存着每一次审查-修改的历史这些历史会干扰它判断。比较有效的做法是每次重试时只保留最初的任务描述和最新的审查意见丢弃中间轮次的修改记录。4.2 上下文溢出与记忆管理第二个高频问题是Agent上下文溢出。在多Agent系统里这个问题比单Agent更容易发生因为每一个Agent都可能在多次任务中累积历史。我踩过的具体场景是协调者一直挂着所有任务的历史结果到第5个任务时上下文里堆了前4个任务的所有输出加上这第5个任务的大输入直接超出上下文窗口。解决的思路是从让Agent记住所有转向只让Agent记住必要协调者的消息历史只保留任务列表和每个任务的摘要完整执行结果存储到内存或临时文件里不放进上下文执行者每次任务执行前重置它的对话历史只注入本轮任务的描述和相关上下文这在架构上给出的启示是Agent的记忆不等于LLM上下文。可以把中间产物当作文件系统来理解——写入磁盘需要时读取不留存在上下文中。这样既节省Token又避免信息稀释带来的质量下降。4.3 输出格式不稳定导致下游解析失败多Agent系统里Agent的输出往往不是给人看的而是给下一个Agent或协调者解析的。这就特别依赖输出格式的稳定性。但LLM天然输出不稳定同样的prompt这次返回了合法JSON下次可能开头多一句好的以下是结果直接把JSON解析器干崩。我的对策是两个第一在prompt中给一个正例明确告诉LLM只输出JSON不要输出任何多余内容。对绝大多数主流模型这个约束能显著降低误格式概率但不能完全消除。第二写一个容错解析函数对返回的字符串做预处理截取第一个{到最后一个}之间的部分或者用正则提取代码块再丢给JSON解析器。这样即使模型多说了废话也能强行提取出结构化数据。import json import re def robust_json_parse(text: str) - dict: # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 提取JSON对象部分 match re.search(r\{.*\}, text, re.DOTALL) if match: return json.loads(match.group()) raise ValueError(fcannot parse json from: {text})这个函数看起来不起眼但在多Agent系统中能救你很多次。建议所有下游解析环节都先过一遍这个容错函数再判断错误。4.4 多Agent测试期如何快速定位问题最后分享一个调试技巧。多Agent系统出问题时最难的往往不是修复Bug而是定位是哪个环节出了问题。我的做法是在每个Agent的run方法入口和出口都打日志记录输入摘要和输出摘要而不是完整内容。def run(self, task: str, additional_context: Optional[dict] None) - str: print(f[Agent:{self.name}] 收到任务: {task[:100]}) ... result self._call_llm(messages) print(f[Agent:{self.name}] 输出: {result[:100]}) return result日志的关键不是记录完整信息而是记录必要的信息片段。加上时间戳你很容易看到一条任务在哪个Agent上停留时间过长或者哪个Agent的输出明显异常比如空字符串、重复内容。多Agent系统的排查本质上就是沿着消息流走查日志日志越结构化排查越快。5. 经验总结与扩展方向5.1 我在实操中的几点体会搭建agency-agents这个项目让我对多Agent系统有了几个很实在的认知分享给你第一多Agent不是银弹。如果你的任务本身很单一比如把这段文字翻译成英文强行拆成多Agent完全是过度设计增加延迟和出错点。多Agent最适合的是有明显的不同知识域、不同判断标准需要交替介入的任务比如写代码和审代码就是天然适合分开的两个环节因为它们的评价标准完全不同。第二角色prompt的质量比Agent数量重要。一个职责清晰、边界明确的Agent胜过三个模糊Agent。我在调优时花的80%时间都在prompt边界定义上——什么样的输入可以接受、什么样的输入应该拒绝、输出格式必须是怎样的。第三系统收敛比系统能力更重要。真实使用中用户最烦的不是Agent能力弱而是一个任务反复执行却看不到终点。所以设计中必须预留收敛阀门最大重试次数、超时时间、强制输出默认值的兜底逻辑。没有这些机制多Agent系统在真实场景里会变成多Agent失控。5.2 这个项目后续可以怎么扩展如果你跟着思路搭了一个最小版本有几个方向可以做进一步扩展加入并行执行当前的调度循环是串行的。如果任务之间没有依赖可以并发执行以提升吞吐量。Python里用concurrent.futures就能很快实现。引入工具调用Function Calling让执行者不仅输出文本还能实际调用工具。这需要在Agent运行逻辑里加入工具选择与执行的步骤我项目里预留了tools字段正是为这个准备的。支持多模型混合不同Agent使用不同的模型。协调者用推理能力强但速度慢的旗舰模型执行者用性价比高的快速模型审查者用注重细节的模型可以在成本和质量之间取平衡。把Agent配置外置将Agent的角色定义、工具列表、模型参数放到配置文件里这样调整Agent行为不用改代码改YAML即可。这几个方向里我最推荐先做并行执行和工具调用它们能立竿见影地提升系统的实用价值。最后说一句心里话多Agent架构真正让我惊喜的地方不在于它把任务完成得多快多好而在于它改变了我思考问题的方式——先问这个任务应该拆成几个什么角色来做而不是这个Agent能不能一个顶仨。这种组织思维才是agency-agents这个项目最核心的收获。如果你也在做Agent应用建议别急着上复杂框架先自己手写一个最小的多Agent协作骨架跑通一遍全流程你会对每个设计决策的理解深入很多。