扩展思考与工具使用

目录

本笔记本演示了如何将 Claude 3.7 Sonnet 的扩展思考功能与工具结合使用。扩展思考功能允许您在 Claude 提供最终答案之前查看其逐步思考过程,从而提高其决定使用哪些工具以及如何解释工具结果的透明度。

在使用扩展思考与工具使用时,模型将在发出工具请求之前显示其思考过程,但在收到工具结果后不会重复思考过程。在下一个非 tool_resultuser 回合之前,Claude 不会输出另一个思考块。有关扩展思考的更多信息,请参阅我们的文档

设置

首先,让我们安装必要的软件包并设置我们的环境。

%pip install anthropic
import anthropic
import os
import json

# 模型和令牌预算的全局变量
MODEL_NAME = "claude-3-7-sonnet-20250219"
MAX_TOKENS = 4000
THINKING_BUDGET_TOKENS = 2000

# 将您的 API 密钥设置为环境变量或直接设置
# os.environ["ANTHROPIC_API_KEY"] = "your_api_key_here"

# 初始化客户端
client = anthropic.Anthropic()

# 辅助函数
def print_thinking_response(response):
    """以思考块的形式美观地打印消息响应。"""
    print("\n==== 完整响应 ====")
    for block in response.content:
        if block.type == "thinking":
            print("\n🧠 思考块:")
            # 为便于阅读,显示截断的思考内容
            print(block.thinking[:500] + "..." if len(block.thinking) > 500 else block.thinking)
            print(f"\n[签名可用: {bool(getattr(block, 'signature', None))}]")
            if hasattr(block, 'signature') and block.signature:
                print(f"[签名(前 50 个字符): {block.signature[:50]}...]")
        elif block.type == "redacted_thinking":
            print("\n🔒 已编辑的思考块:")
            print(f"[数据长度: {len(block.data) if hasattr(block, 'data') else 'N/A'}]")
        elif block.type == "text":
            print("\n✓ 最终答案:")
            print(block.text)

    print("\n==== 响应结束 ====")

def count_tokens(messages, tools=None):
    """计算给定消息列表(可选工具)的令牌数。"""
    if tools:
        response = client.messages.count_tokens(
            model=MODEL_NAME,
            messages=messages,
            tools=tools
        )
    else:
        response = client.messages.count_tokens(
            model=MODEL_NAME,
            messages=messages
        )
    return response.input_tokens

带思考的单次工具调用

此示例演示了如何结合思考并进行单次工具调用,使用一个模拟的天气工具。

def tool_use_with_thinking():
    # 定义一个天气工具
    tools = [
        {
            "name": "weather",
            "description": "获取某地的当前天气信息。",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "获取天气信息的地点。"
                    }
                },
                "required": ["location"]
            }
        }
    ]

    def weather(location):
        # 模拟天气数据
        weather_data = {
            "纽约": {"temperature": 72, "condition": "晴朗"},
            "伦敦": {"temperature": 62, "condition": "多云"},
            "东京": {"temperature": 80, "condition": "部分多云"},
            "巴黎": {"temperature": 65, "condition": "下雨"},
            "悉尼": {"temperature": 85, "condition": "晴朗"},
            "柏林": {"temperature": 60, "condition": "有雾"},
        }

        return weather_data.get(location, {"error": f"无法获取 {location} 的天气数据"})

    # 带有工具使用和思考的初始请求
    response = client.messages.create(
        model=MODEL_NAME,
        max_tokens=MAX_TOKENS,
        thinking={
            "type": "enabled",
            "budget_tokens": THINKING_BUDGET_TOKENS
        },
        tools=tools,
        messages=[{
            "role": "user",
            "content": "今天巴黎天气怎么样?"
        }]
    )

    # 初始响应的详细诊断输出
    print("\n=== 初始响应 ===")
    print(f"响应 ID: {response.id}")
    print(f"停止原因: {response.stop_reason}")
    print(f"模型: {response.model}")
    print(f"内容块: {len(response.content)} 块")

    for i, block in enumerate(response.content):
        print(f"\n块 {i+1}: 类型 = {block.type}")
        if block.type == "thinking":
            print(f"思考内容: {block.thinking[:150]}...")
            print(f"签名可用: {bool(getattr(block, 'signature', None))}")
        elif block.type == "text":
            print(f"文本内容: {block.text}")
        elif block.type == "tool_use":
            print(f"工具: {block.name}")
            print(f"工具输入: {block.input}")
            print(f"工具 ID: {block.id}")
    print("=== 初始响应结束 ===\n")

    # 将思考块提取到对话历史中
    assistant_blocks = []
    for block in response.content:
        if block.type in ["thinking", "redacted_thinking", "tool_use"]:
            assistant_blocks.append(block)

    # 处理工具使用(如果需要)
    full_conversation = [{
        "role": "user",
        "content": "今天巴黎天气怎么样?"
    }]

    if response.stop_reason == "tool_use":
        # 添加包含思考块和工具使用的整个助手响应
        full_conversation.append({
            "role": "assistant",
            "content": assistant_blocks
        })

        # 查找 tool_use 块
        tool_use_block = next((block for block in response.content if block.type == "tool_use"), None)
        if tool_use_block:
            # 执行工具
            print(f"\n=== 执行工具 ===")
            print(f"工具名称: {tool_use_block.name}")
            print(f"要检查的位置: {tool_use_block.input['location']}")
            tool_result = weather(tool_use_block.input["location"])
            print(f"结果: {tool_result}")
            print("=== 工具执行完成 ===\n")

            # 将工具结果添加到对话中
            full_conversation.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_use_block.id,
                    "content": json.dumps(tool_result)
                }]
            })

            # 使用相同的思考配置继续对话
            print("\n=== 发送带有工具结果的后续请求 ===")
            response = client.messages.create(
                model=MODEL_NAME,
                max_tokens=MAX_TOKENS,
                thinking={
                    "type": "enabled",
                    "budget_tokens": THINKING_BUDGET_TOKENS
                },
                tools=tools,
                messages=full_conversation
            )
            print(f"收到后续响应。停止原因: {response.stop_reason}")

    print_thinking_response(response)

# 运行示例
tool_use_with_thinking()
=== 初始响应 ===
响应 ID: msg_01NhR4vE9nVh2sHs5fXbzji8
停止原因: tool_use
模型: claude-3-7-sonnet-20250219
内容块: 3 块

块 1: 类型 = thinking
思考内容: 用户正在询问巴黎当前的天气。我可以使用 `weather` 函数来获取此信息。

`weather` 函数需要一个“地点”参数...
签名可用: True

块 2: 类型 = text
文本内容: 我会立即为您查询巴黎的天气。

块 3: 类型 = tool_use
工具: weather
工具输入: {'location': 'Paris'}
工具 ID: toolu_01WaeSyitUGJFaaPe68cJuEv
=== 初始响应结束 ===


=== 执行工具 ===
工具名称: weather
要检查的位置: Paris
结果: {'temperature': 65, 'condition': 'Rainy'}
=== 工具执行完成 ===


=== 发送带有工具结果的后续请求 ===
收到后续响应。停止原因: end_turn

==== 完整响应 ====

✓ 最终答案:
目前巴黎天气为 65°F(18°C),有雨。如果您要出门,可能需要带把伞!

==== 响应结束 ====

带思考的多项工具调用

此示例演示了如何处理多项工具调用,例如模拟的新闻和天气服务,同时观察思考过程。

def multiple_tool_calls_with_thinking():
    # 定义工具
    tools = [
        {
            "name": "weather",
            "description": "获取某地的当前天气信息。",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "获取天气信息的地点。"
                    }
                },
                "required": ["location"]
            }
        },
        {
            "name": "news",
            "description": "获取某个主题的最新新闻标题。",
            "input_schema": {
                "type": "object",
                "properties": {
                    "topic": {
                        "type": "string",
                        "description": "获取新闻的主题。"
                    }
                },
                "required": ["topic"]
            }
        }
    ]

    def weather(location):
        # 模拟天气数据
        weather_data = {
            "纽约": {"temperature": 72, "condition": "晴朗"},
            "伦敦": {"temperature": 62, "condition": "多云"},
            "东京": {"temperature": 80, "condition": "部分多云"},
            "巴黎": {"temperature": 65, "condition": "下雨"},
            "悉尼": {"temperature": 85, "condition": "晴朗"},
            "柏林": {"temperature": 60, "condition": "有雾"},
        }

        return weather_data.get(location, {"error": f"无法获取 {location} 的天气数据"})

    def news(topic):
        # 模拟新闻数据
        news_data = {
            "technology": [
                "研究实验室宣布了新的人工智能突破",
                "科技公司发布了最新的智能手机型号",
                "量子计算达到了里程碑式的成就"
            ],
            "sports": [
                "当地球队赢得了冠军赛",
                "明星球员签署了创纪录的合同",
                "奥委会宣布了 2036 年的举办城市"
            ],
            "weather": [
                "风暴系统正在大西洋发展",
                "欧洲各地记录到创纪录的高温",
                "气候科学家发布了新的研究成果"
            ]
        }

        return {"headlines": news_data.get(topic.lower(), ["此主题没有新闻可用"])}

    # 初始请求
    response = client.messages.create(
        model=MODEL_NAME,
        max_tokens=MAX_TOKENS,
        thinking={
                "type": "enabled",
                "budget_tokens": THINKING_BUDGET_TOKENS
        },
        tools=tools,
        messages=[{
            "role": "user",
            "content": "伦敦的天气怎么样?另外,能告诉我最新的科技新闻吗?"
        }]
    )

    # 打印初始响应的详细信息
    print("\n=== 初始响应 ===")
    print(f"响应 ID: {response.id}")
    print(f"停止原因: {response.stop_reason}")
    print(f"模型: {response.model}")
    print(f"内容块: {len(response.content)} 块")

    # 打印每个内容块
    for i, block in enumerate(response.content):
        print(f"\n块 {i+1}: 类型 = {block.type}")
        if block.type == "thinking":
            print(f"思考内容: {block.thinking[:150]}...")
            print(f"签名可用: {bool(getattr(block, 'signature', None))}")
        elif block.type == "text":
            print(f"文本内容: {block.text}")
        elif block.type == "tool_use":
            print(f"工具: {block.name}")
            print(f"工具输入: {block.input}")
            print(f"工具 ID: {block.id}")
    print("=== 初始响应结束 ===\n")

    # 处理可能的多项工具调用
    full_conversation = [{
        "role": "user",
        "content": "伦敦的天气怎么样?另外,能告诉我最新的科技新闻吗?"
    }]

    # 跟踪迭代次数以进行多轮工具使用
    iteration = 0

    while response.stop_reason == "tool_use":
        iteration += 1
        print(f"\n=== 工具使用迭代 {iteration} ===")

        # 提取思考块和工具使用以包含在对话历史中
        assistant_blocks = []
        for block in response.content:
            if block.type in ["thinking", "redacted_thinking", "tool_use"]:
                assistant_blocks.append(block)

        # 添加包含思考块和工具使用的助手响应
        full_conversation.append({
            "role": "assistant",
            "content": assistant_blocks
        })

        # 查找 tool_use 块
        tool_use_block = next((block for block in response.content if block.type == "tool_use"), None)
        if tool_use_block:
            print(f"\n=== 执行工具 ===")
            print(f"工具名称: {tool_use_block.name}")

            # 执行相应的工具
            if tool_use_block.name == "weather":
                print(f"要检查的位置: {tool_use_block.input['location']}")
                tool_result = weather(tool_use_block.input["location"])
            elif tool_use_block.name == "news":
                print(f"要检查的主题: {tool_use_block.input['topic']}")
                tool_result = news(tool_use_block.input["topic"])
            else:
                tool_result = {"error": "未知工具"}

            print(f"结果: {tool_result}")
            print("=== 工具执行完成 ===\n")

            # 将工具结果添加到对话中
            full_conversation.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": tool_use_block.id,
                    "content": json.dumps(tool_result)
                }]
            })

            # 继续对话
            print("\n=== 发送带有工具结果的后续请求 ===")
            response = client.messages.create(
                model=MODEL_NAME,
                max_tokens=MAX_TOKENS,
                thinking={
                        "type": "enabled",
                        "budget_tokens": THINKING_BUDGET_TOKENS
                },
                tools=tools,
                messages=full_conversation
            )

            # 打印后续响应详情
            print(f"\n=== 后续响应(迭代 {iteration})===")
            print(f"响应 ID: {response.id}")
            print(f"停止原因: {response.stop_reason}")
            print(f"内容块: {len(response.content)} 块")

            for i, block in enumerate(response.content):
                print(f"\n块 {i+1}: 类型 = {block.type}")
                if block.type == "thinking":
                    print(f"思考内容预览: {block.thinking[:100]}...")
                elif block.type == "text":
                    print(f"文本内容预览: {block.text[:100]}...")
                elif block.type == "tool_use":
                    print(f"工具: {block.name}")
                    print(f"工具输入预览: {str(block.input)[:100]}")
            print(f"=== 后续响应结束(迭代 {iteration})===\n")

            if response.stop_reason != "tool_use":
                print("\n=== 最终响应 ===")
                print_thinking_response(response)
                print("=== 最终响应结束 ===")
        else:
            print("响应中未找到 tool_use 块。")
            break

# 运行示例
multiple_tool_calls_with_thinking()
=== 初始响应 ===
响应 ID: msg_01VwqpBMARVoTP1H8Ytvmvsb
停止原因: tool_use
模型: claude-3-7-sonnet-20250219
内容块: 3 块

块 1: 类型 = thinking
思考内容: 用户要求两项信息:

1. 伦敦的天气
2. 最新的科技新闻

让我看看我有什么可用的工具...
签名可用: True

块 2: 类型 = text
文本内容: 我会立即为您获取信息。

块 3: 类型 = tool_use
工具: weather
工具输入: {'location': 'London'}
工具 ID: toolu_016xHQWMR4JsKtWvH9nbsZyA
=== 初始响应结束 ===


=== 工具使用迭代 1 ===

=== 执行工具 ===
工具名称: weather
要检查的位置: London
结果: {'temperature': 62, 'condition': 'Cloudy'}
=== 工具执行完成 ===


=== 发送带有工具结果的后续请求 ===

=== 后续响应(迭代 1)===
响应 ID: msg_01EhR96Z2Z2t5EDhuWeodUod
停止原因: tool_use
内容块: 1 块

块 1: 类型 = tool_use
工具: news
工具输入预览: {'topic': 'technology'}
=== 后续响应结束(迭代 1)===


=== 工具使用迭代 2 ===

=== 执行工具 ===
工具名称: news
要检查的主题: technology
结果: {'headlines': ['研究实验室宣布了新的人工智能突破', '科技公司发布了最新的智能手机型号', '量子计算达到了里程碑式的成就']}
=== 工具执行完成 ===


=== 发送带有工具结果的后续请求 ===

=== 后续响应(迭代 2)===
响应 ID: msg_01WUEfC4UxPFaJaktjVDMJEN
停止原因: end_turn
内容块: 1 块

块 1: 类型 = text
文本内容预览: 这是您要求的信息:

## 伦敦天气
目前,伦敦天气为 62°F,多云。
...
=== 后续响应结束(迭代 2)===


=== 最终响应 ===

==== 完整响应 ====

✓ 最终答案:
这是您要求的信息:

## 伦敦天气
目前,伦敦天气为 62°F,多云。

## 最新科技新闻头条

- 研究实验室宣布了新的人工智能突破
- 科技公司发布了最新的智能手机型号
- 量子计算达到了里程碑式的成就

==== 响应结束 ====
=== 最终响应结束 ===

保留思考块

在处理扩展思考和工具时,请务必:

  1. 保留思考块签名:每个思考块都包含一个加密签名,用于验证对话上下文。在将思考块传递回 Claude 时,必须包含这些签名。

  2. 避免修改先前上下文:在提交包含工具结果的新请求时,如果修改了任何先前的内容(包括思考块),API 将会拒绝。

  3. 同时处理思考和已编辑的思考块:这两种类型的块都必须保留在对话历史中,即使已编辑块的内容无法被人类阅读。

有关不使用工具的扩展思考的更多详细信息,请参阅主要的“扩展思考”笔记本。

def thinking_block_preservation_example():
    # 定义一个简单的天气工具
    tools = [
        {
            "name": "weather",
            "description": "获取某地的当前天气信息。",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "获取天气信息的地点。"
                    }
                },
                "required": ["location"]
            }
        }
    ]

    def weather(location):
        # 模拟天气数据
        weather_data = {
            "纽约": {"temperature": 72, "condition": "晴朗"},
            "伦敦": {"temperature": 62, "condition": "多云"},
            "东京": {"temperature": 80, "condition": "部分多云"},
            "巴黎": {"temperature": 65, "condition": "下雨"},
            "悉尼": {"temperature": 85, "condition": "晴朗"},
            "柏林": {"temperature": 60, "condition": "有雾"},
        }

        return weather_data.get(location, {"error": f"无法获取 {location} 的天气数据"})

    # 带有工具使用和思考的初始请求
    response = client.messages.create(
        model=MODEL_NAME,
        max_tokens=MAX_TOKENS,
        thinking={
            "type": "enabled",
            "budget_tokens": THINKING_BUDGET_TOKENS
        },
        tools=tools,
        messages=[{
            "role": "user",
            "content": "柏林现在天气怎么样?"
        }]
    )

    # 从响应中提取块
    thinking_blocks = [b for b in response.content if b.type == "thinking"]
    tool_use_blocks = [b for b in response.content if b.type == "tool_use"]

    print("\n=== 初始响应 ===")
    print(f"响应包含:")
    print(f"- {len(thinking_blocks)} 个思考块")
    print(f"- {len(tool_use_blocks)} 个工具使用块")

    # 检查是否触发了工具使用
    if tool_use_blocks:
        tool_block = tool_use_blocks[0]
        print(f"\n调用的工具: {tool_block.name}")
        print(f"要检查的位置: {tool_block.input['location']}")

        # 执行工具
        tool_result = weather(tool_block.input["location"])
        print(f"工具结果: {tool_result}")

        # 首先,我们尝试不包含思考块
        print("\n=== 测试 1:不包含思考块 ===")
        try:
            # 注意我们只包含 tool_use 块,不包含思考块
            partial_blocks = tool_use_blocks

            incomplete_response = client.messages.create(
                model=MODEL_NAME,
                max_tokens=MAX_TOKENS,
                thinking={
                        "type": "enabled",
                        "budget_tokens": THINKING_BUDGET_TOKENS
                },
                tools=tools,
                messages=[
                    {"role": "user", "content": "柏林现在天气怎么样?"},
                    {"role": "assistant", "content": partial_blocks},
                    {"role": "user", "content": [{
                        "type": "tool_result",
                        "tool_use_id": tool_block.id,
                        "content": json.dumps(tool_result)
                    }]}
                ]
            )
            print("成功:在没有思考块的情况下收到了响应(不符合预期)")
        except Exception as e:
            print(f"错误: {e}")
            print("这表明必须保留思考块")

        # 现在尝试包含思考块(正确方法)
        print("\n=== 测试 2:包含思考块(正确方法)===")
        try:
            # 包含响应中的所有块
            complete_blocks = thinking_blocks + tool_use_blocks

            complete_response = client.messages.create(
                model=MODEL_NAME,
                max_tokens=MAX_TOKENS,
                thinking={
                    "type": "enabled",
                    "budget_tokens": THINKING_BUDGET_TOKENS
                },
                tools=tools,
                messages=[
                    {"role": "user", "content": "柏林现在天气怎么样?"},
                    {"role": "assistant", "content": complete_blocks},
                    {"role": "user", "content": [{
                        "type": "tool_result",
                        "tool_use_id": tool_block.id,
                        "content": json.dumps(tool_result)
                    }]}
                ]
            )
            print("成功:在包含思考块的情况下收到了响应")

            # 检查第二个响应是否包含思考块
            second_thinking = [b for b in complete_response.content if b.type == "thinking"]
            second_text = [b for b in complete_response.content if b.type == "text"]

            print(f"\n第二个响应包含:")
            print(f"- {len(second_thinking)} 个思考块")
            print(f"- {len(second_text)} 个文本块")

            if second_text:
                print(f"\n最终答案: {second_text[0].text}")

            print("\n注意:工具使用后的第二个响应不包含思考块。")
            print("这是预期的行为 - 思考在工具使用前显示,但在收到工具结果后不显示。")

        except Exception as e:
            print(f"错误: {e}")

# 取消注释以运行示例
thinking_block_preservation_example()
=== 初始响应 ===
响应包含:

- 1 个思考块
- 1 个工具使用块

调用的工具: weather
要检查的位置: Berlin
工具结果: {'temperature': 60, 'condition': 'Foggy'}

=== 测试 1:不包含思考块 ===
错误: 错误代码: 400 - {'type': 'error', 'error': {'type': 'invalid_request_error', 'message': 'messages.1.content.0.type: 预期为 `thinking` 或 `redacted_thinking`,但发现为 `tool_use`。当启用 `thinking` 时,最后的 `assistant` 消息必须以思考块开头(位于最后设置的 `tool_use` 和 `tool_result` 块之前)。我们建议您包含之前的思考块。要避免此要求,请禁用 `thinking`。请查阅我们的文档 https://docs.anthropic.com/en/docs/build-with-claude/extended-thinking'}}
这表明必须保留思考块

=== 测试 2:包含思考块(正确方法)===
成功:在包含思考块的情况下收到了响应

第二个响应包含:

- 0 个思考块
- 1 个文本块

最终答案: 目前柏林天气为 60°F(约 15.5°C),有雾。

注意:工具使用后的第二个响应不包含思考块。
这是预期的行为 - 思考在工具使用前显示,但在收到工具结果后不显示。

结论

本笔记本展示了如何将 Claude 的扩展思考功能与工具使用相结合。主要优点包括:

  1. 使用工具时 Claude 思考过程的透明度
  2. Claude 决定何时使用工具与内部知识的可见性
  3. 对涉及多项工具调用的多步任务的更好理解
  4. 对 Claude 如何解释工具结果并将其纳入响应的洞察

在使用扩展思考与工具结合使用时,请记住:

  • 为复杂任务设置适当的思考预算(最少 1,024 个令牌)
  • 在传递工具结果时始终保留思考块及其签名
  • 在对话历史中包含常规和已编辑的思考块
  • 确保系统提示、工具和思考配置在调用之间匹配
  • 预期工具结果回合不包含额外的思考块
  • 工具使用和思考结合使用会增加令牌使用量和响应时间