上个月我们在生产环境里做了一次非常难看的实验两个AI Agent各自负责一段业务闭环A负责接收用户需求B负责执行数据回流结果A在大模型能力上表现很好却怎么也够不到B的接口。最后我从日志里翻出来A在上下文里生成了整整5次错误的目标地址每一次都试图猜B的入参格式。这个场景让我确定了一件事大模型驱动的Agent真正缺的从来不是模型智商而是触达——能不能找到该找的系统、能不能把上下文准确递出去、能不能在多个Agent之间建立一条可靠的路。这也就是后来我们内部命名为Agent-Reach的项目真正要回答的问题。这篇文章把我这段时间的调研、设计、踩坑和拆解完整记录下来给正在做Agent协作、编排、跨系统调用的朋友一份可以照着抄的参考。1. 为什么我放弃了一个Agent包打天下的念头1.1 从Demo到生产Agent最缺的不是推理而是触达如果你只是做一个演示用的Agent问题确实很简单把用户的自然语言prompt扔给大模型让大模型输出一段JSON再调用一个搜索引擎或者一个内部API就够了。但一旦进入生产环境Agent要面对的是几十个内部系统、多个业务对象、不同的身份鉴权方式、各不相同的数据结构。这个时候你会发现最消耗时间的不是让模型想明白怎么做而是让模型找到正确的调用目标并成功接通。我接触过不少团队早期都倾向于做一个超级Agent把所有工具塞到一个巨大的function列表里让模型自己选。结果到了30个工具之后模型的选择准确率肉眼可见地下滑经常在相似命名的接口之间徘徊还出现过同时调用两个互斥接口的奇葩行为。本质上这是因为大模型在处理能力选择时依赖的是工具描述的语义匹配而同一组织里大量工具的命名、参数、返回结构本身就是混乱的模型再聪明也容易被带偏。Agent-Reach的思路与此相反承认单Agent的能力是有限的也承认集中式工具注册会让Agent不堪重负。所以我们把一个大Agent拆成多个小Agent每个Agent只持有与自己职责相关的工具。这倒不是新鲜事真正难的是拆完之后怎么让它们互相找得到、叫得通、传得对。1.2 Agent-Reach到底在解决哪一层问题我见过不少团队在讨论多Agent协作时第一反应是要不要用某个开源编排框架、要不要引入消息队列、要不要上K8s服务网格。我的建议是先别管基础设施先看你的Agent之间是怎么通信的。Agent-Reach做的事情本质上只有三件能力注册每个Agent启动时向注册中心声明自己拥有的能力和对应接口意图路由收到任务的一方根据任务描述找到能处理该意图的Agent并完成调用上下文收敛把当前任务的必要上下文打包传给目标Agent而不是把整个会话历史一股脑灌过去。这三件事听起来很简单但每件往深里做都有坑。注册中心需要考虑心跳和失效剔除意图路由需要考虑Agent能力的语义表达和匹配算法上下文收敛则需要考虑token消耗和敏感信息过滤。任何一个环节出问题都会表现为Agent之间明明能互相ping通但业务链路就是跑不通。另外我要说清楚Agent-Reach不是一个通用的分布式中间件。它更接近一个Agent协作协议轻量级运行时的组合。你可以把它理解成多个Agent之间的通信规范再配上一个很薄的网关去落地这套规范。网关不需要承担业务逻辑只需要做寻址、鉴权、转译。2. Agent-Reach的整体设计让每个Agent变成可触达的节点2.1 最小拓扑注册中心、路由网关、消息通道先给出我们最终落地的Agent-Reach最小拓扑方便你对全局有个概念每个Agent实例是一个节点对外暴露HTTP接口并在启动时向注册中心上报自己的元信息注册中心只存三类数据Agent名称、Agent的Base URL、Agent能处理的意图标签列表路由网关是所有跨Agent调用的唯一入口上游Agent不直接访问下游Agent而是把请求交给网关消息通道默认走HTTP同步调用后续可以替换成异步队列。为什么不让Agent之间直接互调因为直接互调会带来两个问题。一是寻址逻辑散落在每个Agent里每个Agent都要维护一份其他Agent地址表地址一变就要到处改配置二是权限和审计无法统一没有经过网关的请求没法做统一的身份校验和日志采集。Agent-Reach把路由网关作为唯一入口之后所有跨Agent调用的路径变得几乎可以从零推理。这里有一个容易被误解的设计点注册中心不能做成强依赖。如果网关在启动后每一笔请求都要去注册中心查一次地址注册中心一挂整个协作网络就瘫痪。更稳的做法是网关启动时拉一次全量注册信息后续依靠本地缓存动态更新注册中心则通过心跳维护Agent的存活状态。我们的Agent-Reach初版就是这么做的后面再演进时可以把注册中心的缓存同步改成订阅推送但初版尽量保持简单可靠。2.2 服务契约Agent之间到底在传什么我在设计Agent-Reach时踩过的第一个认知坑是把Agent之间的消息当成普通的RPC请求来处理。后来发现这完全不对。Agent之间的通信天然带有多轮性和不确定性。下游Agent可能无法一次完成任务可能需要追加信息也可能直接返回这件事我干不了请换人。所以我们给Agent-Reach定义了两层交互模式同步指令型上游明确知道要找谁干什么直接以指令方式调用下游下游返回结构化结果。比如查库存的Agent收到查询SKU A的可用库存返回一个JSON对象。异步协作型上游不确定任务该由谁做或者任务需要分阶段执行则把任务发布到协作通道由路由网关结合注册中心找到目标Agent通过回调或轮询方式收结果。这两层模式的差异对接口设计影响很大。同步指令型可以用标准的HTTPJSON接口上游把参数传给下游就行。异步协作型则需要协议里带上task_id、callback_url、timeout、max_retry等字段否则协作过程根本无法追踪。以Agent-Reach目前定义的标准任务报文为例核心字段包括{ task_id: reach_20250121_001, from_agent: order-agent, to_agent: stock-agent, task_type: sync, payload: { action: query_stock, params: { sku: SKU-A-001 } }, context: { max_tokens: 1200, inherit_history: false, source_trace: [user-session-889] }, callback: { url: http://order-agent/callback/stock, timeout_ms: 30000 } }inherit_history这个字段值得说道。它是我在反复踩过上下文爆炸的坑之后加的默认值是false意思是下游Agent只接收payload里明确传过来的参数不继承上游的冗长会话历史。只有当任务确实需要完整背景时上游才主动把摘要塞进payload。这个设计虽然不太符合Agent应该有全知视角的浪漫想象但它是保证生产可用的关键。2.3 意图路由我该怎么知道找谁干活在注册中心里每个Agent需要上报自己支持的一组意图标签例如订单查询库存变更客户信息更新风险评分计算。路由网关注册时并不懂得每个Agent的业务含义它只知道每条记录里有意图标签和地址。当任务报文到达网关时网关先解析payload里是否有明确的to_agent字段。如果有就按名称直接路由如果没有就做一个语义匹配把任务的意图描述和注册表里的意图标签做相似度计算选最匹配的那个Agent。我强烈建议把明确指定和语义匹配分成两个开关。现实里你总会遇到某些业务必须精确流向某一个Agent比如资金操作只能由风控Agent负责不能因为某个语义相似度更高的Agent被错误命中就换了执行者。Agent-Reach初版里有一个route_mode参数取值可以是exact、semantic、hybrid路由模式行为特征适用场景exact必须按to_agent名称路由找不到就报错资金、审批、写操作等强约束场景semantic忽略to_agent只用语义匹配意图检索、建议、通用问答等弱约束场景hybrid先找to_agent找不到再降级语义匹配大多数内部协作场景hybrid是我们在生产里最常用的模式。它既保留了业务上的显式意图又给了系统一定的容错空间。但这里有个坑语义匹配和to_agent降级如果处理不好很容易把一个本该明确拒绝的任务转发给错误的Agent导致下游莫名其妙收到一堆不该处理的数据。所以降级发生时一定要在响应里增加warning字段并且全链路日志里要记清楚。3. 手把手搭一个Agent-Reach的雏形3.1 技术选型与目录结构我不倾向于在这个环节搞得很重做POC阶段能用一套轻量技术栈快速验证就够了。Agent-Reach的初版我们是用Python FastAPI实现的没有引入重量级注册中心而是先把注册信息存在一个带TTL的本地内存表里后面确认业务方向后再替换成真正的高可用存储。说下为什么选FastAPI一是异步特性对Agent这种可能长时间等待下游回执的场景很友好二是天然支持Pydantic做数据校验正好用在校验路由报文的schema上三是OpenAPI文档可以直接做上手调试省了单独写接口文档的精力。目录结构大致如下agent-reach/ ├── gateway/ │ ├── registry.py # 注册表的读写、心跳、过期 │ ├── router.py # 精确路由和语义路由逻辑 │ ├── sender.py # 统一HTTP转发与超时处理 │ └── api.py # 网关对外API ├── agents/ │ ├── order_agent.py │ ├── stock_agent.py │ └── customer_agent.py ├── schemas/ │ ├── base.py # 任务报文的字段定义 │ └── enums.py └── config/ └── settings.py3.2 注册中心的实现初版注册中心的核心逻辑不复杂。每个Agent启动时调用POST /register上报名称、地址、意图标签之后每隔15秒发送一次心跳网关侧每40秒清理一次过期节点。我先给出注册相关的数据结构定义# gateway/registry.py import time from typing import Dict, List class AgentRegistry: def __init__(self): self._agents: Dict[str, dict] {} self._ttl_seconds 40 def register(self, agent_name: str, base_url: str, intents: List[str]): self._agents[agent_name] { base_url: base_url.rstrip(/), intents: intents, registered_at: time.time(), last_heartbeat: time.time(), } def heartbeat(self, agent_name: str): if agent_name not in self._agents: return False self._agents[agent_name][last_heartbeat] time.time() return True def cleanup(self): now time.time() expired [ name for name, info in self._agents.items() if now - info[last_heartbeat] self._ttl_seconds ] for name in expired: del self._agents[name] return expired def get_agent(self, name: str): return self._agents.get(name)有个细节要提醒你心跳接口和业务接口不要走同一个端口。如果Agent的业务线程被某个长耗时任务阻塞了心跳也会跟着阻塞注册中心会误判Agent下线。Agent-Reach的做法是心跳用单独的轻量级线程或进程去发确保只有进程级别的故障才会导致心跳中断。3.3 路由网关的实现网关的职责是接受上游Agent的转发请求并把它带给目标Agent。路由逻辑里最重要的是异常处理路径也就是目标Agent超时、目标Agent返回5xx、目标Agent不存在这三种情况分别怎么回给上游。网关的转发核心代码大概长这样# gateway/router.py import asyncio from urllib.parse import urljoin from schemas.base import ReachTask, ReachResponse class ReachRouter: def __init__(self, registry): self.registry registry async def route(self, task: ReachTask) - ReachResponse: target None if task.route_mode exact: target self.registry.get_agent(task.to_agent) if not target: return ReachResponse(code404, messagetarget agent not found) elif task.route_mode semantic: target self._semantic_match(task.intent) elif task.route_mode hybrid: target self.registry.get_agent(task.to_agent) if not target: target self._semantic_match(task.intent) if not target: return ReachResponse(code404, messageno agent matched) url urljoin(target[base_url], /invoke) return await self._send_request(url, task) def _semantic_match(self, intent: str): # 用关键词向量相似度叠加计算初版可以先做关键词打分 scored [] for name, info in self.registry._agents.items(): score 0 for tag in info[intents]: if tag in intent or intent in tag: score 1 scored.append((score, name, info)) scored.sort(reverseTrue) return scored[0][2] if scored and scored[0][0] 0 else None注意_semantic_match里的拿法。初版用关键词重叠度做意图匹配完全够用因为Agent注册时写下的意图标签本身是有限集合题目比较简单。但一旦意图标签增长到几十个、上百个就必须引入向量召回否则会出现大量边界case。这个后面在扩展方向里我会细说。3.4 把三个Agent接进网络Agent侧接入方式很简单每个Agent只需要实现一个/invoke接口接收标准的ReachTask报文处理完返回ReachResponse。下面是其中一个Agent的典型实现# agents/stock_agent.py from fastapi import FastAPI, Request from schemas.base import ReachTask, ReachResponse app FastAPI() app.post(/invoke) async def invoke(task: ReachTask): if task.payload.get(action) ! query_stock: return ReachResponse(code400, messageunsupported action) sku task.payload[params][sku] # 模拟库存查询实际接入时替换为真实业务查询 result {sku: sku, available: 42} return ReachResponse(code0, messageok, dataresult) app.post(/heartbeat) async def heartbeat(): return {status: alive}如果你是用FastAPI写Agent这个接入成本基本就是零。但我在落地时发现团队里很多Agent不是Python写的而是跑在Node.js、Java或者更奇怪的环境里。这种异构情况下你只需要保证一个HTTP端点实现了相同的报文规范其余语言差异对Agent-Reach都是透明的。协议比SDK更重要这句话在这种体系里永远是第一原则。三个Agent都接好之后整个链路可以这样跑用户在业务系统里发起一个订单查询请求order-agent收到后发现自己需要库存数据order-agent构造一个ReachTask把action和params塞进payload把to_agent设为stock-agentroute_mode设为exact网关查注册中心拿到stock-agent的地址把报文转发过去stock-agent执行完库存查询把结果原路返回order-agent拿到库存结果后结合自己的领域知识组织最终答复。4. 实测中的意外那些文档上不会写的坑4.1 路由黑洞Agent明明在线却找不到第一次联调的时候我遇到一个很诡异的想象。注册中心里明明能看到stock-agent在线路由网关也返回了目标信息但请求送出去之后stock-agent那边一直没有日志。查了网关的日志显示request已经发出查了stock-agent的网络监听发现根本没有收到任何包。排了很久之后才定位到问题我们给Agent节点接入了一个只允许白名单来源的网络安全策略网关所在的Pod地址不在白名单里。跨Agent通信的包在网络层就被拦截了Agent进程本身完全无感知。这件事给我的教训是Agent-Reach作为一个协作协议框架能保证应用层面的对接但保证不了网络层和基础设施层的打通。你在做压力测试之前一定要先做一次最小链路拨测——从上游Agent发一个最小请求全链路追踪每一个节点的接收和响应确认网络路径、鉴权策略、防火墙规则全部放行。否则后面所有排查都会被这个隐形问题带偏。4.2 上下文炸弹一次委派把3万token塞给下游另一个实际场景里的高频问题上游Agent在和用户连续聊天时上下文里积累了很长一段历史。当它需要委派给下游Agent时为了让下游理解背景我们最初直接把整段历史塞进payload的context里。第一次试运行就出事下游Agent的模型接口直接因为token超限报错。这背后是一个很典型的思维定势我们总觉得关联上下文越完整越好。但Agent协作不是给人看文档而是给另一个模型程序传递必要信息。下游Agent真正需要的往往只是这个用户是谁、目前到了哪一步、需要你来执行什么动作。历史闲聊、中间推理过程、多次失败的尝试记录这些对下游不仅无用反而会干扰注意力。Agent-Reach给出的解法是在发送方侧增加一个context_builder用大模型对原始会话做一次压缩摘要只保留与当前任务相关的实体、用户意图和前置约束条件。然后再决定这些摘要信息是放进payload里面还是不传。压测数据显示加了这个摘要环节后下游Agent的首次响应准确率提升了20%以上很大程度是因为它不必再被无关历史干扰。4.3 循环委派两个Agent互相踢皮球还有一次我们的客服Agent和订单Agent出现了一个死循环客服Agent发现用户问的是退款进度就委托给订单Agent订单Agent发现自己没有退款处理能力又委托回客服Agent。两个Agent在网关的协助下互相调用了几十次直到一个进程把限流阈值打满才停下来。这类问题比看起来更危险因为它不会像超时或404那样直接报错而是持续消耗系统资源表现得像系统变慢。Agent-Reach在协议层加了两个缓解机制。第一个是max_hops。任务报文里记录已转发的跳数网关每次转发前递增超过设定上限就拒绝继续转发直接返回协作链路过深请检查路由设计。这就把无限循环掐死在可控范围内。第二个是task_affinity。每个Agent在context里声明自己已经处理过这个task_id如果路由网关发现目标Agent已经在处理历史中出现过会触发告警并停止自动转发。说实话这两个机制只是兜底最根本的解法还是在设计阶段梳理清楚Agent的职责边界。循环委派本质上不是因为Agent不懂协作而是因为职责划分和意图标签定义不够严格。一个Agent如果既没有能力处理又没有明确意识去拒绝它就会把责任推给下一个Agent。因此在Agent的意图标签里必须包含一个明确不负责的事项列表这个列表用来支持快速拒绝。4.4 权限边界被触达二字冲垮把多个Agent打通之后权限管理会变得非常棘手。单Agent场景下权限问题只在用户-Agent这一层出现Agent协作场景下还有Agent-Agent这一层。如果一个下游Agent拥有写数据库权限而上游Agent可以被任意用户触发那入侵者就可能通过诱导上游Agent转发恶意请求来间接调用下游Agent的高权限能力。我们在生产里第一次出权限事故就是因为这个。客服Agent的prompt注入漏洞被外部触发攻击者伪装成一个查询库存的指令让客服Agent转发给stock-agent。stock-agent只校验了报文格式没有验证最终调用者身份结果把库存数据回传给了越权用户。事后我们在Agent-Reach里加了一层调用链身份透传机制。每个任务报文必须携带source_identity字段记录初始用户或服务账号的ID。目标Agent在执行业务逻辑前要把source_identity和自身接口的权限策略做一次白名单校验。跨Agent转发的过程中该身份字段只允许被覆盖为更小权限范围不能提升。这套机制在落地时有些笨重因为它要求所有Agent在业务逻辑里配合做权限判断而不是依赖网关一肩挑。但我必须说实话靠网关统一做权限判断是可行但不够的因为网关只看到报文看不到业务语义。真正的权限决策必须发生在业务侧由Agent结合自身业务规则来决定是否执行。这也是Agent-Reach协议设计上没有把权限完全收归网关的原因。5. Agent-Reach能带来的真实收益与适用边界5.1 实际收益编排工作量下来了多少在Agent-Reach上线之前我们同一套协作流程是靠一堆if-else脚本和手工配置的HTTP调用来完成的。每新增一个Agent就要改一遍上游的硬编码地址还要处理各种不顺眼的超时和重试。Agent-Reach替换之后最直接的收益是新增Agent的接入成本下降了。新Agent只要注册意图标签并实现标准报文接口网关就能自动路由业务侧无须再逐个修改硬编码。我统计过一轮数据原先新增一个协作节点需要大概2到3人天包含地址配置、联调测试、异常处理改造Agent-Reach模式下如果新Agent的接口做到了协议兼容接入时间能压缩到半天以内。这个收益在Agent数量超过5个之后尤其明显。不稳定性的数据也比较好看。之前直连模式下下游Agent更换地址时上游Agent会在运行时不断报连接错误现在因为统一走网关寻址地址变更只需要更新注册表对上游完全透明。我们上线后的两个月内做过三次地址迁移没有一次对调用链造成影响。5.2 什么场景千万别用Agent-Reach不是所有项目都应该切割成多Agent协作Agent-Reach也一样有自己的适用边界。根据我的经验至少有三类场景是不建议用的流程固定、参数简单的场景。如果系统只有两三个确定性接口调用用硬编码的Python脚本或者工作流引擎就够了。强行引入Agent和路由层只会在维护上制造麻烦。对响应时延极其敏感的场景。Agent-Reach多一跳转发会额外增加几十毫秒到几百毫秒的延迟而且在语义路由时还可能要做向量计算。如果是交易行情或者实时控制类场景这个开销不能接受。团队还搞不清楚Agent职责边界的场景。如果你分不清这个业务到底该谁负责一上来就拼多Agent协作只会把混乱扩散到整个链路。Agent-Reach会放大设计上的缺陷而不是修复它。这部分听起来有点像劝退但我确实见过太多项目在连业务需求都没梳理清楚时就急着搭一套花哨的Agent协作网络。最后Agent和Agent之间天天来回复制消息真正的业务反而没人处理。技术方案的复杂度永远应该向业务复杂度看齐而不是反过来。5.3 往后扩展的大致方向Agent-Reach这个雏形能跑通但它离一个生产级多Agent协作平台还有距离。我对自己项目的下一步规划大致有三个方向第一个方向是把意图路由从关键词匹配升级成真正的语义召回。具体做法是接入一个embedding模型把意图标签和任务描述都映射到向量空间用余弦相似度搜索候选Agent同时对得分过低的候选直接拒绝宁可返回无匹配也不想误路由。这里建议做好缓存因为每次请求都重新对全量标签做向量计算会很浪费。第二个方向是增加异步长任务的支持。现在Agent-Reach的HTTP同步模式适合秒级返回的查询类任务但一旦下游Agent需要几分钟甚至十几分钟才能完成操作同步等待就不现实了。我会给网关加一个任务队列中间层配合任务状态查询接口让上游Agent轮询或订阅任务状态。第三个方向是完善可观测性。多Agent协作最难调的其实是链路追踪。每个任务跨了哪几个Agent、每跳的耗时是多少、上下文在哪一步被截断或篡改这些信息在排障时几乎是命根子。后续我想基于OpenTelemetry的规范扩展一个链路追踪中间件让每个任务报文自动携带trace_id所有Agent的日志都基于这个trace_id做串联。6. 最后给想动手的同学的几点实在话如果你也要做类似Agent-Reach的协作层项目我建议先想清楚一个问题你的Agent协作到底是编排驱动还是自主驱动。编排驱动是指由一个中心节点决定每个步骤找谁自主驱动是指每个Agent自己判断下一步找谁。Agent-Reach目前的设计偏向二者混合但整体上更重路由因为它本质上是帮助Agent互相找到对方而不是替它们做全盘规划。先把这一点想清楚架构才不会走偏。另外我真心建议不要在项目第一天就引入太重的框架。用FastAPI写一个注册表和路由网关用JSON定义报文规范手动让两三个Agent跑通链路你就能对Agent协作的核心痛点有真实体感。相比直接套用那些庞大的多Agent框架这种从零搭一遍的方式更让人理解每行代码在解决什么问题。我在Agent-Reach上最大的体会是能让Agent协作跑起来的不是哪个大模型更聪明而是你有多认真地把注册、路由、报文、权限、追踪这些枯燥的骨架问题逐个解决。先把骨架搭稳Agent的天花板才会真正打开。