并行工具调用
让 Claude 同时调用多个工具以提高效率
🌐 查看英文原文 | 源码 Notebook
Claude 3.7 Sonnet 上的并行工具调用
Claude 3.7 Sonnet 可能不太倾向于在单次回复中进行并行工具调用，即使你没有设置 disable_parallel_tool_use。为解决此问题，我们建议引入一个 “批量工具”（batch tool），它可以作为元工具来同时包装对其他工具的调用。我们发现，如果存在此工具，模型将使用它在并行中同时调用多个工具。
让我们来看看这个问题，并详细探讨这个解决方案。
from anthropic import Anthropic
client = Anthropic()
MODEL_NAME = "claude-sonnet-4-6"执行包含多次工具调用的查询
回顾一下，默认行为是允许 Claude 进行并行工具调用。结合默认的 tool_choice 设置为 auto，这意味着 Claude 可以调用任何指定的工具，或者在单个助手回合中调用多个工具。
让我们为 Claude 设置 get_weather 和 get_time 工具。
def get_weather(location):
# Pretend to get the weather, and just return a fixed value.
return f"The weather in {location} is 72 degrees and sunny."
def get_time(location):
# Pretend to get the time, and just return a fixed value.
return f"The time in {location} is 12:32 PM."
weather_tool = {
"name": "get_weather",
"description": "Gets the weather for in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
"required": ["location"],
},
}
time_tool = {
"name": "get_time",
"description": "Gets the time in a given location",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
},
"required": ["location"],
},
}
def process_tool_call(tool_name, tool_input):
if tool_name == "get_weather":
return get_weather(tool_input["location"])
elif tool_name == "get_time":
return get_time(tool_input["location"])
else:
raise ValueError(f"Unexpected tool name: {tool_name}")接下来，让我们为 Claude 提供这些工具并执行查询。
def make_query_and_print_result(messages, tools=None):
response = client.messages.create(
model=MODEL_NAME,
messages=messages,
max_tokens=1000,
tool_choice={"type": "auto"},
tools=tools or [weather_tool, time_tool],
)
for block in response.content:
match block.type:
case "text":
print(block.text)
case "tool_use":
print(f"Tool: {block.name}({block.input})")
case _:
raise ValueError(f"Unexpected block type: {block.type}")
return response
MESSAGES = [{"role": "user", "content": "What's the weather and time in San Francisco?"}]
response = make_query_and_print_result(MESSAGES)I'll check the current weather and time in San Francisco for you.
Tool: get_weather({'location': 'San Francisco, CA'})注意，尽管我们同时询问了天气和时间，Claude 却只返回了一个查询天气的工具调用？
让我们看看如果我们调用天气工具并继续会发生什么。
last_tool_call = response.content[1]
MESSAGES.append({"role": "assistant", "content": response.content})
MESSAGES.append(
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": last_tool_call.id,
"content": process_tool_call(response.content[1].name, response.content[1].input),
}
],
}
)
response = make_query_and_print_result(MESSAGES)Tool: get_time({'location': 'San Francisco, CA'})现在注意 Claude 发起了第二次工具调用来获取时间。虽然这在技术上确实是立即发生的，但可能是浪费的，因为它需要”来回”交互——首先 Claude 请求天气，然后我们需要处理它，然后 Claude 再请求时间，现在我们又要处理_那个_。
Claude 仍然会对结果做正确的事情，但鼓励 Claude 在一次调用中使用两者可能是有益的，这样我们可以同时处理。
引入批量工具
让我们引入一个 batch_tool，让 Claude 有机会使用它将多个工具调用合并为一个。
import json
batch_tool = {
"name": "batch_tool",
"description": "Invoke multiple other tool calls simultaneously",
"input_schema": {
"type": "object",
"properties": {
"invocations": {
"type": "array",
"description": "The tool calls to invoke",
"items": {
"types": "object",
"properties": {
"name": {
"types": "string",
"description": "The name of the tool to invoke",
},
"arguments": {
"types": "string",
"description": "The arguments to the tool",
},
},
"required": ["name", "arguments"],
},
}
},
"required": ["invocations"],
},
}
def process_tool_with_maybe_batch(tool_name, tool_input):
if tool_name == "batch_tool":
results = []
for invocation in tool_input["invocations"]:
results.append(
process_tool_call(invocation["name"], json.loads(invocation["arguments"]))
)
return "\n".join(results)
else:
return process_tool_call(tool_name, tool_input)现在让我们尝试为 Claude 提供现有的天气和时间工具，以及这个新的批量工具，看看当我们执行需要天气和时间的查询时会发生什么。
MESSAGES = [{"role": "user", "content": "What's the weather and time in San Francisco?"}]
response = make_query_and_print_result(MESSAGES, tools=[weather_tool, time_tool, batch_tool])I can help you check both the weather and the time in San Francisco. Let me get that information for you right away.
Tool: batch_tool({'invocations': [{'name': 'get_weather', 'arguments': '{"location": "San Francisco, CA"}'}, {'name': 'get_time', 'arguments': '{"location": "San Francisco, CA"}'}]})注意这次 Claude 使用了批量工具，一次性查询了时间和天气。这使我们能够同时处理它们，从而可能改善结果的整体延迟。
last_tool_call = response.content[1]
MESSAGES.append({"role": "assistant", "content": response.content})
MESSAGES.append(
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": last_tool_call.id,
"content": process_tool_with_maybe_batch(
response.content[1].name, response.content[1].input
),
}
],
}
)
response = make_query_and_print_result(MESSAGES)Here's the information you requested:
Weather in San Francisco, CA: 72 degrees and sunny
Time in San Francisco, CA: 12:32 PM
Is there anything else you'd like to know about San Francisco?
Tip
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。