搞 AI Agent 的人大概率都经历过这种阶段本地跑 demo 时一切顺利一旦想把多个 Agent 组合起来干一件完整的事立刻就崩了——A 的结果不知道往哪送B 想要的输入格式跟 A 的输出对不上C 处理到一半卡死了你还不知道它卡在哪一步。我这次做的“Agent-Reach”本质就是想解决这一连串“Agent 之间的协作触达”问题让不同职责的 Agent 能互相发现、按能力接活、稳定传消息并且整个过程能追溯、能回放、能排查。这篇文章不聊大而全的架构理论就把我踩过的坑、拆解过的设计、最后能跑通生产环境的方案一次性说清楚适合团队里正在做多 Agent 系统、或者正准备从“单 Agent 硬干”切换到“多 Agent 协作”的开发者。1. 为什么我会自己动手写一个 Agent 协作框架先交代背景。我接手的项目不是那种“一个 Agent 回答所有问题”的玩具而是真实业务里需要 Agent 分角色完成任务的场景有人负责抓数据有人负责分析有人负责生成报告还有人做质检。整个流程串起来之后开始暴露出一堆单 Agent 时代根本不会遇到的问题。1.1 单 Agent 的瓶颈与多 Agent 的隐形成本单个 Agent 的职责边界一旦扩大系统就变得极其难以维护。一方面把所有业务逻辑塞进一个 Agent 的 prompt 和工具集里token 消耗高得离谱而且交互一长模型经常在中间步骤“迷路”答非所问另一方面业务方需要的是能并行处理多个环节的流水线单 Agent 天然做不到“同时抓取十路数据再同时分析”。我尝试按角色拆出多个 Agent 后新的麻烦接踵而至任务该派给谁如果一个 Agent 挂了一次任务怎么重试消息是谁发的、发给谁、经没经过中间节点在引入 Agent-Reach 之前我的代码里全是手写的回调函数第三层嵌套后基本没眼看。1.2 现成框架给我的启发和新的顾虑市面上不是没有现成的多 Agent 编排工具LangGraph、CrewAI、AutoGen 这些我都跑过一遍用起来确实爽开箱即得。但真正放到生产环境里我发现几件难受的事绑定太紧业务代码和框架的图结构、Agent 定义方式深度耦合以后想换或者想落回原生代码会非常痛苦编排过程不够透明任务在框架内部传递时多少人真正见过“某一条消息被重试了三次”这种带现场细节的日志大多数框架的日志和监控对“链路级调试”支持得很弱。于是我用了一个更务实的思路抽象出一个“极薄的消息触达层”也就是 Agent-Reach 的原型。它不是一个重引擎不替你决定 Agent 应该怎么产出内容、怎么调用模型它只解决一件事——让消息在 Agent 之间可靠地流动。1.3 Agent-Reach 的定位注册中心 消息总线 可观测Agent-Reach 的核心概念拆开看就三条注册中心负责让每个 Agent 声明“我能干什么”消息总线负责把任务按声明和规则路由给合适的接收方可观测模块把每次触达、每条消息的状态变化、每一次重试和失败记录成链路日志。整个设计刻意不碰 Agent 的内部推理逻辑所以接入成本极低。“Agent 能干什么”用一套轻量的 schema 声明比如一个“行业信息搜集”Agent 可以声明自己支持collect.industry_news这个 action入参是行业列表和时间范围出参是结构化的文章列表。调度方只需要发起一个动作并附上参数不用关心具体哪个 Agent 来做路由层会根据声明和当前负载去匹配最合适的接收方。这套思路说直白点就是把 Agent 当成一组有“技能标签”的微服务来治理只不过这次“服务”的调用方和接收方都有一层大模型外壳。2. 核心设计思路把 Agent 当成资源而不是代码模块这一节是整个项目真正花我时间最多的地方设计上有几个点我认为是 Agent-Reach 能跑稳的关键。2.1 Agent 能力声明让路由变得可预测很多人在多 Agent 协作里喜欢用“自然语言路由”就是给调度模型一段很长的描述让它自己去判断接下来该调哪个 Agent。我第一次也这么干效果极度不稳定模型偶尔会把一个该发给“数据分析”的任务发给“文案润色”。后来我彻底放弃了模型做路由改为基于结构化的能力声明做精确匹配。每个 Agent 在上线前必须向注册中心提交一份能力声明字段大致是这个样子{ agent_id: industry-collector, host: agent-1.internal, port: 8080, actions: [ { action: collect.industry_news, params_schema: { type: object, properties: { industries: { type: array, items: { type: string } }, time_range_days: { type: integer, minimum: 1, maximum: 30 } }, required: [industries] }, output_schema: { type: array, items: { type: object, properties: { title: { type: string }, url: { type: string }, published_at: { type: string } } } }, available: true } ], load: 0.3, health_status: healthy }这里我把host和port直接写在能力声明里是因为 Agent 大多跑在不同的容器或机器上注册中心需要知道实际触达地址。load字段是动态心跳上报的负载指数路由时会根据负载做软偏好避免所有任务都堆到一个 Agent 上。2.2 消息信封一切的起点与终点消息在 Agent 之间流动最忌讳的是传“裸数据”。我最初调试时直接投递 dict结果下游 Agent 拿到之后完全不知道这个数据是从哪一步来的、该做出什么响应。在设计 Agent-Reach 时我强制所有消息都套一层统一信封字段设计如下{ envelope_id: env_e1f2a3b4c5d6, trace_id: tr_8c9f1a2b3c4d5e6f7a8b, source_agent: orchestrator, target_action: collect.industry_news, priority: 3, created_at: 1714003200.123, expire_at: 1714006800.123, attempt: 1, payload: { industries: [AI infra, Edge AI], time_range_days: 14 } }envelope_id是消息的唯一标识用于幂等和去重trace_id是整条协作链路的追踪 ID所有 Agent 在处理这条消息时都必须把 trace_id 打在自己的日志里否则事后排查时根本连不上attempt表示当前第几次投递用于重试策略判断expire_at用来拦截那些已经超时的过期消息防止僵尸任务无限消耗资源。2.3 路由与触达机制三层匹配优先精确而非发散Agent-Reach 的路由不是靠一个 Agent 去“读懂”任务而是走一个三层匹配流程。第一层是精确匹配任务里的target_action必须与某个 Agent 能力声明里的 action 完全一致能匹配就直接锁定向这个 Agent或负载最低的同一个 action 的 Agent第二层是正则匹配适合一批命名相近的动作例如collect.*可以匹配到信息采集类的任意子动作第三层才使用语义匹配当精确和正则匹配都命中不了时才会调用一个本地小模型把“意图文本”映射到已有的 action 名称并且这种语义匹配的命中结果必须经过一个“相似度阈值”过滤低于 0.85 的直接拒绝。我特意把语义路由放在最后还加了阈值是因为我发现语义匹配的“创造性”在派发任务时是灾难。它会给不太相关的任务硬找一个 Agent 接住宁可拒单也不要错单拒单后至少能被系统看见而错单会让 Agent 静默地产生垃圾输出。2.4 可靠性设计ACK、重试、超时与死信在 Agent-Reach 里消息投递出去不等于任务执行成功。我用的是“手动 ACK 至少一次投递”的模型接收方 Agent 必须在处理完任务后显式调用 SDK 的 ack 接口否则系统判定这一次投递失败。这意味着接收方可能多次收到同一条消息因此幂等处理是必须的通常做法是接收方先检查 envelope_id 是否处理过处理过就直接返回 ACK。重试采用指数退避策略默认初始等待 1 秒每次失败后等待时间乘以 1.8最多尝试 5 次。重试次数达到上限仍没成功消息进死信队列并触发一个回调钩子让编排方感知到这个任务彻底失败。死信队列里保存的不仅是 payload还有全部投递尝试的现场日志这对事后分析 Agent 是“崩了”还是“压根没收到”非常有价值。3. 从零落地一套可复用的 Agent-Reach 配置与接入流程理论说了一堆现在来到最实操的部分。我第一次把 Agent-Reach 接入真实项目时前端界面完全不显示数据后台 task 记录停留在“已派发”状态。经过排查发现是 ACK 没有在任务真正执行后调用而是放在“消息收到”的地方调用。听起来很小儿科但很多新手会踩。这一节我给出一份可以直接照着落地的配置流程。3.1 部署层面的三层架构Agent-Reach 虽然代码量不大但我在部署上分了三个台层控制面部署注册中心与路由决策服务还有管理后台的 API。这个层面的请求量不高但是必须是高可用因为所有 Agent 的上线、心跳、路由查询都依赖它。数据面消息队列与死信存储队列我这里用的是 Redis Streams 的轻量实现。为什么不用 Kafka因为 Agent 之间的消息不是超高吞吐的事件流重点是“各种不同的消息进到对应队列后被独立消费”Redis Streams 足够运维成本低很多。Agent 宿主层每个 Agent 进程独立部署注册后与注册中心维持心跳。底层这三个层面彼此不混跑尤其是 Redis 不要跟控制面混在一个容器里否则一个热 key 就可能把路由决策接口一起拖垮这条是我踩过生产事故得到的教训。3.2 接入 SDK十分钟跑通一个 AgentAgent-Reach 的接入 SDK 最简化下来只需要四个接口足够日常使用reach.ping(timeout200)注册中心探活超时阈值短一点不适合让它慢吞吞等你半天。reach.register(capability)启动时把自己能力声明注册进去返回注册 ID后续心跳带上这个 ID。reach.task(action, payload, options)投递一个新任务这一步等于往总线里放了一条消息信封。reach.ack(envelope_id, status, output)任务完成后显式确认。我把刚才“行业信息搜集”这个 Agent 的接入代码简化之后大概是这个样子from agent_reach import Reach, ActionHandler reach Reach( registry_urlhttp://registry.internal:8600, standby_queues[default-queue] ) def handle_collect(payload): industries payload[industries] days payload[time_range_days] results fetch_news(industries, days) # 业务逻辑 return results handler ActionHandler(collect.industry_news, handle_collect) reach.start(hostagent-1.internal, port8080, handlerhandler)启动之后Agent 会执行注册、上报心跳并且监听队列里投递过来的消息。R业逻辑写在handle_collect里它的返回值会被封装进 ACK 结果中回传给调用方。整个流程没有框架绑架你的 prompt 或模型调用方式业务代码还是你自己的。3.3 一个实际任务竞品舆情周报的协作全链路为了说清楚这套系统真正干了什么我用一个“竞品舆情周报”任务做演示总共四个 Agent 参与采集 Agent、清洗 Agent、分析 Agent、撰写 Agent。编排方把任务信封投给collect.competitor_news路由发现有两个采集 Agent 都声明了这个 action于是挑负载低的load0.2那个下发任务采集 Agent 返回了 200 条原始新闻数据ACK 时附带了输出输出没有直接回给编排方而是写进共享结果 store只回传一个结果 key编排方继续投递transform.deduplicate_and_clean给清洗 Agent清洗 Agent 处理完后投给analyze.sentiment_trend最后拿到分析结论后投给draft.markdown_report每个 Agent 都打印一条日志处理信封 ID、trace ID、耗时、结果 key。运营后台能看到全链路的状态流转从“已派发”到“已 ACK”再到“已生成报告”。3.4 参数调优重试、并发与限流的经验值以下数值是我在生产环境实测一批 Agent 协作任务后归纳的经验值可以直接抄配置项经验值理由心跳间隔5 秒Agent 状态更新够及时也不会压垮注册中心健康检查超时2 秒超过这个时间大概率是 Agent 假活或网络卡了任务默认超时60 秒大模型调用耗时一般不会超过这个值看业务情况调最大重试次数5 次再多大概率不是偶发问题需要人工介入重试退避系数1.8比固定间隔更能避开瞬时故障峰值并发上限单个 Agent 10 个同时处理任务多了模型推理排队反而拖慢整体路由负负载差低于 0.2 才触发转移防止小波动导致反复切换 Agent超时配置得重点讲一下。如果一条任务在“派发成功”后一直处于处理中超过expire_at时间就会被系统主动标记为“疑似卡死”。为什么不直接判失败因为大模型的推理可能真的需要更长时间直接挑断反而浪费前面的计算。我这里的处理是把它标记为超时挂起启动一次 OVerhead-Agent 探活如果 Agent 还在处理中没有返回最终输出但有一堆中间日志就延期expire_at保守地多给 30 秒。4. 生产环境里的坑与排查技巧实录落地过程中我至少踩过几十个具体的坑其中五个是最典型的几乎每次踩都能让我多活十年血压。这里把排查思路和解决方案都整理出来。4.1 路由匹配误命中精确匹配为何还会错现象是明明只希望某个 Agent 处理“电商类目”的采集任务结果另一个只声明了“热点新闻采集”的 Agent 也接到了同样任务。查半天发现问题出在params_schema没有做严格校验路由只看 action 名匹配根本没检查参数结构。修复方式非常简单路由命中后、投递前先拿params_schema对 payload 做一次 JSON Schema 校验不通过的直接拒绝。没加这个校验之前很多 Agent 拿到自己完全无法解析的参数白白产生垃圾输出最后还要求业务方去“猜为什么”。4.2 重复投递与幂等设计信封 ID 是保命绳重试机制必然导致同一条任务可能被接收方多次处理。如果接收方每收到一次就执行一次“插入数据库”的操作那洗数据那步会立刻爆炸。我在 SDK 提供try_duplicate_check(envelope_id)函数接收方在处理任务前先查一下这次任务是不是处理过了处理过就直接返回上一次的 ACK 结果。这个查询用 Redis 里的 envelope_id 做唯一键响应很快能挡住绝大多数的重复。4.3 Agent 假活心跳健康但任务不执行有段时间注册中心里所有 Agent 的状态都是 healthy但任务队列里积压却不断上涨。排查后发现Agent 主进程的心跳线程还活着但业务执行线程池已经全部被一个大任务卡死线程池耗尽后续任务根本排不进去。修改方案是在 SDK 的探活接口里加入“队列深度”和“线程池活跃度”两个指标任何一个高于阈值就向上报告unhealthy。判断逻辑里不能只看进程活着还要看这些“活着的进程”是否真的能干活否则调度就是在给一群假活 Agent 疯狂派单。4.4 调用链爆炸没设置跳数上限的代价有一次我在编排任务时忘了给链路设置最大跳数结果在 “分析 → 撰写 → 分析 → 校验 → 分析” 这类循环协作场景里A 调用 BB 又调回 AA 再调 B几分钟内生成几百条消息。这个问题排查起来非常绝望因为每一条消息本身都长得好像是“正常流程的一部分”。后来我在消息信封里加了一个ttl_step字段每次经过一跳这个值减 1减到 0 就直接丢进死信。默认值是 8能覆盖绝大多数正常业务链路但又能及时切断恶性循环。4.5 全链路日志聚合用 trace_id 抠出完整现场没有链路追踪的分布式系统出问题就是大型考古现场。我最初每台 Agent 只打自己的日志任务出问题后大家都在自己的日志里搜关键字几十个人对着各自的屏幕猜。后来强制所有 Agent 在日志采集时统一按 trace_id 拆包把 trace 相关的整条链路的日志和状态流转汇聚到一个查询入口。现在排查时只需搜索 trace_id就能看到这条消息被谁创建、谁接收、谁 ACK、哪一步耗时爆炸、哪个 Agent 返回了错误码。虽然这个过程要花点开发成本但每次生产事故的定位时间从小时级降到了分钟级我觉得非常值。写到这里我想把 Agent-Reach 这个项目最后再往深挖一层。它解决的核心问题表面看是 Agent 之间的通信和调度但本质上是在为多 Agent 系统建一条“可信的协作管道”。Agent 的单体能力再强在协作环节里一旦消息触达不可靠、链路不可追踪整个系统的可用性就会瞬间垮掉。我个人在做这个项目时的体会是与其把精力都花在调整单一 Agent 的 prompt 上不如分出一部分时间去治理 Agent 之间的“运输系统”。一台车引擎再好路网到处是断头路跑不了长途。最后再分享一个小技巧如果你要把这套方案落到自己的团队我建议从最细的负面场景开始测比如断网、Agent 进程被杀、消息重复投递、队列积压把这些故障注入到测试环境里跑一遍再谈优化业务。我在 Agent-Reach 里做设计时很多参数都被“故障注入测试”说服过那些看起来最优的配置参数真到故障时刻才知道是不是保命设计。搞多 Agent 系统的乐趣也就在这儿十成的功夫往往有三成都花在那些不会被用户直接看见但一崩就是事故的管道里了。