多智能体通信中间层Agent-Reach：注册发现与路由调度实践

做多智能体系统有一段时间了，最头疼的其实不是单个 Agent 的推理能力怎么调，而是多个 Agent 之间的连接问题。业务方一口气接了十几个 Agent，每个都在自己的沙盒里跑，有的走 HTTP 接口，有的走消息队列，有的干脆只能读本地文件，要让它们协同干活，第一步就得解决"谁在哪、谁能干什么、怎么找到它"这三件事。我当时把负责这三件事的中间层抽出来做了一个独立组件，项目名叫Agent-Reach。这篇文章把我从设计到上线的完整思路、关键代码和踩过的坑都写出来，给准备做多 Agent 编排或者正在被 Agent 通信问题折磨的朋友一个直接能用的参考。

Agent-Reach 解决的核心问题很明确：在多 Agent 架构中提供统一的注册、发现、路由和调度能力，让不同技术栈、不同协议、不同部署位置的 Agent 能够像微服务一样被动态发现和互相调用。它有两大特点，一是轻量，不依赖特定的 Agent 框架，不管是 LangChain、自研 Agent 还是纯脚本封装的服务，只要实现一个注册接口就能接入；二是透明，调用方不需要关心目标 Agent 的地址和协议细节，一律通过统一网关转发。适合正在从单 Agent 向多 Agent 演进、被接口对接和链路管理折磨的团队参考，也适合想理解 Agent 间通信机制的开发者当入门骨架用。

1. 项目背景与核心痛点拆解

1.1 三个必须解决的问题

我做 Agent-Reach 之前，团队里其实有一套"能跑"的多 Agent 协作方式：互相之间用 API Key 直连，每个 Agent 的地址写死在配置文件里。一开始只有三五个 Agent 的时候没问题，等规模到了十几个，问题立刻暴露。

第一个问题是信息孤岛。A Agent 不知道 B Agent 有什么能力，业务方要用某个能力的时候，得人工去查文档、确认哪个 Agent 提供什么功能。一旦 Agent 的版本升级、能力列表变化，文档永远跟不上。第二个问题是协议各异。同一个流程里，有的 Agent 是 FastAPI 服务，有的 Agent 是消费 Kafka 消息的消费者，还有一个是跑在边缘盒子上的 Python 脚本，想让它们互相调用，每一对接关系都得单独写适配代码。第三个问题是调度无序。请求量上来之后没有流量控制，慢 Agent 会被打爆，快 Agent 又闲着，更不用说重试、超时、降级这些基础能力，几乎全靠调用方自己写。

这三个问题叠加在一起，本质上就是"多 Agent 缺乏服务化治理"的问题。我当时的判断是，不能靠人肉维护，也不能靠逼着所有 Agent 统一技术栈来解决，正确做法是加一层中间层，把注册、发现、路由、容错变成基础设施。

1.2 方案选型：为什么不做服务网格，而是自建轻量网关

选型的时候我认真对比过 Service Mesh 方案，比如 Istio。它的能力确实强，但适配到 Agent 场景有几个硬伤。一是重，Agent 不一定都是容器化部署，很多是独立进程，甚至可能是别的团队托管的黑盒服务，想给我的每个 Agent 都加 sidecar 根本不现实。二是语义不匹配，服务网格管的是流量的转发和策略，但它不理解"这个 Agent 能做什么任务"，路由只能靠 URL 和标签，做不了基于能力的语义匹配。

所以我选择自建一个轻量的中央网关，思路其实和微服务架构里的注册中心很像：每个 Agent 启动时向 Agent-Reach 注册自己的标识、能力列表、健康状态和通信地址，调用方只和 Agent-Reach 打交道，由它来做寻址和转发。这样一个星期就搭出了能用的版本，后面再逐步加上路由策略、重试、熔断和观测能力。整体架构不复杂，核心就四个模块：注册中心、路由引擎、消息网关、监控模块。

2. 注册中心与动态发现机制

2.1 Agent 注册的数据结构与状态管理

注册中心是整个 Agent-Reach 的起点，每个 Agent 都必须先把"我是谁、我能干什么、怎么联系我"这份元数据交上来。我这里设计的最小注册模型包含四类字段：基础标识、能力声明、通信地址、状态参数。基础标识是 agent_id 和 agent_name，能力声明是一个结构化的 capabilities 列表，通信地址是 transport 信息，状态参数是心跳间隔、超时时间这些运行配置。

@dataclass class AgentRegistration: agent_id: str # 全局唯一 ID agent_name: str # 可读名称 version: str # 版本号，用于灰度 capabilities: list[str] # ["weather.query", "order.create", ...] transport: dict # {"type": "http", "endpoint": "http://10.1.2.3:8000"} heartbeat_interval: int = 30 timeout_ms: int = 5000

这里有个我一开始没想清楚、后来被教训过的点：capabilities 的命名不能随意。最早我让各团队自由发挥，结果是五花八门，有的叫get_weather，有的叫city_weather_query，语义几乎一样却匹配不上。后来我强制所有能力名采用域.动作.对象三段式规范，比如weather.query.city，路由匹配的准确率一下子就从 60% 提到了 92% 以上。如果你也要做类似的注册机制，一定在第一天就定好能力命名规范，这是后续所有路由策略的地基。

状态管理上，每个 Agent 在注册后默认进入 ONLINE 状态，同时启动一个 TTL 计时器。Agent 每隔 30 秒发一次心跳续约，如果连续三个心跳周期内没有收到心跳，也就是 90 秒，就将状态置为 OFFLINE，不再参与路由，同时触发告警通知对应负责人。这个机制参考了 Consul 的 TTL 健康检查实现，用临时节点来保存 Agent 状态，进程挂掉之后节点自动过期，不需要人工清理，很省心。

2.2 基于能力的语义路由策略

路由是整个 Agent-Reach 的核心，它决定了一个调用请求应该被转发给哪个 Agent。原始版本只做精确匹配，请求带的任务类型必须和 Agent 的 capabilities 完全一致，这在小规模下能跑，但 agent 多了之后经常出现"明明某个 Agent 能完成类似的任务，就因为能力名不完全匹配而路由失败"的情况。

我后来实现的是一套两级路由：先做语义召回，再做精排选择。语义召回部分，把每个 Agent 的 capabilities 用文本嵌入模型算一遍向量，保存到内存索引里；请求进来时，把请求意图也转成向量，用余弦相似度找出 Top-K 候选 Agent。精排部分则综合三个维度打分：能力相似度得分、Agent 当前负载、历史平均延迟。最终得分 = 0.5 × 相似度 + 0.3 × (1 - 当前负载率) + 0.2 × (1 - 延迟归一化值)。

这看起来简单，实际调试的时候有个大坑：文本嵌入模型的选择。我一开始用了一个极小的本地模型，向量维度只有 128，结果"天气查询"和"降雨预报"这种高度相关的表述相似度得分只有 0.4，召回效果很差。后来换成维度 768 的模型，得分涨到了 0.81，路由准确率立刻提升。所以如果你要复刻这个方案，嵌入模型至少选一个中等规模的，别追求极致的轻量。

路由结果会做缓存，同一个请求意图在一分钟内直接命中上一次的路由结果，省去向量计算的耗时。实测下来，冷启动路由平均耗时在 80ms 左右，命中缓存后降到 1ms 以内。对于多 Agent 场景，这个性能完全够用。

3. 消息协议与任务调度实现

3.1 统一消息协议设计

Agent 之间通信最怕的就是协议不统一。我在 Agent-Reach 里定了一套基于 JSON 的通用消息信封，所有 Agent 不管底层用什么传输方式，HTTP、gRPC 还是消息队列，发送和接收的都必须遵循统一的消息结构。

{ "message_id": "req_8f6a2e1c93", "trace_id": "trace_7b3a9f", "source": "agent_a", "target": "agent_c", "task": { "type": "weather.query.city", "payload": {"city": "杭州", "date": "2025-01-15"}, "priority": 5, "timeout_ms": 8000 }, "metadata": { "retry_count": 0, "created_at": 1705300000 } }

这个协议的每个字段都有明确的设计意图。trace_id 用来串联一次完整调用的全链路日志，排查问题的时候靠它可以把十几个 Agent 之间的调用关系完整还原出来，没有这个字段的话，线上出了问题根本没法定位。priority 字段用于调度排队，高优先级的任务可以插队。timeout_ms 是链路级的超时控制，我特别强调这一点，因为它决定了后续所有超时和重试的边界，如果每个环节都用自己的超时时间，总超时时间根本不可控。

传输适配层我实现了三种：HTTP 同步调用、消息队列异步投递、本地进程调用。对外暴露的接口是统一的 async_call，内部根据 Agent 注册的 transport 类型自动选择合适的适配器。这样做的好处是调用方完全感知不到目标 Agent 的部署形态，非常符合"透明调用"的设计目标。

3.2 任务调度、重试与背压控制

调度模块的核心是一个优先级队列加多工作线程的模型。任务进来之后先按 priority 排序，同级任务按先进先出处理。每个工作线程向目标 Agent 发起调用时，遵循一个动态并发控制算法：根据目标 Agent 最近 10 秒的平均响应时间和成功率，动态计算最大并发数。公式是 MaxConcurrency = MaxBase × SuccessRate × (TargetLatency / AvgLatency)，其中 SuccessRate 和 AvgLatency 都是滑动窗口统计值。

这个算法本质上是在做背压控制。比如某个 Agent 平时响应 100ms，成功率 99%，那么它的最大并发可以开得很大；一旦响应变慢到 2 秒，成功率掉到 80%，最大并发会自动收缩到原来的四分之一，从源头保护下游 Agent 不被流量冲垮。这个机制上线之后，我再也没遇到过某个 Agent 被打挂导致连锁雪崩的情况。

重试策略我采用了带抖动指数的退避算法。默认最多重试 3 次，重试间隔依次为 1 秒、2 秒、4 秒，并且每次加一个 ±20% 的随机抖动。加抖动的目的是防止多个失败任务同时重试形成重试风暴，这个细节很多文档里不会告诉你，但实际生产环境非常重要。

def next_retry_delay(attempt: int) -> float: base = 2 ** attempt # 1, 2, 4 jitter = random.uniform(0.8, 1.2) return base * jitter

需要特别注意一个容易踩的坑：对于幂等性不确定的任务，重试必须非常克制。我最初对所有消息都做统一重试，结果有一个 Agent 的接口不是幂等的，每次重试都产生了一条重复的业务数据。后来我在注册元数据里增加了一个 idempotent 字段，只有声明了幂等的 Agent 才执行自动重试，其他的最多只重试一次，或者直接转人工处理。

4. 可观测性与性能调优实践

4.1 链路追踪与核心指标采集

Agent-Reach 作为中间层，天然可以看到所有跨 Agent 的调用，这是做可观测性最好的位置。我在消息网关层做了一次彻底的埋点，每一个消息从进入网关到返回结果，记录完整的时间戳序列，包括入站时间、路由决策时间、转发出站时间和响应返回时间。配合统一消息协议里的 trace_id，我可以随时查出一条调用在整个链路中每一跳的耗时占比，快速定位慢在哪一个 Agent。

关键指标我采集了四类：QPS、P95/P99 延迟、成功率、路由命中率。路由命中率这个指标非常容易被忽略，但它能直接反映能力注册和语义匹配的健康度。如果路由命中率低，说明大量请求走了兜底逻辑或者直接失败，大概率是能力命名不规范或者语义嵌入模型需要更新了。我建议你把它当成和成功率同等重要的指标来监控。

提示：中间层埋点要做在主流程里，而不是分支逻辑里。我踩过的教训是，最初我把埋点代码写在容错处理分支里，结果正常路径完全没有日志，排查线上问题的时候等于瞎子。

4.2 三个真实性能瓶颈的调优过程

第一次压测就暴露了性能问题，当时的表现是 200 QPS 下 P99 延迟到了 87ms，对于转发层来说完全不合格。我逐步排查，找到三个瓶颈。第一个是向量计算放在请求关键路径上，每次路由都要做一次文本嵌入，单次耗时 30 到 50ms，这个开销完全没必要。解决方式是加一层路由缓存，并且把嵌入计算改为异步预热，请求命中缓存的直接走内存路由表，只有缓存未命中时才触发计算。

第二个瓶颈是 Python 的 GIL 限制，导致多线程调度时 CPU 利用率上不去。我把调度模块改成了多进程加共享内存的设计，主进程负责接收消息和路由决策，四个工作进程负责转发调用，即使中间转发了 5 万个消息也不超过 50ms 的排队延迟。第三个瓶颈是日志和指标写入的同步阻塞，我在高流量路径上把日志改为异步批量写入，单条日志的耗时从 3ms 降到了 0.2ms。

调优结果我在本地环境实测：纯转发模式（不计算路由）下，单机 2000 QPS 时 P99 延迟稳定在 12ms 左右；完整的注册加语义路由模式下，1200 QPS 时 P99 延迟在 39ms，完全能满足中等规模多 Agent 业务的指标要求。压测表格如下，供你做容量规划时参考：

5. 高频问题排查与实用避坑经验

5.1 高频问题速查表

Agent-Reach 上线半年，我处理过不少工单，把最常见的问题整理成了一张速查表，很多都是配置细节导致的行为异常：

排查这些问题的通用方法我建议遵循三步走的思路：第一，先用 trace_id 拉出完整链路，确认到底哪一跳慢、哪一跳失败，不要猜测；第二，看路由命中率和成功率两个指标联动判断是路由问题还是目标 Agent 问题；第三，验证配置项，尤其是超时时间和重试参数，绝大多数偶发问题都出在这两个参数上。

5.2 两个让我印象深刻的实战教训

第一个教训是关于语义路由的"过度自信"。项目前期我被 92% 的准确率鼓舞，把语义路由设为唯一的路由策略，完全没有兜底方案。直到有一次天气数据源大版本升级，Agent 的能力描述措辞改了，嵌入向量全都偏移，结果 15% 的请求路由失败，排查了三个小时才发现是向量索引没有同步更新。后来我加了一个强制规则层，核心敏感任务走精确匹配的规则路由，普通任务走语义路由，同时建立索引版本管理，每次 Agent 能力更新都会触发索引重建。建议你如果做类似系统，从一开始就保留精确规则优先的兜底机制。

第二个教训是淘汰异常 Agent 的时机。最早的实现是检测到 Agent 成功率低于 50% 就立即从路由表中剔除，结果出现了很尴尬的情况：一个 Agent 只是短暂抖动，30 秒后恢复了，但 30 秒内被剔除导致大量请求打到备用 Agent 上制造了新的压力。后来我把淘汰策略改成连续三个统计窗口即 30 秒内成功率都低于阈值才剔除，同时剔除前先发送探活请求做二次确认。加了这层确认逻辑之后，误杀率降低了 80%。这让我认识到，在分布式系统里，不恰当的快速失败其实是一种更隐蔽的危险。

如果要从这个项目里提炼一句最想说的话，那就是：多 Agent 协作成功与否，前提是通信基础设施的健壮度，而不是 Agent 本身有多聪明。我现在回过头看 Agent-Reach 的定位，它不追求大而全，只把注册、发现、路由、容错四件事做到可靠，这个边界感极其重要。项目后续如果继续演进，我会优先把联邦式部署做出来，让多个 Agent-Reach 实例可以跨网络同步注册信息，目前这个方向在内部仿真里效果不错。最后分享一个很实用的小技巧：给每个 Agent 的注册信息加一个 expires_at 字段，强制定期续期，比只依赖心跳删除更能快速清理僵尸节点，这个细节能让你的注册表长期保持干净稳定。