我为什么要把 Agent 从“能聊”变成“能办”Agent-Reach 的设计与落地实录先交代一下背景。前段时间团队接了一个“自动补货助手”的需求老板的预期很简单让客服在后台问一句“A 类耗材还够用几天”系统不光要回答库存数量还得在低于安全线时自动生成采购申请单。用市面上现成的编排平台搭了一个原型发现最难的从来不是对话和理解意图而是让 Agent 真正触达公司内部的库存系统、审批流和消息通道。折腾了两周我干脆自己写了层连接逻辑后来这个项目内部代号就叫 Agent-Reach意思是“让智能体伸出手够到业务系统”。这篇文章不聊花哨的概念把我做 Agent-Reach 时的核心设计、最小实现、踩坑记录和适用范围都放出来。适合正在做 AI Agent、想给 Agent 接内部工具或第三方 API 的朋友也适合刚入门但被“Agent 只能返回文字”卡住的新手。1. 先搞清楚痛点Agent 缺的不是大脑是手1.1 一个只会“说话”的智能体有多尴尬我见过太多 Demo演示的时候很惊艳Agent 能理解“帮我查一下上个月华东区的退货率”然后流利地输出一段分析。但你仔细看那段分析是模型根据训练数据“编”的它根本没有去查你们公司的数据库。一旦追问“数据源是哪张表”它就含糊其辞。真正的业务场景里用户要的不是一段漂亮的文字而是一个结果查到了、提交了、通知了、办完了。这中间的每一步都需要 Agent 去调用真实世界的系统。所以问题就变成了大语言模型很擅长理解意图和生成回复但它“够不到”外部系统做不到稳定地发起一次 HTTP 请求、读取一份鉴权凭证、解析一个内部接口的返回结构。1.2 Agent-Reach 想解决的问题我给 Agent-Reach 定的目标很朴素在智能体和外部系统之间建立一个轻量的连接层。它做三件事把 Agent 的意图翻译成结构化的工具调用参数代替 Agent 去调用真实的 API、数据库、消息通道把调用结果格式化后送回给模型让模型基于事实继续对话可以把它理解为 Agent 的“手和脚”。模型依然是大脑负责决定“该做什么、按什么顺序做”但真正去执行的是 Agent-Reach 里注册好的一个个工具函数。提示如果你的场景只是“模型回答知识库检索”那不需要 Agent-Reach但只要你需要 Agent 去改数据、发消息、创建工单那这个连接层就绕不开。2. Agent-Reach 的设计骨架三层结构2.1 接入层把所有业务系统收敛成统一接口公司内部系统五花八门老 ERP 只暴露 SOAP 接口新 BI 系统是 REST API还有一堆只支持后端数据库直连的内部系统。如果让 Agent 直接对接每一个系统模型需要记住几十种鉴权方式和数据格式既不稳定也不安全。所以 Agent-Reach 的第一层是接入层统一做三件事用一个工具注册表登记所有可操作的能力查库存、创建工单、发通知把每个能力封装成标准函数对外暴露统一的入参出参结构把鉴权、重试、超时这些通用逻辑收敛到一层不让模型去关心这层设计借鉴了微服务里的 BFFBackend for Frontend思路只不过这里服务的对象不是前端页面而是大模型。2.2 路由层把自然语言请求变成工具调用序列这一层是 Agent-Reach 真正值钱的地方。大模型本身具备函数调用Function Calling / Tool Calling能力但直接裸用有两个问题一是模型可能自己乱造参数二是一次复杂请求可能需要连续调用多个工具中间还需要依赖前一个工具的结果。路由层做的就是把模型的工具调用能力管起来具体包括根据用户问题和工具描述让模型选出合适的工具并生成参数维护多轮调用状态第一次查库存的结果作为第二次创建补货单的输入设置最大调用轮次防止 Agent 在工具链里死循环这里我踩过的最大坑就是一开始没有做路由管控让模型“自由发挥”结果它连续调用同一个查询工具五次因为每次都觉得自己没拿到足够信息。加了轮次上限和中间结果缓存之后这个问题彻底解决。2.3 执行层工具函数内部的细节处理执行层是最朴实也最容易出问题的地方每一个注册进来的工具函数都要处理参数校验模型给出的参数值是否合法比如日期格式、枚举值范围错误兜底上游系统返回 500 或超时时怎么构造对模型友好的错误信息审计日志谁在什么时间调用了什么工具参数是什么结果是什么我建议把执行层和路由层做成进程解耦的。Agent-Reach 初始版本把三层塞在一个进程里结果工具函数里一个死循环直接拖垮了智能体服务。后来拆成执行器独立部署再也没出现过一个工具问题毁掉全部会话的事故。3. 从零实现一个最小可用的 Agent-Reach3.1 技术选型与项目结构实现 Agent-Reach 不需要复杂框架我用了 Python 3.11 FastAPI模型侧走了 OpenAI 兼容的 Function Calling 协议工具注册用了装饰器语法整体结构如下agent-reach/ ├── serve.py # FastAPI 服务入口暴露 /chat 和 /tools 两个接口 ├── router.py # 路由层会话状态、工具选择、调用轮次控制 ├── executor.py # 执行层动态加载并运行工具函数 ├── tools/ │ ├── registry.py # 工具注册表用装饰器登记工具 │ ├── stock.py # 示例工具查库存 │ ├── purchase.py # 示例工具创建补货单 │ └── notify.py # 示例工具发企业微信/钉钉通知 └── config.yaml # 模型配置、系统超时、鉴权地址选 FastAPI 是因为它自带异步支持和数据校验Agent 调用工具时如果有高并发场景异步能力是刚需。工具函数我用装饰器注册这样新增一个能力只需要写一个普通函数团队里不熟悉底层链路的人也能快速上手。3.2 工具注册的核心代码先看工具注册表这是 Agent-Reach 的根基# tools/registry.py from typing import Callable, Dict, Any, Optional import inspect import json class ToolRegistry: def __init__(self): self._tools: Dict[str, Callable] {} self._schemas: Dict[str, dict] {} def register(self, name: str, description: str, parameters: dict): def decorator(func: Callable): self._tools[name] func self._schemas[name] { type: function, function: { name: name, description: description, parameters: parameters, }, } return func return decorator def get_schemas(self) - list: return list(self._schemas.values()) def call(self, name: str, arguments: dict) - Any: if name not in self._tools: raise ValueError(f工具 {name} 未注册) # 这里可以做参数校验、限流、审计 return self._tools[name](**arguments) registry ToolRegistry()这个注册表的核心价值在于get_schemas()它把工具定义直接转成模型需要的 JSON Schema喂给大模型之后模型就知道有哪些工具可用、每个参数应该长什么样。工具函数本身完全不需要关注模型协议只需要接收普通参数、返回普通对象。3.3 一个真实工具函数的写法以查库存为例我把内部库存系统的 HTTP 接口封装成一个工具# tools/stock.py import httpx from tools.registry import registry from config import load_config config load_config() registry.register( namequery_stock, description查询指定SKU的当前库存余额和可用天数, parameters{ type: object, properties: { sku: {type: string, description: 商品编号例如 SKU-10086}, warehouse: {type: string, enum: [华东仓, 华北仓, 华南仓], description: 仓库名称} }, required: [sku, warehouse] } ) async def query_stock(sku: str, warehouse: str) - dict: async with httpx.AsyncClient(timeout10.0) as client: resp await client.get( f{config[stock_api_base]}/api/v1/stock/{sku}, params{warehouse: warehouse}, headers{Authorization: fBearer {config[stock_api_token]}} ) resp.raise_for_status() data resp.json() return { sku: sku, warehouse: warehouse, available: data[available_qty], daily_usage_avg: data[daily_usage_avg], days_left: round(data[available_qty] / max(data[daily_usage_avg], 0.01), 1) }注意几个细节。第一days_left是直接在工具里算好的不要让模型自己去算除法模型做数值计算远不如代码可靠。第二超时设了 10 秒比上游接口自己的超时略短避免 Agent 因为等待迟迟拿不到结果而表现异常。第三返回的是一个普通 dict而不是渲染好的自然语言这样路由层既可以把结果送给模型生成回复也可以直接作为下一个工具的入参。3.4 路由层用工具描述驱动模型决策路由层我封装了一个简单的run_agent函数核心逻辑是循环调用模型直到模型认为不需要再调工具# router.py from openai import AsyncOpenAI from tools.registry import registry client AsyncOpenAI(base_urlconfig[model_base_url], api_keyconfig[model_api_key]) async def run_agent(user_message: str, max_steps: int 5): messages [{role: user, content: user_message}] step 0 while step max_steps: resp await client.chat.completions.create( modelconfig[model_name], messagesmessages, toolsregistry.get_schemas(), tool_choiceauto ) msg resp.choices[0].message messages.append(msg) if not msg.tool_calls: return msg.content for tc in msg.tool_calls: try: result await registry.call(tc.function.name, json.loads(tc.function.arguments)) messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps(result, ensure_asciiFalse) }) except Exception as e: messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps({error: str(e)}, ensure_asciiFalse) }) step 1 return 处理步骤过多已终止请简化请求后重试这个循环是整个 Agent-Reach 最核心的机制每轮把toolsregistry.get_schemas()传给模型模型如果决定调用工具就返回结构化的tool_calls当模型拿到工具返回结果之后会基于真实数据继续推理。这既是能力边界也是稳定性的生命线。4. 我在 Agent-Reach 里踩过的四个坑4.1 模型会编造参数必须加枚举校验第一次联调时我给查询工具传了warehouse华东仓库但工具内部枚举是华东仓上游接口直接 400。模型不会因为你写了enum就每次老实遵守特别是参数来自用户口语时它可能做“合理改写”。我的解决办法是双保险一是在 JSON Schema 里写enum二是工具函数内部再做一次校验和归一化映射。直接在函数入口加一层WAREHOUSE_ALIASES {华东仓库: 华东仓, 上海仓: 华东仓, 北方仓: 华北仓} warehouse WAREHOUSE_ALIASES.get(warehouse, warehouse) if warehouse not in VALID_WAREHOUSES: raise ValueError(f无效仓库: {warehouse}可选值: {VALID_WAREHOUSES})在 Agent-Reach 里永远默认模型是不可靠的所有关键参数都必须在执行层重新校验一遍。4.2 工具返回结果太大模型直接“看不过来”有一次给 Agent 接了一个“查询全部订单明细”的工具返回了 2000 行 JSON。模型拿到这个超大 context 之后不仅响应变慢而且复述数字时开始出错。后来我把所有可能返回大数据的工具都改成了“摘要优先”策略工具先返回前 20 条明细和统计汇总比如总金额、订单数、异常数量如果用户明确需要更多明细再调用第二个分页工具。这一步改动让整个系统的准确率明显上升还顺带降低了 token 成本。4.3 鉴权失效的恢复路径没设计好企业内部系统普遍用 token 鉴权有些 token 一小时就过期。一次会话可能持续十几分钟中途 token 失效工具就直接报错。但这个报错信息如果直接丢给模型模型会一脸茫然地回复“抱歉系统暂时不可用”。我的处理是在执行层捕获 401 错误自动刷新一次 token 后重试并在返回结果里注明“已自动刷新凭证”。只有重试仍然失败时才把错误抛给模型。实测下来会话中断率从大约 10% 降到了 1% 以下。4.4 工具链死循环必须设硬性上限前面说过模型连续调用五次同一个工具的事后来我在路由层不仅设了max_steps5还增加了“同类工具连续调用上限”如果连续三步都调同一个工具就要求模型先说明理由。这样既防了死循环也保留了正常多轮查询的灵活性。5. 一个完整案例用 Agent-Reach 跑通“库存检查自动补货”5.1 定义补充工具查库存和创建补货单为了更直观我再补一个创建补货单的工具# tools/purchase.py registry.register( namecreate_purchase_order, description为指定SKU创建采购补货单, parameters{ type: object, properties: { sku: {type: string, description: 商品编号}, quantity: {type: integer, description: 补货数量}, warehouse: {type: string, description: 目标仓库}, reason: {type: string, description: 补货原因} }, required: [sku, quantity, warehouse, reason] } ) async def create_purchase_order(sku: str, quantity: int, warehouse: str, reason: str) - dict: # 调用内部采购系统创建单据 resp await httpx.post( f{config[purchase_api_base]}/api/v1/purchase-orders, json{sku: sku, quantity: quantity, warehouse: warehouse, reason: reason}, headers{Authorization: fBearer {config[purchase_api_token]}} ) resp.raise_for_status() data resp.json() return {purchase_order_id: data[id], status: data[status], created_at: data[created_at]}5.2 用户请求与 Agent 的完整处理链路用户输入“SKU-10086 在华东仓还够用几天如果低于 7 天自动创建一张补货 100 件的采购单。”Agent-Reach 的处理流程是这样走的第一轮模型判断需要调query_stock参数{sku: SKU-10086, warehouse: 华东仓}执行层调用库存接口返回{available: 35, daily_usage_avg: 6, days_left: 5.8}第二轮模型拿到结果判断 5.8 天低于 7 天阈值决定调create_purchase_order参数{sku: SKU-10086, quantity: 100, warehouse: 华东仓, reason: 库存余量不足7天}同时参考了days_left的计算逻辑执行层创建补货单返回单号和状态第三轮模型把结果汇总成自然语言“SKU-10086 当前可用 35 件按日均 6 件的消耗速度约可支撑 5.8 天低于 7 天安全线。我已自动创建补货单 PO-20250618-001补货 100 件。”整个过程里模型只负责决策所有数字都来自真实接口最终展示给用户的结论是可信的。5.3 这个案例给到的重要启示Agent-Reach 真正厉害的点不是“会调用 API”而是把判断逻辑也交给了模型模型看了days_left之后自己决定是否创建补货单这就是把规则决策从硬编码变成了自然语言驱动。如果你想把阈值改成年订单量动态计算只需要改工具返回值里的指标不需要动路由代码。提示对于补货单这类真实写操作的场景我强烈建议在执行层加“权限校验人工确认”逻辑比如金额超过一定阈值时工具返回“需要审批”Agent 会引导用户在页面点确认。别上来就做全自动。6. Agent-Reach 的适用边界什么场景不要硬上6.1 适合做的场景我验证过几个特别适合 Agent-Reach 的场景效果都不错企业知识库问答 数据查询让 Agent 先定位知识库文档确定口径再查真实数据回答工单处理用户描述问题Agent 根据描述填单、分类、自动指派数据周报生成Agent 自动跑数、汇总趋势、生成简报初稿人只负责审核设备运维辅助Agent 查状态、看日志、定位异常并尝试触发重启流程这些场景的共同特点动作可回滚、单次操作影响面小、失败有补偿路径。6.2 不适合硬上的场景有几类项目我建议离远一点直接操作资金转账、支付流程的场景失败代价极高应该走严格的人工审批流需要为决策负法律责任的场景比如医疗诊断、法律意见上游系统本身极不稳定的场景Agent 会因为频繁的超时和报错变成“客服复读机”6.3 团队落地 Agent-Reach 之前必须准备的几件事把 Agent-Reach 从 Demo 推到生产我建议先做好三件事。第一把工具分级管理只读工具放开给 Agent 自主调用写操作工具默认需要人工确认。第二做好审计每次工具调用的参数和结果都要落日志万一出了问题能追溯。第三给 Agent 设计能力边界工具只能访问它需要的系统用单独的 service 账号跑执行器别用员工个人的 token。这三件事做完Agent-Reach 才真正从“能跑”变成“可靠”。最后说一点我个人的体会吧。最开始我以为 Agent-Reach 最大的难点是接口对接真做下来才发现难点在于如何让模型在每一步都能拿到足够结构化、足够准确的反馈。工具像积木路由像图纸但真正让整个系统转起来的是每一块积木都足够结实。所以如果你也在做类似的东西不要急着堆功能先把一两个工具做到极致可靠再逐步扩展。这条路走通了后面再多的 Agent 场景都是水到渠成的事。
阅读完成 · 觉得有帮助?