Anthropic Prompt Caching 实战：Agent 开发者省下 60% Claude 账单

上个月我在排查一个客服 Agent 的运营成本时，发现了一个让人头皮发麻的数字：每天 $180 的 Claude API 费用。原因很简单——这个 Agent 每小时发起约 200 次调用，每次都带着相同的 8,000 token 系统提示词。每小时有 160 万个 token 在做重复劳动。接入 anthropic prompt caching 后，同样的负载降到了 $72/天，省了 60%，输出质量没有任何变化。

这篇教程会把完整实现过程拆开讲清楚。

要点速览： 在系统提示词的 content block 上加 cache_control: {"type": "ephemeral"}。对于每小时 200 次调用、8K 系统提示词的 Sonnet 5 Agent，每天省 $108。代码改动不超过 20 行。

Prompt Caching 的工作机制

核心逻辑：你告诉 API 哪些 prompt 内容是跨请求不变的，Anthropic 在服务端缓存这些 token 序列。后续请求如果前缀完全匹配，直接从缓存读取，不再重新处理。

请求流程：

第 1 次请求（缓存 MISS）：
  [系统提示词: 8K tokens] ──► cache_control: ephemeral
  [用户消息: 200 tokens]
  费用: 8K × 写入价格（正常价的 1.25 倍）+ 200 × 正常价

第 2~N 次请求（缓存 HIT）：
  [系统提示词: 8K tokens] ──► 从缓存读取
  [用户消息: 200 tokens]
  费用: 8K × 缓存读取价格（正常价的 0.1 倍）+ 200 × 正常价

定价规则：

• 缓存写入： 比正常输入贵 25%
• 缓存读取： 比正常输入便宜 90%
• 缓存有效期： 5 分钟（免费）或 1 小时（付费层级）
• 回本点： 只需 2 次缓存命中就能覆盖写入成本

最小 Token 门槛

不是什么内容都能缓存的：

• Sonnet 5 / Opus 5： 最少 1,024 tokens
• Haiku： 最少 2,048 tokens

如果你的系统提示词低于这个门槛，缓存不会生效。好消息是大多数 Agent 的系统提示词加上工具定义和行为规则，轻松超过 1,024 tokens。

什么场景适合，什么场景不适合

场景	调用频率	静态前缀大小	预计节省	结论
Agent 循环（客服机器人）	200次/小时	8K tokens	60-65%	✅ 非常适合
RAG 流水线（固定指令 + 动态文档）	50次/小时	3K tokens	40-50%	✅ 适合
多工具 Agent（大型工具 schema）	100次/小时	12K tokens	65-70%	✅ 最佳场景
一次性文本摘要	5次/小时	1K tokens	~5%	❌ 收益微薄
每用户独立系统提示词的聊天	不定	不定	0%	❌ 无法命中
批处理（每次 prompt 都不同）	1000次/小时	500 tokens	0%	❌ 低于门槛

我的判断标准： 如果你的 Agent 每小时发起超过 10 次请求，且稳定前缀超过 1,024 tokens，就应该开缓存。回报是即时的。

完整 Python 实现：带缓存的 Agent 循环

以下代码使用 Anthropic 官方 Python SDK，可以直接运行：

import anthropic
from typing import Generator

client = anthropic.Anthropic()  # 读取 ANTHROPIC_API_KEY 环境变量

SYSTEM_PROMPT = """你是 Acme 公司的客服 Agent。
你可以使用以下工具，必须遵守以下规则：

1. 访问账户数据前必须验证客户身份
2. 超过 $500 的账单争议必须转人工
3. 不得向客户暴露内部系统 ID
4. 每个操作都需要记录 reason code

[... 假设这里继续到约 8,000 tokens，
     包含工具定义、策略规则、示例对话和格式要求 ...]
"""

TOOLS = [
    {
        "name": "lookup_customer",
        "description": "根据邮箱或手机号查找客户",
        "input_schema": {
            "type": "object",
            "properties": {
                "email": {"type": "string"},
                "phone": {"type": "string"}
            }
        }
    },
    {
        "name": "get_order_status",
        "description": "根据订单号查询订单状态",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"}
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "create_ticket",
        "description": "创建工单进行人工升级",
        "input_schema": {
            "type": "object",
            "properties": {
                "subject": {"type": "string"},
                "priority": {"type": "string", "enum": ["low", "medium", "high"]},
                "description": {"type": "string"}
            },
            "required": ["subject", "priority", "description"]
        }
    }
]


def run_agent_turn(conversation_history: list[dict]) -> dict:
    """执行一次 Agent 调用，启用 prompt caching。"""
    response = client.messages.create(
        model="claude-sonnet-5-20261001",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"}  # 关键：缓存系统提示词
            }
        ],
        tools=TOOLS,
        messages=conversation_history
    )
    return response


def run_agent_loop(user_messages: Generator[str, None, None]):
    """处理用户消息的主循环。"""
    conversation_history = []
    
    for user_msg in user_messages:
        conversation_history.append({
            "role": "user",
            "content": user_msg
        })
        
        response = run_agent_turn(conversation_history)
        
        # 监控缓存命中情况
        usage = response.usage
        print(f"输入 tokens: {usage.input_tokens}")
        print(f"缓存读取 tokens: {getattr(usage, 'cache_read_input_tokens', 0)}")
        print(f"缓存写入 tokens: {getattr(usage, 'cache_creation_input_tokens', 0)}")
        
        # 处理工具调用循环
        while response.stop_reason == "tool_use":
            tool_results = execute_tools(response.content)
            conversation_history.append({"role": "assistant", "content": response.content})
            conversation_history.append({"role": "user", "content": tool_results})
            response = run_agent_turn(conversation_history)
        
        conversation_history.append({
            "role": "assistant",
            "content": response.content
        })
        
        yield response.content


def execute_tools(content_blocks) -> list[dict]:
    """执行工具调用并返回结果。"""
    results = []
    for block in content_blocks:
        if block.type == "tool_use":
            result = dispatch_tool(block.name, block.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": result
            })
    return results


def dispatch_tool(name: str, params: dict) -> str:
    """工具路由。"""
    handlers = {
        "lookup_customer": lambda p: '{"id": "cust_123", "name": "张三"}',
        "get_order_status": lambda p: '{"status": "shipped", "eta": "2026-08-04"}',
        "create_ticket": lambda p: '{"ticket_id": "TKT-4521"}',
    }
    handler = handlers.get(name, lambda p: '{"error": "unknown tool"}')
    return handler(params)

关键就一行：在 system 消息块上加 "cache_control": {"type": "ephemeral"}。

成本对比：缓存前 vs 缓存后

用 Claude Sonnet 5 的定价来算一笔实账：

前提条件：

• 系统提示词：8,000 tokens
• 平均用户消息：200 tokens
• 平均输出：400 tokens
• 每小时调用：200 次
• 每天运行：24 小时

不用缓存：

每次输入费用: 8,200 tokens × $3.00/百万 = $0.0246
每次输出费用: 400 tokens × $15.00/百万   = $0.006
单次总费用: $0.0306
每日费用: $0.0306 × 200 × 24 = $146.88/天

用缓存（第 1 次写入，后 199 次读取）：

第 1 次（缓存写入）：
  8,000 tokens × $3.75/百万（写入溢价）= $0.030
  200 tokens × $3.00/百万（正常）     = $0.0006
  输出: 400 × $15.00/百万             = $0.006
  小计: $0.0366

后续 199 次（缓存命中）：
  8,000 tokens × $0.30/百万（缓存读取）= $0.0024
  200 tokens × $3.00/百万              = $0.0006
  输出: 400 × $15.00/百万              = $0.006
  单次小计: $0.009

每小时费用: $0.0366 + (199 × $0.009) = $1.83
每日费用: $1.83 × 24 = $43.92/天

节省：$102.96/天（70%）

调用量越大，均摊到写入溢价上的比例越小，节省比例越高。

进阶模式

模式一：启动时预热缓存

如果担心冷启动延迟，在服务启动时发一个轻量请求把缓存填上：

def warm_cache():
    """发送最小请求来预填充缓存。"""
    client.messages.create(
        model="claude-sonnet-5-20261001",
        max_tokens=1,
        system=[{
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}
        }],
        messages=[{"role": "user", "content": "你好"}]
    )
    print("缓存预热完成")

成本不到一分钱，确保第一个真实用户请求就能命中缓存。

模式二：多工具 Agent 的 Schema 缓存

如果你的工具 schema 很大（10 个以上工具很常见），系统提示词和 tools 定义会被作为整体前缀缓存：

response = client.messages.create(
    model="claude-sonnet-5-20261001",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"}
        }
    ],
    tools=TOOLS,  # 工具定义会被包含在缓存前缀中
    messages=conversation_history
)

15 个工具（大约 4K tokens 的 schema）配合 8K 的系统提示词，12K tokens 的前缀全部走缓存读取，省钱效果翻倍。

模式三：对话历史缓存

对于长对话的 Agent，可以把历史对话也标记为缓存：

def build_messages_with_context_cache(conversation_history: list[dict]) -> list[dict]:
    """缓存旧的对话轮次，减少重处理成本。"""
    if len(conversation_history) <= 4:
        return conversation_history
    
    cached_messages = []
    cache_boundary = len(conversation_history) - 4  # 最后 2 轮不缓存
    
    for i, msg in enumerate(conversation_history):
        if i == cache_boundary - 1:
            cached_msg = msg.copy()
            if isinstance(cached_msg["content"], str):
                cached_msg["content"] = [{
                    "type": "text",
                    "text": cached_msg["content"],
                    "cache_control": {"type": "ephemeral"}
                }]
            cached_messages.append(cached_msg)
        else:
            cached_messages.append(msg)
    
    return cached_messages

当对话超过 20 轮、历史记录积累到 5K+ tokens 时，这个模式的价值就体现出来了。

踩坑记录

1. 低流量时缓存反而亏钱

5 分钟的 TTL 是硬性限制。如果你的 Agent 每 30 分钟才来一个请求，每次都是缓存 MISS，每次都付 25% 的写入溢价。监控响应里的 cache_read_input_tokens 字段——如果持续为 0，说明缓存在帮倒忙。

2. 前缀匹配极其严格

缓存匹配的是精确的 token 序列。如果你在系统提示词里注入了动态时间戳、用户 ID 或请求 ID，缓存会在那个位置之后全部失效。动态内容应该放在 user message 里，别碰缓存块。

3. 最小 Token 门槛是针对连续块的

常见错误：把系统提示词拆成多个小块，每个不到 1,024 tokens。门槛要求的是连续缓存内容的总量。把指令合并成一个大块。

4. 切换模型版本会清空缓存

从 claude-sonnet-5-20261001 升级到新版本时，所有缓存失效。在低流量时段做模型升级。

5. 没有手动清除缓存的 API

无法主动失效缓存。如果更新了系统提示词，直接发送新版本即可——新内容会写入新的缓存条目，旧的自然过期。

常见问题

Prompt caching 会影响输出质量吗？

不会。缓存纯粹是计费和延迟层面的优化。模型收到的 token 内容完全相同，无论是从缓存读取还是重新处理。输出结果一致。

能缓存用户消息吗，还是只能缓存系统提示词？

都可以。任何 content block 都支持 cache_control。实际场景中，缓存 few-shot examples 或长文档上下文（对同一文档做多次查询）时特别有用。

缓存在对话中途过期了会怎样？

下一个请求会变成 cache miss——对缓存 token 支付 25% 溢价重新写入，后续请求恢复从缓存读取。不会报错，完全透明。

1 小时的付费 TTL 值得买吗？

对于持续运行的 Agent，5 分钟的免费 TTL 足够了，因为请求足够频繁，缓存一直是热的。1 小时 TTL 适合定时任务型 Agent——每 15-30 分钟执行一次，5 分钟窗口撑不住，但频率又足以从缓存中获益。

流式输出能用缓存吗？

能。Prompt caching 作用在输入端，和输出是否流式完全独立。用 client.messages.stream(...) 配合相同的 cache_control 参数即可。

总结

1. 在系统提示词上加 cache_control: {"type": "ephemeral"}。 对 Agent 负载来说，这一步贡献了绝大部分省钱效果。

2. 2 次调用就回本。 缓存前缀只要在 5 分钟内被命中 2 次就开始省钱，Agent 场景远超这个门槛。

3. 静态内容保持静态。 不要在缓存块里注入时间戳、请求 ID 等动态数据，这些放到 user message 里。

4. 监控 cache_read_input_tokens。 它是缓存生效的信号。如果持续为 0，排查前缀匹配问题。

5. Agent 是 prompt caching 的最佳场景。 大系统提示词 + 高频调用 + 稳定前缀，三个条件 Agent 全占。对于跑在 Claude 上的生产 Agent，没有理由不开启 anthropic caching API 的缓存功能。

---

在 SandBase 上使用 Claude 缓存

SandBase 支持 Anthropic prompt caching——cache_control 参数原样透传。额外好处是：当 Claude 限流时可以自动 fallback 到 DeepSeek、GPT-5.x 等备选模型。

from openai import OpenAI

client = OpenAI(base_url="https://api.sandbase.ai/v1", api_key="your-key")

# 缓存参数原样生效
response = client.chat.completions.create(
    model="anthropic/claude-sonnet-5",
    messages=[...],  # cache_control 结构不变
    extra_body={"anthropic_cache": True}
)

查看 SandBase 上的 Claude 模型：sandbase.ai/vendor/anthropic

Related Reading

• Anthropic Prompt Caching 定价拆解：5 分钟 vs 1 小时
• 费用看板 Agent（Anthropic SDK 教程）
• 2026 LLM API 定价全指南
• Opus 5 vs Sonnet 5：什么时候值得多花 5 倍
• 按次 vs 按 Token：Agent 怎么选
• RAG 成本结构拆解：嵌入 + 搜索 + LLM