★ FEATURED ARTICLE 我自己动手写Agent-Reach这个项目起因其实很朴素一个号称“全能”的家庭助理Agent在集成完天气查询、日程管理、智能家居控制三个外部能力之后彻底变成了“全不能”。问题根子不在模型推理而在触达层——主控模型根本找不到一条稳定、清晰、可观测的路径去碰到它该用的工具。Agent-Reach就是在这个背景下做出来的一个轻量级多智能体触达编排框架。它的核心不是让Agent“更会思考”而是让Agent“更够得着”——把散落在各处的子Agent、API、脚本、数据源统一注册起来再通过一套路由规则和调度机制让主控Agent能可靠地触达它们。这篇文章主要是给正在做多Agent应用、被工具调用混乱和调度可靠性折磨的开发者看的。我会把Agent-Reach的定位、三个核心抽象、最小可运行代码、实测踩坑和选型边界一次讲透中间会穿插不少我实际操作中的教训希望能帮你少走弯路。1. 项目缘起一个“全能Agent”翻车后我只想先把“触达”搞清楚1.1 大Prompt塞工具的方案死在了集成混乱上最早我的想法特别简单把所有工具的描述和调用规则写进一个巨大的System Prompt靠模型自己决定调哪个。开头几个工具确实没问题等工具数量超过十个局面立刻失控。第一个问题是描述互相干扰。各个工具的使用说明挤在一起模型经常分不清“查询天气”和“查询气温趋势”到底该调哪个服务第二个问题是错误传染。只要一个工具接口超时整个对话流程就僵在那里模型反复尝试把上下文塞得乱七八糟第三个问题最致命——完全不可控。你根本不知道模型某个瞬间为什么会调用某个工具也没办法精准地限制某个高风险操作。这段时间我意识到一个之前被忽略的事情大模型的推理能力已经不是瓶颈瓶颈是它和外部世界之间那条“够得着”的路。我们聊AI Agent聊了很久行业内最火的词是“工具调用”但大多数人讨论的重点是“怎么把工具暴露给模型”很少有人关心“工具暴露出来之后调度怎么做”。而后者恰恰是决定Agent能不能上生产的关键。1.2 主流编排框架好用但都没有把“触达”当作一等公民为了解决问题我先后试了CrewAI、AutoGen和LangGraph。有一说一这几个框架都很有价值但用下来总有一个点不对味。CrewAI的抽象层级很高Agent、Task、Crew三个概念就能搭出漂亮的团队协作。但对底层工具调用的控制粒度太粗想要自定义一套精准的路由策略反而要绕很远的路。AutoGen的群聊模式别具一格多Agent互相对话非常灵活。但它天然带有“对话式协商”倾向Agent之间可能来回拉扯好几轮才能落定生产环境的响应时间经不起这么耗。LangGraph是我花时间最多的。它的StateGraph、Node、Edge把流程建模得很彻底适合有状态、分叉复杂的工作流。但项目一大图的节点和边一多维护成本直线上升。而且我能感觉到在LangGraph的世界里“触达某个能力”只是图里的一个节点它没有一个专门的原生概念来表达“我究竟该如何触达”这件事。这三者解决的核心问题其实是同一个方向Agent之间如何协作、流程如何编排。但“Agent如何稳定触达能力”这一层它们默认交给了开发者自己处理给出的支持比较稀薄。这给了我一个明确信号——与其在别人框架的抽象缝隙里补丁摞补丁不如做一个专门的、把“触达”本身做成协议的轻量框架。1.3 Agent-Reach的定位把触达做成协议Agent-Reach这个名字就是我对这件事的答案。Reach这个英文词本身就有“伸手够到、触达、到达”的含义。一个智能体的能力边界不取决于模型参数有多大而取决于它能以多高的可靠性触达多少真实世界的能力。所以Agent-Reach的定位非常聚焦它是介于LLM与具体工具、子Agent之间的一层轻量编排协议。它不做Agent记忆力不做复杂图编排只专注一件事——把“触达”变成可描述、可路由、可观测、可降级的工程对象。我把这种设计叫作“能力触达层”。在这层之上主控Agent只需要发一句话请求在这层之下工具永远不知道是谁在调用自己。中间负责找路、指路、兜底的全部由Agent-Reach承担。2. 架构设计的三个核心抽象Reachable、Route、HubAgent-Reach的整体架构不复杂全部核心概念只有三个Reachable、Route、Hub。项目能保持轻量正是因为边界划得清楚。我曾在好几个项目里吃过“抽象过多”的亏所以这次刻意只保留这三个。2.1 Reachable一切可触达能力的统一描述在Agent-Reach的眼里不管是子Agent、HTTP API、本地脚本还是数据库查询统统是同一个东西——Reachable。我定义了一个统一的数据结构所有能力都必须按这个结构注册。它有几个字段值得你格外重视后面踩坑也主要踩在这几个字段上name全局唯一的触达名称。description能力描述这一行是LLM路由的唯一依据写得好不好直接影响路由命中率后面我会展开说。input_schemaJSON Schema格式的参数说明。为什么必须有因为Hub要在触达前做参数校验不能等错误请求打到下游服务才发现问题。callable真正执行能力的异步函数。tags规则路由用的关键词标签。timeout触达超时时间这个必须由Hub强制不能交给下游自觉。idempotent是否幂等这个字段决定失败后能不能自动重试。version版本号后续灰度升级就靠它。我最早的设计里还有priority、rate_limit这些字段后来全砍了。原因是这些东西不同场景下差别太大硬塞进通用结构反而让注册代码臃肿。有特殊要求的放在具体Reachable的内部逻辑里处理就行。保持核心结构精简是Agent-Reach能长期维护的前提。2.2 Route触达路径的第一公民大多数工具编排框架里路由只是藏在角落里的一堆if-else。Agent-Reach把它提升成了一个显式概念——Route因为所有触达到达前的决策本质上都发生在路由这一步。我的实现里有三类路由策略按顺序执行第一层是声明式规则路由。基于tags和少量关键词规则把高频、明确、无歧义的请求直接命中到对应Reachable成本低、时延小不需要动大模型。第二层是服务降级匹配。规则路由找不到时退一步做宽松匹配比如用户说“下雨”规则库里没有关联但description里出现了“降雨概率”就按模糊匹配继续走。第三层是LLM意图路由。前面两层全部没搭上才动用模型能力。把用户请求和注册表里所有Reachable的描述拼在一起让模型选一个最合适的。这一层准确率最高、成本也最高所以必须放在最后兜底。三层顺序不是随意的而是按照“成本从低到高、覆盖范围从窄到宽”来设计的。高频场景永远走最便宜的路径LLM只在长尾场景才出场。这样生产成本和响应时延都能控制住。2.3 Hub触达的调度中枢Hub是Agent-Reach的心脏所有请求都从它这里进出。它的职责很集中管理注册表、接收请求、执行路由决策、发起触达调用、统计超时重试、记录触达轨迹。Hub做三件事同时它明确不做什么。不做Agent长期记忆不做任务图编排不搞Agent间的复杂协商。Hub就是一个专业的前台接线员你告诉它想办什么事它翻通讯录注册表、按决策规则Route、帮你接通对方Reachable并记录通话质量。这种职责单一的设计让排查问题变得非常简单——出问题先看Hub的日志触达链路一目了然。2.4 与MCP的边界不重复造轮子很多朋友会问现在MCPModel Context Protocol越来越火Agent-Reach和它是不是重叠了这里我明确说它们是两层东西。MCP解决的是“工具如何以标准化方式暴露给模型”的问题相当于统一了工具接入的插头规格。Agent-Reach解决的是“一堆已接入的工具如何被调度、路由、降级、观测”的问题相当于装了一个智能交换台。有了MCP工具接入变规范了但接入之后谁来选路、谁来决定这次触达调用哪个工具、超时了怎么降级MCP自己不管。所以Agent-Reach的定位里有一条明确的设计原则凡是MCP已经做好的绝不重复实现。已存在的MCP Server开发时只写一个薄薄的Adapter包一层就能注册成Reachable投入使用。我甚至可以说Agent-Reach天然是为MCP生态补上调度层而设计的。3. 核心代码实现从注册到触达的完整链路理论讲完了下面进入正经代码。我用Python实现了一个最小可用版本麻雀虽小五脏俱全。你可以直接照着搭先把链路跑通再替换成真实工具。3.1 先定义Reachable的统一结构我用一个dataclass来描述所有可触达能力代码不长但每个字段都对应着前面架构设计里的一个决策。from dataclasses import dataclass, field from typing import Any, Callable, Awaitable dataclass class Reachable: name: str description: str input_schema: dict callable: Callable[[dict], Awaitable[dict]] tags: list[str] field(default_factorylist) timeout: float 5.0 idempotent: bool False version: str 1.0.0这里最重要的一条约定callable必须是一个async函数。为什么因为Hub层要统一用asyncio做超时控制如果某个Reachable是同步阻塞函数整个事件循环会被卡死超时也失去意义。3.2 注册一个子Agent作为Reachable下面我把一个天气服务注册成Reachable。注意看它的description写法我特意写了“用户会怎么问”而不是只写“这个功能做什么”。这个细节是从踩坑里学来的对路由命中率影响极大。async def weather_agent(params: dict) - dict: city params[city] # 这里建议先接本地模拟数据跑通链路再替换为真实天气API return { city: city, temperature: 24, condition: 多云, humidity: 55, updated_at: 2025-06-18T10:20:00Z, } hub.register(Reachable( nameweather_agent, description( 查询任意城市当前的天气状况包括温度、湿度、天气现象、降雨概率。 用户常见说法今天会下雨吗、北京冷不冷、上海天气如何、明天多少度。 ), input_schema{ type: object, properties: { city: {type: string, description: 城市名如 北京、上海} }, required: [city], }, tags[天气, 气温, 下雨, 气象], timeout8.0, idempotentTrue, ))这里idempotentTrue不是随便写的。天气查询就是一个典型的幂等操作——重复查几次结果差别不大但绝不会产生副作用所以后面失败时Hub可以放心重试。3.3 Hub层实现注册表与三层路由接下来是Hub的核心路由逻辑。我把“先规则、后LLM”的三层策略落成代码为了易读省略了部分类型标注。class Hub: def __init__(self): self._registry: dict[str, Reachable] {} def register(self, target: Reachable): self._registry[target.name] target async def dispatch(self, user_request: str, params: dict | None None): target await self._route(user_request) if target is None: return {status: no_route, message: 没有找到可以触达的能力} return await self._invoke_with_safety(target, params or {}) async def _route(self, user_request: str) - Reachable | None: # 第一层规则路由遍历所有Reachable的tags做关键词匹配 for keyword in [天气, 下雨, 温度, 气温]: if keyword in user_request: for target in self._registry.values(): if keyword in target.tags: return target # 第二层宽松匹配寻找描述里包含请求关键词的Reachable for target in self._registry.values(): if any(word in target.description for word in [降雨, 气温, Weather]): if any(word in user_request for word in [雨, 冷, 热, 天气]): return target # 第三层LLM意图路由拼接所有描述请求模型决策 return await self._llm_route(user_request) async def _llm_route(self, user_request: str) - Reachable | None: descriptions \n.join( f[{target.name}] {target.description} for target in self._registry.values() ) prompt f根据用户请求从下列能力中选择最合适的一个直接返回能力名称 用户请求{user_request} 能力列表 {descriptions} 只输出能力名称。 # 这里接入你熟悉的LLM客户端即可 result await llm_complete(prompt) return self._registry.get(result.strip())这段代码里有几个设计细节我想强调一下。第一层规则路由用的是tags匹配所以我注册时必须对tags做覆盖式设计——“下雨”“气温”“天气”全部挂上宁可多挂不可漏挂。第二层宽松匹配是我后续加的它让描述文本在轻量层面上参与路由不消耗LLM成本。第三层才是真正的重武器。3.4 触达执行超时、重试与降级兜底路由找到目标后剩下的就是安全地执行触达。我写了_invoke_with_safety这个函数集中处理三类问题超时、幂等重试、降级兜底。async def _invoke_with_safety(self, target: Reachable, params: dict): # 参数校验严格按照schema过滤防止坏请求打到下游 if not validate_schema(target.input_schema, params): return {status: invalid_params, message: 参数不符合能力要求} try: result await asyncio.wait_for(target.callable(params), timeouttarget.timeout) return {status: ok, result: result, target: target.name} except asyncio.TimeoutError: # 超时后如果能力是幂等的考虑自动重试一次 if target.idempotent: try: result await asyncio.wait_for(target.callable(params), timeouttarget.timeout) return {status: ok, note: timeout_retry, result: result, target: target.name} except asyncio.TimeoutError: pass return {status: timeout, message: f{target.name} 触达超时, target: target.name} except Exception as e: return {status: error, message: f{target.name} 触达异常: {str(e)}, target: target.name}这段代码是我的“保命层”。真实环境中外部API慢上两三秒是家常便饭如果没有这层强制超时主控LLM就会一直在等一个永远不会回来的结果整个会话直接僵死。我在好几个项目里都是因为没写超时被坑惨的现在这层逻辑成了Agent-Reach注册任何新能力时不可删减的标配。3.5 最小可跑通Demo家庭助理的三种能力最后是我最常用的一个Demo把三个能力注册进同一个Hub模拟真实调用场景。async def schedule_agent(params: dict) - dict: return {event: params[event], time: params[time], status: created} async def light_agent(params: dict) - dict: return {device: 台灯, action: params[action], status: ok} hub Hub() hub.register(make_weather_reachable()) hub.register(Reachable( nameschedule_agent, description创建日程提醒用户会说明天下午3点开会、记得提醒我买牛奶。, input_schema{type: object, properties: { event: {type: string}, time: {type: string}}, required: [event, time]}, tags[日程, 提醒, 开会], )) hub.register(Reachable( namelight_agent, description控制家里智能设备开关用户会说把台灯打开、关掉客厅灯。, input_schema{type: object, properties: { device: {type: string}, action: {type: string, enum: [on, off]}}, required: [device, action]}, tags[灯, 台灯, 开关], )) # 模拟三类请求 print(await hub.dispatch(上海明天会下雨吗, {city: 上海})) print(await hub.dispatch(帮我记住周五下午开项目周会, {event: 项目周会, time: 周五14:00})) print(await hub.dispatch(打开卧室台灯, {device: 卧室台灯, action: on}))我给这个Demo的忠告是先在本地把这三条触达链路跑通再做真实集成。很多初学者一上来就接真实硬件、真实第三方API一旦出问题就分不清是网络问题、协议问题还是路由问题。先让Hub在一个完全受控的环境里稳定工作再加入真实外部依赖排查空间会清晰得多。4. 实测与踩坑触达链路里的四个“隐形凶手”框架写好后我拿家庭助理场景做了将近两周的实测。期间遇到的四个问题每一个都让我对“触达”这件事的复杂性有了新认识这里逐一拆给你看。4.1 路由命中率只有71%问题出在描述文本而不是模型能力第一轮实测下来Hub的路由命中率只有71%。这个数字很难看。我原本以为是LLM意图路由选错了后来一查日志才发现大量请求在第一层和第二层路由时就没走对直接被带偏了。根因是我一开始的description写成了“功能说明书”。比如天气服务我写的是“获取天气数据并返回气温和湿度”这句话对机器是有效的但对语义匹配是灾难——用户根本不会说“请获取天气数据”他们会说“今天下雨吗”“明天降温穿什么”这些说法没有一个词能和“获取”匹配上。我把描述全部重写了一遍规则很简单每个Reachable的description至少包含3个典型用户问法声明它对应的意图而不是功能。重写后路由命中率从71%一路涨到94%。这个反差让我意识到在Agent-Reach这类框架里路由的瓶颈往往不是底层模型而是我们对能力本身的“画像”画得够不够准。4.2 超时抖动一个API慢了3秒整个会话卡死第二轮测试我接了一个真实天气API。结果发现只要某个时间段第三方服务响应变慢主控Agent的整个对话流程就停住不动像是被谁按了暂停键。我当时的代码只在Reachable内部设了超时但主控Agent那层并没有统一的兜底。后来把所有超时控制全部上收到Hub层统一强制效果立刻不一样。超时兜底策略开始起作用首次超时返回错误幂等能力自动重试一次重试再失败就返回一条降级提示。这里要记住一个思路超时控制一定要在调度层做而不是依赖每个子Agent自觉。子Agent再多它们也不知道全局的SLA要求只有当Hub这个统一出入口掌握了超时重试这柄大锤整体触达的稳定性才有保障。4.3 上下文窗口的隐形消耗返回结果吃掉了模型注意力触达成功了只是噩梦结束的开始。第三个坑最隐性——我把子Agent返回的完整JSON原样塞回主控LLM的上下文一次两次无所谓十次二十次之后模型连最初的用户需求都“想”不起来了。原因是LLM的注意力是稀缺资源。天气子Agent返回的updated_at、humidity等字段对主控决策毫无用处但它们照样占着上下文窗口稀释了真正关键信息。我的解决思路分两步触达结果在进入上下文之前先做一趟压缩摘要只保留主控Agent做决策必需的核心字段对于特别长的结果不放进上下文而是存到外部存储只给主控Agent一个引用ID。这个改动之后上下文占用率肉眼可见地降了下来模型的决策准确率也随之回升。这里我给自己定了一条硬规矩任何Reachable的返回结果都必须有一个配套的summary函数否则不允许注册。4.4 循环触达两个子Agent在群聊里互相“捧场”最后一个问题是我在扩展多Agent互相触达模式时遇到的。两个子Agent接到同一个用户请求后开始来回调用彼此的能力A说需要B验证B说需要A补充上下文雪球越滚越大整个调度陷入环路。根因是没有触达次数的全局约束。修复方案很直接Hub为每次用户请求分配一个trace_id在每个Reachable的调用记录里增加hop计数一旦某条链路的触达跳数超过预设上限我设为5跳立即停止继续触达返回“需要人工接手”的兜底结果。就这么一个看似朴素的限制却彻底终结了循环问题。它提醒我触达层不光要保证“到得了”还得保证“回得来、停得下”。5. 选型对照什么时候该用Agent-Reach这类轻框架写到这里你可能会问市面上已经有那么多成熟框架了Agent-Reach还有必要吗我的看法是有但要看场景。我把Agent-Reach和三个主流框架放一起做了个详细对照方便你按图索骥。对比维度Agent-ReachLangGraphAutoGenCrewAI核心抽象Reachable、Route、HubStateGraph、Node、EdgeConversableAgent、GroupChatAgent、Task、Crew最擅长场景大量工具/子Agent的触达、路由、降级有状态、分叉明确的复杂工作流多Agent对话协商、群聊协作按角色分工的任务流水线触达控制粒度细路由策略、超时、重试完全由你控制中可在节点里写但无统一模型弱依赖对话对话式协商中偏向任务委派路由机制规则宽松匹配LLM意图三层路由图的边决定流程走向对话决定下一步任务依赖决定执行顺序可观测性内置trace_id和触达轨迹需自行hook靠回调靠回调学习成本低三个概念较高图模型复杂中需理解会话机制低API贴合直觉适合体量单机或小集群的能力触达层大规模生产工作流研究、探索性对话快速原型、组队任务我个人的选择决策清单大致是这样如果你要的是“多个Agent在对话中互相配合、协商完成复杂任务”AutoGen和CrewAI方向对的模型很强协作本身就是重点。如果你要的是“一个严谨的、有状态、可回溯的流程”LangGraph值得你投入学习成本它的图建模能把复杂流程焊死。如果你要的是“把散落各处的能力统一接入、稳定触达、精细调度”或者说你面对的瓶颈是工具太多、调用太乱、出错后不好排查——那就适合Agent-Reach这类轻框架。判断标准其实是你的核心痛点到底是“合作”还是“触达”Agent之间聊得不热闹是合作问题该用重框架能力掉线、调用超时、路由选错是触达问题重框架帮不了你多少轻框架反而一针见血。6. 后续演进把Agent-Reach从项目原型做成通用触达层最后聊聊这个项目的下一步方向。Agent-Reach目前在我本机跑得很稳但我心里很清楚它离一个真正“通用”的触达层还有一段路接下来有三个方向是明确的。6.1 可观测性升级每次触达都有迹可循复盘这轮开发让我受益最多的是给每次触达都加了一个trace_id从用户请求进入Hub开始到路由决策选的是哪一层、命中哪个Reachable、耗了多少毫秒、返回什么结果、中间是否重试、是否超时全部记录下来。这个记录一开始是为了排错后来它成了优化路由策略的重要依据。没有可靠的触达轨迹你永远只会说“感觉不太对劲”有了它你就能精准说出“昨天有37%的请求走到第三层LLM路由说明标签体系得重新设计”。下一步我打算基于轨迹加一个轻量看板让路由决策越来越透明。6.2 多Agent互触达从星型到网状目前的Agent-Reach是典型的星型结构所有能力都挂在Hub上请求从主控Agent进出。但真实应用里会有这样的情况——一个子Agent也需要触达另一个子Agent的能力。扩展方向并不复杂让每个Agent同时扮演“Reachable”和“调用方”两个角色。下级Agent发出的触达请求同样经过Hub做路由和鉴权这样既能控制每个Agent的触达范围又能打破单向星型的局限。核心约束是在Hub层给每个Agent配一份权限清单能触达什么、不能触达什么都要显式声明。6.3 版本灰度新能力旧能力和平共处最后一个演进方向是版本兼容。Reachable注册时带了version字段后续就可以做灰度触达新注册一个能力时先挂10%的流量跑一段时间确认稳定性再把旧版本切掉。这个机制对生产环境非常重要因为触达层一旦上线背后可能连着几十个真实服务全量升级的代价太高。我目前的实现是在Hub的路由层加一个version_selector根据版本号比例决定把请求分给哪个版本。逻辑不复杂但它让“更新一个能力”从一次高风险发布变成了一个可控的渐进过程。项目做到这里我最大的体会是Agent-Reach这类框架的价值不在于代码有多华丽而在于它逼着我把“触达”当成一个正经工程问题来看待——定义统一接口、设计路由策略、做超时兜底、记录全链路轨迹。这一整套方法论比代码本身更值钱。最后再分享一个小习惯每注册一个Reachable我都强制自己写清楚“用户会怎么问”而不是只写“这个功能做什么”。这个小习惯拯救了我后面几乎所有路由问题也建议你从第一个能力注册就开始坚持。
阅读完成 · 觉得有帮助？
