上个月帮一个团队复盘 Agent 落地聊到一半技术负责人拍了下桌子模型选型换了三轮提示词改了上百版最后卡住的居然是“调不到接口”这四个字。模型能推理出该干什么却拿不到订单系统的数据、改不了工单状态、发不出通知——这就是典型的“脑子够用手脚太短”。Agent-Reach 这个项目名我第一次看到时的直觉就是冲着这个问题去的Reach触达。它想解决的不是“智能体够不够聪明”而是“智能体能不能稳定地、可控地、可观测地碰到真实世界里的那些系统和数据”。它本质上是给智能体做一层统一的能力触达层把散落在各处的 HTTP 接口、数据库、内部服务、脚本任务封装成智能体能理解、能调用、能被审计的能力单元。这篇文章适合三类人正在做 Agent 落地被工具调用折磨的工程同学、准备自建能力网关的架构同学、以及想搞清楚“智能体到底怎么接业务系统”的产品同学。下面我把这套东西的设计思路、核心代码、参数算法和踩过的坑完整拆一遍。1. 先搞清楚 Agent-Reach 在整条链路里站在什么位置1.1 一个真实场景模型不笨是手太短我们拿一个售后工单场景举例。用户说“我上周买的那台机器噪音很大想退货”。一个成熟智能体需要做的事包括查订单订单服务、查物流状态物流服务、查售后政策知识库、判断是否在退货窗口内规则引擎、创建退货单工单服务、发短信通知消息服务。这六步里只有“查售后政策”是纯检索剩下五步全是动作。动作意味着三件事有副作用、有失败可能、有权限边界。模型的推理能力再强只要这五步里的任何一步失败率超过 5%整个链路就基本不可用——因为六步串联0.95 的六次方只有 0.73。这就是很多 Demo 惊艳、上线翻车的根本原因。Agent-Reach 想干的活就是把这五步从“模型自己想办法”变成“调用一个已经准备好的能力”。它站在模型和后端系统之间对上提供一份清晰的能力目录对下管理所有连接细节。模型看到的是一张简单的菜单而真正复杂的鉴权、重试、字段映射、限流、审计全被挡在菜单后面。这里有个认知要摆正Agent-Reach 不是让模型变聪明而是让模型的动作变确定。这个定位决定了后面所有的设计取舍——凡是影响“确定性”的花活一律砍掉。1.2 自研工具层为什么最后都会长成一坨泥我见过至少五六个团队自己撸工具调用层路径几乎一模一样。第一周写个字典把函数名和参数描述塞进提示词能跑。第二周加了三个接口发现参数格式不统一开始写 if-else 做适配。第三周下游接口偶发超时加个重试。第四周有人不小心让模型重复提交了两次订单加个开关。第五周运维问“昨天那个工单是谁改的”没人答得上来临时加日志。三个月后这个文件有 2000 行没人敢动。问题出在哪出在一开始就把“能力描述”“连接适配”“调用编排”三件事混在一个文件里了。这三件事的变化频率完全不同能力描述跟着业务走可能一周改一次连接适配跟着下游接口走可能一季度改一次调用编排重试、限流、降级是平台级策略最好半年不动。变化频率不同的东西绑在一起就会互相拖累。Agent-Reach 的分层就是冲这个来的。它把能力注册、适配执行、路由编排、观测审计拆成四层每层有独立的配置文件、独立的发布节奏、独立的负责人。这样改业务的时候不用动平台代码改平台策略的时候不用回头找业务方确认。1.3 四层骨架与关键取舍具体拆开看能力注册层维护一份能力清单每个能力有唯一 ID、自然语言描述、入参出参 schema、权限标签、超时等级。这一层是模型唯一能看到的东西。适配器层每个能力背后挂一个适配器负责把统一签名翻译成具体协议。HTTP 也好、SQL 也好、消息队列也好都在这里消化掉。路由与调度层决定这次调用走哪个实例、要不要重试、什么时候熔断、并发怎么控。观测层trace_id 贯穿全链路记录入参、出参摘要、耗时、结果码出问题能回放。两个取舍值得单独说。第一能力 ID 用稳定的字符串而不是函数名因为函数名会随重构变化而能力 ID 一旦被模型学会就不该轻易变。第二出参一律走“裁剪 摘要”而不是原样透传因为下游返回 200KB 的 JSON 塞进上下文既贵又会让模型抓不住重点。这两条后面有专门的参数章节。2. 能力注册把“能做什么”写成机器和人都不迷糊的契约2.1 能力描述文件长什么样我把注册层的数据结构叫“能力契约”因为它同时约束三方模型怎么调、适配器怎么实现、审计怎么看。下面是我实际在用的一个精简版结构id: order.query version: 1.2.0 title: 根据手机号或订单号查询订单 description: | 查询用户的历史订单。至少提供 phone 或 order_no 其中之一。 返回最近 10 条订单的编号、金额、状态和下单时间。 category: order params: type: object properties: phone: type: string description: 用户手机号11位数字 pattern: ^1[0-9]{10}$ order_no: type: string description: 订单编号通常以 OD 开头 maxLength: 32 limit: type: integer minimum: 1 maximum: 10 default: 5 oneOfRequired: [phone, order_no] returns: type: array maxItems: 10 fields: [order_no, amount, status, created_at] auth: scope: order.read level: read timeout_ms: 1200 cache_ttl_s: 30 idempotent: true几个字段值得展开。oneOfRequired是我自定义的约束表达“这两个参数至少给一个”比在 description 里用自然语言写要可靠得多——模型对自然语言的“至少”经常理解偏。level标成 read写操作标 write路由层会据此决定要不要走二次确认。timeout_ms单独给而不是全局一个值因为查缓存的接口和调外部物流接口的合理超时差一个数量级。idempotent这个标记直接决定重试策略能不能开写错它是会出生产事故的。returns.fields这一项是我强烈建议加的。它让适配器知道该裁掉哪些字段而不是等模型自己在一大坨 JSON 里找。实测下来加了这个字段之后同一个任务的 token 消耗能降 30% 到 50%模型调用准确率也明显上升。2.2 描述信息的 token 预算怎么控能力目录是要塞进系统提示词或者工具列表的所以它直接吃 token。能力一多光目录就能占掉几万 token模型还没开始干活上下文就满了。我的经验值是单个能力的描述控制在 80 到 150 个 token 之间整个目录不超过上下文窗口的 15%。怎么压缩三条做法。第一条description只写“干什么”和“什么时候用”不写“怎么实现”。别写“本接口底层调用 XX 服务的 YY 方法”模型不关心。要写“当用户询问订单状态、退款进度时使用本能力”这才是模型选工具时真正需要的触发条件。第二条参数描述去重。很多团队把枚举值全列在描述里其实枚举放在 schema 的enum字段里更省而且校验更严。只有那些模型容易搞混的值才需要在描述里举例。第三条做目录分级。高频能力放主目录低频能力收进“扩展能力”里模型需要时再通过一个list_more_capabilities能力去查。这个做法有点像菜单和隐藏菜单的关系实测能砍掉一半以上的常驻 token。2.3 版本与灰度能力不能随便改签名能力 ID 一旦被模型学会改签名就是破坏性变更。我的做法是版本号跟能力 ID 分离ID 永久稳定版本独立演进。参数只允许做“加可选字段”这种向后兼容的扩展如果要删字段或者改语义就新开一个 ID比如order.query.v2两个版本并行跑一段时间看日志确认老版本没人调了再下线。灰度这块我在能力契约里加了一个routing段支持按流量比例、按调用方、按用户白名单三种分流方式。比如新适配器上线先切 5% 流量观察错误率和 P99 耗时两条曲线都平稳再逐步放大。注意能力下线前一定要跑一遍“静默期”把老版本适配器保留但打上 deprecated 标记任何一次调用都记一条 warn 日志。我见过直接删能力导致线上智能体整条链路瘫痪的情况恢复花了两个小时。3. 适配器与路由把异构接口抹平成统一签名3.1 适配器层的三条硬规矩适配器是脏活集中营规则必须死。我给自己定了三条规矩一适配器只做转换不做决策。它负责参数校验、字段映射、协议转换、结果裁剪但不判断“要不要重试”“要不要降级”那是路由层的事。判断逻辑一旦渗进适配器二十个适配器就会有二十种重试风格。规矩二所有适配器必须实现同一个抽象基类方法签名固定。Python 里大概是这样from abc import ABC, abstractmethod class BaseAdapter(ABC): capability_id: str abstractmethod async def validate(self, params: dict) - dict: 校验并补默认值抛 ValidationError abstractmethod async def execute(self, params: dict, ctx: CallContext) - dict: 真正执行抛出统一的 UpstreamError abstractmethod def trim(self, raw: dict) - dict: 按 returns.fields 裁剪结果CallContext里带着 trace_id、调用方身份、剩余超时预算。注意“剩余超时预算”这个词——超时是会被上游消耗的适配器拿到的不应该是配置里的完整值而应该是倒推之后剩下的额度。规矩三错误必须归一化。下游可能返回 HTTP 500、可能返回业务码 -1、可能抛连接异常但在适配器出口只能有三种ParamError参数问题不重试、UpstreamError下游问题可重试、FatalError配置错误熔断。错误分类错了重试策略就是瞎打。3.2 路由、重试、降级的策略组合路由层拿到一次调用请求后流程是固定的五步解析能力契约、检查权限、选实例、执行、按结果决定是否重试或降级。实例选择我用的是最简单也最稳的加权轮询加健康度惩罚每个实例有个基础权重连续失败一次扣 20% 健康分健康分低于 30 的实例暂时踢出池子30 秒后以半开状态放一次探测请求进去。这套逻辑和常见的熔断器是一个思路好处是无需额外依赖两百行代码搞定。重试必须满足三个条件才允许触发错误类型是可重试的、能力契约里idempotent: true、剩余超时预算还够一次尝试。第三个条件最容易漏很多团队配了“重试 3 次”单次超时 2 秒结果一个请求挂了 8 秒上游早就断开了。正确做法是重试前算一下剩余预算 单次预期耗时 × 1.5 才重试。退避策略我用的是指数退避加抖动import random def backoff_ms(attempt: int, base: int 100, cap: int 2000) - int: raw min(base * (2 ** attempt), cap) jitter random.uniform(0, raw * 0.3) return int(raw jitter)抖动这一项千万别省。不加抖动的话下游一抖动所有重试会在同一毫秒涌过去把小抖动放大成雪崩。这个坑我在压测里踩过QPS 200 的情况下加抖动前后下游错误率差了四倍。降级策略分两类。读操作可以降级到缓存或者返回“暂时查不到”写操作不能随便降级——如果退货单没创建成功却告诉用户成功了那是数据事故。所以我在契约里对 write 级能力强制要求配一个on_failure行为只能是fail_fast或者queue_for_retry不允许fallback_success。3.3 写操作的幂等与确认机制写操作是整套系统里最需要小心的地方。我做了两层保护。第一层是幂等键。路由层在发起写调用前根据“能力 ID 关键参数 会话 ID”算一个idempotency_key透传给适配器适配器再透传给下游。下游如果支持幂等键就最好了不支持的话我在网关侧维护一个短期的键记录表同一个键在 10 分钟内重复到达直接返回上次结果不再真正调用。第二层是确认机制。write级能力在模型侧看到的描述里会多一句“此操作会修改数据调用前需向用户确认”。同时路由层支持一个require_confirm标记开启后第一次调用只返回一个预演结果展示将要执行的动作和影响范围必须带上confirm_token再调一次才真正执行。这个设计对“删除”“退款”“批量修改”这类高危动作特别必要。提示幂等键的生成里一定要带会话 ID不能只用参数。因为不同用户在同一时刻可能提交完全相同的参数只用参数会导致第二个用户的请求被误判为重复而直接返回第一个用户的结果。4. 从零搭一个最小可用版本4.1 目录结构与依赖先说环境Python 3.11 以上依赖就四个fastapi、httpx、pydantic、pyyaml。不引入重型框架因为这一层越轻越好出问题好定位。目录我是这样组织的agent_reach/ capabilities/ # 能力契约 YAML一个能力一个文件 order.query.yaml ticket.create.yaml adapters/ # 适配器实现 order_query.py ticket_create.py core/ registry.py # 加载契约、建索引 router.py # 路由、重试、熔断 context.py # CallContext、trace_id errors.py # 统一错误类型 server.py # 对模型暴露的入口 config.yaml # 全局策略按能力拆文件而不是按服务拆是为了让“一个能力的全部信息”在同一个地方改的时候不用来回跳。4.2 写第一个能力契约以ticket.create为例几个关键点id: ticket.create version: 1.0.0 title: 创建一个售后工单 description: | 当用户明确要求退货、换货或维修且已完成身份与订单确认时使用。 创建成功后返回工单号用于后续查询进度。 category: ticket params: type: object properties: order_no: {type: string, description: 关联订单号} type: {type: string, enum: [return, exchange, repair]} reason: {type: string, maxLength: 200} contact_phone: {type: string, pattern: ^1[0-9]{10}$} required: [order_no, type, reason] returns: type: object fields: [ticket_no, status, estimated_hours] auth: scope: ticket.write level: write timeout_ms: 3000 retry: {max_attempts: 2, retryable_errors: [UpstreamError]} idempotent: true require_confirm: truedescription里我特意写了触发条件“已完成身份与订单确认”这是给模型看的前置条件提示。实测加不加这句模型在信息不全时就贸然创建工单的比例能差出三倍。4.3 实现适配器import httpx from core.errors import ParamError, UpstreamError from adapters.base import BaseAdapter class TicketCreateAdapter(BaseAdapter): capability_id ticket.create def __init__(self, base_url: str, token: str): self.client httpx.AsyncClient(base_urlbase_url, timeout3.0) self.token token async def validate(self, params: dict) - dict: if params.get(type) not in (return, exchange, repair): raise ParamError(type 必须是 return/exchange/repair 之一) params.setdefault(reason, 用户未填写) return params async def execute(self, params: dict, ctx) - dict: headers { Authorization: fBearer {self.token}, X-Trace-Id: ctx.trace_id, Idempotency-Key: ctx.idempotency_key, } # 关键超时取剩余预算和配置值的较小者 budget min(ctx.remaining_ms / 1000.0, 3.0) try: resp await self.client.post( /api/ticket, jsonparams, headersheaders, timeoutbudget ) except httpx.TimeoutException as e: raise UpstreamError(fticket service timeout: {e}) from e if resp.status_code 500: raise UpstreamError(fupstream {resp.status_code}) if resp.status_code 400: raise ParamError(resp.text[:200]) return resp.json() def trim(self, raw: dict) - dict: return { ticket_no: raw.get(data, {}).get(no), status: raw.get(data, {}).get(status), estimated_hours: raw.get(data, {}).get(eta), }三处值得单独说。X-Trace-Id一定要透传下去否则下游日志跟你这边的日志对不上排查问题等于盲人摸象。Idempotency-Key透传是为了让支持幂等的下游帮着兜一层。trim里做了字段重命名把下游的no映射成语义更清楚的ticket_no这样模型看到的字段名和它在推理时用的名字是一致的减少歧义。4.4 注册、联调与接入模型注册就是把 YAML 加载成契约对象建两张索引id - 契约和id - 适配器。启动时做一次完整性检查每个契约必须有对应适配器适配器的能力 ID 必须在契约里存在任何不匹配直接启动失败。这个检查看起来啰嗦但它能在启动阶段拦掉 90% 的低级错误比运行到线上才发现强。联调我习惯先写一个命令行入口不接模型直接手动传参调用python -m agent_reach.cli call ticket.create \ --params {order_no:OD20240512001,type:return,reason:噪音过大} \ --dry-run--dry-run会走完整链路但把最终写操作替换成打日志用来验证参数映射和权限。确认无误再去掉这个参数跑真实调用。接入模型有两种方式。一种是让 Agent-Reach 作为 MCP 服务端把能力目录暴露出去模型通过标准协议发现和调用。另一种是把能力目录渲染成 JSON Schema 的工具列表直接塞进模型的原生工具调用参数里。前者的好处是能力更新不用重启模型侧后者延迟更低、链路更短。我一般先上后者跑通等能力数量超过三十个再切前者。4.5 压测与观测埋点上线前必须压一轮。我的压测做法是把线上最近一周的真实调用参数脱敏后回放按 1 倍、3 倍、5 倍 QPS 分三档压。观测埋点四个必打调用开始、参数校验结果、下游调用结果、最终返回。每条日志都带 trace_id、capability_id、耗时、结果码。我特别建议加一个“参数校验失败率”的指标这个指标异常升高通常说明模型对某个能力的参数理解出了问题可能得回去改描述。log.info(cap.call, extra{ trace_id: ctx.trace_id, cap: cap.id, ver: cap.version, latency_ms: cost_ms, result: ok if ok else err_code, attempt: attempt, })attempt字段别省它能直接告诉你重试率。重试率超过 3% 就该去看看下游是不是有问题了。5. 参数怎么定超时、并发、截断的算术题5.1 超时预算从外往里倒推超时不能拍脑袋。假设整个用户请求的体验目标是 8 秒内必须有响应一次对话里模型可能串行调 4 个能力模型自身的推理和生成大约要占 2 秒那么留给四次调用总共是 6 秒平均每次 1.5 秒。但这只是平均值还要考虑分布。我给每个能力的超时按“P99 耗时的 1.5 倍”来设且所有能力的超时之和不能超过总预算。如果超了就要做优先级分配核心能力给足边缘能力压缩。能力下游 P99设定超时占比order.query320ms600ms10%policy.search800ms1200ms20%ticket.create1500ms2500ms42%notify.send200ms500ms8%预留缓冲-1200ms20%关键是那个“预留缓冲”。真机上总有意外把预算用满的方案一定会在高峰期出问题。还有一点剩余预算要在整个链路里传递。路由层算出剩余额度传给适配器适配器再设成实际请求超时。这样即使前面某一步慢了点后面的步骤会自动压缩时间总量不会失控。5.2 并发与限流别把下游打穿Agent-Reach 自己也要被限流而且要比下游的限流更保守。原则是网关侧限流阈值 下游安全阈值 × 0.8。做法是按能力分组做信号量隔离而不是全局一个池子。原因很直接慢能力比如创建工单P99 1.5 秒如果和快能力共享并发池慢能力一堆积就会把快能力的位置占满出现“查个订单也超时”的诡异现象。分组之后每类能力有自己的并发额度互不干扰。我用的初始值是读操作单实例并发 50写操作单实例并发 10。写操作给得少是因为写操作的失败重试成本高宁可排队也别打爆。队列策略上超过并发就走排队但排队要设长度上限和等待上限。队列长度上限我一般设并发的 2 倍等待上限设成 200ms。超过就直接返回“系统繁忙”让模型有机会换个说法或者告诉用户稍后再试。比一直挂着强。5.3 截断策略返回值别原样吐给模型下游返回 100KB直接塞进上下文后果是三重的token 成本飙升、模型注意力被稀释导致推理变差、长响应还可能触发上下文截断导致前面的重要信息丢失。我的截断规则是三层。第一层是字段白名单筛选这个在契约的returns.fields里已经定义好了。第二层是长度截断单字段超过 500 字符的保留前 300 加后 100中间用省略标记。第三层是条数限制列表类结果默认最多 10 条超过的返回总数和提示“还有 N 条”。def smart_truncate(text: str, head: int 300, tail: int 100) - str: if len(text) head tail: return text return f{text[:head]}...(共{len(text)}字已省略)...{text[-tail:]}截断时要带上“已省略”的提示这点很重要。否则模型会以为拿到的就是全部信息可能基于不完整数据做判断。带上这个标记模型会自然地选择是否需要追问或者分页获取。6. 常见问题排查与避坑清单6.1 问题速查表现象大概率原因排查动作模型总选错能力描述里没写触发条件多个能力描述重叠检查能力目录把高频混淆的两条描述放一起对比参数校验总是失败schema 太严或模型不知道格式看失败日志里的原始参数必要时放宽 pattern 或加示例偶发超时且集中在某个能力下游慢或者超时预算分配不合理拉下游 P99 曲线核对超时设定同一写操作重复执行幂等键没带或下游不支持查 idempotency_key 是否透传检查网关侧去重表上下文被塞满返回值没截断检查 returns.fields 配置和截断阈值高峰期大量“系统繁忙”并发池太小或分组不合理看各分组并发使用率确认是否被慢能力拖累日志里 trace 断链trace_id 没透传到下游检查适配器 header 是否带 X-Trace-Id老能力突然没人调模型改用了新版本正常确认静默期后可以下线6.2 我踩过的五个坑坑一把重试配在适配器里。结果某个适配器重试 3 次另一个不重试行为不一致压测数据完全没法比。后来全部收到路由层适配器只负责抛正确类型的错误。坑二用函数名当能力 ID。一次重构把query_order改名成fetch_order模型的提示词和缓存里的历史都失效了准确率掉了十几个点。能力 ID 是给模型看的对外接口不该跟代码结构绑定。坑三没区分读写就统一重试。一个创建工单的请求因为超时被重试了结果下游其实成功了生成了两个工单。加了idempotent标记和幂等键之后才彻底解决。坑四日志只记成功不记失败。上线第一天觉得挺稳一周后用户投诉变多回头查日志发现参数校验失败率高得吓人但因为没有记失败详情完全不知道是哪个参数出的问题。失败日志必须记原始入参前提是脱敏。坑五一次性把所有能力都放出去。能力从 5 个涨到 40 个之后模型的工具选择准确率断崖式下跌。后来做了分级目录主目录只留 12 个高频能力准确率才回来。6.3 权限边界与审计权限这块有两个原则不能破。第一是最小权限每个能力只申请它必须的 scope能读的绝不给写。第二是调用方身份要透传不能所有能力共用一个超级账号否则出了事根本查不到是谁干的。审计日志我保留四类信息谁调的调用方身份、调了什么能力 ID 和版本、传了什么脱敏后的参数、结果如何成功失败、耗时、错误码。保留期按业务合规要求定一般 90 天起步。对写操作审计日志还要额外记录执行前后的关键状态方便溯源。注意脱敏规则要单独维护一份清单手机号、身份证、银行卡、地址这些字段在写日志前必须替换。我在适配器出口加了一个统一的脱敏钩子而不是让每个适配器自己处理这样漏掉的可能性小很多。最后分享一个我在实际使用中的体会Agent-Reach 这类能力层最容易犯的错是把它当成一次性工程搭完就不管了。但它的价值恰恰在长期维护里——能力描述的措辞打磨、超时参数的持续校准、失败日志的定期分析这些看着琐碎的事情才是让智能体从“能演示”走到“能用”的分水岭。我现在的习惯是每周花半小时看一遍参数校验失败率排行榜和能力调用重试率排行榜两个榜单的前三名基本就是接下来一周要优化的全部内容。