Claude Prompt Caching 实战
最高降低 90% 的 API 成本

import anthropic

client = anthropic.Anthropic()

# 对话历史
conversation_history = []

def chat(user_message: str) -> str:
    """支持缓存的多轮对话函数"""
    conversation_history.append({
        "role": "user",
        "content": user_message
    })

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        # ✅ 自动缓存：只需这一个字段，无需修改 system/messages 结构
        betas=["prompt-caching-2024-07-31"],
        extra_body={
            "cache_control": {"type": "ephemeral"}  # 顶层缓存控制
        },
        system="你是一位有经验的软件架构师，擅长系统设计与代码审查。",
        messages=conversation_history
    )

    assistant_message = response.content[0].text
    conversation_history.append({
        "role": "assistant",
        "content": assistant_message
    })

    # 输出 Token 使用情况
    u = response.usage
    cache_hit = getattr(u, 'cache_read_input_tokens', 0)
    cache_write = getattr(u, 'cache_creation_input_tokens', 0)
    print(f"[缓存命中: {cache_hit} | 缓存写入: {cache_write} | 输出: {u.output_tokens}]")

    return assistant_message

# 模拟多轮对话
print(chat("请帮我设计一个高并发的消息队列系统"))
print(chat("这个系统应该如何处理消息丢失的问题？"))
print(chat("能给出 Go 语言的实现示例吗？"))

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你的系统提示...",
            # 使用 1 小时缓存（需要额外费用）
            "cache_control": {"type": "ephemeral", "ttl": 3600}
        }
    ],
    messages=[{"role": "user", "content": "用户问题"}]
)

tools = [
    {
        "name": "search_database",
        "description": "搜索产品数据库",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "category": {"type": "string"}
            },
            "required": ["query"]
        },
        # ✅ 工具定义也可以缓存
        "cache_control": {"type": "ephemeral"}
    }
    # 更多工具定义...（最多配合 3 个其他断点，总计不超过 4 个）
]

response = client.messages.create(
    model="claude-opus-4-6",
    max_tokens=1024,
    tools=tools,
    system=[{
        "type": "text",
        "text": "你是一位购物助手...",
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[{"role": "user", "content": "帮我找一款蓝牙耳机"}]
)

import asyncio
import anthropic

client = anthropic.Anthropic()

# 一份固定文档，需要用 N 个不同 Prompt 分析
document_content = "... 大型文档内容 ..."

analysis_tasks = [
    "提取所有人名和机构名",
    "总结文档的主要论点",
    "找出文中的数据和统计信息",
    "分析作者的写作风格",
    "列出所有引用的参考文献",
]

def analyze(task: str) -> dict:
    resp = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        system=[
            {
                "type": "text",
                "text": "你是专业文档分析助手。",
                "cache_control": {"type": "ephemeral"}
            },
            {
                "type": "text",
                "text": f"待分析文档：\n{document_content}",
                "cache_control": {"type": "ephemeral"}  # 核心：文档只写入一次
            }
        ],
        messages=[{"role": "user", "content": task}]
    )
    u = resp.usage
    return {
        "task": task,
        "result": resp.content[0].text,
        "cache_hit": getattr(u, 'cache_read_input_tokens', 0),
        "cache_miss": getattr(u, 'cache_creation_input_tokens', 0)
    }

# 串行执行（或改为并发执行）
results = [analyze(t) for t in analysis_tasks]
for r in results:
    print(f"任务: {r['task'][:20]}...")
    print(f"  缓存命中: {r['cache_hit']} tokens, 缓存写入: {r['cache_miss']} tokens\n")

def calculate_cache_savings(
    doc_tokens: int,
    daily_requests: int,
    model: str = "claude-opus-4-6"
):
    """计算 Prompt Caching 的成本节省"""
    pricing = {
        "claude-opus-4-6": {
            "input": 5.0,          # $5 per MTok
            "cache_write_5m": 6.25, # $6.25 per MTok
            "cache_read": 0.50,    # $0.50 per MTok
        }
    }
    p = pricing[model]
    MTok = 1_000_000

    # 不使用缓存
    cost_no_cache = (doc_tokens * daily_requests / MTok) * p["input"]

    # 使用缓存（每 5 分钟最多被使用一次，假设请求均匀分布 8 小时内）
    cache_writes_per_day = 8 * 60 / 5  # 96 次写入
    cache_reads_per_day = daily_requests - cache_writes_per_day

    cost_with_cache = (
        (doc_tokens * cache_writes_per_day / MTok) * p["cache_write_5m"] +
        (doc_tokens * cache_reads_per_day / MTok) * p["cache_read"]
    )

    saving = cost_no_cache - cost_with_cache
    saving_pct = saving / cost_no_cache * 100

    print(f"=== 文档规模: {doc_tokens:,} tokens | 每日请求: {daily_requests} 次 ===")
    print(f"不使用缓存日成本:  ${cost_no_cache:.4f}")
    print(f"使用缓存后日成本:  ${cost_with_cache:.4f}")
    print(f"每日节省:          ${saving:.4f}  ({saving_pct:.1f}%)")
    print(f"每月节省(30天):    ${saving * 30:.2f}")

# 测试不同场景
calculate_cache_savings(doc_tokens=10_000, daily_requests=100)
print()
calculate_cache_savings(doc_tokens=50_000, daily_requests=500)
print()
calculate_cache_savings(doc_tokens=100_000, daily_requests=1000)