今年以来AI Agent的热度一直没降过我自己也陆续折腾过不少多Agent协作的项目但真正让我卡住很久的不是某个模型的能力上限而是Agent之间怎么找到对方、怎么把任务递过去、怎么确认对方真的收到了。说白了就是一句话Agent的触达问题。后来我花了两周时间做了个小框架代号就叫Agent-Reach专门解决这个事。这篇文章把当初的设计思路、核心模块、部署踩坑和排查经验完整记录下来希望对同样在做多Agent系统的朋友有点帮助。标题里的“Reach”我取了“触达”和“可达”这两层意思。Agent-Reach做的事情可以概括成三点让一个Agent知道外部有哪些能力可以用让Agent能够把任务请求准确送到目标Agent手上以及在传递过程中保证状态可追踪、结果可返回。这不是一个Agent内部怎么推理的问题而是Agent与Agent、Agent与外部工具之间怎么通信的问题。适合正在搭建多Agent系统、做工具编排、或者想从单Agent往多Agent架构演进的同学参考尤其是被“服务发现”和“消息路由”这两个词反复折磨过的人。1. 项目背景与整体设计思路1.1 单Agent到多Agent最先缺的不是模型能力很多团队做AI应用起步都是一个Agent挂一堆工具。模型负责理解意图函数调用负责执行动作跑通几个场景之后看起来不错但一旦任务变复杂单体Agent就会暴露出几个很尴尬的问题。第一上下文窗口再大也装不下全部工具描述。工具一旦超过几十个光是把工具声明塞进prompt就已经占掉大量token更别说还要让模型在这么多工具里精准选。第二不同系统的Agent根本没法互相调用它们各自活在各自的进程里A系统想用B系统的能力得专门写接口。第三没有任何“服务方”的意识。一个Agent调用工具是主动行为但如果反过来另一个Agent想把任务塞给它它连个接收入口都没有。我遇到的具体场景是这样的我这边有几个Agent分别负责数据分析、文档生成和外部API请求。当时想把它们串成一个自动化的调研链路但很快发现最简单的一步——让数据分析Agent把结果发给文档生成Agent——就不知道怎么实现。两个Agent的代码是独立的数据得靠我自己在中间搬运完全没有“让Agent自己去找协作方”的能力。所以Agent-Reach的核心思路特别朴素给每个Agent一个“地址”让它们像微服务一样注册到中心然后通过一个轻量协议互相发消息。你不需要把Agent拆成微服务那么重只要引入“注册”和“路由”这两个概念就行。1.2 Reach的三层含义工具、数据、Agent互达整个项目我从一开始就想清楚了一个模型触达能力分成三层三层的实现方式也完全不同。第一层是Agent对工具的触达也就是函数调用。这一层现在各大模型都原生支持了重点是做好工具描述和参数校验。第二层是Agent对数据的触达包括数据库、文件系统、内部知识库。这一层需要的是连接器和权限控制。第三层才是我真正想做的Agent对Agent的触达。这一层是静态工具描述解决不了的因为协作方不是固定的函数签名而是运行时动态出现的另一个Agent。Agent-Reach架构上最大的决定就是把第三层做成基础设施而不是业务代码。所有Agent接入时不需要关心对方是谁、在哪、用什么语言写的只需要把自己的能力注册上去然后向中心发消息说“我需要某类能力”中心帮你找到合适的接收方并完成投递。这个思路其实跟多年前微服务里的服务注册与发现是一个逻辑只不过服务注册发现的对象从“接口”变成了“Agent”从“RPC调用”变成了“带任务语义的消息”从“负载均衡”变成了“能力匹配路由”。理解了这一点整个项目剩下的设计就顺理成章了。1.3 技术选型为什么不用现成的消息队列和RPC框架做之前我也犹豫过是不是直接用RabbitMQ或者gRPC就能解决问题。评估一圈之后发现传统消息队列解决了消息投递但解决不了“按能力找Agent”。你要自己维护一套Agent能力目录还要自己写匹配逻辑。而RPC框架解决的是点对点通信但每个Agent变成本地接口Agent的独立性又没了。大家都在谈Agent的后端基础设施Agent-Reach的做法其实很朴素在“找”这个问题上用一个小型注册中心把Agent的能力和状态管起来在“递”这个问题上用一个自描述的消息结构把任务原文、来源、目标、回调地址和上下文包在一起。注册中心只是一个轻量服务核心逻辑在Agent侧的SDK里。1.4 整体架构与核心概念Echo Agent-Reach的整体结构不复杂大概分成四部分Reach Registry注册中心负责Agent注册、能力登记、心跳检测。Reach Router路由器根据任务消息里的意图描述匹配到有对应能力的Agent。Reach Agent SDK嵌入在各Agent内部处理注册、发送、接收、回执。Reach Message消息协议统一的任务消息格式贯穿整个链路。每个Agent接入的时候启动就在Registry注册自己的Agent ID、能力列表和回调地址。发任务的时候SDK把任务封装成Reach Message发给RouterRouter根据消息里声明的Capability和Constraints做匹配找到目标Agent之后投递过去。目标Agent处理完按原路返回Result。2. 核心设计思路解析2.1 注册中心轻量但必须有的状态管理有人可能会问Agent-Reach为什么非得要一个注册中心去中心化不行吗确实去中心化方案在理论上存在像点对点的gossip协议也能让节点之间互相发现。但我选择“要一个轻量级注册中心”的原因很现实Agent系统的节点数量通常只有几十上百远远达不到分布式网络需要去中心化的规模。而且注册中心还能集中承载“能力语义”的维护能力这在中心化架构里做起来最简单。每个Agent接入时需要提供三块核心信息Agent ID全局限一的名字我习惯用agent-类型-模块-编号这种格式方便日志里快速定位。Capabilities这个Agent能做什么的能力声明。比如数据分析Agent会写data.analysis和data.visualization文档Agent会写document.generate和document.review。Metadata附加信息例如运行环境、可用时段、当前负载、额外参数给路由阶段做约束过滤用。注册中心在SDK里执行心跳检测。每个Agent每10秒上报一次状态如果连续3次不上报就从可用列表里摘掉不再往里路由任务。2.2 消息协议自描述、带路由信息、可追踪Reach Message是我在这个项目里花最多心思的部分。它是一份JSON格式的数据结构关键字段如下message_id全局唯一的消息ID所有追踪机能都依赖这个字段。source_agent_id和target_agent_id来源和目标Agent ID。如果目标不确定就把target_agent_id留空。intent用一句话说明任务目标这是路由匹配的主要依据。payload任务的具体内容比如数据分析的原始数据、文档生成的大纲等。callback_address结果返回的地址默认回推给来源Agent。context可选字段用于存放路由过的节点信息和全局唯一的追踪ID。created_at和expire_at消息创建和过期时间实现延迟清理。设计取舍在于很多Agent消息协议喜欢定义得非常抽象结果两端还要再对一次schema。我反而让reach消息就是一个低门槛的JSON字段含义在SDK文档里说明这样接入成本降到最低。2.3 路由策略能力匹配不为调度Router在Agent-Reach里的作用不是一个负责任的“调度器”而是“能力匹配器”。它不决定任务分配给哪个实例只决定消息能发给哪个具备相应能力的Agent。匹配逻辑分成两级第一级硬匹配。要求Capability严格匹配。如果注册中心规定data.analysis只能被数据分析Agent声明就不会把这类任务路由给别的Agent。这个匹配强制做能力名的一致性。第二级软过滤。Constraints里的Metadata过滤在硬匹配结果基础上执行例如可用时段、最低版本要求。在这个阶段还要避免把消息发给当前负载过高的Agent虽然这里不在Agent-Reach默认路由里做智能负载均衡但会优先选择当前负载最低的那个可用实例。路由整体是同步的。Agent SDK发送消息后同步阻塞等待Router的投递结果如果Router在5秒内找不到可用目标就返回路由失败错误。这个错误的选择对AI流水线很重要因为Agent往往需要在消息里知道该任务是否真的送给了下一个Agent而不是直接丢给对端就结束了。2.4 SDK的设计取舍侵入性控制在“极简”线附近Agent接入这个体系最怕的就是SDK太重。我在做SDK的时候反复提醒自己一句话接入一个Agent应该像发一个HTTP请求一样简单而不是像改造成微服务一样复杂。最后SDK的核心API收敛到四个方法register(agent_id, capabilities, metadata)启动时注册。send(intent, payload, target_agent_id)发任务。on_task(handler)注册一个处理函数接收Reach Message并返回结果。reply(message_id, result)在handler里返回结果后SDK自动向回调地址发送结果。这样设计的意图很明确Agent业务代码不需要关心消息是怎么路由的也不需要关心注册中心在哪。SDK内部承载了本地缓存、重试、心跳上报等逻辑。这里给一个SDK接入的伪代码示例展示一个Agent侧的样子from reach_sdk import ReachAgent def handle_analysis(data): # 这里是Agent自己的业务逻辑 results run_stats(data) return {status: success, results: results} agent ReachAgent( agent_idagent-analysis-001, capabilities[data.analysis, data.visualization], metadata{load_threshold: 0.8} ) agent.on_task(handle_analysis) agent.start() # 注册 心跳 监听接收2.5 为什么要做消息回执而不是只发不响我在第一个版本里吃过亏。消息发过去就算了目标Agent处理成什么样完全不知道。等到后来的Agent要依赖前序结果时才发现数据是空的根本没法串联整个工作流。Echo Agent-Reach在消息机制里引入回执就是从这次踩坑开始的。Agent收到消息后必须先返回一个RECEIVED回执确认自己收到消息处理完再返回SUCCESS或FAILED结果。这个机制既是链路追踪的依据也是让Agent自然形成“状态机意识”的关键。有了路由回执、接收回执、结果回执整个Agent协作链路每一步都是可追踪的。这套机制做出来后我再也没因为中间某个环节静默失败而大海捞针地查日志。3. 实操过程与核心环节实现3.1 部署一个最小可用的Agent-Reach环境这里直接给出我实测可行的部署步骤。在本地跑起来的话只需要一台装好Docker的开发机器就行。先部署Registry和Router这两个核心组件。Registry我在项目里用的是FastAPI写的Router也一并打包成了同一个Docker镜像里的不同入口。生产环境可以把它们分开部署但本地调试阶段放一起没毛病。# 拉取镜像这里用我打好的开发镜像做示范 docker pull reachdev/agent-reach:0.3.0-dev # 启动Registry默认监听8001端口 docker run -d --name reach-registry -p 8001:8001 reachdev/agent-reach:0.3.0-dev registry # 启动Router默认监听8002端口连接Registry docker run -d --name reach-router -p 8002:8002 reachdev/agent-reach:0.3.0-dev router --registry http://localhost:8001环境变量配置在.env文件里除了组件地址之外还有一个必须认真设的参数是message_expire_seconds我设的是300秒。默认300秒的超时时间是基于经验判断的一般Agent处理任务很少超过几分钟超过之后基本可以认为出了意外。这个时间不是越长越好太长了会让无效任务堆积。3.2 编写一个多Agent触达的完整示例最小可行链路我选了三个Agent一个负责“取数”一个负责“分析”一个负责“汇报”。三个Agent分别部署成三个Docker容器通过Agent-Reach串联起来。第一步先把三个Agent注册好。每个Agent都是一个简单的Python进程接入SDK后启动注册会在进程启动时自动完成。# collect_agent.py from reach_sdk import ReachAgent def handle_collect(meta): data fetch_from_source(meta[source]) return {status: success, data: data} agent ReachAgent( agent_idagent-data-collect, capabilities[data.source.collect], metadata{region: cn-east} ) agent.on_task(handle_collect) agent.start()第二步分析Agent注册两个能力一个接收采集来的原始数据一个产出分析结果。注意这里它接收的数据来源不是某个固定Agent ID而是通过消息的source_agent_id字段辨认来源。def handle_analyze(msg): source msg.get(source_agent_id) data msg.get(payload, {}).get(data) # 业务逻辑分析数据、生成统计、输出图表的描述JSON result run_analysis(data) return {status: success, analysis: result}第三步汇报Agent直接订阅分析Agent的结果。这里的订阅不是标准的消息队列订阅而是通过Agent-Reach的消息回执机制分析Agent回执成功汇报Agent就自动拿到结果并生成最终报告。整个链路串起来我会在Router的监控面板里看到三条消息流第一条是采集到分析的data.source.collect第二条是分析到汇报的data.analysis.done第三条是所有回执的累计视图。这个视图做出来之后整个协作链条在脑海里就特别清楚了。3.3 路由规则实现细节从“只会发广播”到“按能力精确投递”最早做路由时我偷懒用的是广播的形式一条消息发给所有Agent谁有能力谁处理。能跑通但问题很明显所有Agent都会被唤醒token开销、计算开销都是白白的浪费而且结果不可预期。后来经过几轮演进路由落到了三个步骤上第一步能力字典匹配。Router内部维护了一张Capability - AgentIDs的映射表这张表由注册中心实时推送匹配耗时是毫秒级的。第二步Metadata过滤。根据Constraints做过滤比如data.region: cn-east这种要求把不符合条件的Agent剔除。第三步负载评分。Router在心跳数据里附带current_load字段Agent自己上报当前排队任务数量。Router在候选池里选负载最低的。这套三步走的路由机制做出来以后链路平均延迟降了一半以上消息量和唤醒次数大幅下降。这里给一个路由规则的YAML配置示例routes: - name: analysis_pipeline match: intent_contains: analysis target: capability: data.analysis constraints: region: cn-east min_version: 1.2 - name: report_pipeline match: intent_contains: report target: capability: document.report timeout_seconds: 603.4 回执机制与可追踪性的落地实现每当一个Agent收到消息SDK内部会立即自动向Router发送一条RECEIVED回执。这个不用业务代码管SDK自动做的。业务代码的handler运行完之后返回值会由SDK封装成SUCCESS结果发回给消息里的callback_address。如果handler抛异常了SDK会捕获到异常信息转成FAILED结果返回。追踪的关键是消息里的trace_id在链路的每个节点里都会原样携带。排查问题的时候直接拿trace_id去Registry的日志接口一查整条链路谁接的、谁回的、哪个环节慢一目了然。这里分享一个排查耗时问题的实用命令示例我经常用它看链路耗时分布curl -X POST http://localhost:8001/debug/trace \ -H Content-Type: application/json \ -d {trace_id: reach_trace_20241124_001} # 返回每个节点的接收时间、处理时间、回执时间以及耗时分布这个命令是我在联调时最依赖的工具。没有它多Agent联调基本等于摸黑前进。4. 常见问题与排查技巧实录4.1 注册成功但路由不到目标Agent这个问题我遇到太多次了场景基本都是Agent明明在Registry里显示在线但Router就是找不到能投递的目标。排查思路建议往三个方向走。第一确认能力名大小写和命名空间完全一致。Data.Analysis和data.analysis在我这套路由规则里是严格区分大小写的拼写对不上就永远匹配不上。第二确认Metadata过滤条件是不是卡住了。如果要求region是us-west但Agent注册的是cn-east也会导致匹配失败。第三确认注册中心同步路由器缓存有没有延迟。我遇到过Router内存里的能力映射表更新慢导致新注册的Agent要等若干秒才能参与路由。自查的时候建议先直接看注册记录再试一次显式指定target_agent_id的发送方式。如果指定ID能送达而按能力发送不生效基本就是路由规则问题。4.2 消息发送超时但Agent侧显示在正常运行这是典型的“假在线”问题。心跳机制默认10秒一次但如果Agent在处理一个耗时很长的任务心跳线程其实还会继续工作。所以Agent看起来在线其实已经没有窗口接收新任务了消息进了堆积队列等前面的处理完才能轮到后面的。我的SDK里默认给每个Agent开了个10个槽位的并发接收窗口超过10个任务排队就直接返回BUSY回执避免任务排队越积越多。如果你的场景里任务普遍偏重建议做两件事一是调高并发窗口限额二是把耗时长的任务改造成“先收下、立刻回执、后台异步处理、完成后再回结果”的模式。这个模式下Agent接收和处理的解耦做得干干净净整个系统的吞吐量会提升一个台阶。4.3 回执丢了导致上层Agent卡在等待状态这种问题的典型表现是A发给BB处理完了结果也回推了但A一直认为任务没完成。造成回执丢失的原因往往出在回调地址上。如果回调地址用了Agent注册时生成的内网地址而从Router来的回执是从另一个网络环境发起的就会存在网络不通的情况。我自己做示例环境时最常踩的就是这个坑。解决办法是在SDK启动时增加一个public_callback_base配置项显式指定从外网可访问的回调地址前缀。另外务必在消息协议里加上超时重发机制。我目前采用的是指数退避重发超时后1秒、2秒、4秒、8秒逐步重试最多重试5次。如果重试次数耗尽还没收到回执就标记该任务失败并通知上层Agent做兜底处理。4.4 多版本Agent混跑导致路由到旧逻辑节点升级Agent的时候旧版本容器没有及时下线Rrouter可能会依然把消息路由给旧版本节点。旧节点虽然没有崩溃但处理结果可能不符合新逻辑预期这是多Agent系统里相当隐蔽的坑。我在Agent-Reach里引入了min_version约束配置在路由规则里指定最低版本号这样Router会把低于设定版本的Agent过滤出候选池。同时在Agent Metadata里额外加一个deprecated标记计划下线的Agent可以标记为“不参与新任务”让存量任务处理完后再彻底下线。4.5 排查问题必备从日志、指标到链路追踪做多Agent系统必须在一个地方看到所有Agent的实时状态否则任何问题排查都像海底捞针。我在Agent-Reach里集成了一个简单的健康看板列出注册的Agent数量、在线状态、各Agent的负载、消息积压量、路由失败数和平均处理耗时。这些全部来自Registry的API前端轮询就能展示。日志规范我用的是JSON结构化格式核心字段包括time、level、trace_id、source_agent_id、target_agent_id、event、message。有了这些维度的日志即使不做全链路追踪平台用最普通的日志检索也能顺畅地定位问题。5. 踩坑心得与后续规划5.1 如果重来一次我会做什么不同这个项目越做到后面我越觉得很多地方做了过度设计也有不少地方吃亏在了“想得不够远”。如果从头再做一遍我第一件会改的事是把Agent的能力声明规范从固定枚举改成带语义的版本化Schema。当初做能力名匹配图的是简单直接。但能力名冒烟测试太少一旦任务复杂起来光靠一句话很难精准描述能力边界后面维护能力目录的成本越来越高。如果当初就引入JSON Schema来描述能力输入输出后面路由匹配的精细化程度会高很多。第二件会改的是把Webhook回调这种“单向通知”的形态改成支持双向流式通信。现在Agent之间发消息是一问一答但对一些持续性的场景——比如一个Agent实时产出分析结果、另一个Agent持续汇总——问答式交互根本不够用。将来可以支持通过同一个connection_id维持长连接让数据持续流动该有事件的场景有事件该有取消机制的地方有取消机制这个方向我想得很清楚了。5.2 目前还在迭代中的扩展方向短期内的规划有三个。第一个是动态Tool引用让Agent注册的能力可以直接引用外部工具而不用把工具描述写在自身代码里。第二个是联邦注册模式让多个体系的多个注册中心之间可以互相转发消息把触达范围从单系统扩展到整个组织。第三个是多模态Payload支持目前消息的payload以JSON为主后续会加上对大文件、图片、表格的直接引用避免把大对象塞进消息体。个人最看重的其实是第二个方向。Agent的数量一旦过了一定规模跨部门、跨系统的协作会成为一个重要的突破点。就像以前微服务化进程一样Agent协作也要先有“地址簿”和“路由规则”否则数不清的Agent只会变成新的信息孤岛。框架本身是工具工具可以粗糙但结构必须清晰。我现在还在持续打磨这套东西的易用性和可观测性希望它能真正成为Agent协作“最后一公里”的通讯基础设施。