Agent-Reach：智能体工具调用与能力触达层实践

1. Agent-Reach 到底解决什么问题Agent-Reach 这个名字我第一眼看到的时候会心一笑——它戳中的正是做大模型应用的人最疼的那块地方模型能说会道却常常够不着外面的世界。Reach 这个词用得挺准不是调用不是集成而是触达。调用是技术动作触达是业务结果。一个 Agent 真正跑起来卡住它的往往不是模型不够聪明而是它在需要的那一刻拿不到数据、摸不到系统、推不动流程。我把它定位成一个面向智能体的能力触达层把散落在各个系统里的接口、脚本、数据源、内部服务统一注册成模型能理解、路由能分发、权限能管控、审计能追溯的能力单元。它不训练模型不替代编排框架也不抢前端交互的活它只做一件事——让 Agent 说出的每一个意图都能落到一个真实可执行的终点上。这套东西适合谁看如果你正在做企业内部的知识助手、自动化运维助手、客服工单助手、数据分析助手或者任何需要让模型动手而不是动嘴的场景那这里面的取舍和踩坑对你基本都能复用。哪怕你现在只是用一两个接口做个小工具先按这套思路把注册表设计对后面扩展时能省掉大把重构。2. 整体架构设计与选型思路2.1 为什么不做成大一统的 Agent 框架我见过太多团队一开始就想着做平台注册中心、编排引擎、记忆管理、向量库、前端面板全套上。结果三个月过去能演示的还是一个天气查询。Agent-Reach 的设计原则正好相反——只解决触达不解决思考。理由很实际。模型层和编排层这两块变化太快今天流行的工作流范式半年后可能就被新东西替代。你要是把编排逻辑焊死在自己的框架里模型一升级你就得跟着改。而触达这层的抽象相对稳定一个能力有名字、有描述、有入参出参、有权限、有超时这套东西五年内不会变。所以我给它的边界划得很清楚分层归谁管Agent-Reach 的态度模型推理模型服务商完全不碰对话编排、多轮状态编排框架不碰通过标准接口对接能力注册与路由Agent-Reach核心职责权限、限流、审计Agent-Reach核心职责具体业务逻辑业务系统只做适配不重写这个边界感带来一个直接好处你可以先接入再逐步替换。今天用 A 模型明天换 B 模型注册表里的能力定义一个字都不用改。我实测下来这套独立部署的触达层配合不同的模型后端切换业务侧的改动量基本是零。2.2 四层结构注册、路由、执行、回执内部我把它拆成四层每层职责单一出错时能快速定位是哪一层的问题。第一层是注册层。所有能力在这里被描述成结构化元数据包括名称、自然语言描述、入参 schema、出参 schema、超时、幂等策略、所需权限。这一层的关键是描述质量模型的调用准确率八成取决于这里的文字写得好不好后面会细讲。第二层是路由层。用户一句话进来路由层要做的事是从注册表里挑出最可能相关的 Top-K 个能力交给模型做最终决策。注意是挑出来再交给模型选不是让模型看全部。这个顺序很关键原因在 4.4 节会算一笔账。第三层是执行层。真正发起调用处理超时、重试、熔断、并发控制、协议适配。MCP 服务、HTTP 接口、本地脚本、数据库查询都在这一层被抹平成同一种执行语义。第四层是回执层。把执行结果整理成模型能消化、人能看懂的形式。原始接口返回的 JSON 往往又长又乱直接塞给模型既浪费 token 又容易干扰判断这一层要做裁剪、摘要、错误归一化。提示这四层不要合并部署。我早期图省事把路由和执行放在一个进程里结果一次工具执行卡住整个路由都被阻塞所有会话一起超时。拆开之后执行层挂了路由层还能正常回一个服务暂时不可用的友好提示。2.3 协议适配MCP、OpenAPI、脚本工具怎么统一现实环境里能力来源五花八门。有标准的 MCP 服务有现成的 OpenAPI 文档有写着玩的 Python 脚本还有只能通过命令行调用的老系统。Agent-Reach 的做法是不追求原生统一而是做适配收口。适配器的选择有讲究MCP 协议优先接。它的工具描述、参数 schema、调用语义都是现成的适配成本最低。适合新写的内部服务。OpenAPI 文档次之。凡是能拿到规范的直接解析生成能力定义比手写省事得多。但要注意过滤一份大的 OpenAPI 动辄几百个接口全注册进去等于给路由层添堵。自定义适配器兜底。老系统、脚本、数据库这类写个小适配器把输入输出对齐到统一格式就行。命令行放最后。可执行、可测试但错误信息和退出码处理起来麻烦容易把脏数据带进上下文。我一般建议的注册顺序是先接三到五个高频、稳定、幂等的能力跑通全链路再逐步铺开。一上来注册两百个能力路由准确率和调试成本都会爆炸。3. 核心模块拆解与实操配置3.1 能力注册表字段设计决定上限注册表是整个系统的地基。我给每个能力定义的字段大致是这样{ name: query_order_status, display_name: 查询订单状态, description: 根据订单号查询当前订单的物流与支付状态。仅在用户明确提供了订单号时使用如果用户只给了手机号请先调用 search_orders_by_phone。, category: order, parameters: { type: object, properties: { order_id: { type: string, description: 订单号通常是 18 位数字以 20 开头 } }, required: [order_id] }, returns: { type: object, properties: { status: { type: string }, updated_at: { type: string } } }, timeout_ms: 5000, idempotent: true, side_effect: read, required_scopes: [order:read], rate_limit: { qps: 20, burst: 40 }, version: 1.3.0, owner: order-team }几个字段值得单独说说。description不是给人看的注释是给模型看的决策依据。写法上我总结了三要素做什么、什么时候用、什么时候别用。只写查询订单状态这五个字模型在用户说我上周买的东西到哪了的时候很可能不调用因为它不知道这句话跟订单有关。把触发条件和排除条件写进去调用准确率提升非常明显我做过一轮对比同一批测试用例加了排除说明之后误调用率从两成多降到个位数。side_effect用来区分读操作和写操作。读操作可以放心重试、可以并发、可以预取写操作必须严格串行、必须幂等、必须走确认流程。这个字段是后面权限和限流策略的基础。required_scopes是权限的最小单位。我不建议按角色授权角色会膨胀得很快按能力域 读写授权粒度更稳。version和owner是给运维留的后路。线上出问题的时候你得知道找谁、回滚到哪一版。3.2 意图路由向量召回加规则兜底路由层我踩过最大的坑是过度依赖向量检索。早期版本纯靠 embedding 相似度召回测试集上看着挺好一上真实流量就露馅用户说帮我看看昨天那笔语义上跟哪个能力都不像向量检索直接懵。现在的做法是三层叠加第一层是关键词与实体规则。识别出订单号、手机号、时间范围这类强特征实体直接映射到候选能力。规则命中率高、延迟低、可解释作为兜底最合适。第二层是向量召回。把用户 query 和所有能力的 description 一起做 embedding取相似度 Top-K。这一层解决的是用户换了种说法的情况。第三层是上下文继承。多轮对话里上一轮调用过的能力要加权。用户问完订单状态接着问那什么时候能到这句话本身信息量为零但结合上下文就很明确。三层结果的合并策略我用的是加权打分规则命中的权重最高向量相似度次之上下文继承给一个温和的加成。最后取 Top-8 到 Top-12 交给模型。注意K 值不要拍脑袋定。太小会漏掉正确能力太大会把模型的注意力稀释掉。我的经验是从 12 开始往下压每压一档跑一遍评估集看准确率拐点在哪。多数场景下 8 到 10 是甜点区。3.3 权限与审计Agent 不能想调就调这一块最容易被忽略但出事的时候最要命。Agent 自动执行和人工点击执行风险等级完全不是一个量级——人工点击有人的判断Agent 调用只有模型的判断而模型会幻觉。我的做法是三级权限模型第一级是能力级权限。某个会话或某个用户有没有资格触达这个能力域。这一级在路由前就过滤掉没权限的能力根本不出现在候选列表里避免模型看到但调不了的尴尬。第二级是参数级约束。这个更细比如查询订单只能查自己名下的不能查别人的。实现上是在执行前插入一层校验器把会话身份注入参数或者对参数做范围检查。第三级是动作确认。凡是side_effect为write的能力一律走二次确认或者限定在特定沙箱环境里执行。我倾向于把写操作分成可自动和需确认两档退款、删除、发送这类不可逆动作必须确认。审计日志我记录了这几个字段缺一不可会话 ID、用户身份、路由候选列表及打分、模型最终选择、实际执行参数、执行耗时、返回码、结果摘要。中间三项特别重要——出了问题你要能回答模型当时为什么选了这个能力没有候选列表和打分你只能靠猜。3.4 限流、重试与幂等执行层的稳定性八成靠这三个机制。限流要分两个维度做。一个是全局维度保护下游系统一个是单会话维度防止某个 Agent 陷入循环疯狂调用。我见过最离谱的一次一个 Agent 因为返回结果格式不对自己重试了四十多次把下游接口打挂了。单会话维度加上同一能力连续调用超过 N 次就熔断这类问题就基本绝迹。重试的规则要按错误类型区分不能一刀切错误类型是否重试策略连接超时是立即重试 1 次间隔 200ms服务端 5xx是退避重试最多 2 次参数校验失败否直接把错误信息回给模型让它改参数权限不足否直接返回不重试业务逻辑错误否返回错误码和提示由模型决定下一步幂等只对写操作有实际意义。实现方式有两种一是下游系统自带幂等键二是我们在执行层生成一个请求指纹短时间内相同指纹的请求直接返回上次结果。第二种更通用但要注意指纹的生成要把参数归一化别因为字段顺序不同就判定成两次请求。4. 从零搭一套最小可用链路4.1 环境与依赖准备最小可用版本不需要太多东西。我用 Python 演示核心依赖就三个一个 Web 框架、一个向量库、一个 HTTP 客户端。python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn httpx pydantic pip install numpy faiss-cpu pip install sentence-transformers选faiss-cpu而不是上向量数据库是因为能力数量通常在几百到几千这个量级本地索引完全够用启动快、无外部依赖调试的时候少一个故障点。等能力数上万了再考虑换。配置文件我习惯用 YAML注册的能力写在独立目录里一个能力一个文件方便做代码评审# config/settings.yaml router: top_k: 10 rule_weight: 1.0 vector_weight: 0.7 context_weight: 0.3 executor: default_timeout_ms: 5000 max_retries: 2 session_rate_limit: 30 # 每会话每分钟最多 30 次调用 circuit_breaker: consecutive_failures: 5 cooldown_seconds: 60 audit: log_candidates: true log_raw_result: falselog_raw_result我默认关掉。原始返回里经常带手机号、地址这类信息全量落盘风险不小只记摘要和结果长度就够了。4.2 注册第一个工具先写一个最简单的能力适配器把查询订单状态接进来。# capabilities/order_status.py import httpx from pydantic import BaseModel, Field class OrderStatusParams(BaseModel): order_id: str Field(..., description18 位订单号以 20 开头) CAPABILITY { name: query_order_status, display_name: 查询订单状态, description: ( 根据订单号查询订单的支付与物流状态。 当用户提供了明确订单号或上下文中已经确认过订单号时使用。 如果用户只提供了手机号或昵称而没有订单号不要使用本能力 应当先调用 search_orders_by_identity。 ), parameters: OrderStatusParams.model_json_schema(), timeout_ms: 5000, idempotent: True, side_effect: read, required_scopes: [order:read], } async def execute(params: dict, ctx: dict) - dict: order_id params[order_id] if not (len(order_id) 18 and order_id.startswith(20)): return {ok: False, error: INVALID_ORDER_ID, message: 订单号格式不对请确认后再试} async with httpx.AsyncClient(timeout5) as client: resp await client.get( fhttps://internal.example.com/orders/{order_id}, headers{X-Trace-Id: ctx[trace_id]}, ) if resp.status_code 404: return {ok: False, error: NOT_FOUND, message: 没有找到该订单} resp.raise_for_status() data resp.json() return { ok: True, data: { status: data[status], status_text: data[status_text], updated_at: data[updated_at], }, }这里有个细节值得展开参数校验放两道。第一道在 schema 层让模型知道格式要求第二道在 execute 里做真实校验。为什么重复做因为模型不一定听话schema 写了它也可能传个 10 位的数字进来。第二道校验失败时返回的错误信息要写清楚应该是什么样模型看到之后下一轮往往能自己改对。ctx里我固定塞了trace_id、user_id、session_id、scopes。这个上下文贯穿全链路排查问题时靠它串起所有日志。4.3 接上模型跑通一次调用注册完能力接下来是把它暴露给模型。这里我用最朴素的函数调用格式演示换成任何模型服务商的格式思路都一样。# core/router.py import numpy as np from sentence_transformers import SentenceTransformer class CapabilityRouter: def __init__(self, registry, cfg): self.registry registry self.cfg cfg self.model SentenceTransformer(paraphrase-multilingual-MiniLM-L12-v2) self.names list(registry.keys()) texts [registry[n][description] for n in self.names] self.embeddings self.model.encode(texts, normalize_embeddingsTrue) def route(self, query: str, ctx: dict, top_k: int None) - list: top_k top_k or self.cfg[router][top_k] scores {} # 第一层规则命中 for name, cap in self.registry.items(): for kw in cap.get(keywords, []): if kw in query: scores[name] scores.get(name, 0) self.cfg[router][rule_weight] # 第二层向量召回 qvec self.model.encode([query], normalize_embeddingsTrue)[0] sims self.embeddings qvec for name, sim in zip(self.names, sims): scores[name] scores.get(name, 0) float(sim) * self.cfg[router][vector_weight] # 第三层上下文继承 last ctx.get(last_capability) if last in self.registry: scores[last] scores.get(last, 0) self.cfg[router][context_weight] # 权限过滤必须在排序之前 allowed ctx.get(scopes, set()) ranked [ n for n, _ in sorted(scores.items(), keylambda x: -x[1]) if set(self.registry[n][required_scopes]) allowed ] return ranked[:top_k]顺序上有个容易写错的地方权限过滤要在截断之前做。如果你的逻辑是先取 Top-10 再过滤权限很可能过滤完只剩两个候选路由等于白做。这个小细节我在评审别人代码的时候见过三次。拿到候选列表之后把对应能力的 schema 组装成模型能识别的工具定义连同对话历史一起发过去。模型返回工具调用请求后走执行层再把结果拼回对话继续下一轮。# core/executor.py import asyncio, time, hashlib class Executor: def __init__(self, registry, cfg, audit): self.registry registry self.cfg cfg self.audit audit self.session_calls {} async def run(self, name, params, ctx): cap self.registry.get(name) if cap is None: return {ok: False, error: UNKNOWN_CAPABILITY} if not cap[can_execute](ctx): return {ok: False, error: FORBIDDEN} sid ctx[session_id] now time.time() calls [t for t in self.session_calls.get(sid, []) if now - t 60] if len(calls) self.cfg[executor][session_rate_limit]: return {ok: False, error: SESSION_RATE_LIMITED, message: 本会话调用过于频繁请稍后再试} calls.append(now) self.session_calls[sid] calls timeout cap.get(timeout_ms, self.cfg[executor][default_timeout_ms]) / 1000 start time.time() for attempt in range(self.cfg[executor][max_retries] 1): try: result await asyncio.wait_for(cap[execute](params, ctx), timeouttimeout) self.audit.write(name, params, result, time.time() - start, ctx) return result except asyncio.TimeoutError: if attempt self.cfg[executor][max_retries]: return {ok: False, error: TIMEOUT} await asyncio.sleep(0.2 * (attempt 1))这段代码少了熔断逻辑实际用的时候要在attempt循环外面加一层失败计数器连续失败到阈值就把能力标记成不可用冷却期过了自动恢复。4.4 参数预算与调优一笔账算清楚这一节是我最想讲的因为很多团队在这里凭感觉调参调不明白。先说工具描述占用的上下文预算。一个写得比较完整的能力描述包含 name、description、参数 schema大概在 100 到 200 token 之间。取中间值 150。如果你把所有能力全部塞进每一次请求能力数量单次请求工具描述 token20 轮对话累计203000600005075001500001001500030000020030000600000注意这是累积的。多轮对话每一轮都要带上完整的工具列表到了第 20 轮光工具描述就吃掉几十万 token。这还没算对话历史本身。所以全量暴露这条路在能力数超过二三十个之后就走不通了。再说动态裁剪的收益。用 Top-10 的候选代替全量# 上下文预算估算 def estimate_tool_tokens(candidates): per_tool 150 return len(candidates) * per_tool full estimate_tool_tokens(registry) # 全量 trimmed estimate_tool_tokens(candidates) # Top-10 print(f裁剪前: {full}, 裁剪后: {trimmed}, 节省: {1 - trimmed/full:.0%})200 个能力的场景下从 30000 降到 1500节省 95%。这个量级的差别不是优化是能不能跑起来的问题。超时预算也得算。用户能忍受的等待上限我一般按 30 秒设计。拆开看环节预算说明路由召回300ms向量检索本地索引很快模型首 token1200ms取决于模型服务工具执行5000ms单次上限重试预留5000ms最多一次结果整理 二次模型3000ms让模型把结果说成人话缓冲剩余应对抖动加起来 14.5 秒留出足够缓冲。如果你把工具超时设成 30 秒一次超时就把整个预算吃光用户等半分钟什么都拿不到。所以工具超时必须小于端到端预算的三分之一这是我的经验值。并发度也得算。假设单实例 QPS 目标是 50每次调用平均耗时 2 秒那理论上需要的并发数是 100。但实际不可能每个请求都占满按峰值系数 1.5 算连接池开到 150 比较合适。开太小会排队开太大下游扛不住。这个数要压测别抄。提示这一节的数字都是经验值不是标准答案。真正靠谱的做法是搭个评估集几十条真实 query 加期望调用每次调参跑一遍。评估集不用大50 条就够看出趋势但一定要覆盖模糊表达多轮指代无匹配能力这三类难例。5. 常见问题与排查技巧实录5.1 常见问题速查表现象可能原因排查方向解决方式模型该调不调description 太笼统看候选列表里有没有这个能力补触发条件说明模型乱调相似能力没做区分看候选能力描述是否雷同在描述里写明互斥场景参数格式总错schema 描述不具体看模型传了什么加示例值和格式说明调用超时下游慢或超时设太长看执行耗时分布缩短超时加缓存结果答非所问原始返回太长太乱看回执层处理逻辑裁剪字段做摘要重复调用同一能力模型没识别到结果已足够看对话历史里有几次调用加连续调用熔断权限报错但看不出原因缺少 scopes 上下文看 ctx 里的权限集合补全身份注入高并发下大量失败连接池或限流不够看限流触发计数调整池大小和阈值5.2 几个踩过的坑坑一把接口文档直接当能力描述用。早期图省事把 OpenAPI 里的 summary 直接拿来当 description结果模型完全抓不住重点。接口文档是写给开发看的讲的是这个接口干什么能力描述是写给模型看的要讲什么情况下该用它。这两件事不是一回事。后来我加了一道人工改写流程每个上线的能力描述必须经过一次评审准确率才稳下来。坑二忽略了多轮对话里的指代。用户说帮我查一下订单Agent 问请问订单号是多少用户回就是刚才那个。这个刚才那个在纯向量检索里完全找不到对应必须靠上下文继承。我在路由层加了最近提及实体的槽位记忆把用户在前几轮里说过的订单号、手机号存起来路由时优先匹配。这个改动让多轮场景的成功率提升了一大截。坑三错误信息写得太技术。后端返回{code: 40001, msg: invalid param}模型看到完全不知道该怎么改。回执层必须做错误归一化把技术错误翻译成模型能理解的行动指引比如参数 order_id 需要 18 位你提供的是 10 位请向用户确认完整订单号。改完之后模型自己修正参数的成功率明显上升。坑四测好了才上线上线就翻车。测试环境的能力都是通的生产环境有各种奇怪情况某个接口慢、某个返回空、某个鉴权过期。建议在执行层加一个健康探测定期轻量调用每个能力把不可用的提前摘出来别等模型调用了才发现。坑五日志记了但没用起来。一开始我只记成功失败出问题完全没法复盘。后来把候选列表、打分、模型选择、实际参数全记下来才具备了优化能力。这个投入回报比很高强烈建议从第一天就做好。5.3 上线前的检查清单每个能力的 description 都写了什么时候用和什么时候别用。所有写操作能力都有二次确认或沙箱限制。权限过滤逻辑在候选截断之前执行。每个能力都设了独立超时且小于端到端预算的三分之一。单会话调用频次有上限同一能力连续调用有熔断。错误返回都做了归一化带可执行的修正提示。审计日志包含候选列表和打分保留时间符合合规要求。有健康探测异常能力能自动下线。有一个 50 条以上的评估集能一键跑出准确率。回执层对原始返回做了字段裁剪不把敏感信息带进上下文。最后分享一个小技巧。能力描述我习惯留一份反面样本文档记录每次模型调错的案例哪句话、哪个能力、错在哪。攒到二三十条之后你会发现错误有明显的聚类往往是某几个能力的描述边界没划清。针对性地改描述比盲目调路由权重有效得多。这套东西没有一劳永逸的配置能力在增长模型在更新用户的表达方式也在变能持续纠偏的机制比一开始就调对参数更重要。