Agent-Reach这个名字第一次看到的人容易摸不着头脑——Agent我懂Reach是什么鬼其实它解决的是多智能体系统里一个极其普遍、却又常被忽略的问题触达。你可以想象这样一个场景系统里跑着客服Agent、订单查询Agent、优惠计算Agent、库存预测Agent每一个单独拎出来都挺聪明但把它们放到同一个业务链路上麻烦就来了——一个新上线的Agent怎么让其他Agent知道用户抛过来一个模糊请求到底该让哪个Agent去接Agent正在高负载或者已经下线了分发系统知道吗这些问题统称为Agent之间的“触达”。我这段时间完整落地了一遍Agent-Reach的方案设计把能力注册、意图路由、消息投递、结果回传这一整条链路都跑通了踩了不少坑也沉淀了一套可以复用的方案。这篇文章就把整个拆解过程和实操细节全部摊开适合正在做多Agent编排、或者被“Agent一多就乱”困扰的团队参考。1. 项目整体设计与思路拆解1.1 Agent一多“触达”为什么会失效先说一个最直观的现象。Agent数量少的时候两三个Agent靠if-else就能调度清楚但规模一旦上来问题立刻变味。我见过一个实际业务里的调度代码光路由逻辑就堆了上千行全是散装的if条件如果关键词包含“退款”就走客服Agent如果包含“库存”就走库存Agent再叠加什么会员等级、地域、时间段的判断简直是蜘蛛网。这种硬编码路由至少有四个致命伤。第一是能力不透明。老Agent根本不知道新Agent能干什么新能力上线后要人工通知、人工改代码业务方忘了同步就永远没人调用它。第二是状态不同步。Agent本身是有运行状态的可能在过载、在维护、甚至已经宕了但路由方对此一无所知照样把请求往那儿塞。第三是语义缺失。用户的话往往含糊不是每个请求都带着“退款”“库存”这种明确关键词硬编码路由对模糊表达基本无能为力。第四是回传混乱。请求发给Agent之后执行到哪一步了、成没成功、失败原因是什么全链路没有统一跟踪出了问题只能靠翻日志猜。这里有个特别值得强调的点传统开发里模块间的调用关系是编译期就确定了的A方法调B方法写死就完了。但Agent是运行时实体它的能力边界、健康状态、上下文能力都是动态变化的。静态的路由思路天然就不适配动态的Agent体系这就是需要Agent-Reach这类触达层的原因。1.2 三层架构把“触达”这件事拆干净我一开始设计的思路很简单想做一个大网关所有请求进来由网关统一调Agent。但写着写着就发现不行——网关里要干的活太多了既要理解用户意图又要管Agent的注册表还要处理消息投递、重试、结果回传全塞一起的话任何一个模块变更都要重新发布整个网关耦合太严重。后来参考微服务注册中心的设计思路把架构拆成了三层触达层、调度层、通道层。触达层干的事比较纯粹接收外部请求做意图识别和初步校验然后把“要干什么”这个语义描述交给调度层。它本身不依赖任何具体Agent的信息所以可以随便横向扩容。调度层是整个方案的大脑它维护Agent注册表、能力目录、负载状态根据触达层传来的意图做路由决策选出最合适的Agent列表。通道层则负责把任务真正投递出去管理HTTP调用或MQ消息的发送、超时重试、结果回传把“调谁”和“怎么调”彻底隔离。这三层的数据流是单向清晰的请求先进触达层触达层拿着语义描述去问调度层“谁能干这活”调度层查注册表、打分排序后返回目标Agent列表触达层再让通道层去执行投递Agent处理完的结果同样通过通道层回传。每一层只对上一层负责修改任何一层都不影响其他层这在实际维护中的收益非常大。1.3 为什么不直接用消息队列硬扛有人可能会问RabbitMQ、Kafka这些消息中间件不是已经能解决投递问题了吗何必再造一套我的回答是消息中间件解决的是“消息不丢不重”的投递问题但它完全不理解“语义”。你往Kafka里丢一条消息它根本不知道这条消息应该由哪个消费者处理还是得靠你自己写路由规则。而且消息队列的方案里Agent的注册发现、能力匹配、负载感知、意图理解这些核心逻辑一样都省不掉等于该造的轮子一个没少还得额外维护一套MQ基础设施。我把两个方案的核心差异整理成了一个对比表方便大家直接参考。对比维度纯消息中间件方案Agent-Reach触达层方案语义理解不支持消息就是字节流支持意图识别与语义匹配能力发现无需要手动写清消费者映射自动注册、动态发现能力负载感知无消费能力全靠消费者自己声明记录负载状态并参与路由打分兜底策略需自研内置降级、多Agent编队实现复杂度低但路由逻辑全堆在外面中等但逻辑内聚、可复用所以我的结论很简单如果只是做异步任务解耦MQ完全够用别折腾。但如果核心痛点是Agent之间的发现与路由那必须有一个懂语义的触达层Agent-Reach的思路就是针对这个问题设计的。2. 核心模块解析与关键技术细节2.1 能力注册表Agent的“身份证”怎么设计整个Agent-Reach方案的地基是能力注册表。每个Agent上线时必须先在系统里登记之后调度层才能按“能力”而不是按“名字”来找它。这块设计得不好后面路由再花哨也是空中楼阁。我设计的注册信息包含几个关键字段。agent_id是全局唯一标识endpoint是Agent的实际调用地址capabilities是一组能力标签比如“订单查询”“优惠计算”“退款处理”这是路由匹配的主维度。除了标签我还要求每个Agent提交一段自然语言能力描述比如“我能查询订单状态支持按订单号、手机号查询能判断是否超时发货”这段描述会在意图识别阶段做语义相似度计算弥补标签过硬的缺陷。另外还有几个动态字段负载水位当前并发数/最大并发数、健康状态在线/忙碌/离线、可用时段、上下文窗口大小。注册和续约的流程借鉴了服务注册中心的心跳机制。Agent启动时调用注册接口把上面这些信息登记进来然后每隔一段时间我用的30秒发送心跳续约。调度层有一个后台任务在盯着所有Agent的最近心跳时间超过90秒没收到心跳就标记为离线并从候选路由池里摘掉等它恢复心跳后再自动加回来。这套机制保证路由决策永远基于Agent的最新状态而不是停留在它上线那一刻的快照。这里有个我特别想强调的细节能力描述一定要让业务Agent的开发方自己写不要代劳。我之前试过让平台方统一给Agent写描述结果写出来的全是标准化套话语义相似度计算效果很差。后来改成让每个Agent的开发方用自己的话描述能力反而匹配得特别准——因为只有开发方自己最清楚Agent能干什么、擅长干什么。2.2 意图识别与路由打分怎么让请求找到对的Agent触达层收到请求后第一步是意图识别。我的做法不是一上来就做复杂的LLM分类而是先用一个轻量的分类模型提取意图标签和关键实体比如“帮我查一下上周买的手机到哪了”识别出意图是“物流查询”、实体是“手机”然后再把“意图标签 原始文本”一起送进调度层做路由。调度层的路由决策分两步候选集过滤和打分排序。过滤阶段先把不满足硬性条件的Agent排除掉——比如意图标签完全不匹配的、健康状态离线的、不在可用时段的、上下文窗口装不下当前请求的。排除之后剩下的Agent进入打分环节我用的是一个加权公式score 技能相似度 × 0.5 历史成功率 × 0.3 负载余量 × 0.2技能相似度是意图标签和Agent能力标签的匹配程度加上能力描述的语义相似度辅助历史成功率是过去24小时这个Agent处理同类请求的成功比例负载余量由Agent当前并发数和其声明的最大并发数计算而来。三个维度加权汇总后选得分最高的一个或者前几个作为目标。如果得分都低于一个阈值我设的0.35就触发兜底策略返回提示信息或者转人工处理而不是硬塞给一个不合适的Agent——硬塞的结果往往是答非所问比不答更糟。可以打个比方这套机制就像外卖平台点餐先根据你选的美食分类筛掉不相关的店铺再综合评分、配送距离、当前排队情况排序最后选综合体验最好的那家下单。单看评分高没用还得看它忙不忙、送不送得到。2.3 触达通道与消息协议投递不丢不重不漏路由决策只解决了“把任务给谁”接下来还面临“怎么给”的问题。通道层我做了两种适配如果Agent提供的是HTTP接口绝大多数自己开发的业务Agent都是这种就用HTTP调用的方式同步或异步返回如果Agent是订阅MQ消息的一般是外部系统接入的场景就把任务封装成消息投递到指定队列。为了统一屏蔽差异我把两种方式都封装成了同一个调用接口对外只暴露send(agent_id, payload)这个方法。投递这块有个老生常谈但特别容易栽跟头的问题超时与重试。我给HTTP调用设了三档超时连接超时3秒、读超时15秒、业务处理超时30秒。为什么分三档因为它们失败的含义完全不一样连接超时说明网络有问题可以直接快速失败换一个Agent读超时说明Agent可能在处理但在慢慢来业务超时说明Agent已经接收但流程卡住了。重试的时候不能一概而论读超时还可以等等再重试一次业务超时重试往往没有意义反而可能让Agent重复执行。另外所有投递消息都必须带一个幂等键业务Agent端要按这个键做去重防止网络重试导致同一任务被执行两遍。结果回传我用的是一条独立的回调通道。Agent执行完后把处理结果按约定格式POST回触达层如果Agent在调用时携带了request_id回传时会带上同一个request_id这样整条链路就能串起来做监控跟踪。这一块看着简单但如果没有从一开始就约定好协议后期对接外部Agent时非常痛苦。3. 实操过程从零搭建Agent-Reach核心组件3.1 环境准备与最小工程结构我实现Agent-Reach用的是Python 3.10 FastAPI选FastAPI是因为业务侧的Agent也大多是FastAPI写的做集成测试方便。注册信息临时放内存字典生产环境可以替换成Redis或者MySQL这个不影响核心逻辑。最小工程结构我建议这样组织agent_reach/ ├── agent_reach/ │ ├── __init__.py │ ├── register.py # Agent注册与心跳管理 │ ├── router.py # 意图路由与打分排序 │ ├── channel.py # 通道层HTTP投递/回传 │ ├── gateway.py # 触达层FastAPI接口 │ └── model.py # 数据模型注册信息、任务包 ├── agents/ │ ├── order_agent.py # 示例Agent订单查询 │ ├── promo_agent.py # 示例Agent优惠计算 │ └── faq_agent.py # 示例Agent常见问题解答 └── tests/ └── test_route.py # 路由逻辑单元测试生产级工程还会加缓存、监控、配置中心这些但核心链路就靠上面六个模块先把主链路跑通最重要。3.2 核心代码注册中心与路由分发先看注册模块。我用一个AgentRegistry类维护所有Agent的注册信息和状态同时提供注册、心跳续约、状态查询三个接口。核心逻辑是依赖一个loaded_at字段做过期淘汰。import time import threading from typing import Dict, Optional class AgentRegistry: def __init__(self): self._agents: Dict[str, dict] {} self._lock threading.Lock() self._offline_timeout 90 # 秒 def register(self, agent_info: dict) - str: agent_id agent_info[agent_id] with self._lock: now time.time() agent_info[updated_at] now agent_info[status] online self._agents[agent_id] agent_info return agent_id def heartbeat(self, agent_id: str) - bool: with self._lock: if agent_id not in self._agents: return False self._agents[agent_id][updated_at] time.time() self._agents[agent_id][status] online return True def get_online_agents(self) - list: now time.time() online [] with self._lock: for aid, info in self._agents.items(): if now - info.get(updated_at, 0) self._offline_timeout: info[status] online online.append(info) else: info[status] offline return online这里有一个容易忽略的点过期淘汰不能只在查询时做还要有一个后台线程定期把离线Agent踢掉否则Agent如果永久下线注册表里的脏数据会一直占用内存而且可能导致查询老返回一个已经不存在的东西。我在项目里是用一个daemon线程每30秒扫描一次把超时的Agent从字典中移除让健康检查这层逻辑彻底交给触达层来保证。再看路由打分模块这是整套方案的核心。我实现了一个Router类根据传来的任务意图在在线Agent里做过滤和打分然后返回排名列表。import numpy as np class Router: def __init__(self, registry: AgentRegistry): self.registry registry def dispatch(self, intent: str, context: dict) - list: agents self.registry.get_online_agents() candidates [] for agent in agents: # 过滤阶段硬性条件不满足直接跳过 if not self._match_capability(agent[capabilities], intent): continue if agent.get(current_load, 0) agent.get(max_load, 10): continue score self._score(agent, intent, context) candidates.append((score, agent)) candidates.sort(keylambda x: x[0], reverseTrue) return [(score, agent[agent_id]) for score, agent in candidates[:3]] def _match_capability(self, capabilities: list, intent: str) - bool: # 标签完全匹配 if intent in capabilities: return True # 同义词匹配 for cap in capabilities: if self._similarity(intent, cap) 0.6: return True return False def _score(self, agent: dict, intent: str, context: dict) - float: sim self._similarity(intent, .join(agent[capabilities])) success_rate agent.get(success_rate, 0.8) load_ratio 1 - agent.get(current_load, 0) / max(agent.get(max_load, 10), 1) return sim * 0.5 success_rate * 0.3 load_ratio * 0.2 staticmethod def _similarity(text_a: str, text_b: str) - float: # 实际项目中这里用的是文本向量余弦相似度 # 简单演示时用字符集合Jaccard近似 set_a, set_b set(text_a), set(text_b) if not set_a or not set_b: return 0.0 return len(set_a set_b) / len(set_a | set_b)这段代码里的_similarity在生产环境我会换成预训练模型的向量余弦相似度演示用Jaccard只是把雏形跑通。负载余量参与打分这个设计我实际用了很久才发现它的价值如果不考虑负载高并发场景下得分最高的Agent会被连续打满其他Agent干闲着整体吞吐反而下降。加上负载权重之后路由会自动做负载均衡不需要额外写一套流量分配逻辑。3.3 端到端联调模拟三Agent协作工程骨架和核心代码都有了我用一个贴近业务的小场景做了端到端联调三个Agent分别是订单查询Agent、优惠计算Agent、常见问题Agent。用户的一条请求进来“我上周买的手机怎么还没发货另外现在有没有优惠”这条请求同时包含两个意图。触达层先拆解成一个主任务订单查询和一个附加任务优惠计算。两个任务串行执行主任务先由订单Agent查询发货状态结果里带上订单金额接着这个金额作为参数传给优惠计算Agent算出可用的优惠信息最后把两块结果合并返回。FAQ Agent不参与这次路由因为两个意图跟它都不匹配。我在gateway.py里把整条链路串起来模拟运行时的日志大致长这样[触达层] 收到请求: 我上周买的手机怎么还没发货另外现在有没有优惠 [触达层] 意图识别完成: [订单查询(0.87), 优惠计算(0.79)] [调度层] 订单查询候选: [(0.82, order_agent), (0.51, faq_agent)] [调度层] 优惠计算候选: [(0.79, promo_agent)] [通道层] 投递任务 - order_agent (request_id: r_20250214_001) [通道层] order_agent 返回: 订单状态已揽收, 预计送达2月16日, 订单金额3899 [通道层] 投递任务 - promo_agent (request_id: r_20250214_002) [通道层] promo_agent 返回: 可用优惠满3000减200, 有效期至2月28日 [触达层] 合并结果返回用户跑通这组联调花的时间比我预期久主要卡在意图拆解上。最开始我直接把整句话送进意图识别只识别出一个主导意图结果优惠计算那条就被漏了。后来改成“先拆解子任务、再逐个路由”的方式效果才稳定。这个经验分享出来给要做多Agent链路的同学提个醒一个用户请求往往包含多个子任务触达层必须先把任务拆解到底再做逐个子任务的路由否则整条链路的效果会大打折扣。4. 常见问题与排查技巧实录4.1 Agent注册后永远“不在线”我在联调时第一次遇到的坑是Agent启动后调了注册接口注册表里能看到心跳也在更新但路由查询时它就是不出现。排查了半天才发现问题出在Agent的endpoint上——我注册的是内网地址但是路由进程跑在同一台机器上还能通一旦部署到不同环境就出现网络隔离。后来又发现另一个更隐蔽的问题我把健康检查路径写成了/health但Agent实际只实现了/ping注册时检查健康状态直接失败Agent被标记为不健康自动从候选池里摘掉了。这个经验的核心结论是注册信息里的endpoint和健康检查路径一定要做两次验证。第一次是Agent进程启动后自己验证第二次是触达层真的发起一次HTTP探测不能只看注册表的“在线”标记。我后来把checks变成了三重验证心跳活跃、健康检查探活、最近一次路由是否产生过真实调用。就差最后一重验证没加之前线上出现过Agent挂着心跳但实际处理能力已经完全退化的情况。4.2 路由结果不稳定同一请求打到不同Agent还有一个典型问题是一模一样的请求过一段时间再发路由结果变了。这其实不一定算bug因为负载权重本身就在动态变。但如果出现“同一时刻同一个请求两次路由结果不一样”那就要查是不是哈希遍历顺序不稳定导致的。Python3.7之后字典是有序的但如果你用了set或者多线程并发改注册表遍历顺序就可能飘。这类问题我用了一个很简单的修复在路由打分前对整个候选列表做一次确定性排序先按agent_id升序再按分数降序这样即使打分分数一样也能保证同一个请求永远选中同一个Agent。此外阈值和权重不是拍脑袋定的我跑了小批量样本看分布情况发现相似度低于0.35的匹配基本都是错配才把阈值定在那个位置。4.3 投递消息丢失与重复执行通道层最折磨人的问题有两个消息丢了和消息重复。消息丢的根源几乎都是没有确认机制——投递HTTP请求后Agent返回了200但通道层没等响应体完整解析就标记成功结果响应体在网络中被截断数据就丢了。改成必须收到完整响应体才确认后这个问题消失了。消息重复则来自重试机制。我之前写了一个简单的超时重试只要Agent没在5秒内响应就重发一次。有一天排查数据重复发现Agent确实幂等了但重试者的重试计数没有重置导致连续触发了五次重试。正确的做法是重试次数上限与幂等键挂钩同一request_id最多重试两次不管响应多慢都绝不重发第三次同时Agent端根据request_id去重。这两端配合才能既保证可用性又保证一致性。4.4 性能压测与参数调优最后给一组我在这套方案上跑过的压测数据方便作为调参的起点。我的环境是三台服务器触达层两节点调度层单节点三组Agent各单副本。压测工具用locust模拟用户请求总量10000峰值QPS控制在500左右。场景平均耗时(ms)P99耗时(ms)成功率单意图直路由428999.96%双意图拆解路由7615899.82%带Agent执行时间(模拟200ms)28648999.21%调参方面路由缓存TTL我从60秒降到了10秒耗时会增加一点点缓存命中率下降但换来的是Agent状态感知实时性大幅提升我觉得值。心跳续约间隔30秒没变但离线判定超时从120秒收紧到90秒因为通讯正常的情况下60秒内应该能看到活跃心跳。另外我把注册表的锁从粗粒度的全局锁换成了分片锁压测下锁竞争导致的耗时从占比12%降到不足3%。如果你们的Agent数量不超过50个完全不用纠结锁的粒度等量级上来了再优化也不迟。最后分享一点个人体会。Agent-Reach这类触达层本质上是在做“语义层面的服务治理”它不是给Agent增加业务能力而是让已有的Agent能力能被更准确、更高效地调度起来。我在落地的过程中最大的一个感触是先别急着上复杂的路由算法、向量匹配、模型推理先把“能注册、能探活、能路由、能投递、能回传”这条主链路跑通、跑稳再逐步引入更聪明的决策逻辑。很多团队一开始就把调子定得很高结果被底层链路的不稳定拖住了反而是基础方案先跑通、后面再迭代更容易出效果。如果你们团队也正在被类似的问题困扰可以从一个最小可用的注册表加路由服务开始试跑一个月再回头看你会发现自己已经离不开这个触达层了。