Agent-Reach这个名字最开始并不是我们拍脑袋起的。团队把十几个Agent接入生产环境之后我发现自己面临一个特别尴尬的境地每个Agent单独拉出来都能跑通Demo但一旦组合起来没有人能说清楚——系统里今天到底有多少能力在线哪些能力其实已经静默失效某一次用户请求应该交给哪个Agent又该在什么时候放弃重试我们试过在代码里写死路由试过靠群聊里的口头约定来维护Agent清单结果都不理想。于是就有了Agent-Reach一套让Agent的能力触达范围可见、可配置、可治理的连接层。如果你也在多个Agent之间做调度和编排这篇文章应该能给你一些可复用的思路。1. 为什么会有Agent-Reach从Agent调度的痛点说起1.1 我遇到的三个真实问题先说说动手之前的背景。我们的平台接入的Agent形态很杂有基于大模型封装的通用助手有专门做SQL查询的数据库代理有负责工单流转的流程Agent还有对接第三方供应商API的只读型Agent。最初大家各自开发、各自对接谁需要调用另一个Agent就在代码里直接声场一个客户端实例。这种直连模式在Agent数量少的时候没有问题但涨到十几二十个之后问题开始集中爆发。第一个问题是硬编码。某个Agent需要调用另一个Agent的能力时必须在调用方代码里写清楚地址、密钥、超时时间和参数格式。每次有新的Agent上线就要有人去改调用方的代码每次有Agent下线或者更换服务地址又要全链路通知一遍。我统计过一个星期内因为Agent地址变更导致的线上报错至少有七次是可以通过集中注册解决的。第二个问题是能力边界不透明。因为没有人维护一份动态清单A Agent不知道B Agent其实已经具备了某项技能于是把本该转给B的请求自己硬扛结果生成质量很差或者直接拒绝。反过来C Agent明明只能查近30天的数据调度层却把需要查历史全量的任务派给C最终返回一个看似完整、但实际缺失数据口径的结果。这种错误非常隐蔽比直接报错更危险因为用户不会立刻感知到结果有问题只会慢慢失去信任。第三个问题是没有统一的降级策略。外部API一抖动所有依赖它的Agent一起超时日志里全是超时重试的堆叠下游服务被打到限流然后超时时间又被调大进一步拖垮系统。我们甚至出现过一次因为重试策略冲突导致的雪崩A Agent超时后重试了3次每次重试又触发B Agent的重试最后同一个请求在五分钟内被重复执行了几十次产生了一笔不小的成本账单。1.2 Agent-Reach要解决的边界做了几次线上事故复盘之后我意识到问题的核心不是哪一个Agent实现得不好而是Agent之间缺少一层统一的“连接控制面”。我需要的不是把Agent逻辑重写一遍也不是做一个新的Agent编排框架而是一个轻量的中间层它至少能提供三件事让每个Agent主动声明自己能做什么、不能做什么、依赖什么让调度器基于声明和实时状态做路由而不是靠代码硬编码让每一次触达都有记录可以被回溯、被统计。我把这个中间层命名为Agent-Reach。Reach这个词想表达的是“触达范围”一个Agent到底能触达哪些工具、哪些数据源、哪些业务流程这个信息在Agent系统里极具价值。过去它散落在各处的配置文件和代码注释里Agent-Reach要做的就是把这些信息集中起来变成可执行的路由依据。需要说明的是Agent-Reach刻意避开了几个边界它不去管大模型本身怎么推理不提供提示词工程能力也不涉足Agent训练。它只关心一件事当一个任务到达时如何根据当前所有Agent的能力状态找到一条最合理的触达路径并且把这条路走完、走好。2. Agent-Reach的核心设计触达层、调度层、反馈层2.1 触达层定义能力而不是定义功能Agent-Reach的第一个模块是触达层解决的是“声明”的问题。一开始我们想用功能点来定义比如给每个Agent列一个功能清单写上“支持查天气”“支持查新闻”“支持数据库问答”。但很快发现功能点描述有两个缺点第一无法做推理查天气和查天气预警是两种能力还是同一种能力写的人和一个箱子读的人读出另一个意思第二功能描述颗粒度太粗根本没有办法判断某个Agent能不能处理多轮对话的上下文。后来我参考了云资源标签的思路改用“能力声明”的格式。每个Agent在上线时必须提供一份能力描述文件里面包含三个关键部分基础信息、能力标签、约束条件。# capabilities/db-agent.yaml name: db_agent version: 2025.03.01 capabilities: - type: database_query mode: readonly engines: [mysql, postgresql] time_range: min: 30d max: 365d permissions: - select - aggregate rejected_permissions: [drop, update, delete, insert] - type: natural_language_to_sql supports_multiturn: true complexity: medium constraints: max_concurrent: 10 max_latency_ms: 3000 cost_priority: low data_scope: internal_warehouse你能看到这里刻意强调了约束条件。permissions和rejected_permissions是我们后来补的因为在真实环境里一个只读Agent被分配了写入任务轻则任务失败重则产生脏数据。声明“我不做什么”和声明“我能做什么”同等重要这是我踩过坑之后才真正理解的一句话。2.2 调度层从“找Agent”变成“找能力网络”有了能力声明之后下一步是调度。Agent-Reach的调度层不再是“把任务发给某个具体的Agent”而是“根据任务需求匹配能力网”。我把任务本身也声明一下再和Agent的能力声明做匹配。# task example: user request task { intent: database_query, engine: postgresql, time_range: 90, permissions: [select], mode: readonly, complexity: low }匹配逻辑我会在第三章详细展开这里先讲一下设计思路我们采用“分层匹配”先按硬性条件过滤引擎类型、数据范围、权限再按软性条件打分延迟偏好、成本偏好、并发余量。硬性条件不满足就直接排除避免Agent接受自己根本不胜任的任务软性条件打分则是为了在多个候选里选出最优解。同时调度层还维护了一张“能力-依赖关系图”某个Agent依赖外部服务X外部服务X当前健康度如何直接影响了这个Agent的可用性。调度器在做任务分发之前会先检查依赖链的健康状态。如果外部服务正在抖动调度器就能提前把流量切换到备用Agent而不是等超时之后再补救。2.3 反馈层把每次触达变成数据讲完触达和调度这里必须提一下反馈层因为Agent-Reach这个项目能持续做下去全靠反馈层的数据支撑。每个任务从进入调度器到最后返回结果都会生成一条事件记录。最开始我们只记录了失败异常后来发现远远不够。一条完整的事件记录应该包含任务ID、发起方、进入时间匹配到了哪些候选Agent、匹配分数、最终选中的Agent实际执行耗时、重试次数、降级动作Token消耗、API费用估算输出结果的质量评分人工反馈或规则评判是否触发了熔断、是否产生了告警。这个设计一开始让我觉得重但实际跑起来之后价值非常明显。运营Agent系统最怕的就是不知道成本和效果花在哪儿了。有了反馈层我们可以在一个面板上准确看到上周调用量最高的三个Agent是谁、成功率最低的是谁、每一类任务的平均响应耗时和平均成本是多少。这些数据反向驱动我们调整调度策略形成了一个闭环。3. 关键实现细节一套可以直接改来用的骨架这一章我给出Agent-Reach在实现层面的核心骨架。为了讲清楚我用Python写了一套简化版的示例代码。你在自己的项目里不一定照搬但核心思路可以直接迁移。3.1 注册中心让每个Agent“自我声明”Agent-Reach的最低一层是注册中心。每个Agent在启动时通过一个Provider接口注册自己的能力声明注册中心把这些声明保存在内存中并维护一个基于标签的索引。# registry.py from typing import Dict, List class AgentProvider: def __init__(self, agent_id: str, capability: dict): self.agent_id agent_id self.capability capability class Registry: def __init__(self): self._providers: Dict[str, AgentProvider] {} def register(self, provider: AgentProvider) - None: self._providers[provider.agent_id] provider # 更新索引为每一种能力标签建立agentId索引 for cap in provider.capability.get(capabilities, []): key cap[type] index self._label_index.setdefault(key, set()) index.add(provider.agent_id) def unregister(self, agent_id: str) - None: self._providers.pop(agent_id, None) def find_by_capability(self, cap_type: str) - List[AgentProvider]: agent_ids self._label_index.get(cap_type, set()) return [self._providers[aid] for aid in agent_ids if aid in self._providers]这里有个容易被忽略的细节注册中心一定要提供“心跳过期”机制。Agent虽然是程序但也会因为网络分区、进程被杀等原因失联如果不做心跳淘汰调度器会把任务派给一个已经死掉的Agent然后白白等超时半天。我们在AgentReach里引入了30秒一次的心跳超过90秒没有心跳就自动把Agent标记为offline不再参与路由。3.2 路由匹配规则优先语义兜底路由匹配是整个系统最核心的模块。我用了两层策略规则硬过滤和软性评分。# router.py from dataclasses import dataclass dataclass class TaskRequirement: intent: str engine: str time_range: int 0 permissions: list None mode: str readonly complexity: str low class HardFilter: staticmethod def passes(task: TaskRequirement, capability: dict) - bool: # 引擎必须匹配 if task.engine and task.engine not in capability.get(engines, []): return False # 时间范围必须在能力范围内 if task.time_range: tr capability.get(time_range, {}) if task.time_range tr.get(max, 0): return False # 权限必须满足不能要求一个只有select权限的Agent去做update if task.permissions: req_set set(task.permissions) allowed set(capability.get(permissions, [])) if not req_set.issubset(allowed): return False return True class SoftScorer: staticmethod def score(task: TaskRequirement, capability: dict) - float: score 0.0 # 偏向并发余量大的Agent score capability.get(max_concurrent, 0) * 0.3 # 偏向延迟低的Agent score 10 - capability.get(max_latency_ms, 10000) / 1000 # 偏向成本优先级匹配的Agent if capability.get(cost_priority) low: score 2 return score实际匹配流程是先用HardFilter把所有候选Agent过滤一遍再用SoftScorer对剩余候选排序取分数最高的。代码很简单设计上的关键是硬过滤条件一定要足够严格。我宁可因为过滤条件太严导致“无候选Agent”也不希望因为过滤太宽而把任务派给一个不合适的Agent。前者我们还可以触发降级策略或者直接返回失败后者则会产生一个错误的成功结果排查起来要花大得多的成本。3.3 执行层的并发、超时与降级路由确定了目标Agent之后剩下的是执行。这个阶段最大的风险是外部依赖抖动所以Agent-Reach的执行层做了一套完整的超时和降级机制。# executor.py import asyncio class Executor: def __init__(self, timeout5, max_retries3): self.timeout timeout self.max_retries max_retries async def run(self, agent_provider, task) - dict: attempt 0 while attempt self.max_retries: attempt 1 try: result await asyncio.wait_for( self._invoke(agent_provider, task), timeoutself.timeout ) return {ok: True, data: result, attempts: attempt} except asyncio.TimeoutError: if attempt self.max_retries: return {ok: False, reason: timeout, attempts: attempt} except Exception as e: return {ok: False, reason: str(e), attempts: attempt} async def _invoke(self, agent_provider, task): # 实际调用Agent内部逻辑通过标准接口 return await agent_provider.handle(task)这里有一个我不会写进短代码示例、但真实环境里必须考虑的点重试要有指数退避并且要有最大累计成本上限。如果一次任务的调用成本高比如需要几百个Token的推理无限重试会带来不可控的费用。我在执行层加了一个“成本预算”参数重试一次就扣减预算预算耗尽后即使还没达到最大重试次数也直接走降级分支。降级分支一般有几种降级到另一个相似能力的Agent降级到规则模板或者给用户返回一个明确的“当前暂时无法处理”的响应。需要强调的是降级到规则模板要谨慎——它看起来能兜底但生成质量差异很大用户会明显感知到响应变“笨”。我现在的策略是规则模板只在低风险任务上使用涉及金额、合同、对外承诺这类敏感场景时宁可拒绝也不要降级。4. 真实环境下的效果数据与问题复盘4.1 一组有代表性的指标Agent-Reach在我们内部跑了大概两个月以下是第二个月的一组关键指标和接入前做了对比指标接入前接入后路由准确率任务最终找到正确Agent的比例74%91%P95响应耗时2800ms1900ms因Agent地址变更导致的线上故障周均6次接近0重复执行/重复调用事故每月4-5次0单次任务平均Token成本基准值下降约22%路由准确率的提升主要来自能力声明的硬过滤。过去很多任务被发错了Agent要么因为调用方根本不知道还有个更合适的Agent要么因为代码里硬编码了旧地址。现在调度器每次都基于最新的注册信息做匹配那些“先入为主”的调用路径就被自动纠正了。响应耗时的下降则和“健康度分流”有关外部依赖一抖动流量就会自动切到健康节点而不是继续排队等一个大概率失败的超时。4.2 踩过的三个坑数据好看但过程远没有这么顺畅。这个项目我们走过不少弯路挑三个最典型的说一下。第一个坑是能力描述写得过细导致标签体系迅速膨胀。我们第一版放开让团队自由填标签结果一周后注册中心里出现了上百种标签其中“查询”“数据查询”“数据库查询”“查数据”表达的是同一个意思但并存的标签就占了十几个路由匹配经常漏掉正确候选。后来我们定了两级标签体系一级标签是官方规定的能力类型一共只有二十多个强制选择二级标签才是自定义描述允许补充细节但不会直接影响路由。这算是用“有管理的自由”解决了分类爆炸的问题。第二个坑是超时时间设置过于激进。最开始我们希望所有请求都在2秒内返回所以把超时都设成2秒。结果数据好看实际用户反馈却变差了——因为很多Agent内部的推理本身就需要3秒以上激进超时导致大量请求被提前掐断然后重试最终既没有省时间还白白增加了重复计算成本。后来我们根据每个Agent的实测P95耗时动态设置超时P95耗时小的Agent超时设成P95的两倍P95耗时大的Agent单独放宽不为统一指标牺牲真实质量。第三个坑是反馈层的数据链路一开始很松散。我们让各个Agent在执行完任务之后自行上报事件结果有些Agent忘了上报有些Agent上报的事件ID对不上导致反馈面板统计出的成功率和真实情况偏差很大。反思之后我们把事件上报从“Agent主动上报”改成了“执行层统一记录”任务进出执行器的时候自动打点Agent只负责填充结果数据和质量评分其他信息都不需要Agent关心。这样虽然改动大一些但数据完整性提升非常明显。4.3 现在的定位与后续规划Agent-Reach目前在我们团队里的定位不是一个大而全的平台而是一层“轻量化Agent连接协议”。它能帮我们解决能力注册、路由分发、执行反馈这些底层的共性问题而每个Agent怎么实现、用哪个模型、走什么提示词框架Agent-Reach一概不管。这种边界感很重要——如果什么都管这个项目可能到现在还躺在设计文档里。后续我们计划做三件事第一把策略配置从代码里抽出来做热更新让运维同学可以在不重启服务的情况下调整路由权重和降级策略第二建设一个公共能力池把一些高频的通用能力比如格式化、聚合统计、数据脱敏从具体Agent里抽出来让所有Agent共享触达第三进一步完善成本预算功能把每次触达的费用和业务价值关联起来帮助我们判断哪些Agent值得加大调用量、哪些该降级处理。最后说一点个人体会。Agent-Reach的核心价值不在于它用了多复杂的算法而在于它逼着我们提前想清楚了一个问题一个Agent接进来到底能做什么、不能做什么、边界在哪里。在Agent数量少的初期这类信息靠文档和记忆还能勉强维护一旦规模上来没有一套机制去统一声明和检查系统就会慢慢失控。先把触达范围管起来再去谈智能和编排这个顺序我希望你也能试一试。