多智能体协作框架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 常见问题速查表

5.6 几个我踩过的坑和对应技巧

第一个坑是过度依赖模型做路由决策。早期我让一个模型代理根据任务内容决定下一步走哪个分支，结果模型偶尔会给出不在预设范围内的路由，导致流程异常。后来改成模型只做建议，编排器根据建议和预设规则做最终决策，稳定性大幅提升。

第二个坑是忽略日志的可追踪性。多代理系统中，一个任务经过多个代理，如果没有统一的追踪 ID，出问题后很难把各环节日志串起来。后来我在初始消息里加了 trace_id，所有代理的日志都带上这个 ID，排查效率提高了很多。

第三个坑是代理能力配置过宽。前面提过，给代理开放过多工具权限会导致误调用。后来我改成按最小必要原则授权，并且对敏感操作加二次确认，问题就少了。

第四个坑是忽略冷启动和重试的幂等性。有些代理在重试时会重复执行副作用操作，比如重复写入数据。后来我给每个任务加唯一 ID，代理在执行副作用操作前先检查该任务是否已处理过，保证了幂等性。

6. 扩展方向与个人实践体会

这个框架跑通之后，可以扩展的方向其实不少。一个方向是引入更细粒度的代理能力描述，让编排器可以根据能力匹配自动选择代理，而不是硬编码代理名称。另一个方向是加入代理性能监控，记录每个代理的处理时间、成功率和资源消耗，为优化提供数据支撑。还有一个方向是支持代理的动态加载和卸载，让系统可以在运行时调整代理组合。

我自己在实际操作中的体会是，多代理系统的价值不在于代理数量多，而在于职责划分清晰、消息流转可靠、编排逻辑可维护。我见过一些项目堆了很多代理，但职责重叠、消息格式混乱、编排逻辑写死在代码里，结果维护成本比单代理还高。所以我的建议是，先从两三个代理的最小协作开始，把消息协议和编排逻辑打磨好，再逐步扩展。每加一个代理，都要问自己：这个代理的职责是否足够独立？它的输出是否对下游有明确价值？如果答案不清晰，就不要急着加。

最后分享一个小技巧：在开发阶段，给编排器加一个“单步执行”模式，每执行一个代理就暂停，打印当前消息状态，确认无误后再继续。这个模式在调试复杂流程时非常有用，能帮你快速定位是哪个环节出了问题。等流程稳定后，再切换到全自动模式运行。