Agent-Reach：让AI代理互相发现、路由与调用的统一接入网关-平芜编程栈

AI 代理这个赛道最近有多热，不用我多说了。但真正把代理落到生产环境里跑起来的人，大概率都会撞上同一个问题：代理之间互相不认识，调用链全靠手写，今天接 A 框架，明天接 B 框架，每个代理的接口风格还不一样，最后整个系统变成一盘散沙。我最近折腾完的这套 Agent-Reach，就是专门解决这个问题的——它不是一个用来写 agent 逻辑的框架，而是一层让 agent 能被发现、被路由、被统一调用的中间桥梁。这篇文章我会把整个设计思路、核心代码、踩坑过程都摊开来讲，适合正在做多代理系统、或者准备把 agent 从 demo 往生产环境推的团队参考。

Agent-Reach 说白了就是一个“代理接入网关”。它把每个独立运行的 agent（不管你是用 LangChain、AutoGen、CrewAI 还是纯手写）当成一个可以被注册、被发现、被调用的服务节点，然后通过一个统一入口对外提供调用能力。你不需要改 agent 内部的业务逻辑，只需要给它套一层薄薄的适配壳，就能接入整个网络。

1. 项目全貌：Agent-Reach 到底解决了什么问题

1.1 一个现实问题：AI 代理各自为政

我先描述一个场景，你感受一下。一个公司里可能有多个 AI 代理在跑：一个做订单查询，一个做售后话术生成，一个跑数据分析，还有一个做内部知识库问答。它们可能是不同团队在不同时间用不同框架搭的，接口有的是 HTTP，有的是 WebSocket，有的甚至只是个命令行脚本。此时业务方说，帮我做一个统一的智能助手入口，它要根据用户的问题自动分流到正确的代理去处理。

你打算怎么做？最朴素的做法是写一堆 if-else，把用户问题硬编码跟代理能力匹配：包含“订单”就走订单代理，包含“退款”就走售后代理。但用户说的话你永远猜不全，规则越写越多，到最后连你自己都不知道该加到哪里。而且一旦某个代理要升级、下线或迁移地址，所有上游调用方都得跟着改。

这个场景的核心痛点有三层：

发现难：系统里到底有哪些代理可用，各自能干什么，没有统一的地方能查。
接入难：每个代理的协议、参数格式、鉴权方式都不一样，调用方要单独适配。
运维难：代理挂了、变慢了、或返回格式异常了，没有统一的观测手段，只能等用户投诉。

1.2 Agent-Reach 的定位与边界

Agent-Reach 的核心思想是“把 agent 当成一种可路由的服务”。它自己不生产代理由于业务逻辑，只做四件事：

注册中心：代理用一份标准化的元数据登记自己，说明名称、版本、能力描述、调用地址、鉴权方式。
统一入口：所有外部请求只打到 Agent-Reach 一个地址，由它内部负责把请求转发到具体代理。
语义路由：不只靠 keyword 做匹配，而是用 embedding 计算用户请求跟 agent 能力描述之间的相似度，再决定转给谁。
可观测性：每次调用都带着全局唯一的 request_id，从网关到目标 agent 的整个链路状态可追踪。

同时，Agent-Reach 严格限定自己不做的事：

不做 agent 内部实现。你的 agent 就算是拿定时任务脚本写的，也能接。
不做对话状态管理。多轮对话的记忆和上下文由各个 agent 自己负责，或者由你外面再接一层会话层。
不做业务数据存储。它只存“哪个 agent 能干什么、地址是什么、是否健康”，不碰业务数据。

这种边界的价值在于：它可以在不侵入你现有代码的前提下，把原先互不相通的 agent 串成一张网。

1.3 Agent-Reach 和 LangGraph、AutoGen 这类编排框架的关系

很多朋友一听说多代理，第一反应是拿 AutoGen 或者 LangGraph 来编排。但你要区分清楚，编排框架解决的是“多个 agent 之间的对话流程怎么走、消息怎么传递、任务怎么协同”，而 Agent-Reach 解决的是“一个 agent 怎么被别人找到并调用”。前者是“人与人之间怎么配合干活”，后者是“怎么让别人准确找到你的工位和电话分机”。

实际操作中，这两者完全可以叠加使用。就拿我自己搭的系统来说，最外层是一个 AutoGen 的 orchestrator，它负责拆解任务、决定要不要调用子代理；但它并不直接一个个去连各个底层服务，而是把每个子代理都注册到 Agent-Reach 上，orchestrator 统一通过 Agent-Reach 网关去发起子调用。这样编排逻辑和底层服务的位置、形态、存活状态就彻底解耦了，升级某个子代理时，orchestrator 这一层一行代码都不用改。

2. 整体设计与方案选型

2.1 为什么用网关模式而不是点对点直连

先说结论：点对点直连在实验阶段完全可行，一旦 exceeding 3-4 个 agent，就开始失控。类似“电话总机”和“人人互留手机号”的区别：人少的时候记几个号码没问题，人一多你就需要总机来干两件事——登记谁是谁，以及让来电者不用记每个人的分机。

网关模式最大的好处是三点：

调用方只需要知道 Agent-Reach 一个地址，不需要维护几十个 agent 地址。
这个代理网络的控制策略（鉴权、限流、路由）可以集中管理，不用在每个上游调用方里各写一遍。
网络里新增、下线、换址、升级 agent 时，调用方无感知，因为地址只在注册中心里变，调用方永远打的是网关。

当然，网关模式也会引入一个新的单点风险：网关挂了，所有调用都断了。这个问题我用三招来缓解：第一，网关本身无状态，可以多实例部署，前面挂负载均衡；第二，所有请求都带超时和重试，在网关层面做兜底；第三，代理注册信息的变更会实时同步到每个网关实例的本地缓存里，避免每次转发都强依赖中心化的注册存储。

2.2 注册中心的数据结构设计

注册中心是整个系统里最需要想清楚的部分。我最终采用的注册模型长这样：

class AgentRegistration(BaseModel): agent_id: str # 全局唯一，比如 "cs_order_query" name: str # 人类可读的名字 version: str # 代理版本，用于灰度 capabilities: list[str] # 自然语言能力描述，供语义路由使用 endpoint: str # http://host:port/invoke authn: str # 无认证 / api_key / jwt owner: str # 负责团队，便于追踪 extra: dict = {} # 扩展字段，比如最大并发、超时阈值

这里有两个细节值得展开。

第一个细节：capabilities 字段我强烈建议用自然语言描述，不要用一堆标签枚举。例如“订单查询”这个代理，它注册时写的能力描述可以是“查询订单状态、查看物流轨迹、获取订单详情、退改签订单”。这些话不是给代码用的，是给 embedding 模型用的。语义路由做匹配时，它把用户的问题也转成向量，再跟这几句话的向量算相似度，无需硬编码规则。

第二个细节：注册信息里不存任何密钥。网关转发请求时，只在它自己的配置里存一份“哪个代理用哪个密钥”的对照关系，而且密钥是放在环境变量或 secret 管理服务里，不落在注册信息里。你要是在注册表里存密钥，等哪天注册中心数据被人 dump 出来，整个系统的代理密钥就全暴露了。这个大家务必要注意。

2.3 路由策略：规则优先、语义兜底

路由是整个系统里最容易过度设计的地方。我见过有团队上了特别复杂的意图识别模型，结果线上效果还不如一个精确匹配加兜底。Agent-Reach 的做法很简单，三步走：

如果请求里明确指定了 agent_id，直接精确路由，不做任何语义计算。这适合集成方知道自己要找谁的情况。
如果请求里没有指定 agent_id，就走语义路由。先把用户原始输入转成 embedding，然后跟所有在线代理的 capability 向量做相似度计算，取最相似并且相似度超过阈值的那一个。
如果所有相似度都低于阈值，返回“无法匹配到合适代理”，并且把候选代理的 top 3 信息一并返回。这样做一方面方便调用方做二次确认，另一方面也方便我调阈值。

我当时的阈值经验值是 0.72（基于 OpenAI text-embedding-3-small 算的余弦相似度）。但这玩意跟你的能力描述怎么写、用户问题风格和模型版本都有关系，别直接照抄，需要拿你自己的数据跑一版再定。低于这个值的场景，宁可返回“无法匹配”，也别硬找一个完全不沾边的代理去消费用户的时间。

2.4 通信协议与容错策略

通信协议我选了 HTTP + JSON，而不是 gRPC。原因很朴素：

HTTP + JSON 的调试成本低，curl 一敲就能看出问题在哪，gRPC 还要装各种工具。
agent 服务的实现方五花八门，你没法要求所有人都懂 protobuf。
对 agent 调用这种场景，HTTP 的延迟开销完全在可接受范围内，真正的时间大头基本都消耗在模型推理上。

Agent-Reach 与后端 agent 之间的约定非常简单：请求体为{"params": <任意 JSON>, "request_id": "xxx"}，响应体为{"result": <任意 JSON>, "request_id": "xxx"}。这个协议刻意保持极简，所有的业务语义都放在 params 和 result 里面自由发挥。协议越小，接入方的学习成本就越低。

容错上，我做了三层：一是对后端 agent 的健康检查采用心跳上报，agent 每隔 15 秒向注册中心上报一次状态，超过 45 秒没上报就标记为 offline，路由时直接排除。二是转发请求时网关侧默认设置 30 秒超时，超时后按幂等接口标准重试一次。三是所有转发请求的目标地址只从注册中心拿，拿到之后在本地缓存 10 秒，避免每个请求都打注册中心。

3. 核心实现：从零跑通 Agent-Reach

3.1 环境与依赖

我的整个实现是基于 Python 3.11 + FastAPI + Redis 的，依赖列表如下：

fastapi==0.110.0 uvicorn[standard]==0.29.0 redis==5.0.0 pydantic==2.6.0 httpx==0.27.0 numpy==1.26.4 sentence-transformers==2.5.1 opentelemetry-api==1.22.0 opentelemetry-sdk==1.22.0 opentelemetry-instrumentation-fastapi==0.43b0 opentelemetry-instrumentation-httpx==0.43b0

关于 embedding 的选型，我生产环境用的是 OpenAI 的 text-embedding-3-small，但开发环境为了不烧 token，用的是本地 sentence-transformers 的all-MiniLM-L6-v2模型。两者切换只涉及一个函数，下面代码里我会统一用get_embedding()这个抽象函数，你根据自己实际情况注入实现即可。

3.2 注册中心实现

注册中心的核心职责就两个：写入注册信息和读取可用的代理列表。我选了 Redis 做存储，一方面是它的 TTL 天然适合做心跳过期，另一方面它的数据结构足够简单，不需要引入重型数据库。

import json import time import redis redis_client = redis.Redis(host="localhost", port=6379, db=0) REG_KEY = "ar:agent:{agent_id}" IDX_KEY = "ar:agents:ids" HEARTBEAT_TTL = 45 # 秒 def register_agent(reg: AgentRegistration): key = REG_KEY.format(agent_id=reg.agent_id) payload = reg.model_dump_json() redis_client.set(key, payload) # 用 ZSET 记录所有 agent_id，score 存注册时间，方便后续遍历 redis_client.zadd(IDX_KEY, {reg.agent_id: time.time()}) # 更新能力向量缓存 update_capability_vector(reg) def heartbeat(agent_id: str): key = REG_KEY.format(agent_id=agent_id) if not redis_client.exists(key): raise KeyError(f"agent {agent_id} not registered") redis_client.expire(key, HEARTBEAT_TTL) def list_agents() -> list[AgentRegistration]: agent_ids = redis_client.zrange(IDX_KEY, 0, -1) agents = [] for aid in agent_ids: raw = redis_client.get(REG_KEY.format(agent_id=aid)) if raw: agents.append(AgentRegistration.model_validate_json(raw)) return agents

心跳逻辑需要注意：心跳只做两件事，检查 key 是否存在，以及重置 TTL。它不重写注册信息。注册信息是重活，只在代理启动和配置变更时才执行。把这两者分开，可以有效避免心跳频率过高把 Redis 写成热点。

另一个容易被忽略的坑是：代理进程是多实例部署时，每个实例都要心跳，但注册信息只能由主实例写一次。否则多个实例同时写同一个 agent_id，可能会互相覆盖配置。我的办法是在代理端做一个简单的 leader 选举，或者让平台上只有第一个启动的实例负责注册，其余实例只负责心跳续期。

3.3 语义路由与统一调用入口

语义路由这块，核心是用 embedding 把“用户问题”和“agent 能力描述”映射到同一向量空间，然后算相似度。我实现了一个向量存储类，内部用 numpy 数组维护，生产环境可以换成向量数据库，但小规模场景 numpy 完全够用。

import numpy as np from functools import lru_cache capability_vectors: dict[str, np.ndarray] = {} def update_capability_vector(reg: AgentRegistration): text = " ".join(reg.capabilities) vec = get_embedding(text) capability_vectors[reg.agent_id] = vec / np.linalg.norm(vec) def semantic_route(query: str, agents: list[AgentRegistration], threshold=0.72): target_vec = get_embedding(query) target_vec = target_vec / np.linalg.norm(target_vec) scored = [] for agent in agents: vec = capability_vectors.get(agent.agent_id) if vec is None: update_capability_vector(agent) vec = capability_vectors[agent.agent_id] sim = float(np.dot(target_vec, vec)) scored.append((sim, agent)) scored.sort(reverse=True, key=lambda x: x[0]) best_sim, best_agent = scored[0] if best_sim >= threshold: return best_agent return None

统一调用入口我直接做成了 FastAPI 的两个路由：一个是POST /v1/invoke，接收{"agent_id": "", "params": {...}}，精确调用指定代理；一个是POST /v1/chat，接收{"query": "..."}, 自动语义路由到合适的代理。外部业务方日常用的基本都是第二个，这样他们的调用方式就和“跟一个大模型对话”没什么区别了。

网关转发逻辑用 httpx 的异步客户端实现：

import httpx from fastapi import FastAPI, HTTPException app = FastAPI() api_client = httpx.AsyncClient(timeout=30.0) @app.post("/v1/invoke") async def invoke_agent(req: InvokeRequest): reg = get_agent_registration(req.agent_id) if not reg or not is_agent_alive(reg.agent_id): raise HTTPException(status_code=404, detail="agent not found or offline") payload = { "params": req.params, "request_id": req.request_id or generate_request_id(), } resp = await api_client.post(reg.endpoint, json=payload) if resp.status_code >= 500: # 幂等场景下重试一次 resp = await api_client.post(reg.endpoint, json=payload) return resp.json()

注意我上面特意区分了 5xx 和非 5xx：只有 5xx（目标服务自己挂了或者超时）才重试，4xx 说明是请求本身有问题，重试也没意义。

3.4 为一个真实代理接入 Agent-Reach

光说框架肯定不行，我写一个真实的接入案例。假设我有一个订单查询 agent，原本是用 Flask 写的，就一个/query_order接口。现在要接入 Agent-Reach，我需要做两件事：第一，写一个启动时注册的动作；第二，把原来的接口包一层，让它接收标准协议。

改造后的代理端入口长这样：

# order_agent/main.py from flask import Flask, request, jsonify import agent_reach_sdk # 这是 Agent-Reach 的客户端 SDK app = Flask(__name__) AR_ENDPOINT = "http://agent-reach-gateway:8000" @app.post("/invoke") def invoke(): data = request.get_json() params = data["params"] request_id = data["request_id"] # 这里把你原本的业务函数对接进来 result = query_order(params) return jsonify({"result": result, "request_id": request_id}) if __name__ == "__main__": # 启动时注册 agent_reach_sdk.register( gateway=AR_ENDPOINT, registration={ "agent_id": "order_query", "name": "订单查询代理", "version": "1.0.0", "capabilities": [ "查询订单状态", "查看物流轨迹", "获取订单详情", "退改签订单", ], "endpoint": "http://order-agent:5000/invoke", "authn": "api_key", }, ) # 启动心跳线程 agent_reach_sdk.start_heartbeat(interval=15) app.run(host="0.0.0.0", port=5000)

接入过程我实际跑下来，算上写注册信息的时间，一个熟悉自己代码的人大概 15-20 分钟能完成。整个适配层就只有两段代码：一个注册动作，一个标准协议入口。你原本的业务函数一行没改。

这里有一个非常容易踩坑的地方：注册时填写的 endpoint 必须是 Agent-Reach 网关能从自己的网络访问到的地址。如果代理和网关不在同一个容器网络或者同一个 VPC 里，你填localhost:5000网关是找不到的，要填代理服务的真实内网地址。

3.5 链路追踪：让每一次调用都有迹可循

生产环境里最怕的就是“用户说我的请求挂了，但日志里啥也没有”。Agent-Reach 默认给每次调用生成一个request_id（UUID 短格式），这个 ID 会从入口一直传递到目标 agent。Agent-Reach 本身不统一收集日志，而是把这个 ID 放进 OpenTelemetry 的 span 属性里，让下游的 tracing 系统按 ID 查整条链路。

from opentelemetry import trace from opentelemetry.trace import SpanKind tracer = trace.get_tracer(__name__) @app.post("/v1/invoke") async def invoke_agent(req: InvokeRequest): with tracer.start_as_current_span("ar.invoke", kind=SpanKind.CLIENT) as span: span.set_attribute("ar.request_id", req.request_id) span.set_attribute("ar.agent_id", req.agent_id) span.set_attribute("ar.endpoint", reg.endpoint) # ... 转发逻辑

这样的好处是，线上排查问题时，直接拿ar.request_id一查，就能看到请求在哪一步耗时最长、哪个上游返回了 4xx、哪个环节发生了超时重试。对定位多代理系统中的疑难杂症帮助非常大。

4. 常见问题与排查实录

4.1 网关转发请求超时

我遇到的第一类高频问题就是超时。现象是网关日志里出现httpx.ReadTimeout，或者客户端侧看到 504。

这里我先纠正一个常见误解：超时不一定是目标 agent 处理得慢，很多时候是 agent 自己也在等别的上游。比如你接入的 agent 内部又调了一个 LLM 接口，LLM 流式生成要 40 秒，而你 Agent-Reach 网关只等了 30 秒，自然就超时了。

解决思路分成两路：

如果目标是短任务，把超时时间调整为 60 秒，并且要求 agent 在处理期间周期性地把中间状态上报回网关，避免反压。
如果目标是长任务（超过 1 分钟），不要让调用方同步等待，改成提交任务后立即返回 task_id，再由 agent 通过 webhook 或轮询接口回调结果。这个模式虽然要多写点代码，但才是生产级的长任务正确姿势。

我自己的经验准则是：Agent-Reach 网关保守一点，默认 60 秒超时；超过 60 秒的业务，一律走异步任务模式。别硬调超时时间来解决长任务，不然并发一高，网关线程池就被拖死了。

4.2 语义路由匹配到了错误代理

语义路由在开发环境测试都正常，一上真实流量就偶发错配。这是我整个项目里调试时间最长的问题。

排查下来原因主要有三个：

第一，能力描述写得太笼统。比如某代理写“处理用户咨询”，另一代理写“处理用户售后”，这两个描述在向量空间里相似度极高，用户问“我的订单怎么还没到”，可能就被分到咨询代理去了。解决办法是能力描述要写“动词 + 对象 + 典型问题”，越具体越好。

第二，embedding 模型对相似措辞区分度不够。我后来在能力描述里故意加上反例，比如“不处理价格谈判、不处理合同审核”，把边界画出来，正确率有明显提升。

第三，阈值设低了。生产环境中我把阈值从 0.72 提高到了 0.76，宁缺毋滥，错配率从 8% 降到不足 2%。

4.3 注册信息抖动导致路由不稳定

有个阶段我发现线上请求时而成功时而失败，查了半天，发现是有两个 agent 实例注册的 agent_id 相同，但 endpoint 不同：一个还是旧地址，一个是新地址。注册信息发生抖动后，网关拿到的 endpoint 时对时错，重试自然也是对的打一次错的打一次。

后来我改了两处：一是注册时强制校验版本号，如果新实例版本与现有注册版本不一致，必须要先等旧实例把 TTL 过期；二是在 SDK 里授信同一个 agent_id 只能由一个主实例执行注册，其余实例只做心跳。这个第 2.2 节提过，我觉得值得反复强调，多实例部署时必须处理，否则就是一个隐蔽的大坑。

4.4 鉴权与数据安全的几个坑

Agent-Reach 网关对外暴露的是一个统一入口，这个入口一旦被打穿，攻击者可以调用所有代理。所以我在网关层做了自己的鉴权：所有进来请求必须带有效的 API key，网关校验通过后才允许转发。调用方传进来的 key 不是目标 agent 的 key，而是 Agent-Reach 自己的调用凭证。网关自己维护一份目标 agent 所需的密钥，从环境变量读取，绝不入库。

另一个需要注意的点是日志脱敏。网关如果直接打印转发请求的 params，客户个人信息可能就落盘了。我的做法是：日志默认只打印 request_id、agent_id、耗时和状态码，不打印完整 params。只有在显式开启 debug 模式时，才打印参数体，而且要对常见敏感字段（身份证号、手机号、卡号）做脱敏。

4.5 问题速查表

现象	可能原因	快速检查方法	解决办法
请求报 404	agent_id 填写错误或未注册	查注册中心是否有该 agent_id	检查代理启动日志，确认注册成功
请求报 502	目标 agent 地址不可达	ping 一下 endpoint，或看 agent 容器日志	修正注册信息中的 endpoint
请求超时	agent 内部处理太久	查 trace 中 span 耗时	加超时时间或转异步模式
路由匹配错误	能力描述不规范或阈值过低	打印 top3 相似度，观察分数	重写能力描述，提高阈值
间歇性 500	多实例注册信息互相覆盖	对比各实例注册时间戳	强制单主实例注册，其余仅心跳
网关并发打满	超时时间过长 + 同步阻塞	看网关线程池占用率	排查转异步任务模式