Tool Calling从模型协议到安全执行完整原理

Tool Calling 让模型根据自然语言选择受控业务能力并生成结构化参数，例如查询采集状态、搜索知识库、创建工单或生成报表。模型并没有直接执行 Java/Python 方法，也没有自动获得数据库权限；它只生成“建议调用哪个工具、参数是什么”的协议消息，真正执行由应用后端完成。

生产系统必须把下面三件事分开：

模型提出Tool Call
≠ 后端授权执行
≠ 业务系统已经成功提交

只会给模型传一个 JSON Schema 还不够。写工具还要处理权限、参数、幂等、不可变确认计划、超时结果未知、对账、审计和恢复。本章将完整解释这些过程。

一、学习目标

学完后，你应该能够：

• 区分 Tool Calling、Function Calling、插件、工作流和 Agent。
• 解释工具 Schema 怎样进入模型上下文，模型怎样返回 Tool Call。
• 解释 tool_call_id 为什么用于关联调用请求和工具结果。
• 设计名称、描述、参数、返回值和错误契约清晰的工具。
• 实现 JSON 语法、Schema、业务、权限四层校验。
• 说明用户身份、tenantId 和数据范围为什么不能来自模型参数。
• 区分查询、低风险写入、高风险写入和禁止开放的工具。
• 设计“计划—确认—执行”流程，并防止确认后参数变化。
• 使用幂等键、唯一约束和执行记录避免重复副作用。
• 正确处理写操作超时后的 UNKNOWN 状态。
• 区分可重试、不可重试、需澄清、需审批和需对账错误。
• 处理并行 Tool Call、依赖顺序、Deadline 和取消。
• 裁剪、脱敏并隔离不可信工具结果。
• 运行本页状态机 Demo，观察 UNKNOWN 如何对账恢复为 SUCCEEDED。

二、先区分五个概念

概念	核心含义	谁控制执行顺序
普通模型调用	输入消息，输出自然语言或结构化文本	应用单次调用
Tool/Function Calling	模型生成工具名和结构化参数	模型建议，应用授权执行
固定工作流	步骤和分支由代码/BPMN预定义	后端流程引擎
Agent	模型根据State和Observation动态选择下一步	模型建议，应用状态机约束
插件/远程能力	工具定义和实现的打包或接入形态	仍由宿主应用执行与授权

Function Calling 常是某些 Provider 对 Tool Calling 的早期或具体命名。协议字段、并行调用、强制工具、流式事件和错误格式因 Provider/模型而异，必须通过适配层和契约测试确认，不能假设完全一致。

三、模型到底看到了什么

应用请求通常包含：

messages
+ tools数组
+ tool_choice或同类策略
+ 模型参数
+ 响应格式

工具定义方向：

{
  "type": "function",
  "function": {
    "name": "get_collect_status",
    "description": "查询当前租户内指定数据资产的采集状态。只查询状态，不创建任务。",
    "parameters": {
      "type": "object",
      "additionalProperties": false,
      "required": ["assetId"],
      "properties": {
        "assetId": {
          "type": "string",
          "pattern": "^asset-[A-Za-z0-9_-]{1,50}$",
          "description": "资产编号，例如asset-lis-result"
        }
      }
    }
  }
}

模型读取工具名称、描述和参数 Schema 后，可能输出自然语言，也可能输出一个或多个 Tool Call。工具定义会占用上下文 Token；把数百个工具全部暴露给每次请求会增加成本、选择混淆和攻击面。

四、一次Tool Calling完整协议链

flowchart TD
    A["用户请求"] --> B["认证、限流与场景识别"]
    B --> C["按场景和权限选择最小工具集合"]
    C --> D["发送消息、工具Schema和选择策略"]
    D --> E["模型生成自然语言或Tool Call"]
    E --> F["应用解析tool_call_id、名称和参数"]
    F --> G["语法、Schema、业务与权限校验"]
    G --> H{"是否需要确认或审批"}
    H -- "是" --> I["生成不可变计划并等待确认"]
    H -- "否" --> J["创建执行记录与幂等键"]
    I --> J
    J --> K["后端调用真实业务服务"]
    K --> L["持久化成功、失败或UNKNOWN"]
    L --> M["裁剪、脱敏并构造Tool Result"]
    M --> N["使用原tool_call_id回传模型"]
    N --> O{"是否还需要下一次调用"}
    O -- "是" --> E
    O -- "否" --> P["校验最终回答并返回用户"]

4.1 第一轮模型响应

逻辑示例：

{
  "toolCalls": [
    {
      "id": "call_01HXYZ",
      "name": "get_collect_status",
      "arguments": "{\"assetId\":\"asset-lis-result\"}"
    }
  ]
}

有些 SDK 已将 arguments 解析为对象，有些仍是 JSON 字符串。无论 SDK 是否自动解析，都要校验。

4.2 tool_call_id有什么用

一次模型响应可能包含多个 Tool Call。应用执行后，要把每个结果与原调用 ID 关联：

{
  "role": "tool",
  "tool_call_id": "call_01HXYZ",
  "content": "{\"status\":\"FAILED\",\"reasonCode\":\"CONNECT_TIMEOUT\"}"
}

模型据此知道结果属于哪次调用。不能把多个结果拼成一段无身份文本，否则并行调用时容易错配。

4.3 什么时候结束

模型收到 Tool Result 后可能：

• 直接生成最终回答。
• 请求另一个工具。
• 修正参数后再次调用。
• 表示需要用户澄清。

应用必须限制最大工具次数、总 Deadline、Token、金额和同类错误次数，不能让模型无限循环。

五、Tool Choice有哪些方向

Provider 常提供类似策略，字段名和能力以实际接口为准：

策略方向	含义	风险
auto	模型选择回答还是调用工具	可能该用却不用，也可能误用
none	禁止调用工具	适合纯解释或降级
required	必须调用某个工具方向	信息不足时可能硬凑参数
specific tool	强制指定工具	应用已通过规则确定工具时使用

实时事实场景可以由后端路由强制进入查询工具，而不是只用 Prompt 写“请务必调用”。高风险写操作不能因为 required 就跳过审批。

六、工具注册表为什么需要

生产系统应有服务端 Tool Registry：

toolName
schemaVersion
descriptionVersion
riskLevel
requiredPermissions
timeoutBudget
retryPolicy
idempotencyPolicy
confirmationPolicy
resultProjection
owner
enabledTenants

请求到达后，根据场景、用户权限、租户开关、模型能力和风险动态选择工具集合。不要让前端或用户传任意工具名，也不要把禁用工具的 Schema 发给模型。

6.1 名称必须稳定且具体

推荐：

get_order_status
get_collect_status
create_recovery_ticket

不推荐：

query
execute
do_action

6.2 描述要写适用和不适用边界

查询当前租户内数据资产的最近采集状态。
不查询患者明细，不创建补采任务，不修改调度配置。

这能减少模型误选，但不能替代后端权限。

6.3 Schema版本为什么要记录

模型输出与旧 Schema 一致，但后端实现已经变更时可能解析失败或语义漂移。审计中要记录模型实际看到的 Tool Schema Version，而不是只记录当前代码版本。

七、参数Schema怎样设计

7.1 使用对象和明确required

顶层通常为 object，列出必填字段和 additionalProperties: false 方向，减少模型塞入未定义参数。

7.2 使用业务类型，不要自由SQL

错误：

{"sql": "select * from ..."}

改进：

{
  "assetId": "asset-lis-result",
  "from": "2026-07-16T00:00:00+08:00",
  "to": "2026-07-16T23:59:59+08:00"
}

后端使用参数化查询和固定数据范围实现。

7.3 使用enum、范围和格式

{
  "priority": {
    "type": "string",
    "enum": ["LOW", "MEDIUM", "HIGH"]
  },
  "limit": {
    "type": "integer",
    "minimum": 1,
    "maximum": 20
  }
}

Schema 能改善结构正确率，但业务归属、状态机和权限仍需后端验证。

7.4 避免歧义参数

date 不明确是自然日、UTC 还是用户时区。应使用带时区的时间或明确业务日期与 timezone。金额使用最小货币单位整数或后端 Decimal/BigDecimal，不使用浮点金额。

八、四层校验不能少

flowchart TD
    A["模型arguments"] --> B{"JSON语法可解析"}
    B -- "否" --> C["格式错误：有限修复或澄清"]
    B -- "是" --> D{"符合Schema"}
    D -- "否" --> E["字段、类型、枚举错误"]
    D -- "是" --> F{"符合业务规则"}
    F -- "否" --> G["状态、范围或资源不存在"]
    F -- "是" --> H{"通过身份和数据权限"}
    H -- "否" --> I["拒绝并审计"]
    H -- "是" --> J["进入风险和执行流程"]

例子：

• JSON 合法：{"orderNo":"SO-1"}。
• Schema 合法：orderNo 是字符串。
• 业务合法：订单存在且处于可查询状态。
• 权限合法：订单属于当前 tenantId，用户有查看权限。

模型不能通过在参数中写 tenantId=admin 获得权限。

九、身份和tenantId必须来自后端

错误工具：

{
  "name": "get_patient",
  "parameters": {
    "userId": {"type": "string"},
    "tenantId": {"type": "string"},
    "patientId": {"type": "string"}
  }
}

用户可诱导模型传其他租户或管理员 ID。

正确方向：

模型参数：业务对象自然键或受控查询条件
后端上下文：currentUser、tenantId、roles、dataScope、traceId

Repository 查询必须带服务端 tenantId 和数据范围条件，不能先查全局对象再仅靠模型决定是否展示。

十、工具风险分级

等级	示例	主要控制
L0公开只读	天气、公开术语	参数、限流、日志
L1内部只读	订单、资产、日志	认证、租户、字段裁剪、审计
L2可逆写入	创建草稿、工单	权限、幂等、确认、频控
L3高风险写入	退款、停用、批量通知	不可变计划、强确认/审批、双人复核方向
L4禁止开放	任意SQL、任意Shell、修改权限根角色	不向模型暴露

风险不仅取决于工具名，还取决于参数范围。例如“导出 10 条自己可见数据”和“导出全院数据”不是同一风险。

十一、查询工具怎样设计

查询工具没有业务副作用，但仍存在：

• 越权读取。
• 大结果集拖垮数据库。
• 敏感字段泄露。
• 慢 SQL。
• 工具结果 Prompt Injection。
• 查询成本和频率滥用。

治理：

• 固定查询模板和参数化 SQL。
• 服务端 ACL。
• Limit、时间范围和分页上限。
• 只返回必要字段。
• 手机号、身份证、Token、内部成本脱敏或删除。
• 超时、只读事务和数据库资源隔离。
• 结果标记为不可信数据，不执行其中命令。

十二、写工具为什么需要计划—确认—执行

直接执行：

用户自然语言
→ 模型猜参数
→ 立即产生副作用

风险很高。推荐：

flowchart TD
    A["模型提出写Tool Call"] --> B["后端校验并生成不可变Plan"]
    B --> C["显示对象、参数、影响和风险"]
    C --> D{"用户或审批人确认"}
    D -- "否" --> E["取消并记录"]
    D -- "是" --> F["验证Plan未过期且未变化"]
    F --> G["创建幂等执行记录"]
    G --> H["执行真实业务写入"]
    H --> I["返回权威业务结果"]

12.1 不可变Plan包含什么

planId
toolName和schemaVersion
规范化参数
目标资源版本
发起用户和租户
风险等级
影响摘要
planHash
expiresAt

确认必须绑定 planHash。用户确认后如果模型修改金额、对象或范围，Hash 变化，旧确认失效，必须重新展示。

12.2 确认不是一句聊天消息

“好的”“确认”可能被多轮上下文误关联。应用应使用结构化 confirmationToken 或审批单，绑定：

• planId 与 planHash。
• 用户和租户。
• 过期时间。
• 一次性随机数。
• 审批策略。

12.3 资源版本

确认时订单状态可退款，执行时可能已退款。Plan 可携带 expectedVersion，业务写入使用条件更新或状态机再次校验，防止确认后的并发变化。

十三、幂等不只是生成一个UUID

幂等目标：同一个业务意图被重复提交时，只产生一次副作用，并返回同一权威结果。

13.1 幂等键由谁生成

应用根据稳定业务意图生成或持久化，不应让模型随每次重试随机生成。可包含：

tenantId + toolName + planId或businessRequestId

13.2 服务端如何保证

工具执行表：

idempotencyKey UNIQUE
toolName
planHash
status
businessId
resultSnapshot
attempts
lastError
version

数据库唯一约束是并发兜底。仅“先查再插”会有竞态：两个请求同时查不到，然后都执行。

13.3 幂等记录和业务写入的事务边界

若工具与业务表在同一数据库，可在一个本地事务中写业务结果和成功记录。若跨系统，需要下游也支持同一幂等键和结果查询；调用方记录不能单独阻止下游重复提交。

13.4 相同Key但参数不同

必须拒绝并告警。幂等键应绑定 planHash；否则攻击者或代码错误可能用旧 Key 提交新金额。

十四、为什么写工具超时后是UNKNOWN

超时只证明调用方在 Deadline 前没收到响应，不证明下游未执行。

flowchart TD
    A["应用发送创建工单请求"] --> B["下游提交数据库事务"]
    B --> C["下游返回响应"]
    C --> D["网络响应丢失或超过Deadline"]
    D --> E["调用方观察到Timeout"]
    E --> F["真实业务可能已经成功"]

如果收到 Timeout 后直接换新幂等键重试，可能创建两张工单或退款两次。

14.1 正确状态

PREPARED
→ RUNNING
→ SUCCEEDED
→ FAILED_FINAL
→ UNKNOWN

UNKNOWN 表示提交结果待查证，不是普通失败。

14.2 对账恢复

1. 使用原 idempotencyKey 查询下游。
2. 若查到业务结果，更新为 SUCCEEDED 并返回原 businessId。
3. 若下游明确证明未执行，才按原 Key 重试。
4. 若无法证明，继续 UNKNOWN、后台对账或转人工。
5. 绝不能仅因“查询暂时没找到”立即断言没执行；需要考虑副本延迟和最终一致性。

十五、错误分类和重试策略

类别	示例	自动重试
参数可修复	枚举错误、缺字段	可让模型有限修正，不执行副作用
需要澄清	同名资产多个	请求用户选择
权限拒绝	无创建权限	不重试，审计
业务冲突	状态已关闭、版本冲突	重新读取状态，通常重新确认
瞬时读取失败	连接重置、部分5xx	白名单内退避重试
写入超时	响应未知	标记UNKNOWN，先按Key查证
永久失败	非法业务状态	FAILED_FINAL
风控拦截	频率、金额或批量过大	审批或拒绝

重试使用剩余 Deadline，指数退避加抖动，并限制次数。每一层重新给予完整超时会让总请求远超用户 Deadline。

十六、并行Tool Call怎么处理

模型可能一次返回多个调用：

查采集状态
查最近告警
查资产负责人

16.1 只有独立只读工具才适合并行

若 B 依赖 A 的结果，必须串行。两个写工具即使看似独立，也可能共享库存、额度或状态，不能仅因为模型并行返回就并行执行。

16.2 每个调用独立状态

为每个 tool_call_id 记录：

• 工具名和 Schema Version。
• 参数摘要。
• Deadline。
• 状态和尝试次数。
• 权限结果。
• 结果或错误。

返回模型时按原 ID 关联，不能因完成顺序不同错配。

16.3 部分成功

查询 A 成功、查询 B 超时时，应用可以把两个独立结果分别回传并明确 B 不可用；写操作部分成功则必须依赖业务补偿和状态机，不能让模型用自然语言声称“已全部完成”。

十七、工具结果为什么也不可信

工具可能返回：

• 用户可控备注。
• 网页内容。
• 工单描述。
• 文档中的 Prompt Injection。
• 下游异常堆栈和内部地址。
• 数据库中不应展示的敏感字段。

因此 Tool Result 要经过：

业务结果
→ 字段允许列表
→ 数据范围复核
→ 脱敏
→ 长度与条数限制
→ 错误归一化
→ 标记为不可信数据
→ 发送模型

不要把完整数据库 Entity、HTTP Header、SQL、Token 或异常堆栈直接放入 Prompt。

十八、工具返回契约怎样设计

推荐返回稳定机器字段：

{
  "status": "FAILED",
  "reasonCode": "CONNECT_TIMEOUT",
  "occurredAt": "2026-07-16T09:30:00+08:00",
  "assetId": "asset-lis-result",
  "evidenceIds": ["log-801", "alarm-992"],
  "canRetry": true
}

而不是只返回：

好像连不上，你再试试。

稳定错误码便于应用判断流程；自然语言解释可由模型生成，但不能替代权威状态。

18.1 错误结果也要结构化

{
  "status": "ERROR",
  "errorType": "PERMISSION_DENIED",
  "retryable": false,
  "userMessage": "没有查看该资产的权限"
}

不要把内部堆栈给模型，也不要让模型看到权限失败后尝试“换一个工具绕过”。

十九、Deadline、取消和资源隔离

一次请求总 Deadline 假设 8 秒，可分：

认证与路由 300ms
模型第一次调用 2500ms
工具 2000ms
模型总结 2200ms
校验和网络余量 1000ms

下游收到剩余时间，不是重新获得完整 8 秒。

查询工具可以在用户断开后传播取消；写工具取消只表示“不再等待”，不能假设下游事务回滚。还要使用：

• 每工具并发 Bulkhead。
• 数据库连接池隔离。
• 租户配额。
• 单次和累计工具次数。
• 熔断与降级。
• 大导出转异步任务。

二十、同步、异步和长任务

短查询适合同步。批量报表、全库扫描、文件处理和长时间补采应创建异步任务：

模型提出create_report_task
→ 后端校验与确认
→ 创建taskId
→ 返回QUEUED
→ Worker执行并保存Checkpoint
→ 用户通过get_task_status查询
→ 完成后返回受控下载地址

模型不能在聊天上下文里“等待十分钟”。异步任务需要状态机、Worker Lease、Heartbeat、重试、取消、过期和幂等。

二十一、可运行Demo：确认、幂等与UNKNOWN恢复

下面只使用 Python 标准库，模拟一个“创建补采工单”写工具：

• 模型只提供资产和原因。
• tenantId、userId、权限来自后端 Principal。
• Plan 使用规范化参数和 SHA-256 Hash。
• Confirmation Token 用 HMAC 绑定用户、Plan 和过期时间。
• 第一次执行模拟“业务已提交，但响应超时”。
• 执行记录进入 UNKNOWN。
• 按同一幂等键对账后恢复 SUCCEEDED。
• 再次执行返回同一工单，不重复创建。

from __future__ import annotations

import hashlib
import hmac
import json
import time
from dataclasses import dataclass
from enum import Enum
from typing import Any


class Status(str, Enum):
    PREPARED = "PREPARED"
    RUNNING = "RUNNING"
    SUCCEEDED = "SUCCEEDED"
    FAILED_FINAL = "FAILED_FINAL"
    UNKNOWN = "UNKNOWN"


@dataclass(frozen=True)
class Principal:
    user_id: str
    tenant_id: str
    permissions: frozenset[str]


@dataclass(frozen=True)
class Plan:
    plan_id: str
    tool_name: str
    schema_version: str
    arguments: dict[str, Any]
    principal_user_id: str
    tenant_id: str
    expires_at: int
    plan_hash: str


@dataclass
class ExecutionRecord:
    idempotency_key: str
    plan_hash: str
    status: Status
    business_result: dict[str, Any] | None = None
    attempts: int = 0


def canonical_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def create_plan(
    principal: Principal,
    plan_id: str,
    arguments: dict[str, Any],
    now: int,
) -> Plan:
    if "ticket:create" not in principal.permissions:
        raise PermissionError("没有创建工单权限")
    allowed_keys = {"assetId", "reason", "priority"}
    if set(arguments) != allowed_keys:
        raise ValueError("参数字段不符合Schema")
    if not str(arguments["assetId"]).startswith("asset-"):
        raise ValueError("assetId格式错误")
    if len(str(arguments["reason"]).strip()) < 5:
        raise ValueError("reason过短")
    if arguments["priority"] not in {"LOW", "MEDIUM", "HIGH"}:
        raise ValueError("priority枚举错误")

    payload = {
        "planId": plan_id,
        "toolName": "create_recovery_ticket",
        "schemaVersion": "v2",
        "arguments": arguments,
        "userId": principal.user_id,
        "tenantId": principal.tenant_id,
        "expiresAt": now + 300,
    }
    plan_hash = hashlib.sha256(canonical_json(payload).encode("utf-8")).hexdigest()
    return Plan(
        plan_id=plan_id,
        tool_name=payload["toolName"],
        schema_version=payload["schemaVersion"],
        arguments=dict(arguments),
        principal_user_id=principal.user_id,
        tenant_id=principal.tenant_id,
        expires_at=payload["expiresAt"],
        plan_hash=plan_hash,
    )


def confirmation_token(plan: Plan, secret: bytes) -> str:
    message = f"{plan.plan_hash}|{plan.principal_user_id}|{plan.expires_at}"
    return hmac.new(secret, message.encode("utf-8"), hashlib.sha256).hexdigest()


class TicketSystem:
    def __init__(self) -> None:
        self.results_by_key: dict[str, dict[str, Any]] = {}
        self.created_count = 0

    def create(
        self,
        idempotency_key: str,
        tenant_id: str,
        arguments: dict[str, Any],
        simulate_timeout_after_commit: bool,
    ) -> dict[str, Any]:
        existing = self.results_by_key.get(idempotency_key)
        if existing is not None:
            return existing

        self.created_count += 1
        result = {
            "ticketNo": f"TK-{self.created_count:04d}",
            "tenantId": tenant_id,
            "assetId": arguments["assetId"],
            "status": "CREATED",
        }
        # 模拟下游事务已经提交，并按幂等键保存了权威结果。
        self.results_by_key[idempotency_key] = result
        if simulate_timeout_after_commit:
            raise TimeoutError("响应丢失，但业务可能已经提交")
        return result

    def find_by_idempotency_key(self, key: str) -> dict[str, Any] | None:
        return self.results_by_key.get(key)


class SafeExecutor:
    def __init__(self, secret: bytes, ticket_system: TicketSystem) -> None:
        self.secret = secret
        self.ticket_system = ticket_system
        self.records: dict[str, ExecutionRecord] = {}

    def execute(
        self,
        principal: Principal,
        plan: Plan,
        token: str,
        now: int,
        simulate_timeout_after_commit: bool = False,
    ) -> ExecutionRecord:
        if principal.user_id != plan.principal_user_id or principal.tenant_id != plan.tenant_id:
            raise PermissionError("确认人与计划主体不一致")
        if now > plan.expires_at:
            raise ValueError("确认计划已过期")
        expected = confirmation_token(plan, self.secret)
        if not hmac.compare_digest(token, expected):
            raise PermissionError("确认Token无效或计划已变化")
        if "ticket:create" not in principal.permissions:
            raise PermissionError("执行时权限已失效")

        key = hashlib.sha256(
            f"{plan.tenant_id}|{plan.tool_name}|{plan.plan_id}".encode("utf-8")
        ).hexdigest()
        existing = self.records.get(key)
        if existing is not None:
            if existing.plan_hash != plan.plan_hash:
                raise ValueError("同一幂等键绑定了不同计划")
            if existing.status in {Status.SUCCEEDED, Status.UNKNOWN}:
                return existing

        record = existing or ExecutionRecord(key, plan.plan_hash, Status.PREPARED)
        self.records[key] = record
        record.status = Status.RUNNING
        record.attempts += 1
        try:
            result = self.ticket_system.create(
                idempotency_key=key,
                tenant_id=principal.tenant_id,
                arguments=plan.arguments,
                simulate_timeout_after_commit=simulate_timeout_after_commit,
            )
            record.business_result = result
            record.status = Status.SUCCEEDED
        except TimeoutError:
            record.status = Status.UNKNOWN
        return record

    def reconcile(self, record: ExecutionRecord) -> ExecutionRecord:
        if record.status is not Status.UNKNOWN:
            return record
        result = self.ticket_system.find_by_idempotency_key(record.idempotency_key)
        if result is not None:
            record.business_result = result
            record.status = Status.SUCCEEDED
        return record


if __name__ == "__main__":
    now = int(time.time())
    principal = Principal(
        user_id="u-1001",
        tenant_id="hospital-a",
        permissions=frozenset({"ticket:create"}),
    )
    plan = create_plan(
        principal=principal,
        plan_id="plan-20260716-001",
        arguments={
            "assetId": "asset-lis-result",
            "reason": "采集连接连续超时",
            "priority": "HIGH",
        },
        now=now,
    )
    secret = b"demo-secret-replace-with-kms-managed-key"
    token = confirmation_token(plan, secret)
    ticket_system = TicketSystem()
    executor = SafeExecutor(secret, ticket_system)

    first = executor.execute(
        principal,
        plan,
        token,
        now,
        simulate_timeout_after_commit=True,
    )
    assert first.status is Status.UNKNOWN
    assert ticket_system.created_count == 1

    reconciled = executor.reconcile(first)
    assert reconciled.status is Status.SUCCEEDED
    assert reconciled.business_result is not None
    assert reconciled.business_result["ticketNo"] == "TK-0001"

    retry = executor.execute(principal, plan, token, now)
    assert retry.business_result == reconciled.business_result
    assert ticket_system.created_count == 1

    print("after timeout:", Status.UNKNOWN.value)
    print("after reconcile:", reconciled.status.value, reconciled.business_result)
    print("created count after retry:", ticket_system.created_count)

Demo 中的 Secret 仅为教学常量，生产应由 KMS/Secret 管理并轮换；执行记录和业务幂等键必须落数据库，不能只放进单进程字典。

二十二、商业场景：排查采集失败并创建补采工单

用户：

帮我查看LIS检验结果资产为什么停止采集；如果是连接超时，创建一张高优先级补采工单。

完整链路：

flowchart TD
    A["用户请求"] --> B["认证并确认可见资产范围"]
    B --> C["模型提出get_collect_status"]
    C --> D["后端用当前tenantId和asset ACL查询"]
    D --> E["工具返回CONNECT_TIMEOUT与证据ID"]
    E --> F["模型提出create_recovery_ticket"]
    F --> G["后端生成不可变Plan"]
    G --> H["用户确认资产、原因、优先级和影响"]
    H --> I["验证Confirmation Token和当前权限"]
    I --> J["按幂等键创建执行记录"]
    J --> K["工单系统提交"]
    K --> L{"调用方是否收到确定响应"}
    L -- "是" --> M["记录SUCCEEDED和ticketNo"]
    L -- "否" --> N["记录UNKNOWN并按Key对账"]
    N --> M
    M --> O["模型只总结权威状态和证据"]

22.1 查询工具

get_collect_status：

• 只读。
• tenantId 来自认证上下文。
• assetId 必须在用户数据范围内。
• 日志仅返回错误码、时间和证据 ID。
• 不返回数据库密码、连接串和完整异常栈。

22.2 写工具

create_recovery_ticket：

• L2/L3 风险，取决于影响范围。
• 生成 Plan，不立即执行。
• 确认绑定 planHash。
• 工单系统接受幂等键并可按 Key 查询。
• Timeout 进入 UNKNOWN。
• 最终回答只在查到 ticketNo 后声称已创建。

22.3 如果模型错误判断原因

写工具业务规则应再次验证 reasonCode=CONNECT_TIMEOUT 是否仍是最新权威状态，或者 Plan 明确由人工确认该判断。模型自然语言推断不能替代采集系统状态。

二十三、审计日志和Trace

至少记录：

requestId、conversationId、agentTaskId
modelRoute、promptVersion
toolCallId、toolName、schemaVersion
参数Hash和脱敏摘要
principalUserId、tenantId、权限结果
riskLevel、planId、planHash、confirmationId
idempotencyKey
状态迁移、attempts、Deadline
providerRequestId、downstreamTraceId
businessId、resultCode
耗时和错误类型

不要在普通日志记录密码、Token、患者隐私、完整工具结果和 HMAC Secret。高敏审计存储应限权、加密并设置保留周期。

二十四、工具评估怎样做

评估集不能只测“模型有没有选中工具”，还要拆开：

层	指标
工具选择	正确工具率、不该调用时的误调用率
参数	JSON通过率、Schema通过率、字段准确率
安全	越权阻断率、高风险确认覆盖率
执行	成功率、UNKNOWN率、幂等重复副作用数
恢复	UNKNOWN对账完成率、平均恢复时间
结果	字段裁剪、敏感泄露率、最终陈述一致性
性能	Tool P95/P99、总Deadline、重试成本

样本覆盖：

• 信息完整和缺字段。
• 同名资源需要澄清。
• 无权限资源。
• 直接/间接 Prompt Injection。
• 重复提交。
• 写入提交后响应丢失。
• 下游明确失败。
• 确认后资源版本变化。
• 并行只读与有依赖调用。

二十五、生产故障排查Runbook

25.1 模型一直不用工具

1. 检查场景路由是否把工具 Schema 发给当前模型。
2. 检查模型是否支持该 Tool 协议和当前 Provider 版本。
3. 检查工具名称、描述和 Query/写入边界。
4. 实时事实是否应由后端规则强制 Tool，而不是依赖 auto。
5. 查看 finish reason、原始适配层事件和工具选择评估集。

25.2 模型总选错工具

• 工具名称和描述是否过于相似。
• 是否一次暴露太多工具。
• 工具职责是否重叠。
• Few-shot 是否包含错误示例。
• 场景 Router 是否应先缩小工具集合。
• 高风险工具是否被错误暴露给普通场景。

25.3 参数格式经常错误

• Schema 是否过深、字段过多或相互矛盾。
• 模型是否真正支持当前 JSON Schema 子集。
• 流式 arguments 是否在完整结束前解析。
• max output tokens 是否截断参数。
• Provider 适配是否重复拼接增量片段。
• 记录语法、Schema、业务三类失败，不能都归为模型错。

25.4 出现重复工单或重复通知

这是业务事故：

1. 暂停相关写工具或切只读模式。
2. 按 requestId、toolCallId、planId、idempotencyKey 和 businessId 对账。
3. 检查是否每次重试生成新 Key。
4. 检查下游是否真正以 Key 建唯一约束。
5. 检查 Timeout 是否被标 FAILED 后直接重试。
6. 修复后加入并发、超时后提交和重复确认测试。

25.5 工具超时但业务方说已成功

1. 找到原 idempotencyKey。
2. 将本地执行状态视为 UNKNOWN，不覆盖为 FAILED。
3. 调用下游按 Key 或业务流水查询权威结果。
4. 查到结果后绑定 businessId 并更新 SUCCEEDED。
5. 未查到时考虑副本延迟，按对账策略继续，不更换 Key 盲重试。

25.6 权限拦截突然增多

• 用户角色或租户上下文是否丢失。
• Tool Schema 是否错误要求模型传 tenantId。
• 资源归属数据是否延迟或缓存过期。
• 新版本是否扩大了工具目标范围。
• 区分攻击尝试、正常权限变化和认证链故障。

25.7 最终回答说成功，但工具实际失败

• Tool Result 是否使用稳定 status/errorType。
• Prompt 是否允许模型覆盖权威状态。
• 多个 tool_call_id 是否错配。
• 部分成功是否被模型总结为全部成功。
• 最终输出层应从执行记录渲染 businessId 和状态，而不是只信模型文本。

二十六、常见误区与后果

误区	正确理解
模型调用了Java方法	模型只生成协议，应用映射并执行
JSON Schema保证业务正确	还需业务、权限和状态校验
tenantId让模型填写即可	身份和数据范围必须来自认证上下文
UUID就是幂等	还需稳定意图、持久化记录和唯一约束
Timeout就是失败	写入可能已提交，应进入UNKNOWN并对账
用户说确认就能执行任何参数	确认必须绑定不可变planHash和有效期
Tool Result来自后端所以可信	内容可能含用户文本、注入和敏感字段
并行Tool Call都可并发执行	只对独立、低风险调用并发
模型说已完成就是完成	权威执行记录和业务ID才是事实
工具越多模型越强	过多工具增加Token、混淆和攻击面

二十七、面试标准回答

27.1 Tool Calling完整流程是什么

应用根据场景和权限选择工具 Schema，与用户消息一起发给模型；模型返回带 tool_call_id 的工具名和参数；应用执行 JSON、Schema、业务和权限校验，按风险进行确认和幂等处理，再调用真实业务服务；结果裁剪脱敏后以原调用 ID 回传模型，模型再决定继续调用或生成最终回答。

27.2 模型为什么不能直接执行工具

模型输出是概率生成，可能选错工具、传错参数或受 Prompt Injection 影响。真实执行需要后端身份、租户、权限、状态机、幂等、审批和审计。模型只能提出意图，能否执行由确定性代码和业务系统决定。

27.3 写工具怎样做二次确认

先把模型参数校验成不可变 Plan，包含工具版本、规范化参数、资源版本、用户、租户、风险、过期时间和 planHash；向用户展示影响，确认 Token 绑定用户、planHash 和有效期。确认后任何参数变化都要重新确认，执行时还要重新鉴权和检查资源当前状态。

27.4 Tool Calling为什么需要幂等

模型重试、网络超时、用户重复提交和 Worker 恢复都可能重复执行。应用使用稳定业务意图生成幂等键，执行表以唯一约束绑定 planHash，下游也按相同 Key 去重并可查询原结果；相同 Key 参数不同必须拒绝。

27.5 写工具超时后为什么不能直接重试

超时只表示调用方没及时收到响应，下游事务可能已经提交。应把状态记为 UNKNOWN，使用原幂等键查询权威结果；查到则恢复 SUCCEEDED，明确未执行后才按原 Key 重试，无法判断时继续对账或转人工。

27.6 并行Tool Call怎样保证结果不错配

每个调用使用独立 tool_call_id、状态、Deadline 和审计记录，只并发执行无依赖的低风险调用；完成后结果以原 ID 回传，即使返回顺序不同也能关联。部分成功要逐项表达，写操作不能由模型把部分成功说成全部成功。

更多简洁回答见 AI应用工程化面试题，完整原理以本页为准。

二十八、关联知识点

• Prompt任务编译、安全与发布：理解工具 Schema 和不可信数据边界。
• AI Agent：理解多步骤 State、Action、Observation 与 Checkpoint。
• AI应用架构：学习 Deadline、状态机、Lease、Outbox 和多Region。
• AI安全：学习 Prompt Injection、越权和数据最小化。
• Spring AI Tool Calling：查看 Java 注解、Schema 映射和执行实现。
• 分布式幂等与事务：深入业务唯一约束、消息和补偿。

二十九、学习验收清单

• [ ] 能区分模型Tool Call、后端授权和业务提交。
• [ ] 能画出两轮模型调用与Tool Result回传过程。
• [ ] 能说明tool_call_id为什么不能丢。
• [ ] 能设计名称、描述、参数、返回值和错误契约。
• [ ] 能实现语法、Schema、业务和权限四层校验。
• [ ] 能说明tenantId为什么不能来自模型。
• [ ] 能按风险判断直接查询、确认、审批或禁止开放。
• [ ] 能设计不可变Plan、planHash和confirmationToken。
• [ ] 能设计稳定幂等键、执行表和数据库唯一约束。
• [ ] 能解释Timeout、FAILED和UNKNOWN的区别。
• [ ] 能按原Key对账并恢复权威businessId。
• [ ] 能判断哪些Tool Call可以并行。
• [ ] 能裁剪、脱敏并隔离不可信工具结果。
• [ ] 能根据Trace排查误调用、重复写、越权和假成功。

达到这些标准后，才算掌握生产级 Tool Calling，而不是只会让模型返回一个函数名和 JSON 参数。