显式缓存最佳实践

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

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]}...")

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: QW-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}")