多智能体协作框架agency-agents：架构拆解与实操复现

1. 从“agency-agents”这个名字说起一个多智能体协作框架的拆解与复现第一次看到“agency-agents”这个项目名我的直觉是这大概率是一个围绕“代理agent”构建的协作系统而且“agency”这个词暗示了它不只是单个代理的简单封装而是强调代理的自主性、组织性和协同能力。在当前的开发社区里agent 这个词已经被用得很泛了——从简单的 API 调用封装到复杂的多角色决策系统都可以叫 agent。但“agency”加“agents”的组合通常指向一个更具体的场景多个具备一定自主决策能力的代理在一个统一的框架下分工协作完成单个代理难以独立完成的任务。我花了一些时间梳理这个项目的核心逻辑也结合自己在实际开发中踩过的坑把它的设计思路、关键实现、实操步骤和常见问题整理成这篇博文。如果你正在做多智能体系统、任务编排、自动化工作流相关的事情或者单纯对“让多个 AI 角色协同干活”这件事感兴趣这篇文章应该能给你一些可以直接参考的东西。先说清楚这个项目大概是什么。agency-agents 本质上是一个多智能体协作框架它提供了一套机制来定义代理的角色、能力边界、通信方式和任务分配策略。你可以把它理解为一个“代理团队的管理系统”每个代理有自己的专长和职责框架负责协调它们之间的消息传递、任务流转和结果汇总。它解决的问题是当单个代理无法高效处理复杂任务时如何通过多个代理的分工与协作来提升整体处理能力和结果质量。适合谁来参考我认为有三类人比较适合第一类是做自动化工作流编排的开发者想从单线程的脚本式调用升级到多代理协作第二类是在做 AI 应用层产品的工程师需要让不同模型或不同提示词策略的代理协同工作第三类是对多智能体系统感兴趣的学习者想找一个结构清晰、可上手修改的参考实现。不管你是哪一类下面的内容都会从设计思路讲到具体操作尽量让不同基础的人都能跟上。2. 整体架构与设计思路拆解2.1 为什么是“多代理协作”而不是“单代理堆功能”很多人一开始会想我为什么不把一个代理做得足够强大把所有能力都塞进去这个问题我在早期项目里也纠结过。实际做下来会发现单代理堆功能有几个绕不过去的坎。第一是提示词膨胀当你把任务规划、工具调用、结果校验、格式输出全部写在一个系统提示里提示词会变得又长又杂模型很容易顾此失彼。第二是错误累积一个环节出错后面全盘皆输而且很难定位是哪个能力模块出了问题。第三是复用性差你为一个场景调好的全能代理换一个场景几乎要重写。agency-agents 选择多代理协作路线核心逻辑就是“分而治之”。每个代理只负责一个相对窄的职责比如有的专门做任务拆解有的专门做信息检索有的专门做结果校验有的专门做最终输出。这样做的好处很明显每个代理的提示词可以写得很聚焦行为更可控单个代理出错时影响范围被限制在它负责的环节排查和替换都更容易不同场景之间可以复用同一批代理只需要调整编排逻辑。这个思路和现实中的团队协作是一样的。你不会让一个人同时做需求分析、写代码、做测试、写文档而是让不同角色各司其职通过沟通和交接来完成项目。agency-agents 就是把这种协作模式搬到了代理系统里。2.2 核心抽象代理、消息、编排器拆开来看agency-agents 的核心抽象其实不复杂主要是三个东西代理Agent、消息Message和编排器Orchestrator。代理是基本执行单元。每个代理有名字、角色描述、能力配置和对应的执行逻辑。角色描述决定了它在协作中扮演什么角色能力配置决定了它能调用哪些工具或模型执行逻辑决定了它收到消息后怎么处理、怎么回复。消息是代理之间通信的载体。消息通常包含发送方、接收方、内容类型和负载数据。内容类型可以是任务指令、中间结果、校验反馈、最终输出等。消息机制的设计直接决定了协作的灵活性和可追踪性。编排器是协调中心。它负责根据任务类型和当前状态决定下一步该由哪个代理处理把消息路由到正确的代理并维护整个任务的状态流转。编排器可以很简单比如一个固定的流水线也可以很复杂比如基于规则或模型决策的动态路由。我自己的经验是刚开始不要追求编排器太智能。先用固定流程跑通再逐步加入条件分支和动态决策。很多项目一上来就想做“全自动智能编排”结果调试成本极高反而拖慢进度。2.3 通信模式的选择同步、异步还是事件驱动代理之间怎么通信是一个需要提前想清楚的问题。agency-agents 支持几种常见的通信模式我分别说一下适用场景和注意点。同步请求-响应模式最简单代理 A 发消息给代理 B然后等待 B 返回结果再继续下一步。这种模式逻辑清晰容易调试适合流程固定、依赖关系明确的场景。缺点是如果某个代理处理慢整个流程会被阻塞。异步消息模式允许代理发完消息后继续做别的事结果通过回调或轮询获取。这种模式吞吐量更高适合代理处理时间差异大的场景。但异步带来的状态管理复杂度会上升你需要仔细处理消息顺序和超时。事件驱动模式是异步的进阶版代理不直接指定接收方而是发布事件由编排器或订阅机制决定谁来处理。这种模式解耦最彻底扩展性最好但调试难度也最高。我一般建议在代理数量超过五个、流程分支明显增多时再考虑事件驱动。注意通信模式的选择没有绝对优劣关键是和你的任务特征匹配。流程固定就用同步处理时间差异大就用异步代理数量多且关系复杂再考虑事件驱动。不要为了“架构先进”而过度设计。3. 核心细节解析与实操要点3.1 代理定义的关键字段与配置逻辑定义一个代理时有几个字段是必须想清楚的。第一个是角色名称和描述这不仅是给人类看的也会影响代理自身的提示词构建。描述要具体比如“负责从用户需求中提取关键约束条件”就比“负责分析需求”要好得多。第二个是能力配置。这包括这个代理可以调用哪些工具、访问哪些数据源、使用哪个模型。能力边界要收窄不要给一个代理开放所有工具。我见过一个项目所有代理都配置了全部工具权限结果某个代理在应该只做文本校验的时候误调了一个数据写入工具造成了脏数据。后来改成按需授权问题就消失了。第三个是输入输出格式。代理之间的消息最好有明确的结构比如 JSON 格式包含字段类型和必填项。这样编排器和其他代理都能可靠地解析。自由文本虽然灵活但在多代理流转中很容易因为格式不一致导致解析失败。第四个是超时和重试策略。每个代理的处理时间可能不同设置合理的超时时间可以避免整个流程卡死。重试策略要区分错误类型网络抖动可以重试逻辑错误重试多少次都没用应该直接上报。3.2 消息协议设计让代理之间“说同一种语言”消息协议是多代理系统的血脉。设计得不好后面全是坑。我的经验是消息至少包含这几个部分消息 ID、发送方、接收方、消息类型、时间戳、负载和元数据。消息 ID 用于追踪和去重。发送方和接收方用于路由。消息类型决定了接收方怎么处理比如 task_assign、task_result、validation_feedback、final_output 等。时间戳用于排序和超时判断。负载是实际数据。元数据可以放优先级、重试次数、关联任务 ID 等辅助信息。负载的结构建议用 schema 约束。比如任务分配消息的负载可以定义为包含 task_id、task_description、constraints、deadline 等字段。这样每个代理在处理消息时可以按固定字段读取减少解析错误。提示消息协议一旦确定尽量保持稳定。如果确实需要变更做好版本兼容比如在元数据里加 protocol_version让新旧代理可以共存一段时间。3.3 编排器的决策逻辑从固定流水线到动态路由编排器的决策逻辑决定了整个系统的灵活度。最简单的做法是固定流水线任务按预设顺序依次经过代理 A、B、C、D。这种模式适合流程稳定的场景比如“需求解析 - 信息检索 - 内容生成 - 质量校验 - 输出”。稍微复杂一点的是条件分支根据某个代理的输出结果决定下一步走哪条路径。比如质量校验不通过时回到内容生成环节重新处理而不是直接输出。这种模式需要编排器能够解析代理的输出并做出判断。更复杂的是动态路由编排器根据当前任务状态和历史信息动态选择下一个代理。这通常需要一定的决策逻辑可以是基于规则的也可以是基于模型的。我的建议是动态路由先从简单规则开始比如“如果任务包含数据查询需求则路由到检索代理否则路由到生成代理”不要一上来就用模型做路由决策因为模型决策的不确定性会给调试带来很大困难。3.4 状态管理与上下文传递多代理协作中状态管理是一个容易被低估的难点。每个代理处理完任务后产生的中间结果需要被后续代理使用。这些结果怎么存、怎么传直接影响系统的可靠性和可扩展性。常见做法有两种一种是把所有中间结果放在一个共享的上下文对象里每个代理都可以读写。这种方式简单直接但代理之间的耦合度较高一个代理修改了上下文可能影响其他代理。另一种是每个代理只输出自己的结果由编排器负责汇总和分发。这种方式解耦更好但编排器的负担更重。我倾向于混合方式共享上下文只放全局状态比如任务 ID、当前阶段、全局约束每个代理的中间结果单独存储通过消息传递给需要的下游代理。这样既保证了全局一致性又避免了代理之间的意外干扰。上下文传递时要注意数据量。如果中间结果很大比如大段文本或大量数据不要直接塞进消息里而是存到外部存储消息里只传引用 ID。这样可以避免消息体积膨胀导致性能下降。4. 实操过程与核心环节实现4.1 环境准备与依赖安装假设你已经决定动手搭一个 agency-agents 风格的协作系统第一步是准备环境。我以 Python 技术栈为例因为它在 AI 应用开发中生态最成熟。首先创建虚拟环境这是基本习惯避免依赖冲突python -m venv venv source venv/bin/activate # Linux/macOS # 或者 venv\Scripts\activate # Windows然后安装核心依赖。通常需要 HTTP 客户端、异步框架、数据校验库和日志库pip install httpx pydantic loguru如果你打算用异步模式还需要安装异步支持库。如果要用模型做决策还需要对应的模型调用 SDK。具体装哪些取决于你的代理需要调用什么能力。注意依赖版本要锁定。我踩过的坑是开发环境跑得好好的换一台机器因为某个库的小版本差异消息序列化行为变了导致解析失败。后来用 requirements.txt 锁定版本问题就没了。4.2 定义第一个代理从最小可用开始不要一上来就定义五个代理。先定义一个最小可用的代理跑通单代理的输入输出再逐步加。一个代理的基本结构包括初始化配置、消息处理方法、结果返回方法。初始化时传入角色描述、能力配置和依赖项。消息处理方法接收消息对象根据消息类型执行对应逻辑返回结果消息。我建议第一个代理就做最简单的事比如“接收一段文本返回其字数统计”。这个代理不需要调用任何外部工具逻辑完全确定用来验证消息流转和编排器是否工作正常。class WordCountAgent: def __init__(self, name: str): self.name name def handle(self, message: dict) - dict: text message.get(payload, {}).get(text, ) count len(text) return { sender: self.name, type: task_result, payload: {word_count: count} }这个例子虽然简单但它包含了代理的核心要素有名字、有处理方法、有结构化输出。跑通它之后你就有了一个可以扩展的基础。4.3 搭建编排器任务流转的核心逻辑编排器的工作是接收初始任务按顺序调用代理传递消息收集结果。最简单的编排器就是一个循环class Orchestrator: def __init__(self, agents: list): self.agents agents def run(self, initial_message: dict) - dict: message initial_message for agent in self.agents: message agent.handle(message) return message这个编排器按固定顺序执行所有代理每个代理的输出作为下一个代理的输入。它适合流水线式的任务。当你需要条件分支时编排器就要能读取代理输出并做判断。比如def run(self, initial_message: dict) - dict: message initial_message message self.agents[parser].handle(message) if message[payload].get(need_retrieval): message self.agents[retriever].handle(message) message self.agents[generator].handle(message) message self.agents[validator].handle(message) return message这种写法比固定循环灵活但编排逻辑和具体代理耦合较紧。如果代理数量继续增加建议把编排逻辑抽象成配置或规则表让编排器根据配置决定流转路径。4.4 消息队列与异步处理的实际配置当代理处理时间较长或者需要并行处理多个任务时同步编排器会成为瓶颈。这时候引入消息队列是合理的选择。我一般用轻量级的方案起步比如基于内存的队列先把异步流程跑通。核心思路是编排器把任务消息放入队列代理作为消费者从队列取消息处理处理完把结果消息放回另一个队列编排器再从结果队列取消息决定下一步。import asyncio from collections import deque class AsyncOrchestrator: def __init__(self): self.task_queue deque() self.result_queue deque() async def dispatch(self, message: dict): self.task_queue.append(message) async def collect(self): while not self.result_queue: await asyncio.sleep(0.1) return self.result_queue.popleft()实际生产环境可能会用更成熟的消息中间件但核心逻辑是一样的解耦生产者和消费者通过队列缓冲任务通过异步提升吞吐。提示异步模式下一定要设置超时和死信处理。我遇到过代理因为某个外部调用卡住消息一直不返回整个流程挂起的情况。后来加了超时机制超时后把消息转入死信队列并告警问题就可控了。4.5 结果校验与输出格式化最后一个环节通常是结果校验和输出格式化。校验代理负责检查前面代理的输出是否符合要求比如字段是否完整、格式是否正确、内容是否满足约束。如果校验不通过编排器应该决定是重试、回退还是直接报错。输出格式化代理负责把最终结果整理成目标格式比如 JSON、Markdown 或特定模板。这个环节看起来简单但实际项目中经常因为格式细节没对齐而导致下游系统解析失败。我的做法是输出格式化代理的配置里明确写清楚目标格式的 schema并且在校验环节就按这个 schema 检查。def format_output(message: dict) - dict: data message[payload] return { status: success, result: { summary: data.get(summary, ), details: data.get(details, []), confidence: data.get(confidence, 0.0) } }格式化完成后整个协作流程就算走完了。你可以把结果返回给调用方也可以存入数据库或推送到下游系统。5. 常见问题与排查技巧实录5.1 代理之间消息丢失或顺序错乱这是异步模式下最常见的问题。表现是某个代理没有收到预期消息或者收到的消息顺序和发送顺序不一致。排查思路先检查消息队列的配置确认是否有消息过期时间设置过短导致消息被丢弃。然后检查代理的消费逻辑确认是否有异常被吞掉导致消息没有正确确认。最后检查是否有多个消费者竞争同一个队列导致消息被错误消费。解决方法给消息加唯一 ID 和序列号接收方按序列号排序和去重。设置合理的消息过期时间不要过短。代理消费消息时做好异常捕获和日志记录确保消息处理失败时能被发现和重试。5.2 某个代理处理超时导致整个流程卡住这个问题在同步模式下尤其明显。一个代理卡住后面所有代理都等着。排查思路先定位是哪个代理超时然后看这个代理在处理什么类型的消息时超时。常见原因是外部调用没有设置超时、代理内部逻辑有死循环、或者消息负载过大导致处理缓慢。解决方法给每个代理设置独立的超时时间超时后编排器决定是重试、跳过还是报错。外部调用必须设置超时。消息负载过大时改用引用传递而不是直接传数据。5.3 代理输出格式不一致导致解析失败多代理系统中不同代理可能由不同人开发或者在不同时间开发输出格式容易不一致。排查思路收集失败消息的原始内容对比预期格式和实际格式的差异。常见差异包括字段名大小写不一致、必填字段缺失、数据类型不对、嵌套层级不同。解决方法定义统一的消息 schema所有代理的输出都按 schema 校验。校验不通过时要么让代理重新输出要么由编排器做格式转换。schema 变更时做好版本管理。5.4 编排器逻辑复杂到难以维护当代理数量增多、分支条件变多时编排器的代码会变得又长又难懂。排查思路统计编排器中的条件分支数量和嵌套层级。如果超过五层嵌套或者分支数量超过十个基本可以确定需要重构。解决方法把编排逻辑从代码中抽出来用配置或规则表描述。比如用 YAML 定义流程节点和跳转条件编排器读取配置执行。这样修改流程不需要改代码也更容易理解和审查。5.5 常见问题速查表问题现象可能原因排查方法解决措施消息丢失队列过期时间过短、消费异常被吞检查队列配置和消费日志调整过期时间、完善异常捕获顺序错乱多消费者竞争、异步无排序检查消费者数量和消息序列号加序列号排序、单消费者或分区消费代理超时外部调用无超时、死循环、负载过大定位超时代理和消息类型设置超时、优化逻辑、改用引用传递格式解析失败schema 不统一、字段缺失或类型错误对比实际输出和预期 schema统一 schema、加校验和转换编排器难维护分支过多、嵌套过深统计分支和嵌套层级抽离配置、规则表驱动整体吞吐低同步阻塞、串行处理分析各环节耗时引入异步、并行化独立环节5.6 几个我踩过的坑和对应技巧第一个坑是过度依赖模型做路由决策。早期我让一个模型代理根据任务内容决定下一步走哪个分支结果模型偶尔会给出不在预设范围内的路由导致流程异常。后来改成模型只做建议编排器根据建议和预设规则做最终决策稳定性大幅提升。第二个坑是忽略日志的可追踪性。多代理系统中一个任务经过多个代理如果没有统一的追踪 ID出问题后很难把各环节日志串起来。后来我在初始消息里加了 trace_id所有代理的日志都带上这个 ID排查效率提高了很多。第三个坑是代理能力配置过宽。前面提过给代理开放过多工具权限会导致误调用。后来我改成按最小必要原则授权并且对敏感操作加二次确认问题就少了。第四个坑是忽略冷启动和重试的幂等性。有些代理在重试时会重复执行副作用操作比如重复写入数据。后来我给每个任务加唯一 ID代理在执行副作用操作前先检查该任务是否已处理过保证了幂等性。6. 扩展方向与个人实践体会这个框架跑通之后可以扩展的方向其实不少。一个方向是引入更细粒度的代理能力描述让编排器可以根据能力匹配自动选择代理而不是硬编码代理名称。另一个方向是加入代理性能监控记录每个代理的处理时间、成功率和资源消耗为优化提供数据支撑。还有一个方向是支持代理的动态加载和卸载让系统可以在运行时调整代理组合。我自己在实际操作中的体会是多代理系统的价值不在于代理数量多而在于职责划分清晰、消息流转可靠、编排逻辑可维护。我见过一些项目堆了很多代理但职责重叠、消息格式混乱、编排逻辑写死在代码里结果维护成本比单代理还高。所以我的建议是先从两三个代理的最小协作开始把消息协议和编排逻辑打磨好再逐步扩展。每加一个代理都要问自己这个代理的职责是否足够独立它的输出是否对下游有明确价值如果答案不清晰就不要急着加。最后分享一个小技巧在开发阶段给编排器加一个“单步执行”模式每执行一个代理就暂停打印当前消息状态确认无误后再继续。这个模式在调试复杂流程时非常有用能帮你快速定位是哪个环节出了问题。等流程稳定后再切换到全自动模式运行。