工具调用经验怎样沉淀成规则在 Function Calling 或 Agent 工具调用中模型输出可能不符合参数契约也可能选择了不该自主执行的动作。Prompt 可以帮助说明意图但权限、校验和审计必须由程序控制。工具治理应把模型建议与实际执行分开先解析参数再按工具风险级别判断授权条件最后保留足够的调用记录。这样即使模型行为变化执行边界仍由明确规则约束。1. 用可复现输入检查工具调用边界可以用一组受控输入检验治理规则缺字段的参数、超出范围的参数、未知工具名、重复调用以及未获批准的写操作。检查结果应告诉调用方是哪条规则拒绝了请求同时在审计记录里保留工具名、输入摘要、决定和结果。这样讨论的是可验证的边界不是编造一段线上事故来说明风险。2. 工具调用治理与决策沉淀架构可将工具调用组织为分层的检查流程这套架构核心保证了三点类型安全参数先通过 Pydantic 或 JSON Schema 校验校验失败时返回可修正的信息。风险隔离高危工具如删除、修改、重启强制需要经过批准低危工具限频执行。可追溯性记录调用标识、必要的输入摘要、结果和耗时。审计存储的防篡改能力取决于具体存储与权限配置不能仅凭日志名称保证。3. Python 生产级安全 Function Calling 治理组件实现下面是一个完整的 Python 生产级工具调用治理与审计组件。代码实现了基于 Pydantic 的 Schema 校验、高危工具拦截与全局审计日志记录import json import time import logging from typing import Dict, Any, Callable, Optional, List from pydantic import BaseModel, Field, ValidationError logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s) # 1. 工具参数 Schema 契约定义 class QueryLogArgs(BaseModel): service_name: str Field(..., description服务名称) lines: int Field(default50, ge1, le500, description读取行数 1-500) class RestartServiceArgs(BaseModel): service_name: str Field(..., description服务名称) force: bool Field(defaultFalse, description是否强制重启) reason: str Field(..., min_length5, description重启原因说明不得少于5字) # 2. 工具定义与风险等级注册 class ToolDefinition: def __init__(self, name: str, schema: type[BaseModel], is_high_risk: bool, func: Callable): self.name name self.schema schema self.is_high_risk is_high_risk self.func func # 3. 生产级安全 Tool Invoker 调度器 class GovernanceToolInvoker: def __init__(self): self.tools: Dict[str, ToolDefinition] {} self.audit_logs: List[Dict[str, Any]] [] self.call_counters: Dict[str, int] {} # 频次限制 def register_tool(self, name: str, schema: type[BaseModel], is_high_risk: bool, func: Callable): self.tools[name] ToolDefinition(name, schema, is_high_risk, func) def invoke(self, tool_name: str, raw_args_json: str, require_human_approved: bool False) - str: start_time time.time() tool self.tools.get(tool_name) # 检查 1工具是否存在 if not tool: err_msg f错误: 工具 {tool_name} 不存在 self._log_audit(tool_name, raw_args_json, False, err_msg, start_time) return json.dumps({status: error, message: err_msg}) # 检查 2频次与死循环防护 (同一工具单任务不得超过 3 次调用) current_calls self.call_counters.get(tool_name, 0) if current_calls 3: err_msg f硬拦截: 工具 {tool_name} 在当前会话中超过最大允许调用次数 (3次)触发熔断 self._log_audit(tool_name, raw_args_json, False, err_msg, start_time) return json.dumps({status: error, message: err_msg}) # 检查 3JSON 解析与 Pydantic 强 Schema 校验 try: parsed_json json.loads(raw_args_json) validated_args tool.schema(**parsed_json) except (json.JSONDecodeError, ValidationError) as e: err_msg f Schema 校验失败: {str(e)} logging.error(f[{tool_name}] {err_msg}) self._log_audit(tool_name, raw_args_json, False, err_msg, start_time) return json.dumps({status: error, message: f参数格式不符: {err_msg}}) # 检查 4高危工具拦截与人工审批查验 if tool.is_high_risk and not require_human_approved: err_msg f安全拦截: 工具 {tool_name} 属于高风险写操作需要人工确认审批后方可执行。 logging.warning(f[{tool_name}] {err_msg}) self._log_audit(tool_name, raw_args_json, False, err_msg, start_time) return json.dumps({status: need_approval, message: err_msg}) # 执行真实工具逻辑 try: self.call_counters[tool_name] current_calls 1 result tool.func(validated_args) self._log_audit(tool_name, raw_args_json, True, result, start_time) return json.dumps({status: success, data: result}) except Exception as ex: err_msg f工具执行抛出运行时异常: {str(ex)} self._log_audit(tool_name, raw_args_json, False, err_msg, start_time) return json.dumps({status: error, message: err_msg}) def _log_audit(self, tool_name: str, args: str, success: bool, output: Any, start_time: float): latency_ms (time.time() - start_time) * 1000 audit_entry { timestamp: time.strftime(%Y-%m-%d %H:%M:%S), tool_name: tool_name, raw_args: args, success: success, output: str(output)[:200], latency_ms: round(latency_ms, 2) } self.audit_logs.append(audit_entry) # 4. 模拟真实工具与测试 def mock_query_log(args: QueryLogArgs) - Dict[str, Any]: return {logs: [f[{args.service_name}] 内存占用 85%, f[{args.service_name}] HTTP 200 OK]} def mock_restart_service(args: RestartServiceArgs) - Dict[str, Any]: return {message: f服务 {args.service_name} 已成功重启原因: {args.reason}} if __name__ __main__: invoker GovernanceToolInvoker() # 注册工具 invoker.register_tool(query_log, QueryLogArgs, is_high_riskFalse, funcmock_query_log) invoker.register_tool(restart_service, RestartServiceArgs, is_high_riskTrue, funcmock_restart_service) print( 测试 1: 正常低风险查询工具调用 ) res1 invoker.invoke(query_log, {service_name: order_api, lines: 20}) print(f返回结果: {res1}\n) print( 测试 2: 畸形参数触发表单 Schema 校验拦截 ) res2 invoker.invoke(query_log, {service_name: order_api, lines: 999}) # 超出 lines500 规则 print(f返回结果: {res2}\n) print( 测试 3: 未经批准调用高危重启工具触发安全拦截 ) res3 invoker.invoke(restart_service, {service_name: order_api, reason: 线上内存泄漏紧急排查}) print(f返回结果: {res3}\n) print( 测试 4: 带有人工批准标志的高危工具正常执行 ) res4 invoker.invoke(restart_service, {service_name: order_api, reason: 线上内存泄漏紧急排查}, require_human_approvedTrue) print(f返回结果: {res4}\n) print( 查看沉淀的 Tool Audit Trail 审计日志 ) print(json.dumps(invoker.audit_logs, indent2, ensure_asciiFalse))运行输出逻辑展现了防线如何精准拦截非法的 Function Calling 参数与未经授权的高危动作 测试 1: 正常低风险查询工具调用 返回结果: {status: success, data: {logs: [[order_api] 内存占用 85%, [order_api] HTTP 200 OK]}} 测试 2: 畸形参数触发表单 Schema 校验拦截 2026-08-26 10:50:00 [ERROR] [query_log] Schema 校验失败: 1 validation error for QueryLogArgs... 返回结果: {status: error, message: 参数格式不符: Schema 校验失败...} 测试 3: 未经批准调用高危重启工具触发安全拦截 2026-08-26 10:50:00 [WARNING] [restart_service] 安全拦截: 工具 restart_service 属于高风险写操作需要人工确认审批后方可执行。 返回结果: {status: need_approval, message: 安全拦截: 工具 restart_service 属于高风险写操作...} 测试 4: 带有人工批准标志的高危工具正常执行 返回结果: {status: success, data: {message: 服务 order_api 已成功重启原因: 线上内存泄漏紧急排查}}4. 将工具排障经验沉淀为工程规则的 3 项要求要把工具调用规则留在工程里可先落实以下三项要求第一所有工具的声明必须绑定强类型 Pydantic/JSON Schema 约束。禁止向 LLM 暴露参数类型为任意字符串或未限制范围数值的裸 Tool。第二读写工具分离与权限分级。查询类只读工具Read-only由 Agent 自主调用并设置频次上限变更类高危工具Write/Mutate必须经过人工审批或携带特权签名。第三调用审计留痕。记录tool_name、经过脱敏的输入摘要、响应结果与耗时并按访问控制要求写入日志或审计系统。Prometheus 更适合保存聚合指标原始调用记录通常需要单独的日志或审计存储。规则进入仓库后最好用几组失败输入做回归缺字段、重复调用、超时返回和越权工具名都应能得到稳定结果。规则不是越多越好一条连续误伤正常请求的限制也要记录原因并及时收窄。还应区分“建议执行”和“实际执行”。模型可以提出重启、删除或导出的建议但执行层需要再次读取当前授权、目标环境和变更单状态。对读操作也不要完全放开查询范围、返回行数和数据脱敏同样是工具契约的一部分。把这些条件写在工具注册处比散落在提示词里更容易审查和测试。审计记录的用途是复查决策不是无边界地保存原始数据。记录字段时应优先保存调用标识、版本、规则命中情况和可脱敏的摘要需要原文排查时再通过受控权限关联回去。这样既能定位规则问题也不会把治理系统变成新的数据风险点。