★ FEATURED ARTICLE 做多智能体系统最头疼的从来不是单个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_idcontent_publish_flow, steps[ StepSpec( step_iddrafting, agentwriter_agent, actioncreate_article, input_reftask.inputs.topic, # 从任务输入取参数 output_reftask.results.draft, # 产出放到共享状态指定位置 ), StepSpec( step_idreviewing, agentreviewer_agent, actionreview_article, input_reftask.results.draft, output_reftask.results.review, ), StepSpec( step_idpublishing, agentpublisher_agent, actionpublish_article, input_reftask.results.draft, output_reftask.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_tasksCPU核数×2并发任务太多会导致模型服务和工具调用互相争抢资源agent_timeout60秒单个Agent处理单个步骤的超时超过即标记失败max_retries3次每个步骤最多重试次数超过则任务进入failedretry_backoff1秒/2秒/4秒重试间隔递增避免风暴式重试heartbeat_interval15秒Agent心跳频率太低会浪费资源太高则故障发现慢lease_ttl30秒执行租约有效期网络波动时要给足余量关于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实现乐观锁CASCompare 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协作的坑有千千万万但绝大多数可以在一个简单的原型里提前暴露出来——前提是你真的动手把原型写出来而不是只在文档里画架构图。
阅读完成 · 觉得有帮助？
