Agent-Reach：为大模型打造安全可控的工具触达层

前阵子有个朋友问我你给大模型接了多少个工具了我说二三十个吧。他接着问那你怎么管的我一下愣住了。说实话最早一版就是写 if-else模型吐出函数名应用层照着调函数。一开始确实挺爽直到工具数量涨到两位数、用 Agent 的用户也开始变多我才意识到问题根本不在“能不能调”而在“怎么管”。Agent-Reach 这个项目就是在这个背景下折腾出来的——它是 Agent 工程化过程中的一套能力触达层capability reachability layer核心解决的事情非常聚焦让大模型驱动的应用可以安全、可控、可观测地触达真实世界里的工具、接口和数据。如果你正在做 Agent 相关开发尤其是已经过了“跑通 Demo”的阶段开始思考工具多了怎么办、权限怎么分、线上怎么排障这类问题这篇文章应该能帮你少走不少弯路。下面我把这套东西从动机、架构、代码实现到踩坑经历完整过一遍。1. 为什么需要 Reach 层大模型有“大脑”但没有“手”1.1 模型的输出本质上只是“建议”很多人第一次接触大模型工具调用时都有一个误解觉得模型“会”调用工具。实际上模型本身是一个文本进文本出的系统你给它一段 Prompt它返回一段文本。所谓 Function Calling 也好、Tool Use 也好本质上只是让模型在回答里附带一个结构化的 JSON例如{function: search_orders, params: {user_id: u_123}}。真正执行这个函数、去查数据库、去请求第三方 API 的是应用层代码。也就是说模型的输出只是一个“意图”能不能落成真实世界里的操作取决于应用层的执行链路。这条链路我称之为 Agent 的“手脚”。问题在于大部分人刚开始做这条链路时用的都是最简单的方式在代码里写死一个函数映射表。模型说要调 A你就调 A模型说要调 B你就调 B。在工具数量不超过十个的时候这种方式完全够用。1.2 硬编码工具调用的三个天花板但当一个 Agent 需要接入二三十个工具并且这些工具背后连接着不同的系统、归属于不同的团队、对不同用户有不同的开放权限时硬编码方式会很快撞上三个天花板。天花板具体表现维护天花板每新增一个工具都要改代码、改提示词、重新测试发布。需求方排着队等你发版一个工具上线周期可能拖一两周。权限天花板权限逻辑散落在 if-else 里权限一变就要改代码。多租户场景下什么样的用户能调什么工具代码层面根本管不过来。可观测天花板Agent 哪次对话调了哪些工具、传了什么参数、执行结果如何没有统一出口。日志各家系统自己打排障的时候翻半天还拼不出一条完整链路。这三个天花板的本质是把“接入工具”和“治理工具”这两件事混在了一起。接入是必要的但治理才是工具多起来以后的核心矛盾。你需要的不是一个又一个 if-else而是一个独立的层专门负责能力的注册、发现、路由和管控。这就是 Agent-Reach 存在的意义。1.3 Agent-Reach 的定位和它做的三件事Agent-Reach 不是一个模型也不是一个 Agent 框架。它处于模型和应用系统之间像一个调度中枢加门禁系统。它的名字里 Reach 就是“触达”的意思——模型负责思考Reach 负责让思考变成行动而且是在受控范围内的行动。具体来说它做三件事。第一是注册所有工具、API、数据源统一抽象成“连接器”Connector以声明式的方式登记自己的职责、参数、权限、超时等属性。第二是路由意图进来之后引擎根据语义找到最合适的连接器并且按需加载工具描述避免把几十个工具的完整文档一次性塞给模型。第三是治理策略引擎在路由和执行之前做授权判断、限流、审计对高风险操作挂起等待人工确认。这三件事拆开看都不复杂但合在一起就构成了一条完整的“能力触达链路”。接下来我从架构层面讲清楚这条链路是怎么运转的。2. Reach 引擎的核心骨架注册表、路由器、连接器的协作流程2.1 四个核心组件的职责划分Agent-Reach 的运行时核心由四个组件构成注册表Registry、路由器Router、策略引擎PolicyEngine和执行器Executor。它们各管一摊职责划分非常明确。组件职责一句话总结Registry维护所有已注册连接器的元数据提供能力发现接口“有什么能力”Router接收意图从注册表中筛选匹配的候选能力并做最终路由决策“选哪个能力”PolicyEngine在路由前后做授权、限流、条件判断拦截未授权请求“能不能用它”Executor负责真正调用连接器处理超时、重试、并发控制、结果封装“怎么执行它”状态放在 Registry决策放在 Router约束放在 PolicyEngine执行放在 Executor。这四者各司其职你才能在任何一环出问题时单独排查、单独升级而不是面对一坨纠缠在一起的代码。2.2 一次完整的 Agent 触达流程一次完整的触达从大模型产生意图开始到结果回到模型上下文结束一共经过九个步骤模型根据系统提示词和用户对话输出一个结构化意图 JSON包含intent和params两个字段。Router 接收这个意图将其转换为内部统一的事件对象。Router 调用 Registry 的 discover 接口基于意图语义和参数特征做能力发现得到一组候选连接器。PolicyEngine 对候选列表做预过滤剔除当前上下文无权调用的连接器并记录拒绝原因。如果候选仍然超过一个Router 将候选列表的“精简摘要”交给模型做最终选择或者用规则评分自动裁决。选定连接器后PolicyEngine 再次做执行前授权校验这一步是最终防线。Executor 调用连接器应用超时、重试、并发信号量等策略。执行结果封装成统一的 ReachResult回填到模型的上下文让模型基于真实结果继续推理。全链路日志写入审计存储关联 trace_id后续可以完整回放。第 5 步是一个容易被忽略但极其关键的设计当工具数量很多时你不能把全部工具的完整描述都塞给模型做选择否则 Prompt 会被撑爆。正确的做法是先粗筛、再精排只把候选工具的摘要交给模型做最终判断。2.3 最小实现几十行代码跑通 Reach理论讲完得动手。一个最小可用的 Agent-Reach 骨架其实不用写多少代码。下面是我在项目里用的核心抽象去掉业务细节后的简化版import asyncio import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any, AsyncIterator class ReachConnector(ABC): 所有连接器的基类一个工具、一个API、一个数据源统一实现这个接口。 name: str description: str keywords: list[str] [] timeout: float 5.0 scopes: list[str] field(default_factorylist) abstractmethod async def execute(self, params: dict[str, Any], ctx: dict[str, Any]) - Any: ... class ReachRegistry: 连接器注册表登记能力、提供发现接口。 def __init__(self): self._connectors: dict[str, ReachConnector] {} def register(self, connector: ReachConnector) - None: self._connectors[connector.name] connector def discover(self, intent: str, params: dict[str, Any]) - AsyncIterator[ReachConnector]: # 基于关键词做一次粗筛 intent_lower intent.lower() for conn in self._connectors.values(): if intent_lower in conn.name.lower() or any( kw in intent_lower for kw in conn.keywords ): yield conn class PolicyEngine: 策略引擎授权判断未授权直接拒绝。 def authorize(self, connector: ReachConnector, ctx: dict[str, Any]) - bool: # 简化逻辑检查用户上下文中的 scopes 是否覆盖连接器所需 scopes user_scopes set(ctx.get(user_scopes, [])) return user_scopes.issuperset(connector.scopes) class ReachRouter: 路由器发现 - 过滤 - 路由决策。 def __init__(self, registry: ReachRegistry, policy: PolicyEngine): self.registry registry self.policy policy async def route(self, intent: str, params: dict[str, Any], ctx: dict[str, Any]) - Any: candidates [c for c in self.registry.discover(intent, params)] allowed [c for c in candidates if self.policy.authorize(c, ctx)] if not allowed: raise PermissionError(no allowed connector for intent: intent) # 简化为规则裁决优先选择名称/关键词匹配得分最高的连接器 selected max( allowed, keylambda c: (1 if intent.lower() in c.name.lower() else 0) ) try: result await asyncio.wait_for( selected.execute(params, ctx), timeoutselected.timeout, ) except asyncio.TimeoutError: return {error: timeout, connector: selected.name} return {connector: selected.name, result: result}这段代码去掉注释大概六七十行一个能跑通“意图发现 权限过滤 超时控制”的最小 Reach 引擎就出来了。你可以在上面继续加向量检索、重试逻辑、审计日志但这些核心抽象的方向是对的连接器面向接口编程注册表只负责登记和发现路由器只管决策权限作为独立关卡嵌在路由过程中。3. 连接器生态与动态注册机制让能力变成可插拔的资源3.1 为什么动态注册远比硬编码好维护早期我把工具直接写在业务代码里每接一个第三方 API 就要在 Service 层加一个方法。后来工具多了我发现真正的痛苦不是写调用逻辑而是需求方永远在等排期。一个运营同学过来说“帮我接一个查询物流的接口”本来接口文档都备好了但还是得等我改代码、发版明明能五分钟做完的事情要拖一个迭代。动态注册机制解决的就是这个协作问题。我把连接器做成一个可独立开发、独立部署的单元任何团队按照约定写一个连接器类在启动时通过一行代码注册到 Registry 里这个能力就“上线”了。发布不再依赖整个应用的版本节奏也不需要改 Agent 的系统提示词。这就是把接入从“开发任务”变成了“配置任务”。这个思路类比一下就是插座协议电器厂商不用知道你家电视里有什么电路只要按国家标准做插头插上就能用。Agent-Reach 里的连接器协议就是那个国家标准。3.2 连接器的声明字段一份 Schema 说清所有事一个标准连接器我建议至少声明以下字段{ name: order_query, description: 根据订单号查询订单详情返回状态、金额、物流信息, owner: order-team, version: 1.2.0, input_schema: { type: object, properties: { order_id: { type: string } }, required: [order_id] }, output_fields: [order_status, amount, logistics], scopes: [order:read], timeout: 3.0, retry: { max_retries: 2, backoff: 0.5 }, semaphore_limit: 5 }这里input_schema非常重要它既是给模型看的参数契约也是执行前校验器。模型生成参数后Reach 会先做一次 JSON Schema 校验不合法直接返回格式化错误让模型重新生成而不是把坏参数传到业务系统。description和keywords是给注册表做发现用的scopes是权限声明timeout和retry是执行策略semaphore_limit是并发上限。有一点要提醒Schema 不要写得太死。我早期把参数类型校验写得很严比如金额字段强制number但模型理解“一百二十块”变成120没问题可它偶尔会生成120.00元这种字符串。一旦 Schema 过严你会发现大量合法调用被误杀。现在的做法是核心校验只保证类型和必填字段其余交给业务侧做语义校验给模型留出容错空间。3.3 插拔带来的新问题健康检查、版本和灰度动态注册解决了上新慢的问题但引入了新的问题你怎么知道一个连接器是好用的我遇到过最离谱的情况是一个连接器注册上去了但底层依赖的服务早就下线了每次调用都报 500。模型在对话里反复尝试用户体验极差。所以注册表必须配合健康检查机制运行。我的实现方式是给 Registry 加一个心跳字段连接器每次成功执行后上报耗时和错误率Registry 定期扫描连续 20 次调用中错误率超过 60%自动将该连接器标记为 degraded。路由时优先跳过 degraded 连接器除非没有替代能力才降级使用。运维侧可以手动摘除一个连接器摘除后新请求不再路由到它但进行中的调用不受影响。版本管理上引入了连接器的version字段同时支持多版本共存。灰度时把新版本注册为query_order_v2用路由权重把 10% 的请求切过去观察稳定性后再逐步放量。这比改代码回滚方便太多了。4. 权限边界让 Agent 够得着但不乱碰4.1 从“能用”到“只能给你用”一个 Agent 能调用工具和能安全地为特定用户调用工具是两码事。我在做 Agent-Reach 之前走查过一段自己的代码发现一个非常惊险的问题我提供给用户的 Agent 工具列表里有个“删除用户”的功能Model 在权限判断上完全没有拦截也就是说任何一个登录了 portal 的普通用户理论上都能通过构造意图让 Agent 去执行删除操作。这本质上是把我自己的操作权限直接给了所有用户。正确的原则应该是Agent 的能力边界必须继承“当前宿主用户”的权限边界甚至要比宿主权限更保守。也就是说用户没有权限调用的接口Agent 也不能替他调用。当时就想如果没有一个集中的授权层这种风险根本防不住。这就是为什么我把 PolicyEngine 设计成路由链路里不可跳过的一环。4.2 策略引擎的三级维度资源、动作、条件Agent-Reach 的策略模型我按三个维度组织资源维度连接器操作的资源是什么例如order.api、db.users、oss.file。动作维度对资源做什么例如read、write、execute、delete。每个连接器在编写时声明自己的动作类型审计和授权都依赖这个声明。条件维度除了“能不能”之外还看“当前场景允不允许”例如时间窗口、调用来源 IP、用户风险分、是否是敏感数据。条件维度特别适合做细粒度控制。比如“订单查询”这个连接器普通用户可以查自己的订单客服角色可以查全量订单运营角色可以加一点限流。这些在策略里写成 JSON{ policies: [ { id: order_read_rule, effect: allow, resource: order.api, action: read, conditions: { user_role: [customer, support, ops], rate_limit: 100 } }, { id: order_delete_rule, effect: deny, resource: order.api, action: delete, conditions: { user_role: [customer] } } ] }这套策略在 PolicyEngine 中被解释执行完全不用改业务代码。要加一条规则更新策略配置即可配合配置中心下发能做到秒级生效。4.3 高风险操作的兜底人工确认回路策略引擎能挡住明确的越权操作但挡不住“合法但危险”的操作。比如一个管理员角色确实有权删除生产环境数据但如果 Agent 因为上下文误判主动执行了删除呢这就是必须引入 human-in-the-loop 的原因。我对高风险连接器做了一个requires_approval标记。当 Router 匹配到这类连接器时不会直接进入 Executor而是生成一个待确认请求携带approval_token推送到审批通道。用户看到的是“Agent 正在请求执行 XX 操作”确认后审批通道回调 Executor超时未确认请求自动作废。这个机制同时解决了两个问题一是给关键操作留一道人工闸门二是让 Agent 的自主性边界变得清晰可控。5. 实测中的瓶颈与调优经验从跑通到抗住5.1 工具描述别一股脑塞给模型一开始我把所有连接器的完整描述拼成一段超长 system prompt想着让模型“知道所有工具”。上线后发现两个问题一是 Prompt 长度暴涨Token 成本飙升二是模型在长上下文里的注意力会被稀释经常在工具选择上犯低级错误比如明明有个精确匹配订单号的工具它偏要选一个模糊查询的。解决办法就是我前面说的“两段式描述”。Reach 层在路由前只给模型一份摘要索引每条大概是一句话加参数列表可用能力 - order_query: 根据订单号精确查订单详情。参数: order_id - logistics_trace: 查物流轨迹。参数: order_id, carrier - coupon_validate: 校验优惠券有效性。参数: coupon_code, amount模型要选哪个Reach 再把对应连接器的完整描述和 Schema 拉到上下文里。这相当于从“百科全书”变成了“目录 按需查阅”实测工具选择准确率反而提升了Token 成本降了接近一半。5.2 超时和失败语义别让 Agent 卡死在第三方接口Agent 的推理循环依赖工具结果如果某个连接器迟迟不返回模型就会一直空转等待用户感知就是“机器人卡住了”。我吃了好几次亏之后把超时策略改成了连接器维度可配置。外部的 API 服务给 3 到 5 秒内部数据库查询给 8 秒需要跑批的任务直接设计成异步模式先返回“任务已提交”再通过查询接口拿结果。同时设置重试策略只对幂等连接器开启自动重试非幂等的坚决不重试否则会造成重复扣款之类的事故。5.3 并发隔离一个连接器的雪崩不能让全员陪葬Agent 同时在线时会出现某个热门连接器被打爆拖垮整个路由进程的情况。我在 Executor 里给每个连接器加了一个独立的信号量控制它的最大并发数class SemaphoreConnector(ReachConnector): def __init__(self, inner: ReachConnector, limit: int): self.inner inner self.sem asyncio.Semaphore(limit) async def execute(self, params, ctx): async with self.sem: return await self.inner.execute(params, ctx)当一个连接器超过并发上限新请求快速失败并返回“当前能力繁忙”而不是无限排队。代价是少部分请求失败但保护了整体稳定性。对依赖方来说一次失败还能重试总比所有 Agent 一起挂掉强。5.4 排障的核心全链路 trace 贯穿意图到执行工具多了以后排障最大的困难是“这一轮模型为什么不选 A 工具而是选了 B”。我在 Reach 的入口处生成一个trace_id从模型意图产生、能力发现、策略过滤、路由决策到连接器执行、结果回填全部在日志里带上这个 ID。这样一次会话的完整链路可以像看剧本一样回放快速定位是发现环节漏了还是权限环节拒了还是执行环节超时了。实际操作中这类排查日志帮了大忙。有一次用户反馈“同样的订单有时候能查到有时候查不到”追踪下来发现问题不在 Reach而在底层订单库项目切换了不同分片连接器的筛选参数没带上分片键。没有全链路 trace这种问题靠肉眼排查会非常奔溃。如果你也准备把 Agent 从 Demo 推向生产我建议你第一步不是选什么框架而是先把“能力触达层”想清楚。模型会越来越聪明但工具接入、权限管控、可观测性这些工程问题不会自己消失。Agent-Reach 这套思路做下来我最深的体会是好的架构不是追求复杂的技术而是把原本纠缠在一起的关注点拆成边界清晰、各自可治理的模块让 Agent 在足够大的空间里安全地够到它需要的东西。