多Agent协同触达框架Agent-Reach：注册、发现与路由实践

先交代一下背景2024年下半年开始我这边项目里的Agent从“单兵作战”转向“多Agent协同”结果最大的瓶颈反而不是单个Agent的能力而是Agent之间怎么互相找到、怎么把活儿派下去、怎么把上下文交接清楚。试过不少方案最后沉淀出一个轻量级的多Agent触达框架就叫Agent-Reach。这套东西折腾了大概一个月踩了不少坑也趟出了一些经验趁着这几天把设计思路、核心代码精简版和排查实录整理出来希望能给正在做同类事情的人一些参考。适合的读者有三类刚接触多Agent编排、想在项目里让多个智能体互相协作的开发者以及正在对比各种Agent通信方案的架构师。如果你只有单个Agent那这篇东西暂时用不上先收藏也行。1. 为什么我需要一个“Agent-Reach”1.1 单Agent再强也只是个“单兵”在聊Agent-Reach之前先说个我自己的直观感受。单Agent的能力边界其实很清晰给它一个任务、配好工具和模型它就能干活。但你一旦把好几个Agent放在同一个系统里问题就变得微妙了。打个比方单Agent像是一个全能的“个体户”什么活儿都能接但都自己干。多Agent协同则像是你开了一家公司有销售、有技术、有客服每个人各有所长。公司要运转光有“能干的人”不够还得有通讯录、工单系统、派单规则。Agent-Reach扮演的正是这个“内部通讯和调度系统”。我在实际项目中遇到的第一波问题不是某个Agent能力不够而是Agent之间根本没建立起“连接”。比如我有一个“数据分析Agent”和一个“报表生成Agent”数据分析Agent算完了一堆指标怎么把这些指标塞给报表生成Agent最初的做法是代码里硬编码A调用B的直接API。两个人还好等Agent到了七八个硬编码关系就变成了一张蜘蛛网改一个Agent的接口牵连三四个地方谁都不敢动。这也是我决定专门做一个触达层的原因把Agent之间的连接从“点对点的硬编码”变成“注册—发现—路由—传递”的统一通道。Agent不需要知道对方在哪里、用什么接口协议只需要知道自己要发出什么类型的请求剩下的交给Agent-Reach来解决。1.2 多Agent协作的三个硬痛点把问题再拆细一点我觉得多Agent协作里最磨人的是下面三个东西。第一个是能力发现。系统里有七八个Agent每个Agent能干什么、接受什么样的输入、产出什么样的结果如果没有一个统一的注册信息你就只能用嘴硬记。开发阶段还能翻代码到了运行阶段一个Agent扩容了三个实例或者某个能力下线了调用方根本不知道。这个痛点有点像你手机里存了一堆联系人但没有搜索功能找人只能翻通讯录。第二个是任务路由。假设用户说了一句“帮我把最近三十天的销售数据做成可视化看板”这个任务该给数据分析Agent还是报表生成Agent还是两个都要参与如果没有路由逻辑要么全广播给所有Agent造成浪费要么靠代码里的if-else判断加一个新任务类型就得改一遍路由代码。我当时维护的那套if-else已经膨胀到了让人头皮发麻的程度。第三个是上下文传递。多个Agent协作时前置Agent的输出往往是后置Agent的输入。分析Agent得出一个“华东区销售额下滑18%”的结论如果这个结论不能被报表Agent理解、接住那这个Agent协同就只是形式上的“调用了”实际上还是各干各的。上下文传什么、传多少、要不要做格式转换这些都需要一个可配置的规则而不是在每个调用处临时拼接。1.3 Agent-Reach想解决的事Agent-Reach的核心目标其实就一句话让Agent像服务一样被注册、被发现、被调用同时把Agent间的上下文流转纳入统一的收口管理。它的设计理念跟微服务治理有不少相似之处但又有明显的区别。微服务治理处理的是“服务”和“接口”Agent-Reach处理的是“能力”和“任务语义”。服务接口是稳定的而Agent的输入输出天然带有不确定性可能是一段自然语言、一个结构化JSON、甚至是一个多模态的对象。所以Agent-Reach不能简单套用REST的思维它需要更灵活的消息模型和触达策略。另外Agent-Reach面向的核心场景是“工作流里的智能体接力”。这个接力过程里哪个Agent先上、哪个Agent后上往往不是固定的而是需要根据上下文动态判断。这意味着它不仅要管“怎么把消息送到”还要管“这一棒应该交给谁”。这个“交给谁”的判断放在Agent-Reach里被称为触达策略我后面会详细讲。注意Agent-Reach不是搜索引擎也不是RAG管道。它不做知识检索只做Agent之间的发现与触达。如果大家发现某个框架频繁出现“检索增强”“向量库”之类的功能大概率是走偏了。2. Agent-Reach 的核心设计拆解2.1 整体架构注册中心 消息总线 触达路由Agent-Reach的架构不复杂核心就三块注册中心、消息总线、触达路由。注册中心负责保存所有Agent的元信息包括Agent标识、能力列表、当前状态、服务地址、版本信息等。Agent启动时向注册中心登记关闭时注销运行期定期发送心跳表示自己还活着。消息总线解决的是“消息怎么传”的问题。它屏蔽了底层的传输细节不管Agent是在同一个进程、同一台机器还是跨机器部署对Agent而言都是发一条消息给某个逻辑标识。这个设计在前期可能显得有点重但Agent数量一旦超过十个消息总线的价值就会非常明显你可以统一做日志、统一做监控、统一做限流不需要每个Agent自己实现一套通信逻辑。触达路由是Agent-Reach最有区分度的模块。它的职责是接收一个“任务意图”结合注册中心里的能力元信息决定把消息发给哪一个或哪一组Agent。路由规则可以是静态配置的比如“所有销售数据分析类的任务都发给分析Agent”也可以是动态计算的比如根据当前各Agent的负载、近期的成功率、上下文里提到的实体来动态选择触达目标。拿一个实际场景来说。业务侧发来一个任务“用户问客服为什么上周的退款没有到账。”触达路由拿到这个意图拆解出关键字“退款”“客服”“上周”于是将消息同时触达“客服Agent”和“财务流水Agent”让客服Agent负责组织话术财务流水Agent负责查证退款状态。没有Agent-Reach这套路由这两个Agent根本不会知道彼此在这个任务里都扮演了角色。2.2 Agent身份与能力声明在Agent-Reach里每个Agent都有一份“身份档案”这是它在整个系统里被识别和调用的基础。身份档案的组成不是随便定义的我整理下来大概包含这么几类信息基础标识agent_id、名称、版本、能力描述capabilities每项能力要有能力名、输入参数定义、输出结果定义、示例、运行信息地址、协议、心跳时间、负载状态、语义信息适用的任务类型、领域标签、约束条件。能力描述这个字段最容易被人忽略但恰恰是最影响整个系统体验的一个点。你想想触达路由需要在运行期决定“这个消息谁能接”它就是靠分析能力描述来匹配的。如果能力描述写得很含糊比如就写了一句“处理各种任务”那路由基本只能靠猜。反之如果能力描述写得很结构化比如“能根据销售明细生成按地区、按产品线的汇总表输入是CSV或JSON输出是Markdown表格”路由就能做比较精准的匹配。我自己的经验是能力声明一定要围绕“输入输出可预期”来写。Agent的模型再强对外提供的能力边界如果不可预期别人就不敢调它。你可以允许Agent内部自由发挥但在能力声明的边界上尽量把事情说清楚。开发时我们调侃过一句话——“声明能力时请把自己当作一个严肃的API供应商不要当作一个天马行空的创作者。”这句话后来直接写进了团队的开发规范。2.3 服务调用的消息模型Agent-Reach里所有触达行为都封装成一种统一的消息结构它跟普通RPC请求最大的不同是消息里除了“参数”还带了一截“上下文”。我们定义的消息头包含这几项消息ID全链路唯一用于追踪和关联式日志。任务ID多次触达共享同一个任务ID表示它们属于同一条业务链路。源Agent、目标Agent或目标能力描述两者至少有一个。消息类型指令型请执行、咨询型请判断、通知型不需要返回结果。调用策略同步等待、异步回调、超时时间、重试次数。消息体里除了调用参数还有一块designated_context这是给目标Agent的上下文补全。设计这块的时候我们刻意让上下文分为两层全局上下文和局部上下文。全局上下文是链路里所有Agent都应该看到的比如用户ID、业务线、总任务目标局部上下文只对当前目标Agent可见比如某个内部推理过程。这个分层设计避免了两个常见的问题。一是上下文爆炸如果每经过一个Agent就把所有历史信息都带上消息体很快就会膨胀到不划算二是上下文越权有些中间过程不该让下游Agent知道做了分层就能控制信息的可见范围。很多失败的Agent协作根源就出在“该传给下游的没传不该传的倒是全给了”。2.4 为什么不用“直连”而用“间接”在多Agent架构里有人会问一个很直接的问题Agent之间直接互相调用API不就行了吗为什么非要引入一个中间层我的答案是当Agent数量少、协作关系固定时直连完全够用但一旦规模上来中间层带来的收益会远超它的成本。直连最大的问题是耦合。Agent A要知道Agent B的地址、接口格式、参数要求这意味着双方要同时发布、同时修改、同时验证。我在项目里经历过一次特别难受的联调分析Agent改了输出字段名报表Agent没同步改结果线上数据报表直接空了一列。这种问题在直连架构里只能靠“人盯人”来避免而在Agent-Reach里因为触达路由负责能力匹配Agent B只知道“有人需要我生成报表”具体的字段变化可以通过消息模型里的schema版本处理掉不会牵一发动全身。另一个直连解决不了的问题是“系统视角”。直连模式下没有一个地方能看到全链路的调用关系。Agent运行期出了故障你只能每个Agent挨个翻日志。Agent-Reach因为所有消息都经过统一的总线和路由天然就有全链路的日志和监控能力。哪个Agent响应慢了、哪个Agent拒绝服务了、哪条链路断了在总线上都能看到排查效率完全不是一个量级。当然间接不等于低效。Agent-Reach在设计上允许“已建立的触达通道走直连快路径”类型于在总线上先完成寻址和鉴权再把后续通信切到点对点。这个优化我们在高吞吐场景下验证过能够把额外的路由开销控制在可接受范围。3. 动手实现一个精简版 Agent-Reach3.1 环境准备与目录结构直接上代码。我在这里实现的是一个教学向的精简版Agent-Reach核心能力包括注册、心跳、发现、路由、同步调用。生产环境的版本比这个复杂不少但核心骨架一致。看完这个精简版大家再去看完整的开源实现就不会迷路。环境方面没有特殊要求基于Python 3.10以上就行。依赖只用到了FastAPI和uvicorn为了演示的简单消息总线直接用进程内的字典和队列实现。如果要在分布式环境跑替换成Redis或消息队列就好。目录结构我按模块拆成下面这样agent_reach_demo/ ├── main.py # 启动入口拉起路由服务 ├── registry.py # 注册中心 ├── router.py # 触达路由 ├── agent_runtime.py # Agent运行框架提供装饰器和注册方法 ├── agents/ │ ├── analysis_agent.py # 数据分析Agent示例 │ └── report_agent.py # 报表生成Agent示例 └── tasks/ └── sample_task.py # 演示任务入口这个结构不要嫌小麻雀虽小五脏俱全。我把注册、路由、运行时分开了为的是让大家看清楚各自的职责边界。实际项目里还会加一个message_bus模块和monitor模块这里为了聚焦核心逻辑先不展开。3.2 注册中心的实现注册中心要干的事就三件登记Agent、更新心跳、按能力查找Agent。先看代码。import time import threading from collections import defaultdict from typing import Optional class AgentRegistry: def __init__(self, heartbeat_timeout15): self._agents {} # agent_id - info self._lock threading.Lock() self._heartbeat_timeout heartbeat_timeout self._gc_thread threading.Thread(targetself._gc_loop, daemonTrue) self._gc_thread.start() def register(self, agent_id: str, address: str, capabilities: list[dict]) - None: 注册或刷新Agent信息,同时重置心跳时间 with self._lock: self._agents[agent_id] { agent_id: agent_id, address: address, capabilities: capabilities, status: UP, last_heartbeat: time.time(), registered_at: time.time(), } def heartbeat(self, agent_id: str) - bool: Agent心跳上报,返回是否仍然被注册中心认领 with self._lock: if agent_id not in self._agents: return False self._agents[agent_id][last_heartbeat] time.time() self._agents[agent_id][status] UP return True def unregister(self, agent_id: str) - None: with self._lock: self._agents.pop(agent_id, None) def find_by_capability(self, requirement: str) - list[dict]: 根据能力声明查找候选Agent。 精简版用关键字匹配,生产环境会结合向量/规则做更精细的匹配。 requirement requirement.lower() results [] with self._lock: for info in self._agents.values(): if info[status] ! UP: continue for cap in info[capabilities]: if requirement in cap[name].lower(): results.append({ agent_id: info[agent_id], address: info[address], capability: cap, }) break return results def _gc_loop(self): 定时清理失活Agent while True: time.sleep(self._heartbeat_timeout) now time.time() with self._lock: dead_ids [ agent_id for agent_id, info in self._agents.items() if now - info[last_heartbeat] self._heartbeat_timeout ] for agent_id in dead_ids: self._agents[agent_id][status] DOWN这里有一个细节值得多说一句GC线程里我没有直接把失活Agent从字典里删掉而是把状态置为DOWN。原因很简单——如果你想排查一个Agent为什么反复失联留痕比清理更重要。生产版本里我还会保留一份DOWN状态的归档日志方便事后复盘。直接删除的话问题Agent就“蒸发”了排查线索也就断了。3.3 Agent端封装装饰器 生命周期有了注册中心Agent端怎么接进来不能要求每个Agent开发者都去手动调用registry.register那太原始了。我在agent_runtime.py里做了一个装饰器让Agent开发者只需要写清身份和能力剩下的注册、心跳、接收路由都由框架处理。import time import threading import uuid from registry import agent_registry class AgentRuntime: def __init__(self, agent_id, address, capabilities, handler): self.agent_id agent_id self.address address self.capabilities capabilities self.handler handler agent_registry.register(agent_id, address, capabilities) self._stop False threading.Thread(targetself._heartbeat_loop, daemonTrue).start() def _heartbeat_loop(self): while not self._stop: agent_registry.heartbeat(self.agent_id) time.sleep(5) def handle(self, payload): return self.handler(payload) def stop(self): self._stop True agent_registry.unregister(self.agent_id) class AgentMeta(type): 通过元类把Agent类定义转为运行时实例 def __new__(mcs, name, bases, namespace): cls super().__new__(mcs, name, bases, namespace) agent_id namespace.get(agent_id) if agent_id: # 实际应该从配置读取address,这里示例直接构造 address fhttp://localhost:9010/{agent_id} runtime AgentRuntime( agent_idagent_id, addressaddress, capabilitiesnamespace.get(capabilities, []), handlernamespace.get(handle), ) cls._runtime runtime return cls这样Agent开发者定义一个类就算完事class AnalysisAgent(metaclassAgentMeta): agent_id analysis_agent capabilities [ { name: sales_analysis, inputs: [sales_data, group_by], outputs: [summary, dimensions], description: 对销售明细做多维度分析,输出汇总结论, }, { name: trend_forecast, inputs: [history_data, horizon], outputs: [forecast_result], description: 基于历史序列做趋势预测, }, ] def handle(self, payload): # 实际的模型推理与计算逻辑 data payload[input][sales_data] group_by payload[input].get(group_by, all) # 省略计算过程 return { summary: f按{group_by}汇总完成,指标数{len(data)}, dimensions: group_by, }Agent只关心自己的业务逻辑连接的事交给框架这在实际落地时对团队开发的友好度提升非常明显。我们团队当时把几个存量Agent接入Agent-Reach纯改造成本基本上半天一个。3.4 路由与消息传递路由模块是整个系统最核心的地方它的职责是把外部请求翻译成“对某个能力的触达”。import time import uuid from registry import agent_registry class ReachRouter: def __init__(self): self._callbacks {} def dispatch(self, task_definition: dict) - dict: task_definition结构: { task_id: 可选, requirement: sales_analysis, input: {...}, context: {...}, strategy: first_match | broadcast | dynamic } requirement task_definition[requirement] input_data task_definition[input] context task_definition.get(context, {}) task_id task_definition.get(task_id, str(uuid.uuid4())) candidates agent_registry.find_by_capability(requirement) if not candidates: return {status: NO_TARGET, message: f没有Agent声明能力:{requirement}} target candidates[0] # 精简版取第一个,生产版会加负载和策略判断 request_id str(uuid.uuid4()) message { request_id: request_id, task_id: task_id, source: router, target: target[agent_id], type: COMMAND, input: input_data, context: context, } # 这里在精简版中直接调用本地handler # 生产环境会换成基于消息总线的传输 handler self._callbacks.get(target[agent_id]) if not handler: return {status: ERROR, message: fAgent离线或未注册回调:{target[agent_id]}} try: result handler(message) return {status: OK, request_id: request_id, result: result} except Exception as exc: return {status: ERROR, message: str(exc)} def register_callback(self, agent_id, fn): self._callbacks[agent_id] fn我特意在注释里标注了哪些环节在生产版本会扩容候选选取这里的selected策略、handler调用里的传输机制、以及回调的注册方式。大家在学习这段代码时先不要纠结“这么简单能生产用吗”先把消息流走通。路由是“触达”这个词落地的地方。我在设计时刻意把触达分为“发现”和“送达”两个步骤。发现是找到能力匹配的Agent送达是把消息可靠地交给目标。两个步骤分离的好处是将来任何一个步骤要升级都很独立。比如我要给“发现”加上基于历史成功率的选择不需要动“送达”的任何代码。3.5 跑通一个跨Agent调用下面我们把两个Agent串起来跑一个最简单的“分析并生成报表”任务。这里用一个上游Agent分析Agent和下游Agent报表Agent中间通过Agent-Reach路由触发。# tasks/sample_task.py from agent_runtime import AgentMeta from router import reach_router class ReportAgent(metaclassAgentMeta): agent_id report_agent capabilities [ { name: report_generation, inputs: [title, content_blocks], outputs: [markdown_report], description: 把结构化的内容块生成为markdown报表, } ] def handle(self, payload): title payload[input][title] blocks payload[input][content_blocks] # 简化处理,真实场景会调用模板引擎 md f# {title}\n\n for block in blocks: md f- {block}\n return {markdown_report: md} def run_sample(): # 触达1:让分析Agent干活 task1 { requirement: sales_analysis, input: { sales_data: [ {region: 华东, amount: 100}, {region: 华南, amount: 80}, ], group_by: region, }, context: {biz_line: 电商}, } result1 reach_router.dispatch(task1) # 把分析Agent的产出作为上下文,触达2:交给报表Agent task2 { requirement: report_generation, input: { title: 销售分析简报, content_blocks: [result1[result][summary]], }, context: {upstream_agent: analysis_agent}, } result2 reach_router.dispatch(task2) print(result2[result][markdown_report]) if __name__ __main__: run_sample()这个演示虽然简单但已经覆盖了Agent-Reach三个核心动作能力发现、上下文传递、结果接力。大家可以把这段代码放在自己的环境里跑一遍看看完整调用链。实际做项目时两个Agent之间往往还要插入校验环节比如分析Agent输出的summary格式是否满足报表Agent的输入要求这个校验在Agent-Reach里由“协议适配”模块完成这里就不展开了。3.6 心跳与故障摘除心跳机制是Agent-Reach的稳定性基石。没有心跳路由永远不知道一个Agent已经挂了还会继续往里派任务造成一堆超时报错。我在心跳设计上踩过的坑是把心跳和业务处理放在同一个线程里。某次数据分析Agent处理一个超大查询阻塞了整整两分钟心跳线程被拖累注册中心判定它失联触达路由就把它摘除了。等大查询跑完Agent已经是“已下线”状态后续任务全部落空。这个经历告诉我们心跳发送必须与业务主流程分离要么独立线程要么独立进程。代码里AgentRuntime的做法就是独立的daemon线程业务代码阻塞再久心跳不受影响。故障摘除还有个细节摘除后Agent恢复如何重新加入。Agent-Reach的策略是Agent重启后主动重新注册注册中心对比版本号如果版本号有更新就刷新能力清单如果版本号相同只恢复心跳状态。这种策略可以避免Agent滚动更新时出现能力信息不一致的窗口期。4. 实操过程中踩过的坑与排查实录4.1 注册丢失时序问题的高发区我们最早跑起来的时候经常出现“Agent明明启动了路由却找不到目标”。排查到最后问题几乎全部指向启动时序。Agent启动时的注册是异步的而任务可能更早到达。如果路由在注册还没完成时就做了发现自然找不到目标。这个问题的解法分两层Agent启动后增加一个就绪态标记路由发现“所有候选Agent都还没就绪”时不直接返回失败而是一边重试一边把消息暂存到一个pending队列。重试策略用指数退避退避阈值别超过心跳时间的三倍。示例排查流程我整理成一个口诀先看心跳时间是否连续再看注册中心里的状态是否为UP最后看路由发现时刻的Agent状态。这三个位置按顺序检查基本能定位绝大多数注册丢失问题。4.2 消息超时与重试策略多Agent链路里消息超时是必然事件不是你代码写得好就能避免的。模型推理的延迟天然有波动同一个Agent在高峰期可能要跑十几秒平时只要两三秒。超时重试这块最容易犯的错是“盲目重试”。尤其是消息已经送达且Agent处理了一半客户端因为超时就重发了一条一模一样的消息结果Agent又处理了一遍。这在涉及写操作的场景下非常危险。Agent-Reach的处理办法是给每条消息带上request_idAgent端对request_id做幂等如果Agent发现request_id已经处理过就直接返回上一次的处理结果不再重复执行。具体的超时参数我建议分成三层连接超时1秒、读超时默认10秒可按Agent调整、任务总超时30秒以上。三层超时各管一段定位问题时能很快缩小范围。4.3 上下文漂移Agent之间的“失忆”上下文漂移是我在这套系统里最头疼的问题。表现是这样的链路第一阶段分析Agent产出了结论第二阶段报表Agent引用了这个结论但因为上下文传递时少带了一个字段报表Agent只能靠猜测补全最后生成的报表结论跟分析结果对不上。后来查根因发现是路由里选错了上下文层级。我前面提到全局上下文和局部上下文分离但在实现时局部上下文被忽略了。分析Agent输出的summary被放进了全局上下文本意是让后续所有Agent都能看到结果全局上下文里的键名越来越杂强耦合的数据被塞了进来下游Agent不知道该信任谁。解决方式是在上下文传递时加一个Schema校验。每个能力声明里写清楚它依赖哪些全局上下文键、接收哪些局部上下文键路由在触达前做一次校验缺了键就直接返回“上下文缺失”。这在前期会带来一些字段对齐的工作量但比起排查“失忆”问题的成本这点工作量太划算了。4.4 安全边界与权限控制Agent-Reach本质上让每个Agent的能力变成可被路由调用的开放接口这就带来一个安全问题不是所有Agent都应该允许触达所有其他Agent。我们在生产环境里遇到过内部Agent误触达的情况一个测试用的Agent忘记下线把生产环境的订单Agent当成目标发出了测试指令。虽然最后没有造成实质影响但它暴露了一个事实——很多多Agent系统默认“内网安全”结果内部权限完全裸奔。Agent-Reach的权限控制在消息层解决每条触达消息都携带来源Agent的凭证路由在转发前校验“源Agent是否有触达目标能力的权限”。权限配置采用最小授权原则没有显式授权就默认拒绝。配置项可以通过一个简单的关系表维护Agent数量上来后建议换成策略引擎。我个人的建议是别把权限问题拖到Agent数量很大时再来设计六个Agent左右就要考虑模型了。规模越大越改越痛苦。4.5 常见问题速查表现象可能原因排查步骤与解法路由提示NO_TARGET注册信息缺失或Agent DOWN检查Agent启动注册是否成功、心跳是否连续、能力名是否与服务声明匹配消息持续超时网络分区或模型推理过慢先看三层超时配置再看目标Agent日志最后检查是否触发幂等逻辑下游Agent收到脏数据上下文层级传错或缺少Schema校验检查任务ID链路日志确认全局/局部上下文边界补全Schema校验Agent被反复摘除心跳线程被业务阻塞心跳必须独立线程确认注册中心GC逻辑避免单点误判同一任务被重复执行重试时未携带request_id强制消息幂等Agent端对request_id做去重新Agent上线不生效路由缓存未刷新注册中心状态发布是实时还是轮询确认缓存过期时间这张表里的每一条都在我做的真实项目里发生过。做Agent-Reach最大的感受就是技术上真正的难点不是实现一个功能而是预估它会在什么场景下出问题。5. 从Agent-Reach到更广阔的Agent互联5.1 与MCP等协议定位的差异聊Agent-Reach时总有人会问“这和MCP有什么区别”。我统一说下我的理解。MCP这类协议解决的是“Agent如何统一访问外部工具和数据源”重点在Agent与工具之间的标准化接口相当于给Agent装了一套通用的“手脚”驱动。Agent-Reach解决的是Agent与Agent之间如何发现、触达、协作重点在Agent之间的组织和调度相当于Agent的“神经系统”。两者不是替代关系而是互补关系。一个Agent内部的工具调用走MCP没问题多个Agent之间的接力协作需要Agent-Reach这类触达层。在实践中我发现用了Agent-Reach之后反而让MCP的价值更容易落地。因为Agent-Reach把Agent之间的调用标准化了每个Agent暴露能力时底层仍然可以借助MCP去访问各自需要的工具上层不用关心这些细节。5.2 Agent-Reach适合什么场景不适合什么我给Agent-Reach划了三个特别适合的场景。第一个是客服工单全流程处理。进来的工单经过分类、检索、内容生成、审核等多个Agent接力每个Agent只负责一段触达路由负责判断当前工单要走哪条分支。第二个是数据分析到报告生成的多步流水线。分析Agent、图表Agent、报告Agent串联任务上下文逐级传递。第三个是跨领域专家协同。一个法律Agent和一个业务运营Agent协同处理合同审核两者能力差异大需要通过统一路由交换信息。不适合的场景也很明确。如果你的场景是单Agent内多工具调用Agent-Reach帮不上什么忙如果你的Agent之间协作模式极其固定且数量长期不超过三个直连可能更轻量如果你的Agent之间完全不需要互相发现能力只是按固定脚本调用那上一套编排引擎就够了不必上Agent-Reach。5.3 后续扩展方向Agent-Reach现在这版还只是我内部的沉淀。后续有几个方向是我明确想往深里做的。一是语义级触达。现在的能力匹配大多还要靠结构化的能力名和规则将来希望Agent可以用自然语言描述自己能干什么路由通过语义向量匹配触达目标这样新增Agent的接入成本更小。二是动态路由策略。不只是根据能力匹配还叠加负载、成功率、成本、延迟等多个维度做多目标优化让触达不仅在“能不能”的层面可用还在“好不好”的层面可调。三是可编程的Agent协同协议。现在路由更多是预配置的期望以后能在任务外部动态注入一段协作策略像写SQL一样写Agent的路由规则。这会让整个系统的灵活度再上一个台阶。说起来Agent-Reach这个名字里的Reach我其实更偏好“触达”这个翻译。它不仅是把消息发出去的deliver更是一种“知道该找谁、如何接上头、怎样接着往下走”的机制。这比单纯的传递更像Agent协作的本质。最后分享一个我自己的体会做这种框架类的东西别上来就追求大而全。先把注册、发现、路由、心跳这四个基础动作做到位跑通一条真实业务链路再逐步加语义匹配、权限、动态策略。你后面遇到的问题很多都会在设计基础阶段就埋下伏笔。前面多花点功夫把消息模型和上下文边界想清楚后面会省下大量排查时间。这次把核心设计和代码拆出来分享也是希望大家能少走一遍我走过的弯路。