使用扩展思维进行构建

import anthropic

client = anthropic.Anthropic()

# 使用触发被编辑思维的特殊提示(仅用于演示目的)
response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 16000
    },
    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("Response contains redacted thinking blocks")
    # 这些块在后续请求中仍然可用

    # 提取所有块(包括被编辑和未被编辑的)
    all_thinking_blocks = [
        block for block in response.content
        if block.type in ["thinking", "redacted_thinking"]
    ]

    # 在传递给后续请求时,包含所有未修改的块
    # 这保持了 Claude 推理的完整性

    print(f"Found {len(all_thinking_blocks)} thinking blocks total")
    print(f"These blocks are still billable as output tokens")

curl https://api.anthropic.com/v1/messages \
     --header "x-api-key: $ANTHROPIC_API_KEY" \
     --header "anthropic-version: 2023-06-01" \
     --header "content-type: application/json" \
     --data \
'{
    "model": "claude-3-7-sonnet-20250219",
    "max_tokens": 20000,
    "stream": true,
    "thinking": {
        "type": "enabled",
        "budget_tokens": 16000
    },
    "messages": [
        {
            "role": "user",
            "content": "ANTHROPIC_MAGIC_STRING_TRIGGER_REDACTED_THINKING_46C9A13E193C177646C7398A98432ECCCE4C1253D5E2D82641AC0E52CC2876CB"
        }
    ]
}'

import anthropic

client = anthropic.Anthropic()

# 第一条启用扩展思维的消息
first_message = [{
    "role": "user",
    "content": "Explain quantum entanglement"
}]

# 获取带有扩展思维的第一个响应
response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 16000
    },
    messages=first_message
)

# 计算第一次交换的令牌数(仅用户输入)
first_count = client.messages.count_tokens(
    model="claude-3-7-sonnet-20250219",
    messages=first_message
)
print(f"First message input tokens: {first_count.input_tokens}")

# 准备带有之前响应的第二次交换(包括思维块)
second_message = first_message + [
    {"role": "assistant", "content": response.content},
    {"role": "user", "content": "How does this relate to quantum computing?"}
]

# 计算第二次交换的令牌数
second_count = client.messages.count_tokens(
    model="claude-3-7-sonnet-20250219",
    messages=second_message
)
print(f"Second message input tokens: {second_count.input_tokens}")

# 提取仅文本块进行比较
text_only_blocks = [block for block in response.content if block.type == "text"]
text_only_content = [{"type": "text", "text": block.text} for block in text_only_blocks]

# 创建仅包含文本块的消息进行比较
text_only_message = first_message + [
    {"role": "assistant", "content": text_only_content},
    {"role": "user", "content": "How does this relate to quantum computing?"}
]

# 计算此仅文本消息的令牌数
text_only_count = client.messages.count_tokens(
    model="claude-3-7-sonnet-20250219",
    messages=text_only_message
)

# 比较令牌计数以证明之前的思维块不被计数
print(f"Are they equal? {second_count.input_tokens == text_only_count.input_tokens}")

from anthropic import Anthropic
import requests
from bs4 import BeautifulSoup

client = anthropic.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": "You are an AI assistant that is tasked with literary analysis. Analyze the following text carefully.",
    },
    {
        "type": "text",
        "text": LARGE_TEXT,
        "cache_control": {"type": "ephemeral"}
    }
]

MESSAGES = [
    {
        "role": "user",
        "content": "Analyze the tone of this passage."
    }
]

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

print(f"First response usage: {response1.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response1.content
})
MESSAGES.append({
    "role": "user",
    "content": "Analyze the characters in this passage."
})
# 第二个请求 - 相同的思维参数(预期缓存命中)
print("\nSecond request - same thinking parameters (cache hit expected)")
response2 = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000  # 相同的思维预算
    },
    system=SYSTEM_PROMPT,
    messages=MESSAGES
)

print(f"Second response usage: {response2.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response2.content
})
MESSAGES.append({
    "role": "user",
    "content": "Analyze the setting in this passage."
})

# 第三个请求 - 不同的思维预算(由于系统提示缓存预期缓存命中)
print("\nThird request - different thinking budget (cache hit expected)")
response3 = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 8000  # 不同的思维预算 - 仍然保持缓存!
    },
    system=SYSTEM_PROMPT,
    messages=MESSAGES
)

print(f"Third response usage: {response3.usage}")

from anthropic import Anthropic
import requests
from bs4 import BeautifulSoup

client = anthropic.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": "Analyze the tone of this passage."
            }
        ]
    }
]

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

print(f"First response usage: {response1.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response1.content
})
MESSAGES.append({
    "role": "user",
    "content": "Analyze the characters in this passage."
})
# 第二个请求 - 相同的思维参数(预期缓存命中)
print("\nSecond request - same thinking parameters (cache hit expected)")
response2 = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 4000  # 相同的思维预算
    },
    messages=MESSAGES
)

print(f"Second response usage: {response2.usage}")

MESSAGES.append({
    "role": "assistant",
    "content": response2.content
})
MESSAGES.append({
    "role": "user",
    "content": "Analyze the setting in this passage."
})

# 第三个请求 - 不同的思维预算(预期缓存未命中)
print("\nThird request - different thinking budget (cache miss expected)")
response3 = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 8000  # 不同的思维预算破坏缓存
    },
    messages=MESSAGES
)

print(f"Third response usage: {response3.usage}")

weather_tool = {
    "name": "get_weather",
    "description": "Get current weather for a location",
    "input_schema": {
        "type": "object",
        "properties": {
            "location": {"type": "string"}
        },
        "required": ["location"]
    }
}

# 第一个请求 - Claude 用思维和工具请求响应
response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 16000
    },
    tools=[weather_tool],
    messages=[
        {"role": "user", "content": "What's the weather in Paris?"}
    ]
)

# 提取思维块和工具使用块
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-3-7-sonnet-20250219",
    max_tokens=20000,
    thinking={
        "type": "enabled",
        "budget_tokens": 16000
    },
    tools=[weather_tool],
    messages=[
        {"role": "user", "content": "What's the weather in Paris?"},
        # 注意 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"Current temperature: {weather_data['temperature']}°F"
        }]}
    ]
)