最近不少做多智能体项目的朋友都在聊 Agent-Reach 这个名字。其实最早看到它时我也没太当回事直到自己手底下的 Agent 多了、乱起来才明白这项目到底卡在哪个痛点上。简单说Agent-Reach 解决的是一个特别朴素又特别现实的问题当系统里有几十上百个 Agent它们可能来自不同的团队、跑在不一样的模型上、接口风格五花八门怎么让这些 Agent 互相之间稳定地触达、传消息、把活儿递到对的人或者说对的 Agent手里。单个 Agent 的能力再强如果接不上、呼不动、消息发出去没人认一切都白搭。我这次基于 Agent-Reach 的核心理念自己从零搭了一套最小可用的 Agent 互联调度机制跑通了一个跨 Agent 协作的完整任务流。这篇文章不聊虚的直接讲清楚 Agent-Reach 解决什么、架构怎么拆、代码怎么落以及我实际操作中踩过的坑和调参经验。想自己动手复现的可以直接抄作业遇到问题的也可以拿后面的排查表一个个对照。1. Agent-Reach 到底是什么它捅破的窗户纸在哪1.1 多 Agent 系统最常见的三个尴尬场景我先讲三个真实场景你看看自己中过几枪。第一个场景是“各自为政”。团队 A 做了一个订单查询 Agent团队 B 做了个用户画像 Agent两边都做得挺漂亮。结果产品经理说“帮我做个新功能用户问订单时顺带推荐点东西”。你一看这不就是让订单 Agent 调用户画像 Agent 嘛。但问题来了订单 Agent 是 Python 写的用户画像 Agent 是 Java 写的两边自定义了一套 HTTP 接口参数一个叫order_id一个叫orderId时间格式一个是2025-01-01 00:00:00一个是时间戳。光对齐这些就耗掉大半天。第二个场景是“呼叫无门”。Agent B 挂在某个内网环境里或者跑在某个云函数上没有固定的对外地址。Agent A 想调用它却根本不知道该怎么找到它。“我知道有这么一个 Agent但我联系不上它”——这比“没有这个 Agent”更让人崩溃。因为前者意味着功能明明存在却卡在最后一公里的触达问题上。第三个场景是“重复劳动”。六个 Agent 都需要调用同一个“天气查询”能力每个 Agent 都自己接一遍上游接口自己处理鉴权、超时、错误重试。六份代码逻辑一样写法各不相同哪天上游接口改了个字段六个 Agent 挨个出问题排查起来直接怀疑人生。这三个场景本质上就是同一个问题Agent 之间的互联机制缺失。Agent-Reach 这个名字取得很到位——reach触达。它的核心目标就是让任何 Agent 能以统一、可控、可观测的形式触达其他 Agent。1.2 从名字拆解它的设计哲学我理解 Agent-Reach 的设计哲学可以从三个关键词拆开看。第一个是 Agent。它面向的不是某个具体的 Agent 框架或语言而是“广义的 Agent”。不管你是 LangChain 的 Agent还是自研的状态机 Agent甚至只是一个“看起来像 Agent 的 HTTP 服务”在 Agent-Reach 的视角里都算是可接入的节点。这一点很关键它决定了这套机制的普适性。第二个是 Reach。英文里 reach 表示“到达、影响、联系”。在设计上它就对应着“可达性”和“触达能力”两层意思。可达性是指系统需要有一个机制让 Agent 能被其他人发现并连上触达能力是指消息发出去后对方能收到、能理解、能回应。很多 Agent 协作方案只解决了“能发消息”这一层却没有解决“对方能不能正确理解并执行”这就导致协作流程走走就断了。第三个是 Agent-Reach 整体的定位——它是一层“连接层”或者说“协作中间层”而不是一个 Agent 运行时。它不负责你 Agent 内部怎么思考、怎么规划它只管 Agent 与 Agent 之间的连接、翻译、调度、反馈。这个边界只要清晰了整套系统的复杂度就可控了。1.3 什么场景值得上 Agent-ReachAgent-Reach 不是银弹我并不建议所有项目一上来就搞它。如果你的项目只有两三个 Agent彼此直接写接口调用就行引入额外的中间层反而是负担。但当你的系统出现以下特征时就该认真考虑这层机制了Agent 数量超过 5 个两两之间的调用关系开始变得像蜘蛛网。Agent 由不同团队维护接口风格、数据结构、通信协议不统一。有 Agent 跑在无法直接暴露固定地址的环境里但其他 Agent 必须调用它。需要统一管控“谁能调用谁”的权限边界。需要给跨 Agent 的调用链路做日志、排查、审计。拿我自己这个项目来说最开始只有三个 Agent后来加到七八个每个 Agent 都要互相调用。代码层面撑了俩月实在扛不住了才决定动手做个类似 Agent-Reach 的中间层。做完之后最大的感受是当 Agent 之间不再互相 “直接调接口” 而是“通过连接层间接触达” 时系统的可维护性会有质的变化。好处是你改一个 Agent不需要通知另外五个同步改接口坏处是你要多维护一层基础设施配置和排障的复杂度转移到了这一层。2. 整体架构拆解Agent-Reach 的三大核心组件2.1 核心组件之一Registry——Agent 的通讯录一个多 Agent 系统里如果每个 Agent 都要记住其他 Agent 的地址、端口、接口路径、鉴权方式那配置文件的复杂度会爆炸。Registry 组件就是来解决这个问题的——它是全套系统的“通讯录”或者“服务中心”。每个 Agent 启动时会向 Registry 注册自己的“Agent 描述信息”。这条描述信息至少包含几个字段agent_id全局唯一的 Agent 标识name人类可读的 Agent 名称endpoint这个 Agent 对外接收消息的地址可以是一个 HTTP URL也可以是一个消息队列的主题名甚至是一个内部服务名protocol它听懂什么协议比如agent-reach/json-v1capabilities能力说明用一组固定的能力 ID 来描述比如order.query、user.profile.getschema_url这个 Agent 提供的接口定义文件用于参数校验和协议转换Registry 做了什么维护一张“存活表”。Agent 注册后要定期发送心跳超过阈值没心跳的会自动被标记为离线。其他 Agent 要调用某个能力时不是自己去找连接信息而是先问 Registry谁有order.query这个能力Registry 告诉它一个可用的 agent_id 和 endpoint调用方再去发起连接。我实际做的时候Registry 直接用 Redis 做存储给每个 Agent 一个 hash 结构字段就是上面那些超时时间设成 3 个心跳周期简单够用。但需要注意一个点Registry 如果自己有状态它就变成了单点。解决思路有两个一是让每个 Agent 本地缓存一份 Registry 的全量镜像定期拉取或订阅变更二是 Registry 本身做集群化部署。小项目用方案一性价比最高后续我会细说。2.2 核心组件之二Router——消息的传令兵Router 是 Agent-Reach 里最黏人的组件。它不干重活但所有跨 Agent 的请求都过它。设计 Router 时我纠结过一个问题为什么不走“直连”路线后来想明白了直连模式省事但你要在调用方写一堆容错、超时、重试逻辑。这些逻辑是通用的不该每个 Agent 各写一遍。Router 把这些通用逻辑收敛到了一处。Router 接收调用方 Agent 发来的“意图请求”格式类似{ request_id: req_001, from: agent_a, target_capability: order.query, payload: { order_id: 20250117001 }, timeout_ms: 3000, retry: 2 }Router 拿到这份请求后做四件事先查 Registry找到具备order.query能力的可用 Agent 列表。按调度策略挑一个 Agent。默认策略是“如果只有一个就选那个如果有多个轮询或者随机”。做协议转换。上游 Agent 用的是下划线字段名目标 Agent 用的是驼峰字段名Router 在这里把order_id翻译成orderId。把请求转发给目标 Agent等待响应。如果出现超时Router 会按retry配置在候选列表里换一个重试。Router 还承担一个容易被忽视的任务请求上下文的透传。一次完整的业务请求可能串联 Agent A → B → C中间路由了两次。如果每次转发都把request_id丢掉后面排查链路就断了。我在 Router 里强制要求任何转发请求必须带上原始request_id并且记录每一跳的耗时。2.3 核心组件之三DIA——屏蔽 Agent 方言的翻译层DIA 全名是 Dynamic Invocation Adapter动态调用适配器。名字是我自己取的换到什么项目里都适用。它的作用是解决开头说的“一个字段名不统一、时间格式不统一、接口风格不统一”的问题。我的实现很粗暴但有效每个 Agent 在注册时除了声明 endpoint还要声明 schema。schema 描述它期望的输入输出结构我用的是一份简化版 JSON Schema。DIA 维护了一个类似“翻译字典”的映射表把统一格式的请求翻译成每个 Agent 自己习惯的格式。举个例子。上层统一请求是{ user_id: u_123, start_date: 2025-01-01, end_date: 2025-01-31 }某个 Agent B 接口期望的是{ uid: 123, dateRange: 2025-01-01~2025-01-31 }正常直连的话调用方得写死针对 Agent B 的转换逻辑。有了 DIA这些转换全部配置化管理。这让新的 Agent 接入成本变得极低——新的 Agent 只需要注册 schema 和翻译规则不用改任何现有调用方的代码。DIA 还有一个隐藏功能它会记录每一次调用的“样本”。包括入参、出参、耗时、错误码。这些数据平时看没什么等你想做 Agent 健康度评估或者性能分析时直接拉这份日志就能完成分析。2.4 为什么我没用 MCP 作为唯一的通信协议聊到 Agent 互联很多朋友第一反应是 MCPModel Context Protocol。说实话MCP 解决了很大的问题尤其是让模型能力接入变得更标准化。但我在实际项目里没有把 Agent-Reach 完全押在 MCP 上核心原因有三个一是 MCP 在当前语境下更侧重于“模型访问外部工具/资源”的能力协议而我这里的 Agent 互联还包括业务编排、状态同步、权限管控等场景纯 MCP 表达起来不够自然。二是历史系统不兼容。团队里老 Agent 的接口五花八门先做协议规整已经够费力了再让它们全部实现一遍 MCP推动成本太高。Agent-Reach 通过 DIA 这层反而能兼容新老接口。三是我想保持协议的轻量可控。Agent-Reach 的协议本质就是一个带路由信息的 JSON 信封理解成本极低任何语言都能实现。MCP 可以作为一个 Agent 内部能力暴露的方式而 Agent-Reach 管的是 Agent 之间的互联触达。两者可以共存并不互斥。3. 实操从零搭一套 Mini 版 Agent-Reach3.1 环境准备与最小依赖我这次搭的 Mini 版用到的东西很简单Python 3.10FastAPI做 Agent 的 HTTP 服务Redis做 Registry 存储Docker Compose起一个 Redis 实例顺手把两个测试 Agent 也容器化目录结构大概是这样agent-reach-mini/ ├── registry.py # 服务中心管理 Agent 注册与发现 ├── router.py # 路由转发协议翻译 ├── adapter.py # DIA动态适配 ├── agents/ │ ├── order_agent.py # 订单查询 Agent │ └── user_agent.py # 用户画像 Agent └── broker.py # 简单的“连接层”入口组合上面三块先用 Docker 起 Redisdocker run -d --name agent-reach-redis -p 6379:6379 redis:7-alpine3.2 定义 Agent 注册的 Schema我的 Agent 注册机制不搞复杂的格式就一个 JSON 结构。让我直接用代码说话# registry.py import json import time import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def register_agent(agent_info: dict, ttl: int 30): key fagent:{agent_info[agent_id]} agent_info[last_seen] time.time() r.hset(key, mappingagent_info) r.expire(key, ttl) return key def heartbeat(agent_id: str): r.hset(fagent:{agent_id}, last_seen, time.time()) r.expire(fagent:{agent_id}, 30) def discover(capability: str): found [] for key in r.scan_iter(agent:*): info r.hgetall(key) if capability in json.loads(info.get(capabilities, [])): info[agent_id] key.split(:)[-1] found.append(info) return found每个 Agent 启动时调用register_agent并起一个 timer 每隔 10 秒心跳一次。这样 Registry 的“存活表”基本是实时的。一个 Agent 的注册信息长这样agent_order { agent_id: order_agent_v1, name: 订单查询Agent, endpoint: http://order_agent:8001/process, protocol: agent-reach/json-v1, capabilities: json.dumps([order.query, order.refund]), input_schema: json.dumps({ order_id: {type: string}, user_id: {type: string} }), output_schema: json.dumps({ order_status: {type: string}, amount: {type: number} }) }这里有个细节值得说我把capabilities声明为字符串类型因为 Redis hash 里嵌套结构要么序列化要么拆字段。序列化省事但要注意读取时一致地做json.loads。项目一大建议封装成一个 Agent SDK 类别人就不用关心序列化细节了。3.3 Router 与 DIA 的最小实现Router 的转发逻辑核心是“抽捡目标”和“统一请求格式”。我这个版本做得比较细把超时和重试都写了# router.py import json import uuid import httpx from registry import discover async def route_request(request: dict): capability request[target_capability] candidates [a for a in discover(capability) if a.get(status) ! offline] if not candidates: return {ok: False, error: no_available_agent, capability: capability} last_error None for attempt in range(request.get(retry, 1) 1): target candidates[attempt % len(candidates)] try: async with httpx.AsyncClient(timeoutrequest.get(timeout_ms, 3000) / 1000) as client: # DIA 的翻译逻辑在这里调用详见 adapter.py payload translate_payload(target, request[payload]) resp await client.post(target[endpoint], json{ request_id: request.get(request_id, uuid.uuid4().hex), from: request.get(from), payload: payload }) resp.raise_for_status() result resp.json() return {ok: True, request_id: request[request_id], response: result} except Exception as e: last_error str(e) # 标记该 Agent 状态准备换一个重试 mark_agent_error(target[agent_id], last_error) return {ok: False, error: all_agents_failed, detail: last_error}注意一个优化点retry策略我默认是“换一个候选 Agent 重试”。如果一个能力只有一个 Agent 实现重试其实没有意义只会重复打一个已经超时的服务。所以我在候选列表剩一个时直接补齐retry1最多打两次避免无限等待。DIA 的翻译逻辑我把它做成了一个函数核心是维护一份字段映射表# adapter.py import json FIELD_MAP { order_agent_v1: { order_id: orderId, user_id: uid, start_date: dateStart, end_date: dateEnd } } def translate_payload(target: dict, payload: dict): mapping FIELD_MAP.get(target.get(agent_id), {}) new_payload {} for k, v in payload.items(): new_key mapping.get(k, k) new_payload[new_key] v return new_payload这版 DIA 是很简化的只做了字段改名。如果要处理“字符串日期改成时间戳”“数组拍平”“嵌套结构重组”建议直接引入 jq 风格的转换表达式或者干脆让每个 Agent 的 schema 里自带一个transform_script用一小段 JSONata 表达式描述转换规则。字段名映射是保底方案永远不过时。3.4 两个测试 Agent 与一个完整任务流我写了一个订单查询 Agent 和一个用户画像 Agent互相之间通过 Router 连接。订单 Agent 收到请求后不自己处理用户信息而是转调用用户 Agent# agents/order_agent.py from fastapi import FastAPI, Request import httpx app FastAPI() app.post(/process) async def process(req: Request): body await req.json() payload body[payload] if user_id in payload and not payload.get(user_profile_loaded): # 向 Router 请求用户画像能力 async with httpx.AsyncClient() as client: user_resp await client.post(http://broker:8000/route, json{ request_id: body[request_id], from: order_agent_v1, target_capability: user.profile.get, payload: {user_id: payload[user_id]}, timeout_ms: 2000, retry: 1 }) profile user_resp.json() payload[user_level] profile[response][user_level] return { order_status: shipped, amount: 299.0, user_level: payload.get(user_level, normal) }整个流程跑起来是这样的外部请求打到 broker 的/route入口要求查一个订单并附带用户会员等级。Router 发现order.query能力由订单 Agent 提供把请求转过去。订单 Agent 发现自己需要用户等级数据再向 Router 发起一次user.profile.get请求。Router 把这次请求转给用户 Agent用户 Agent 返回会员等级。订单 Agent 组装最终响应再通过 Router 返回给外部调用方。这个链路的价值在于订单 Agent 不需要知道用户 Agent 在哪个地址、端口多少它只需要声明“我要这个能力”。真正找到人、把消息送达、把结果带回来全都由 Agent-Reach 这层完成。我实测跑了一百次串行调用成功率稳定在 97% 以上剩下的 3% 失败大多是超时设置太短导致的重试超限后文会细说怎么调。4. 关键细节触达的稳定性、安全性与可观测性4.1 超时、重试与熔断的参数计算“触达”最怕的不是失败而是“失败的姿势很难看”。我起初遇到的典型问题是A 调 BB 卡住了 8 秒A 一直在等连带整条链路的请求都堆积起来。后来我把超时和重试的参数当成一门课来研究简单分享几个结论。超时时间不能拍脑袋设。我常用的估算方式是“链路总超时 每个 Agent 平均处理时间 × 单跳超时系数 网络开销”。假设目标 Agent 的 p95 处理时间是 800ms那我给单跳设置的超时就是 800 × 2 200 1800ms留足抖动余地又不至于让调用方等太久。重试次数要看“重试的成本”。Agent 接口如果没有副作用纯查询重试是安全的可以设 2 次。如果 Agent 接口会创建工单、发起支付、扣减库存重试必须极其谨慎最好配合幂等键。Agent-Reach 里我强制约定所有 Agent 接口必须接受request_id作为幂等键。执行型能力若发现相同request_id已经处理过直接返回上次的结果不再重复执行。熔断也是必须的。我自定义了一个简化版熔断连续失败 3 次Router 标记该 Agent 为degraded接下来 30 秒内不再往它转发请求只保留少量探测请求。探测成功了状态恢复。这个机制能防止“一个 Agent 出故障拖垮整条链路上的所有调用方”。参数我用的默认值调整依据单跳超时2s目标 Agent p95 处理时间的 2 倍重试次数1可配置为 2只对无副作用接口开放 2 次熔断阈值连续 3 次失败观察实际可用性调大调小均可熔断恢复窗口30s短于 5 秒容易抖动长于 60 秒恢复太慢心跳周期10s至少留 3 个周期窗口才判离线4.2 谁允许触达谁权限建模的实操Agent 互联时“权限”是一个特别容易被糊弄过去的问题。因为 Agent 之间的调用跟人没关系传统 OAuth 那套太重了。我采用的是一个简化版的“能力访问控制列表”。每个 Agent 注册时带一个acl字段声明“我可以被谁调用”agent_user { agent_id: user_agent_v1, capabilities: [user.profile.get], acl: [order_agent_v1, recommend_agent_v1] }Router 在转发请求前检查请求头里的from字段是否在目标 Agent 的acl列表里。不在就直接拒绝返回permission_denied。这个机制很轻量却天然阻止了很多低级事故。比如订单 Agent 被攻破或误配置它也无法随意去访问所有 Agent。再进阶一点可以在 ACL 里增加条件表达式比如“只允许在timeout_ms 3000时访问”但这种需求少不必一开始就做复杂。先解决“谁能调谁”再考虑“什么条件下能调”。4.3 踩过的坑有状态 Agent 与无状态 Agent做 Agent-Reach 之前我默认所有 Agent 都是无状态的后来遇到一个“会话记忆型 Agent”直接破功。这个 Agent 会记住当前对话上下文比如用户问“我刚才那个订单呢”它需要结合之前的会话状态才答得上来。问题在于 Router 每次转发请求都可能选到不同实例Agent 就无法拿到同一份上下文。排查了很久最后解决方案是在注册信息中显式声明stateful: true。Router 在调度这类 Agent 时按请求里的user_id或session_id做哈希固定分发到同一个实例。这里有个取舍——如果这个 Agent 只有一个实例那一切好说如果有多个实例需要引入共享状态存储比如 Redis或者会话亲和。另外一个印象很深的坑是“Agent 的限流静默失败”。有一个 Agent 服务自身带了限流流量超了直接拒绝返回 429。Router 第一次漏掉了对 429 的特殊处理把 429 当成普通请求失败并触发了熔断。结果这个 Agent 明明只是短暂限流却被熔断了 30 秒导致整体可用性更差。后来我在 Router 里加了判断429 和 503 不要触发熔断只做退避重试只有 500、超时这类才计入熔断阈值。可观测性方面我给的土办法是结构化日志。每次路由请求打一行 JSON 日志包含request_id、from、to、capability、status、cost_ms。出了问题grep request_id一下就能看到全链路。不需要上昂贵的调用链系统小项目先把日志打规范比什么监控都管用。5. 常见问题排查与避坑实录现象可能原因排查方法解决方案请求发出去Router 报 no_available_agentRegistry 里没有可用 Agent或 Agent 心跳过期检查 Agent 是否启动、心跳是否正常重启 Agent确认注册成功调用报 permission_denied调用方不在目标 Agent 的 ACL 列表查看目标 Agent 的注册信息在 ACL 中加入对应 agent_id有时成功有时失败候选 Agent 有多个部分实例异常查看 Router 日志里每跳的耗时实例全部恢复正常后重试429 后被熔断Router 把限流误判为故障看日志里是否连续 3 次 429调整熔断策略429 不触发熔断参数传过去字段对不上DIA 翻译映射未配置检查 Field Map 或 schema补充翻译规则链路过长整体超时单跳超时参数加总过大打印每跳耗时分析调整单跳超时缩短链路层数5.1 注册失败或心跳超时的排查注册失败最常见的坑是 Redis 连接串没配对或者 Agent 启动顺序问题——Agent 先启动Redis 还没起来注册时报错但 Agent 本身没退出导致状态看起来“活着”实际没注册上。我的经验是Agent 启动时注册动作要带重试逻辑连不上 Redis 就退避重试不要只试一次。心跳超时的坑则来自于 TTL 设置和心跳周期不匹配。比如 TTL 设 10 秒心跳也是 10 秒一次网络稍一抖动Redis 就把 Agent 标记为过期了。安全做法是 TTL 至少是心跳间隔的 3 倍。我上面的代码里心跳间隔 10 秒TTL 设 30 秒实测稳得很。5.2 消息重复传递的幂等处理跨 Agent 的调用网络超时后重试极容易产生重复消息。Agent A 发了一个“创建订单”请求给订单 Agent CRouter 等了 2 秒没响应就重试了一次但第一次请求其实已经到达 C只是返回响应时网络断了。结果就是同一个订单被创建了两次。解法分两层。第一层调用方在请求里带request_id第二层目标 Agent 在处理前先查 Redis# 幂等判断 if r.setnx(fidempotent:{request_id}, 1): # 第一次处理 result do_create_order(payload) r.setex(fresult:{request_id}, 3600, json.dumps(result)) return result else: cached r.get(fresult:{request_id}) if cached: return json.loads(cached)这个方法成本极低效果立竿见影。我在所有 Agent 的入口处都加了这个逻辑之后重复触发的问题基本清零。5.3 Agent 不可达网络分区与部署形态的墙前面提到有两种 Agent 是不可达状态。一种是网络分区A 和 B 明明在同一个系统里但部署在不同机房网络配置不对Router 转发时连不上。排查办法是先用curl手动打一下目标 endpoint确认基础连通性。还有一种是 Agent 跑在无固定地址的环境里比如某个任务型云函数运行时才拉起没有固定的服务地址。这时候 Router 不能直接 HTTP 转发需要改走消息队列或反向通道。我的 Mini 版里对第二种情况做了个变通Agent 启动时主动连上 Router 建立一个 WebSocket 长连接Router 收到请求后通过这个 WebSocket 把消息推给 AgentAgent 处理完再原路返回。这种“Agent 主动外联”模式在很多内网环境下是唯一可行的方案。实现起来也不复杂就是在 Router 里维护一个agent_id - WebSocket连接的映射表。最后分享两个让 Agent-Reach 更顺手的小技巧第一Registry 的本地缓存。不要让每个请求都打 Redis带一个小 TTL 的本地缓存就能把压力降一个量级。我这里用了 5 秒的缓存窗口Agent 注册状态变更后最多延迟 5 秒被感知完全够用。第二给每个 Agent 的注册信息里加一个version字段。Agent 接口升级时用新版本号重新注册Router 根据灰度比例把流量分配到 v1 和 v2相当于白拿了一个轻量灰度发布能力。我在实际把这套机制跑起来之后最大的体会是Agent 互联这件事难的不是某个组件的实现而是你要先想清楚边界。Agent 内部怎么想是它自己的事Agent 之间怎么通是你的事。把触达这一层做好了后面的调度、编排、权限才有着力点。目前这套 Mini 版在我的项目里稳定跑了三周下一步我计划把 DIA 的翻译规则改成可配置的 JSONata 文件让运营同事也能自己维护映射。如果你们也正在被多 Agent 互联折磨不妨从这个最小骨架开始先把通讯录建起来、把路由跑通后面的一切都好谈。
最近不少做多智能体项目的朋友都在聊 Agent-Reach 这个名字。其实最早看到它时我也没太当回事直到自己手底下的 Agent 多了、乱起来才明白这项目到底卡在哪个痛点上。简单说Agent-Reach 解决的是一个特别朴素又特别现实的问题当系统里有几十上百个 Agent它们可能来自不同的团队、跑在不一样的模型上、接口风格五花八门怎么让这些 Agent 互相之间稳定地触达、传消息、把活儿递到对的人或者说对的 Agent手里。单个 Agent 的能力再强如果接不上、呼不动、消息发出去没人认一切都白搭。我这次基于 Agent-Reach 的核心理念自己从零搭了一套最小可用的 Agent 互联调度机制跑通了一个跨 Agent 协作的完整任务流。这篇文章不聊虚的直接讲清楚 Agent-Reach 解决什么、架构怎么拆、代码怎么落以及我实际操作中踩过的坑和调参经验。想自己动手复现的可以直接抄作业遇到问题的也可以拿后面的排查表一个个对照。1. Agent-Reach 到底是什么它捅破的窗户纸在哪1.1 多 Agent 系统最常见的三个尴尬场景我先讲三个真实场景你看看自己中过几枪。第一个场景是“各自为政”。团队 A 做了一个订单查询 Agent团队 B 做了个用户画像 Agent两边都做得挺漂亮。结果产品经理说“帮我做个新功能用户问订单时顺带推荐点东西”。你一看这不就是让订单 Agent 调用户画像 Agent 嘛。但问题来了订单 Agent 是 Python 写的用户画像 Agent 是 Java 写的两边自定义了一套 HTTP 接口参数一个叫order_id一个叫orderId时间格式一个是2025-01-01 00:00:00一个是时间戳。光对齐这些就耗掉大半天。第二个场景是“呼叫无门”。Agent B 挂在某个内网环境里或者跑在某个云函数上没有固定的对外地址。Agent A 想调用它却根本不知道该怎么找到它。“我知道有这么一个 Agent但我联系不上它”——这比“没有这个 Agent”更让人崩溃。因为前者意味着功能明明存在却卡在最后一公里的触达问题上。第三个场景是“重复劳动”。六个 Agent 都需要调用同一个“天气查询”能力每个 Agent 都自己接一遍上游接口自己处理鉴权、超时、错误重试。六份代码逻辑一样写法各不相同哪天上游接口改了个字段六个 Agent 挨个出问题排查起来直接怀疑人生。这三个场景本质上就是同一个问题Agent 之间的互联机制缺失。Agent-Reach 这个名字取得很到位——reach触达。它的核心目标就是让任何 Agent 能以统一、可控、可观测的形式触达其他 Agent。1.2 从名字拆解它的设计哲学我理解 Agent-Reach 的设计哲学可以从三个关键词拆开看。第一个是 Agent。它面向的不是某个具体的 Agent 框架或语言而是“广义的 Agent”。不管你是 LangChain 的 Agent还是自研的状态机 Agent甚至只是一个“看起来像 Agent 的 HTTP 服务”在 Agent-Reach 的视角里都算是可接入的节点。这一点很关键它决定了这套机制的普适性。第二个是 Reach。英文里 reach 表示“到达、影响、联系”。在设计上它就对应着“可达性”和“触达能力”两层意思。可达性是指系统需要有一个机制让 Agent 能被其他人发现并连上触达能力是指消息发出去后对方能收到、能理解、能回应。很多 Agent 协作方案只解决了“能发消息”这一层却没有解决“对方能不能正确理解并执行”这就导致协作流程走走就断了。第三个是 Agent-Reach 整体的定位——它是一层“连接层”或者说“协作中间层”而不是一个 Agent 运行时。它不负责你 Agent 内部怎么思考、怎么规划它只管 Agent 与 Agent 之间的连接、翻译、调度、反馈。这个边界只要清晰了整套系统的复杂度就可控了。1.3 什么场景值得上 Agent-ReachAgent-Reach 不是银弹我并不建议所有项目一上来就搞它。如果你的项目只有两三个 Agent彼此直接写接口调用就行引入额外的中间层反而是负担。但当你的系统出现以下特征时就该认真考虑这层机制了Agent 数量超过 5 个两两之间的调用关系开始变得像蜘蛛网。Agent 由不同团队维护接口风格、数据结构、通信协议不统一。有 Agent 跑在无法直接暴露固定地址的环境里但其他 Agent 必须调用它。需要统一管控“谁能调用谁”的权限边界。需要给跨 Agent 的调用链路做日志、排查、审计。拿我自己这个项目来说最开始只有三个 Agent后来加到七八个每个 Agent 都要互相调用。代码层面撑了俩月实在扛不住了才决定动手做个类似 Agent-Reach 的中间层。做完之后最大的感受是当 Agent 之间不再互相 “直接调接口” 而是“通过连接层间接触达” 时系统的可维护性会有质的变化。好处是你改一个 Agent不需要通知另外五个同步改接口坏处是你要多维护一层基础设施配置和排障的复杂度转移到了这一层。2. 整体架构拆解Agent-Reach 的三大核心组件2.1 核心组件之一Registry——Agent 的通讯录一个多 Agent 系统里如果每个 Agent 都要记住其他 Agent 的地址、端口、接口路径、鉴权方式那配置文件的复杂度会爆炸。Registry 组件就是来解决这个问题的——它是全套系统的“通讯录”或者“服务中心”。每个 Agent 启动时会向 Registry 注册自己的“Agent 描述信息”。这条描述信息至少包含几个字段agent_id全局唯一的 Agent 标识name人类可读的 Agent 名称endpoint这个 Agent 对外接收消息的地址可以是一个 HTTP URL也可以是一个消息队列的主题名甚至是一个内部服务名protocol它听懂什么协议比如agent-reach/json-v1capabilities能力说明用一组固定的能力 ID 来描述比如order.query、user.profile.getschema_url这个 Agent 提供的接口定义文件用于参数校验和协议转换Registry 做了什么维护一张“存活表”。Agent 注册后要定期发送心跳超过阈值没心跳的会自动被标记为离线。其他 Agent 要调用某个能力时不是自己去找连接信息而是先问 Registry谁有order.query这个能力Registry 告诉它一个可用的 agent_id 和 endpoint调用方再去发起连接。我实际做的时候Registry 直接用 Redis 做存储给每个 Agent 一个 hash 结构字段就是上面那些超时时间设成 3 个心跳周期简单够用。但需要注意一个点Registry 如果自己有状态它就变成了单点。解决思路有两个一是让每个 Agent 本地缓存一份 Registry 的全量镜像定期拉取或订阅变更二是 Registry 本身做集群化部署。小项目用方案一性价比最高后续我会细说。2.2 核心组件之二Router——消息的传令兵Router 是 Agent-Reach 里最黏人的组件。它不干重活但所有跨 Agent 的请求都过它。设计 Router 时我纠结过一个问题为什么不走“直连”路线后来想明白了直连模式省事但你要在调用方写一堆容错、超时、重试逻辑。这些逻辑是通用的不该每个 Agent 各写一遍。Router 把这些通用逻辑收敛到了一处。Router 接收调用方 Agent 发来的“意图请求”格式类似{ request_id: req_001, from: agent_a, target_capability: order.query, payload: { order_id: 20250117001 }, timeout_ms: 3000, retry: 2 }Router 拿到这份请求后做四件事先查 Registry找到具备order.query能力的可用 Agent 列表。按调度策略挑一个 Agent。默认策略是“如果只有一个就选那个如果有多个轮询或者随机”。做协议转换。上游 Agent 用的是下划线字段名目标 Agent 用的是驼峰字段名Router 在这里把order_id翻译成orderId。把请求转发给目标 Agent等待响应。如果出现超时Router 会按retry配置在候选列表里换一个重试。Router 还承担一个容易被忽视的任务请求上下文的透传。一次完整的业务请求可能串联 Agent A → B → C中间路由了两次。如果每次转发都把request_id丢掉后面排查链路就断了。我在 Router 里强制要求任何转发请求必须带上原始request_id并且记录每一跳的耗时。2.3 核心组件之三DIA——屏蔽 Agent 方言的翻译层DIA 全名是 Dynamic Invocation Adapter动态调用适配器。名字是我自己取的换到什么项目里都适用。它的作用是解决开头说的“一个字段名不统一、时间格式不统一、接口风格不统一”的问题。我的实现很粗暴但有效每个 Agent 在注册时除了声明 endpoint还要声明 schema。schema 描述它期望的输入输出结构我用的是一份简化版 JSON Schema。DIA 维护了一个类似“翻译字典”的映射表把统一格式的请求翻译成每个 Agent 自己习惯的格式。举个例子。上层统一请求是{ user_id: u_123, start_date: 2025-01-01, end_date: 2025-01-31 }某个 Agent B 接口期望的是{ uid: 123, dateRange: 2025-01-01~2025-01-31 }正常直连的话调用方得写死针对 Agent B 的转换逻辑。有了 DIA这些转换全部配置化管理。这让新的 Agent 接入成本变得极低——新的 Agent 只需要注册 schema 和翻译规则不用改任何现有调用方的代码。DIA 还有一个隐藏功能它会记录每一次调用的“样本”。包括入参、出参、耗时、错误码。这些数据平时看没什么等你想做 Agent 健康度评估或者性能分析时直接拉这份日志就能完成分析。2.4 为什么我没用 MCP 作为唯一的通信协议聊到 Agent 互联很多朋友第一反应是 MCPModel Context Protocol。说实话MCP 解决了很大的问题尤其是让模型能力接入变得更标准化。但我在实际项目里没有把 Agent-Reach 完全押在 MCP 上核心原因有三个一是 MCP 在当前语境下更侧重于“模型访问外部工具/资源”的能力协议而我这里的 Agent 互联还包括业务编排、状态同步、权限管控等场景纯 MCP 表达起来不够自然。二是历史系统不兼容。团队里老 Agent 的接口五花八门先做协议规整已经够费力了再让它们全部实现一遍 MCP推动成本太高。Agent-Reach 通过 DIA 这层反而能兼容新老接口。三是我想保持协议的轻量可控。Agent-Reach 的协议本质就是一个带路由信息的 JSON 信封理解成本极低任何语言都能实现。MCP 可以作为一个 Agent 内部能力暴露的方式而 Agent-Reach 管的是 Agent 之间的互联触达。两者可以共存并不互斥。3. 实操从零搭一套 Mini 版 Agent-Reach3.1 环境准备与最小依赖我这次搭的 Mini 版用到的东西很简单Python 3.10FastAPI做 Agent 的 HTTP 服务Redis做 Registry 存储Docker Compose起一个 Redis 实例顺手把两个测试 Agent 也容器化目录结构大概是这样agent-reach-mini/ ├── registry.py # 服务中心管理 Agent 注册与发现 ├── router.py # 路由转发协议翻译 ├── adapter.py # DIA动态适配 ├── agents/ │ ├── order_agent.py # 订单查询 Agent │ └── user_agent.py # 用户画像 Agent └── broker.py # 简单的“连接层”入口组合上面三块先用 Docker 起 Redisdocker run -d --name agent-reach-redis -p 6379:6379 redis:7-alpine3.2 定义 Agent 注册的 Schema我的 Agent 注册机制不搞复杂的格式就一个 JSON 结构。让我直接用代码说话# registry.py import json import time import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def register_agent(agent_info: dict, ttl: int 30): key fagent:{agent_info[agent_id]} agent_info[last_seen] time.time() r.hset(key, mappingagent_info) r.expire(key, ttl) return key def heartbeat(agent_id: str): r.hset(fagent:{agent_id}, last_seen, time.time()) r.expire(fagent:{agent_id}, 30) def discover(capability: str): found [] for key in r.scan_iter(agent:*): info r.hgetall(key) if capability in json.loads(info.get(capabilities, [])): info[agent_id] key.split(:)[-1] found.append(info) return found每个 Agent 启动时调用register_agent并起一个 timer 每隔 10 秒心跳一次。这样 Registry 的“存活表”基本是实时的。一个 Agent 的注册信息长这样agent_order { agent_id: order_agent_v1, name: 订单查询Agent, endpoint: http://order_agent:8001/process, protocol: agent-reach/json-v1, capabilities: json.dumps([order.query, order.refund]), input_schema: json.dumps({ order_id: {type: string}, user_id: {type: string} }), output_schema: json.dumps({ order_status: {type: string}, amount: {type: number} }) }这里有个细节值得说我把capabilities声明为字符串类型因为 Redis hash 里嵌套结构要么序列化要么拆字段。序列化省事但要注意读取时一致地做json.loads。项目一大建议封装成一个 Agent SDK 类别人就不用关心序列化细节了。3.3 Router 与 DIA 的最小实现Router 的转发逻辑核心是“抽捡目标”和“统一请求格式”。我这个版本做得比较细把超时和重试都写了# router.py import json import uuid import httpx from registry import discover async def route_request(request: dict): capability request[target_capability] candidates [a for a in discover(capability) if a.get(status) ! offline] if not candidates: return {ok: False, error: no_available_agent, capability: capability} last_error None for attempt in range(request.get(retry, 1) 1): target candidates[attempt % len(candidates)] try: async with httpx.AsyncClient(timeoutrequest.get(timeout_ms, 3000) / 1000) as client: # DIA 的翻译逻辑在这里调用详见 adapter.py payload translate_payload(target, request[payload]) resp await client.post(target[endpoint], json{ request_id: request.get(request_id, uuid.uuid4().hex), from: request.get(from), payload: payload }) resp.raise_for_status() result resp.json() return {ok: True, request_id: request[request_id], response: result} except Exception as e: last_error str(e) # 标记该 Agent 状态准备换一个重试 mark_agent_error(target[agent_id], last_error) return {ok: False, error: all_agents_failed, detail: last_error}注意一个优化点retry策略我默认是“换一个候选 Agent 重试”。如果一个能力只有一个 Agent 实现重试其实没有意义只会重复打一个已经超时的服务。所以我在候选列表剩一个时直接补齐retry1最多打两次避免无限等待。DIA 的翻译逻辑我把它做成了一个函数核心是维护一份字段映射表# adapter.py import json FIELD_MAP { order_agent_v1: { order_id: orderId, user_id: uid, start_date: dateStart, end_date: dateEnd } } def translate_payload(target: dict, payload: dict): mapping FIELD_MAP.get(target.get(agent_id), {}) new_payload {} for k, v in payload.items(): new_key mapping.get(k, k) new_payload[new_key] v return new_payload这版 DIA 是很简化的只做了字段改名。如果要处理“字符串日期改成时间戳”“数组拍平”“嵌套结构重组”建议直接引入 jq 风格的转换表达式或者干脆让每个 Agent 的 schema 里自带一个transform_script用一小段 JSONata 表达式描述转换规则。字段名映射是保底方案永远不过时。3.4 两个测试 Agent 与一个完整任务流我写了一个订单查询 Agent 和一个用户画像 Agent互相之间通过 Router 连接。订单 Agent 收到请求后不自己处理用户信息而是转调用用户 Agent# agents/order_agent.py from fastapi import FastAPI, Request import httpx app FastAPI() app.post(/process) async def process(req: Request): body await req.json() payload body[payload] if user_id in payload and not payload.get(user_profile_loaded): # 向 Router 请求用户画像能力 async with httpx.AsyncClient() as client: user_resp await client.post(http://broker:8000/route, json{ request_id: body[request_id], from: order_agent_v1, target_capability: user.profile.get, payload: {user_id: payload[user_id]}, timeout_ms: 2000, retry: 1 }) profile user_resp.json() payload[user_level] profile[response][user_level] return { order_status: shipped, amount: 299.0, user_level: payload.get(user_level, normal) }整个流程跑起来是这样的外部请求打到 broker 的/route入口要求查一个订单并附带用户会员等级。Router 发现order.query能力由订单 Agent 提供把请求转过去。订单 Agent 发现自己需要用户等级数据再向 Router 发起一次user.profile.get请求。Router 把这次请求转给用户 Agent用户 Agent 返回会员等级。订单 Agent 组装最终响应再通过 Router 返回给外部调用方。这个链路的价值在于订单 Agent 不需要知道用户 Agent 在哪个地址、端口多少它只需要声明“我要这个能力”。真正找到人、把消息送达、把结果带回来全都由 Agent-Reach 这层完成。我实测跑了一百次串行调用成功率稳定在 97% 以上剩下的 3% 失败大多是超时设置太短导致的重试超限后文会细说怎么调。4. 关键细节触达的稳定性、安全性与可观测性4.1 超时、重试与熔断的参数计算“触达”最怕的不是失败而是“失败的姿势很难看”。我起初遇到的典型问题是A 调 BB 卡住了 8 秒A 一直在等连带整条链路的请求都堆积起来。后来我把超时和重试的参数当成一门课来研究简单分享几个结论。超时时间不能拍脑袋设。我常用的估算方式是“链路总超时 每个 Agent 平均处理时间 × 单跳超时系数 网络开销”。假设目标 Agent 的 p95 处理时间是 800ms那我给单跳设置的超时就是 800 × 2 200 1800ms留足抖动余地又不至于让调用方等太久。重试次数要看“重试的成本”。Agent 接口如果没有副作用纯查询重试是安全的可以设 2 次。如果 Agent 接口会创建工单、发起支付、扣减库存重试必须极其谨慎最好配合幂等键。Agent-Reach 里我强制约定所有 Agent 接口必须接受request_id作为幂等键。执行型能力若发现相同request_id已经处理过直接返回上次的结果不再重复执行。熔断也是必须的。我自定义了一个简化版熔断连续失败 3 次Router 标记该 Agent 为degraded接下来 30 秒内不再往它转发请求只保留少量探测请求。探测成功了状态恢复。这个机制能防止“一个 Agent 出故障拖垮整条链路上的所有调用方”。参数我用的默认值调整依据单跳超时2s目标 Agent p95 处理时间的 2 倍重试次数1可配置为 2只对无副作用接口开放 2 次熔断阈值连续 3 次失败观察实际可用性调大调小均可熔断恢复窗口30s短于 5 秒容易抖动长于 60 秒恢复太慢心跳周期10s至少留 3 个周期窗口才判离线4.2 谁允许触达谁权限建模的实操Agent 互联时“权限”是一个特别容易被糊弄过去的问题。因为 Agent 之间的调用跟人没关系传统 OAuth 那套太重了。我采用的是一个简化版的“能力访问控制列表”。每个 Agent 注册时带一个acl字段声明“我可以被谁调用”agent_user { agent_id: user_agent_v1, capabilities: [user.profile.get], acl: [order_agent_v1, recommend_agent_v1] }Router 在转发请求前检查请求头里的from字段是否在目标 Agent 的acl列表里。不在就直接拒绝返回permission_denied。这个机制很轻量却天然阻止了很多低级事故。比如订单 Agent 被攻破或误配置它也无法随意去访问所有 Agent。再进阶一点可以在 ACL 里增加条件表达式比如“只允许在timeout_ms 3000时访问”但这种需求少不必一开始就做复杂。先解决“谁能调谁”再考虑“什么条件下能调”。4.3 踩过的坑有状态 Agent 与无状态 Agent做 Agent-Reach 之前我默认所有 Agent 都是无状态的后来遇到一个“会话记忆型 Agent”直接破功。这个 Agent 会记住当前对话上下文比如用户问“我刚才那个订单呢”它需要结合之前的会话状态才答得上来。问题在于 Router 每次转发请求都可能选到不同实例Agent 就无法拿到同一份上下文。排查了很久最后解决方案是在注册信息中显式声明stateful: true。Router 在调度这类 Agent 时按请求里的user_id或session_id做哈希固定分发到同一个实例。这里有个取舍——如果这个 Agent 只有一个实例那一切好说如果有多个实例需要引入共享状态存储比如 Redis或者会话亲和。另外一个印象很深的坑是“Agent 的限流静默失败”。有一个 Agent 服务自身带了限流流量超了直接拒绝返回 429。Router 第一次漏掉了对 429 的特殊处理把 429 当成普通请求失败并触发了熔断。结果这个 Agent 明明只是短暂限流却被熔断了 30 秒导致整体可用性更差。后来我在 Router 里加了判断429 和 503 不要触发熔断只做退避重试只有 500、超时这类才计入熔断阈值。可观测性方面我给的土办法是结构化日志。每次路由请求打一行 JSON 日志包含request_id、from、to、capability、status、cost_ms。出了问题grep request_id一下就能看到全链路。不需要上昂贵的调用链系统小项目先把日志打规范比什么监控都管用。5. 常见问题排查与避坑实录现象可能原因排查方法解决方案请求发出去Router 报 no_available_agentRegistry 里没有可用 Agent或 Agent 心跳过期检查 Agent 是否启动、心跳是否正常重启 Agent确认注册成功调用报 permission_denied调用方不在目标 Agent 的 ACL 列表查看目标 Agent 的注册信息在 ACL 中加入对应 agent_id有时成功有时失败候选 Agent 有多个部分实例异常查看 Router 日志里每跳的耗时实例全部恢复正常后重试429 后被熔断Router 把限流误判为故障看日志里是否连续 3 次 429调整熔断策略429 不触发熔断参数传过去字段对不上DIA 翻译映射未配置检查 Field Map 或 schema补充翻译规则链路过长整体超时单跳超时参数加总过大打印每跳耗时分析调整单跳超时缩短链路层数5.1 注册失败或心跳超时的排查注册失败最常见的坑是 Redis 连接串没配对或者 Agent 启动顺序问题——Agent 先启动Redis 还没起来注册时报错但 Agent 本身没退出导致状态看起来“活着”实际没注册上。我的经验是Agent 启动时注册动作要带重试逻辑连不上 Redis 就退避重试不要只试一次。心跳超时的坑则来自于 TTL 设置和心跳周期不匹配。比如 TTL 设 10 秒心跳也是 10 秒一次网络稍一抖动Redis 就把 Agent 标记为过期了。安全做法是 TTL 至少是心跳间隔的 3 倍。我上面的代码里心跳间隔 10 秒TTL 设 30 秒实测稳得很。5.2 消息重复传递的幂等处理跨 Agent 的调用网络超时后重试极容易产生重复消息。Agent A 发了一个“创建订单”请求给订单 Agent CRouter 等了 2 秒没响应就重试了一次但第一次请求其实已经到达 C只是返回响应时网络断了。结果就是同一个订单被创建了两次。解法分两层。第一层调用方在请求里带request_id第二层目标 Agent 在处理前先查 Redis# 幂等判断 if r.setnx(fidempotent:{request_id}, 1): # 第一次处理 result do_create_order(payload) r.setex(fresult:{request_id}, 3600, json.dumps(result)) return result else: cached r.get(fresult:{request_id}) if cached: return json.loads(cached)这个方法成本极低效果立竿见影。我在所有 Agent 的入口处都加了这个逻辑之后重复触发的问题基本清零。5.3 Agent 不可达网络分区与部署形态的墙前面提到有两种 Agent 是不可达状态。一种是网络分区A 和 B 明明在同一个系统里但部署在不同机房网络配置不对Router 转发时连不上。排查办法是先用curl手动打一下目标 endpoint确认基础连通性。还有一种是 Agent 跑在无固定地址的环境里比如某个任务型云函数运行时才拉起没有固定的服务地址。这时候 Router 不能直接 HTTP 转发需要改走消息队列或反向通道。我的 Mini 版里对第二种情况做了个变通Agent 启动时主动连上 Router 建立一个 WebSocket 长连接Router 收到请求后通过这个 WebSocket 把消息推给 AgentAgent 处理完再原路返回。这种“Agent 主动外联”模式在很多内网环境下是唯一可行的方案。实现起来也不复杂就是在 Router 里维护一个agent_id - WebSocket连接的映射表。最后分享两个让 Agent-Reach 更顺手的小技巧第一Registry 的本地缓存。不要让每个请求都打 Redis带一个小 TTL 的本地缓存就能把压力降一个量级。我这里用了 5 秒的缓存窗口Agent 注册状态变更后最多延迟 5 秒被感知完全够用。第二给每个 Agent 的注册信息里加一个version字段。Agent 接口升级时用新版本号重新注册Router 根据灰度比例把流量分配到 v1 和 v2相当于白拿了一个轻量灰度发布能力。我在实际把这套机制跑起来之后最大的体会是Agent 互联这件事难的不是某个组件的实现而是你要先想清楚边界。Agent 内部怎么想是它自己的事Agent 之间怎么通是你的事。把触达这一层做好了后面的调度、编排、权限才有着力点。目前这套 Mini 版在我的项目里稳定跑了三周下一步我计划把 DIA 的翻译规则改成可配置的 JSONata 文件让运营同事也能自己维护映射。如果你们也正在被多 Agent 互联折磨不妨从这个最小骨架开始先把通讯录建起来、把路由跑通后面的一切都好谈。
