AI Agent 的 Tool Calling 机制深度解析

引言

在上一篇文章《AI Agent 中的 Tool 系统》中，我们从应用层面讨论了工具的定义、分类和设计原则。本文则将镜头拉近，深入到 Tool Calling 的底层机制——LLM 如何学会调用工具、API 层面如何交互、解析和验证如何完成、以及生产环境中如何处理各种边界情况。

理解这些底层机制，就像理解编译原理对编程的帮助一样——日常不一定直接用到，但遇到问题时，你才能知道该往哪个方向排查。

LLM 如何学会调用工具

从文本补全到结构化调用

LLM 本质上是文本生成模型。让它们输出结构化的工具调用指令，需要额外的训练。目前主流的做法是在预训练后加入专门的工具调用微调：

1. 构建训练数据：将大量”用户请求 → 工具调用 → 工具返回 → 最终回答”的对话编入训练数据。
2. 注入特殊标记：在训练数据中使用特殊 token 标记工具调用的开始和结束，让模型学会区分”普通文本”和”工具调用指令”。
3. 多轮对话微调：重点训练模型在收到工具返回结果后的”下一步”行为——是继续调用工具，还是基于已有信息给出最终回答。

特殊 Token 的作用

不同的模型家族使用不同的方式标记工具调用：

OpenAI GPT 系列：在 API 层面将工具调用与文本生成分离。模型内部使用特殊的 token 标记，但开发者只需处理结构化的 tool_calls 数组。当模型决定调用工具时，finish_reason 被设为 tool_calls，message.tool_calls 包含函数名和参数的 JSON 字符串。

Anthropic Claude 系列：工具调用作为特殊的 content block 嵌入到消息流中。tool_use block 包含 id、name 和 input（JSON 对象），与普通的 text block 并列存在于同一个 assistant 消息中。这种设计的优势在于：工具调用与文本推理在上下文中是统一的，模型可以自然地交替输出推理文本和工具调用。

开源模型（Llama、Qwen 等）：通常使用类似于 <function_call> </function_call> 的 XML 风格标签，或者遵循 OpenAI 兼容格式。各开源模型在工具调用的实现质量上参差不齐，这是选择开源模型时需要重点评估的维度。

原始 API 交互格式

OpenAI 格式

一次完整的 OpenAI 工具调用交互包含以下消息序列：

[System Message]
{"role": "system", "content": "你是一个有帮助的助手。"}

[User Message]
{"role": "user", "content": "帮我查一下北京和上海明天的天气"}

[Assistant Message (工具调用)]
{"role": "assistant", "content": null, "tool_calls": [
    {"id": "call_001", "type": "function",
     "function": {"name": "get_weather", "arguments": "{\"city\": \"北京\"}"}},
    {"id": "call_002", "type": "function",
     "function": {"name": "get_weather", "arguments": "{\"city\": \"上海\"}"}}
]}

[Tool Result Messages]
{"role": "tool", "tool_call_id": "call_001",
 "content": "北京：晴，15-25°C"}
{"role": "tool", "tool_call_id": "call_002",
 "content": "上海：多云，18-26°C"}

[Assistant Message (最终回答)]
{"role": "assistant",
 "content": "北京明天晴，15-25°C；上海明天多云，18-26°C。两个城市天气都不错！"}

Anthropic 格式

Claude 的消息结构更加扁平，工具调用和结果都作为 content block：

[User Message]
{"role": "user", "content": "帮我查一下北京明天的天气"}

[Assistant Message (含工具调用)]
{"role": "assistant", "content": [
    {"type": "text", "text": "我来帮你查询北京的天气。"},
    {"type": "tool_use", "id": "toolu_001", "name": "get_weather",
     "input": {"city": "北京"}}
]}

[User Message (工具结果)]
{"role": "user", "content": [
    {"type": "tool_result", "tool_use_id": "toolu_001",
     "content": "北京：晴，15-25°C"}
]}

[Assistant Message (最终回答)]
{"role": "assistant", "content": [
    {"type": "text", "text": "北京明天晴，气温在 15 到 25 摄氏度之间。"}
]}

值得注意的一个设计细节是：Claude 中，工具结果以 user 角色 的消息回传，而不是像 OpenAI 那样有专门的 tool 角色。这个设计使得工具结果被模型视为”来自外部世界的输入”，而非”自己的输出”——这在 prompt injection 防御上有一定优势。

工具调用的解析与验证

JSON Schema 验证

模型生成的工具调用参数需要经过严格的验证才能执行：

import json
from jsonschema import validate, ValidationError

def validate_tool_call(tool_call_args: str, schema: dict) -> dict:
    """验证并解析工具调用参数"""
    try:
        # 第一步：解析 JSON
        args = json.loads(tool_call_args)
    except json.JSONDecodeError as e:
        raise ValueError(f"工具调用参数不是合法的 JSON：{e}")

    try:
        # 第二步：Schema 验证
        validate(instance=args, schema=schema)
    except ValidationError as e:
        raise ValueError(f"参数不符合 Schema 定义：{e.message}")

    return args

验证分为两层：

1. 语法层：参数必须是合法的 JSON。
2. 语义层：参数必须符合工具的 JSON Schema 定义——类型正确、必填参数齐全、约束条件满足。

处理畸形输出

模型可能产生各种格式错误：

# 常见的畸形输出模式
malformed_cases = [
    '{"city": "北京",}',           # 末尾多余逗号
    '{"city": "北京"',             # 缺少闭合括号
    "{'city': '北京'}",             # 单引号代替双引号
    '{"city": "北京", "unit": c}', # 未加引号的值
]

应对策略：

1. 宽容解析：使用 json5 或自定义解析器，在严格解析失败后尝试修复常见错误。
2. 错误反馈重试：如果无法修复，将错误信息反馈给模型，让模型重新生成。

def robust_parse_tool_args(raw_args: str, schema: dict, max_retries=3):
    for attempt in range(max_retries):
        try:
            return json.loads(raw_args)
        except json.JSONDecodeError:
            if attempt == max_retries - 1:
                raise
            # 将错误反馈给 LLM，让它重新生成
            raw_args = llm_regen(f"上次的参数格式错误：{raw_args}，请修正")

并行工具调用

机制原理

当 LLM 判断多个工具调用之间没有依赖关系时，可以在一次响应中同时发出多个工具调用。这一特性能够显著减少延迟。

OpenAI：模型在单次响应中返回多个 tool_calls（数组形式）。框架负责并行执行这些调用，将各自的结果以独立的 tool 消息回传。

Anthropic：模型在 assistant 消息中生成多个 tool_use block。由于每个 tool_use 有独立的 ID，结果可以按任意顺序回传。

执行逻辑

def execute_parallel_tool_calls(tool_calls: list) -> list:
    """并行执行多个无依赖的工具调用"""
    import concurrent.futures
    results = {}
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = {
            executor.submit(execute_tool, tc): tc.id
            for tc in tool_calls
        }
        for future in concurrent.futures.as_completed(futures):
            tc_id = futures[future]
            results[tc_id] = future.result()
    # 按原始顺序返回结果
    return [results[tc.id] for tc in tool_calls]

需要注意的边界情况：

• 某些工具调用不能在并行场景下安全执行（如文件系统的写操作）。
• 如果并行执行的总时间超过单次 API 超时，可能需要降级为串行执行。

流式工具调用

流式（Streaming）与工具调用的结合是一个技术挑战。因为工具调用本身是结构化的，而流式响应是增量的。

OpenAI SSE 流

OpenAI 的流式工具调用通过 tool_call_delta chunk 增量传输：

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_001",
      "function":{"name":"get_weather"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,
      "function":{"arguments":"{\"city\":"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,
      "function":{"arguments":" \"北京\"}"}}]}}]}

开发者需要在客户端维护一个缓冲区，将每个 delta 中的 arguments 片段拼接起来，直到接收到完整的 JSON 字符串。

Anthropic SSE 流

Anthropic 的流式响应通过 Server-Sent Events 协议传输，每个 content_block_start、content_block_delta、content_block_stop 事件构成了一个完整的工具调用生命周期：

event: content_block_start
data: {"type":"content_block_start","content_block":{"type":"tool_use","id":"toolu_001","name":"get_weather"}}

event: content_block_delta
data: {"type":"content_block_delta","delta":{"type":"input_json_delta","partial_json":"{\"city\":\"北京\"}"}}

event: content_block_stop

构建最小化的工具调用循环

以下是用简化的方式展示一个工具调用循环的核心逻辑（纯 Python 伪代码，约 30 行）：

class ToolCallingLoop:
    def __init__(self, model_client, tools: list, max_rounds=25):
        self.client = model_client
        self.tools = {t["name"]: t for t in tools}
        self.max_rounds = max_rounds

    def run(self, user_message: str) -> str:
        messages = [{"role": "user", "content": user_message}]

        for _ in range(self.max_rounds):
            response = self.client.send(messages, tools=self.tools)

            if response.has_tool_calls():
                # 执行工具调用并追加结果到消息历史
                for tc in response.tool_calls:
                    result = self.execute_tool(tc.name, tc.args)
                    messages.append(tc.to_message())
                    messages.append(self.format_result(tc.id, result))
            else:
                return response.text  # 模型给出最终回答

        raise RuntimeError("达到最大轮次，Agent 未完成任务")

    def execute_tool(self, name: str, args: dict) -> str:
        tool = self.tools[name]
        try:
            return tool["function"](**args)
        except Exception as e:
            return json.dumps({"error": str(e)})

这个循环体现了 Agent 的核心逻辑：迭代调用 LLM，当模型需要工具时执行工具并反馈结果，当模型认为可以给出最终答案时结束循环。

上下文窗口经济学

工具调用对上下文窗口的消耗涉及三个部分：

Token 消耗分析

组成部分	Token 消耗特点
工具定义	每个工具约 100-500 tokens，在每次请求中重复发送
调用请求	Tool Call 的 JSON 结构，通常 20-100 tokens/次
返回结果	可能很大，尤其是搜索或文件读取类的工具
对话历史	所有之前的消息，线性增长

优化策略

1. 工具数量控制：保持每次请求的工具定义在 10-20 个以内。超出时可以使用”工具路由”策略——先用一个轻量级分类器选择相关的子集。
2. 结果截断：为工具返回结果设置 token 上限（如 4000 tokens），超出部分截断并标注。
3. 历史压缩：在对话过长时，使用摘要或滑动窗口策略管理上下文。
4. 工具定义缓存：某些 API 提供商支持工具定义的 prompt caching，大幅减少重复发送的成本。

# 工具数量优化的示例：两级工具选择
def select_relevant_tools(user_query: str, all_tools: list) -> list:
    """先选择相关的工具子集，再传给 LLM"""
    # 用 embedding 或简单的关键词匹配预选工具
    query_embedding = embed(user_query)
    scores = []
    for tool in all_tools:
        tool_embedding = embed(tool["description"])
        score = cosine_similarity(query_embedding, tool_embedding)
        scores.append((score, tool))
    scores.sort(reverse=True)
    return [t for _, t in scores[:15]]  # 保留 top 15

常见故障与诊断

故障模式清单

故障类型	表现	排查方向
工具选择错误	调用不相关的工具	检查工具描述是否准确、是否与其他工具混淆
参数幻觉	虚构不存在的参数值	检查参数描述是否清晰、是否提供了足够的示例
JSON 格式错误	解析失败	模型对特殊字符的转义是否正确
无限循环	重复调用同一工具	检查工具返回结果是否满足模型的预期
空调用	finish_reason 为 tool_calls 但没有实际调用	API 或框架的边界 bug
参数缺失	Schema 校验不通过	检查 required 字段定义是否合理

调试工具链

# 一个简单的工具调用调试器
class ToolCallDebugger:
    def __init__(self):
        self.call_log = []

    def log_call(self, round_num, tool_name, args, result, latency_ms):
        self.call_log.append({
            "round": round_num,
            "tool": tool_name,
            "args": args,
            "result_preview": str(result)[:200],
            "latency_ms": latency_ms
        })

    def analyze(self):
        """分析调用模式，发现异常"""
        tool_counts = {}
        for entry in self.call_log:
            tool_counts[entry["tool"]] = tool_counts.get(entry["tool"], 0) + 1
        # 查找过度使用的工具（可能是循环的征兆）
        suspicious = {k: v for k, v in tool_counts.items() if v > 5}
        return {
            "total_rounds": len(self.call_log),
            "tool_distribution": tool_counts,
            "suspicious_loops": suspicious
        }

生产环境的最佳实践

1. 超时与重试

def execute_tool_with_retry(tool_name, args, max_retries=2, timeout=30):
    for attempt in range(max_retries + 1):
        try:
            result = tool_registry[tool_name](**args, _timeout=timeout)
            return result
        except TimeoutError:
            if attempt == max_retries:
                return json.dumps({"error": "工具执行超时"})
        except Exception as e:
            if attempt == max_retries:
                return json.dumps({"error": str(e)})
    # 指数退避
    time.sleep(2 ** attempt)

2. 权限沙箱

工具执行应在受限环境中进行，特别是代码执行和文件操作类工具。Docker 容器、WebAssembly 沙箱、或操作系统级别的权限控制（如 seccomp）都是可选的隔离方案。

3. 审计日志

记录每一次工具调用的完整信息（时间、参数、返回值、耗时），便于事后排查问题和安全审计。这不仅是工程需要，在金融、医疗等领域也是合规要求。

总结

Tool Calling 的底层机制看似简单——模型输出 JSON，系统解析并执行——但真正做好需要考虑的细节非常多：JSON 解析的鲁棒性、并行执行的协调、流式传输的状态管理、上下文窗口的经济性、以及生产环境的可靠性保障。

理解这些底层机制的价值在于：当你的 Agent “莫名其妙”地失败了，你不会停留在”模型不够聪明”的模糊判断上，而是能够精准定位到是 JSON 解析的问题，还是工具描述误导了模型，抑或是上下文窗口溢出导致模型”遗忘”了之前的工具调用结果。

调试 Agent 行为时的一条黄金法则：先怀疑工具系统，再怀疑模型本身。80% 的 Agent 故障源于工具定义不清晰或工具执行的边界情况处理不当，只有 20% 是模型本身的问题。