使用扩展思考构建

weather_tool = {
    "name": "get_weather",
    "description": "获取某个位置的当前天气",
    "input_schema": {
        "type": "object",
        "properties": {
            "location": {"type": "string"}
        },
        "required": ["location"]
    }
}

# 第一个请求 - Claude用思考和工具请求回应
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[weather_tool],
    messages=[
        {"role": "user", "content": "巴黎的天气如何？"}
    ]
)

# 提取思考块和工具使用块
thinking_block = next((block for block in response.content
                      if block.type == 'thinking'), None)
tool_use_block = next((block for block in response.content
                      if block.type == 'tool_use'), None)

# 调用您的实际天气API，这里是您实际API调用的地方
# 让我们假设这是我们得到的回应
weather_data = {"temperature": 88}

# 第二个请求 - 包含思考块和工具结果
# 响应中不会生成新的思考块
continuation = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[weather_tool],
    messages=[
        {"role": "user", "content": "巴黎的天气如何？"},
        # 注意thinking_block和tool_use_block都被传入
        # 如果不传入，会引发错误
        {"role": "assistant", "content": [thinking_block, tool_use_block]},
        {"role": "user", "content": [{
            "type": "tool_result",
            "tool_use_id": tool_use_block.id,
            "content": f"当前温度：{weather_data['temperature']}°F"
        }]}
    ]
)

import anthropic

client = anthropic.Anthropic()

# 定义工具
calculator_tool = {
    "name": "calculator",
    "description": "执行数学计算",
    "input_schema": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "要计算的数学表达式"
            }
        },
        "required": ["expression"]
    }
}

database_tool = {
    "name": "database_query",
    "description": "查询产品数据库",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "要执行的SQL查询"
            }
        },
        "required": ["query"]
    }
}

# 第一个请求 - Claude在所有工具调用之前思考一次
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[calculator_tool, database_tool],
    messages=[{
        "role": "user",
        "content": "如果我们以每个50美元的价格销售150个产品A，总收入是多少，这与我们从数据库中的平均月收入相比如何？"
    }]
)

# 响应包括思考后跟工具使用
# 注意：Claude在开始时思考一次，然后做出所有工具决策
print("第一个响应：")
for block in response.content:
    if block.type == "thinking":
        print(f"思考（总结）：{block.thinking}")
    elif block.type == "tool_use":
        print(f"工具使用：{block.name}，输入{block.input}")
    elif block.type == "text":
        print(f"文本：{block.text}")

# 您将执行工具并返回结果...
# 获得两个工具结果后，Claude直接响应而不进行额外思考

import anthropic

client = anthropic.Anthropic()

# 与之前相同的工具定义
calculator_tool = {
    "name": "calculator",
    "description": "执行数学计算",
    "input_schema": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "要计算的数学表达式"
            }
        },
        "required": ["expression"]
    }
}

database_tool = {
    "name": "database_query",
    "description": "查询产品数据库",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "要执行的SQL查询"
            }
        },
        "required": ["query"]
    }
}

# 启用交错思考的第一个请求
response = client.beta.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[calculator_tool, database_tool],
    betas=["interleaved-thinking-2025-05-14"],
    messages=[{
        "role": "user",
        "content": "如果我们以每个50美元的价格销售150个产品A，总收入是多少，这与我们从数据库中的平均月收入相比如何？"
    }]
)

print("初始响应：")
thinking_blocks = []
tool_use_blocks = []

for block in response.content:
    if block.type == "thinking":
        thinking_blocks.append(block)
        print(f"思考：{block.thinking}")
    elif block.type == "tool_use":
        tool_use_blocks.append(block)
        print(f"工具使用：{block.name}，输入{block.input}")
    elif block.type == "text":
        print(f"文本：{block.text}")

# 第一个工具结果（计算器）
calculator_result = "7500"  # 150 * 50

# 继续第一个工具结果
response2 = client.beta.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[calculator_tool, database_tool],
    betas=["interleaved-thinking-2025-05-14"],
    messages=[
        {
            "role": "user",
            "content": "如果我们以每个50美元的价格销售150个产品A，总收入是多少，这与我们从数据库中的平均月收入相比如何？"
        },
        {
            "role": "assistant",
            "content": [thinking_blocks[0], tool_use_blocks[0]]
        },
        {
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use_blocks[0].id,
                "content": calculator_result
            }]
        }
    ]
)

print("\n计算器结果后：")
# 使用交错思考，Claude可以在决定查询数据库之前思考计算器结果
for block in response2.content:
    if block.type == "thinking":
        thinking_blocks.append(block)
        print(f"交错思考：{block.thinking}")
    elif block.type == "tool_use":
        tool_use_blocks.append(block)
        print(f"工具使用：{block.name}，输入{block.input}")

# 第二个工具结果（数据库）
database_result = "5200"  # 示例平均月收入

# 继续第二个工具结果
response3 = client.beta.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    tools=[calculator_tool, database_tool],
    betas=["interleaved-thinking-2025-05-14"],
    messages=[
        {
            "role": "user",
            "content": "如果我们以每个50美元的价格销售150个产品A，总收入是多少，这与我们从数据库中的平均月收入相比如何？"
        },
        {
            "role": "assistant",
            "content": [thinking_blocks[0], tool_use_blocks[0]]
        },
        {
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use_blocks[0].id,
                "content": calculator_result
            }]
        },
        {
            "role": "assistant",
            "content": thinking_blocks[1:] + tool_use_blocks[1:]
        },
        {
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use_blocks[1].id,
                "content": database_result
            }]
        }
    ]
)

print("\n数据库结果后：")
# 使用交错思考，Claude可以在制定最终响应之前思考两个结果
for block in response3.content:
    if block.type == "thinking":
        print(f"最终思考：{block.thinking}")
    elif block.type == "text":
        print(f"最终响应：{block.text}")

from anthropic import Anthropic
import requests
from bs4 import BeautifulSoup

client = Anthropic()

def fetch_article_content(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')

    # 移除脚本和样式元素
    for script in soup(["script", "style"]):
        script.decompose()

    # 获取文本
    text = soup.get_text()

    # 分解为行并移除每行的前导和尾随空格
    lines = (line.strip() for line in text.splitlines())
    # 将多标题分解为每行一个
    chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
    # 删除空行
    text = '\n'.join(chunk for chunk in chunks if chunk)

    return text

# 获取文章内容
book_url = "https://www.gutenberg.org/cache/epub/1342/pg1342.txt"
book_content = fetch_article_content(book_url)
# 仅使用足够的文本进行缓存（前几章）
LARGE_TEXT = book_content[:5000]

SYSTEM_PROMPT=[
    {
        "type": "text",
        "text": "您是一个负责文学分析的AI助手。请仔细分析以下文本。",
    },
    {
        "type": "text",
        "text": LARGE_TEXT,
        "cache_control": {"type": "ephemeral"}
    }
]

MESSAGES = [
    {
        "role": "user",
        "content": "分析这段文字的语调。"
    }
]

# 第一个请求 - 建立缓存
print("第一个请求 - 建立缓存")
response1 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000
    },
    system=SYSTEM_PROMPT,
    messages=MESSAGES
)

print(f"第一个响应使用量：{response1.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response1.content
})
MESSAGES.append({
    "role": "user",
    "content": "分析这段文字中的人物。"
})
# 第二个请求 - 相同的思考参数（预期缓存命中）
print("\n第二个请求 - 相同的思考参数（预期缓存命中）")
response2 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000
    },
    system=SYSTEM_PROMPT,
    messages=MESSAGES
)

print(f"第二个响应使用量：{response2.usage}")

# 第三个请求 - 不同的思考参数（消息缓存未命中）
print("\n第三个请求 - 不同的思考参数（消息缓存未命中）")
response3 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 8000  # 更改了思考预算
    },
    system=SYSTEM_PROMPT,  # 系统提示保持缓存
    messages=MESSAGES  # 消息缓存失效
)

print(f"第三个响应使用量：{response3.usage}")

from anthropic import Anthropic
import requests
from bs4 import BeautifulSoup

client = Anthropic()

def fetch_article_content(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')

    # 移除脚本和样式元素
    for script in soup(["script", "style"]):
        script.decompose()

    # 获取文本
    text = soup.get_text()

    # 分解为行并移除每行的前导和尾随空格
    lines = (line.strip() for line in text.splitlines())
    # 将多标题分解为每行一个
    chunks = (phrase.strip() for line in lines for phrase in line.split("  "))
    # 删除空行
    text = '\n'.join(chunk for chunk in chunks if chunk)

    return text

# 获取文章内容
book_url = "https://www.gutenberg.org/cache/epub/1342/pg1342.txt"
book_content = fetch_article_content(book_url)
# 仅使用足够的文本进行缓存（前几章）
LARGE_TEXT = book_content[:5000]

# 没有系统提示 - 在消息中缓存
MESSAGES = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": LARGE_TEXT,
                "cache_control": {"type": "ephemeral"},
            },
            {
                "type": "text",
                "text": "分析这段文字的语调。"
            }
        ]
    }
]

# 第一个请求 - 建立缓存
print("第一个请求 - 建立缓存")
response1 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000
    },
    messages=MESSAGES
)

print(f"第一个响应使用量：{response1.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response1.content
})
MESSAGES.append({
    "role": "user",
    "content": "分析这段文字中的人物。"
})
# 第二个请求 - 相同的思考参数（预期缓存命中）
print("\n第二个请求 - 相同的思考参数（预期缓存命中）")
response2 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000  # 相同的思考预算
    },
    messages=MESSAGES
)

print(f"第二个响应使用量：{response2.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response2.content
})
MESSAGES.append({
    "role": "user",
    "content": "分析这段文字中的背景。"
})

# 第三个请求 - 不同的思考预算（预期缓存未命中）
print("\n第三个请求 - 不同的思考预算（预期缓存未命中）")
response3 = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 8000  # 不同的思考预算破坏缓存
    },
    messages=MESSAGES
)

print(f"第三个响应使用量：{response3.usage}")

import anthropic

client = anthropic.Anthropic()

# 使用触发编辑思考的特殊提示（仅用于演示目的）
response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    messages=[{
        "role": "user",
        "content": "ANTHROPIC_MAGIC_STRING_TRIGGER_REDACTED_THINKING_46C9A13E193C177646C7398A98432ECCCE4C1253D5E2D82641AC0E52CC2876CB"
    }]
)

# 识别编辑的思考块
has_redacted_thinking = any(
    block.type == "redacted_thinking" for block in response.content
)

if has_redacted_thinking:
    print("响应包含编辑的思考块")
    # 这些块在后续请求中仍然可用

    # 提取所有块（编辑和非编辑的）
    all_thinking_blocks = [
        block for block in response.content
        if block.type in ["thinking", "redacted_thinking"]
    ]

    # 当传递到后续请求时，包含所有块而不修改
    # 这保持了Claude推理的完整性

    print(f"总共找到{len(all_thinking_blocks)}个思考块")
    print(f"这些块仍然作为输出token计费")