去年下半年我在公司内部做过一次小调研结果有点意外——我们团队同时维护着七八个不同用途的Agent服务有做客服问答的、有自动排班的、有写报表分析的但这几个Agent之间几乎没有直接协作。客服Agent要查订单得去数据库里捞数据排班Agent想用团队里已有的翻译能力根本不知道谁做了更别提两个Agent需要互相交接任务的时候各说各话格式五花八门。这个问题的本质是Agent本身不缺单点能力缺的是被其他Agent触达的能力。所以我搭了一个叫Agent-Reach的轻量互连层——它做的事情很简单让任何Agent都能对外发布自己能干吗同时让任何Agent都能找到别人并发出任务然后把结果安全地带回来。这篇文章把Agent-Reach的设计思路、核心实现和踩过的坑完整写下来给同样在折腾多Agent协同的朋友一个参考。如果你现在只有两三个Agent跑着可能觉得没必要但只要Agent数量一多、协作关系一复杂你会发现这层互连基础设施几乎是刚需。1. 为什么会有Agent-Reach三个让我头疼的真实场景1.1 客服Agent和订单Agent互不通信先说说最直接的痛点。我们线上客服用的是基于大模型RAG的问答Agent用户问我的订单到哪了客服Agent能理解意图但真要查订单状态它只能走一条很绕的路拼SQL去查订单库或者调一个内部HTTP接口。这个HTTP接口是订单团队的Agent对外暴露的但接口文档只有order_id、返回JSON这种最原始的信息客服Agent为了接它写了一堆解析逻辑。问题在哪里首先是耦合。订单Agent只要改字段名客服Agent这边就得跟着改其次是权限。客服Agent拿着数据库账号理论上能查到所有用户订单这在数据安全上是个隐患最后是体验。一次查询要经历意图识别→拼SQL→查库→再生成回答四条链路随便哪一步出错用户看到的就是抱歉我现在无法查询。这种场景其实很多人都有Agent的能力散落在各自的服务里别人要用要么走后门直连数据要么写死接口。本质上缺一层能力注册与调用的中间媒介。Agent-Reach最初就是为了解决这个原始问题才立项的。1.2 Agent的能力无法被发现第二个场景更尴尬。我们数据分析团队做过一个很不错的翻译Agent支撑多语言报表。后来排班Agent的项目启动负责人不知道内部已经有这个能力自己又调了一个外部翻译API花了三周效果还没内部的好。这不是个例而是多Agent体系膨胀后的必然结果——Agent越来越多能力清单只存在各项目负责人的脑子里。这让我意识到Agent系统缺的不仅是通信还有发现。一个Agent应该能对外回答这样一个问题你会什么你的输入是什么格式你现在的可用状态如何没有这套描述Agent越多重复建设越严重。后来接入Agent-Reach之后我们发现光是内部能力发现这一个功能就至少帮团队省掉了一次完全重复的Agent开发ROI已经值回票价。1.3 Agent之间没有标准的任务交接方式第三个场景是关于任务交接。我们有一个流程是客服Agent识别到用户要改地址需要把任务转给订单Agent执行。最初的做法是客服Agent直接把改好的地址JSON POST给订单Agent的某个接口。看起来没什么问题但等到第三个Agent也要参与比如需要先风控审核问题就来了——谁来记录这单子现在在谁手里如果订单Agent处理失败任务要不要重新排队回执是200就代表改成功了吗没有统一的任务交接协议每次新增协作都是重新发明轮子而且极容易出错。有一次就是因为客服Agent在订单Agent返回处理中时直接判定成功导致用户地址没改成客诉压了一整天。这三个场景叠加在一起结论就非常清晰需要一层独立的Agent互连基础设施让Agent之间的协作从点对点写接口升级成注册-发现-路由的标准模式。这就是Agent-Reach的由来。2. Agent-Reach的整体设计思路把Agent协作抽象成三件事2.1 核心思想注册、发现、路由设计Agent-Reach的时候我一直在找一个足够简单的抽象。后来发现不需要发明新概念把Agent看成人就很好理解注册相当于每个人在通讯录里登记自己的名字、职位和擅长的事。Agent启动时把自己的能力描述上报给Agent-Reach中心。发现相当于查通讯录。调用方Agent告诉Agent-Reach我要找一个能做订单状态查询的AgentAgent-Reach返回匹配的结果。路由相当于接线员转接电话。调用方不直接连目标Agent而是把任务交给Agent-Reach由它按规则转给最合适的Agent并把结果带回来。这个三件事模型最大的好处是每个Agent只需要跟Agent-Reach打交道不需要知道其他Agent的地址、协议和实现方式。新增一个Agent对既有系统零侵入——它只要注册别人就能在发现阶段找到它下线一个Agent也只要反注册调用方那边不需要改任何代码。这个解耦的价值在多Agent项目规模从三五个扩展到十几个时体会最深。2.2 技术选型为什么自研而不是直接用现成的服务发现组件说实话当初我也纠结过注册、发现、健康检查这在微服务领域太成熟了Consul、Nacos、Etcd随便挑一个不就行了我甚至先拿Nacos搭了个原型三天就跑通了。但越往后梳理越发现两个关键差异让我必须在它们之上再包一层。第一Agent的能力描述比微服务实例复杂得多。微服务注册的维度通常是服务名实例地址健康状态而一个Agent的能力需要描述输入输出结构、参数示例、适用的意图标签、调用方式是同步还是异步、是否需要人工审批。这是一套领域语义直接塞进Consul的KV里会很别扭查询匹配也没法做——比如找一个能处理多语言并且支持批量翻译的AgentKV只能精确匹配做不了语义级查询。第二任务路由不是简单的负载均衡。微服务通常按服务名做轮询或一致性哈希Agent的任务路由往往要看意图匹配度、Agent当前负载、任务类型。比如一个翻译请求普通翻译Agent能做但如果是涉外法律文档可能得路由给专门做过法律语料训练的Agent。这种按能力语义路由负载均衡器做不了。所以最终的架构是Agent-Reach自己做核心的注册表、能力描述引擎和任务路由逻辑底层存储直接复用Redis做状态暂存消息投递用RabbitMQ。不引入太重的框架保持整个中间层足够轻方便其他团队快速接入。Redis和RabbitMQ在公司内部本来就是基础设施运维成本几乎为零这显著降低了Agent-Reach的推广门槛。2.3 目录结构与模块划分Agent-Reach我用Python重写了最早是个内部脚本堆出来的原型最终目录结构如下agent-reach/ ├── agent_reach/ │ ├── registry/ # 注册中心能力注册、实例管理、状态维护 │ ├── discovery/ # 发现服务能力查询、意图匹配 │ ├── router/ # 路由引擎任务派发、结果回传、重试策略 │ ├── protocol/ # 能力描述Schema、消息协议定义 │ ├── transport/ # 底层消息通道封装Redis/RabbitMQ适配 │ ├── sdk/ # 供各Agent接入的客户端SDK │ └── web/ # 管理控制台APIAgent列表、调用记录、拓扑图 ├── examples/ # 三个完整示例Agent客服、订单、翻译 ├── tests/ # 单元测试与集成测试 └── pyproject.toml模块职责上我有意做了严格区隔registry只负责记录状态不关心业务消息router只负责把任务从A送到B并带结果回来不做任何业务判断SDK是所有Agent接入的唯一入口避免各团队自己写HTTP调用导致协议漂移。这套设计可能不够炫但它在实际项目里的价值是出问题的时候我能快速判断问题出在哪个模块不需要翻完整个代码库。3. 关键实现细节从零搭建Agent互连层3.1 Agent注册与能力描述先统一你会什么这个问题Agent-Reach的核心协议是能力描述文件Capability Descriptor。每个Agent启动时通过SDK上报一份JSON描述自己的会什么。我最初设计这份描述时参考了OpenAPI和JSON Schema的哲学但刻意做了简化——不是每个Agent团队都有精力写完整的OpenAPI文档门槛一高大家就不愿意接了。一个实际的描述文件长这样{ agent_id: order-agent-01, name: 订单助手, namespace: commerce, version: 1.3.0, skills: [ { id: order.status.query, name: 订单状态查询, description: 查询订单当前物流与状态信息, intent_tags: [order_query, 物流查询, order_status], params: { type: object, properties: { order_id: {type: string, required: true, example: SO20250311001}, customer_id: {type: string, required: false} } }, response_schema: { type: object, properties: { status: {type: string, enum: [pending, shipping, delivered, cancelled]}, tracking: {type: string} } }, invocation: {mode: sync, timeout_ms: 5000}, auth_required: false } ], heartbeat_interval: 30 }这里每个字段都有用意intent_tags是给发现模块做语义匹配用的后面会细说invocation.mode决定路由器的调用策略同步任务挂起等待结果异步任务只回执接收auth_required标记该能力是否需要额外权限校验这直接影响了Router在派发前要不要走审批链路。注册流程本身很简单Agent启动后调用SDK的register()方法SDK做两件事把能力描述写入注册表同时启动一个后台心跳协程。这个流程必须是幂等的——Agent重启、网络抖动导致的重复注册不能产生脏数据。我在注册表里用(agent_id, skill_id)作为唯一键每次注册直接整组覆盖省去版本比对。3.2 健康检查与状态管理让可达这件事可度量Reach这个词在Agent-Reach里有两层含义一是发现层面的触达二是运行时层面的可达。后面这层就是健康检查要解决的问题。一个Agent注册了10个能力但它的进程已经卡死这时候它对外是不可达的路由器就不该再把任务派给它——否则调用方Agent会白白等超时。Agent-Reach的健康状态是三级模型UP心跳正常最近一次心跳在允许窗口内。SUSPECT错过了一次心跳暂时不派新任务但已有任务继续执行。DOWN连续错过心跳超过阈值从可用列表中剔除并触发告警。这个模型借鉴了分布式系统里常见的怀疑而非直接判定死亡的思路因为Agent处理长任务时可能阻塞心跳线程直接判DOWN会导致任务无故中断。SUSPECT状态给了系统一个缓冲。实现上心跳线程在SDK内独立运行与Agent的业务逻辑隔离避免业务阻塞影响心跳上报。这个设计后来救过我们一次某个Agent跑一个20分钟的报表任务时主线程完全阻塞但心跳线程正常任务没有被误判中断。3.3 消息路由与任务派遣一个请求是怎么被送到目标Agent的接下来是重头戏——路由。一个调用方Agent发起任务请求完整路径是这样的调用方SDK把请求格式化为标准Task消息包含task_id、target_skill可以是精确能力ID或意图描述、payload、callback字段。发现模块先做能力匹配。如果是精确ID直接查表如果是意图描述比如帮我找个能翻译的Agent则对注册表中的intent_tags做文本相似度匹配我用的方案是预计算标签的向量化表示查询时用余弦相似度排序取Top-3返回。路由引擎从匹配结果中过滤掉状态不是UP的Agent然后根据策略选出一个目标。默认策略是可用性优先即只选UP且当前积压任务数最少的也可以按Agent配置的权重做加权选择。任务投递到底层消息通道。同步调用走RabbitMQ的RPC模式即发送消息后等待reply队列异步调用直接投递到目标Agent的work queue回执即返回。回传结果同样走SDK封装的reply机制附上task_id调用方按task_id追踪。这里的核心设计是调用方永远不需要知道目标Agent的地址。哪怕订单Agent从节点A迁移到节点B只要它注册的agent_id不变对调用方来说就毫无感知。这彻底解决了最初接口文档过时、地址写死的问题。另外我特意在Task消息里保留了task_id的全链路透传这让后面排查问题尤其是第4.4节那种任务风暴变得轻松很多。3.4 与主流Agent框架的适配怎么让已有Agent快速接入很多团队已经有基于LangChain、Dify、FastGPT或者自研框架跑起来的Agent不想为了接入Agent-Reach重写。所以我把SDK设计成一个薄适配层接入成本控制在最低。这个判断非常重要——如果接入成本超过半天团队就不愿意用了。以我们内部一个基于LangChain写的报表Agent为例接入只需要三步from agent_reach.sdk import AgentReachAgent # 1. 实例化传入能力描述文件路径 reach_agent AgentReachAgent( descriptor_pathcapabilities/report-agent.json, registry_urlredis://registry.internal:6379/0 ) # 2. 把原有业务函数注册为可暴露的能力 reach_agent.skill(report.generate) def generate_report(params: dict) - dict: # 这里直接复用原有LangChain chain的逻辑 return chain.invoke(params) # 3. 启动 reach_agent.start()原有的业务逻辑完全不用动只是一个装饰器把函数暴露为能力。如果Agent需要反过来调用别的AgentSDK也提供了call_skill()方法result reach_agent.call_skill( skillorder.status.query, payload{order_id: SO20250311001}, modesync, timeout5.0 )写SDK时我踩过一个有意思的细节装饰器的函数签名和真实JSON参数之间怎么映射。很多Agent团队的函数参数是Python风格下划线命名而能力描述用的是JSON字段可能是驼峰或业务命名。如果直接生硬映射调用方传customerId而函数参数是customer_id就会莫名其妙报错。我在SDK里加了参数名映射配置装饰器声明时可以用reach_agent.skill(report.generate, param_map{start_date: startDate})显式转换避免每个团队在业务代码里做脏活。4. 实测结果与踩坑记录上线两周遇到的三件事4.1 先看压测数据Agent-Reach内部跑了两周后我们在测试环境做了简单压测。部署规模是8个Agent实例、12个能力Skill、1个注册中心节点Redis、1个路由器节点RabbitMQ Router进程。压测工具用Locust模拟20个并发调用方Agent持续10分钟覆盖80%同步调用和20%异步调用。结果数据如下指标数值任务成功率99.6%同步调用P95延迟312ms其中业务耗时约80%同步调用P99延迟680ms附加的路由开销35~45ms注册表稳定运行时间2周无重启路由本身的开销控制在几十毫秒级别对于Agent间协作来说完全可以接受。真正的耗时大头还是Agent内部的模型推理和业务逻辑。这也验证了一个判断互连层不应该做太多复杂逻辑它的核心竞争力是稳定和轻量而不是计算。4.2 踩坑一能力描述文件的版本兼容问题第一个坑出在能力升级上。订单Agent把order.status.query的响应里tracking字段从字符串改成了对象包含carrier和number。升级当天客服Agent的调用直接报错——它还在按字符串解析。这是典型的能力版本漂移。排查链路是这样的先看客服Agent日志报错是解析tracking字段时TypeError: dict object has no attribute startswith定位到是订单Agent响应格式变了。再看Agent-Reach的调用记录确认调用方版本和提供方版本都没有被记录。问题根源在于注册表只存了当前能力描述没有存历史的兼容版本信息。修复方案分两层。短期是回滚订单Agent响应格式长期是在能力描述里加了deprecated标记和compatibility_route字段如果提供方要改响应必须先在描述文件里标记旧字段为deprecated登记录入时Agent-Reach会在调用方SDK侧做一次响应格式检查并给出明确报错而不是让调用方在业务代码里稀里糊涂地崩。现在我们的约定是能力描述文件的变更必须走MR评审且变更前要在测试环境模拟旧版本调用方跑一遍回归。这个流程看着繁琐但对比一次线上事故的排查成本完全值得。4.3 踩坑二任务回执的确认机制第二个坑和可靠投递有关。上线第二天运营反馈改地址的任务偶尔没生效。查日志发现路由器的确认机制不够严格。最初的逻辑是router把任务投递到RabbitMQ队列后只要RabbitMQ确认了消息接收router就认为投递成功订单Agent消费消息并处理后也会发一个执行完成的reply。但问题在于订单Agent在处理过程中如果崩溃消息已经消费掉了没有做ack任务丢失调用方Agent还一直等reply等到超时。表现就是任务不见了也没有回执。这个在消息投递里是经典的at-most-once问题。修复方案是按at-least-once重构订单Agent消费消息时不立即ack而是先执行业务逻辑把结果写入自己的持久化存储再ack并发送reply。如果Agent在处理中崩溃消息会被重新投递配合业务侧的幂等键用task_id去重保证任务不丢不重。这个坑也让我反思Agent任务协作和普通消息消费最大的不同在于任务是有业务语义和状态流转的不能简单当普通消息处理。现在的Agent-Reach里每个Task都有一个完整状态机CREATED → DISPATCHED → RUNNING → SUCCEEDED / FAILED路由器和调用方SDK都会记录状态迁移排障时直接看task_id的状态历史比翻日志高效得多。4.4 踩坑三Agent循环调用导致的任务风暴第三个坑最惊险。某天下午注册表里的Agent-Reach路由器CPU飙到90%RabbitMQ队列积压了几万条消息。查看调用拓扑发现客服Agent在回答用户问题时调用了订单Agent订单Agent内部又调用了翻译Agent翻译Agent为了理解上下文又调用回客服Agent……三个Agent形成了一个调用环。每个用户问题都会触发一轮循环每条消息又会产生新的子任务任务量指数级爆炸。修复的思路分两层第一层是调用深度限制。在Task消息里加max_depth字段默认5跳路由器每转发一次就减1减到0直接拒单并返回调用链过深错误。这一层挡住了环路的无限扩张。第二层是环路检测。路由器在派发任务时在Task消息里附加一个trace_chain记录这条请求已经经过的agent_id列表。如果发现目标agent_id已经在trace_chain里说明发生了循环调用直接拒绝并告警。这个方案类似RPC框架里的分布式链路追踪的变体实现成本很低但非常有效。任务风暴之后我们加了一个监控看板按agent_id聚合的每分钟接收任务数和平均任务深度。这两个指标能提前暴露循环风险。说实话如果没有这次事故我不会意识到多Agent协作的图一旦带上环会比普通微服务调用更容易失控——因为每个Agent都有自主决策能力它会自己决定要不要再调下一个Agent。5. Agent-Reach还能怎么玩三个实用扩展方向5.1 把Agent-Reach接到RPA和其他外部系统上Agent-Reach本身不限定于大模型Agent之间协作。它的能力描述协议足够通用完全可以把RPA机器人、定时脚本、甚至老旧的核心系统包成一个Agent注册进来。我们内部后来就把一个RPA机器人接了进来——它注册了一个erp.invoice.export的能力由数据分析Agent在生成月报前自动触发。这让Agent-Reach实际上变成了企业内部系统的统一能力网关而不仅仅是Agent之间的小圈子。这一步扩展的思路其实很简单能力描述协议定了接入方是谁反而不重要。5.2 支持异步长任务与人工审批介入Agent的实际任务经常不只是查一下这种同步小操作还有导出三个月的数据报表这种长任务。Agent-Reach的异步模式支持这种场景但更实用的是人工审批介入某些能力比如订单退款、修改核心数据必须有人类确认。我们在能力描述里加了approval_required字段路由器遇到这类任务会先把任务挂起到待审批状态推送到审批台审批通过后才真正派发给目标Agent。这个扩展把Agent的自主性和安全边界揉在了一起——让Agent能干活但关键的活必须有人兜底。5.3 权限细粒度控制的思路最后提一下权限。早期版本的Agent-Reach是信任内网的但接入RPA和外部系统之后权限必须做细。我们的做法是在能力描述里增加access_policy字段结合统一的token做调用方身份识别再结合每个Agent按调用方身份能力ID维度做授权。这个改动不大但能把谁能调用谁的规则从业务代码里抽离出来变成Agent-Reach的安全策略配置。如果你计划把Agent-Reach对外开放这一步建议提前做不要等出了安全事故再补。最后聊聊我的体会。Agent-Reach整个项目的出发点其实特别朴素Agent之间要能互相触达先把你会什么、我在哪、任务怎么交接这三件事标准化。技术选型上我没有追求新颖的协议或者复杂的编排引擎而是把注册、发现、路由这三个成熟概念用Agent的语义重新组织了一遍。事实证明这个老概念新包装的思路在落地时阻力最小——团队接入的意愿高因为接入成本几乎只有SDK一行初始化。如果你也在折腾多Agent协同我建议别一开始就想做一个万能编排平台。先把触达做好能让一个Agent发现另一个Agent能把一个任务可靠地送到它手上能把结果带回来就已经解决了80%的问题。剩下的编排、审批、权限都是在这个地基上长出来的。一个小技巧收尾能力描述文件里的intent_tags一定要让业务团队填而不是技术团队编。业务人员给的词帮我查个单翻译这段话才是真实用户会说的Agent-Reach的语义匹配和后续的编排规则全靠这批词的质量。这一步做扎实了后面会省很多事。