跳至主要內容

AI 应用的缓存设计——语义缓存和精确缓存的结合

老张2026/4/30大约 8 分钟

AI 应用的缓存设计——语义缓存和精确缓存的结合

有段时间我们的 AI 客服系统账单涨得很快。我去翻了翻调用日志，发现一个规律：每天大概有 40% 的问题，都是几十个高频问题的"近似变体"。

比如"如何修改密码"，用户的实际提问可能是：

  • "怎么改密码？"
  • "密码忘了怎么找回？"
  • "我想修改我的登录密码"
  • "password 怎么 change？"
  • "密码在哪里设置"

这五个问题的本质是一样的，但如果每次都调用 GPT-4 回答，就是五次 API 调用、五次 token 费用。

这就是引入语义缓存要解决的核心问题：相似问题只需要回答一次。

两种缓存策略的对比

在讲实现之前，先把两种缓存策略的适用场景说清楚。

精确缓存（Exact Match Cache）

基于请求内容的精确 Hash 缓存。逻辑很简单：对请求内容做 MD5，去 Redis 查，命中就返回，未命中就调用模型然后缓存结果。

适用场景：

  • API 文档问答、FAQ 机器人（问题文本高度标准化）
  • 系统 Prompt 固定、用户输入有限的场景
  • 对延迟要求极高、不允许向量计算开销的场景

局限：只要用户措辞稍有变化，就会 miss 缓存。对于自然语言输入，命中率通常很低（< 5%）。

语义缓存（Semantic Cache）

基于向量相似度的缓存。把历史问答的问题文本做 Embedding，存入向量库；新问题进来时，先 Embed，然后做向量相似度搜索，如果找到相似度 > 阈值的历史问题，直接返回历史答案。

适用场景：

  • 知识库问答、客服机器人（自然语言输入多样）
  • 问题领域相对固定（同一业务域内的问答）
  • 对 token 成本敏感的场景

局限：需要向量计算，有额外延迟（50-200ms）；相似度阈值难以调优，太高命中率低，太低可能返回错误答案。

两者结合：分层缓存策略

实际生产中，两者都要用，分层处理：

  1. L1：精确缓存（Redis Hash）

    • 命中条件：完全相同的请求 Hash
    • 延迟：< 1ms
    • 命中率：5-10%（针对完全重复的请求）

  2. L2：语义缓存（Redis + 向量相似度）

    • 命中条件：相似度 > 阈值（默认 0.92）
    • 延迟：50-150ms（向量计算 + Redis 查询）
    • 命中率：20-40%（针对语义相似的请求）

  3. L3：原始 LLM 调用

    • 命中条件：L1、L2 都未命中
    • 延迟：1-10 秒
    • 处理结果同时写入 L1 和 L2

语义缓存的核心实现

用 Redis 的向量搜索能力（RedisSearch with HNSW index）来实现语义缓存：

@Service
public class SemanticCacheService {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @Autowired
    private JedisPooled jedis; // 用 Jedis 直接操作 Redis Stack
    
    @Autowired
    private EmbeddingService embeddingService;
    
    // 语义缓存的向量索引配置
    private static final String CACHE_INDEX_NAME = "idx:semantic_cache";
    private static final String CACHE_KEY_PREFIX = "semantic_cache:";
    
    // 相似度阈值（余弦相似度，越接近 1 越相似）
    private static final double SIMILARITY_THRESHOLD = 0.92;
    
    // 缓存 TTL（24 小时，避免过期信息影响用户）
    private static final long CACHE_TTL_SECONDS = 24 * 60 * 60;
    
    /**
     * 初始化向量索引（应用启动时调用）
     */
    @PostConstruct
    public void initIndex() {
        try {
            // 如果索引已存在则跳过
            jedis.ftInfo(CACHE_INDEX_NAME);
        } catch (Exception e) {
            // 创建向量索引
            Schema schema = new Schema()
                .addTextField("tenant_id", 1.0)
                .addTextField("question", 1.0)
                .addTextField("answer", 1.0)
                .addNumericField("created_at")
                .addVectorField("embedding", 
                    Schema.VectorField.VectorAlgo.HNSW,
                    Map.of(
                        "TYPE", "FLOAT32",
                        "DIM", 1536,           // text-embedding-3-small 的维度
                        "DISTANCE_METRIC", "COSINE"
                    )
                );
            
            IndexDefinition indexDef = new IndexDefinition(IndexDefinition.Type.HASH)
                .setPrefixes(new String[]{CACHE_KEY_PREFIX});
            
            jedis.ftCreate(CACHE_INDEX_NAME, IndexOptions.defaultOptions().setDefinition(indexDef), schema);
            log.info("Created semantic cache index: {}", CACHE_INDEX_NAME);
        }
    }
    
    /**
     * 语义缓存查询
     * @return 命中的缓存答案，或 empty
     */
    public Optional<CacheEntry> get(String tenantId, String question) {
        // 1. 先做精确 Hash 匹配（L1 缓存）
        String exactKey = buildExactKey(tenantId, question);
        String exactCached = redisTemplate.opsForValue().get(exactKey);
        if (exactCached != null) {
            CacheEntry entry = deserialize(exactCached);
            entry.setHitType(CacheHitType.EXACT);
            incrementHitCounter(tenantId, CacheHitType.EXACT);
            return Optional.of(entry);
        }
        
        // 2. 语义向量搜索（L2 缓存）
        float[] queryEmbedding = embeddingService.embed(question);
        
        // 构建 KNN 搜索查询
        // 只在同一 tenant 范围内搜索
        Query query = new Query(
            String.format("(@tenant_id:{%s})=>[KNN 5 @embedding $vec AS score]", 
                escapeTenantId(tenantId))
        )
        .addParam("vec", floatsToBytes(queryEmbedding))
        .returnFields("question", "answer", "score")
        .setSortBy("score", true)  // 按分数升序（余弦距离越小越相似）
        .limit(0, 1);
        
        SearchResult searchResult = jedis.ftSearch(CACHE_INDEX_NAME, query);
        
        if (searchResult.getTotalResults() == 0) {
            incrementMissCounter(tenantId);
            return Optional.empty();
        }
        
        Document doc = searchResult.getDocuments().get(0);
        double score = Double.parseDouble(doc.getString("score"));
        
        // Redis 返回的是余弦距离（0 = 完全相同），转换为相似度
        double similarity = 1.0 - score;
        
        if (similarity < SIMILARITY_THRESHOLD) {
            log.debug("Semantic cache miss: best similarity {:.4f} < threshold {}", 
                similarity, SIMILARITY_THRESHOLD);
            incrementMissCounter(tenantId);
            return Optional.empty();
        }
        
        String cachedQuestion = doc.getString("question");
        String cachedAnswer = doc.getString("answer");
        
        log.debug("Semantic cache hit: similarity {:.4f}, matched question: {}", 
            similarity, cachedQuestion);
        
        incrementHitCounter(tenantId, CacheHitType.SEMANTIC);
        
        return Optional.of(CacheEntry.builder()
            .question(cachedQuestion)
            .answer(cachedAnswer)
            .similarity(similarity)
            .hitType(CacheHitType.SEMANTIC)
            .build());
    }
    
    /**
     * 将新的问答对写入缓存
     */
    public void put(String tenantId, String question, String answer) {
        // 生成 Embedding
        float[] embedding = embeddingService.embed(question);
        
        // 生成缓存 ID
        String cacheId = UUID.randomUUID().toString().replace("-", "");
        String redisKey = CACHE_KEY_PREFIX + cacheId;
        
        // 写入 Redis Hash（RedisSearch 会自动索引）
        Map<String, String> fields = new HashMap<>();
        fields.put("tenant_id", tenantId);
        fields.put("question", question);
        fields.put("answer", answer);
        fields.put("created_at", String.valueOf(System.currentTimeMillis()));
        fields.put("embedding", bytesToBase64(floatsToBytes(embedding)));
        
        jedis.hset(redisKey, fields);
        jedis.expire(redisKey, CACHE_TTL_SECONDS);
        
        // 同时写入精确 Hash 缓存（L1）
        String exactKey = buildExactKey(tenantId, question);
        redisTemplate.opsForValue().set(
            exactKey, 
            serialize(CacheEntry.builder().question(question).answer(answer).build()),
            CACHE_TTL_SECONDS,
            TimeUnit.SECONDS
        );
    }
    
    private String buildExactKey(String tenantId, String question) {
        String hash = DigestUtils.md5Hex(tenantId + ":" + question);
        return "exact_cache:" + hash;
    }
    
    private byte[] floatsToBytes(float[] floats) {
        ByteBuffer bb = ByteBuffer.allocate(floats.length * 4);
        bb.order(ByteOrder.LITTLE_ENDIAN);
        for (float f : floats) {
            bb.putFloat(f);
        }
        return bb.array();
    }
    
    // 命中率统计
    private void incrementHitCounter(String tenantId, CacheHitType type) {
        String key = String.format("cache:metrics:%s:%s:hit", tenantId, type.name().toLowerCase());
        redisTemplate.opsForValue().increment(key);
        redisTemplate.expire(key, 7, TimeUnit.DAYS);
    }
    
    private void incrementMissCounter(String tenantId) {
        String key = String.format("cache:metrics:%s:miss", tenantId);
        redisTemplate.opsForValue().increment(key);
        redisTemplate.expire(key, 7, TimeUnit.DAYS);
    }
}

在 LLM 调用链中集成语义缓存

@Service
public class CachedAIService {
    
    @Autowired
    private SemanticCacheService cacheService;
    
    @Autowired
    private AICapabilityService aiService;
    
    @Autowired
    private CostTrackingService costTracker;
    
    public ChatResponse chat(String tenantId, String question) {
        // 1. 查缓存
        Optional<CacheEntry> cached = cacheService.get(tenantId, question);
        
        if (cached.isPresent()) {
            CacheEntry entry = cached.get();
            
            // 记录缓存命中（节省了多少成本）
            costTracker.recordCacheHit(tenantId, entry.getHitType());
            
            return ChatResponse.builder()
                .content(entry.getAnswer())
                .fromCache(true)
                .cacheHitType(entry.getHitType().name())
                .cachedQuestion(entry.getQuestion()) // 告知调用方命中了哪个问题
                .similarity(entry.getSimilarity())
                .build();
        }
        
        // 2. 调用 LLM
        long startMs = System.currentTimeMillis();
        ChatResponse response = aiService.chat(ChatRequest.builder()
            .appId("knowledge-base")
            .userMessage(question)
            .build());
        long latencyMs = System.currentTimeMillis() - startMs;
        
        // 3. 写入缓存（异步，不阻塞响应）
        CompletableFuture.runAsync(() -> {
            try {
                cacheService.put(tenantId, question, response.getContent());
            } catch (Exception e) {
                log.warn("Failed to write semantic cache: {}", e.getMessage());
                // 缓存写入失败不影响主流程
            }
        });
        
        costTracker.recordLLMCall(tenantId, response.getUsage().getTotalTokens(), latencyMs);
        
        return response;
    }
}

语义缓存的痛点：相似度阈值怎么调

这是语义缓存最难的地方，没有通用答案，只有原则：

阈值太低（比如 0.80）

  • 优点：命中率高，省钱
  • 问题：会把不同问题的答案张冠李戴。比如"如何添加用户"和"如何删除用户"的相似度可能有 0.85，但答案完全相反

阈值太高（比如 0.98）

  • 优点：准确率高，几乎不会返回错误答案
  • 问题：命中率很低，接近精确匹配

我的调优方法：

  1. 从 0.95 开始，用生产数据跑一周，收集所有命中的问答对
  2. 人工抽样审核 50 对命中数据，检查语义是否真的等价
  3. 如果错误率 > 5%，把阈值提高 0.01，重复这个过程
  4. 找到"准确率 >= 99%"时对应的最低阈值，这就是你的目标阈值

我们系统里最终用的是 0.92，这个值下：

  • 命中率约 35%（相对于所有 LLM 调用）
  • 语义错误率约 0.3%（用户反馈"答非所问"的比例）

数据：引入语义缓存的实际效果

上线语义缓存的第一个完整月，数据如下：

总 AI 调用请求：428,000 次
L1（精确缓存）命中：18,200 次（4.3%）
L2（语义缓存）命中：131,900 次（30.8%）
实际 LLM 调用：277,900 次（65.0%）

节省的 token 消耗（相比无缓存）：
  节省 Prompt tokens：约 3,100 万
  节省 Completion tokens：约 1,800 万

API 费用对比：
  无缓存：$3,240/月（估算）
  有缓存：$2,106/月（实际）
  节省：$1,134/月（34.9%）

实际节省了 35% 的 API 费用，比预期的 "API 调用减少了 35%"更接近，因为 L1 精确缓存的命中通常是短问题（token 少），L2 语义缓存命中的问题 token 分布和总体差不多。

语义缓存的几个注意事项

1. 不是所有内容都适合缓存

以下内容不应该缓存（或者 TTL 设很短）：

  • 查询实时数据的问题："今天的股价是多少"、"现在几点了"
  • 带有用户个人信息的问题："我的订单状态是什么"
  • 需要考虑最新知识的问题："最新的法规变化"

可以在 Prompt 里添加时间敏感标记，或者对特定类型的问题跳过缓存：

public boolean shouldCache(String question) {
    // 检测时间敏感问题
    List<String> timeSensitivePatterns = List.of(
        "今天", "现在", "最新", "当前", "实时", "最近"
    );
    for (String pattern : timeSensitivePatterns) {
        if (question.contains(pattern)) return false;
    }
    return true;
}

2. 缓存污染问题

如果某个高频问题对应的答案后来更新了（比如产品功能变化），缓存里的旧答案会持续返回。解决方案：

  • 知识库更新时，清理相关缓存（通过 tag 或者 doc_id 关联）
  • 设置合理的 TTL，不要缓存太久

3. 个性化和缓存的冲突

如果不同用户对同一问题应该得到不同答案（比如角色权限不同），语义缓存要按用户角色分桶，不能跨角色共享缓存。

小结

语义缓存是 AI 应用成本优化最有效的手段之一，核心设计要点：

  1. L1（精确） + L2（语义）分层：覆盖不同的命中场景
  2. 相似度阈值经验调优：从 0.95 开始，用真实数据逐步下调
  3. 时间敏感内容排除：不要缓存会过期的答案
  4. 缓存失效机制：知识库变更时能清理相关缓存
  5. 异步写缓存：不让缓存写入影响正常响应延迟

引入语义缓存之后节省了 35% 的 API 费用，这个数字在不同的系统里会有差异，但通常都在 20-50% 之间。这是投入产出比极高的优化。

编辑此页
最近更新:
贡献者: xiaozhangge