9.3 缓存策略

在上一节中，我们讨论了如何通过请求队列和限流算法应对高并发。但无论限流做得多好，只要请求最终打到了 LLM API，就既慢又贵。有没有办法让一部分请求根本不走到 LLM？答案是缓存。

缓存是计算机科学中"用空间换时间"的经典策略，但在 Agent 系统中，它的价值被放大了——因为 LLM 调用既慢（秒级）又贵（按 Token 计费），缓存命中意味着延迟从秒级降到毫秒级，成本从几美分降到几乎为零。

9.3.1 为什么 Agent 系统需要缓存？

假设你的 Agent 服务每天处理 100 万次请求，其中 30% 是相似或重复的问题。如果每次请求都调用 GPT-4o（输入 $2.50/1M tokens，输出 $10/1M tokens），日均成本可能高达数千美元。而通过缓存，这 30% 的请求可以在毫秒级返回，成本几乎为零。

生活类比：缓存就像你手边的笔记本。同事问你公司 Wi-Fi 密码，你不需要每次都去翻 IT 部门的文档，因为上次问过了，记在笔记本上了。翻笔记本（缓存命中）比打电话问 IT（调用 LLM）快得多，而且不花钱。

缓存带来的三大收益：

• 降低延迟：缓存命中时响应时间从秒级降至毫秒级
• 降低成本：减少昂贵的 LLM API 调用
• 提升可靠性：LLM 服务不可用时，缓存可以作为降级方案

9.3.2 Prompt 缓存（Prefix Cache）

工作原理

Prompt 缓存（也称 Prefix Cache 或 KV Cache）是 LLM 推理层面的一种优化。当多个请求共享相同的 System Prompt 或长文档前缀时，推理引擎可以复用已计算的 Key-Value 张量，跳过重复的 Prefill 计算。

生活类比：就像你写了一篇万字长文的开头，发给三个不同的人续写。你不需要把开头重新打印三遍——开头只算一次的工夫，后面只需要写不同的结尾。

请求1: System Prompt (1000 tokens) + User Message (50 tokens)
        └── Prefill 计算 1000 tokens ──┘  Decode 50 tokens

请求2: System Prompt (1000 tokens) + User Message (30 tokens)
        └── 复用缓存的 KV 张量 ──────────┘  Decode 30 tokens
        ✨ 节省 1000 tokens 的 Prefill 计算！

各大厂商支持情况

厂商	缓存机制	最小缓存长度	价格折扣	说明
DeepSeek	Prefix Cache	64 tokens	输入价格 -90%	自动启用，无需配置
OpenAI	Prompt Caching	1024 tokens	输入价格 -50%	需手动标记缓存区域
Anthropic	Prompt Caching	1024 tokens	输入价格 -90%	需 cache_control 标记
阿里云 DashScope	Context Cache	1024 tokens	输入价格 -80%	显式配置
智谱 GLM	Prefix Cache	1024 tokens	输入价格 -50%	自动启用

实战：利用 OpenAI Prompt Caching

import openai                                  # 导入 OpenAI SDK
import time                                    # 导入时间模块

client = openai.OpenAI()                       # 创建客户端

# ============ 不使用缓存（首次请求） ============
start = time.time()                            # 记录开始时间
response1 = client.chat.completions.create(    # 调用 LLM
    model="gpt-4o",                            # 使用 GPT-4o
    messages=[
        {"role": "system", "content": "你是一个专业的法律顾问，请用中文回答。\n" * 50},  # 长 System Prompt
        {"role": "user", "content": "什么是合同法的基本原则？"}
    ]
)
time1 = time.time() - start                    # 计算耗时
print(f"第一次请求耗时: {time1:.2f}s, "        # 打印耗时
      f"输入 tokens: {response1.usage.prompt_tokens}, "
      f"缓存 tokens: {response1.usage.prompt_tokens_details.cached_tokens}")

# ============ 使用缓存（相同的 System Prompt） ============
start = time.time()                            # 记录开始时间
response2 = client.chat.completions.create(    # 第二次调用
    model="gpt-4o",                            # 同一个模型
    messages=[
        {"role": "system", "content": "你是一个专业的法律顾问，请用中文回答。\n" * 50},  # 完全相同的前缀
        {"role": "user", "content": "违约责任有哪些形式？"}  # 不同的用户问题
    ]
)
time2 = time.time() - start                    # 计算耗时
print(f"第二次请求耗时: {time2:.2f}s, "
      f"缓存 tokens: {response2.usage.prompt_tokens_details.cached_tokens}")
print(f"缓存加速: {time1/time2:.1f}x")         # 打印加速比

最佳实践

1. 将静态内容放在前面：System Prompt、文档、示例等固定内容放在消息列表的最前面
2. 避免在前缀中插入动态内容：时间戳、用户 ID 等动态内容会破坏缓存
3. 使用结构化分隔符：用 --- 或特定标记分隔缓存区域和非缓存区域
4. 监控缓存命中率：通过 cached_tokens 字段追踪缓存效果

9.3.3 Response 缓存

Response 缓存是最直观的缓存策略：将用户请求的哈希值作为 Key，LLM 的完整响应作为 Value 存储。

生活类比：就像 FAQ 页面。用户问"怎么重置密码"，你直接把写好的答案甩过去，不需要每次都找技术支持重新回答一遍。

import hashlib                                  # 导入哈希库
import json                                    # 导入 JSON 处理
import redis.asyncio as aioredis               # 导入异步 Redis
from typing import Optional                    # 导入类型注解
import time                                    # 导入时间模块

class ResponseCache:
    """基于 Redis 的 Response 缓存"""
    
    def __init__(self, redis_url: str = "redis://localhost", ttl:***@dataclass
class CacheEntry:
    """缓存条目"""
    question: str                              # 原始问题
    answer: str                                # 缓存的答案
    embedding: np.ndarray                      # 问题的向量表示
    timestamp: float                           # 存储时间
    hit_count: int = 0                         # 命中次数

class SemanticCache:
    """语义缓存：基于向量相似度匹配"""
    
    def __init__(
        self,
        redis_url: str = "redis://localhost",  # Redis 连接地址
        similarity_threshold:***@property
    def hit_rate(self) -> float:
        """缓存命中率"""
        total = self.hits + self.misses        # 总查询次数
        return self.hits / total if total > 0 else 0

# ============ 使用示例 ============
semantic_cache = SemanticCache(similarity_threshold=0.90)  # 相似度阈值 90%

async def smart_chat(question: str) -> str:
    """智能聊天：先查语义缓存，未命中再调 LLM"""
    # 第一步：查询语义缓存
    cached = await semantic_cache.query(question)  # 查缓存
    if cached:
        return cached                          # 命中，直接返回
    
    # 第二步：缓存未命中，调用 LLM
    response = await openai.AsyncOpenAI().chat.completions.create(
        model="gpt-4o",                        # 使用 GPT-4o
        messages=[{"role": "user", "content": question}]
    )
    answer = response.choices[0].message.content  # 提取回复
    
    # 第三步：写入缓存
    await semantic_cache.store(question, answer)  # 存入缓存
    
    return answer                              # 返回结果

# 测试
async def test():
    result1 = await smart_chat("Python中如何实现单例模式？")  # 第一次查询
    print(f"结果1: {result1[:100]}...\n")
    
    result2 = await smart_chat("Python单例模式怎么做？")     # 语义相似，应命中缓存
    print(f"结果2: {result2[:100]}...")
    
    print(f"\n缓存命中率: {semantic_cache.hit_rate:.1%}")   # 打印命中率

# asyncio.run(test())

GPTCache 开源方案

如果你不想从零实现语义缓存，可以使用 GPTCache——目前最成熟的 LLM 缓存开源项目。

from gptcache import cache                     # 导入 GPTCache 核心
from gptcache.adapter import openai            # 导入 OpenAI 适配器
from gptcache.embedding import Onnx            # 导入 ONNX 嵌入模型
from gptcache.manager import CacheBase, VectorBase, get_data_manager  # 数据管理
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation  # 相似度评估
import numpy as np                             # 导入 NumPy

# 初始化 GPTCache
onnx_embedding = Onnx()                        # 创建 ONNX 嵌入器
data_manager = get_data_manager(               # 创建数据管理器
    CacheBase("sqlite"),                       # 缓存存储用 SQLite
    VectorBase("faiss", dimension=onnx_embedding.dimension)  # 向量索引用 FAISS
)

cache.init(                                    # 初始化缓存
    embedding_func=onnx_embedding.to_embeddings,  # 嵌入函数
    data_manager=data_manager,                 # 数据管理器
    similarity_evaluation=SearchDistanceEvaluation(),  # 相似度评估器
    similarity_threshold=0.85                  # 相似度阈值
)

# 使用方式与普通 OpenAI 调用完全一致——GPTCache 自动拦截
response = openai.ChatCompletion.create(
    model="gpt-4o",                            # 指定模型
    messages=[{"role": "user", "content": "什么是深度学习？"}]
)

9.3.6 缓存策略对比

缓存类型	延迟	成本节省	命中率	实现复杂度	适用场景
Prefix Cache	减少 Prefill	高（-90%）	取决于前缀复用	零（自动）	长 System Prompt、长文档
Response Cache	毫秒级	中（-100%调用）	低（精确匹配）	低	FAQ、固定问题
Embedding Cache	毫秒级	中（-100% API）	高	中	RAG 系统
Semantic Cache	毫秒级	高（-100%调用）	中高	高	通用问答 Agent

常见误区

1. "缓存命中率越高越好"——不尽然。如果语义缓存的相似度阈值设得太低（比如 0.70），可能返回不相关的答案，用户体验反而变差。命中率和准确率需要平衡。

2. "Prefix Cache 不需要做任何事"——虽然很多厂商自动启用，但如果你在前缀中放了动态内容（如当前时间、用户 ID），缓存就不会命中。合理安排消息顺序很重要。

3. "Response Cache 适用于所有场景"——对于需要实时信息的请求（如"今天的天气"），缓存结果反而是错误的。缓存策略需要区分"可缓存"和"不可缓存"的请求。

4. "缓存不需要过期"——模型更新后，旧缓存的结果可能不再准确。设置合理的 TTL（如 1~7 天）是必要的。

5. "语义缓存可以替代 LLM"——语义缓存只适合"有标准答案"的问题。对于需要个性化、需要上下文推理的请求，缓存可能返回不合适的答案。

6. "缓存只是性能优化"——缓存还有一层重要的战略意义：当 LLM 服务不可用时，缓存是降级方案的第一选择（下一节会详细讨论）。

本节小结

要点	说明
Prefix Cache	推理层优化，自动生效，将静态内容放在前面以最大化命中率
Response Cache	精确匹配，适合 FAQ 场景，实现简单但命中率有限
Embedding Cache	缓存文本向量，RAG 系统的必备优化
Semantic Cache	基于语义相似度，命中率最高、实现最复杂、收益最大
相似度阈值	0.85~0.92 是常用范围，过高命中率低，过低可能返回不准确的缓存
GPTCache	最成熟的开源 LLM 缓存方案，生产环境推荐使用

缓存策略让系统更快更省，但缓存只是"锦上添花"——它能让一部分请求秒回，却无法保证 LLM 服务永远可用。当 LLM API 真的挂了怎么办？下一节我们将讨论降级与容错：如何在风雨中让系统继续提供服务。

---

下一节：9.4 降级与容错——模型切换、熔断器、重试策略，让系统在故障中依然坚挺。