提示缓存（Prompt Caching）原理与收益

第26章：提示缓存（Prompt Caching）原理与收益

你每次向 Hermes Agent 发起调用，都在重复"阅读"同一份 13,900 tokens 的框架指令。Anthropic 的 Prompt Caching 技术让这份重复劳动的代价降低 90%。本章将深入解析这一技术的工作原理，以及在 Hermes 框架中如何配置并实现约 2 倍的成本节省。

26.1 Anthropic Prompt Caching 机制原理

什么是 Prompt Caching？

Prompt Caching 是 Anthropic 在 2024 年 8 月推出的一项服务端缓存技术。其核心思想是：对于在多次 API 调用中保持不变的内容，只需在第一次调用时完整处理，后续调用直接复用已处理的 KV（Key-Value）缓存。

从技术层面理解：

传统 API 调用（无缓存）：
┌─────────────────────────────────────┐
│  每次调用：                          │
│  [系统提示 4,200 tokens] ──► 完整处理 │
│  [工具定义 6,800 tokens] ──► 完整处理 │
│  [记忆注入 2,100 tokens] ──► 完整处理 │
│  [用户消息  xxx tokens]  ──► 完整处理 │
│  计费：所有 tokens 按"输入价格"计费   │
└─────────────────────────────────────┘

使用 Prompt Caching：
┌─────────────────────────────────────┐
│  第 1 次调用（缓存写入）：            │
│  [系统提示 4,200 tokens] ──► 完整处理 + 写缓存（加收 25%） │
│  [工具定义 6,800 tokens] ──► 完整处理 + 写缓存（加收 25%） │
│  [用户消息  xxx tokens]  ──► 完整处理 │
│                                     │
│  第 2–N 次调用（缓存命中）：          │
│  [系统提示 4,200 tokens] ──► 读缓存（按 10% 价格计费）    │
│  [工具定义 6,800 tokens] ──► 读缓存（按 10% 价格计费）    │
│  [用户消息  xxx tokens]  ──► 完整处理                   │
└─────────────────────────────────────┘

Token 计费规则

缓存生命周期

• 缓存有效期：默认 5 分钟；在 5 分钟内发起的后续调用将命中缓存
• 缓存刷新：每次命中都会重置 5 分钟计时器
• 最小缓存单元：需要至少 1,024 tokens 才能启用缓存（防止小内容缓存的低效）
• 缓存边界：缓存在 cache_control 标记处截断，标记后的内容不缓存

26.2 在 Hermes 中的配置方法

基础配置：标记缓存断点

import anthropic

client = anthropic.Anthropic()

def create_hermes_request_with_caching(
    user_message: str,
    conversation_history: list[dict] = None
) -> dict:
    """
    构建带有 Prompt Caching 的 Hermes 请求。
    
    缓存策略：
    - 系统提示：完整缓存（变化频率最低）
    - 工具定义：完整缓存（变化频率低）
    - 记忆内容：可选缓存（变化频率中等）
    - 用户消息：不缓存（每次不同）
    """
    
    # 1. 系统提示：加 cache_control，整体缓存
    system_content = [
        {
            "type": "text",
            "text": load_hermes_system_prompt(),
            "cache_control": {"type": "ephemeral"}  # 标记缓存断点
        }
    ]
    
    # 2. 工具定义：在 tools 参数中标记缓存
    tools = load_hermes_tools_with_cache_control()
    
    # 3. 构建消息列表
    messages = []
    
    # 注入记忆（带缓存标记）
    if memory_content := load_memory():
        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "text", 
                    "text": f"<memory>\n{memory_content}\n</memory>",
                    "cache_control": {"type": "ephemeral"}  # 记忆内容缓存
                }
            ]
        })
        messages.append({
            "role": "assistant",
            "content": "I have reviewed the memory context. How can I help?"
        })
    
    # 添加对话历史（不缓存，或仅缓存较早的历史）
    if conversation_history:
        messages.extend(conversation_history)
    
    # 添加当前用户消息（不缓存）
    messages.append({
        "role": "user",
        "content": user_message
    })
    
    return {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 4096,
        "system": system_content,
        "tools": tools,
        "messages": messages
    }


def load_hermes_tools_with_cache_control() -> list[dict]:
    """加载工具定义，并在最后一个工具上标记缓存断点"""
    tools = load_hermes_tools()  # 加载标准工具列表
    
    # 仅在最后一个工具上标记 cache_control
    # Anthropic 会缓存到这个断点之前的所有工具定义
    if tools:
        tools[-1]["cache_control"] = {"type": "ephemeral"}
    
    return tools

高级配置：多层缓存断点

对于复杂场景，可以设置多个缓存断点，精细控制哪些内容被缓存：

def create_multi_level_cache_request(
    user_message: str,
    project_context: str,       # 项目特定上下文（中等变化频率）
    conversation_history: list,
    recent_tool_results: list
) -> dict:
    """
    多层缓存策略：
    层级 1：系统提示 + 工具定义（最稳定，缓存效益最高）
    层级 2：项目上下文（中等稳定，缓存效益中等）
    层级 3：对话历史（仅缓存较旧的部分）
    不缓存：最新消息、工具调用结果
    """
    messages = []
    
    # 层级 2：项目上下文（变化周期：小时/天级别）
    if project_context:
        messages.append({
            "role": "user",
            "content": [{
                "type": "text",
                "text": f"<project_context>\n{project_context}\n</project_context>",
                "cache_control": {"type": "ephemeral"}  # 断点 2
            }]
        })
        messages.append({
            "role": "assistant",
            "content": "Project context loaded."
        })
    
    # 层级 3：较早的对话历史（超过 10 轮的对话）
    if len(conversation_history) > 10:
        old_history = conversation_history[:-5]  # 保留最近 5 轮不缓存
        # 在旧历史的最后一条消息上打标记
        if old_history:
            last_msg = old_history[-1].copy()
            if isinstance(last_msg["content"], str):
                last_msg["content"] = [{
                    "type": "text",
                    "text": last_msg["content"],
                    "cache_control": {"type": "ephemeral"}  # 断点 3
                }]
            messages.extend(old_history)
        messages.extend(conversation_history[-5:])  # 最近 5 轮不缓存
    else:
        messages.extend(conversation_history)
    
    # 当前消息（不缓存）
    messages.append({"role": "user", "content": user_message})
    
    return build_request(messages=messages, ...)

配置验证：读取缓存使用情况

def call_with_cache_monitoring(request: dict) -> tuple[dict, dict]:
    """调用 API 并监控缓存命中情况"""
    response = client.messages.create(**request)
    
    usage = response.usage
    cache_stats = {
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cache_creation_input_tokens": getattr(usage, "cache_creation_input_tokens", 0),
        "cache_read_input_tokens": getattr(usage, "cache_read_input_tokens", 0),
    }
    
    # 计算实际成本
    # Claude 3.5 Sonnet 价格（2024 Q4）
    PRICE_INPUT = 0.003        # $3 / 1M tokens
    PRICE_CACHE_WRITE = 0.00375  # $3.75 / 1M tokens（125%）
    PRICE_CACHE_READ = 0.0003    # $0.30 / 1M tokens（10%）
    PRICE_OUTPUT = 0.015         # $15 / 1M tokens
    
    cost = (
        cache_stats["input_tokens"] / 1_000_000 * PRICE_INPUT +
        cache_stats["cache_creation_input_tokens"] / 1_000_000 * PRICE_CACHE_WRITE +
        cache_stats["cache_read_input_tokens"] / 1_000_000 * PRICE_CACHE_READ +
        cache_stats["output_tokens"] / 1_000_000 * PRICE_OUTPUT
    )
    
    cache_stats["estimated_cost_usd"] = cost
    cache_stats["cache_hit"] = cache_stats["cache_read_input_tokens"] > 0
    cache_stats["savings_vs_no_cache"] = calculate_savings(cache_stats)
    
    return response, cache_stats

def calculate_savings(stats: dict) -> float:
    """计算与不使用缓存相比的节省金额"""
    # 如果不使用缓存，缓存读取的 token 也要按普通输入价格计费
    no_cache_cost = (
        (stats["input_tokens"] + stats["cache_read_input_tokens"]) 
        / 1_000_000 * 0.003
    )
    actual_cost = stats["estimated_cost_usd"]
    return max(0, no_cache_cost - actual_cost)

26.3 缓存命中率的影响因素

影响命中率的核心因素

常见的缓存命中率下降原因

# 问题示例 1：在缓存内容中嵌入动态信息
# 这会导致每次调用的缓存内容不同，命中率为 0%

# 错误做法 ❌
system_prompt = f"""
You are Hermes. Current time: {datetime.now()}.  # 动态内容在缓存区内！
Today's date is {date.today()}.
User ID: {user_id}.
...（静态内容）
"""

# 正确做法 ✓
system_prompt_static = """
You are Hermes, an AI assistant by NousResearch.
...（纯静态内容）
"""

# 动态信息放在用户消息中，不被缓存
user_message_with_context = f"""
[Context: {datetime.now().strftime('%Y-%m-%d %H:%M')} | User: {user_id}]

{actual_user_message}
"""

# 问题示例 2：工具定义中包含动态内容

# 错误做法 ❌
def build_tools_with_dynamic_desc():
    return [{
        "name": "get_data",
        "description": f"Get data. Last updated: {datetime.now()}",  # 动态！
        ...
    }]

# 正确做法 ✓
STATIC_TOOLS = [
    {
        "name": "get_data",
        "description": "Get data from the configured data source.",  # 纯静态
        ...
    }
]

缓存预热策略

import asyncio

async def warmup_cache(client: anthropic.Anthropic, system_prompt: str, tools: list):
    """
    在服务启动时预热缓存，确保第一个真实用户请求就能命中缓存。
    """
    warmup_request = {
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 10,
        "system": [{
            "type": "text",
            "text": system_prompt,
            "cache_control": {"type": "ephemeral"}
        }],
        "tools": tools,  # 最后一个工具带 cache_control
        "messages": [{"role": "user", "content": "warmup"}]
    }
    
    response = await client.messages.create(**warmup_request)
    created = getattr(response.usage, "cache_creation_input_tokens", 0)
    print(f"缓存预热完成：写入 {created} tokens 到缓存")

# 在应用启动时调用
asyncio.run(warmup_cache(client, SYSTEM_PROMPT, TOOLS))

26.4 实测成本节省数据

实测场景设计

测试环境：Claude 3.5 Sonnet，标准 Hermes 配置，100 次连续调用（模拟活跃会话）

关键发现：用户消息越短（即固定开销占比越高），缓存节省的比例越大。这与 73% 固定开销的发现高度一致——大部分开销恰好是最适合缓存的内容。

月度成本对比（基准：1,000 次调用/天，典型消息长度）

不使用缓存：
  - 每次调用输入 Token：~15,900（固定13,900 + 用户2,000）
  - 每次调用费用：$0.0477
  - 月度费用（30天）：$1,431

使用 Prompt Caching（第 N+1 次起命中缓存）：
  - 每次调用输入 Token（不含缓存读取）：~2,000（仅用户消息）
  - 缓存读取 Token：~13,900（按10%计费）
  - 每次调用费用：$0.0102
  - 月度费用（30天）：$306

节省：$1,125/月（节省 78.6%）

实际项目中的测量结果

以下是来自社区分享的三个真实项目数据（已匿名）：

26.5 与其他成本优化手段的叠加效果

叠加优化策略矩阵

# 优化策略叠加示例：同时使用缓存 + 按需工具 + 输出压缩

class FullyOptimizedHermesAgent:
    """整合多种成本优化策略的 Hermes Agent"""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.tool_classifier = ToolClassifier()
        self.memory_retriever = SmartMemoryLoader()
        self.history_compressor = HistoryCompressor()
    
    async def run(self, user_message: str, session_id: str) -> str:
        # 策略 1：按需加载工具（节省 ~3,000–5,000 tokens）
        relevant_tools = self.tool_classifier.get_tools(user_message)
        
        # 策略 2：智能记忆检索（节省 ~800–1,500 tokens）
        relevant_memory = self.memory_retriever.retrieve_relevant(
            user_message, max_tokens=600
        )
        
        # 策略 3：历史压缩（节省 ~1,000–5,000 tokens）
        history = self.history_compressor.get_compressed(
            session_id, max_tokens=3000
        )
        
        # 策略 4：Prompt Caching（节省 50%–85% 的固定开销计费）
        request = self._build_cached_request(
            user_message=user_message,
            tools=relevant_tools,
            memory=relevant_memory,
            history=history
        )
        
        response, cache_stats = call_with_cache_monitoring(request)
        
        # 记录优化效果
        self._log_optimization_stats(cache_stats)
        
        return response.content[0].text

综合优化效果对比

叠加效益：全部策略组合使用时，成本可降至基线的 11%，相当于节省 89% 的成本。这是一个值得投入工程资源的优化方向。

成本优化决策流程

开始：评估当前月费
  │
  ├─► 月费 < $50？ → 暂不优化，等规模扩大后再投入工程资源
  │
  ├─► 月费 $50–$500？
  │     → 优先：配置 Prompt Caching（1天工程量，节省50%+）
  │     → 其次：精简系统提示（半天工程量，节省10%–20%）
  │
  └─► 月费 > $500？
        → 全面优化：
           1. Prompt Caching（必做）
           2. 按需工具加载（高收益）
           3. 记忆检索优化（中收益）
           4. 历史压缩（高收益）
           5. 监控 Dashboard（持续优化）

26.6 小结

Prompt Caching 是 Hermes Agent 成本优化中性价比最高的单一措施：

• 原理：将不变的系统提示、工具定义等内容缓存在服务端，后续调用只需支付 10% 的读取费用
• 配置要点：在 system 和 tools 的合适位置添加 cache_control: {"type": "ephemeral"} 标记
• 命中率关键：保持缓存内容纯静态、控制调用间隔在 5 分钟内、做好缓存预热
• 实测收益：典型场景节省 50%–80% 成本；与其他策略叠加后可节省高达 89%
• ROI 极高：配置工作量约 1 天，长期收益持续且无副作用

思考题

1. Prompt Caching 的 5 分钟有效期是一把双刃剑。对于批处理场景（一次性处理大量任务），如何设计调用节奏以最大化缓存命中率？

2. 如果系统提示需要每小时更新一次（如注入最新的业务数据），Prompt Caching 还适用吗？如何权衡"动态内容的必要性"与"缓存效益"？

3. 在多用户并发场景下，缓存是共享的还是每用户独立的？这对你的系统设计有什么影响？

4. 缓存写入比普通输入贵 25%。设计一个数学模型，计算在什么条件下（最少需要多少次缓存命中）Prompt Caching 才能实现正收益？