通过Claude API实现提示缓存:降低延迟与成本的完整指南
技术背景
在大语言模型应用中,长上下文处理常导致高延迟与高成本,尤其在多轮对话、文档分析等重复性任务中,相同前缀被反复处理。提示缓存通过存储和重用提示中的上下文,避免重复计算,显著提升响应速度并降低费用。适用于需要频繁使用固定系统提示、长文档或历史对话的场景,如客服机器人、代码助手、文档问答等。主要限制包括最小可缓存长度、缓存TTL(默认5分钟)及断点数量限制(最多4个显式断点),且缓存写入成本略高,需权衡使用频率。
核心原理
提示缓存的核心原理是识别请求中可复用的前缀部分,将其计算结果存储,后续请求若前缀相同则直接读取缓存,跳过重复计算。自动缓存通过在请求顶层添加cache_control字段,系统自动在最后一个可缓存块设置断点,并随对话动态前移。显式缓存断点则允许在特定内容块(如系统提示、长文档)手动放置cache_control,实现细粒度控制。缓存命中时,API返回cache_read_input_tokens,未命中则返回cache_creation_input_tokens。缓存TTL默认5分钟,每次命中刷新,1小时TTL需额外付费。
架构分析
本教程示例采用简单的客户端-服务器架构:Python客户端通过anthropic SDK调用Claude API。核心组件包括:1) 内容获取模块,使用requests和BeautifulSoup抓取并清洗文本;2) 对话管理模块,维护消息列表并处理缓存控制;3) API调用模块,封装messages.create方法。自动缓存模式下,系统自动管理断点,无需额外逻辑;显式模式通过ConversationWithExplicitCaching类手动在最后用户消息和系统提示上放置cache_control。设计上,自动缓存优先,显式断点用于特殊需求,两者可组合。生产环境可扩展为服务化架构,将缓存管理集成到业务逻辑中。
完整实战教程
使用Claude API进行提示缓存
提示缓存允许您存储和重用提示中的上下文,对于重复性任务,延迟可降低2倍以上,成本最高可降低90%。
有两种启用提示缓存的方式:
- 自动缓存(推荐):在请求的顶层添加一个
cache_control字段。系统会自动为您管理缓存断点。 - 显式缓存断点:在单个内容块上放置
cache_control,以精确控制缓存的内容。
本教程演示了这两种方法,从更简单的自动方法开始。
设置
%pip install --upgrade 'anthropic>=0.83.0' bs4 requests python-dotenv --quiet
import time
import anthropic
import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic()
MODEL_NAME = "claude-sonnet-4-6"
# Unique prefix to ensure we don't hit a stale cache from a previous run
TIMESTAMP = int(time.time())
让我们获取《傲慢与偏见》的全文(约18.7万tokens)作为我们的长上下文。
def fetch_article_content(url):
response = requests.get(url, timeout=30)
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)
print(f"Fetched {len(book_content)} characters from the book.")
print("First 500 characters:")
print(book_content[:500])
我们还会定义一个小的辅助函数来打印使用统计信息:
def print_usage(response, elapsed):
"""Print token usage and timing for an API response."""
usage = response.usage
cache_create = getattr(usage, "cache_creation_input_tokens", 0)
cache_read = getattr(usage, "cache_read_input_tokens", 0)
print(f" Time: {elapsed:.2f}s")
print(f" Input tokens: {usage.input_tokens}")
print(f" Output tokens: {usage.output_tokens}")
if cache_create:
print(f" Cache write tokens: {cache_create}")
if cache_read:
print(f" Cache read tokens: {cache_read}")
- --
示例 1:自动缓存(单轮)
自动缓存是最简单的入门方式。在 messages.create() 调用的顶层添加 cache_control={"type": "ephemeral"},系统会处理其余部分——自动将缓存断点放在最后一个可缓存块上。
我们将比较三种场景:
- 无缓存 — 基线
- 首次缓存调用 — 创建缓存条目(与基线时间相近)
- 第二次缓存调用 — 从缓存读取(大幅加速)
基线:无缓存
start = time.time()
baseline_response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
messages=[
{
"role": "user",
"content": str(TIMESTAMP)
+ "<book>"
+ book_content
+ "</book>"
+ "\n\nWhat is the title of this book? Only output the title.",
}
],
)
baseline_time = time.time() - start
print(f"Response: {baseline_response.content[0].text}")
print_usage(baseline_response, baseline_time)
首次调用与自动缓存(缓存写入)
唯一的变化是顶层的 cache_control 参数。首次调用会写入缓存,因此耗时与基线相近。
start = time.time()
write_response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
cache_control={"type": "ephemeral"}, # <-- one-line change
messages=[
{
"role": "user",
"content": str(TIMESTAMP)
+ "<book>"
+ book_content
+ "</book>"
+ "\n\nWhat is the title of this book? Only output the title.",
}
],
)
write_time = time.time() - start
print(f"Response: {write_response.content[0].text}")
print_usage(write_response, write_time)
第二次调用,自动缓存(缓存命中)
再次发送相同的请求。这次会复用缓存的前缀,因此你应该会看到显著的加速效果。
start = time.time()
hit_response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
cache_control={"type": "ephemeral"},
messages=[
{
"role": "user",
"content": str(TIMESTAMP)
+ "<book>"
+ book_content
+ "</book>"
+ "\n\nWhat is the title of this book? Only output the title.",
}
],
)
hit_time = time.time() - start
print(f"Response: {hit_response.content[0].text}")
print_usage(hit_response, hit_time)
print("\n" + "=" * 50)
print("COMPARISON")
print("=" * 50)
print(f"No caching: {baseline_time:.2f}s")
print(f"Cache write: {write_time:.2f}s")
print(f"Cache hit: {hit_time:.2f}s")
print(f"Speedup: {baseline_time / hit_time:.1f}x")
- --
示例2:多轮对话中的自动缓存
自动缓存在多轮对话中尤其出色。随着对话的增长,缓存断点自动向前移动——您无需自行管理任何标记。
| 请求 | 缓存行为 |
|---------|----------------|
| 请求1 | 系统 + 用户:A 缓存(写入) |
| 请求2 | 系统 + 用户:A 从缓存读取;助手:B + 用户:C 写入缓存 |
| 请求3 | 系统至用户:C 从缓存读取;助手:D + 用户:E 写入缓存 |
system_message = f"{TIMESTAMP} <file_contents> {book_content} </file_contents>"
questions = [
"What is the title of this novel?",
"Who are Mr. and Mrs. Bennet?",
"What is Netherfield Park?",
"What is the main theme of this novel?",
]
conversation = []
for i, question in enumerate(questions, 1):
print(f"\n{'=' * 50}")
print(f"Turn {i}: {question}")
print("=" * 50)
conversation.append({"role": "user", "content": question})
start = time.time()
response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
cache_control={"type": "ephemeral"}, # automatic caching
system=system_message,
messages=conversation,
)
elapsed = time.time() - start
assistant_reply = response.content[0].text
conversation.append({"role": "assistant", "content": assistant_reply})
print(f"\nAssistant: {assistant_reply[:200]}{'...' if len(assistant_reply) > 200 else ''}")
print()
print_usage(response, elapsed)
在第一轮之后,几乎每一轮后续对话中,100%的输入令牌都会从缓存中读取。对话代码只是一个普通的消息列表——不需要在单个块上添加特殊的cache_control标记。
示例 3:显式缓存断点
为了获得更多控制,你可以直接在各个内容块上放置 cache_control。这在以下情况下很有用:
- 你想用不同的 TTL 缓存不同的部分
- 你需要独立于消息内容缓存系统提示词
- 你想要对缓存内容进行细粒度控制
你也可以结合两种方法:为系统提示词使用显式断点,同时让自动缓存处理对话。
下面,我们将 cache_control 直接放在书籍内容块上,并在每一轮手动将断点向前移动。
class ConversationWithExplicitCaching:
"""Multi-turn conversation that manually places cache_control on the last user message."""
def __init__(self):
self.turns = []
def add_user(self, content):
self.turns.append({"role": "user", "content": [{"type": "text", "text": content}]})
def add_assistant(self, content):
self.turns.append({"role": "assistant", "content": [{"type": "text", "text": content}]})
def get_messages(self):
"""Return messages with cache_control on the last user message."""
result = []
last_user_idx = max(i for i, t in enumerate(self.turns) if t["role"] == "user")
for i, turn in enumerate(self.turns):
if i == last_user_idx:
result.append(
{
"role": "user",
"content": [
{
"type": "text",
"text": turn["content"][0]["text"],
"cache_control": {"type": "ephemeral"},
}
],
}
)
else:
result.append(turn)
return result
conv = ConversationWithExplicitCaching()
for i, question in enumerate(questions, 1):
print(f"\n{'=' * 50}")
print(f"Turn {i}: {question}")
print("=" * 50)
conv.add_user(question)
start = time.time()
response = client.messages.create(
model=MODEL_NAME,
max_tokens=300,
system=[
{
"type": "text",
"text": system_message,
"cache_control": {"type": "ephemeral"}, # explicit breakpoint on system
},
],
messages=conv.get_messages(),
)
elapsed = time.time() - start
assistant_reply = response.content[0].text
conv.add_assistant(assistant_reply)
print(f"\nAssistant: {assistant_reply[:200]}{'...' if len(assistant_reply) > 200 else ''}")
print()
print_usage(response, elapsed)
选择方法
| | 自动缓存 | 显式断点 |
|---|---|---|
| 易用性 | 一行更改 | 必须放置和移动 cache_control 标记 |
| 多轮对话 | 断点自动向前移动 | 您管理断点位置 |
| 细粒度控制 | 无 | 最多4个独立断点 |
| 混合TTL | 自动断点单一TTL | 每个断点不同TTL |
| 可组合性 | 是——自动+显式一起 | 是 |
从自动缓存开始。 它用最少的努力覆盖大多数用例。仅在需要细粒度控制时切换到显式断点。
关键细节
- 最小可缓存长度: Sonnet为1,024个token;Opus和Haiku 4.5为4,096个token
- 缓存TTL: 默认5分钟(每次命中刷新)。1小时TTL可用,价格为基础输入价格的2倍。
- 定价: 缓存写入成本为基础输入价格的1.25倍。缓存读取成本为基础输入价格的0.1倍。
- 断点限制: 每个请求最多4个显式断点。自动缓存使用一个槽位。
有关完整详细信息,请参阅提示缓存文档。
常见问题
常见错误:1) 缓存未命中——检查请求前缀是否完全一致,包括时间戳等动态内容;2) 缓存写入成本高——确保长上下文复用频率足够;3) 断点位置不当——显式断点需放在可缓存块上,且不超过4个;4) 忽略最小可缓存长度——Sonnet需1024 tokens,Opus/Haiku需4096 tokens;5) 缓存过期——默认5分钟TTL,长时间间隔后需重新写入。排查方法:使用print_usage函数检查cache_read_input_tokens和cache_creation_input_tokens,对比时间差异;确保使用最新SDK版本。生产优化:使用1小时TTL(成本2倍)减少频繁写入;结合自动与显式缓存;监控缓存命中率并调整断点策略。
企业应用扩展
生产环境优化建议:1) 将缓存管理封装为独立服务,支持多租户隔离;2) 使用Redis等外部存储扩展缓存TTL,但需注意API限制;3) 结合流式响应降低首字延迟;4) 对长文档进行分块缓存,动态组合;5) 监控成本与延迟,设置告警。进阶学习:探索Claude API的其他功能如工具调用、多模态;研究提示工程优化缓存效率;对比不同模型的缓存特性。
总结
本教程面向有编程基础的开发者,深入讲解Claude API的提示缓存机制。通过自动缓存与显式缓存断点两种方式,结合《傲慢与偏见》全文作为长上下文示例,演示如何将重复任务的延迟降低2倍以上、成本降低90%。内容涵盖环境配置、代码实现、性能对比、多轮对话场景及方法选择,并补充技术原理、架构设计、常见错误排查与生产优化建议,帮助开发者高效落地提示缓存技术。
- --
内容来源与验证
- 内容等级:PREMIUM_A
- 质量等级:S_CANDIDATE
- 难度:中级
- 验证等级:V1
- 原文作者:alexalbertt
- 原文日期:2024-08-14
- 来源许可:MIT
- 来源版本:
085a3c299fedcd9612bbfce5fac648f22184ed00 - 查看原文
站内还有更多完整实战教程
加入会员,解锁全部教程、完整代码与持续更新内容。
加入会员