第 18 章

Tool Use 完全指南:单工具 / 并行调用 / 工具链设计模式与 token 开销计算

第十八章:Tool Use 架构总览:定义、调用与结果注入

18.1 Tool Use 的核心设计理念

Tool Use(工具使用)是 Claude API 中最重要的能力扩展机制。它允许模型识别何时需要外部能力,构造结构化的函数调用请求,等待结果返回,然后将结果整合进最终答案。

与简单的提示词嵌入相比,Tool Use 的本质区别在于结构化契约:工具的输入输出格式是严格定义的,Claude 不是在文本中猜测结果,而是在明确的 Schema 约束下调用真实的外部系统。

Tool Use 解决了三类核心问题:

  1. 实时信息问题:Claude 的训练数据有截止日期,通过搜索工具可以获取最新信息
  2. 计算精确性问题:复杂数学和数据处理通过代码执行工具完成,避免 LLM 直接计算的误差
  3. 系统集成问题:通过数据库查询、API 调用工具,将 Claude 接入企业现有系统

18.2 完整的 Tool Use 交互流程

理解 Tool Use 的关键是掌握它的多轮交互模式。一次完整的工具调用涉及以下步骤:

用户消息 → Claude 分析 → tool_use block(调用请求)
    ↓
开发者执行工具 → 获取结果
    ↓
tool_result block(结果注入)→ Claude 生成最终答案

这个过程在 API 层面由以下消息序列表示:

# 第一轮:发送用户请求,附带工具定义
messages = [
    {"role": "user", "content": "查询用户 ID 为 12345 的账户余额"}
]

response1 = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=[...],  # 工具定义
    messages=messages
)
# response1.stop_reason == "tool_use"
# response1.content 包含 tool_use block

# 第二轮:注入工具结果
messages.append({"role": "assistant", "content": response1.content})
messages.append({
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": response1.content[0].id,
        "content": "账户余额:¥15,234.56"
    }]
})

response2 = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=[...],
    messages=messages
)
# response2.stop_reason == "end_turn"
# 包含最终文本答案

18.3 工具定义的完整 JSON Schema

基础工具结构

每个工具由以下字段定义:

{
  "name": "get_account_balance",
  "description": "查询指定用户的账户余额。返回当前余额(人民币)和最后更新时间。",
  "input_schema": {
    "type": "object",
    "properties": {
      "user_id": {
        "type": "string",
        "description": "用户唯一标识符,格式为数字字符串"
      },
      "currency": {
        "type": "string",
        "enum": ["CNY", "USD", "EUR"],
        "description": "返回余额的货币单位,默认 CNY",
        "default": "CNY"
      }
    },
    "required": ["user_id"]
  }
}

多工具定义示例

实际生产系统中通常需要定义多个工具:

import anthropic
import json

tools = [
    {
        "name": "search_database",
        "description": "在企业数据库中搜索记录。支持模糊匹配和字段过滤。",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "table": {
                    "type": "string",
                    "enum": ["users", "orders", "products", "invoices"],
                    "description": "要搜索的数据表"
                },
                "limit": {
                    "type": "integer",
                    "description": "返回结果数量上限",
                    "minimum": 1,
                    "maximum": 100,
                    "default": 10
                },
                "filters": {
                    "type": "object",
                    "description": "字段过滤条件,键为字段名,值为过滤值",
                    "additionalProperties": {
                        "type": "string"
                    }
                }
            },
            "required": ["query", "table"]
        }
    },
    {
        "name": "send_email",
        "description": "发送电子邮件。用于通知用户或发送报告。",
        "input_schema": {
            "type": "object",
            "properties": {
                "to": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "收件人邮箱地址列表"
                },
                "subject": {
                    "type": "string",
                    "description": "邮件主题"
                },
                "body": {
                    "type": "string",
                    "description": "邮件正文,支持 HTML"
                },
                "cc": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "抄送地址列表(可选)"
                }
            },
            "required": ["to", "subject", "body"]
        }
    },
    {
        "name": "execute_calculation",
        "description": "执行精确的数学计算。对于复杂的数值计算,始终使用此工具而非直接计算。",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式,使用 Python 语法。例如:'2 ** 32' 或 'sum([1,2,3,4,5])'"
                }
            },
            "required": ["expression"]
        }
    }
]

18.4 tool_use block 的完整结构

当 Claude 决定调用工具时,响应的 content 数组中会出现 tool_use 类型的块:

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "搜索所有状态为 pending 的订单"}]
)

# 检查 stop_reason
print(f"Stop reason: {response.stop_reason}")  # "tool_use"

# 检查 content
for block in response.content:
    print(f"Block type: {block.type}")
    if block.type == "tool_use":
        print(f"Tool name: {block.name}")
        print(f"Tool use ID: {block.id}")
        print(f"Input: {json.dumps(block.input, ensure_ascii=False, indent=2)}")

典型的 tool_use block 数据结构(JSON 表示):

{
  "type": "tool_use",
  "id": "toolu_01XFDUDYJgAf9n7mP7YT7V5H",
  "name": "search_database",
  "input": {
    "query": "pending",
    "table": "orders",
    "filters": {
      "status": "pending"
    },
    "limit": 50
  }
}

18.5 tool_result block 的注入格式

执行工具后,需要将结果以 tool_result 格式注入消息历史:

成功结果

tool_result_success = {
    "type": "tool_result",
    "tool_use_id": "toolu_01XFDUDYJgAf9n7mP7YT7V5H",
    "content": json.dumps({
        "results": [
            {"order_id": "ORD-001", "user_id": "U123", "amount": 299.00, "status": "pending"},
            {"order_id": "ORD-002", "user_id": "U456", "amount": 1599.00, "status": "pending"}
        ],
        "total_count": 2,
        "query_time_ms": 45
    }, ensure_ascii=False)
}

错误结果

tool_result_error = {
    "type": "tool_result",
    "tool_use_id": "toolu_01XFDUDYJgAf9n7mP7YT7V5H",
    "content": "查询失败:数据库连接超时(已重试3次)",
    "is_error": True
}

图像结果

某些工具(如截图工具)可以返回图像内容:

import base64

with open("screenshot.png", "rb") as f:
    image_data = base64.standard_b64encode(f.read()).decode("utf-8")

tool_result_image = {
    "type": "tool_result",
    "tool_use_id": "toolu_01XFDUDYJgAf9n7mP7YT7V5H",
    "content": [
        {
            "type": "image",
            "source": {
                "type": "base64",
                "media_type": "image/png",
                "data": image_data
            }
        },
        {
            "type": "text",
            "text": "截图已捕获,分辨率 1920x1080"
        }
    ]
}

18.6 完整的端到端工具调用实现

工具执行引擎

import anthropic
import json
from typing import Any, Dict, List, Callable

class ToolExecutor:
    """工具执行引擎"""
    
    def __init__(self):
        self.tools: Dict[str, Callable] = {}
        self.tool_definitions: List[dict] = []
    
    def register(self, tool_definition: dict, func: Callable):
        """注册工具"""
        name = tool_definition["name"]
        self.tools[name] = func
        self.tool_definitions.append(tool_definition)
        return self
    
    def execute(self, tool_name: str, tool_input: dict) -> Any:
        """执行工具调用"""
        if tool_name not in self.tools:
            raise ValueError(f"未知工具: {tool_name}")
        return self.tools[tool_name](**tool_input)
    
    def build_result_block(self, tool_use_id: str, result: Any, 
                           is_error: bool = False) -> dict:
        """构建 tool_result block"""
        if isinstance(result, str):
            content = result
        else:
            content = json.dumps(result, ensure_ascii=False, default=str)
        
        block = {
            "type": "tool_result",
            "tool_use_id": tool_use_id,
            "content": content
        }
        if is_error:
            block["is_error"] = True
        return block


class ToolUseAgent:
    """支持工具调用的 Agent"""
    
    def __init__(self, executor: ToolExecutor, model: str = "claude-opus-4-5"):
        self.client = anthropic.Anthropic()
        self.executor = executor
        self.model = model
        self.max_iterations = 10  # 防止无限循环
    
    def run(self, user_message: str, system: str = "") -> str:
        """运行工具调用循环"""
        messages = [{"role": "user", "content": user_message}]
        create_kwargs = {
            "model": self.model,
            "max_tokens": 4096,
            "tools": self.executor.tool_definitions,
            "messages": messages
        }
        if system:
            create_kwargs["system"] = system
        
        for iteration in range(self.max_iterations):
            response = self.client.messages.create(**create_kwargs)
            
            # 如果完成,返回文本答案
            if response.stop_reason == "end_turn":
                return ' '.join(
                    b.text for b in response.content if b.type == "text"
                )
            
            # 处理工具调用
            if response.stop_reason == "tool_use":
                tool_results = []
                
                for block in response.content:
                    if block.type != "tool_use":
                        continue
                    
                    print(f"[工具调用] {block.name}({json.dumps(block.input, ensure_ascii=False)})")
                    
                    try:
                        result = self.executor.execute(block.name, block.input)
                        result_block = self.executor.build_result_block(block.id, result)
                        print(f"[工具结果] 成功")
                    except Exception as e:
                        result_block = self.executor.build_result_block(
                            block.id, str(e), is_error=True
                        )
                        print(f"[工具结果] 错误: {e}")
                    
                    tool_results.append(result_block)
                
                # 更新消息历史
                messages.append({"role": "assistant", "content": response.content})
                messages.append({"role": "user", "content": tool_results})
                create_kwargs["messages"] = messages
            else:
                # 未预期的 stop_reason
                break
        
        return "达到最大迭代次数,任务可能未完成。"


# 使用示例
def setup_demo_agent():
    executor = ToolExecutor()
    
    # 注册搜索工具
    executor.register(
        {
            "name": "search_web",
            "description": "搜索互联网获取最新信息",
            "input_schema": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        },
        lambda query: f"搜索结果:关于'{query}'的最新信息..."  # 实际替换为真实搜索
    )
    
    # 注册计算工具
    executor.register(
        {
            "name": "calculate",
            "description": "执行数学计算",
            "input_schema": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Python 数学表达式"}
                },
                "required": ["expression"]
            }
        },
        lambda expression: eval(expression, {"__builtins__": {}}, 
                                 {"sum": sum, "abs": abs, "round": round, "len": len})
    )
    
    return ToolUseAgent(executor)

18.7 tool_choice 参数:控制工具选择行为

tool_choice 参数可以精确控制 Claude 是否使用工具以及使用哪个工具:

# 模式一:auto(默认)——Claude 自行决定是否使用工具
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "auto"},
    messages=messages
)

# 模式二:any——必须使用至少一个工具
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "any"},
    messages=messages
)

# 模式三:tool——强制使用指定工具
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={
        "type": "tool",
        "name": "search_database"  # 强制使用此工具
    },
    messages=messages
)

# 模式四:none——禁止使用工具(仅文本输出)
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "none"},
    messages=messages
)

使用场景

tool_choice 模式 适用场景
auto 通用场景,让 Claude 判断是否需要工具
any 确保 Claude 不直接用记忆回答,必须查询最新数据
tool + name 数据提取任务,强制结构化输出
none 已收集所有信息,只需 Claude 汇总分析

18.8 错误处理与重试策略

工具执行错误处理

class RobustToolExecutor(ToolExecutor):
    """带重试和错误处理的工具执行器"""
    
    def __init__(self, max_retries: int = 3, retry_delay: float = 1.0):
        super().__init__()
        self.max_retries = max_retries
        self.retry_delay = retry_delay
    
    def execute(self, tool_name: str, tool_input: dict) -> Any:
        """带重试的工具执行"""
        import time
        
        last_error = None
        for attempt in range(self.max_retries):
            try:
                return super().execute(tool_name, tool_input)
            except Exception as e:
                last_error = e
                if attempt < self.max_retries - 1:
                    print(f"工具 {tool_name} 执行失败(第{attempt+1}次),{self.retry_delay}秒后重试...")
                    time.sleep(self.retry_delay)
        
        raise RuntimeError(f"工具 {tool_name} 在 {self.max_retries} 次尝试后仍失败: {last_error}")

Claude 处理工具错误的行为

is_error: True 时,Claude 会:

  1. 理解工具调用失败
  2. 可能尝试用不同参数重新调用
  3. 告知用户无法完成任务的原因
  4. 提供替代建议(如果有的话)
# 测试错误处理
messages = [{"role": "user", "content": "查询用户 99999 的订单"}]
messages.append({"role": "assistant", "content": [
    {"type": "tool_use", "id": "toolu_test", "name": "search_database",
     "input": {"query": "99999", "table": "orders"}}
]})
messages.append({"role": "user", "content": [{
    "type": "tool_result",
    "tool_use_id": "toolu_test",
    "content": "错误:用户 99999 不存在于数据库中",
    "is_error": True
}]})

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=512,
    tools=tools,
    messages=messages
)
# Claude 会优雅地处理此错误,向用户说明情况

18.9 Token 计算与性能优化

工具定义消耗的 token

工具定义会作为系统提示的一部分发送,消耗输入 token:

# 估算工具定义的 token 消耗
def estimate_tool_tokens(tools: list) -> int:
    """粗略估算工具定义消耗的 tokens"""
    tool_text = json.dumps(tools, ensure_ascii=False)
    # 英文约 4 字符/token,中文约 2 字符/token
    return len(tool_text) // 3  # 粗略估算

print(f"工具定义预计消耗 ~{estimate_tool_tokens(tools)} tokens")

优化策略

  1. 精简工具描述:每个工具描述控制在 100 字以内,避免冗余
  2. 按需加载工具:根据对话上下文动态选择工具子集
  3. 结构化输出工具:使用 tool_choice: tool 模式提取结构化数据时,比让 Claude 自由输出 JSON 更可靠
  4. 工具结果精简:避免返回超大 JSON,只返回 Claude 需要的字段
def filter_tool_result(raw_result: dict, needed_fields: list) -> dict:
    """过滤工具结果,只保留必要字段"""
    return {k: v for k, v in raw_result.items() if k in needed_fields}

小结

Tool Use 架构的核心是一个严格的契约:工具的 Schema 定义了输入格式,tool_use block 是 Claude 的调用请求,tool_result block 是开发者的结果注入,整个循环由 stop_reason 驱动。

掌握这个基础架构后,下一章将深入探讨如何设计高质量的工具 Schema,让 Claude 在复杂场景下做出更精准的工具选择。

本章评分
4.7  / 5  (18 评分)

💬 留言讨论