最近几个月我一直在折腾一个项目代号Agent-Reach。起因特别简单——我手上有几个由不同框架搭出来的 AI Agent各自都能跑但彼此根本不“认识”Agent A 需要查资料时不知道 Agent B 能帮忙Agent B 处理完数据后想通知 Agent C也只能靠我自己写胶水代码。这种状态做 Demo 还行一旦进入真实业务你会发现大量时间都浪费在“让 Agent 之间怎么找到对方”这件事上。Agent-Reach 就是为了把这些孤岛连起来而生的一套轻量级 Agent 互联互通中间件通过统一的注册与发现、消息路由、能力适配让不同框架的 Agent 可以互相看见、互相调用。这篇文章会把完整的架构、核心代码、落坑过程都整理出来适合正在做多 Agent 系统、或者准备把 Agent 从单机流程升级成协作网络的开发者参考。我不敢说它是终极方案但它确实解决了我实际项目里最痛的那几个问题。更重要的是这套思路不绑定任何特定框架也不要求你把自己的 Agent 重写一遍——它做的是“连接”和“翻译”的活。1. 项目背景与核心痛点拆解1.1 AI Agent 生态的碎片化现状这两年 Agent 框架层出不穷LangChain、CrewAI、AutoGen、Dify、Coze甚至自己手写的 State Machine Agent每个框架都有自己的消息格式、工具调用约定、运行生命周期。框架本身没有对错但落到一个真实项目里问题就来了你不可能用一个框架解决所有问题。比如我在一个项目里同时用到了 LangChain 做 RAG 问答用 AutoGen 做多角色讨论还自己写了几个轻量 Agent 处理定时任务。每个 Agent 单独跑都很稳定可一旦需要它们协作比如“从知识库检索资料 - 让写作 Agent 生成报告 - 再让翻译 Agent 转成英文版”整个链路就得靠我手动传递数据。最开始我用的是脚本硬编码把上一个 Agent 的输出保存成 JSON 文件下一个 Agent 读文件。麻烦不说Agent 一多依赖关系很快就乱成一团。这个痛点在业内其实很普遍。很多团队做了十几个 Agent但实际是“十三个独立的小系统”根本没有形成“网络效应”。单个 Agent 的能力再强也只能在它自己的数据孤岛里打转价值大打折扣。1.2 Agent-Reach 究竟要解决什么问题我定义这个项目时给自己提了三个问题第一个发现Agent A 怎么知道系统中存在 Agent BAgent B 又能做什么事这需要一套“能力注册与发现机制”类似服务注册中心但注册的不是 IP 和端口而是 Agent 的“能力描述”。第二个沟通Agent A 和 Agent B 用不用“同一种语言”现实情况是它们大概率不用同一种语言。所以 Agent-Reach 需要做“协议转换”让各方用自己的母语说话由 Reach 负责翻译和路由。第三个信任Agent A 凭什么让你把任务转给 Agent B这就需要身份标识、权限控制、调用审计。连接越方便安全问题就越不能忽略否则 Agent 之间乱调用出事的风险比单机时代大得多。Agent-Reach 本质上就是在解决这三个问题它的定位不是另一个 Agent 框架而是一个Agent 互联互通网关层处于 Agent 和 Agent 之间负责撮合、转发、翻译、审计。1.3 常见方案的对比与选型思路在动手之前我调研过当时已有的方案。一个是 MCPModel Context Protocol它的核心是解决“应用怎么调用外部工具”的问题从 Model 到 Tool 方向是一套很标准的工具调用协议。但 MCP 本身不解决“Agent 之间互相发现和动态路由”的问题它更多是单向上下的调用关系。另一个是新出的 A2AAgent2Agent协议方向完全正确就是冲着 Agent 互联去的。但它规范还在快速迭代阶段协议细节和 SDK 都不算稳定直接用于生产环境有点冒险。我还考虑过直接用消息队列做中转比如用 Redis Stream 或 RabbitMQ。消息队列能解决通信问题但“能力发现”和“动态路由”依然得自己实现还得额外处理协议层的东西。综合评估下来我的选择是自研一个轻量协议核心借鉴 JSON-RPC 2.0同时兼容 MCP 的工具调用格式。这样既能满足 Agent 互联的需求又能蹭上 MCP 生态里已有的工具调用习惯老 Agent 接入成本很低。2. 整体架构设计与关键决策2.1 第一版架构Registry Router AdapterAgent-Reach 的架构用一个公式就能说清楚Agent-Reach Registry Router Adapter。Registry 是注册中心负责记录当前系统里所有 Agent 的身份、能力、地址、状态。Router 是路由层根据调用方给出的“能力需求”或者“明确的 Agent 名字”找到最合适的接收方然后把请求转发过去。Adapter 是适配器负责把不同 Agent 的接口形态包装成统一的协议格式。这三个部分我设计成了两个进程一个 Agent-Reach Server中心服务负责注册、路由、认证、审计还有一个 agent-reach-client SDK嵌入到各个 Agent 进程内负责注册上报、接收请求、返回响应。这样设计的好处是中心服务只做“转发”和“管理”不直接参与 Agent 的业务逻辑性能瓶颈不会出现在最前端。第一版我没搞微服务拆分的花活。所有模块放在同一个服务里数据用 Redis 存因为注册信息有天然的时效性——Agent 会上下线心跳过期就要摘除。Redis 的 TTL 机制正好能解决这个问题。2.2 协议层选型为什么选择 JSON-RPC 与 MCP 兼容层Agent 之间通信需要协议我花了挺长时间纠结这个。直接用 REST 行不行行但难受。REST 的语义偏向“资源操作”而 Agent 之间的调用本质是“请求执行一个任务”任务可能有返回结果也可能是异步执行、稍后回调。用 REST 来表达这层语义需要额外约定很多东西任务状态查询、结果回传、幂等控制、超时取消。这些做多了REST 就会变成“披着 REST 外衣的 RPC”。所以第一版定了 JSON-RPC 2.0。它有request、response、notification三种消息形态天然支持请求-响应模式和异步通知模式有id字段做消息关联有error字段统一错误结构轻量且足够用。同时我增加了一层 MCP compatibility如果一个 Agent 的某个能力本质上就是调用一个 MCP 工具那注册的时候可以直接声明type: mcp并附上 MCP 工具声明。Agent-Reach 收到转发请求后再按 MCP 协议去调远端工具这样等于把 MCP 生态的工具也接入了 Agent 协作网络。2.3 通信链路与消息模型同步调用 vs 异步任务设计消息模型时我特意区分了两种模式同步调用和异步任务。同步调用适合“马上要有结果”的短任务比如查天气、算数学、检索资料。调用方发出请求后阻塞等待响应超时时间一般在 10 到 30 秒。异步任务适合“执行时间很长”的流程比如生成一份完整报告、批量处理一批数据。调用方发出请求后立刻拿到一个task_id之后可以通过轮询或者回调方式拿结果。Agent-Reach 在这里做了一个很关键的设计把异步任务的执行状态统一管理起来不依赖某个 Agent 自己的存储这样即使接收方 Agent 重启了任务状态也可以恢复。消息模型的统一非常关键。不管是同步还是异步消息头都包含以下几个字段trace_id全链路追踪、from_agent、to_agent或target_capability、expire_at。这套消息头在后面排查问题的时候帮了我大忙——哪个环节慢、哪个任务断了trace_id 一查就清楚。3. 核心模块实现与实操细节3.1 服务注册与发现模块Agent-Reach 的注册模块核心数据结构长这样。一个 Agent 启动时会把以下信息上报到服务端agent_id全局唯一的 Agent 标识我用UUID4生成。agent_name人类可读的名称比如customer-service-agent。endpoint这个 Agent 接收回调消息的地址可以是 HTTP 地址也可以是 WebSocket甚至是本地函数回调。capabilities能力列表每一项都包括能力名、描述、输入输出 JSON Schema。ttl心跳时长默认 30 秒。服务端拿到注册信息后会写入 Rediskey 是agent:{agent_id}value 是 JSON 序列化的 Agent 元信息。同时给每个 Agent 维护一个agent:{agent_id}:capabilities的 Set存放它提供的能力名列表方便按能力检索。关键就在“心跳”这里。Redis 的EXPIRE机制天然适合做健康检查Agent 每次心跳就将 TTL 重置如果服务端在 TTL 时间内没收到心跳这个 Agent 的 key 就会自动消失路由时就不会再把它作为候选对象。第一版我用redis.setex实现很简单稳定。3.2 能力描述与技能路由能力描述是整个路由的基石。我参考了 Function Calling 里流行的 JSON Schema 方式每个能力描述包含四部分能力名、一句话说明、输入参数 Schema、输出结果 Schema。例如一个“知识库检索”能力{ name: knowledge_search, description: 在企业知识库中检索与关键词相关的文档片段, input_schema: { type: object, properties: { query: { type: string, description: 检索关键词或自然语言问题 }, top_k: { type: integer, description: 返回的文档片段数量默认 5, default: 5 } }, required: [query] }, output_schema: { type: array, items: { type: object, properties: { content: { type: string }, score: { type: number }, source: { type: string } } } } }路由模块的核心逻辑是先把调用方的请求意图映射到一个能力名上。这个过程我做了两层第一层是精确匹配如果调用方明确写了target_capabilityknowledge_search就直接找到对应 Agent第二层是语义匹配如果调用方只写了一段自然语言描述比如“帮我查一下公司今年的营收数据”系统会先用嵌入模型把描述和所有 Agent 的 capability description 做向量相似度比对返回候选 Agent 列表再由调用方选择一个最合适的。这个语义匹配是我做得比较值的一项功能虽然它增加了一点复杂度但它让 Agent 之间的协作从“我自己知道找谁”升级成了“我只需要说我要什么”。3.3 身份认证与权限控制Agent 之间调用必须有边界。我不能让一个只应该查天气的 Agent 去调一个能删数据库的 Agent这太危险了。第一版的认证方式采用 API Key 机制每个 Agent 注册成功后会拿到一个client_id和一个client_secret。后续所有 RPC 请求都要在 Header 里带上这两个字段由服务端做验证。为了防止密钥在网络传输中被截获所有通信我都要求走 TLS。权限控制我用了 ACL 白名单。每个 Agent 的注册信息里可以声明一个allowed_caller_ids字段只有白名单里的 Agent 能调用它的能力。如果这个字段为空表示默认拒绝所有调用需要显式配置allow_all: true才会开放给所有 Agent。这套机制在第一版已经够用。生产环境如果有更强的安全需求可以后续扩展 mTLS 双向认证或者基于 OAuth 的授权流程我留了接口。3.4 Agent 适配器与消息转换这是整个项目里最“接地气”的部分。不同 Agent 的接入方式千奇百怪有的是 HTTP 接口有的走 WebSocket有的是本地 Python 函数还有的是 LangGraph 里的节点。Adapter 的作用就是把这些差异屏蔽掉。我定义了一个统一的 AgentBackend 抽象每个后端只需要实现两个方法handle(request)和health_check()。class AgentBackend(ABC): abstractmethod async def handle(self, request: dict) - dict: 处理一次 RPC 请求返回统一响应格式 raise NotImplementedError abstractmethod async def health_check(self) - bool: 健康检查由注册中心周期性调用 raise NotImplementedErrorHTTP 后端就负责把请求体解析成内部字典、调用远端接口、再解析响应WebSocket 后端就维护一个连接池通过连接发送和接收消息本地方函数后端最简单直接通过函数名和参数调用。实际做的时候我发现适配器最大的坑不在“怎么调”而在“错误怎么映射”。HTTP 接口返回 500到底对应 RPC 的哪种错误超时了是返回错误还是重试这些细节得在 Adapter 层统一处理好否则每个 Agent 的错误五花八门调用方根本没法做异常处理。4. 环境搭建与关键代码实现4.1 技术栈与安装Agent-Reach 服务端用了 Python 3.11 FastAPI RedisSDK 也是 Python 包。选 Python 的原因很直接现有 Agent 生态大部分是 Python接入门槛低改造成本小。FastAPI 自带接口文档调试和对接都非常舒服。依赖安装就三行命令pip install fastapi uvicorn redis httpx pydanticRedis 我直接用了本机 Docker 跑避免污染系统环境docker run -d --name agent-reach-redis -p 6379:6379 redis:7-alpine4.2 注册中心的数据结构与核心代码注册中心是整个项目的心脏。我先把 Agent 的注册模型写好from pydantic import BaseModel, Field from typing import List, Optional class CapabilitySchema(BaseModel): name: str description: str input_schema: dict output_schema: Optional[dict] None class AgentRegistration(BaseModel): agent_id: str agent_name: str endpoint: str endpoint_type: str Field(defaulthttp, descriptionhttp/websocket/local) capabilities: List[CapabilitySchema] allowed_caller_ids: List[str] [] allow_all: bool False status: str online注册接口的路由处理函数长这样from fastapi import FastAPI, HTTPException from redis import asyncio as aioredis import json app FastAPI() redis aioredis.from_url(redis://localhost:6379/0) app.post(/v1/agents/register) async def register_agent(reg: AgentRegistration): key fagent:{reg.agent_id} reg.status online # 注册信息写入 RedisTTL 30 秒 await redis.setex(key, 30, reg.model_dump_json()) # 为每个能力建立倒排索引 for cap in reg.capabilities: await redis.sadd(fcapability:{cap.name}, reg.agent_id) # 记录 Agent 与能力名集合的映射 cap_names [cap.name for cap in reg.capabilities] await redis.delete(fagent:{reg.agent_id}:caps) if cap_names: await redis.sadd(fagent:{reg.agent_id}:caps, *cap_names) return {status: ok, agent_id: reg.agent_id}心跳接口更简单就是续期app.post(/v1/agents/heartbeat) async def heartbeat(agent_id: str): key fagent:{agent_id} if not await redis.exists(key): raise HTTPException(status_code404, detailAgent not registered) await redis.expire(key, 30) return {status: ok}注意这里的setex和expire它们是整个注册发现机制能稳定工作的基石。每个 Agent 必须每 15 秒左右打一次心跳留足余量避免网络抖动就掉线。4.3 消息路由的代码实现路由模块的职责是收到 RPC 请求后决定把请求转给哪个 Agent。这里给出最核心的路由匹配逻辑。app.post(/v1/rpc/call) async def rpc_call(req: RpcRequest): # 1. 先查调用方是否注册 if not await redis.exists(fagent:{req.from_agent}): raise HTTPException(status_code401, detailCaller not registered) # 2. 确定目标 Agent if req.to_agent: target_agent req.to_agent if not await redis.exists(fagent:{target_agent}): raise HTTPException(status_code404, detailTarget agent not found) elif req.target_capability: agent_ids await redis.smembers(fcapability:{req.target_capability}) if not agent_ids: raise HTTPException(status_code404, detailNo agent provides this capability) target_agent await route_to_most_suitable(agent_ids, req) else: raise HTTPException(status_code400, detailNo routing target specified) # 3. 检查权限 target_info json.loads(await redis.get(fagent:{target_agent})) if not target_info[allow_all] and req.from_agent not in target_info[allowed_caller_ids]: raise HTTPException(status_code403, detailPermission denied) # 4. 转发请求伪代码实际走 AgentBackend response await dispatch_request(target_agent, req) return responseroute_to_most_suitable在这里是最有意思的函数。当多个 Agent 提供同一个能力时我用了“加载评分”策略每个 Agent 上报心跳时顺带上一个当前 pending 任务数路由时选择 pending 数最小的那个。这个策略实现简单效果却很好相当于一个轻量负载均衡。后来我加了基于响应时间的 EWMA 指数加权移动平均选路就更有谱了。4.4 对接演示让两个异构 Agent 互相调用理论讲再多不如看一次实际对接。我拿“翻译 Agent”和“工单 Agent”做例子翻译 Agent 用 FastAPI 自研工单 Agent 用 LangGraph 里的节点封装成 HTTP 服务。第一步翻译 Agent 启动后注册到 Reachfrom agent_reach import AgentReachClient client AgentReachClient( server_urlhttp://localhost:8000, agent_idtranslator-agent, agent_name翻译助手 ) client.register( endpointhttp://translator:9001/translate, capabilities[ { name: translate_text, description: 将文本翻译成指定目标语言, input_schema: { type: object, properties: { text: {type: string}, target_lang: {type: string} }, required: [text, target_lang] } } ] )第二步工单 Agent 在处理用户请求时需要把一段回复翻译成英文。它自己并不“认识”翻译 Agent只知道“需要一个翻译能力”于是向 Reach 发起调用response client.call_capability( capabilitytranslate_text, params{ text: 您的工单已经处理完成感谢您的耐心等待。, target_lang: en }, timeout10 )整个过程工单 Agent 没有直接拼接翻译 Agent 的 URL也没关心翻译 Agent 是用什么框架写的。它只需要告诉 Reach“我要什么”Reach 去查注册表、过权限、路由、转发然后把结果拿回来。这就是 Agent-Reach 的核心价值调用方与实现方解耦。这个演示跑通后我确认了整套设计是成立的。后面再接入新 Agent流程就完全机械化了写一个注册文件 - 写一段业务代码 - 启动后自动加入协作网络。5. 生产环境避坑与问题排查实录5.1 常见问题速查表我整理了一张问题排查表都是我在迭代过程中真遇到过的直接列出便于对照解决。现象可能原因解决方式路由时报“Agent not found”心跳中断Redis key 过期检查 Agent 心跳任务是否正常运行看 RedisTTL剩余时间调用方刚注册就被拒绝ACL 白名单未配置检查目标 Agent 的allowed_caller_ids确认调用方在名单内请求超时但目标 Agent 明明处理完了响应体过大或链路有缓冲导致延迟用trace_id排查每一跳耗时必要时增大内部接口超时Agent 频繁上下线心跳间隔与 TTL 设置不合理心跳间隔设置为 TTL 的 1/2避免网络抖动导致误判语义路由选错 Agent能力描述写得太模糊重写 capability description用具体业务词汇避免通用大词异步任务丢失Agent 重启后任务状态丢失中心维护任务状态表Agent 重连后自动拉取未完成任务5.2 三个让人印象深刻的坑第一个坑是 Redis 大 key 问题。第一个月跑着没事后来某个 Agent 挂了十几个小时没上报它的 capability 集合里存了太多历史 Agent ID。每次路由要遍历整个集合做过滤Redis 响应慢到几十毫秒整个路由链路被拖垮。解决方法是给集合键加上业务前缀并且定期清理离线 Agent 的残留索引。这个坑让我明白了注册索引和业务数据一样不做生命周期管理就一定会出问题。第二个坑是语义路由的“幻觉”。有次一个 Agent 想调“生成回复”能力语义匹配居然选到了“知识库检索”Agent因为两个能力的描述都含有“用户问题”这个词。从那以后我引入了路由置信度机制向量相似度低于阈值的请求一律不自动路由返回候选列表让调用方自己选。机器觉得“应该对”和业务上“确实对”之间永远要留一个人工确认的缓冲带。第三个坑是超时和重试的幂等性。早期有些 Agent 收到请求后处理超时调用方立刻重发导致同一个翻译任务被处理了两次客户收到重复邮件。后来我在消息模型里强制加上了idempotency_key由调用方生成服务端和接收方都做去重。这个字段是全局唯一的处理完后存入缓存重复请求直接返回第一次的结果。任何涉及外部调用的系统幂等性都是跑不掉的课题。5.3 性能与稳定性调优建议Agent-Reach 单机版用 FastAPI Redis 能扛住每秒几百次的 RPC 转发单机不够时再考虑加一层 Nginx 负载均衡和 Redis 集群。性能优化我做了三件事。第一把路由时用的 Agent 元信息加了一层本地缓存Redis 的 key 过期回调只能保证最终一致性但本地缓存的命中率能到 95% 以上实际延迟从 5 毫秒降到了 1 毫秒以内。第二把日志从同步改成异步批量写入避免高并发下日志拖累接口响应。第三把转发请求的超时时间设置成分层管理网络超时 3 秒、调用方业务超时 15 秒、整个链路总超时 30 秒。这样既不会让调用方无谓等待又给了长任务足够的执行空间。稳定性方面最重要的是给了 Server 一个优雅停机机制收到 SIGTERM 信号后先摘掉注册列表里的 Agent 标记将 status 置为 offline再等待当前 in-flight 请求处理完毕最后退出进程。这个过程能避免服务升级期间“请求发到半开却没人接”的尴尬。6. 后续扩展方向与个人体会Agent-Reach 目前的状态对我来说已经是很顺手的基础设施了。不过我脑子里还有一些明确想继续做的方向这里也列出来给想做类似事情的朋友参考。第一个方向是支持联邦注册。多个团队可以各自部署一套 Agent-Reach然后通过联邦机制互相注册“上游 Agent”这样不同团队甚至不同公司的 Agent 也能安全地互相调用有点像 Agent 界的“域名解析”。这个方向的价值在于让 Agent 协作从小规模内部试点走向生态级别。第二个方向是加强任务编排能力。现在 Agent-Reach 只负责路由和转发不编排任务。但实际业务里一个请求可能需要串行调用多个 Agent甚至要根据前一个 Agent 的结果决定下一个调谁。把这种编排逻辑做成可视化配置是很多非技术用户的需求。第三个方向是更深度的 MCP 集成。现在只是协议兼容后续可以直接做成“MCP Registry 网关”把任意 MCP Server 自动包装成一个 Agent 能力让 Agent 调用 MCP 工具像调用一个本地函数一样简单。最后分享一下我个人的实际操作体会。Agent-Reach 这个项目做下来我最大的感受是Agent 互联互通的技术难点根本不在网络通信而在于“怎么描述能力”和“怎么建立信任”。能力强描述得不好路由就不准权限设置得太严协作就死掉设定得太松出事又没人担责。这两件事值得花 80% 的时间去设计和打磨。另外一个很实用的小技巧给每个 Agent 的能力描述开头加一个“常用触发场景”段落例如“当用户需要查询天气时”而不是只写“查询天气”。这样的描述无论对语义匹配还是对人工维护都友好得多。这个细节是我在一次选错路由的故障中总结出来的现在每次写 capability 都会用到。Agent-Reach 不会取代任何 Agent 框架它只是让不同的 Agent 有个共同的“社交广场”。如果你手上的 Agent 也开始多到你记不清谁有什么能力不妨也搭一个这样的中间层。连接本身不会创造智能但连接的广度决定了 Agent 能力的边界。