以下是翻译结果:

在 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_weatherget_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?