Agent-Reach：面向多智能体协作的轻量级触达与调度中间件

1. 项目概述：Agent-Reach 到底解决什么问题

先说结论：Agent-Reach 是一个面向多智能体（Multi-Agent）协作场景的轻量级触达与调度中间件。它解决的核心问题不是"让单个 Agent 变聪明"，而是"让多个 Agent 找得到彼此、连得上服务、收得回结果"——这恰恰是实际落地中最容易翻车的一段。

如果你做过 Agent 类项目，大概率遇到过这些情况：写好的工具函数散落在各个服务里，Agent 调用时靠硬编码 URL；新加了一个 Agent 能力，老模块的代码要跟着改；任务分发出去之后，某个子任务超时到底是谁的锅，排查起来全靠日志翻找。Agent-Reach 就是把这些"脏活累活"收敛成一个统一层：服务注册、路由选择、消息投递、结果回传、异常兜底，全都在这一层完成。

这套东西适合谁来用？两类人最合适。一类是正在做多 Agent 编排的技术人，团队里已经有三五个 Agent 服务在跑，但互相之间是"手拉手"式的点对点调用，需要一个更规范的调度层；另一类是准备从单体 Agent 往多 Agent 架构迁移的同学，想提前把"触达协议"和"路由策略"设计好，避免后期推倒重来。单个 Agent 的玩具 Demo 用不上它，但一旦你的 Agent 数量超过 3 个、依赖的服务超过 5 个，这套思路的价值就会立刻显现。

我最初设计这个项目时，参考的是微服务里非常成熟的"注册中心 + 网关"模式，但做了大量裁剪和面向 Agent 场景的重构。Agent 和普通微服务有个本质区别：微服务的调用方是人写的代码，参数和返回结构相对固定；Agent 的调用方可能是另一个 Agent，它自己会用自然语言理解任务、拆解步骤，于是"调用"这件事本身就充满了不确定性。Agent-Reach 要做的，就是把这种不确定性挡在调度层之外，让上层 Agent 只管"表达意图"，底层服务只管"执行能力"。

2. 整体设计与思路拆解

2.1 为什么不能用普通 HTTP 硬调

最早我也试过最朴素的做法：每个 Agent 暴露一个 HTTP 接口，其他 Agent 直接 requests.post。这套方案在小规模下完全能用，问题出在三个地方。

第一是地址管理混乱。Agent A 要调用 B 的能力，开发时写的是 localhost:8001，部署后要改成服务发现地址，环境一多配置就爆炸。更麻烦的是，B 升级后换了端口或者拆成了 B1、B2 两个实例，A 的代码就要改。第二是协议不统一。有的 Agent 接受 JSON，有的接受 XML，有的要表单格式，调用方得为每个被调对象写一套适配逻辑。第三是失败处理完全靠自觉。超时重试、熔断降级、结果校验这些事，每个 Agent 各写各的，质量参差不齐。

Agent-Reach 的设计目标就是把这三点一次性解决。它把"触达"抽象成三个动作：发现（Discovery）、路由（Routing）、投递（Delivery）。发现解决"谁能干什么"，路由解决"这件事找谁干最合适"，投递解决"怎么把任务送出去、结果怎么拿回来"。

2.2 核心概念：三种角色

Agent-Reach 里有三个核心角色：能力提供方（Provider）、能力消费方（Consumer）、调度中枢（Broker）。

能力提供方就是真正干活的服务。它在启动时向 Broker 注册自己拥有的能力，比如"我可以做文档摘要""我可以调用天气 API""我可以计算物流运费"。注册信息包括能力名称、入参 schema、出参 schema、QPS 预估、超时上限等元数据。

能力消费方是发起任务的一方。它不需要知道 Provider 的地址、端口、鉴权方式，只需要告诉 Broker 自己要什么能力，Broker 负责把请求路由到合适的 Provider。

调度中枢是全部核心逻辑的承载者。它维护一份能力注册表，处理注册、注销、心跳续约；它执行路由策略，决定一次请求交给哪个 Provider；它还要做结果汇聚，把多个 Provider 的返回组装成统一结构。

这个三角色模型其实是参照了微服务架构里注册中心 + 服务消费者的关系，但有一处关键改动：Agent-Reach 的路由不是简单按权重轮询，而是支持按"能力匹配度"和"上下文约束"来做决策。比如某个 Agent 声称自己能做"文本分类"，但注册时填的入参是英文文本，这时候来一条中文文本请求，路由层就要能识别这个不匹配，宁可报错也不要乱转发。

2.3 设计时优先考虑的三个原则

我在做架构取舍时，给自己定了三条原则，全程没有偏离。

第一条，调度层不感知业务语义。Broker 只做路由和投递，不解析任务内容。它知道"这条请求对应卡片分类能力"，但不知道"卡片分类"具体怎么实现。这样可以保证 Broker 本身很轻，不会变成业务逻辑的大杂烩。

第二条，Provider 无感接入。任何已有的工具函数、API 服务、模型推理接口，只要包一层 Agent-Reach 的适配器就能接入，不需要改内部实现。这一点非常重要，因为实际项目里很多老代码是不能动的，能用一个薄适配层接进来是最优解。

第三条，失败是可观测的。每一次触达都有全链路追踪 ID，从请求进入直到结果返回，每个环节的耗时、状态、路由决策理由都会被记录下来。这样出了问题不是靠猜，而是直接看链路数据。

3. 核心细节解析与实操要点

3.1 Provider 注册协议设计

注册协议是整个系统的地基。我设计了一个 JSON 格式的注册信息，核心字段包括能力标识、版本、入参 Schema、出参 Schema、运行时信息。这里我踩过一个坑：早期版本把 Schema 直接写死在代码里，Provider 升级后忘记更新注册信息，Broker 还在按旧 Schema 校验请求，导致一大批调用失败。后来我把 Schema 校验挪到了 Provider 侧启动时做一次自检，Broker 只做字段格式的粗校验。

注册信息的核心结构大致是这样的：

{ "service": "text-analyzer", "version": "1.2.0", "skills": [ { "name": "sentiment_analysis", "description": "对中文文本进行情感极性判断", "input_schema": { "type": "object", "properties": { "text": {"type": "string", "minLength": 1}, "lang": {"type": "string", "enum": ["zh", "en"], "default": "zh"} }, "required": ["text"] }, "output_schema": { "type": "object", "properties": { "label": {"type": "string", "enum": ["positive", "negative", "neutral"]}, "confidence": {"type": "number", "minimum": 0, "maximum": 1} }, "required": ["label", "confidence"] }, "timeout_ms": 3000, "qps_limit": 50 } ], "heartbeat_interval": 15 }

注意几个容易被忽视的细节。timeout_ms 是 Provider 自己声明的最大处理时长，Broker 路由时会参考这个值来决定请求级超时，避免出现 Provider 明明只能 3 秒完成、调度层却给了 30 秒等待期的错配。qps_limit 用来做服务端的自我保护，Broker 发现有 Provider 接近瓶颈时，会自动把新请求路由到其他副本。

3.2 路由策略的实现逻辑

路由是整个系统里最有嚼头的部分。我实现了三种策略，可以组合使用。

第一种是最小连接数（Least Connections）。Broker 会统计每个 Provider 当前在途的请求数量，新请求优先分给在途数最少的那个。这个策略实现简单，对大多数场景已经够用。

第二种是基于标签的亲和路由（Affinity Routing）。有些 Agent 服务绑定了特定数据源，比如某个 Provider 只处理华东区域的订单，那它的注册信息里会带一个region=cn-east的标签。请求里如果带有同样的上下文标签，就优先路由到这个 Provider。这个策略要求请求方在发起时能正确传标签，所以我会在 SDK 里做一个隐式注入：只要 Consumer 初始化时声明了自己的区域信息，所有请求都会自动带上。

第三种是基于能力的语义预筛（Semantic Pre-filter）。这是 Agent-Reach 区别于普通网关的地方。Broker 会在路由前做一次轻量级的入参预检，把请求里的关键参数和 Provider 注册的 Schema 做一次结构比对。比如请求里要求返回格式是{"label": string, "confidence": number}，而候选 Provider 注册的输出里根本没有 confidence 这个字段，那这个提供方会被直接跳过，即使它的 QPS 最低。这个预筛不涉及语义理解，只是格式层面的匹配，所以耗时可以控制在毫秒级。

路由优先级我定为：语义预筛通过 > 标签亲和 > 最小连接数。语义预筛是第一道闸门，保证候选集是"有能力做"的；标签亲和是第二道闸门，保证候选集是"应该做"的；最小连接数做最终分配。

3.3 消息投递与结果回传机制

投递这块我用了两种模式：同步等待（request-reply）和异步回调（callback）。同步模式适合耗时短、对实时性要求高的任务，比如"帮我查一下天气"。异步模式适合耗时长、可以慢慢跑的任务，比如"帮我生成一份完整的数据分析报告"。

同步模式要注意超时控制。我采用的是"阶梯式超时"：Broker 给 Provider 的时间是 Provider 自报的 timeout_ms 加 20% 的缓冲；Consumer 侧的超时则会根据任务优先级再乘一个系数。这样设计是为了避免一个慢任务把整个消费线程池占死。实测里，把超时从固定 10 秒改成阶梯式之后，同资源下的吞吐量提升了将近一倍。

异步模式的回传地址不固定。我的做法是 Consumer 在发起任务时带一个reply_to字段，这个字段可以是一个消息队列的 Topic，也可以是一个 HTTP 回调地址。Broker 收到 Provider 的结果后，不做任何加工，原样投递给reply_to。这里有一个非常重要的约定：结果必须带request_id，否则回调方根本不知道这是哪次任务的结果。

异步模式还有一个"结果状态"的概念。Provider 处理超时后，Broker 会主动标记该任务失败，并按规则决定是否重投。这里我实现了两个重试级别：Provider 级重试和 Consumer 级重投。Provider 级重试是同一个请求在 Provider 内部再跑一次，适合那种"上次是因为依赖服务抖动才失败"的情况；Consumer 级重投是 Consumer 重新发起整个任务，适合"Provider 本身已经不可用"的情况。

4. 实操过程与核心环节实现

4.1 基础设施选型

我实现 Agent-Reach 时用的技术栈比较克制：Python 3.10 + FastAPI + Redis + PostgreSQL。选择这些不是为了炫技，而是综合考虑了生态成熟度和维护成本。

Broker 用 FastAPI 写，因为异步支持好，天然适合高并发 IO 密集场景。Redis 用来做注册表的缓存和心跳计时，因为注册信息读多写少，放 Redis 里性能足够，还能顺带做分布式锁。PostgreSQL 保存注册历史、路由日志、任务状态这些结构化数据，方便后续做分析查询。

消息投递层我没有引入重量级的 MQ，初期用了 Redis Stream 做了个轻量版。等到任务量真的上来，再平滑迁移到 RabbitMQ 或者 Kafka 也不难，因为我在投递接口上做了抽象，底层实现可以替换。

4.2 Broker 核心模块实现

Broker 拆了四个子模块：注册管理、路由引擎、投递执行、监控统计。

注册管理的核心是一个SkillRegistry类。它维护三张表：技能表（skill_code -> 提供方列表）、提供方表（service_id -> 实例元数据）、路由缓存表（路由结果的 LRU 缓存）。

class SkillRegistry: def __init__(self, redis_client): self.redis = redis_client self._cache = {} async def register(self, service_meta: ServiceMeta): key = f"skill:{service_meta.service_id}" # 写入 Redis 哈希结构 await self.redis.hset(key, mapping=service_meta.to_dict()) # 更新技能索引 for skill in service_meta.skills: await self.redis.sadd(f"skill_code:{skill.name}", service_meta.service_id) # 同时标记缓存失效 self._cache.clear() async def get_candidates(self, skill_name: str) -> list: cached = self._cache.get(skill_name) if cached: return cached service_ids = await self.redis.smembers(f"skill_code:{skill_name}") candidates = [] for sid in service_ids: raw = await self.redis.hgetall(f"skill:{sid}") candidates.append(ServiceMeta.from_dict(raw)) self._cache[skill_name] = candidates return candidates

这里我做过一个优化：候选列表缓存了 10 秒，而不是每次请求都实时查询 Redis。因为技能注册表的更新频率远低于查询频率，10 秒的延迟完全不影响正确性，却能显著降低 Redis 的压力。

路由引擎是独立于注册管理的纯函数模块。它接收四个输入：当前请求的 SkillRequest、候选 Provider 列表、每个 Provider 的实时在途数、请求上下文标签。输出是一个选定 Provider 的 service_id 和路由理由。

async def route(router, request: SkillRequest, candidates: list, in_flight: dict, context: dict): # 第一层：语义预筛 valid = [] for cand in candidates: if not validate_schema(cand.output_schema, request.expected_output): continue if not validate_schema(cand.input_schema, request.input_args): continue valid.append(cand) if not valid: return None, "no_candidate_after_prefilter" # 第二层：标签亲和 tagged = [c for c in valid if match_labels(c.labels, context)] pool = tagged if tagged else valid # 第三层：最小连接数 best = min(pool, key=lambda c: in_flight.get(c.service_id, 0)) return best.service_id, "least_connections"

选型时我纠结过一件事：要不要把路由逻辑做成插件化。后来还是决定先用简单的策略函数 + 工厂模式，不用太早引入插件机制。原因很简单，路由策略的变更是低频的，比起灵活，现阶段更看重代码的简单可读。真到了需要扩展策略的时候，把 if-else 替换成策略注册表成本也不高。

4.3 Consumer SDK 封装要点

Consumer SDK 是给上层 Agent 用的接口，必须足够"无感"。我提供了 Python 和 TypeScript 两个版本，核心 API 就三四个。

from agent_reach import Consumer # 初始化时指定所属应用名 consumer = Consumer(app_name="billing-assistant", broker_url="http://broker:8080") # 定义期望的输出结构 expected = { "type": "object", "properties": { "express_no": {"type": "string"}, "status": {"type": "string"} } } # 请求某个技能，自动完成路由、投递、结果校验 resp = consumer.invoke_skill( skill_name="express_query", input_args={"order_id": "SO202405001"}, expected_output=expected, is_async=False, timeout_hint=8 )

SDK 内部做的事情比我一开始想象的多得多。它要自动完成注册发现、鉴权 token 刷新、上下文标签注入、全链路 trace 拼接、超时控制、异常重试。我写了将近 800 行代码才把这些细节全部收干净。

有个细节值得分享：SDK 启动时会做一次"预热连接"，也就是先从 Broker 拉一次全量技能清单缓存到本地，之后每次请求先去本地缓存查一下目标技能是否存在。如果本地缓存没有，才真正去 Broker 查询。这样设计可以减少一次跨网络请求，在请求量大的时候效果很明显。

4.4 参数配置参考表

我整理了一份关键参数清单，基本都是我用默认值跑了一段时间后手动调过的，可以直接抄作业：

这些参数没有"标准答案"，一定要根据自己的部署环境、任务类型、流量模型来定。我刚开始就是照搬了微服务那套默认值，结果在公网环境频繁误摘健康节点，后来逐项调才稳定下来。

5. 常见问题与排查技巧实录

5.1 注册风暴与集群脑裂

第一种我实际遇到过的"事故"是注册风暴。一个 Provider 因为网络抖动反复尝试重连，每秒钟发出几十次注册请求，直接把 Broker 的 CPU 拉满了。我排查时先从监控面板看到register接口的 QPS 异常飙升，然后顺着日志找到了那个疯狂重连的服务实例。

解决方案是在 SDK 侧加指数退避的重试机制：第一次重试等待 1 秒，第二次 2 秒，第四次 8 秒，最多退避到 60 秒。同时 Broker 侧再加一层限流，同一个 service_id 每 10 秒最多接受 5 次注册请求，超出的直接丢弃并返回429 Too Many Registrations。这样即使 SDK 没更新，也不会拖垮 Broker。

第二种坑是集群脑裂导致的注册信息丢失。我早期只部署了一个 Broker 节点，后来为了高可用加了第二个节点，但注册表只在单节点的 Redis 上存了一份，节点切换后注册信息全丢，所有 Provider 被迫重新注册。后来我把注册存储改成了 Redis Cluster 并配置了持久化，同时 Broker 启动时会触发一次全量re-sync请求，强制所有在线 Provider 重新上报元数据，才算把这个洞堵住。

5.2 路由到不可用节点

这是多实例部署时最容易踩的坑。你明明在注册表里能看到两个 Provider 实例，但请求发过去后，其中一个老是超时，另一个却稳定返回。问题往往出在"注册信息和真实运行状态脱节"。

我在一次复盘中发现，某个实例已经内存泄漏到只剩下 5% 可用内存，但它的心跳还在正常上报，因为心跳线程只检查进程存活，不检查资源水位。Broker 把这个"僵尸实例"认为健康节点，继续往它头上堆请求，结果当然是超时一堆。

解决办法是在注册信息里加一个health_score字段，Provider 每 10 秒过来心跳时，把当前的 CPU、内存、在途请求数一起带过来。Broker 侧设置一个阈值，比如health_score < 60的节点自动摘除，不参与路由。这个字段用不着像监控系统那么精确，能反映相对健康度就够了。

5.3 异步任务结果对不上账

异步模式最容易出的问题就是"结果回来没人认领"。我碰到过 Consumer 方填的reply_to地址写错，结果 Provider 确实处理完了，但结果投递到了不存在的 Topic 上，任务状态永远停在"处理中"。

后来我在日志里专门加了"孤儿任务检测"：如果一个异步任务超过最大处理时间两倍后既没有回传结果，也没有被标记失败，Broker 会主动发提醒事件给 Consumer。Consumer SDK 收到这个事件后，可以查询任务状态并给出告警或自动降级。

还有一次是 Consumer 消费回调消息时挂了，重启后那批结果消息全丢了。这个问题是因为我早期用 Redis Stream 消费后没有保存处理进度。后来改成了消费 + 持久化两步式：先保存到本地数据库，再确认消息。虽然多了一步磁盘写入，但保证了不丢消息。

5.4 问题排查速查表

关于排查，我还有一个小建议：不要把日志当成事后武器，而是要在系统设计阶段就把关键链路的数据留全。Agent-Reach 里每一次路由决策都会记录路由理由，每一次投递都会记录 target 和耗时。这些数据平时没人看，但一旦出问题，它们就是唯一的线索。

6. 项目扩展方向与个人建议

Agent-Reach 目前的版本已经能支撑我手上的三个实际项目稳定运行，但我知道它离"完美"还差得远。接下来我想做两件事。

第一件事是增加"能力协商"机制。现在的模式是 Consumer 声明需求，Broker 匹配 Provider。但实际场景里，Provider 的能力可能不是静态的，比如某个模型服务同时支持摘要和分类，但摘要的并发一高分类的响应就变慢。我计划让 Provider 在心跳上报时带一个动态能力水位，Broker 在路由时综合考虑水位和请求优先级。

第二件事是完善多级路由中的上下文传播。当前标签亲和依赖 Consumer 主动传标签，但很多场景下上下文是需要自动推导的，比如根据请求里的 IP 判断区域、根据订单号识别业务线。我准备在 Broker 里加一个"上下文解析器"的插件位，把这类推导逻辑从路由主流程里拆出去。

最后分享一点个人体会。做这类中间件项目，最忌讳一上来就追求功能大全、架构完美。我一开始也很想把路由策略做成 SPI 插件、想引入规则引擎、想做跨集群联邦，但后来都被自己一个个砍掉了。真正的经验是：先把一条主链路跑通、跑稳，让三个 Provider、两个 Consumer 在真实环境里连续跑一周不出问题，再谈扩展。稳定性和可维护性，永远是这类系统的生命线。