以下是翻译结果:
在 Claude 3.7 Sonnet 上并行调用工具
Claude 3.7 Sonnet 可能不太可能在响应中进行并行工具调用,即使您没有设置 disable_parallel_tool_use
。为了解决这个问题,我们建议引入一个“批处理工具”,它可以充当元工具,同时包装对其他工具的调用。我们发现,如果存在此工具,模型将使用它来同时为您调用多个工具。
让我们看一下问题,并更详细地研究此解决方法。
from anthropic import Anthropic
client = Anthropic()
MODEL_NAME = "claude-3-7-sonnet-20250219"
执行具有多个工具调用的查询
请记住,Claude 的默认行为是允许并行工具调用。结合默认的 tool_choice
设置为 auto
,这意味着 Claude 可以调用指定的任何工具,或者在一个助手回合中调用多个工具。
让我们为 Claude 设置 get_weather
和 get_time
工具。
def get_weather(location):
# 假装获取天气,只返回一个固定值。
return f"The weather in {location} is 72 degrees and sunny."
def get_time(location):
# 假装获取时间,只返回一个固定值。
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?