Redis 缓存策略：降低 LLM 调用成本

LLM API 调用是 AI 应用最主要的成本来源。GPT-4o 每百万输入 token 约 2.5 美元，一个中等规模的企业应用每天可能消耗数百万 token。更头疼的是，相当一部分调用是重复的：同样的问题被不同用户反复提问，同样的文档被反复 Embedding，同样的系统 prompt 每次都要计入 token 计数。

缓存是降低这部分成本最直接有效的手段。本章系统介绍 AI 应用中的缓存机会点、实现方案和失效策略。

1.1 AI 应用中的缓存机会

三种Redis缓存模式对比——精确匹配、语义相似、分层缓存的命中率与成本节省关系

不同类型的 AI 操作，缓存收益差异很大：

操作类型	重复率	缓存难度	推荐策略
相同 prompt 的 LLM 调用	高（FAQ 场景可达 60%+）	低（精确匹配）	Exact Match 缓存
语义相似问题	中（同类问题变体多）	中（需向量相似度）	Semantic Cache
文档 Embedding 计算	极高（同文档多次处理）	低（文档 hash 匹配）	Embedding 缓存
会话历史	高（同一会话多轮）	低	Session 缓存
工具调用结果（天气、搜索）	中	低	TTL 缓存

1.2 Exact Match 缓存

最简单也最高效的缓存策略：完全相同的 prompt 直接返回缓存结果，不调用 LLM。

python

# llm_cache.py
import hashlib
import json
import redis
from openai import OpenAI
from typing import Optional

redis_client = redis.Redis(host="localhost", port=6379, db=3, decode_responses=True)
openai_client = OpenAI()


def _make_cache_key(model: str, messages: list, **kwargs) -> str:
    """
    生成缓存 key。
    把所有影响输出的参数都纳入 key，避免参数不同却命中相同缓存。
    temperature=0 时输出确定性高，适合缓存；temperature>0 时输出有随机性，
    是否缓存取决于业务场景。
    """
    payload = {
        "model": model,
        "messages": messages,
        # 只把影响输出的参数纳入 key
        "temperature": kwargs.get("temperature", 1.0),
        "max_tokens": kwargs.get("max_tokens"),
    }
    content = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return f"llm:exact:{hashlib.md5(content.encode()).hexdigest()}"


def cached_chat_completion(
    model: str,
    messages: list,
    ttl: int = 3600,  # 默认缓存 1 小时
    cache_enabled: bool = True,
    **kwargs,
) -> dict:
    """
    带精确匹配缓存的 LLM 调用。
    cache_enabled=False 可以在调试或需要新鲜结果时跳过缓存。
    """
    if not cache_enabled:
        return _call_llm(model, messages, **kwargs)

    cache_key = _make_cache_key(model, messages, **kwargs)

    # 尝试从缓存读取
    cached = redis_client.get(cache_key)
    if cached:
        result = json.loads(cached)
        result["_cached"] = True  # 标记为缓存结果，方便监控统计
        return result

    # 缓存未命中，调用 LLM
    result = _call_llm(model, messages, **kwargs)
    result["_cached"] = False

    # 写入缓存，temperature=0 的确定性输出可以缓存更长时间
    actual_ttl = ttl * 24 if kwargs.get("temperature", 1.0) == 0 else ttl
    redis_client.setex(cache_key, actual_ttl, json.dumps(result, ensure_ascii=False))

    return result


def _call_llm(model: str, messages: list, **kwargs) -> dict:
    """实际调用 LLM，返回统一格式的结果。"""
    response = openai_client.chat.completions.create(
        model=model,
        messages=messages,
        **kwargs,
    )
    return {
        "content": response.choices[0].message.content,
        "model": response.model,
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
        },
    }

1.3 Semantic Cache（语义缓存）

精确匹配缓存有一个明显局限：用户用不同措辞提同一个问题，缓存就失效了。"北京今天天气怎么样"和"北京今天的天气如何"语义相同，但 MD5 完全不同。

语义缓存的思路：把问题向量化，存入 Redis（搭配 RediSearch），新问题进来先搜索相似向量，相似度超过阈值就返回对应的缓存结果。

python

# semantic_cache.py
import numpy as np
import json
import redis
from openai import OpenAI

# 需要安装 redis[hiredis] 和 redis-py >= 4.6
# 并在 Redis 中加载 RediSearch 模块
redis_client = redis.Redis(host="localhost", port=6379, db=4)
openai_client = OpenAI()

SIMILARITY_THRESHOLD = 0.92  # 相似度阈值，越高越严格
EMBEDDING_DIM = 1536          # text-embedding-3-small 的维度


def setup_semantic_cache_index():
    """
    创建 RediSearch 向量索引，只需运行一次。
    使用 HNSW 算法，支持近似最近邻搜索，速度远快于暴力扫描。
    """
    try:
        redis_client.execute_command(
            "FT.CREATE", "semantic_cache_idx",
            "ON", "HASH",
            "PREFIX", "1", "sem_cache:",
            "SCHEMA",
            "question", "TEXT",
            "embedding", "VECTOR", "HNSW", "6",  # HNSW（Hierarchical Navigable Small World，分层可导航小世界图），一种高效的近似最近邻搜索算法，比暴力扫描快数百倍
            "TYPE", "FLOAT32",
            "DIM", str(EMBEDDING_DIM),
            "DISTANCE_METRIC", "COSINE",
        )
    except Exception:
        pass  # 索引已存在时忽略错误


def get_embedding(text: str) -> np.ndarray:
    """获取文本的向量表示。实际项目中应该对这个调用本身也做缓存（见下节）。"""
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return np.array(response.data[0].embedding, dtype=np.float32)


def semantic_cache_get(question: str) -> Optional[str]:
    """
    语义缓存查询：找到语义最相似的缓存问题，相似度超阈值则返回缓存答案。
    """
    query_embedding = get_embedding(question)

    # KNN 向量搜索，找最相似的 1 条
    query = (
        f"*=>[KNN 1 @embedding $vec AS score]"
    )
    results = redis_client.execute_command(
        "FT.SEARCH", "semantic_cache_idx", query,
        "PARAMS", "2", "vec", query_embedding.tobytes(),
        "RETURN", "3", "question", "answer", "score",
        "SORTBY", "score",
        "DIALECT", "2",
    )

    if results[0] == 0:  # 没有任何缓存
        return None

    # results 格式：[count, key, [field, value, ...], ...]
    fields = dict(zip(results[2][::2], results[2][1::2]))
    score = float(fields.get(b"score", 0))

    # COSINE 距离：0 表示完全相同，2 表示完全相反
    # 转换为相似度：1 - distance/2
    similarity = 1 - score / 2

    if similarity >= SIMILARITY_THRESHOLD:
        return fields.get(b"answer", b"").decode()

    return None


def semantic_cache_set(question: str, answer: str, ttl: int = 86400):
    """将问答对存入语义缓存。"""
    embedding = get_embedding(question)
    import hashlib
    cache_key = f"sem_cache:{hashlib.md5(question.encode()).hexdigest()}"

    redis_client.hset(cache_key, mapping={
        "question": question,
        "answer": answer,
        "embedding": embedding.tobytes(),
    })
    redis_client.expire(cache_key, ttl)

1.4 Embedding 缓存

Embedding 计算往往是最容易被忽视的成本。一个 RAG 系统每次处理查询都要计算 query 的 Embedding，如果相同的查询被反复提交，就是重复付钱。更重要的是，文档的 Embedding 一旦计算完成，只要文档不变就可以永久复用。

python

# embedding_cache.py
import hashlib
import json
import numpy as np
import redis
from openai import OpenAI

redis_client = redis.Redis(host="localhost", port=6379, db=5)
openai_client = OpenAI()


class CachedEmbedder:
    """
    带 Redis 缓存的 Embedding 计算器。
    文档 Embedding 使用内容 hash 作为 key，内容不变则永久复用。
    Query Embedding 设置较短 TTL，因为 query 种类繁多，内存有限。
    """

    def __init__(self, model: str = "text-embedding-3-small"):
        self.model = model

    def embed_document(self, text: str, doc_id: str = None) -> np.ndarray:
        """
        文档 Embedding 缓存。
        优先用 doc_id 作为 key（语义清晰），没有 doc_id 则用内容 hash。
        文档内容不变，Embedding 就不变，设置极长 TTL（30天）。
        """
        content_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
        cache_key = f"emb:doc:{doc_id or content_hash}"

        cached = redis_client.get(cache_key)
        if cached:
            return np.frombuffer(cached, dtype=np.float32)

        embedding = self._compute_embedding(text)
        # 文档 Embedding 不轻易失效，TTL 30 天
        redis_client.setex(cache_key, 86400 * 30, embedding.tobytes())
        return embedding

    def embed_query(self, text: str) -> np.ndarray:
        """
        查询 Embedding 缓存。
        TTL 1 小时：查询词汇多变，长期缓存意义不大，但短期内重复提问很常见。
        """
        cache_key = f"emb:query:{hashlib.md5(text.encode()).hexdigest()}"

        cached = redis_client.get(cache_key)
        if cached:
            return np.frombuffer(cached, dtype=np.float32)

        embedding = self._compute_embedding(text)
        redis_client.setex(cache_key, 3600, embedding.tobytes())
        return embedding

    def embed_batch(self, texts: list[str]) -> list[np.ndarray]:
        """
        批量 Embedding，先查缓存，只对未命中的文本调用 API，降低 API 调用次数。
        """
        results = [None] * len(texts)
        uncached_indices = []
        uncached_texts = []

        # 先批量查缓存
        for i, text in enumerate(texts):
            cache_key = f"emb:doc:{hashlib.sha256(text.encode()).hexdigest()[:16]}"
            cached = redis_client.get(cache_key)
            if cached:
                results[i] = np.frombuffer(cached, dtype=np.float32)
            else:
                uncached_indices.append(i)
                uncached_texts.append(text)

        # 只对未命中的文本批量调用 API
        if uncached_texts:
            response = openai_client.embeddings.create(
                model=self.model,
                input=uncached_texts,
            )
            for idx, embedding_obj in zip(uncached_indices, response.data):
                embedding = np.array(embedding_obj.embedding, dtype=np.float32)
                results[idx] = embedding
                # 写入缓存
                text = texts[idx]
                cache_key = f"emb:doc:{hashlib.sha256(text.encode()).hexdigest()[:16]}"
                redis_client.setex(cache_key, 86400 * 30, embedding.tobytes())

        return results

    def _compute_embedding(self, text: str) -> np.ndarray:
        response = openai_client.embeddings.create(model=self.model, input=text)
        return np.array(response.data[0].embedding, dtype=np.float32)

1.5 缓存失效策略

缓存最难的部分不是命中，而是失效。AI 应用的缓存失效需要应对几种场景：

TTL（Time-To-Live）：最简单的失效方式，适合有自然时效性的数据（天气、新闻摘要）。

LRU（Least Recently Used）：Redis 内置，通过配置 maxmemory-policy allkeys-lru 自动淘汰最久未使用的缓存。适合内存有限时的自动管理。

主动失效：内容变化时立即删除对应缓存，适合知识库更新场景。

python

# cache_invalidation.py

class CacheManager:
    """统一管理缓存失效，避免散落在各处的 redis.delete() 调用。"""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def invalidate_document(self, doc_id: str):
        """
        文档更新时主动失效：
        1. 删除文档的 Embedding 缓存
        2. 删除所有与该文档相关的 LLM 响应缓存
        这里用 Redis SCAN 代替 KEYS，避免在大 key 空间下阻塞。
        """
        # 删除文档 Embedding
        self.redis.delete(f"emb:doc:{doc_id}")

        # 扫描并删除所有引用了该文档的 LLM 缓存
        # 实际实现中应在写入缓存时用 Set 记录 doc_id -> [cache_keys] 的映射
        # 这里展示简化版本
        cursor = 0
        pattern = f"llm:*"
        deleted_count = 0
        while True:
            cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
            for key in keys:
                value = self.redis.get(key)
                if value and doc_id in value.decode():
                    self.redis.delete(key)
                    deleted_count += 1
            if cursor == 0:
                break

        return deleted_count

    def get_cache_stats(self) -> dict:
        """获取缓存统计信息，用于监控成本节省效果。"""
        info = self.redis.info("stats")
        hits = info.get("keyspace_hits", 0)
        misses = info.get("keyspace_misses", 0)
        total = hits + misses
        return {
            "hit_rate": hits / total if total > 0 else 0,
            "total_requests": total,
            "hits": hits,
            "misses": misses,
        }

1.6 Redis 数据结构选择

数据结构	适用缓存场景	优势
String	LLM 响应、Embedding 二进制	简单，支持 TTL，原子操作
Hash	会话状态（多个字段）	字段级读写，节省内存
List	流式 token 队列	有序，支持阻塞读取
Sorted Set	热门问题排行、LRU 手动实现	按 score 排序，范围查询
Stream	事件日志、审计记录	持久化，消费者组

会话状态用 Hash 比用 String 更高效：

python

# 用 String 存会话（低效）：每次都要序列化整个会话
redis_client.setex(f"session:{session_id}", 3600, json.dumps(all_messages))

# 用 Hash 存会话（高效）：可以只读取最近几条，不用加载整个会话
redis_client.hset(f"session:{session_id}", f"msg:{timestamp}", json.dumps(message))
redis_client.expire(f"session:{session_id}", 3600)
# 读取最近 10 条时，只需知道 key，无需反序列化全部数据

1.7 小结

AI 应用的缓存策略应该分层次实施：优先做 Embedding 缓存（成本高、重复率高、实现简单），再做 Exact Match LLM 缓存（FAQ 场景效果显著），最后视业务需要引入 Semantic Cache（需要额外的向量搜索基础设施）。

缓存命中率和成本节省需要持续监控，cache_stats 中的命中率是核心指标。命中率低于 20% 时，需要重新审视缓存 key 的设计和 TTL 策略。

下一步是将缓存与监控告警系统打通：当 LLM API 成本突然飙升时，能立即发现是缓存失效了、还是流量真的增长了，而不是月底看账单时才发现问题。