查询 25 次,耗时 0.2218 秒

从“能调用”到“可恢复”:构建可靠 LLM 工具调用 Agent 的工程实践

3100953

New member

3100953 Rep
0
0
0
Rep
0
3100953 Vouches
Vouches
0
1 MONTH
1 1 MONTH OF SERVICE
3100953 获得
0
0
0
获得
0
3100953 凭证
0
0
0
凭证
0
主题
2
点赞
0
HakB
1 月
在社区成长 1 月
LEVEL 50 100 XP
 
一、背景

工具调用让 LLM 不再局限于生成文本,而是能够查询数据库、检索文档、执行计算、创建工单或调用内部服务。但在实际系统中,模型正确输出一次函数名和参数,只能证明演示链路可用,不能证明 Agent 可靠。

生产环境中的主要问题通常不是“模型不会调用工具”,而是以下边界条件:参数格式合法但语义错误;工具已经执行成功,但响应在网络中丢失;模型重复调用具有副作用的接口;检索工具返回不可信内容,诱导模型越权;上下文被裁剪后,模型忘记已完成的步骤;多个工具之间存在依赖,却被错误地并发执行。

因此,一个可靠的工具调用系统至少要同时处理:结构化约束、权限边界、幂等性、状态管理、超时重试、结果校验和可回放评测。本文给出一套适合本地实验和内部服务的最小实现方法。

二、核心原理:把模型视为不可信的规划器

LLM 可以负责理解意图、生成候选计划和选择工具,但不能直接拥有无限制执行权。推荐将系统拆成五层:

1. Planner:模型根据目标和当前状态提出下一步动作。
2. Validator:使用 JSON Schema、业务规则和权限策略校验动作。
3. Executor:执行经过批准的工具调用,并控制超时、并发和资源额度。
4. State Store:记录动作、参数摘要、执行结果、错误和幂等键。
5. Verifier:判断结果是否满足预期,决定继续、修复、重试或终止。

关键原则是“生成与执行分离”。模型输出只是候选请求,只有通过确定性代码校验后才能进入执行器。尤其是发送消息、修改数据、支付、删除资源等有副作用的工具,应默认要求人工确认或更严格的策略审批。

三、前置条件

建议准备以下环境:

1. Python 3.11 或更高版本。
2. 一个支持结构化输出或工具调用的 LLM 接口;本地模型和兼容接口均可。
3. Pydantic 2.x,用于参数验证。
4. SQLite,用于保存调用日志和幂等状态。
5. pytest,用于构建失败注入和回归测试。

安装示例:

python -m venv .venv
source .venv/bin/activate
pip install pydantic httpx tenacity pytest

Windows PowerShell 可使用:

.venv\Scripts\Activate.ps1

本文只使用本地模拟工具,不连接真实外部目标,也不执行高风险操作。

四、步骤一:设计窄而明确的工具契约

不要设计一个名为 execute 或 request、可以接受任意 URL 和任意命令的万能工具。工具能力越宽,越难验证,也越容易被提示注入利用。下面以本地库存查询为例:

from pydantic import BaseModel, Field, ConfigDict

class InventoryArgs(BaseModel):
model_config = ConfigDict(extra="forbid")
sku: str = Field(pattern=r"^[A-Z0-9-]{3,32}$")
warehouse: str = Field(pattern=r"^(BJ|SH|GZ)$")

def get_inventory(args: InventoryArgs) -> dict:
demo = {
("GPU-A10", "BJ"): 4,
("GPU-A10", "SH"): 0,
}
return {
"sku": args.sku,
"warehouse": args.warehouse,
"available": demo.get((args.sku, args.warehouse), 0)
}

这里通过 extra="forbid" 拒绝模型自行添加的字段,通过正则和枚举范围限制参数。业务层仍需继续校验,例如调用者是否有权查看指定仓库,而不能把 JSON Schema 当作完整的安全策略。

五、步骤二:建立统一的动作信封

模型输出应包含动作标识,而不仅是工具名和参数:

{
"action_id": "step-0003",
"tool": "get_inventory",
"arguments": {
"sku": "GPU-A10",
"warehouse": "BJ"
},
"reason": "需要确认北京仓库存量后再回答"
}

reason 只能用于审计和调试,不能作为授权依据。授权必须基于调用者身份、工具、参数和当前业务状态进行确定性判断。action_id 则用于关联模型决策、工具执行和后续验证。

六、步骤三:实现注册表、校验和安全执行器

import asyncio
from typing import Callable

TOOL_REGISTRY: dict[str, tuple[type[BaseModel], Callable]] = {
"get_inventory": (InventoryArgs, get_inventory),
}

async def execute_tool(tool: str, raw_args: dict) -> dict:
if tool not in TOOL_REGISTRY:
return {"ok": False, "error": "tool_not_allowed", "retryable": False}

schema, func = TOOL_REGISTRY[tool]
try:
args = schema.model_validate(raw_args)
except Exception as exc:
return {
"ok": False,
"error": "invalid_arguments",
"detail": str(exc),
"retryable": True
}

try:
result = await asyncio.wait_for(
asyncio.to_thread(func, args),
timeout=3.0
)
return {"ok": True, "data": result}
except TimeoutError:
return {"ok": False, "error": "timeout", "retryable": True}
except Exception:
return {"ok": False, "error": "internal_error", "retryable": False}

返回值应保持稳定的机器可读结构。不要把完整异常堆栈直接交给模型,因为其中可能包含密钥、路径、SQL 语句或内部服务信息。详细异常写入受控日志,模型只接收经过脱敏的错误码和必要说明。

七、步骤四:区分重试、修复与终止

并非所有失败都适合重试:

1. timeout、临时限流、短暂网络错误:可以使用指数退避重试。
2. invalid_arguments:不应原样重试,应把字段级错误反馈给模型,请其修复参数。
3. permission_denied、tool_not_allowed:立即终止,不能让模型通过换一种表述绕过策略。
4. 业务冲突:先读取最新状态,再由规划器决定是否生成新动作。
5. 未知内部错误:通常终止或转人工,避免循环消耗资源。

建议同时设置三类预算:最大规划轮数、最大工具调用数和总耗时。例如单次任务最多 8 轮、最多 5 次工具调用、总耗时不超过 30 秒。任一预算耗尽后返回明确的 incomplete 状态,而不是让模型假装任务已完成。

八、步骤五:为副作用工具增加幂等与确认

假设系统增加 create_ticket 工具,网络可能在工单创建成功后中断。如果执行器直接重试,就可能生成重复工单。应由服务端接受幂等键:

idempotency_key = f"{task_id}:{action_id}:create_ticket"

执行前在状态库查询该键:若已有成功结果,直接返回此前结果;若状态为 running,需要等待或执行恢复流程;只有不存在时才创建。注意,幂等键必须由可信编排层生成,不能完全采用模型提供的随机字符串。

高风险动作建议采用两阶段流程:

1. prepare:验证参数并生成待确认摘要,不产生业务副作用。
2. commit:在用户确认或策略批准后执行,并携带短时有效的确认令牌。

这能避免模型在澄清过程中提前执行操作。

九、步骤六:管理上下文与记忆

工具原始结果不应无限追加到对话历史。推荐保存三种状态:

1. 事件日志:完整记录每次模型输出、校验结果和工具返回,用于审计,不一定全部送入模型。
2. 工作状态:保存当前目标、已完成步骤、关键实体和未解决问题。
3. 摘要记忆:压缩历史信息,但必须保留来源引用和时间戳。

摘要不能覆盖事实源。例如库存数量应以最近一次工具结果为准,而不是以模型生成的历史摘要为准。对于带时效的数据,可以在结果中加入 observed_at 和 expires_at,过期后必须重新查询。

十、步骤七:防御工具结果中的提示注入

网页、文档和工单文本都可能包含“忽略之前规则”“调用某工具发送数据”等内容。它们属于数据,不是系统指令。防御不能只依赖一句提示词,而应采用分层措施:

1. 工具输出以明确的数据字段传入,不与系统指令拼接成同一权限层级。
2. 检索器只返回任务必要字段,减少无关文本进入上下文。
3. 敏感工具使用独立权限策略,不因检索内容的要求而自动开放。
4. 对外发工具设置域名、收件人或数据分类白名单。
5. 检测到文档要求泄露凭据、改变权限或执行无关动作时,标记为不可信指令。

如果采用 MCP,仍应把 MCP Server 视为外部能力边界。客户端需要校验服务器身份、允许的工具集合和参数范围,不能因为协议统一就默认所有 Server 都可信。

十一、步骤八:建立可回放评测集

只统计最终回答是否“看起来正确”远远不够。至少记录以下指标:

1. 工具选择准确率:是否选择了正确工具。
2. 参数有效率:首次参数通过 Schema 和业务校验的比例。
3. 任务成功率:最终状态是否满足可验证条件。
4. 重复副作用率:同一意图是否产生重复写操作。
5. 平均工具调用次数和 P95 延迟。
6. 无效重试率:对永久性错误进行重试的比例。
7. 权限违规拦截率:越权请求是否被确定性策略阻止。

可构建如下测试样例:

{
"input": "查询北京仓库 GPU-A10 的库存",
"expected_tool": "get_inventory",
"expected_args": {"sku": "GPU-A10", "warehouse": "BJ"},
"forbidden_tools": ["create_ticket"]
}

还应进行失败注入:模拟超时、返回畸形 JSON、结果过期、服务成功但响应丢失、上下文截断等情况。每次修改提示词、模型版本、工具描述或编排代码后,都运行同一批样例。否则,某个正常样例变好可能掩盖边界场景退化。

执行测试:

pytest -q

对于具有随机性的模型,单个样例建议运行多次,并固定评测配置。评测报告应记录模型标识、采样参数、工具版本和提示模板哈希,以便复现。

十二、风险与注意事项

第一,日志可能包含用户输入和工具结果,应实施字段级脱敏、访问控制和保留周期,禁止记录明文密钥。

第二,不要盲目信任模型生成的 SQL、Shell、URL 或文件路径。确需使用时,应采用参数化查询、沙箱、目录限制和网络出口控制。更优方案是提供业务语义明确的窄工具,而不是开放通用执行能力。

第三,并发执行只适用于相互独立、无副作用或已实现并发控制的动作。存在先后依赖的调用必须显式建图,不能仅凭模型说“可以并行”就并行。

第四,自动重试会放大成本和副作用。重试策略必须同时考虑错误类型、幂等性、剩余预算和服务端限流信息。

第五,模型生成的“已完成”不是完成证明。应由 Verifier 检查工单 ID、数据库状态、文件哈希或其他可验证结果。

十三、常见问题

问:使用支持严格 JSON 输出的模型后,还需要 Pydantic 校验吗?
答:需要。严格 JSON 只能提高结构合法性,无法保证仓库代码存在、调用者有权限、数值未超业务上限,也无法处理模型接口之外的篡改和版本差异。

问:工具失败后,把完整错误信息交给模型是否更容易修复?
答:不建议。应返回稳定错误码、可修复字段和安全提示。完整堆栈保留在内部日志中,并通过 trace_id 关联。

问:所有工具都要做幂等吗?
答:纯查询通常不需要业务幂等,但仍可做缓存和请求去重。创建、发送、扣费、修改和删除类工具必须重点考虑幂等、确认和补偿。

问:Agent 循环何时停止?
答:满足可验证成功条件、遇到不可恢复错误、需要人工确认或耗尽预算时停止。不要只依赖模型输出 finish,因为模型可能在证据不足时提前结束。

问:提示词能否替代权限系统?
答:不能。提示词只能影响模型行为,无法构成安全边界。权限、资源范围、参数限制和审计必须在模型之外实现。

十四、总结

可靠的 LLM 工具调用系统,本质上不是给模型更多权限,而是把不确定的语言决策包裹在确定性的工程控制中。实践顺序可以概括为:先收窄工具契约,再分离规划与执行;使用 Schema 和业务策略双重校验;对副作用操作加入幂等和确认;按错误类别决定修复、重试或终止;通过事件日志和状态存储支持恢复;最后使用失败注入和可回放评测持续验证。

当系统能够回答“执行了什么、为什么允许、失败后如何恢复、是否产生重复副作用、最终结果如何验证”这些问题时,工具调用 Agent 才从演示功能进入可维护、可审计的工程系统阶段。
 
顶部