Agent-Reach：多智能体协作的任务路由与消息协调框架

发布时间：2026/10/15 9:08:57

1. Agent-Reach 是什么从一个真实需求说起大概半年前我在做一个内部工作流自动化项目一开始以为最难的环节是模型能力结果真正逼疯我的反而是 agent 之间的通信问题。项目里有三个角色一个负责解析用户意图的入口 agent一个负责查库存数据的数据 agent还有一个负责发通知的执行 agent。单看每个 agent 都跑得好好的一旦串起来就各种掉链子——A agent 调用 B agent 时不知道 B 挂了没有B 返回的结果 A 读不懂任务中途超时后没有人接着处理。那段时间我大部分精力不是在调模型而是在写各种胶水代码处理 agent 之间的超时、重试和状态同步。后来我把这层胶水逻辑从业务代码里抽出来做成了一个独立的协调服务这就是 Agent-Reach 的雏形。Agent-Reach 本质上是一个面向多 agent 协作场景的轻量级任务路由与消息协调框架。它解决的问题可以概括成一句话让一群彼此不知道对方存在的 agent能够稳定、可靠地互相找到对方并把任务传递下去。如果你也在做 agent 编排、工具调用链、或者任何形式的 autonomous agent 协作这个项目和相关思路应该能给你一些启发。这套框架的核心价值有三个。第一是注册与发现agent 启动时自动登记自己的能力和地址其他 agent 不需要写死调用关系靠名字和能力标签就能找到对方。第二是任务路由消息不是直接点对点投递而是按路由规则走支持优先级、超时、重试、死信队列。第三是状态追踪每个任务从发起到完成的全链路状态都记录在案出问题时可以按 trace 回溯到底哪一步断了。我把它开源出去之后不少朋友来问实现细节也有人提出想直接用在生产环境。这篇文章就趁热把 Agent-Reach 的设计思路、核心实现、以及我在实际跑的过程里踩过的坑系统梳理一遍给正在做 agent 编排的同学一个可以抄作业的参考。2. 设计取舍为什么我放弃了“直连调用”而选择“路由中转”2.1 直连模式的问题最早我的设计其实是直连调用——agent A 直接通过 HTTP 调 agent B 的接口。这种方式写起来最省事一两个 agent 的时候也确实够用。但当 agent 数量增加到五六个、调用链变成 A - B - C - D 这种多级传递时问题就全冒出来了任何一个节点挂掉整条链就断。B 挂了之后 A 不知道只能反复重试到超时然后整个任务失败。我那时候加了一堆 try-catch 和重试逻辑代码越写越乱。层级调用没法追踪。A 调 BB 又调 C如果 C 返回了一个错误A 根本不知道这个错误是 C 来的还是 B 处理过程中出的。耦合太深。agent 之间互相知道对方的接口地址和参数格式改一个 agent 的入参结构所有上游调用方都要跟着改。这就是典型的分布式系统里“点对点连接不可扩展”的问题。我自己以前写过消息队列相关的中间件心里其实清楚解决方案是什么但一开始为了图快还是走了直连的老路结果就是不断返工。2.2 路由中转带来的三个关键能力后来我痛定思痛把所有 agent 之间的调用全部改成走 Agent-Reach 中转agent 之间不再直接通信。这个改动带来了三个直接的好处解耦。agent 只需要知道自己能干什么、需要什么格式的输入不需要关心调用方是谁。新加一个 agent 的时候只需要让它启动时向 Agent-Reach 注册一下能力和地址其他 agent 完全不用改代码。可控。所有的消息流经一个中心节点我可以在中间加拦截器做日志、限流、鉴权、格式化转换。以前想统计每个 agent 的调用量得在每段代码里埋点现在直接在 Agent-Reach 里统一统计方便太多。可靠。中转节点可以缓存消息、重试投递、兜底处理。即使目标 agent 暂时不可用消息也不会立刻丢失可以有策略地等待或转给备用节点。当然引入中心节点不是没有代价。最大的代价就是多了一跳网络开销以及 Agent-Reach 本身成了单点。如果它挂了整个协作链路就瘫了。这个问题的应对我在后面第五节详细讲先不展开。提示如果你的场景里只有两个固定的 agent 互调且调用频率极低、失败影响可控直连确实也没毛病。Agent-Reach 这类框架是给“多个 agent、动态编排、链路较长”的场景用的别为了用而用。3. 核心实现注册表、路由规则与状态追踪3.1 Agent 注册表的数据结构Agent-Reach 的注册表是整套系统的地基。每个 agent 启动时先调 Agent-Reach 的注册接口上报自己的基本信息。我用的结构大致是这样的{ agent_name: inventory_query, version: 1.2.0, capabilities: [inventory.query, inventory.batch_query], endpoint: http://10.0.0.12:8081, health_check: /healthz, max_concurrency: 20, timeout_ms: 3000, priority: 5 }这里几个字段的设计逻辑说一下。capabilities是关键它决定了路由能力。调用方发任务的时候不需要指定“我要发给 inventory_query 这个 agent”而是指定“我要完成 inventory.query 这个能力”由 Agent-Reach 根据能力标签找到当前可用的 agent。这样做的好处是如果有两个 agent 都提供 inventory.query 能力它们可以做负载均衡如果一个 agent 下线了路由会自动指向另一个调用方无感知。health_check字段是我后来加的。Agent-Reach 会定期默认每 10 秒去请求一下这个健康检查端点如果连续两次超时就把对应 agent 标记为不可用不再给它路由新任务。这个机制看着简单但实际效果非常好避免了之前那种“消息发出去了对方已经死了”的无效调用。3.2 任务消息的封装格式路由的核心是消息格式。我在 Agent-Reach 里定义了一套统一的任务信封所有 agent 之间传的数据都塞在这个信封里。结构如下dataclass class TaskEnvelope: task_id: str # 全局唯一任务IDUUID trace_id: str # 链路追踪ID同一条调用链共享 capability: str # 需要的能力标签如 inventory.query payload: dict # 业务数据 timeout_ms: int # 本次任务的超时时间 max_retries: int # 最大重试次数 callback_topic: str # 结果回执的Topic用于链式传递 priority: int # 优先级0-9数字越大优先级越高 created_at: floattrace_id这个字段强烈建议从一开始就留好。我在早期版本里没有这个字段导致排查链路问题时极其痛苦——A 调 B 调 C中间某一步出错我只能顺着日志一条条人肉翻。后来加了trace_id每次调用在 Agent-Reach 里都会打一条结构化日志带上 trace_id 就能把所有相关环节一次性捞出来。现在排查一个失败任务基本是秒级定位。3.3 路由选择器的实现思路路由选择器负责回答一个问题给定一个 capability 标签应该把任务投给哪个 agent我实现的是一个带权重的最小连接数算法大体思路是这样def select_target(capability, available_agents): candidates [a for a in available_agents if capability in a[capabilities]] if not candidates: raise NoAvailableAgent(capability) # 按当前并发数排序优先选最空闲的 candidates.sort(keylambda a: a[current_load] / a[max_concurrency]) return candidates[0]再加上一层权重分配。比如两个节点机器配置不同一个 16 核一个 4 核那我希望压力尽量倾斜到大机器上就给大机器设一个更高的权重系数实际负载当前任务数 / (max_concurrency * weight)。这个权重在注册的时候通过priority字段间接控制也在 API 里开放了手动调整接口。3.4 任务状态机Agent-Reach 里的每个任务都走一条明确的状态流CREATED - ROUTED - DELIVERED - PROCESSING - SUCCEEDED / FAILED。特殊情况下还会进入RETRYING、DEAD_LETTER。状态机这个东西不算复杂但它把所有边界情况都统一到了一条流程里。“目标 agent 没回应”和“目标 agent 返回了错误码”这两种情况处理逻辑完全不同。前者是投递层的问题需要重试投递后者是业务层的问题应该直接记失败并回执给调用方。如果没有状态机很容易把这两种情况混在一起处理重试策略就很难写对。我把状态流转图直接写成了代码里的一个字典避免状态流转失控STATE_TRANSITIONS { CREATED: {ROUTED, FAILED}, ROUTED: {DELIVERED, RETRYING, FAILED}, DELIVERED: {PROCESSING, RETRYING, FAILED}, PROCESSING: {SUCCEEDED, FAILED, RETRYING}, RETRYING: {DELIVERED, FAILED}, FAILED: {DEAD_LETTER}, SUCCEEDED: set(), }每次状态变化都会附上时间戳这样全链路的时间消耗一目了然。哪一步最慢、卡了多久直接从状态里读出来就行。4. 消息投递与超时重试稳定性是怎么抠出来的4.1 HTTP 投递还是消息队列投递Agent-Reach 现在的默认投递方式走的是 HTTP 长轮询其实早期我纠结了很久要不要直接依赖现成的消息队列中间件。后来决定不依赖主要原因有两个。第一我不想给使用者增加额外的部署负担。消息队列意味着多维护一套集群单机小规模场景里属于杀鸡用牛刀。第二agent 之间的通信天然是请求-响应模式和单纯的消息异步推送不完全一样——A agent 发了任务给 B它通常还是想知道 B 处理结果的。用消息队列实现请求-响应模式需要自己维护关联 ID 和等待队列复杂度反而上去了。所以 Agent-Reach 的做法是投递通过 HTTP 接口调用目标 agent 的/agent/invoke端点目标 agent 处理完异步回调 Agent-Reach 的回执接口把处理结果附上task_id传回来。这样既保留了 HTTP 的简单直接又不用调用方同步阻塞等待。4.2 三种超时及其处理方式超时是 agent 协作里最常见的问题没有之一。我在 Agent-Reach 里把超时分成了三层连接超时Agent-Reach 向目标 agent 发起 HTTP 请求时建立连接的时间。这个值一般设 500ms 就够了如果 500ms 连不上说明目标 agent 大概率已经不可用或者网络有问题。读取超时请求发出后等待目标 agent 返回响应头的时间。我默认设为 3 秒。agent 的 HTTP 端点接收任务后应该立刻返回 202 Accepted表示“我收到任务了”然后异步处理处理完再回调。如果超过 3 秒连个 202 都没拿到说明目标 agent 已经被卡死了按投递失败处理。业务处理超时这是从投递成功到收到回执之间的最大允许时间。这个值没法硬编码因为不同任务复杂度差异太大——查个库存可能 1 秒跑一次深度分析可能需要十几秒。所以我在 TaskEnvelope 里增加了timeout_ms字段由调用方按需指定。Agent-Reach 会起一个定时器检查如果超时未收到回执就把任务标记为FAILED并触发重试逻辑。三种超时对应三种失败处理策略连接超时直接换节点重投读取超时做最多 2 次重试业务处理超时则要谨慎——因为任务可能其实已经在目标 agent 那边执行成功了只是回执丢了贸然重试可能导致重复执行。这种情况我会在任务信封里带一个idempotency_key目标 agent 收到带相同 key 的重复任务时直接返回上一次的结果缓存不做二次执行。这是所有 agent 协作场景都建议提前设计的一个机制。4.3 指数退避重试重试不能死心眼地固定间隔。Agent-Reach 用的是指数退避加抖动import random import time def next_retry_delay(attempt, base_delay1.0, max_delay30.0): exp_delay min(max_delay, base_delay * (2 ** attempt)) jitter random.uniform(0, exp_delay * 0.1) return exp_delay jitter为什么要加抖动如果没有抖动多个失败的任务可能同时重试形成惊群效应反而把目标 agent 打得更死。加一个 10% 范围内的随机偏移代价可以忽略不计收益是能明显平滑重试流量的尖峰。这个细节是我实际压测时发现的——三个 agent 同时失败后不带抖动的重试在下一分钟制造了三次几乎同步的流量洪峰加上抖动之后流量曲线平滑了很多。4.4 死信队列的兜底重试到最大次数还是失败的任务Agent-Reach 不会直接丢掉而是丢进死信队列。死信队列每条记录保留完整的 task_id、payload、失败原因、重试历史。我提供了一个管理界面可以手动从死信队列里捞任务重放也可以设置定时任务在目标 agent 恢复后自动重放。这是我们生产环境里一个特别实用的功能。有一次数据 agent 因为一个 SQL 写错了导致反复失败重试全部耗尽了任务全进了死信队列。修完 SQL 之后我直接在管理界面上按条件筛选把积压的几百个任务一键重放全部跑通了。如果没有这个兜底那几百个任务就只能靠人肉补数据了。5. 生产环境实战我踩过的三个大坑5.1 Agent-Reach 自身单点挂了怎么办这是中心化架构永远躲不开的问题。第一版上线跑了一周Agent-Reach 所在进程因为内存泄漏挂了结果所有 agent 互相之间再也找不到对方整条业务链路瘫痪了一个多小时。这真是血泪教训。后续我做了三个加固措施。一是给 Agent-Reach 加了一个轻量的嵌入式存储把注册信息和任务状态实时落盘重启后自动恢复现场。二是加了主动健康检查探针配合进程守护工具自动拉起。三是关键的我修改了 agent 端的 SDK——假设 Agent-Reach 不可用agent 启动时会打印错误日志并进入待命状态但已经缓存了能力路由表的 agent 可以继续处理本地的任务只是无法接受新路由。这套举措上线后再没有出现过全链路瘫痪。单点问题不可能彻底消除但把故障半径从“全挂”缩小到“部分新任务无法路由”这是可接受的。5.2 回执延迟导致的重试风暴上线后不久我观察到一个奇怪的现象某个 agent 偶发性地处理完任务后回执延迟了十几秒才发出来。这本身可能只是目标 agent 调度线程池不够导致的但问题在于Agent-Reach 看在眼里的是“业务处理超时”于是触发重试。而目标 agent 那边第一次的任务其实已经执行完了重试任务又执行了一次两条写入操作同时发生直接把库存数据写重了。这就是我在前面说的业务处理超时重试的陷阱。最终我采取了三层解决方案所有任务信封强制带idempotency_key目标 agent SDK 里默认实现幂等检查。业务处理超时的值从硬编码改为按 capability 的 P95 耗时动态估算默认取最近 50 次成功任务的 P95 再乘 1.5。控制面增加告警如果某个 capability 的 P95 持续上升说明目标 agent 已经过载了这时不应重试而应该直接熔断一段时间。这里顺带说一句观测数据的价值就在这种时候体现。如果一开始没做链路追踪我可能到现在还在人肉排查为什么数据会写重。5.3 能力标签命名失控这是个小问题但很影响体验。项目初期大家命名能力标签比较随意有人写inventory.query有人写query_inventory还有人写InventoryQuery。Agent-Reach 的匹配是对大小写敏感的于是经常出现“明明有这个 agent但路由就是找不到”的情况。后来我定了一套命名规范能力标签统一用领域.动词的小写形式比如inventory.query、order.create、customer.get。同时注册接口里加了一个标签归一化步骤如果发现新注册的标签和已有标签格式不一致会自动在 API 里给出警告。这不是什么高深的技术但制定规范工具检查双管齐下之后这类问题基本绝迹了。6. 从 Agent-Reach 到更通用的建议如果你想做类似的编排框架6.1 先想清楚你的 agent 数量级Agent-Reach 这个方案适合的规模大概是 5 到 50 个 agent、单日任务量在百万以内的场景。超过这个规模中心节点就会成为性能瓶颈需要拆分成多个 Agent-Reach 实例或者引入更成熟的分布式消息中间件。反过来如果你只有 2、3 个 agent那真的不需要这套框架直连或者其他轻量方案更划算。这个规模判断我建议在动手前就做好不然后面重构很痛苦。6.2 从第一天就做链路追踪这个建议我重复了三次因为真的太重要。统一 trace_id、统一日志格式、统一状态流转记录这三件事必须在框架的第一版就内置好。不要等项目上线了再补。我们的经验是agent 数量只要超过四个人肉排查调用链就已经不现实了。链路追踪不是为了好看它是所有稳定性建设的基础——没有它你就无法知道超时发生在哪一跳、失败是投递失败还是业务失败、重试是否真的生效。6.3 预留人工干预的入口自动化系统永远需要人工兜底的开关。Agent-Reach 里的死信队列、管理界面、任务重放本质上都是人工干预的入口。设计这些功能的时候我心里想的场景是凌晨三点某个 agent 挂了运维同学收到告警后能不能不登录业务后台直接在 Agent-Reach 的管理界面里看到故障范围、定位到具体任务、手动改路由或重放把“人最后兜底”这个场景想清楚框架的可用性和成熟度会高一个台阶。Agent-Reach 现在的管理界面做得很朴素但真到了事故发生时它是我第一优先打开的工具——稳定、直观、能直接操作这就够了。坦白说Agent-Reach 到现在还有很多粗糙的地方比如 HTTP 投递的性能上限、中心节点的容灾粒度、管理界面的交互体验这些都有提升空间。但作为一套可以稳定跑业务的 agent 协作骨架它帮我省掉的“胶水代码”和“半夜救火”时间已经远超写它所花的精力。如果你也在做多 agent 编排建议从注册表、统一消息信封、状态机这三个最小要素开始先让链路能跑通、能追踪再一步步补超时重试和死信兜底。架构做得再花哨不如先把这三块吃透。