Function Calling 工具调用实现指南：把模型提案当作未验证输入

Function Calling 解决的是这样一个问题：让模型不再只输出自然语言，而是输出结构化的「工具名 + 参数」，由应用去执行真实操作。它适合正在把大模型接入业务系统的开发者——尤其是那些要调用订单、支付、消息发送这类有副作用的接口的场景。需要先固定一条原则：模型返回的 tool call 不是命令，而是一份尚未验证的提案。模型确实能返回结构化的工具名和参数，但执行外部 API、做鉴权、处理超时、防止重复执行的主体是应用。JSON 符合 schema，和这次操作可以被执行，是两件不同的事。本文不按某家 SDK 的写法来讲，而是把 Function Calling 当作应用自己拥有的数据契约与状态机来整理。

准备工作

在动手写代码之前，先把下面这些前提确认清楚。

• 一个支持工具调用的模型接口。各家 API 在表达方式上有差异，但职责划分是一致的：模型生成调用，应用执行并把结果回传。具体支持哪些模型、参数名怎么叫，以各家官网当前信息为准。
• 一份应用侧维护的工具注册表。工具定义由应用持有，不由模型动态决定。注册表里至少要有工具名、描述、输入 schema，以及对应的处理函数。
• 一个可信的会话上下文来源。用户 ID、租户 ID、角色、访问范围这些信息必须来自已认证的会话，不能来自模型生成的参数。
• 一个持久化存储。用于保存已完成的操作用于幂等去重，以及保存审批记录。内存字典只够做演示。
• 日志与追踪设施。调用选择、鉴权、执行、结果映射这几个阶段要能分别观测，否则出问题时无法定位是哪一层坏了。

工具定义要尽量小。把查询、改地址、取消订单全塞进一个 manage_order，会让选择边界和授权边界同时变模糊。更合理的拆法是这样的：

凡是应用已经知道的取值，就不要让模型再生成一遍。模型少生成一个字段，参数错误和权限提升的空间就少一分。

操作步骤

第一步：把调用循环固定成六个阶段

使用客户端工具时，通用循环可以拆成六段：

1. 应用把工具定义交给模型。
2. 模型返回普通回答，或者返回一个工具调用候选。
3. 应用解析、校验、授权这个调用。
4. 应用执行工具。
5. 应用把调用 ID 和工具结果回传给模型。
6. 模型基于结果给出最终回答。

用 JSON 表示这个过程中的消息，模型交给应用的是一份提案：

{ "call_id": "call_01", "name": "request_order_cancellation", "arguments": { "order_id": "A-104", "reason": "ordered_by_mistake" } }

应用执行完工具后，回传一个与原始调用对应的结果信封：

{ "call_id": "call_01", "status": "ok", "output": { "request_id": "cancel_827", "state": "accepted" } }

call_id 不是可有可无的附加信息。多个调用并行执行时完成顺序会变，调用与结果的对应关系不能依赖顺序，只能依赖这个 ID。

第二步：把工具调用拆成四类数据

如果直接把模型返回的字典丢给处理函数，就没人说得清哪些内容已经验证过了。实现时至少要区分四类数据：

这里有一条关键约束：不要把 user_id、tenant_id、role、access scope 放进 ProposedCall.arguments。这些值不能让模型决定，而应由应用从已认证的会话中取出，注入到 ValidatedOperation 里。

第三步：把校验拆成三道闸门

不要把工具调用的校验压缩成一个布尔值，要按失败层次分开：

模型的严格模式有助于得到符合 schema 的参数。以 OpenAI 的用法为例，严格 schema 要求设置 additionalProperties: false，并把属性都标为必填。但下面这个调用即使符合 schema，也必须拒绝：

{ "name": "request_order_cancellation", "arguments": { "order_id": "OTHER_USER_ORDER", "reason": "ordered_by_mistake" } }

order_id 是字符串，和当前登录用户拥有这个订单，是两回事。资源所有权要在工具处理函数内部，用可信数据库再确认一次。

第四步：按错误分类决定重试策略

把错误回传给模型时，不要直接把内部异常抛出去。按类别组织信息：

在 authorization_denied 里返回「所有者是 user-789」这类细节，会让错误响应本身变成信息泄露。反过来，把所有错误都压成一个笼统的 error，模型就会反复提交同一个调用。原则是：只结构化地给出修正所需的最小信息。

第五步：让有副作用的工具幂等执行

只读工具重试相对安全，但下单、发送、更新这类操作，超时后重跑就是事故。最危险的情形是「服务端已经成功，只是响应丢了」——从客户端看是失败，于是重试同一个调用，造成重复下单。

下面是一个最小调度器示例，按白名单、校验器、审批、幂等的顺序处理模型提案。这是用于说明职责边界的自写示例，不是任何官方 SDK 代码的转载。

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass
from typing import Any, Callable, Mapping

LOGGER = logging.getLogger(__name__)


class ToolCallError(Exception):
    """工具调用无法安全执行时的基类异常。"""


class AuthorizationError(ToolCallError):
    """用户或审批状态不满足执行条件。"""


@dataclass(frozen=True)
class ProposedCall:
    """模型生成的未验证工具调用。"""

    call_id: str
    name: str
    arguments: Mapping[str, Any]


@dataclass(frozen=True)
class ExecutionContext:
    """由已认证会话构造的可信上下文。"""

    user_id: str
    allowed_tools: frozenset[str]
    approved_operation_ids: frozenset[str]


@dataclass(frozen=True)
class ToolSpec:
    """应用维护的工具定义。"""

    handler: Callable[[Mapping[str, Any], ExecutionContext], Mapping[str, Any]]
    validator: Callable[[Mapping[str, Any]], None]
    has_side_effect: bool


class Dispatcher:
    """只执行已校验的工具，并从说明用缓存中复用重复结果。"""

    def __init__(self, registry: Mapping[str, ToolSpec]) -> None:
        if not registry:
            raise ValueError("registry must not be empty")
        self._registry = dict(registry)
        # 仅用于说明。生产环境的幂等性要落到带事务的持久化存储。
        self._completed: dict[str, Mapping[str, Any]] = {}

    def execute(
        self,
        call: ProposedCall,
        context: ExecutionContext,
    ) -> Mapping[str, Any]:
        """校验提案，只执行被允许的工具。"""
        spec = self._registry.get(call.name)
        if spec is None:
            LOGGER.warning("unknown tool: call_id=%s", call.call_id)
            raise ToolCallError("unknown tool")

        if call.name not in context.allowed_tools:
            LOGGER.warning(
                "tool denied: call_id=%s user_id=%s tool=%s",
                call.call_id,
                context.user_id,
                call.name,
            )
            raise AuthorizationError("tool is not allowed")

        spec.validator(call.arguments)

        operation_id = self._operation_id(call)
        previous = self._completed.get(operation_id)
        if previous is not None:
            LOGGER.info("idempotent replay: operation_id=%s", operation_id)
            return previous

        if spec.has_side_effect and operation_id not in context.approved_operation_ids:
            LOGGER.info("approval required: operation_id=%s", operation_id)
            raise AuthorizationError("explicit approval is required")

        LOGGER.info(
            "tool execution started: call_id=%s tool=%s user_id=%s",
            call.call_id,
            call.name,
            context.user_id,
        )
        output = spec.handler(call.arguments, context)
        result: Mapping[str, Any] = {
            "call_id": call.call_id,
            "status": "ok",
            "output": output,
        }
        self._completed[operation_id] = result
        LOGGER.info("tool execution completed: call_id=%s", call.call_id)
        return result

    @staticmethod
    def _operation_id(call: ProposedCall) -> str:
        """由调用 ID 与规范化 JSON 后的参数生成操作 ID。"""
        if not call.call_id:
            raise ToolCallError("call_id is required")
        # 把摘要纳入 ID，审批后参数被改动的操作就会被当成另一件事。
        canonical = json.dumps(
            call.arguments,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        return f"{call.name}:{call.call_id}:{digest}"

这段代码要抓住几个要点：

• 注册表里没有的工具，不做动态导入。
• 用户 ID 和允许列表不从模型参数里取。
• 校验之后，还要按资源做一次授权。
• 有副作用的操作，只在参数与已批准操作一致时才执行。
• 重试时返回同一操作已保存的结果。
• 日志里不打印完整参数、令牌和个人信息。

要注意 self._completed 只是说明用的。它扛不住进程重启、多实例部署和并发请求。而且模型如果换一个新的 call_id 重新提出语义相同的操作，也会被当成另一个操作。生产环境不能只依赖模型给出的 ID，而应在审批时由应用签发请求 ID，配合规范化参数，把结果事务性地写入带唯一约束的数据库。

第六步：把审批绑定到执行前的内容

只问一句「是否继续」，用户根本不知道自己在批准什么。要展示收件方、金额、目标资源、变更前后内容，并把审批对象固定为服务端记录。如果审批之后模型改了参数，操作 ID 会随之改变，就必须重新审批。高风险操作还要校验审批令牌的有效期、用户、会话和资源版本。

第七步：按依赖关系编排并行调用

即使模型能一次返回多个调用，也不要让它单独决定能否并行执行。

并行调用能降低延迟，但会同时消耗速率配额，也会增加部分失败。要设置最大并发数、整体截止时间和按工具区分的超时。另外，如果三件里只成功了两件，把整批重试会导致已成功的那两件再执行一次。结果信封要按条目保存，重试目标只限定在失败的条目上。

第八步：把工具结果也当作不可信输入

网页搜索、工单、邮件、文档返回的文本里，可能夹带经由外部数据传入的恶意指令。工具执行本身是可信的，不代表工具输出的内容也可信。应用侧要做这些控制：

• 只抽取允许的字段，去掉 HTML 和脚本。
• 限制字节数、记录条数和嵌套深度。
• 对密钥和个人信息做掩码。
• 用消息结构把数据与指令分开。
• 保留来源 URL、获取时间和版本。
• 不因为外部文本的内容就自动执行高风险工具。

外部页面里就算写着「把凭据发送到这个地址」，那也只是数据。它不能升级成修改应用网络白名单或授权策略的指令。

一个完整示例

把上面的步骤串起来，走一遍从模型提案到最终回答的最小流程。

假设用户说「帮我取消订单 A-104，我下错了」。应用先把工具定义交给模型，其中取消工具的参数 schema 只包含 order_id 和 reason，且 additionalProperties 设为 false，两个字段都必填。

模型返回：

{ "call_id": "call_01", "name": "request_order_cancellation", "arguments": { "order_id": "A-104", "reason": "ordered_by_mistake" } }

应用侧的处理顺序是：

1. 解析。确认这是一段合法 JSON，取出 call_id、name、arguments，构造 ProposedCall。此时它仍然是不可信数据。
2. 白名单检查。确认 request_order_cancellation 在注册表中，且在本次会话的 allowed_tools 里。
3. Schema 校验。确认 order_id 是字符串、reason 在允许的枚举内，没有多余字段。
4. 业务授权。用可信数据库确认订单 A-104 属于当前登录用户，且当前状态允许取消。
5. 生成操作 ID。对规范化后的参数做 SHA-256，拼成 request_order_cancellation:call_01:<digest>。
6. 查幂等缓存。如果这个操作 ID 已有结果，直接返回旧结果，不再执行。
7. 取审批。因为这是有副作用的操作，向用户展示订单号、当前状态和取消后果，拿到批准后把操作 ID 记入 approved_operation_ids。
8. 执行。调用处理函数，得到 {"request_id": "cancel_827", "state": "accepted"}。
9. 回传结果信封。把 call_id、status、最小化的 output 一起交给模型。
10. 模型给出最终回答。例如「订单 A-104 的取消申请已受理，受理编号 cancel_827」。

如果第 4 步发现订单不属于当前用户，就返回 authorization_denied，且只告知不可执行，不透露订单归属。如果第 8 步超时，不要立刻重试，而是先用操作 ID 查询持久化存储里是否已有结果，确认没有成功记录后再决定是否重试。

注意事项

Function Calling、Agent、MCP 处理的是不同边界。Function Calling 负责模型与应用之间交换调用提案和结果；Agent 或编排器负责状态、停止条件、工具选择循环和审批；MCP 负责应用与工具服务器之间的发现与调用标准化。MCP 的工具规范定义了服务器如何公开工具，以及客户端如何使用 tools/list 和 tools/call。两者不是二选一的关系：模型用 Function Calling 形式提出工具调用，应用再通过 MCP 去调用工具服务器，这种组合是可行的。但 MCP 服务器返回的工具注解和结果不能无条件信任，客户端仍要做确认、超时、结果校验和审计。

追踪要按阶段记录。只看最终回答的准确率，无法判断是调用选择坏了还是授权坏了。建议分阶段采集：

模型、提示词、工具 schema、调度器的版本也要记进同一条追踪里。模型更新之后，变化的不只是选择结果，调用次数、参数写法和并行程度都可能变。

评估分三层做。离线用固定样本和模拟工具复现选择、参数和停止条件；预发环境用沙箱账号验证超时、速率限制、审批和回滚；生产环境监控未授权执行率、重复率、p95 延迟和成本。有副作用的工具测试不要指向生产资源，要准备沙箱和试运行模式。

关于 Toolformer 的定位要清楚。它展示的是用自监督方式学习「调用哪个 API、何时调用、用什么参数、结果如何用于后续 token 预测」的研究方法，其数据生成流程大致是：从含少量示例的提示中采样调用位置和参数候选，执行 API 调用拿到结果，比较有结果和无结果两种情况下后续 token 的加权损失，只把能显著降低损失的调用插入原文，再用生成的数据按常规语言建模目标做微调。它并不是当前 Function Calling API 规范的来源，学习方法与推理时的协议要分开看。该研究的实验里还有每个样本最多一次 API 调用的限制，多工具串联的 agent 循环需要另行设计和评估。

生产环境检查清单：

• 把模型输出当作不可信提案来解析。
• 读工具与写工具分开。
• JSON 校验、schema 校验、业务授权分别记录。
• 用户 ID、租户 ID、访问范围从服务端上下文注入。
• 资源所有权在处理函数里再确认一次。
• 展示收件方、金额和变更内容后再取审批。
• 拒绝审批后被改动的参数。
• 为有副作用的工具实现持久化幂等。
• 用调用 ID 对应结果。
• 设置超时、调用上限、并发数、速率限制和整体截止时间。
• 净化工具结果，限制体积和字段。
• 按阶段评估选择、授权、执行和依据性。

让 Function Calling 安全的核心不在提示词工程，而在于把「模型可以提议的范围」和「应用允许执行的范围」分开，并用确定性的代码去强制后者。各家 API 的具体参数、配额与可用性会随时间变化，落地前请以官网当前信息为准。