显式缓存最佳实践

核心要点

• 在需要缓存的消息上添加 "cache_control": {"type": "ephemeral"}，从 messages 数组开头到该标记位置之间的所有内容将被创建为缓存块。

• 缓存内容最少需要 1024 Token。

• 单次请求最多支持 4 个缓存标记。

• 缓存有效期为 5 分钟，每次命中自动续期。

• Tools 定义是 System Prompt 的一部分参与缓存计算，如果 Tools 改变则无法命中缓存。

快速开始

以下示例展示了显式缓存的基本使用方式：第一次请求创建缓存，第二次请求命中缓存。

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 需要缓存的长文本（需超过 1024 Token）
long_text_content = "<Your Long Text Here>" * 400

def get_completion(user_input):
    messages = [
        {
            "role": "system",
            "content": [
                {
                    "type": "text",
                    "text": long_text_content,
                    # 添加缓存标记：从 messages 开头到此位置的内容将被缓存
                    "cache_control": {"type": "ephemeral"},
                }
            ],
        },
        {"role": "user", "content": user_input},
    ]
    completion = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages,
        extra_body={"enable_thinking": False},
    )
    return completion

# 第一次请求：创建缓存
first = get_completion("请总结文档的核心要点")
print(f"创建缓存 Token：{first.usage.prompt_tokens_details.cache_creation_input_tokens}")
print(f"命中缓存 Token：{first.usage.prompt_tokens_details.cached_tokens}")

# 第二次请求：相同 system 内容，不同问题，命中缓存
second = get_completion("文档中提到了哪些注意事项？")
print(f"创建缓存 Token：{second.usage.prompt_tokens_details.cache_creation_input_tokens}")
print(f"命中缓存 Token：{second.usage.prompt_tokens_details.cached_tokens}")

import anthropic
import os

client = anthropic.Anthropic(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/apps/anthropic",
)

# 需要缓存的长文本（需超过 1024 Token）
long_text_content = "<Your Long Text Here>" * 400

def get_completion(user_input):
    response = client.messages.create(
        model="qwen3.7-max",
        max_tokens=1024,
        system=[
            {
                "type": "text",
                "text": long_text_content,
                # 添加缓存标记
                "cache_control": {"type": "ephemeral"},
            }
        ],
        messages=[
            {"role": "user", "content": user_input},
        ],
    )
    return response

# 第一次请求：创建缓存
first = get_completion("请总结文档的核心要点")
print(f"创建缓存 Token：{first.usage.cache_creation_input_tokens}")
print(f"命中缓存 Token：{first.usage.cache_read_input_tokens}")

# 第二次请求：命中缓存
second = get_completion("文档中提到了哪些注意事项？")
print(f"创建缓存 Token：{second.usage.cache_creation_input_tokens}")
print(f"命中缓存 Token：{second.usage.cache_read_input_tokens}")

运行上述代码，预期输出类似如下：

创建缓存 Token：2005
命中缓存 Token：0
创建缓存 Token：0
命中缓存 Token：2005

第一次请求时系统创建缓存块，第二次请求因 System Prompt 内容完全一致，成功命中缓存。命中缓存的 Token 仅按标准输入价格的 10% 计费。

确认缓存状态

请求完成后，可以通过响应中的 usage 字段确认缓存状态：

• cache_creation_input_tokens：本次请求新创建缓存的 Token 数。该值大于 0 说明创建了新缓存。

• cached_tokens（OpenAI 兼容）或 cache_read_input_tokens（Anthropic 兼容）：本次请求命中缓存的 Token 数。该值大于 0 说明成功命中缓存。

多轮对话场景

场景特点：

• 用户与模型进行多轮交互，每轮请求携带完整对话历史

• 典型应用：客服对话、知识问答、代码辅助等

最佳实践：在每次请求的最后一条消息上添加 cache_control 标记。每轮对话都会命中上一轮创建的缓存（对话历史部分），同时为下一轮创建包含当前完整对话的新缓存。

示例：

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# System Prompt（产品手册，需超过 1024 Token）
product_manual = """你是智能家居产品"百炼智家"的客服助手。以下是完整产品手册：

## 产品概述
百炼智家 是一款全屋智能中控设备，支持语音控制、场景联动、能耗管理等功能...

## 安装指南
1. 选择中心位置安装，确保 WiFi 信号覆盖...
2. 连接电源适配器（5V/2A）...

## 常见问题
Q: 设备无法连接 WiFi？A: 请确认路由器支持 2.4GHz...
""" * 80  # 重复以确保超过 1024 Token

messages = [{"role": "system", "content": product_manual}]

def chat(user_input):
    # 关键：在最后一条用户消息上添加 cache_control
    messages.append({
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": user_input,
                "cache_control": {"type": "ephemeral"},
            }
        ],
    })
    completion = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages,
        extra_body={"enable_thinking": False},
    )
    assistant_msg = completion.choices[0].message.content
    messages.append({"role": "assistant", "content": assistant_msg})

    usage = completion.usage
    created = usage.prompt_tokens_details.cache_creation_input_tokens
    cached = usage.prompt_tokens_details.cached_tokens
    print(f"  [缓存] 创建: {created} Token, 命中: {cached} Token")
    return assistant_msg

# 模拟多轮客服对话
print("用户: 百炼智家 支持哪些语音助手？")
print(f"客服: {chat('百炼智家 支持哪些语音助手？')[:60]}...\n")

print("用户: WiFi 连不上怎么办？")
print(f"客服: {chat('WiFi 连不上怎么办？')[:60]}...\n")

print("用户: 可以同时控制多少个设备？")
print(f"客服: {chat('可以同时控制多少个设备？')[:60]}...")

运行结果示例：

用户: 百炼智家支持哪些语音助手？
  [缓存] 创建: 8658 Token, 命中: 0 Token
客服: 百炼智家支持天猫精灵、小爱同学、Siri等主流语音助手...

用户: WiFi 连不上怎么办？
  [缓存] 创建: 149 Token, 命中: 8658 Token
客服: WiFi 连接问题请按以下步骤排查：1. 确认路由器支持 2.4GHz...

用户: 可以同时控制多少个设备？
  [缓存] 创建: 162 Token, 命中: 8807 Token
客服: 百炼智家最多可同时控制 256 个智能设备...

从第二轮开始，每轮对话都命中了上一轮创建的缓存（即之前完整的对话历史），同时创建包含当前轮新内容的缓存。对话轮数越多，节约越显著。

复杂工业级 Agent 场景

场景特点：

• 超长多轮对话，包含：长 System Prompt + skills/tools 说明 + project 上下文 + 用户对话 / 工具调用

• 不同部分的变化频率不同

• 典型应用：AI 编程助手（如 Claude Code、OpenClaw）、RAG 问答系统等

最佳实践：使用多个缓存标记（最多 4 个），分别标记不同稳定性层级的内容。每个标记需放在不同的 message 上才能作为独立截断点：

• System Prompt 加一个（几乎不变）

• skills/tools 说明加一个（可能出现组合变化）

• project 上下文加一个（可能切换/压缩）

• 用户对话 / 工具调用加一个（每轮增长）

示例：以下示例中，系统人设几乎不变（缓存标记 1），知识库随商品切换而变化（缓存标记 2），对话历史每轮增长（缓存标记 3）。注意：知识库放在 user message 中，以确保它有独立的缓存截断点——多条 system message 会被内部合并，无法作为独立截断点：

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 第一层：系统人设（几乎不变）
system_persona = """你是"百炼电子"的高级客服 AI 助手。你需要：
1. 基于知识库内容准确回答用户问题
2. 对于知识库中没有的信息，如实告知"我需要为您转接人工客服"
3. 始终保持专业、友善的语气
4. 如用户表示不满，先致歉再解决问题

以下是你的完整服务规范和话术指南：
""" + "服务规范详细说明..." * 200  # 确保超过 1024 Token

# 第二层：知识库检索结果（半稳定，随用户咨询的商品变化）
knowledge_base_product_a = """### 当前咨询商品：百炼 Pro Max 无线耳机
- SKU: BL-PM-2024
- 价格: 599 元
- 颜色: 极夜黑 / 星云白 / 冰晶蓝
- 续航: 主动降噪开启 8 小时，关闭 12 小时
- 防水等级: IPX5
- 保修: 1 年质保，支持 7 天无理由退换
- 当前库存: 极夜黑（充足）/ 星云白（少量）/ 冰晶蓝（缺货）
""" * 50  # 确保超过 1024 Token

def ask_about_product_a(user_question, history=None):
    if history is None:
        history = []
    messages = [
        {
            "role": "system",
            "content": [
                {
                    "type": "text",
                    "text": system_persona,
                    "cache_control": {"type": "ephemeral"},  # 缓存标记 1：系统人设
                }
            ],
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": f"以下是当前商品的知识库信息：\n{knowledge_base_product_a}",
                    "cache_control": {"type": "ephemeral"},  # 缓存标记 2：知识库
                }
            ],
        },
        {"role": "assistant", "content": "好的，我已了解该商品的详细信息，请问有什么可以帮您？"},
    ]
    # 添加对话历史
    messages.extend(history)
    # 添加当前问题（带缓存标记 3）
    messages.append({
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": user_question,
                "cache_control": {"type": "ephemeral"},  # 缓存标记 3：对话历史
            }
        ],
    })

    completion = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages,
        extra_body={"enable_thinking": False},
    )
    usage = completion.usage
    print(f"  创建缓存: {usage.prompt_tokens_details.cache_creation_input_tokens}, "
          f"命中缓存: {usage.prompt_tokens_details.cached_tokens}")
    return completion.choices[0].message.content

# 第一次：用户询问商品 A
print("Q1: 这款耳机有冰晶蓝色吗？")
a1 = ask_about_product_a("这款耳机有冰晶蓝色吗？")
print(f"A1: {a1}\n")

# 第二次：继续追问商品 A（系统人设 + 知识库均命中）
history = [
    {"role": "user", "content": "这款耳机有冰晶蓝色吗？"},
    {"role": "assistant", "content": a1},
]
print("Q2: 那什么时候能补货？")
a2 = ask_about_product_a("那什么时候能补货？", history)
print(f"A2: {a2}")

运行结果示例：

Q1: 这款耳机有冰晶蓝色吗？
  创建缓存: 7394, 命中缓存: 0
A1: 百炼 Pro Max 无线耳机确实有冰晶蓝配色，不过...目前该颜色暂时缺货...

Q2: 那什么时候能补货？
  创建缓存: 0, 命中缓存: 7394
A2: 关于冰晶蓝的具体补货时间...我需要为您转接人工客服...

第二轮中，从请求起始到缓存标记 2（系统人设 + 知识库 = 7,394 Token）的前缀完全一致，因此全部命中缓存。仅标记 2 之后的新增内容（对话历史 + 新问题）需要正常处理。

多标记缓存的命中逻辑：

• 用户继续追问同一商品：系统人设 + 知识库均未变化，命中缓存标记 2 处的缓存（最长前缀匹配），节约最大。

• 对话轮次增加：前面的内容（系统人设 + 知识库 + 历史对话）命中上一轮的缓存，仅新增部分需创建新缓存。

任务完成型场景（批量处理）

场景特点：

• 单轮对话，不需要上下文记忆

• 不变的长 System Prompt（任务说明）+ 变化的用户输入（待处理数据）

• 典型应用：文本分类、意图识别、数据提取、内容审核等

最佳实践：仅在 System Prompt 上添加 cache_control 标记。后续每次请求只要 System Prompt 不变，即可命中缓存。

示例：

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 长 System Prompt：包含详细的分类规则说明（需超过 1024 Token）
classification_prompt = """你是一个电商评论分类助手。请将用户评论分类为以下类别之一：
- 正面评价
- 负面评价
- 中性评价
- 咨询问题
- 投诉建议

只输出类别名称，不要其他内容。

以下是详细的分类规则和示例：
""" + """规则说明：
1. 正面评价：包含积极情感词汇（如"好"、"棒"、"满意"、"推荐"等），或表达对产品/服务的肯定。
2. 负面评价：包含消极情感词汇（如"差"、"失望"、"退货"等），或表达对产品/服务的不满。
3. 中性评价：情感倾向不明显，仅陈述事实。
4. 咨询问题：以疑问句形式出现，询问产品信息。
5. 投诉建议：表达改进意见或提出投诉。
""" * 100

# 待分类的评论列表（模拟批量处理场景）
reviews = [
    "这个产品太棒了，质量超好，强烈推荐！",
    "发货太慢了，等了一个星期才到，包装还破损了",
    "请问这个商品有红色的吗？尺码偏大还是偏小？",
    "建议增加更多尺码选择，M码对我来说太大了",
    "东西还行吧，中规中矩，没什么惊喜",
]

print("=== 批量文本分类（显式缓存）===")
for i, review in enumerate(reviews):
    completion = client.chat.completions.create(
        model="qwen3.7-max",
        messages=[
            {
                "role": "system",
                "content": [
                    {
                        "type": "text",
                        "text": classification_prompt,
                        "cache_control": {"type": "ephemeral"},  # 缓存分类规则
                    }
                ],
            },
            {"role": "user", "content": review},
        ],
    )
    result = completion.choices[0].message.content
    cached = completion.usage.prompt_tokens_details.cached_tokens
    created = completion.usage.prompt_tokens_details.cache_creation_input_tokens
    print(f"评论{i+1}: \"{review[:20]}...\" → {result}")
    print(f"  创建缓存: {created}, 命中缓存: {cached}")

运行结果示例：

评论1: "这个产品太棒了，质量超好，强烈推..." → 正面评价
  创建缓存: 5353, 命中缓存: 0
评论2: "发货太慢了，等了一个星期才到，包装..." → 负面评价
  创建缓存: 0, 命中缓存: 5353
评论3: "请问这个商品有红色的吗？尺码偏大还..." → 咨询问题
  创建缓存: 0, 命中缓存: 5353
评论4: "建议增加更多尺码选择，M码对我来说..." → 投诉建议
  创建缓存: 0, 命中缓存: 5353
评论5: "东西还行吧，中规中矩，没什么惊喜..." → 中性评价
  创建缓存: 0, 命中缓存: 5353

第一条请求创建缓存后，后续所有请求均命中缓存。处理 1000 条数据时，999 次请求的输入 Token 成本降低 90%。

Function Calling 时缓存工具列表

场景特点：

• 使用 Function Calling 功能，工具定义列表较长

• 工具定义在多次请求间保持不变

最佳实践：tools 参数的内容会作为 System Prompt 的一部分参与缓存。只需确保每次请求的工具定义完全一致（工具顺序、字段顺序、字段结构），并在 messages 的 content 上添加 cache_control 标记即可。

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 模拟长文本确保超过 1024 Token 阈值
long_text_content = "<Your Code Here>" * 400

# 工具定义：确保每次请求完全一致
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的当前天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_flights",
            "description": "搜索两个城市之间的航班",
            "parameters": {
                "type": "object",
                "properties": {
                    "origin": {"type": "string", "description": "出发城市"},
                    "destination": {"type": "string", "description": "目的城市"},
                    "date": {"type": "string", "description": "出发日期，格式 YYYY-MM-DD"}
                },
                "required": ["origin", "destination", "date"]
            }
        }
    }
]

def ask(user_input):
    messages = [
        {
            "role": "system",
            "content": [
                {
                    "type": "text",
                    "text": long_text_content,
                    # cache_control 只能加在 messages 的 content 上，不能加在 tools 上
                    "cache_control": {"type": "ephemeral"},
                }
            ],
        },
        {"role": "user", "content": user_input},
    ]
    completion = client.chat.completions.create(
        model="qwen3.7-max",
        messages=messages,
        tools=tools,
        extra_body={"enable_thinking": False},
    )
    usage = completion.usage
    print(f"  创建缓存: {usage.prompt_tokens_details.cache_creation_input_tokens}, "
          f"命中缓存: {usage.prompt_tokens_details.cached_tokens}")
    tool_calls = completion.choices[0].message.tool_calls
    if tool_calls:
        print(f"  调用工具: {[t.function.name for t in tool_calls]}")
    return completion

# 第一次请求：创建缓存（包含 tools 定义）
print("Q1: 北京今天天气怎么样？")
ask("北京今天天气怎么样？")

# 第二次请求：命中缓存
print("\nQ2: 帮我查明天从上海到北京的航班")
ask("帮我查明天从上海到北京的航班")

运行结果示例：

Q1: 北京今天天气怎么样？
  创建缓存: 1995, 命中缓存: 0
  调用工具: ['get_weather']

Q2: 帮我查明天从上海到北京的航班
  创建缓存: 0, 命中缓存: 1995
  调用工具: ['search_flights']