☰
Agent-Reach:打造高可用工具触达层,让智能体真正‘够得着‘外部系统
2026/10/26 7:05:03 网站建设 项目流程

很多人做 Agent 开发,最开始关注的都是模型选型、Prompt 怎么写、记忆怎么管,但真正把 Agent 推到生产环境之后,你发现最卡脖子的其实是“触达能力”——也就是 Agent 到底能碰到多少外部系统、能调多少工具、能多稳地拿到结果。我最近把一个内部项目整理成了开源形态,代号就叫 Agent-Reach,核心就是解决这个“触达半径”的问题。这篇就结合我这个项目的完整落地过程,把思路、实现、参数选择、排坑全过程都摊开聊一遍,给正在做 Agent 工具接入、函数调用、MCP 集成的人一个可以直接抄作业的参考。

1. 内容整体设计与思路拆解

1.1 为什么叫“Reach”:先想清楚 Agent 的能力边界在哪

做 Agent 之前,我习惯先画一张图:左边是模型本身的推理能力,右边是 Agent 能触达的真实世界。很多项目死在半路,不是因为模型不够聪明,而是因为右边那半张图太小——工具只有一两个,API 没接进来,数据库摸不到,最后模型再聪明也只能“空想”。

Reach 这个词掏出来,就是在强调这个被忽略的边界。一个 Agent 的 Reach 半径,直接决定了它能完成多复杂的任务链。比如你要做一个“查天气然后帮用户订餐厅”的 Agent,模型推理部分是弱的,真正的工作量全在“能不能查到天气”“能不能调用订餐 API”“失败了怎么办”这三件事上。Agent-Reach 这个项目,本质就是把这三件事变成一套可复用、可观测、可兜底的工程框架。

我设计它的目标不是再造一个 Agent 框架,而是做一个“任何 Agent 框架都能接的触达层”。不管你的调度器是 LangChain、LlamaIndex、Semantic Kernel 还是自己写的一套流程编排,Agent-Reach 只负责一件事:把模型意图变成可靠的外部调用。

1.2 核心设计思路:把工具触达拆成三层

整个项目我拆了三层,每一层只干一件事,避免揉成一团后期没法演进。

第一层是“声明层”。所有要开放给 Agent 的工具,统一用 JSON Schema 描述,不写逻辑,只写“这个工具是什么、参数有哪些、约束是什么”。模型看到的就是这一层,它根据这些声明决定调哪个工具、填什么参数。

第二层是“调度执行层”。负责接收模型发出的工具调用请求,做参数校验、超时控制、重试、并发限制。这一层是整个触达质量的胜负手,因为大模型生成的参数经常不符合接口要求,直接透传出去,一会儿类型错误、一会儿缺字段,全得在这一层拦住并修正。

第三层是“回传观测层”。每个工具调用的结果、耗时、token 消耗、错误原因,全部结构化记录。这一层不是为了给老板看报表,而是为下一轮 prompt 优化和工具描述优化提供数据支撑——我后面会详细说,很多工具“明明可用却总是没被模型选中”,问题就出在描述上,而这些信息只能从这里挖出来。

这三层对应到代码里就是registry、executor、tracer三个核心模块。设计的时候我没有让它们互相依赖,registry 只负责注册和校验 Schema,executor 只负责执行和重试,tracer 只负责记录。这样任何一个模块都能单独替换。

1.3 为什么不用硬编码、不用裸 Function Calling

如果你只是做 Demo,直接在代码里写if tool == "get_weather": return xxx()就够了,但一旦工具数量超过十个、参数超过几十个,硬编码就是灾难。每次加一个工具都要改主流程代码,模型和工具之间的映射关系也完全不可见,出了错误根本没法排查。

裸 Function Calling 的问题更大。模型返回一个{"name": "get_weather", "arguments": {...}},你确实拿到了意图,但后面的活儿全得自己干:参数要自己校验、超时要自己管、错误要自己兜底、工具间依赖要自己编排。这些逻辑每个项目都写一遍,写出来还不一样,质量全靠个人功力。Agent-Reach 把这几段“脏活”收口,工具接入方只写一个函数加一段 Schema 就结束。

另外我特意兼容了 MCP(Model Context Protocol),因为现在不少现成工具都有 MCP Server 实现,直接连能省掉很多重复开发。Agent-Reach 里我写了一个 adapter,把 MCP 工具动态转换成内部的工具声明——这个点很多人问,我后面会在实操部分单独演示。

2. 核心细节解析与实操要点

2.1 工具 Schema 怎么设计才不会坑到模型

工具 Schema 是模型做工具选择的唯一依据,它的质量直接决定触达成功率。我见过很多团队拿后端接口的参数列表直接当 Schema 用,字段名是后端缩写风格,枚举值不写说明,每个字段还都是 required——这种 Schema 喂给模型,它不瞎猜才怪。

我沉淀了几个关键习惯。字段名优先用自然语言全名,比如用cityName而不是cn,用partySize而不是ps,模型对语义明确的名字理解准确率高非常多。每个参数都要写description,而且要写“这个字段应该填什么”,不是写“这个字段是什么”。举个例子,表示人数那个字段,如果你只写“人数”,模型可能填 4 也可能填 “4” 或 “four”;如果你写“参与用餐的人数(整数,最小值为1,特殊需求如婴儿座椅请在备注中说明)”,模型就很少出错。

参数的required我建议极度克制。经验数据是,required 越多,模型为了凑齐参数的“幻填率”越高——它真的会编一个不存在的城市代码填进去。能不 required 就不 required,宁可后端多做一次默认值处理,也比模型瞎填好。

枚举值一定要给description。当你限制cuisine: ["chinese", "japanese"]时,模型经常会问“韩餐算哪种”,但如果写成"chinese"(中餐,包含川菜、粤菜等),它就能正确归类。这个细节让我整体工具调用成功率提升了将近 20 个点。

2.2 工具描述就是给模型看的“首页”

很多人不重视description字段,写一行“获取天气”就完事了。我实测下来,工具描述对模型选型的影响比参数 Schema 还大。原因是模型在每一步都要决定“当前这个任务,要不要使用工具、用哪一个”——这个决策基本全靠描述判断。描述写得含糊,模型宁可自己编一个结果也不用你的工具,这就是“工具叫不动”的最大原因。

我写描述有一条固定模板:触发场景 + 数据来源 + 典型问题示例。比如 “当用户询问当前或未来几天的天气情况时使用,数据来源为中央气象台接口,支持国内城市。例如用户可以这样问:北京明天会下雨吗?” 就这么一段话,模型几乎不会漏选。

如果工具来自 MCP,很多时候原始描述是从服务端代码注释生成的,读起来像文档而不是决策依据。我在 adapter 里加了一个可选的descriptionTemplate替换机制,接入方可以覆盖原始描述。别偷懒,这一步很值得做。

2.3 执行层:超时、重试、并发一个都不能少

执行层是触达质量的关键。工具调用本质上是网络 IO,网络 IO 就有三个逃不掉的问题:慢、挂、并发冲。

超时我分成“连接超时”和“整体超时”两层。连接超时统一设 3 秒,整体超时看工具本身特性,普通 HTTP 查询类给 10 秒,流式接口或批量任务给到 60 秒。绝不给“无限超时”,因为模型等待工具返回的过程一直在消耗上下文窗口,挂一个工具,整条 Agent 链路就废了。

重试是必要的,但重试不是“失败就再试一次”,那样只会把网络抖动放大成雪崩。我采用的策略是:只有网络错误(超时、连接中断、5xx)才重试,业务错误(参数校验失败、4xx)不重试,因为业务错误重试多少次都是同一个结果。重试次数上限 2 次,每次间隔采用指数退避加随机抖动,比如第一次等 1 秒、第二次等 2~3 秒。并发限制也必须有,我把每个工具单独配了一个信号量,默认同时最多 4 个并发,防止某个慢工具把整个 Agent 的线程池占满。

2.4 最小权限与安全边界

给 Agent 开放工具,本质上就是给一段不可完全控的代码开权限。模型再聪明,也保不齐在复杂上下文里产生非预期调用,所以权限控制必须做成“默认拒绝,显式放行”。

我在 Agent-Reach 里维护了一张permission.json,记录每个工具的角色白名单、可执行操作、访问域名白名单。模型生成调用意图后,执行层第一步不是执行,而是校验权限。比如“查询订单”这个工具,对前台助手角色只开放状态查询,对运营角色才开放全量列表。域名级别也做了限制,工具内部如果发现目的域名不在白名单中,直接拒绝并记录高危事件,防止模型被注入后访问内网地址。这一块不是“功能需求”,是“生存需求”,一定要从第一天就带上。

3. 实操过程与核心环节实现

3.1 环境准备与依赖清单

整个项目我用的 Python 3.10,依赖尽量精简。核心就四个包:

pip install openai jsonref pydantic aiohttp

openai是模型调用客户端,jsonref用来解析带$ref的 Schema,pydantic做参数校验和数据类定义,aiohttp做异步 HTTP 工具调用。没有引入重型 Web 框架,因为 Agent-Reach 是库不是服务,被打包在其他服务里运行。

项目目录结构如下:

agent_reach/ ├── __init__.py ├── registry.py # 工具注册与 Schema 管理 ├── executor.py # 调度执行、超时重试、并发控制 ├── tracer.py # 调用链路记录 ├── adapters/ │ ├── __init__.py │ └── mcp_adapter.py # MCP 工具适配层 ├── schemas/ # 内置工具 Schema │ └── weather.json └── permission.json # 权限白名单

3.2 工具注册:一行代码接入一个能力

我先定义registry.py的核心逻辑。工具注册使用装饰器风格,接入方只需要关注“函数实现”和“Schema 声明”两件事。

# registry.py from typing import Callable, Dict, Any import jsonref class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict[str, Any]] = {} def register(self, schema: dict): """注册一个工具,传入其 JSON Schema 声明""" def decorator(func: Callable): resolved_schema = jsonref.replace_refs(schema) self._tools[resolved_schema["name"]] = { "schema": resolved_schema, "func": func, } return func return decorator def get_tool_models(self) -> list[dict]: """返回给模型用于 Function Calling 的工具声明列表""" return [t["schema"] for t in self._tools.values()] def get(self, name: str): return self._tools.get(name)

接入方注册一个工具只需要这样写:

# weather_tool.py from agent_reach.registry import ToolRegistry registry = ToolRegistry() weather_schema = { "type": "function", "function": { "name": "get_weather", "description": "当用户询问当前或未来几天的天气情况时使用,数据来源为中央气象台接口,支持国内城市。例如用户可以这样问:北京明天会下雨吗?", "parameters": { "type": "object", "properties": { "cityName": { "type": "string", "description": "城市中文名,例如:北京、上海、广州" }, "date": { "type": "string", "description": "查询日期,格式为 YYYY-MM-DD,不传默认当天" } }, "required": ["cityName"] } } } @registry.register(weather_schema) async def get_weather(cityName: str, date: str = None) -> dict: # 实际这里访问天气 API,省略 return {"city": cityName, "date": date, "temp": 26, "condition": "晴"}

写到这里有个小提醒:如果用 OpenAI 的 Function Calling 格式,Schema 外面有一层type: "function"和function包裹,很多测试过的人容易漏掉,导致模型端返回 400。Agent-Reach 的get_tool_models()会原样返回,所以注册时一定要把完整包裹结构写对。

3.3 执行层:模型意图到可靠调用的转换

执行层是核心中的核心。我从模型拿到结构化输出后,要走“解析、校验、执行、兜底”四步,缺一步都会出问题。

# executor.py import asyncio import logging from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type logger = logging.getLogger(__name__) class ToolExecutor: def __init__(self, registry, max_concurrency: int = 4): self._registry = registry self._semaphore = asyncio.Semaphore(max_concurrency) async def execute(self, call: dict) -> dict: tool_name = call.get("name") arguments = call.get("arguments", {}) tool = self._registry.get(tool_name) if not tool: return {"error": f"Unknown tool: {tool_name}", "success": False} # 参数校验:用 pydantic 或轻量规则,这里省略细节 validated_args = self._validate(arguments, tool["schema"]) if validated_args is None: return {"error": "Invalid arguments", "success": False} # 并发受限执行,带超时和重试 async with self._semaphore: try: result = await asyncio.wait_for( self._call_with_retry(tool["func"], validated_args), timeout=self._timeout_for(tool_name) ) return {"result": result, "success": True} except asyncio.TimeoutError: return {"error": "Tool timeout", "success": False} except Exception as e: logger.exception("Tool execution failed: %s", tool_name) return {"error": str(e), "success": False} async def _call_with_retry(self, func, args): try: return await func(**args) except (TimeoutError, ConnectionError): raise

模型返回的arguments是 JSON 字符串,不是对象,很多第一次对接的人会栽在这里。如果模型给出非法 JSON(截断、转义错误),我会做一个轻量修复:先尝试json.loads,失败则用正则抽取最外层花括号再试一次。实测这个修复能把参数解析成功率从 94% 拉到 99% 以上。

另一个点是参数校验。我基于 Schema 写了一套轻量校验器,类型不符时容忍转换:字符串形式的 “4” 转成整数 4,保留数字转字符串。这个宽容策略很关键,因为大模型经常把数字类型的参数格式化成字符串,直接严格模式会导致大量“参数非法”的伪失败。

3.4 让模型补全缺失参数:不要一丢了之

参数不全的情况非常常见。比如用户只说了“帮我查明天北京的天气”,但工具要求传cityName,模型可能没经过地理编码,直接传了 “北京” 进来。这个没问题。但如果工具要求传一个不在 Schema 枚举里的districtCode,模型没传,你怎么处理?

我的方案是“参数补齐回调”。当校验发现缺少 required 参数时,不直接返回失败,而是生成一条“补齐请求”信息回传给模型,让它结合历史对话继续补全。比如:

调用天气工具时缺少参数:districtCode(城市编码)。当前可用编码:110000(北京)、310000(上海)、440100(广州)。请问您要查询哪个城市?

这样模型就有了第二次机会。实测这个机制把多参数工具的成功率从 76% 拉到 92%,代价是多消耗一轮模型调用,但对用户体感来说,明显比“工具调用失败”要强得多。

3.5 接入 MCP 工具与权限检查

MCP 工具现在越来越常见,官方生态里有大量现成的mcp-server-xxx。Agent-Reach 不直接连 MCP 服务,而是写一个适配器获取其工具列表,转换成内部注册格式:

# adapters/mcp_adapter.py from mcp.client import ClientSession # 示意,具体取决于所用 MCP 客户端库 async def convert_mcp_tools(session: ClientSession): tools = await session.list_tools() schemas = [] for tool in tools: schemas.append({ "type": "function", "function": { "name": tool.name.strip(".").replace(".", "_"), "description": tool.description, "parameters": tool.input_schema } }) return schemas

注意我特意把工具名里的点号转成下划线,这类坑很隐蔽——MCP 的 root 工具名里可能带/或.,直接给 Function Calling 用,有些模型会输出不合法字符导致执行层找不到工具。

权限检查放在 executor 内。每个工具注册时可以从permission.json读取限定角色,执行器在execute()最前面校验一次。这听起来简单,但很多人会遗漏“Agent 内部调用 Agent 的场景”,也就是子 Agent 调用工具的链路。我的做法是把权限上下文作为参数一路传下去,子 Agent 只能继承父级权限的子集,不能扩大。

4. 常见问题与排查技巧实录

这个项目从第一版到现在,我踩过的坑足够写一本小册子,这里挑典型问题整理成清单。你在自己的 Agent 接入里遇到类似的,直接按表排查通常很快能定位。

4.1 模型始终不调用工具,“叫不动”

这是最让人头疼的问题。我排查时按顺序看三样东西:

第一,工具描述是否符合“触发场景 + 数据来源 + 典型问题示例”的模板。如果描述只是“查询天气”,模型很可能不知道什么时候该调它。第二,工具是否真的进入了模型 API 的 tools 列表。调试时可以打印模型的完整请求体,看看 tools 数组里有没有你注册的工具。第三,系统 Prompt 有没有和工具描述冲突。比如你在大模型 Prompt 里写了“你是一个无所不知的助手”,它会觉得没必要调工具,自己编答案算了。

一个方向对了但效果不够的情况,我会在示例里增加“典型问题”,给模型多几个对标的场景。

4.2 模型调用了错误工具或参数严重偏差

当参数偏差严重时,优先检查你的枚举值和描述。比如模型把 “Shanghai” 传给你,但你要求的是中文 “上海”——这是 Schema 里没写清楚取值范围的典型案例。补一句“城市名必须为中文全称”立刻改善。

如果模型老是调错工具,八成是两个工具的触发场景重叠了。比如“查天气”和“查历史天气”,在描述里一定要明确区分:前者用于“未来 7 天”,后者用于“历史日期查询”。否则模型会随机选一个。

4.3 工具调用耗时太长,Agent 整体体验被拖垮

我的经验是把所有工具分成“快速工具”和“慢工具”两条链路。快速工具单次调用不超过 3 秒,慢工具可以 60 秒,但要给用户返回“正在处理”的中间状态。Agent-Reach 里我给慢工具加了一个可选的pending_feedback回调,一旦超过 2 秒未完成,就生成一条中间消息插入对话流,用户不会以为是卡死了。

排查超时时,先看 tracer 记录到底卡在连接建立、服务端响应还是数据解析。我曾经遇到一个工具本身很快,但因为被并发限流排队了 20 秒,导致整体变慢——后来把并发上限从 4 调整成 2 就解决了,所以限流参数同样需要压测拟合,不是拍脑袋定的。

4.4 上下文爆炸:单轮对话塞进太多工具结果

模型每调用一次工具,结果就会塞进上下文。如果工具返回的是大 JSON(比如 Top 100 搜索结果),几轮对话下来 token 就爆了。我在执行层里加了一个summarize_result钩子,当返回内容的预估 token 超过阈值(默认 1500)时,自动做一轮“压缩摘要”,只保留任务目标相关的结构化字段,而不是把完整 JSON 原样塞回去。

这个方法除了省 token,还能减少模型被无关字段干扰的概率。比如查询订单接口返回 30 个字段,但任务只关心订单状态,压缩后只保留状态字段加一句“共 X 个订单,其中 Y 个待支付”,模型做后续决策的准确率反而更高。

4.5 问题排查速查表

现象常见原因快速处理建议
模型不调用工具描述太抽象 / tools 没传全 / 系统提示冲突重写描述,按“触发场景+示例问题”模板;打印请求体验证
参数总是字符串Schema 缺类型说明校验器加宽容转换或用description强制格式
工具调用超时慢查询没区分链路调整阈值,增加 pending 反馈
结果太长未做摘要开 summarizer,限制 token
工具权限越级permission 未下沉检查子 Agent 的权限继承

4.6 调试技巧:你缺一个好 tracer

我最后想重点表扬一下 tracer 模块。早期没做它的时

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询