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-Reach3.1 环境与依赖我的整个实现是基于 Python 3.11 FastAPI Redis 的依赖列表如下fastapi0.110.0 uvicorn[standard]0.29.0 redis5.0.0 pydantic2.6.0 httpx0.27.0 numpy1.26.4 sentence-transformers2.5.1 opentelemetry-api1.22.0 opentelemetry-sdk1.22.0 opentelemetry-instrumentation-fastapi0.43b0 opentelemetry-instrumentation-httpx0.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(hostlocalhost, port6379, db0) REG_KEY ar:agent:{agent_id} IDX_KEY ar:agents:ids HEARTBEAT_TTL 45 # 秒 def register_agent(reg: AgentRegistration): key REG_KEY.format(agent_idreg.agent_id) payload reg.model_dump_json() redis_client.set(key, payload) # 用 ZSET 记录所有 agent_idscore 存注册时间方便后续遍历 redis_client.zadd(IDX_KEY, {reg.agent_id: time.time()}) # 更新能力向量缓存 update_capability_vector(reg) def heartbeat(agent_id: str): key REG_KEY.format(agent_idagent_id) if not redis_client.exists(key): raise KeyError(fagent {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_idaid)) 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], threshold0.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(reverseTrue, keylambda 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(timeout30.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_code404, detailagent not found or offline) payload { params: req.params, request_id: req.request_id or generate_request_id(), } resp await api_client.post(reg.endpoint, jsonpayload) if resp.status_code 500: # 幂等场景下重试一次 resp await api_client.post(reg.endpoint, jsonpayload) 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( gatewayAR_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(interval15) app.run(host0.0.0.0, port5000)接入过程我实际跑下来算上写注册信息的时间一个熟悉自己代码的人大概 15-20 分钟能完成。整个适配层就只有两段代码一个注册动作一个标准协议入口。你原本的业务函数一行没改。这里有一个非常容易踩坑的地方注册时填写的 endpoint 必须是 Agent-Reach 网关能从自己的网络访问到的地址。如果代理和网关不在同一个容器网络或者同一个 VPC 里你填localhost:5000网关是找不到的要填代理服务的真实内网地址。3.5 链路追踪让每一次调用都有迹可循生产环境里最怕的就是“用户说我的请求挂了但日志里啥也没有”。Agent-Reach 默认给每次调用生成一个request_idUUID 短格式这个 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, kindSpanKind.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 问题速查表现象可能原因快速检查方法解决办法请求报 404agent_id 填写错误或未注册查注册中心是否有该 agent_id检查代理启动日志确认注册成功请求报 502目标 agent 地址不可达ping 一下 endpoint或看 agent 容器日志修正注册信息中的 endpoint请求超时agent 内部处理太久查 trace 中 span 耗时加超时时间或转异步模式路由匹配错误能力描述不规范或阈值过低打印 top3 相似度观察分数重写能力描述提高阈值间歇性 500多实例注册信息互相覆盖对比各实例注册时间戳强制单主实例注册其余仅心跳网关并发打满超时时间过长 同步阻塞看网关线程池占用率排查转异步任务模式5. 我在实际使用中攒下来的几点体会Agent-Reach 不是一个大而全的平台它更像一管润滑剂让原本互不相干的 agent 变成一张可以协作的网。我实际用下来最深的感受是接入的难易度主要取决于你对协议的态度只要能坚持“小协议 标准化注册 全局链路 ID”这三个原则后面几乎不会出大乱子。再分享一个小经验很多团队会把能力描述写成给开发看的说明书但其实那是给模型看的你应该用“用户在搜索什么话时会觉得这个代理能帮他”这种视角去写。我后来把能力描述改成了一组典型问题的集合类似“用户会问XX 商品几天能到”语义路由的效果立刻提升了一个档次。如果你后续想扩展Agent-Reach 还可以加编排策略、灰度发布、多环境隔离这些能力。但我的建议是先把最基础的四件事做扎实注册、发现、路由、追踪。这些做稳了多代理系统才能谈得上真正可用。