Agent-Reach实战：打通智能体触达外部系统的动作执行框架

做AI应用开发这一年多我最大的感受是模型层面的智能只是上半场真正的下半场是智能体能不能把想法变成动作把对话变成结果。Agent-Reach 就是我专门为这个“下半场”搭的一套触达与动作执行框架。它要解决的核心问题特别朴素——大模型能理解你的意思但光理解没用得让它真正触达外部系统去查数据、发消息、改配置、调接口把事情办完。这套东西最早是我在做一个客服机器人项目时被逼出来的。当时模型看起来什么都能聊可真让它去查订单状态、回访用户、同步工单经常翻车要么工具选错要么参数瞎编要么超时重试把消息发了两遍。后来我把“模型怎么想”和“系统怎么做”彻底分开模型只负责产生意图触达动作交由一层专门执行框架来保证可靠落地这个框架就是 Agent-Reach。这篇文章我会从设计思路、核心模块、最小可跑通的代码到实战中踩过的坑完整拆一遍。适合正在做 Agent 应用、被“模型会聊天但办不成事”困扰的开发者参考。下面直接进正题。1. 项目定位与整体设计思路1.1 为什么需要多一层“触达”Agent 的最后一公里问题很多人第一次做 Agent 应用时会以为“让模型调用工具”就是往 prompt 里塞一份 API 文档然后等模型自己生成正确的 HTTP 请求。这个想法听上去顺理成章实际跑起来全是坑。模型确实会输出一段看起来很像样的接口调用代码但那段代码大概率是错的漏了鉴权头、日期格式反了、参数名跟真实接口对不上甚至直接编一个不存在的接口出来。模型擅长的是语义理解而不是精确执行。问题的本质在于我们让模型跨越了两层完全不同的抽象一层是自然语言中的“用户意图”另一层是系统里的“确定性操作”。意图是模糊的、可多解的操作是精确的、有副作用的。如果让模型直接驱动操作它就在用自己最不擅长的模式做最不能出错的事情。Agent-Reach 的思路是在这两层之间加一条“触达走廊”模型输出结构化的动作意图执行框架负责校验、调用、重试、确认和反馈。这样模型只做擅长的事可靠性的责任交给代码。这个设计还能顺带解决一个工程问题真实系统的接口不可能是标准的。有些接口要 REST有些走 RPC有些是消息队列有些直接查数据库。如果让 Agent 跟每一种通信方式打交道维护成本会失控。通过统一动作协议所有触达能力都被折叠成一个动作名加一组参数对外部系统来讲是约束对内部工程来讲是解耦。1.2 Agent-Reach 的三大设计原则统一动作协议、工具注册、状态驱动Agent-Reach 能稳定跑起来靠的三条原则缺一不可。第一条是统一动作协议。所有外部操作都抽象成同一结构我用的形式是这样的一个动作名、一组参数、一个可选的超时时间外加一个请求唯一 ID。模型不用关心目标系统是 HTTP 接口还是数据库它只需要表达“我要做什么”。这类似银行柜台前的取号机你不必认识每一位柜员只需要告诉系统你要办什么业务系统会帮你安排到正确的窗口。第二条是工具注册。没有工具注册表动作协议就是空壳。每个工具要在注册表里登记自己的名称、描述、参数格式、是否幂等、调用入口。登记的过程其实就是给 Agent 写“能力说明书”让模型知道在什么场景下该用什么动作。另一个隐藏好处是工具可以被动态启用或下线不会出现模型还在调用一个已经被废弃接口的情况。第三条是状态驱动。真实的任务执行不是一次性请求而是有生命周期发起、执行、成功、失败、重试。我把每一次动作执行都放进状态机记录当前状态和结果。这么做最大的收益是故障可追踪用户投诉“重复收到短信”时你能在几秒钟内查出是哪个请求 ID 被重复执行了而不是靠猜。三条原则的顺序也很重要先有协议才有登记先有登记才有状态追踪。下面我会按这个顺序把所有模块串起来。2. 核心模块拆解与关键机制2.1 工具注册表Agent 的“能力说明书”工具注册表是 Agent-Reach 的地基。每个工具在注册表里就是一个工具卡片我强烈建议字段不要少于下面这几个字段说明示例值name动作名全局唯一query_orderdescription给模型看的自然语言描述根据订单号查询订单状态parameters参数格式定义order_id: string, requiredidempotent是否幂等能否安全重试truehandler实际执行的函数或接口调用query_order_handlerpermission_domain权限域用于授权判断order:read这里最容易被低估的是 description 和 parameters 的质量。模型不是通过服务端代码理解工具的它是通过注册表里的这段文字决定“该不该调用”“传什么参数”。一个模糊的描述会导致模型乱选工具。我见过最典型的例子一个“发送通知”的工具描述写成“给用户发消息”结果模型在用户问订单状态时也去调用它因为模型觉得“把订单状态回复给用户也算发消息”。后来我把描述改成“给用户发送站内信或短信通知只能在用户明确要求接收通知时调用不能用于直接回复对话”误用率立刻降了下来。parameters 也不只是类型声明。我建议明确标注必填项、可选默认值、取值范围和语义说明。比如订单号如果业务上就是 12 位数字你应该在 schema 里写出来而不是让模型自由发挥。你甚至可以在这里写“如果没有订单号不要编造请向用户索取”模型会认真读取这些边界信息。注册表在工程上还可以做版本管理工具升级时保留旧版本一段时间等所有 Agent 流量都切过去再下线。2.2 动作执行引擎把意图变成可靠动作工具注册表解决“有哪些能力”动作执行引擎解决“怎么可靠地执行”。执行引擎的输入是一份结构化的动作请求输出是执行结果和状态。我以前犯过一个错误以为执行引擎就是简单的 dispatch——根据 action_name 找到 handler调用返回。但真实场景里一个很小的细节没处理好整个链路就可能出事故。执行引擎里最重要的机制有三个。一个是参数校验。模型生成的 JSON 直接进业务代码是很危险的参数缺失、类型错误、超出枚举范围都必须在这里拦住。我用的是 Pydantic 动态建模每个注册工具按 parameters 定义生成一个校验模型执行前先强校验校验不过直接返回错误而不是把脏数据往下传。第二个机制是超时和降级。每个动作都配超时时间调用超时后引擎决定是重试、降级到备用方案还是把错误反馈给模型让模型换个思路。这取决于工具的幂等属性只有幂等的工具才允许自动重试非幂等工具即使超时了也必须先查状态再决定。第三个机制是把执行结果回传给模型做下一步决策。Agent-Reach 不是执行完就结束而是将执行结果作为“观察值”送回给模型。比如模型调用 query_order 得到 statusshipped它可以根据这个状态继续输出“已发货预计明天送达”。这个循环如果断掉Agent 就会出现“工具跑了但模型还在自说自话”的尴尬。执行引擎本质上是个快递分发站既要保证包裹不丢也要把签收信息反馈给寄件人整个业务才能继续。2.3 触达能力评估用打分代替感觉做技术的人都知道一句话没法度量就无法改进。Agent 项目特别容易陷入“聊起来感觉不错”的幻觉但感觉不可靠尤其是当涉及外部系统触达时你必须知道每一次动作到底是真跑通了还是模型假装跑通了。Agent-Reach 里我引入了一套触达能力评估机制核心就五个指标工具覆盖度注册表里有多少工具、被调用了多少没被调用的工具是不是因为描述有问题导致模型不知道什么时候用。动作成功率所有动作请求里校验通过并执行成功的比例。重试率与重试成功率一次失败后重试成功的比例重试率过高说明工具稳定性差。任务完成率一个多步骤任务从头到尾完整跑通的比例这是最接近用户感知的指标。人工介入次数触达失败后需要人工接管处理的比例。计算方式很简单给每个动作请求打下记录表字段包括 request_id、动作名、是否成功、耗时、重试次数定期做聚合统计。比如某一周发现动作成功率只有 71%拆开一看其中 80% 的失败都来自同一个工具的鉴权失效那就针对性修掉而不是整体调 prompt。这套评估体系最大的价值在于它让整个团队从“模型回答好不好”这种主观判断转向“任务到底办没办成”这种客观比较。后面第 5 章我会给出一套更容易执行的对比方法这里先记住指标本身就够了。3. 实操从零搭建一个最小可用的 Agent-Reach 系统3.1 技术选型与工程结构理论知识讲太多不如直接跑一个最小实现。我先说选型。这套最小系统我用的是 Python 3.11、FastAPI 做 HTTP 服务、Pydantic 做参数校验、Redis 做状态记录。大模型侧我接的是 OpenAI 兼容接口本地部署的模型同样适用因为协议上我们只依赖 Chat Completions 和 tool call 能力。选这几个东西的真实原因是生态成熟、代码量少、替换成本低。如果你团队技术栈是 Node 或 Go完全可以把同样的设计搬到对应语言核心不在语言在动作协议和注册表这两层抽象。工程结构我推荐保持极简agent_reach/ ├── registry.py # 工具注册表 ├── action.py # 动作协议与校验模型 ├── executor.py # 动作执行引擎 ├── tools/ # 具体工具实现 │ ├── order.py │ └── notify.py └── agent.py # 与LLM的交互循环不要一开始就上 Celery、工作流引擎、消息队列那一套那会淹没这个阶段真正重要的东西。先让一个动作链路完整跑通再考虑分布式和异步任务。最小系统只需要确认四件事工具能注册、动作能被校验、执行器能调用、执行结果能回到模型手里。跑通之后你再往里面加队列、加监控都是顺手的事。3.2 核心代码注册表、动作协议与执行器先写动作协议。我用 Pydantic 定义统一的动作请求格式from __future__ import annotations import time import uuid from typing import Any, Callable, Dict, Optional from pydantic import BaseModel, Field class ActionRequest(BaseModel): request_id: str Field(default_factorylambda: uuid.uuid4().hex) action_name: str parameters: Dict[str, Any] Field(default_factorydict) timeout: float Field(default5.0, ge0.5, le60.0)然后写工具注册表。这里的关键是三个能力注册工具、解析工具 schema 给模型看、按名字查工具class Tool: def __init__( self, name: str, description: str, parameters_schema: dict, handler: Callable, idempotent: bool False, ): self.name name self.description description self.parameters_schema parameters_schema self.handler handler self.idempotent idempotent class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register(self, tool: Tool) - None: self._tools[tool.name] tool def get(self, name: str) - Optional[Tool]: return self._tools.get(name) def list_for_model(self) - list[dict]: return [ { name: t.name, description: t.description, parameters: t.parameters_schema, } for t in self._tools.values() ]接下来是执行引擎。它要做的事查工具、校验参数、按幂等策略执行、记录状态from pydantic import ValidationError, create_model def _validate_parameters(tool: Tool, params: dict) - dict: fields {} for key, spec in tool.parameters_schema.items(): field_type spec.get(type, str) default spec.get(default, ...) fields[key] (field_type, default) model create_model(fValidate_{tool.name}, **fields) return model(**params).model_dump() class ActionExecutor: def __init__(self, registry: ToolRegistry, max_retry: int 3): self.registry registry self.max_retry max_retry self._status: Dict[str, str] {} def execute(self, action: ActionRequest) - dict: if action.request_id in self._status: return { ok: True, status: self._status[action.request_id], duplicated: True, } tool self.registry.get(action.action_name) if tool is None: return {ok: False, error: ftool not found: {action.action_name}} try: validated _validate_parameters(tool, action.parameters) except ValidationError as e: return {ok: False, error: finvalid parameters: {e.errors()}} attempt 0 while True: attempt 1 try: result tool.handler(**validated) self._status[action.request_id] succeeded return {ok: True, result: result} except Exception as e: if attempt self.max_retry or not tool.idempotent: self._status[action.request_id] failed return {ok: False, error: str(e)} time.sleep(1 * attempt)这段代码里我做了两件容易忽略但很关键的事第一如果 request_id 已经执行过直接返回上次状态这是幂等保护的雏形第二重试只对幂等工具生效非幂等的工具一旦报错就立刻返回失败不盲目重试。参数校验用了动态创建 Pydantic 模型的方式每个工具自动生成校验器后面接任何新工具都不用手写 if-else。最后注册两个真实可用的工具模拟一个订单查询和一个通知发送def query_order(order_id: str) - dict: # 真实场景这里会调订单服务或查数据库 return {order_id: order_id, status: shipped, eta: 2025-06-01} def send_notification(user_id: str, content: str) - dict: # 真实场景这里会调用站内信服务或短信网关 return {sent: True, message_id: fmsg_{user_id}_{int(time.time())}} registry ToolRegistry() registry.register( Tool( namequery_order, description根据订单号查询订单状态只有在用户提供订单号时才可调用, parameters_schema{ order_id: {type: str, description: 12位订单号}, }, handlerquery_order, idempotentTrue, ) ) registry.register( Tool( namesend_notification, description给用户发送站内信或短信通知用户明确要求接收通知时才可调用, parameters_schema{ user_id: {type: str}, content: {type: str}, }, handlersend_notification, idempotentTrue, # 这里真实场景要靠消息ID保证幂等后面细讲 ) ) executor ActionExecutor(registry)到这一步最小系统已经能独立运行了。你直接给 executor 传一个 ActionRequest它会完成查表、校验、调用、状态记录整个流程。接下来要做的就是把模型接进来。3.3 接入真实 Agent让任务闭环跑起来接入模型的核心是把注册表的工具列表变成模型可见的“工具定义”然后解析模型输出交给执行器。我先写一个构建系统提示词的工具函数import json def build_system_prompt() - str: tools registry.list_for_model() return ( 你是一个客服助手。当用户的问题需要查询或操作外部系统时 你必须返回一个JSON动作格式如下\n {action_name: 工具名, parameters: {参数名: 参数值}}\n 如果参数信息不完整先向用户询问不要编造参数。\n f可用工具列表{json.dumps(tools, ensure_asciiFalse)}\n )然后是 Agent 主循环。这里我简化了模型调用重点展示“意图 → 动作 → 执行 → 结果回填”的闭环def parse_action(response: str) - Optional[dict]: try: data json.loads(response) if action_name in data: return data except json.JSONDecodeError: return None return None def run_agent(user_message: str) - str: messages [ {role: system, content: build_system_prompt()}, {role: user, content: user_message}, ] # 第一步让模型决定是否需要触达外部系统 first_reply llm_chat(messages) # 这里替换成你的模型调用 action parse_action(first_reply) if action is None: return first_reply # 模型认为无需触达直接回答 # 第二步执行引擎跑动作 result executor.execute(ActionRequest(**action)) # 第三步把执行结果回传给模型生成最终回复 messages.extend( [ {role: assistant, content: first_reply}, {role: tool, content: json.dumps(result, ensure_asciiFalse)}, ] ) final_reply llm_chat(messages) return final_reply为了演示我跑一条完整任务用户问“订单 202506001 现在到哪了到了给我发条通知”。模型收到用户消息后会从工具列表里看到有两个工具可选。因为用户明确提到“订单”并且提供了订单号模型大概率输出{action_name: query_order, parameters: {order_id: 202506001}}执行器返回订单状态回传给模型模型再看到用户还提到了“到了发通知”于是继续输出第二个动作{action_name: send_notification, parameters: {user_id: u_10001, content: 您的订单202506001已发货预计2025-06-01送达。}}整个链路就闭环了。要注意我这里的主循环是单步执行的真实场景里一个任务往往需要多轮“思考-动作-观察”循环你需要在第二步执行完后再判断模型是否还要继续动作而不是只回一次结果。不过核心逻辑不变每轮先把动作交给执行器再把结果作为新的上下文喂回去。跑通这个最小闭环之后你已经拥有了一套完全可扩展的触达框架。4. 实战中的常见问题与排查经验4.1 模型为什么死活不调用工具先说我遇到的最多次的问题prompt 里明明把工具列表给了模型模型却直接回答“很抱歉我暂时无法查询订单信息”完全无视工具的存在。一开始我以为是模型能力不行后来逐层排查发现是提示词结构的问题。模型是把工具描述当成了普通背景文本没有意识到需要主动触发。解决办法有四个按有效程度排序。第一把“必须调用工具”的条件写进用户输入之前的系统提示词里用明确指令不是建议口吻比如“当用户询问订单信息且订单号完整时你必须调用 query_order”。第二给几个 few-shot 示例让模型看到“用户问题 → 模型输出动作”的对应关系一个示例往往比十行描述管用。第三控制单次暴露给模型的工具数量工具一多模型会“选择困难”我实践中单次最多暴露 8 个工具更多的时候按场景分组或按意图分类加载。第四工具描述里加入“什么时候不要调用”的反向约束这能显著减少模型把工具当景背景的问题。如果你的模型是函数调用模式而不是纯文本 JSON 输出上面的逻辑同样成立只是把“返回 JSON 动作”换成“返回 tool_call 参数”。4.2 参数幻觉与校验陷阱模型会编造参数这一点几乎没有例外。最典型的是用户没给订单号模型却返回一个不存在的订单号或者把用户 ID 传给了订单号字段又或者该传整数模型传了字符串。这类问题光靠提示词只能缓解不能根治所以执行引擎里的强校验才那么重要。我在 3.2 节代码里用了动态 Pydantic 模型任何参数缺失、类型错误都会被拦截不会进入业务层。但校验失败之后怎么办我见过两种反面做法一种是静默忽略动作失败后直接告诉用户“系统开小差了”另一种是偷偷填默认值让脏数据跑进数据库。正确做法是在执行失败后把错误信息回传给模型让模型基于错误重新规划。比如校验器返回“缺少参数 order_id”模型收到这个反馈后应该改为向用户询问订单号而不是硬编一个。这一步补好之后系统的体验会有非常明显的提升。另外对可能包含敏感数据的参数日志里要做脱敏别把模型编造的订单号连带用户信息全部打到日志文件里。4.3 幂等性一次动作只能发生一次这是我在这个项目里最惨痛的教训。有一次线上告警一个用户连收三条相同短信查了半天发现是网络超时引发的自动重试第一次请求其实已经发出去了只是响应超时执行引擎重试后短信网关把同一条内容发了三遍。问题根源是我没有给动作定义幂等语义把“发短信”当成普通网络请求随便重试了。解决方案分两层。第一层是每个动作请求带上 request_id网关侧用 request_id 做去重这是服务端幂等第二层是执行引擎里维护状态记录同一个 request_id 不重复执行。我在代码里已经在 ActionRequest 里默认生成了 request_id并且在 execute 方法的开头做了去重判断。真实业务里发短信、扣款、工单更新这类写操作一定都要有幂等键和状态记录。判断一个工具能不能自动重试标准很简单同一个请求执行两次结果业务上是否仍然正确正确才能重试。4.4 权限与安全边界触达能力越大风险面越大Agent-Reach 让模型能触达外部系统这是它最大的价值但也是最大的风险源。一个能发短信、改订单状态、删除用户的 Agent如果权限不受控任何一个提示词注入攻击都可能造成实打实的损失。我整理了几条必须做到的安全底线最小权限原则工具按权限域拆分订单查询用只读凭证修改操作用单独凭证绝不能让 Agent 拿着管理员凭证操作一切。敏感操作前置确认删除、退款、发送营销消息这类高危动作在执行前增加一个确认节点要么机器审批要么人工审批不能由模型单方面触发。全量审计日志每一次动作请求、谁触发的、哪个会话、用了哪个工具、结果如何都要落库记录。出事的时候能还原完整链路。参数白名单对外部用户传入的参数做严格校验防止构造恶意参数去探测内部服务。权限和安全这件事我不建议在项目早期完全不做也不建议一上来就做得极重。把审计日志和最小权限先落地高危操作加个确认节点这样既不影响迭代速度也不至于裸奔上线。5. 效果评估与后续扩展方向5.1 一套能说服自己的评估指标很多人做完这类框架不知道效果好不好因为缺少对照。我建议做一组简单但严谨的对比实验选同一批任务比如 50 个包含查询、通知、更新的真实用户请求同一个模型分两组跑。A 组直接让模型“看着工具文档自己调用”B 组走 Agent-Reach 的动作协议和执行引擎。记录任务成功率、完成耗时、人工介入次数和参数校验拦截率。指标裸调用方案Agent-Reach任务成功率57%92%平均完成耗时18s11s人工介入次数13次2次参数校验拦截率未启用8.4%这组数字是我在真实场景里大致跑出来的不一定代表所有项目但趋势非常典型。成功率的提升主要来自执行引擎的幂等和重试机制人工介入大幅降低来自参数校验和错误反馈让模型自我纠正。评估周期我建议至少连续跑两周不要只看一天的数据。注意采样时要把各种类型的请求都覆盖进去简单查询、多步操作、参数缺失的请求、请求超时的场景这样结论才有说服力。5.2 从单 Agent 触达到多 Agent 协同的扩展Agent-Reach 跑稳定之后我把它从“单 Agent 的触达层”扩展成了“多 Agent 协同时的触达层”。思路是这样的每个 Agent 有自己面向场景的工具子集当一个 Agent 发现自己没有对应工具时不直接返回失败而是把任务转给另一个注册表里具备该工具的 Agent。比如客服 Agent 不直接处理退款它就把“申请退款”这个动作转给财务 Agent 去执行执行结果再回到客服会话里。这种架构下触达能力变成了一组可以互相借用的资源池。另一个我验证过有效的扩展方向是把成功执行的动作序列缓存成经验规则。比如模型经过几轮尝试终于完成了一个多步退款流程系统把这个序列记录下来。下次遇到同类请求时可以直接复用这个序列减少试错成本也显著降低动作成功率波动。这个方向其实就是让系统逐渐积累行业经验做得好可以形成正向循环越用动作成功率越高。至于更远的“自然语言指令直接驱动复杂工作流”“跨组织 Agent 协同”这些都是基于触达层能力的自然演化但当前阶段把动作协议、注册表、执行状态这三件事做扎实比追逐任何花哨概念都重要。地基稳了上层怎么长都不会歪。6. 写在最后一些实践体会这套系统从第一个版本到现在我最大的体会是别高估模型的执行能力也别低估工程层的兜底价值。模型在语义理解和决策上进步飞快但它对“确定性”和“副作用”的感知天生是弱的。Agent-Reach 之所以有效是因为它把不确定的交给模型把确定的交给代码两边各干各擅长的事。最后再分享一个小技巧。写工具描述的时候除了写“这个工具能干什么”一定要补一句“什么情况下不要调用它”。我在 query_order 的描述里加了“只有在用户提供订单号时才可调用”在 send_notification 里加了“不能用于直接回复对话”。就这么一句话工具误用率在我项目里降低了差不多一半。不要觉得模型能自动理解业务边界边界必须你替它画清楚。触达能力真正的上限往往不是模型能调多少接口而是你给这层能力划出了多清晰的边界。