ROI 最高的优化手段——不改逻辑,只改结构,就能省 50-90% 的输入 token 成本。OpenAI 自动缓存、Anthropic 显式缓存,两者机制不同但目标一致。
# OpenAI Prompt Caching 工作流程
#
# 1. 请求路由:基于 prompt 前缀哈希路由到同一服务器
# - 哈希通常使用前 256 个 token
# - 可选 prompt_cache_key 参数影响路由
#
# 2. 缓存查找:检查服务器上是否有匹配的前缀缓存
#
# 3a. 缓存命中 (Cache Hit):
# - 跳过前缀的重复计算
# - 输入 token 成本降低 50%(截至2025年5月:缓存价=输入价×10%)
# - 延迟降低最高 80%
#
# 3b. 缓存未命中 (Cache Miss):
# - 正常处理完整 prompt
# - 结果自动缓存供后续请求使用
#
# 缓存条件:
# - prompt 至少 1024 token
# - 前缀完全匹配
# - 同一前缀约 15 请求/分钟以内
import openai
client = openai.OpenAI()
# ❌ 低效:动态内容在前,静态内容在后
# 每次请求缓存都会 miss,因为前缀不同
def bad_request(user_question, conversation_history):
return client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": user_question}, # 动态(每次不同)
{"role": "assistant", "content": "..."},
{"role": "system", "content": LONG_SYSTEM_PROMPT}, # 静态(应该在前)
{"role": "user", "content": TOOLS_DEFINITION}, # 静态
]
)
# ✅ 高效:静态内容在前,动态内容在后
# 系统提示词 + 工具定义作为前缀,可以被缓存
def optimized_request(user_question, conversation_history):
return client.chat.completions.create(
model="gpt-4o",
messages=[
# === 缓存前缀开始 ===
{"role": "system", "content": LONG_SYSTEM_PROMPT}, # 静态
{"role": "system", "content": TOOLS_DEFINITION}, # 静态
# === 缓存前缀结束 ===
# 以下是动态内容
*conversation_history, # 对话历史
{"role": "user", "content": user_question}, # 最新用户输入
]
)
# 使用 prompt_cache_key 提高命中率
def optimized_with_cache_key(user_question, conversation_history,
cache_key="my-app-v1"):
return client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": LONG_SYSTEM_PROMPT},
{"role": "system", "content": TOOLS_DEFINITION},
*conversation_history,
{"role": "user", "content": user_question},
],
# 额外参数:影响缓存路由
extra_body={"prompt_cache_key": cache_key}
)
cache_control 标记来显式指定哪些内容需要缓存。相比 OpenAI 的自动缓存,Anthropic 给你更多控制,但也需要你手动标记。
import anthropic
client = anthropic.Anthropic()
# 方式1: 自动缓存(推荐)—— 顶部 cache_control
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
# 自动缓存:系统自动在最后一个可缓存块设置断点
cache_control={"type": "ephemeral"},
system="你是一个专业的技术助手。你的任务是..." * 10, # 需要足够长
messages=[
{"role": "user", "content": "解释 Python 的 GIL"}
]
)
# 方式2: 显式缓存断点(精细控制)
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
system=[
{
"type": "text",
"text": "你是一个专业的技术助手。" * 50,
"cache_control": {"type": "ephemeral"} # 缓存这个块
}
],
tools=[
{
"name": "search",
"description": "搜索文档",
"input_schema": {...},
"cache_control": {"type": "ephemeral"} # 也可以缓存工具定义
}
],
messages=[
{"role": "user", "content": "解释 Python 的 GIL"}
]
)
# 检查缓存命中
usage = response.usage
print(f"Input tokens: {usage.input_tokens}")
print(f"Cache creation: {usage.cache_creation_input_tokens}") # 首次缓存
print(f"Cache read: {usage.cache_read_input_tokens}") # 缓存命中
# 缓存命中的 token 按 90% 折扣计费!
# OpenClaw 使用 CACHE_BOUNDARY 分隔系统上下文和用户输入
# 这个分隔符不仅用于缓存优化,也用于安全隔离
# 在 OpenClaw 的上下文构建中:
# 1. 系统文件(AGENTS.md, SOUL.md, USER.md 等)→ 缓存前缀
# 2. CACHE_BOUNDARY 标记
# 3. 用户消息 → 动态后缀
# 这确保了:
# - 系统上下文变化时才需要重新缓存
# - 用户消息变化不影响缓存命中
# - 同时实现了安全隔离(系统指令 vs 用户输入)
# 等效的 API 调用结构:
"""
messages = [
# 缓存前缀(系统上下文)
{"role": "system", "content": AGENTS_MD + SOUL_MD + USER_MD + ...},
# 缓存边界
# 动态后缀(用户消息)
{"role": "user", "content": latest_user_message}
]
"""
| 特性 | OpenAI 自动缓存 | Anthropic 显式缓存 |
|---|---|---|
| 启用方式 | 自动(≥1024 token) | 手动 cache_control |
| 缓存粒度 | 前缀匹配 | 指定块缓存 |
| 折扣幅度 | 缓存输入 = 正常输入 × 10% | 缓存输入 = 正常输入 × 10% |
| TTL | 5-10 分钟(非公开) | 5 分钟(ephemeral) |
| 最低 token 数 | 1024 | 1024(Sonnet)/ 2048(Opus) |
| 控制程度 | 低(自动) | 高(手动标记) |
| 多模态缓存 | 支持(图片需一致) | 支持 |
| 费用 | 缓存命中免费折扣;无额外费 | 缓存创建额外收费(25% 加价) |
class CachedConversation:
"""多轮对话缓存优化"""
def __init__(self, model: str = "gpt-4o"):
self.model = model
self.system_prompt = None
self.conversation_history = []
def set_system_prompt(self, prompt: str):
"""设置系统提示词(缓存前缀)"""
self.system_prompt = prompt
async def chat(self, user_message: str) -> str:
"""发送消息并优化缓存"""
# 构建消息列表:静态前缀 + 动态后缀
messages = []
# 缓存前缀:系统提示词(不变的部分)
if self.system_prompt:
messages.append({
"role": "system",
"content": self.system_prompt
})
# 动态后缀:对话历史 + 新消息
messages.extend(self.conversation_history)
messages.append({"role": "user", "content": user_message})
# 调用 API
response = await self._call_api(messages)
assistant_message = response.choices[0].message.content
# 更新历史
self.conversation_history.append({"role": "user", "content": user_message})
self.conversation_history.append({"role": "assistant", "content": assistant_message})
return assistant_message
async def _call_api(self, messages):
return await openai.AsyncOpenAI().chat.completions.create(
model=self.model,
messages=messages
)
# 长对话的缓存策略
class LongConversationOptimizer:
"""长对话优化——避免历史越来越长导致缓存 miss"""
def __init__(self, max_history_turns: int = 10, summary_model: str = "gpt-4o-mini"):
self.max_history = max_history_turns
self.summary_model = summary_model
async def optimize_history(self, messages: list) -> list:
"""优化对话历史,保持缓存命中"""
if len(messages) <= self.max_history * 2:
return messages
# 保留最近 N 轮 + 历史摘要
recent = messages[-(self.max_history * 2):]
old = messages[:-(self.max_history * 2)]
# 生成摘要
summary = await self._summarize(old)
# 返回:摘要 + 最近对话
return [
{"role": "system", "content": f"对话历史摘要:{summary}"},
*recent
]
async def _summarize(self, messages: list) -> str:
"""用小模型生成历史摘要"""
text = "\n".join(f"{m['role']}: {m['content'][:200]}" for m in messages)
response = await openai.AsyncOpenAI().chat.completions.create(
model=self.summary_model,
messages=[{
"role": "user",
"content": f"请用 200 字以内总结以下对话的关键信息:\n\n{text}"
}],
max_tokens=300
)
return response.choices[0].message.content
## Prompt Caching 最佳实践
### 1. 结构优化
✅ 系统提示词 → 工具定义 → few-shot 示例 → 对话历史 → 用户输入
❌ 用户输入 → 对话历史 → 系统提示词
### 2. 减少前缀变化
✅ 版本化的系统提示词(只在版本变更时修改)
❌ 每次请求都修改系统提示词(如加入时间戳)
### 3. 使用 cache_key(OpenAI)
✅ 为不同用户群/功能设置 cache_key
❌ 所有请求使用相同的 cache_key(超过15 req/min会溢出)
### 4. 工具定义缓存(Anthropic)
✅ 大量工具定义加 cache_control
❌ 每次请求动态增减工具列表
### 5. 多模态缓存
✅ 图片 URL 不变(缓存图片处理结果)
❌ 每次请求用不同编码的相同图片
### 6. 监控缓存命中率
✅ 跟踪 cache_read vs cache_creation 比例
❌ 假设缓存一定有效
### 7. 缓存 TTL 注意
- OpenAI: ~5-10 分钟无访问会失效
- Anthropic: 5 分钟
- 高频场景保持热度,低频场景考虑 Batch API