做多智能体系统最头疼的，从来不是单个Agent的能力强不强，而是怎么让一群Agent像同一家公司的同事那样各司其职、顺畅协作。我最近把一个叫agency-agents的轻量级多Agent协作框架从玩具原型推进到了能扛住内部业务的可用版本，趁热把设计思路、核心实现和踩过的坑都记下来，希望对正在折腾Agent编排的朋友有点用。
这个项目解决的核心问题很具体：多个Agent如何组织在一起，完成一个跨步骤、多角色、带审核和返工循环的业务任务。它不是要替代任何大模型推理框架，而是做在Agent之上的一层“组织管理”：谁负责什么、消息怎么传、状态怎么存、任务挂了怎么恢复。如果你已经有几个能独立干活的Agent，或者手里有一堆bot、脚本、API服务想让它们协同工作，这篇东西值得看看。
1. agency-agents到底解决什么问题
1.1 从“一个Agent”到“一支队伍”
单个Agent跑一个简单问答、写一段摘要、调一个工具，这类事情现在已经非常成熟，框架也很多。可一旦任务复杂起来，问题就变了。比如一个“内容发布”流程：先有人写初稿，然后要合规审核，审核不过要打回修改，审核过了要排期发布，发布后还要收集数据反馈。用一个超级Agent去完成所有这些步骤，意味着你要在一个上下文窗口里塞进所有角色提示词、所有工具定义、所有中间状态——没跑几步就开始上下文污染，模型输出质量断崖式下跌，而且某个环节出错很难定位。
更麻烦的是权限问题。写初稿的Agent理论上不应该能直接点发布按钮，审核Agent也不该去调用数据统计接口。如果所有工具都挂在一个Agent身上，权限隔离就是一句空话。agency-agents的出发点就是把“一个人干所有事”改成“一个机构里不同岗位的人分工协作”，每个Agent只持有自己岗位需要的工具和权限，任务通过消息传递，而不是把所有人都塞进同一个提示词里。
实际做的时候，我还有个很深的体会：团队里不同Agent往往是用不同技术栈写的，有些是Python服务，有些是Node脚本，甚至还有老的Shell任务。硬性要求统一到某个Agent框架根本不现实。所以agency-agents从一开始就没打算规定“你必须用哪个模型SDK、哪个Agent基类”，而是做一个语言无关的协作层，让任何能收发消息的程序都可以作为Agent加入。
1.2 和常见编排框架的定位差异
市面上已经有不少多Agent框架，比如自动编排对话流的CrewAI、擅长图结构状态机的LangGraph、偏研究探索的AutoGen。它们各有优势，但我在实际项目里总感觉差一层东西：这些框架大多假设所有Agent运行在同一个进程、同一个语言生态里，而且流程结构往往在代码里写死，运行期想动态调整角色、加一个临时审批节点就很别扭。
agency-agents的定位是“机构化协作层”，它刻意做了三件不一样的事：
| 维度 | agency-agents | 典型单语言Agent框架 |
|---|---|---|
| Agent语言 | 不限，只要走协议 | 通常绑定某一种语言 |
| 协作拓扑 | 支持编排、市场、管道三种模式 | 偏重链式或图式编排 |
| 状态存储 | 外置Redis，可跨进程恢复 | 多在内存中维护 |
| 权限模型 | 显式声明工具白名单 | 多数依赖调用方自觉 |
| 消息追踪 | 内置trace_id全链路记录 | 需要额外接APM |
这么说不是要贬低别的框架——它们在自己的场景里都很好用。但如果你和我一样，面对的是一堆存量系统、异构服务，想快速把它们组合成“虚拟团队”，那一个薄薄的协议层比全家桶框架实用得多。后面所有内容，都围绕这个协议层展开。
2. 先把架构想清楚再写代码
动笔写业务代码之前，我在架构上卡了最久。多Agent协作系统最大的风险是“看似灵活，实则失控”。如果每个Agent都能自由地给任何其他Agent发消息，运行十分钟后你根本不知道是谁触发了谁，任务就成了一团乱麻。agency-agents的处理方式不是限制自由度，而是用模式来约束协作结构。
2.1 三种协作模式：编排、市场与管道
我最终把协作模式收敛为三种，所有复杂场景都能落到这三种模式的组合上。
第一种是编排者模式（Orchestrator）。有一个中心调度Agent负责任务拆解、派发、回收和异常兜底，其他Agent只干活不决策。这个模式最适合“流程明确、容错要求高”的场景，比如把用户需求分发给专业Agent，然后汇总结果。市面上大多数框架的默认模式就是这个。优点是好控制，缺点是中心节点容易成为瓶颈，而且调度Agent本身可能因为提示词太长产生决策漂移。
第二种是市场模式（Marketplace）。没有强中心，每个Agent把自己能做的事情发布成“服务”，需要协作时通过一个轻量的服务目录寻找合适的Agent，直接点对点通信。这种模式灵活、扩展性强，很适合工具数量庞大、调用关系动态变化的场景。代价是排查链路困难、可能出现多个Agent抢同一个任务，需要引入leader选举或者租约机制。
第三种是管道模式（Pipeline）。任务按照一个预先定义的拓扑，像一个流水线一样依次经过多个Agent。这种模式最简单，也最容易预测，适合内容处理类任务：抓取、清洗、分析、生成。它的问题在于拓扑写死之后，动态跳过某个环节或者插入新环节不太方便。
我在框架里把三种模式都实现了，但默认推荐编排者模式。理由很简单：可控性永远比炫技重要。市场模式和管道模式作为扩展能力存在，绝大多数业务流程，先用编排者模式跑通，再考虑优化成更扁平的结构。
2.2 消息协议与共享状态的设计取舍
多个Agent协作，首先要解决的就是“说什么语言”。我定义了一套极简的消息结构，核心字段只有六个：消息ID、链路ID、发送方、接收方、消息类型、载荷。所有消息统一走这个结构，用JSON传递，接收方只需要解析标准头部，业务内容放在payload里自由发挥。
这套设计参考了公司内部各部门之间传工单的方式：不管你是哪个部门，工单都得有编号、有发起人、有接收人、有优先级。至于工单正文怎么写，每个部门可以有自己的格式。放在Agent协作里，好处是——所有Agent都只需要实现一个消息处理入口，业务逻辑完全解耦。
共享状态是个大问题。我一开始图省事，让所有Agent把中间结果都放在一个全局字典里，跑了不到两个复杂任务就放弃了。多个Agent同时读写，会出现“你在改A字段的时候我把整个记录覆盖了”的经典问题。后来我采用“三区分离”的做法：
- 任务元数据区：存任务状态、当前阶段、负责人，只能由编排者Agent或者持有调度权限的Agent写。
- 数据结果区：存各Agent产出的中间结果，按agent_id分目录，只允许对应Agent写自己的目录。
- 共享黑板区：存跨Agent需要共同消费的信息，比如业务上下文、临时变量，写入必须带版本号。
这三类数据都放在Redis里，但Key前缀完全不同。这样既能保证状态外置可恢复，又能从存储层面减少冲突。
2.3 为什么坚持语言无关的通信层
我见过太多团队栽在这个选择上：为了用某个Agent框架，逼着所有服务都改成Python重写，最后项目死在迁移路上。agency-agents没有走那条路，通信层就是一个HTTP接口加一组消息格式约定，Agent可以是任何语言写的，只要能解析HTTP请求、能返回JSON。
这意味着你现有的一个Java写的文档解析服务，外面套一层薄薄的封装，注册一下就变成了“文档解析Agent”。一个Node写的发信bot，包一层也能参与协作。语言无关带来一个额外的好处：Agent的故障不会拖垮整个进程。如果是同一个进程内直接函数调用，一个Agent抛异常可能把主程序打挂；走HTTP通信，调用失败最多拿到一个超时，隔离性天然就好。
当然语言无关也有代价，主要是序列化和网络开销，以及调试时需要同时看多个服务的日志。我是用“链路ID贯穿所有日志”的方式解决这个问题的，后面排查部分会细说。
3. 核心实现：从零把agency-agents跑起来
架构想清楚了，实现其实不复杂。整体技术栈选型上我比较保守：Python写通信层和编排器，FastAPI做HTTP接口，Redis做状态存储和消息队列，没有引入额外重组件。之所以这样选，是因为这些组件都足够普及，出了问题网上一搜一大把解决方案，团队成员接手也快。
3.1 环境依赖与最小工程骨架
项目仓库里我建议按下面这个结构组织代码：
agency-agents/ ├── agent_agency/ │ ├── __init__.py │ ├── protocol.py # 消息协议定义 │ ├── orchestrator.py # 编排者模式核心 │ ├── agent_runtime.py # Agent运行时封装 │ ├── registry.py # Agent注册中心 │ ├── store.py # Redis状态与黑板 │ └── tracing.py # 链路追踪 ├── agents/ # 具体业务Agent │ ├── writer_agent.py │ ├── reviewer_agent.py │ └── publisher_agent.py ├── tests/ ├── config.yaml └── requirements.txt依赖方面我只用了四个核心库：fastapi提供HTTP服务，redis做客户端，pydantic做消息校验，uvicorn做服务启动。如果你需要定时调度，再加一个apscheduler就足够了，不要急着上消息队列中间件。
安装依赖后，第一件事是把Redis跑起来，因为所有Agent注册信息、任务状态、消息的临时交换都依赖它。Redis挂了，整个协作系统就会停摆，所以生产环境至少要做主从或者集群，这个后面说高可用的时候再展开。
3.2 注册一个带工具权限的Agent
要让一个程序成为协作团队的一员，核心是注册信息。我把注册信息设计成Agent Manifest，一份YAML文件，包含身份、能力、权限三个部分。看一个具体例子：
agent_id: writer_agent display_name: 内容创作Agent description: 负责根据选题生成文章初稿 capabilities: - text_generation - outline_planning tools: - name: web_search action: tool_call description: 搜索公开资料 enabled: true - name: create_draft action: tool_call description: 创建内容初稿 enabled: true permissions: can_send_to: - reviewer_agent - orchestrator_agent can_read_blackboard: true can_write_blackboard: false注册这个Manifest后，注册中心会做几件事：检查Agent ID是否重复，解析能力列表，生成一个通信凭证，最后把Agent的状态标记为idle。重点在于permissions字段，它规定了这台Agent能跟谁说话。这个设计是为了控制“调度风暴”——如果每个Agent都能给任意Agent发消息，系统很快就会进入不可预测的混沌状态。实际经验是，权限定义越严格，越容易定位和排查问题。
Agent注册完成后，它会启动一个HTTP服务并主动向注册中心发心跳。心跳间隔我默认设置为15秒，超过45秒没有心跳，注册中心就标记该Agent离线，编排者在分配任务时会自动越过离线节点。
3.3 定义一个跨Agent执行的任务流
注册好Agent，下一步就是定义任务流程。agency-agents支持把任务定义成流程图结构，但在代码层面我看到太多项目被“流程图框架”绑架了——为了画一张漂亮的可视化编排图，引入了一堆概念和依赖，结果业务逻辑依然是一团乱。
所以我选择了轻量的方式：用Python声明式地描述每一步，并把步骤之间的转移关系写成数据。直接看代码更直观：
# 定义一个“内容发布”任务流 from agent_agency.protocol import WorkflowSpec, StepSpec workflow = WorkflowSpec( workflow_id="content_publish_flow", steps=[ StepSpec( step_id="drafting", agent="writer_agent", action="create_article", input_ref="task.inputs.topic", # 从任务输入取参数 output_ref="task.results.draft", # 产出放到共享状态指定位置 ), StepSpec( step_id="reviewing", agent="reviewer_agent", action="review_article", input_ref="task.results.draft", output_ref="task.results.review", ), StepSpec( step_id="publishing", agent="publisher_agent", action="publish_article", input_ref="task.results.draft", output_ref="task.results.publish_url", ), ], transitions=[ # drafting完成 -> 进入reviewing ("drafting", "success", "reviewing"), # reviewing结果为不合格 -> 回到drafting ("reviewing", "rejected", "drafting"), # reviewing结果为合格 -> 进入publishing ("reviewing", "approved", "publishing"), ], )这个定义放在一个集中配置里，修改流程不需要重新部署Agent，只需编排者重新加载配置。input_ref和output_ref的设计非常关键，它让每个Agent之间不需要知道对方的数据结构细节，只通过任务状态区域里约定的Key传递数据。
执行引擎拿到这个WorkflowSpec后，会把它编译成状态机：每个步骤有pending、running、success、failed、rejected几种状态，转换条件就是transitions列表。编排者会根据当前状态决定下一步派发给谁。
3.4 状态追踪与断点恢复机制
多Agent系统一旦跑起来，最大的噩梦就是进程崩溃或网络波动导致任务卡在半路。内存里的状态变量在重启后全部清零，这就是为什么我强烈建议把任务状态放到Redis里持久化。
我实现了一个简单的任务状态机，每个任务在Redis里存一份完整的状态快照：
agency:task:{task_id} { "workflow_id": "content_publish_flow", "current_step": "reviewing", "step_states": { "drafting": "success", "reviewing": "running" }, "messages_sent": 87, "version": 12 }执行过程中，每个步骤开始和结束都会把最新状态写回Redis，并递增版本号。崩溃恢复的流程是：编排者重启后扫描所有running状态的步骤，通过心跳检查对应Agent是否还活着；如果Agent活着，发送一个status_probe消息询问当前进度；如果在超时窗口内没有回应，就把状态重置为failed并将任务回退到上一个成功步骤，允许重新调度。
这里踩过一个大坑：如果恢复逻辑写得过于激进，网络闪断时会误判Agent故障，导致同一个任务被两个Agent同时执行。解决方案是给每个步骤加一个执行租约，只有持有租约的Agent才被允许执行该步骤。租约默认30秒，快到期时必须续租，否则由编排者重新分配。
4. 实战示例：用一个“内容发布小组”验证
只讲框架不跑真实案例是耍流氓。我用一个虚拟的“内容发布小组”来完整演示agency-agents的使用方式，这个小组有三个角色：选题加工Agent、内容审核Agent、发布执行Agent。整体流程贴近很多内容团队的真实工作方式，也方便你举一反三。
4.1 三个Agent的分工设计
内容发布小组的Agent分工逻辑，我参考的是现实公司的岗位职责边界：
- writer_agent（创作Agent）：负责根据给定的主题，查询公开资料，撰写结构完整的文章初稿。它的工具权限最宽，能搜索、能生成文本，但不能直接提交发布。
- reviewer_agent（审核Agent）：负责审核文章内容质量，输出结论
approved或rejected，并给出修改建议。它的工具权限很窄，只能读稿子、查询审核规范，不能修改原稿。 - publisher_agent（发布Agent）：负责把审核通过的文章发布到目标平台，并返回发布链接。它的工具权限只有一个——发布接口，每次调用都要记录理由，方便审计。
这个分工的核心在于：没有任何一个Agent拥有完成全链路的完整权限。尽管writer_agent也可以去调发布接口把稿子发出去，但从权限设计上就杜绝了这种行为。这跟真实公司设置不同岗位权限是同一个道理。
每个Agent启动时，我会在系统提示词里明确它的角色边界：
你是内容创作Agent，负责根据主题撰写初稿。 你的产出必须提交给审核Agent，不得自行发布内容。 如果收到审核返工意见，根据意见修改原稿后重新提交。不要小看这几句话，模型再强也需要明确的角色约束。agent协作中80%的失控问题，都源于角色边界没有在提示词层面对齐。
4.2 任务执行链路与消息时序
假设现在有一个新任务：写一篇关于“个人知识管理工具选型”的科普文章。任务会以一条task.submit消息进入系统，随后进入工作流。完整的消息链路如下：
- 编排者收到任务，创建任务ID，读取发布流程配置。
- 编排者向writer_agent发送
task.execute消息，载荷为选题主题和输出位置。 - writer_agent处理完，把初稿写入Redis中的
task.results.draft，然后向编排者返回task.completed。 - 编排者根据状态机，向reviewer_agent发送
task.execute消息，指向task.results.draft。 - reviewer_agent审核后返回
task.reviewed，如果结论是approved，编排者进入下一步；如果是rejected，编排者带着修改意见回到步骤2。 - 审核通过后，编排者向publisher_agent发送
task.execute消息，publisher完成发布，返回publish_url。 - 编排者更新任务状态为
success，把完整链路标记为trace_id结尾，归档入库。
每一步消息都会带上同一个trace_id。这个trace_id是排障的第一利器——无论是查日志、查状态、还是查消息流，只需要按trace_id过滤，就能还原全过程。我还在Redis里维护了一个消息时序表，记录每条消息的时间戳和发送接收方，后期复盘非常有用。
4.3 参数调优：并发度、超时与重试
跑通流程只是第一步，想要稳定运行必须调参。很多人写多Agent应用时直接忽略参数配置，结果一上生产，任务一多就乱套。我总结了一套比较稳的参数起点，你可以按自己的场景调整：
| 参数 | 推荐值 | 说明 |
|---|---|---|
max_concurrent_tasks | CPU核数×2 | 并发任务太多会导致模型服务和工具调用互相争抢资源 |
agent_timeout | 60秒 | 单个Agent处理单个步骤的超时，超过即标记失败 |
max_retries | 3次 | 每个步骤最多重试次数，超过则任务进入failed |
retry_backoff | 1秒/2秒/4秒 | 重试间隔递增，避免风暴式重试 |
heartbeat_interval | 15秒 | Agent心跳频率，太低会浪费资源，太高则故障发现慢 |
lease_ttl | 30秒 | 执行租约有效期，网络波动时要给足余量 |
关于agent_timeout，我吃过亏。调小了模型偶尔响应慢就误判失败，调大了遇到真故障又要等很久。后来我采用“双阈值”：首次调用用60秒超时，重试时用90秒超时，因为重试场景下模型需要更多时间处理上下文，而失败的Agent很可能已经缓存了部分计算结果，第二次往往更快。
还有一个容易被忽略的参数：任务队列长度。我默认限制为100个等待中的任务，超过之后新的task.submit直接返回“系统繁忙”。这个拒绝机制虽然看起来不友好，但能防止任务积压导致所有Agent都在处理旧任务，新任务反而饿死。生产环境里，“有损限流”比“无限排队”对用户体验更好。
5. 实战中的高频问题与排查手册
跑了一个多月，零散遇到十几个问题，最典型的集中在下面四类。每个问题我都记录了现象、原因和解决办法，整理成了速查表。
5.1 Agent互相等待导致死锁
现象：任务状态长时间停在running，没有任何Agent在干活，日志里看不到新的消息输出。
这个问题的本质是循环依赖。比如下面这种场景：Agent A完成步骤后，按照流程需要Agent B确认，但Agent B又在等待Agent A先完成某个数据准备——两边都在等对方，谁也动不了。在编排者模式里不容易出现，但在市场模式里非常常见。
排查步骤：先查看trace_id对应的消息时序，看看最后一条消息是发给谁的；如果最后一条消息发给了A，但A当前状态不是running，说明A在等一个永远不会到来的消息。解决方法是两招：一是给所有task.execute消息设置等待超时，超时未回即产生告警；二是在编排者心里维护一个依赖等待图，检测到循环依赖就主动干预，抛出异常并人工处理。
我后来干脆在框架里加了一个规则：任何Agent在一个步骤上的等待时间不能超过5分钟，超时默认回退到编排者重新决策。这就把“无限等待”变成了“有限等待加点重试”。
5.2 工具调用风暴与循环放大
现象：任务数量没有增长，但日志里Agent调用工具的次数指数级上涨，系统负载飙升。
原因通常是某个Agent在提示词里被允许“反复改进”，它在一次任务里不断调用搜索、重复生成，还每次都把结果写回共享黑板，其他Agent读到后又触发自己的工具调用，来回放大。有一次我的系统一晚上跑了10万次搜索调用，就是因为一个Agent在循环优化一个标题，停不下来。
解决办法是三层：第一，给每个Agent设置单任务的工具调用配额，默认10次；第二，在共享黑板写入前计算消息频率，如果同一Agent在1分钟内写入超过20次，直接熔断；第三，在消息协议里加一个max_trace_depth字段，任何链路超过15层直接截断并报错。这三层下来，循环放大基本被堵死了。
5.3 状态不一致：黄金令牌问题
现象：两个Agent同时处理同一个任务的不同步骤，它们读到的状态不一致，最后写回的时候相互覆盖，任务最终状态和实际执行情况完全对不上。
这是分布式系统里最经典的并发写问题，我称之为黄金令牌问题：谁手里的令牌是真正有效的？在单进程时代，一个全局锁就能解决；在多Agent、多进程场景下，锁还得放在共享存储里。
解决方案是用Redis实现乐观锁CAS（Compare And Set）。每次写状态时带上当前的version，写操作通过Lua脚本原子性比较版本号，如果版本号不匹配就拒绝写入并返回冲突。业务代码收到冲突后，重新读取最新状态，决定是丢弃还是合并。这个机制代码量不大，但能避免99%的状态覆盖问题。
5.4 消息时序错乱与丢失
现象：任务步骤A已经完成了，但它的后续步骤B比前置步骤C更晚收到消息，导致处理乱套；或者偶发消息丢失，任务卡住。
消息丢失一般发生在Agent进程重启的瞬间——消息已经发出去，但接收方还没注册完成，消息就被丢弃了。我的处理方式是在所有Agent之间使用一个“确认+重发”机制：每条消息都有唯一ID，接收方处理后返回msg.ack；发送方如果在一定时间内没收到ack，就自动重发。重发会导致重复消费，所以接收方在幂等表里记录已处理的消息ID，重复消息直接丢弃。
时序错乱则需要靠两个字段：一个是sequence_no，按发送顺序递增；另一个是depends_on，标明本消息依赖的上一条消息ID。接收方如果发现depends_on还没收到，会先缓存当前消息，等待前置消息到达后再处理。这其实就是消息队列里的顺序消费逻辑，但用很轻量的方式实现了。
6. 一些经验与后续扩展方向
最后分享几条这一路踩坑攒下来的体会。第一条：不要一上来就设计一个万能架构。我最初试图把编排者模式、市场模式、管道模式都做成高度可插拔的模块，结果代码写得非常抽象，真正跑业务时根本调试不动。后来砍掉一半抽象层，用最朴素的函数调用实现，反而稳定得多。分布式协作系统的价值在于清晰，不在于花哨。
第二条：链路追踪是第一位的基础设施。多Agent系统调试和单体应用完全不同，你不能靠watch一个进程去排查。从第一天起就让所有消息带上trace_id，把每次消息发送和返回的完整链路记录到一个日志文件里，后期能省下无数时间。看到任务卡住，先按trace_id把链路拉出来，比盲猜强一百倍。
第三条：测试里一定要引入“坏Agent”。普通的单元测试只能证明所有Agent正常工作时流程能走通，但真实世界里Agent会因为模型返回格式错误、API超时、工具调用失败等各种原因不按常理出牌。我专门写了一个chaos_agent，随机在30%的步骤上返回格式错误或直接超时，用来验证编排者的异常处理逻辑是否足够稳健。这套混沌测试帮我发现了至少五六个隐藏的恢复bug。
关于扩展方向，我自己接下来的计划有两块：一是把权限体系接到企业的SSO认证上，让Agent的注册、撤销、审计能和管理员的账号体系打通；二是引入基于向量检索的记忆模块，让Agent能跨任务使用历史经验，避免每个新任务都从零开始思考。这些方向我自己也还在摸索，等跑出结果再回来更新。
如果你也在做一个类似的多Agent协作系统，最想提的建议是：先把一个最简单的流程用最笨的方式跑通，再加复杂特性。Agent协作的坑有千千万万，但绝大多数可以在一个简单的原型里提前暴露出来——前提是你真的动手把原型写出来，而不是只在文档里画架构图。
