【Claude】缓存机制与性能调优指南 — 已解决

【Claude】缓存机制与性能调优指南 — 已解决适用版本Claude Code v1.0.x 及以上受影响场景重复请求优化、长上下文缓存、API 延迟降低、批量任务加速阅读时长约 25 分钟目录问题现象原理深挖Prompt Caching 机制根因分析性能瓶颈六大根源多方案解决从缓存到调优验证回归性能调优验证避坑最佳实践附录缓存参数速查表1. 问题现象1.1 典型问题表现问题一相同上下文重复发送导致高成本# 场景10 轮对话每轮系统提示 CLAUDE.md 15K tokens # 每轮都重新发送这 15K tokens # 10 轮 150K tokens 重复发送 × $3/M $0.45 # 如果启用缓存只首次全价后续 10% → 节省 80%问题二首次响应延迟过高# 冷启动系统提示 工具定义 CLAUDE.md 用户输入 # 30K tokens 输入 → API 处理 3-5 秒 # 用户感知延迟高问题三批量任务慢# 100 个文件逐个分析 # 每次都重新构建上下文 → 重复开销 # 总耗时: 100 × 5s 500s问题四缓存未命中不知原因# 启用了 cache_control 但 cache_creation_input_tokens 总是 0 # 不知道为什么缓存不生效问题五缓存过期导致重复计费# 缓存 TTL 5 分钟 # 第 6 分钟发送请求 → 缓存已过期 → 重新全价计费 # 长对话中频繁过期2. 原理深挖Prompt Caching 机制2.1 什么是 Prompt CachingAnthropic API 提供的 Prompt Caching 允许将请求的前缀部分缓存后续请求如果前缀相同只需支付 10% 的输入 Token 费用。无缓存请求: 请求 1: [系统提示 10K][CLAUDE.md 5K][对话历史 5K] → 全价 20K tokens 请求 2: [系统提示 10K][CLAUDE.md 5K][对话历史 8K] → 全价 23K tokens 请求 3: [系统提示 10K][CLAUDE.md 5K][对话历史 12K] → 全价 27K tokens 总计: 70K tokens × $3/M $0.21 有缓存请求: 请求 1: [系统提示 10K][CLAUDE.md 5K (cached)][对话历史 5K] → 15K 全价 10K 缓存写入(1.25倍) $0.056 请求 2: [系统提示 10K (cached)][CLAUDE.md 5K (cached)][对话历史 8K] → 3K 全价 15K 缓存读取(0.1倍) $0.011 请求 3: [系统提示 10K (cached)][CLAUDE.md 5K (cached)][对话历史 12K] → 4K 全价 15K 缓存读取(0.1倍) $0.015 总计: ~$0.082 (节省 61%)2.2 缓存定价操作价格倍率说明正常输入1.0×标准 Token 价格缓存写入1.25×首次缓存稍贵缓存读取0.1×命中缓存极便宜Sonnet 4 示例: 正常输入: $3.00/M tokens 缓存写入: $3.75/M tokens (1.25×) 缓存读取: $0.30/M tokens (0.1×)2.3 缓存规则缓存条件: 1. 前缀必须完全匹配包括空格、换行 2. 最小缓存长度: 1024 tokens (Sonnet/Opus) / 2048 tokens (Haiku) 3. 缓存 TTL: 5 分钟 4. 最多 4 个 cache_control 断点 5. 缓存按组织级别隔离 缓存位置: 请求结构: [system][messages[0]][messages[1]]... cache_control 只能放在 content block 的最后一个 system: { type: text, text: ..., cache_control: {type: ephemeral} ← 缓存点 1 } messages: [ {role: user, content: [ {type: text, text: 固定前缀, cache_control: {type: ephemeral}}, ← 缓存点 2 {type: text, text: 变化部分} ]} ]2.4 Claude Code 的缓存策略Claude Code 内部自动使用 Prompt CachingClaude Code 缓存层次: Layer 1: System Prompt (系统提示) - 工具定义 行为规则 安全约束 - ~10K tokens - 始终缓存每次请求前缀相同 Layer 2: CLAUDE.md (记忆文件) - 项目配置和规范 - ~3K tokens - 始终缓存会话内不变 Layer 3: 对话历史 - 前面的对话消息 - 增量缓存新消息追加到尾部 - 最近的 N 轮对话被缓存 Layer 4: 当前用户输入 - 不缓存每次不同2.5 缓存命中与未命中# API 响应中的缓存指标 response client.messages.create(...) response.usage.cache_creation_input_tokens # 缓存写入的 Token 数 response.usage.cache_read_input_tokens # 缓存读取的 Token 数 response.usage.input_tokens # 未缓存的输入 Token 数 # 判断缓存状态 if response.usage.cache_read_input_tokens 0: print(✓ 缓存命中) elif response.usage.cache_creation_input_tokens 0: print(→ 缓存写入首次) else: print(✗ 无缓存)3. 根因分析性能瓶颈六大根源3.1 根源一未利用 Prompt CachingSDK 直接调用时不自动启用缓存需要手动设置cache_control。3.2 根源二上下文顺序不固定如果系统提示或对话前缀在每次请求中有微小变化多一个空格、顺序变化缓存就会失效。3.3 根源三缓存过期5 分钟 TTL 内没有发送请求缓存自动清除。长对话中如果思考时间超过 5 分钟每次都要重新缓存。3.4 根源四批量任务无复用批量处理多个文件时如果每个请求都构建新的上下文无法复用缓存。3.5 根源五工具定义重复发送每次 API 调用都发送完整的工具定义~5K tokens即使工具列表没有变化。3.6 根源六网络延迟未优化没有使用流式输出、连接复用、就近接入等网络优化手段。4. 多方案解决从缓存到调优4.1 方案一SDK 手动启用缓存import anthropic client anthropic.Anthropic(api_keysk-ant-xxx) SYSTEM_PROMPT 你是一个代码审查助手。遵循以下规范: 1. 检查安全漏洞 2. 检查性能问题 3. 检查代码风格 ...长系统提示 2000 tokens def cached_chat(user_message, messagesNone): 启用 Prompt Caching 的调用 # 系统提示加缓存标记 system_block { type: text, text: SYSTEM_PROMPT, cache_control: {type: ephemeral} } # 构建消息 if messages is None: messages [] messages.append({role: user, content: user_message}) # 对历史消息的最后一条加缓存 if len(messages) 2: # 对话历史的前缀加缓存 history_text json.dumps(messages[:-1]) # 实际操作中cache_control 加在 content block 上 response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens4096, system[system_block], # 系统提示缓存 messagesmessages, toolsget_tools_with_cache() # 工具定义缓存 ) # 报告缓存状态 cache_read response.usage.cache_read_input_tokens cache_write response.usage.cache_creation_input_tokens non_cached response.usage.input_tokens if cache_read 0: print(f✓ 缓存命中: {cache_read:,} tokens (0.1×)) if cache_write 0: print(f→ 缓存写入: {cache_write:,} tokens (1.25×)) print(f 非缓存: {non_cached:,} tokens (1.0×)) return response # 第一次调用: 缓存写入 cached_chat(审查 auth.py) # 输出: → 缓存写入: 10000 tokens (1.25×) # 5 分钟内第二次调用: 缓存命中 cached_chat(审查 user.py) # 输出: ✓ 缓存命中: 10000 tokens (0.1×)4.2 方案二多断点缓存策略def multi_breakpoint_cache(system_prompt, claude_md, conversation_history, current_input): 多断点缓存策略 在 system、CLAUDE.md、对话历史三个位置设置缓存断点 最大化缓存命中率 messages [] # 将对话历史转为 content blocks if conversation_history: history_content [] for msg in conversation_history: history_content.append({ type: text, text: f[{msg[role]}]: {msg[content]}\n }) # 在历史末尾加缓存断点 history_content[-1][cache_control] {type: ephemeral} messages.append({ role: user, content: history_content }) # 当前输入 messages.append({role: user, content: current_input}) # 系统提示 CLAUDE.md 两个缓存断点 system_blocks [ { type: text, text: system_prompt, cache_control: {type: ephemeral} # 断点 1 }, { type: text, text: claude_md, cache_control: {type: ephemeral} # 断点 2 } ] response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens4096, systemsystem_blocks, messagesmessages ) return response # 使用 system 你是代码助手... # 固定 claude_md open(CLAUDE.md).read() # 固定 history [ {role: user, content: 分析 auth.py}, {role: assistant, content: auth.py 分析完成...} ] # 第一次写入缓存 multi_breakpoint_cache(system, claude_md, history, 继续分析 user.py) # 第二次5分钟内命中缓存 history.append({role: assistant, content: user.py 分析完成...}) multi_breakpoint_cache(system, claude_md, history, 分析 order.py)4.3 方案三缓存预热策略 缓存预热在批量任务开始前先写入缓存 def warmup_cache(system_prompt, tools_definition): 预热缓存确保后续请求命中 # 发送一个最小请求来写入缓存 response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens10, # 最小输出 system[{ type: text, text: system_prompt, cache_control: {type: ephemeral} }], toolstools_definition, messages[{role: user, content: ok}] ) cache_written response.usage.cache_creation_input_tokens print(f✓ 缓存预热完成: {cache_written:,} tokens 已缓存) return response def batch_process_with_cache(files, system_prompt): 批量处理 缓存复用 # 1. 预热缓存 warmup_cache(system_prompt, get_tools()) # 2. 批量处理5 分钟内完成以确保缓存命中 results [] for filepath in files: content open(filepath).read() response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens2048, system[{ type: text, text: system_prompt, cache_control: {type: ephemeral} }], messages[{ role: user, content: f审查: {filepath}\n\n{content} }] ) cache_read response.usage.cache_read_input_tokens cost calculate_cost(response) results.append({ file: filepath, result: response.content[0].text, cache_hit: cache_read 0, cost: cost }) status ✓ if cache_read 0 else ✗ print(f {status} {filepath}: ${cost:.4f}) # 汇总 total_cost sum(r[cost] for r in results) cache_hits sum(1 for r in results if r[cache_hit]) print(f\n总计: {len(results)} 文件, ${total_cost:.4f}, f缓存命中 {cache_hits}/{len(results)}) return results # 使用 files [fsrc/{f} for f in os.listdir(src) if f.endswith(.py)] batch_process_with_cache(files, SYSTEM_PROMPT)4.4 方案四缓存保活策略 缓存保活在长对话中定期发送请求保持缓存不过期 import threading import time class CacheKeepAlive: 缓存保活管理器 def __init__(self, client, system_prompt, interval240): 参数: interval: 保活间隔秒默认 240s4分钟小于 5分钟 TTL self.client client self.system_prompt system_prompt self.interval interval self._running False self._thread None def _keepalive_loop(self): 保活循环 while self._running: time.sleep(self.interval) if not self._running: break # 发送最小请求保持缓存 try: self.client.messages.create( modelclaude-sonnet-4-20250514, max_tokens1, system[{ type: text, text: self.system_prompt, cache_control: {type: ephemeral} }], messages[{role: user, content: .}] ) print(f [keepalive] 缓存已刷新) except Exception as e: print(f [keepalive] 失败: {e}) def start(self): 启动保活 self._running True self._thread threading.Thread(targetself._keepalive_loop, daemonTrue) self._thread.start() print(✓ 缓存保活已启动) def stop(self): 停止保活 self._running False if self._thread: self._thread.join(timeout5) print(✓ 缓存保活已停止) # 使用 keepalive CacheKeepAlive(client, SYSTEM_PROMPT, interval240) keepalive.start() # 长对话中即使用户思考超过 5 分钟缓存也不会过期 # ... 对话 ... # ... 用户思考 10 分钟 ... # 缓存仍然有效 keepalive.stop()4.5 方案五性能基准测试 性能基准测试对比缓存前后的延迟和成本 import time import anthropic client anthropic.Anthropic(api_keysk-ant-xxx) SYSTEM_PROMPT 你是代码助手。 * 500 # ~2500 tokens def benchmark_no_cache(calls5): 无缓存基准 total_time 0 total_cost 0 for i in range(calls): start time.time() response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens100, systemSYSTEM_PROMPT, # 字符串形式无缓存 messages[{role: user, content: ftest {i}}] ) elapsed time.time() - start input_t response.usage.input_tokens output_t response.usage.output_tokens cost (input_t * 3 output_t * 15) / 1_000_000 total_time elapsed total_cost cost print(f #{i1}: {elapsed:.2f}s, ${cost:.4f}) print(f\n无缓存: 平均 {total_time/calls:.2f}s, ${total_cost:.4f}) return total_time / calls, total_cost def benchmark_with_cache(calls5): 有缓存基准 total_time 0 total_cost 0 for i in range(calls): start time.time() response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens100, system[{ type: text, text: SYSTEM_PROMPT, cache_control: {type: ephemeral} }], messages[{role: user, content: ftest {i}}] ) elapsed time.time() - start input_t response.usage.input_tokens output_t response.usage.output_tokens cache_read response.usage.cache_read_input_tokens cache_write response.usage.cache_creation_input_tokens # 计算实际成本 cost ( input_t * 3 cache_write * 3.75 cache_read * 0.3 output_t * 15 ) / 1_000_000 cache_status 命中 if cache_read 0 else 写入 if cache_write 0 else 无 total_time elapsed total_cost cost print(f #{i1}: {elapsed:.2f}s, ${cost:.4f} (缓存:{cache_status})) print(f\n有缓存: 平均 {total_time/calls:.2f}s, ${total_cost:.4f}) return total_time / calls, total_cost # 运行基准 print( 无缓存 ) no_cache_time, no_cache_cost benchmark_no_cache() print(\n 有缓存 ) with_cache_time, with_cache_cost benchmark_with_cache() print(f\n 对比 ) print(f延迟: {no_cache_time:.2f}s → {with_cache_time:.2f}s f({(1-with_cache_time/no_cache_time)*100:.0f}% 提升)) print(f成本: ${no_cache_cost:.4f} → ${with_cache_cost:.4f} f({(1-with_cache_cost/no_cache_cost)*100:.0f}% 节省))4.6 方案六网络层优化 网络层优化连接复用、超时调优、就近接入 import anthropic import httpx # 1. 连接复用SDK 默认已启用 # httpx.Client 自动复用 TCP 连接 # 2. 自定义 HTTP 客户端调优连接池 custom_http httpx.Client( timeouthttpx.Timeout( connect5.0, # 连接超时 read60.0, # 读取超时 write10.0, # 写入超时 pool5.0 # 连接池等待 ), limitshttpx.Limits( max_connections10, max_keepalive_connections5, keepalive_expiry30.0 ), http2True # 启用 HTTP/2 多路复用 ) client anthropic.Anthropic( api_keysk-ant-xxx, http_clientcustom_http ) # 3. 流式输出减少首字节延迟 def stream_first_byte(prompt): 使用流式输出获取首个 token 的延迟 start time.time() first_byte_time None with client.messages.stream( modelclaude-sonnet-4-20250514, max_tokens4096, messages[{role: user, content: prompt}] ) as stream: for event in stream: if event.type content_block_delta and first_byte_time is None: first_byte_time time.time() - start print(f 首字节延迟: {first_byte_time:.2f}s) break total_time time.time() - start print(f 总耗时: {total_time:.2f}s) # 流式 vs 非流式: # 非流式: 等待完整响应 → 用户等待 total_time # 流式: 首字节后即可开始阅读 → 用户感知延迟 first_byte_time4.7 方案七并发优化 并发优化多请求并行处理 import anthropic from concurrent.futures import ThreadPoolExecutor, as_completed import threading # 线程安全的统计 class CostTracker: def __init__(self): self._lock threading.Lock() self.total_cost 0 self.total_tokens 0 def add(self, cost, tokens): with self._lock: self.total_cost cost self.total_tokens tokens tracker CostTracker() def process_file(filepath, client, system_prompt): 处理单个文件可并发 content open(filepath).read() response client.messages.create( modelclaude-sonnet-4-20250514, max_tokens2048, system[{ type: text, text: system_prompt, cache_control: {type: ephemeral} }], messages[{role: user, content: f审查: {filepath}\n{content}}] ) cost (response.usage.input_tokens * 3 response.usage.output_tokens * 15) / 1_000_000 tracker.add(cost, response.usage.input_tokens response.usage.output_tokens) return filepath, response.content[0].text def concurrent_batch(files, system_prompt, max_workers5): 并发批量处理 client anthropic.Anthropic(api_keysk-ant-xxx) start time.time() results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(process_file, f, client, system_prompt): f for f in files } for future in as_completed(futures): filepath futures[future] try: result future.result() results.append(result) print(f ✓ {filepath}) except Exception as e: print(f ✗ {filepath}: {e}) elapsed time.time() - start print(f\n并发 {max_workers}: {len(results)} 文件, {elapsed:.1f}s, f${tracker.total_cost:.4f}) return results # 对比不同并发度 import os files [fsrc/{f} for f in os.listdir(src) if f.endswith(.py)][:20] for workers in [1, 3, 5, 10]: tracker CostTracker() concurrent_batch(files, SYSTEM_PROMPT, max_workersworkers)5. 验证回归性能调优验证5.1 缓存命中率验证def verify_cache(): 验证缓存是否生效 system 你是助手。 * 500 # 1024 tokens # 第一次请求 r1 client.messages.create( modelclaude-sonnet-4-20250514, max_tokens10, system[{type: text, text: system, cache_control: {type: ephemeral}}], messages[{role: user, content: hi}] ) assert r1.usage.cache_creation_input_tokens 0, 首次应写入缓存 print(f✓ 首次缓存写入: {r1.usage.cache_creation_input_tokens} tokens) # 第二次请求应命中 r2 client.messages.create( modelclaude-sonnet-4-20250514, max_tokens10, system[{type: text, text: system, cache_control: {type: ephemeral}}], messages[{role: user, content: hello}] ) assert r2.usage.cache_read_input_tokens 0, 第二次应命中缓存 print(f✓ 缓存命中: {r2.usage.cache_read_input_tokens} tokens) # 成本对比 cost1 r1.usage.cache_creation_input_tokens * 3.75 / 1_000_000 cost2 r2.usage.cache_read_input_tokens * 0.3 / 1_000_000 print(f 首次: ${cost1:.4f}, 命中: ${cost2:.4f}) print(f 节省: {(1-cost2/cost1)*100:.0f}%) verify_cache()5.2 验证清单#验证项预期方法1缓存写入cache_creation 0首次请求检查2缓存命中cache_read 05分钟内二次请求3缓存过期5分钟后 cache_read0等待后请求4成本降低50%对比有/无缓存5延迟降低20%对比响应时间6并发安全无竞态多线程测试7保活有效超时后仍命中keepalive 测试8流式首字节2sstream 测试6. 避坑最佳实践6.1 缓存使用原则原则 1: 固定前缀 — 系统提示和 CLAUDE.md 保持不变 原则 2: 5分钟窗口 — 批量任务在 5 分钟内完成 原则 3: 最小 1024 — 缓存内容至少 1024 tokens 原则 4: 最多 4 断点 — cache_control 最多 4 个 原则 5: 保活策略 — 长对话用 keepalive 原则 6: 预热缓存 — 批量前先 warmup 原则 7: 监控命中 — 检查 cache_read_input_tokens 原则 8: 并发复用 — 多线程共享缓存6.2 缓存失效原因原因检查方法解决前缀变化diff 前缀保持完全一致TTL 过期间隔 5 分钟用 keepalive内容 1024 tokens检查长度增加缓存内容超过 4 断点检查 cache_control 数减少断点不同组织API Key 不同同一组织6.3 性能调优检查清单系统提示使用 cache_controlCLAUDE.md 使用 cache_control对话历史增量缓存批量任务在 5 分钟内完成长对话使用 keepalive流式输出减少感知延迟并发处理批量任务HTTP/2 连接复用监控缓存命中率定期成本对比7. 附录缓存参数速查表7.1 缓存定价模型正常输入缓存写入 (1.25×)缓存读取 (0.1×)Opus 4$15/M$18.75/M$1.50/MSonnet 4$3/M$3.75/M$0.30/MHaiku 3.5$0.25/M$0.31/M$0.025/M7.2 缓存限制限制值最小缓存长度 (Opus/Sonnet)1,024 tokens最小缓存长度 (Haiku)2,048 tokens缓存 TTL5 分钟最大 cache_control 断点4缓存隔离按组织7.3 性能优化手段对比手段成本影响延迟影响复杂度Prompt Caching-50~90%-20~40%低流式输出0%-60% 感知低并发处理0%-N倍中keepalive1%0%中HTTP/20%-10%低模型降级-80%-50%低结语Prompt Caching 是 Anthropic API 提供的强大成本优化机制。通过固定前缀、多断点缓存、缓存预热、保活策略、并发复用等手段可以在长对话和批量任务中节省 50-90% 的输入 Token 成本。核心要点回顾固定前缀系统提示和 CLAUDE.md 保持不变确保缓存命中多断点缓存在 system、CLAUDE.md、对话历史设置缓存断点缓存预热批量任务前先 warmup后续请求全部命中保活策略长对话中定期请求保持缓存不过期并发复用多线程共享同一缓存前缀流式输出减少用户感知延迟监控命中检查cache_read_input_tokens确认缓存生效成本对比定期对比有/无缓存的成本差异