上线了一个客服机器人之后，你会发现一个规律：70%的问题都是重复的。“怎么退款”、“物流到哪了”、“密码忘了怎么办”——每天被问一万遍。每次都调一遍LLM，钱烧得心疼，延迟还高。

这篇文章分享一套基于向量相似度的语义缓存方案，实测能拦截80%的重复请求，响应时间从2秒降到50毫秒。

为什么需要语义缓存

传统缓存用精确匹配，“我的订单在哪"和"订单怎么查"意思一样但缓存命中率为零。语义缓存的核心思想是：把问题编码成向量，用相似度判断是否命中。

用户问："怎么退货"
缓存里有："如何办理退款"
→ 字面匹配：miss
→ 语义匹配：similarity=0.92，hit！直接返回缓存答案

架构设计

整体流程分三层：向量化层、缓存层、决策层。

flowchart TD A[用户提问] --> B[Embedding] B --> C[向量检索] C --> D{是否命中?} D -->|是| E[返回缓存] D -->|否| F[调用LLM] F --> G[写入缓存]

核心实现

向量化与存储

用sentence-transformers做embedding，Redis Stack做向量存储：

import numpy as np
from sentence_transformers import SentenceTransformer
import redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition

class SemanticCache:
    def __init__(self, redis_url="redis://localhost:6379",
                 model_name="BAAI/bge-small-zh-v1.5",
                 threshold=0.85):
        self.model = SentenceTransformer(model_name)
        self.redis = redis.from_url(redis_url)
        self.threshold = threshold
        self.dim = 512  # bge-small维度
        self._ensure_index()

    def _ensure_index(self):
        try:
            self.redis.ft("cache_idx").info()
        except:
            schema = (
                TextField("question"),
                TextField("answer"),
                VectorField("embedding", "HNSW", {
                    "TYPE": "FLOAT32",
                    "DIM": self.dim,
                    "DISTANCE_METRIC": "COSINE"
                })
            )
            self.redis.ft("cache_idx").create_index(
                schema,
                index_definition=IndexDefinition(prefix=["cache:"])
            )

    def _embed(self, text: str) -> np.ndarray:
        return self.model.encode(text, normalize_embeddings=True)

缓存查询

用KNN做近似最近邻搜索，O(log n)时间复杂度：

    def get(self, question: str) -> str | None:
        vec = self._embed(question)
        vec_bytes = vec.astype(np.float32).tobytes()

        results = self.redis.ft("cache_idx").search(
            f"*=>[KNN 1 @embedding $vec AS score]",
            query_params={"vec": vec_bytes}
        )

        if results.docs:
            score = float(results.docs[0].score)
            similarity = 1 - score  # Redis返回距离，转为相似度
            if similarity >= self.threshold:
                return results.docs[0].answer
        return None

缓存写入

写入时附带TTL，热点问题自动续期：

    def put(self, question: str, answer: str, ttl: int = 86400):
        vec = self._embed(question)
        key = f"cache:{hash(question)}"
        self.redis.hset(key, mapping={
            "question": question,
            "answer": answer,
            "embedding": vec.astype(np.float32).tobytes()
        })
        self.redis.expire(key, ttl)

集成到业务

cache = SemanticCache(threshold=0.85)

async def ask(question: str) -> str:
    # 先查缓存
    cached = cache.get(question)
    if cached:
        logger.info(f"Cache hit for: {question}")
        return cached

    # 缓存未命中，调LLM
    response = await llm.chat(question)

    # 写入缓存
    cache.put(question, response)
    return response

阈值调优

相似度阈值是最关键的参数。太低会误命中（答非所问），太高会漏掉语义相同的变体。

阈值    命中率    误命中率    推荐场景
0.95    45%      <0.1%     医疗/法律等高精度场景
0.85    72%      ~1%       客服/FAQ（推荐）
0.75    85%      ~5%       闲聊机器人
0.65    92%      ~15%      不推荐

实际调优方法：收集一批历史问答，人工标注哪些应该命中、哪些不应该，然后网格搜索找最优阈值。

def evaluate_threshold(pairs, thresholds):
    for t in thresholds:
        cache = SemanticCache(threshold=t)
        hits, false_hits = 0, 0
        for q, expected_answer in pairs:
            cached = cache.get(q)
            if cached:
                if cached == expected_answer:
                    hits += 1
                else:
                    false_hits += 1
        print(f"阈值={t}: 命中率={hits/len(pairs):.1%}, 误命中={false_hits/len(pairs):.1%}")

缓存失效策略

不是所有问题都适合缓存。时间敏感的问题需要特殊处理：

import re

NO_CACHE_PATTERNS = [
    r"今天", r"现在", r"最新", r"实时",
    r"\d{4}年\d{1,2}月",  # 具体日期
]

def should_cache(question: str) -> bool:
    for pattern in NO_CACHE_PATTERNS:
        if re.search(pattern, question):
            return False
    return True

另外加一个手动清缓存的接口，答案过期时主动删除：

    def invalidate(self, question: str):
        """精确删除某个缓存"""
        key = f"cache:{hash(question)}"
        self.redis.delete(key)

    def invalidate_pattern(self, keyword: str):
        """按关键词批量删除"""
        results = self.redis.ft("cache_idx").search(
            f"@question:*{keyword}*"
        )
        for doc in results.docs:
            self.redis.delete(doc.id)

性能数据

在生产环境跑了30天，数据如下：

总请求量：      120,000
缓存命中：       89,400 (74.5%)
LLM调用节省：    89,400 次
平均响应时间：
  缓存命中：     52ms
  缓存未命中：   1,840ms
API成本节省：    ~¥3,600/月（按GPT-4o计费）

写在最后

语义缓存本质上是用空间换时间、用向量距离换调用成本。实现不复杂，但效果很实在——尤其是客服、FAQ这类重复率高的场景。

唯一的坑是阈值调优，建议先上线观察一周，根据误命中反馈逐步收紧。另外记得给缓存加监控，命中率跌破60%就要排查是不是用户提问分布变了。