Tool Calling 如何让 LLM 从“会说”变成“会做”

内容摘要：工具调用是 Agent 从「会说」到「会做」的关键。本章先讲 JSON Schema 工具定义与参数校验，再给出一个完整、可离线测试的最小 Agent：含一个只读工具、参数校验、超时与错误处理、最大循环次数、工具调用日志、停止条件、单元测试，以及一次成功轨迹与一次失败轨迹。难度：进阶　|　前置知识：03 模型 API、04 Prompt

常见误区

模型输出了合法的工具名和 JSON 参数，就等于工具已经安全执行；工具描述本身也可以被完全信任。

先给结论

工具调用（Tool Calling）只是模型提出结构化动作请求。应用代码仍要完成工具白名单、Schema 校验、业务校验、权限判断、超时执行、结果回传和循环终止，写操作还需审批与幂等。

带着这三个问题阅读：

模型、编排层和工具执行器之间的责任边界在哪里？
为什么 JSON Schema 合法不代表业务参数安全？
一条成功轨迹和一条失败轨迹分别应该记录哪些证据？

一、概念与用途

工具调用（Tool Calling）/ 函数调用（Function Calling）：模型不直接执行动作，而是输出一个结构化的调用请求（工具名 + 参数），由应用代码执行真正的函数，再把结果回传给模型。

关键边界：模型只建议调用；是否执行、用什么参数执行、有没有权限执行，全部由应用决定。绝不能直接信任模型生成的参数。

二、工具定义：完整 JSON Schema

工具声明必须给出明确类型与约束。以只读工具 get_slow_queries 为例：

json

{
  "name": "get_slow_queries",
  "description": "查询指定测试/预发数据库实例的慢查询列表（只读，无副作用）。",
  "parameters": {
    "type": "object",
    "properties": {
      "instance_id": {
        "type": "string",
        "pattern": "^[a-z0-9-]{3,40}$",
        "description": "实例 ID，仅小写字母数字与连字符"
      },
      "top_n": {
        "type": "integer",
        "minimum": 1,
        "maximum": 20,
        "default": 5,
        "description": "返回前 N 条最慢的查询"
      }
    },
    "required": ["instance_id"],
    "additionalProperties": false
  }
}

要点：pattern 限制格式、minimum/maximum 限制范围、additionalProperties: false 拒绝未知字段、required 声明必填。这些约束在校验时逐条执行。

三、参数校验：Schema + 业务规则双层

Schema 校验保证「格式合法」，业务规则校验保证「业务允许」。二者缺一不可：

python

from jsonschema import validate, ValidationError

GET_SLOW_QUERIES_SCHEMA = {
    "type": "object",
    "properties": {
        "instance_id": {"type": "string", "pattern": "^[a-z0-9-]{3,40}$"},
        "top_n": {"type": "integer", "minimum": 1, "maximum": 20},
    },
    "required": ["instance_id"],
    "additionalProperties": False,
}

# 业务规则：只允许非生产实例（示例白名单）
ALLOWED_INSTANCES = {"mysql-test-01", "mysql-staging-02", "redis-test-01"}

def validate_get_slow_queries(args: dict) -> dict:
    """先做 Schema 校验，再做业务规则校验；任一失败都抛错，绝不放行。"""
    try:
        validate(instance=args, schema=GET_SLOW_QUERIES_SCHEMA)
    except ValidationError as e:
        raise ValueError(f"参数不符合 Schema：{e.message}")
    instance_id = args["instance_id"]
    if instance_id not in ALLOWED_INSTANCES:
        raise ValueError(f"实例 {instance_id} 不在允许列表内（仅限非生产实例）")
    args.setdefault("top_n", 5)  # 应用默认值，不依赖模型
    return args

四、完整最小 Agent

下面是一个不依赖网络、可离线运行的最小 Agent。它用一个 ModelClient 协议抽象模型，真实环境用 03 的客户端，测试用 Mock。它满足清单：只读工具、Schema 校验、超时、错误处理、最大循环次数、工具日志、停止条件。

python

"""minimal_agent.py —— 安全的最小 Agent（可离线测试）。
真实模型接入见 03；本文件不含任何真实密钥或网络请求。
"""
from __future__ import annotations
import json
import time
import logging
from dataclasses import dataclass, field
from typing import Protocol, Callable

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("agent")

# ---------- 工具层 ----------
from jsonschema import validate, ValidationError

GET_SLOW_QUERIES_SCHEMA = {
    "type": "object",
    "properties": {
        "instance_id": {"type": "string", "pattern": "^[a-z0-9-]{3,40}$"},
        "top_n": {"type": "integer", "minimum": 1, "maximum": 20},
    },
    "required": ["instance_id"],
    "additionalProperties": False,
}
ALLOWED_INSTANCES = {"mysql-test-01", "mysql-staging-02"}

# 示例数据源（真实实现会连只读副本，这里用内存数据模拟）
_FAKE_DB = {
    "mysql-test-01": [
        {"sql": "SELECT * FROM orders WHERE user_id=?", "seconds": 4.2},
        {"sql": "SELECT COUNT(*) FROM logs", "seconds": 2.1},
    ],
}

def get_slow_queries(instance_id: str, top_n: int = 5) -> list[dict]:
    """只读工具：返回实例的慢查询（示例数据）。"""
    rows = _FAKE_DB.get(instance_id, [])
    return sorted(rows, key=lambda r: r["seconds"], reverse=True)[:top_n]

def validate_get_slow_queries(args: dict) -> dict:
    try:
        validate(instance=args, schema=GET_SLOW_QUERIES_SCHEMA)
    except ValidationError as e:
        raise ValueError(f"参数不符合 Schema：{e.message}")
    if args["instance_id"] not in ALLOWED_INSTANCES:
        raise ValueError(f"实例 {args['instance_id']} 不在允许列表内")
    args.setdefault("top_n", 5)
    return args

@dataclass
class Tool:
    name: str
    validator: Callable[[dict], dict]
    func: Callable[..., object]
    timeout_s: float = 5.0

TOOLS: dict[str, Tool] = {
    "get_slow_queries": Tool("get_slow_queries", validate_get_slow_queries, get_slow_queries),
}

# ---------- 模型抽象 ----------
class ModelClient(Protocol):
    def decide(self, messages: list[dict]) -> dict:
        """返回决策：{"action":"call_tool","tool":..,"args":..} 或 {"action":"final","answer":..}"""
        ...

# ---------- Agent 循环 ----------
@dataclass
class AgentResult:
    answer: str
    status: str                      # "ok" | "max_steps" | "error"
    trace: list[dict] = field(default_factory=list)

def run_agent(client: ModelClient, goal: str, *, max_steps: int = 5) -> AgentResult:
    """执行 Agent 循环：决策→(校验→执行工具→回填)→再决策，直到 final 或达上限。"""
    messages = [
        {"role": "system", "content": "你是只读排障助手，只能调用提供的工具，不得臆造数据。"},
        {"role": "user", "content": goal},
    ]
    trace: list[dict] = []
    for step in range(1, max_steps + 1):
        decision = client.decide(messages)
        logger.info("step=%d decision=%s", step, decision.get("action"))

        if decision.get("action") == "final":
            trace.append({"step": step, "type": "final"})
            return AgentResult(decision.get("answer", ""), "ok", trace)

        if decision.get("action") != "call_tool":
            trace.append({"step": step, "type": "bad_decision", "raw": decision})
            return AgentResult("模型返回了无法识别的决策", "error", trace)

        tool_name = decision.get("tool", "")
        raw_args = decision.get("args", {})
        tool = TOOLS.get(tool_name)
        if tool is None:
            # 失败路径①：模型幻想出不存在的工具
            msg = f"未知工具：{tool_name}"
            trace.append({"step": step, "type": "tool_error", "error": msg})
            messages.append({"role": "tool", "content": msg})
            continue

        # ① 参数校验（Schema + 业务规则）
        try:
            args = tool.validator(dict(raw_args))
        except ValueError as e:
            # 失败路径②：参数非法，把错误回传模型让其纠正，不执行
            err = f"参数校验失败：{e}"
            logger.info("  %s", err)
            trace.append({"step": step, "type": "validation_error", "error": str(e)})
            messages.append({"role": "tool", "content": err})
            continue

        # ② 带超时执行
        started = time.monotonic()
        try:
            result = tool.func(**args)
        except Exception as e:  # 失败路径③：工具内部异常
            err = f"工具执行异常：{e}"
            trace.append({"step": step, "type": "exec_error", "error": str(e)})
            messages.append({"role": "tool", "content": err})
            continue
        elapsed = time.monotonic() - started
        if elapsed > tool.timeout_s:
            # 失败路径④：超时（示例中同步执行，真实场景用可取消的执行器）
            trace.append({"step": step, "type": "timeout", "elapsed": elapsed})
            messages.append({"role": "tool", "content": "工具超时"})
            continue

        # ③ 工具调用日志 + 回填结果
        logger.info("  tool=%s args=%s elapsed=%.3fs", tool_name, args, elapsed)
        trace.append({"step": step, "type": "tool_ok", "tool": tool_name,
                      "args": args, "elapsed": elapsed})
        messages.append({"role": "tool",
                         "content": json.dumps(result, ensure_ascii=False)})

    # 停止条件：达到最大步数仍未结束
    return AgentResult("已达最大步数仍未得出结论", "max_steps", trace)

五、成功轨迹与失败轨迹（含 Mock 模型）

用脚本化的 Mock 模型驱动 Agent，无需网络即可复现两条轨迹：

python

"""demo_traces.py"""
from minimal_agent import run_agent

class ScriptedModel:
    """按预设脚本逐步返回决策，用于确定性地复现轨迹。"""
    def __init__(self, steps: list[dict]):
        self._steps = steps
        self._i = 0
    def decide(self, messages: list[dict]) -> dict:
        step = self._steps[min(self._i, len(self._steps) - 1)]
        self._i += 1
        return step

# —— 成功轨迹：先调工具，拿到结果后给出最终答案 ——
success = ScriptedModel([
    {"action": "call_tool", "tool": "get_slow_queries", "args": {"instance_id": "mysql-test-01", "top_n": 2}},
    {"action": "final", "answer": "mysql-test-01 最慢查询是 orders 全表扫描(4.2s)，建议加索引。"},
])
r1 = run_agent(success, "排查 mysql-test-01 的慢查询")
print("SUCCESS:", r1.status, "| steps:", len(r1.trace))

# —— 失败轨迹：模型给了非法实例，被校验拦截，随后放弃 ——
failure = ScriptedModel([
    {"action": "call_tool", "tool": "get_slow_queries", "args": {"instance_id": "prod-01"}},
    {"action": "final", "answer": "无法处理生产实例，已拒绝。"},
])
r2 = run_agent(failure, "把 prod-01 的慢查询干掉")
print("FAILURE:", r2.status, "| 轨迹类型:", [t["type"] for t in r2.trace])

预期输出（[非确定性] 不涉及真实模型，故本例确定）：

text

SUCCESS: ok | steps: 2
FAILURE: ok | 轨迹类型: ['validation_error', 'final']

失败轨迹里，prod-01 因不在白名单被 validation_error 拦下——校验在工具执行前生效，危险参数根本没机会跑到数据源。

六、单元测试

python

"""test_minimal_agent.py — pytest"""
import pytest
from minimal_agent import (validate_get_slow_queries, run_agent, get_slow_queries)

class OneShot:
    def __init__(self, decision): self.d = decision
    def decide(self, messages): return self.d

def test_validate_ok():
    args = validate_get_slow_queries({"instance_id": "mysql-test-01"})
    assert args["top_n"] == 5           # 默认值由应用注入

def test_validate_rejects_bad_pattern():
    with pytest.raises(ValueError):
        validate_get_slow_queries({"instance_id": "PROD 01!"})

def test_validate_rejects_non_allowlisted():
    with pytest.raises(ValueError):
        validate_get_slow_queries({"instance_id": "prod-01"})

def test_validate_rejects_out_of_range():
    with pytest.raises(ValueError):
        validate_get_slow_queries({"instance_id": "mysql-test-01", "top_n": 999})

def test_agent_stops_on_final():
    r = run_agent(OneShot({"action": "final", "answer": "done"}), "任意目标")
    assert r.status == "ok" and r.answer == "done"

def test_agent_unknown_tool_then_max_steps():
    # 模型一直幻想不存在的工具 → 触发 max_steps 停止条件
    r = run_agent(OneShot({"action": "call_tool", "tool": "rm_rf", "args": {}}), "x", max_steps=3)
    assert r.status == "max_steps"
    assert all(t["type"] == "tool_error" for t in r.trace)

def test_tool_readonly_returns_sorted():
    rows = get_slow_queries("mysql-test-01")
    assert rows and rows[0]["seconds"] >= rows[-1]["seconds"]

运行：pip install jsonschema pytest && pytest -q。预期：全部通过。

七、失败路径（本章覆盖的完整清单）

失败类型	触发条件	Agent 行为
未知工具	模型幻想工具名	回传错误，不执行，继续/最终触发 max_steps
Schema 非法	类型/格式/范围不符	校验拦截，回传错误让模型纠正
业务规则拒绝	非白名单实例	拦截，绝不触达数据源
工具异常	函数内部报错	捕获并回传，不崩溃
超时	执行超时	记录并回传超时
达步数上限	循环不收敛	安全停止，返回 `max_steps`

八、测试与验证

单测覆盖校验的每条约束与每条失败路径（见上）。
轨迹断言：对成功/失败轨迹断言 trace 中出现的事件类型顺序，而非只看最终文本。
这正是「不以模型看起来对为验收标准」的落地方式（16）。

九、安全与权限

最小权限：本章只放只读工具；写/高危工具见 13、14、15。
不信任模型参数：一切参数经 Schema + 业务规则校验后才执行。
白名单优先：实例、路径、命令都用允许列表，而非黑名单。

十、性能与成本

max_steps 同时是成本闸门：它限制了最坏情况下的模型调用与工具调用次数。
工具超时避免慢依赖拖垮整个循环。

十一、框架差异

OpenAI SDK 用 tools/tool_calls，LangChain 用 @tool + create_agent，MCP 用统一协议暴露工具（07）。无论何种，参数校验与权限必须由你的代码负责，框架不替你保证业务安全。

十二、关联章节

下一章：07 MCP、08 Agent Loop 与规划
相关：13 HITL、14 安全、16 测试

面试复述

Tool Calling 的本质是“模型提议、代码裁决、工具执行、结果回传”。模型不能直接碰外部系统，任何参数都要经过 Schema、业务规则与权限校验，并受超时、步数和预算约束。

复述时先给结论，再补机制、适用边界和一个可验证指标；不要只罗列框架名称或产品功能。

迁移练习

在本章最小 Agent 中新增一个写工具，要求加入幂等键、人工审批和审计记录；再构造一次非法参数轨迹，验证工具函数没有被调用。

检索关键词

Tool Calling 原理 JSON Schema 参数校验 最小 Agent 示例 最大循环次数 工具超时 成功轨迹失败轨迹 allowlist 白名单

官方来源

OpenAI Function/Tool Calling 文档、JSON Schema 规范（2020-12）；各框架工具定义文档（verified_at: 2026-09-07）。

Tool Calling 如何让 LLM 从“会说”变成“会做” ​

先给结论 ​

一、概念与用途 ​

二、工具定义：完整 JSON Schema ​

三、参数校验：Schema + 业务规则双层 ​

四、完整最小 Agent ​

五、成功轨迹与失败轨迹（含 Mock 模型） ​

六、单元测试 ​

七、失败路径（本章覆盖的完整清单） ​

八、测试与验证 ​

九、安全与权限 ​

十、性能与成本 ​

十一、框架差异 ​

十二、关联章节 ​

面试复述 ​

迁移练习 ​

检索关键词 ​

官方来源 ​

Tool Calling 如何让 LLM 从“会说”变成“会做”

先给结论

一、概念与用途

二、工具定义：完整 JSON Schema

三、参数校验：Schema + 业务规则双层

四、完整最小 Agent

五、成功轨迹与失败轨迹（含 Mock 模型）

六、单元测试

七、失败路径（本章覆盖的完整清单）

八、测试与验证

九、安全与权限

十、性能与成本

十一、框架差异

十二、关联章节

面试复述

迁移练习

检索关键词

官方来源