第 18 章
Tool Use 完全指南:单工具 / 并行调用 / 工具链设计模式与 token 开销计算
第十八章:Tool Use 架构总览:定义、调用与结果注入
18.1 Tool Use 的核心设计理念
Tool Use(工具使用)是 Claude API 中最重要的能力扩展机制。它允许模型识别何时需要外部能力,构造结构化的函数调用请求,等待结果返回,然后将结果整合进最终答案。
与简单的提示词嵌入相比,Tool Use 的本质区别在于结构化契约:工具的输入输出格式是严格定义的,Claude 不是在文本中猜测结果,而是在明确的 Schema 约束下调用真实的外部系统。
Tool Use 解决了三类核心问题:
- 实时信息问题:Claude 的训练数据有截止日期,通过搜索工具可以获取最新信息
- 计算精确性问题:复杂数学和数据处理通过代码执行工具完成,避免 LLM 直接计算的误差
- 系统集成问题:通过数据库查询、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 会:
- 理解工具调用失败
- 可能尝试用不同参数重新调用
- 告知用户无法完成任务的原因
- 提供替代建议(如果有的话)
# 测试错误处理
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")
优化策略
- 精简工具描述:每个工具描述控制在 100 字以内,避免冗余
- 按需加载工具:根据对话上下文动态选择工具子集
- 结构化输出工具:使用
tool_choice: tool模式提取结构化数据时,比让 Claude 自由输出 JSON 更可靠 - 工具结果精简:避免返回超大 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 在复杂场景下做出更精准的工具选择。