Marico's space

花半天写个缓存，一周省回成本：100行代码的 LLM 缓存方案

AI技术与应用 2026-04-27 15:07:41 1

前几天看到一张账单，一周下来 LLM 调用费用近 1.5 万美元——一个客服 Copilot，把同一套 FAQ 上下文反复塞进 prompt，一周之内同一段输入 token 被重复付费了约 41000 次。🤯

第一反应当然是去找 GPTCache 之类的中间件。但看了一眼它的依赖图，陷入了沉思……真的需要那么多轮子吗？

于是我认真想了想：100 行 Python 能搞定吗？

答案是：能。而且效果还不错。

先说结论：在典型的聊天式 Copilot、RAG 文档问答、内部工具等场景下，花半天写一个轻量缓存，一周之内就能回本，之后持续省钱。

各家的"免费午餐"是什么？

动手之前，先搞清楚免费午餐有哪些。

OpenAI 的自动 Prompt Caching（提示词缓存）：触发门槛 1024 tokens，每 128 tokens 为一个增量单位命中，对命中的输入 token 给你打五折——零代码改动。

Anthropic 的提示词缓存需要手动设置 cache breakpoints，属于 opt-in（主动开启）模式。

这两种方案都适合「前缀稳定」的场景：系统提示词很长、用户消息很短。但下面这些情况它俩都帮不上忙：

输入本身就短（OpenAI 要求 1024 tokens 以上才触发）
两句话意思一样但字不同——比如「密码怎么重置」和「我忘记密码了」
用私有部署模型，没有任何缓存层
你想缓存完整响应，而不仅仅是输入前缀

最后一条是关键。厂商的 Prompt Caching 只帮你省了输入 token 的费用，而一个真正的响应缓存，能把整个调用——包括输出 token——全部省掉。

缓存第一层：精确匹配 + TTL + LRU

先上代码（建议结合注释看）：

from __future__ import annotations
import hashlib, json, time
from collections import OrderedDict
from dataclasses import dataclass
from threading import Lock
from typing import Callable, Optional

@dataclass(frozen=True)
class CacheKey:
    prompt: str
    system: str
    model: str
    temperature: float
    max_tokens: int

    def hash(self) -> str:
        payload = json.dumps(
            {"p": self.prompt, "s": self.system, "m": self.model,
             "t": round(self.temperature, 4), "mt": self.max_tokens},
            sort_keys=True, ensure_ascii=False
        ).encode("utf-8")
        return hashlib.blake2b(payload, digest_size=16).hexdigest()

@dataclass
class CacheEntry:
    response: str
    embedding: Optional[list[float]]
    created_at: float
    hits: int = 0

class LLMCache:
    def __init__(self, max_size: int = 5000, ttl_seconds: int = 3600):
        self._store: OrderedDict[str, CacheEntry] = OrderedDict()
        self._max_size = max_size
        self._ttl = ttl_seconds
        self._lock = Lock()

    def _is_fresh(self, entry: CacheEntry) -> bool:
        return (time.time() - entry.created_at) < self._ttl

    def get_exact(self, key: CacheKey) -> Optional[str]:
        h = key.hash()
        with self._lock:
            entry = self._store.get(h)
            if entry and self._is_fresh(entry):
                self._store.move_to_end(h)  # LRU 更新
                entry.hits += 1
                return entry.response
            if entry:
                del self._store[h]  # 过期了直接删
        return None

    def put(self, key: CacheKey, response: str,
            embedding: Optional[list[float]] = None) -> None:
        h = key.hash()
        with self._lock:
            self._store[h] = CacheEntry(response=response, embedding=embedding, created_at=time.time())
            self._store.move_to_end(h)
            while len(self._store) > self._max_size:
                self._store.popitem(last=False)  # 踢掉最老的

5 个值得关注的细节：

CacheKey 包含 temperature——temperature 0 和 0.7 的调用不是同一回事
不包含 request ID / user ID / 时间戳——加上会让每次请求 key 都不同，缓存形同虚设
OrderedDict + move_to_end = 4 行 LRU——不需要 cachetools 或 Redis
blake2b 摘要长度 16 字节——32 位十六进制 key，冲突风险可忽略，速度比 sha256 快
Lock 是必要的——OrderedDict 的 move_to_end + popitem 组合在并发场景下有竞争风险

缓存第二层：语义相似度兜底

精确匹配层抓重复，语义兜底层抓近义问题——同一个问题换种说法，这是实际生产中收益最大的部分。

import math

def cosine(a, b):
    dot = sum(x*y for x,y in zip(a,b))
    na = math.sqrt(sum(x*x for x in a))
    nb = math.sqrt(sum(y*y for y in b))
    if na == 0 or nb == 0: return 0.0
    return dot / (na * nb)

class SemanticLLMCache(LLMCache):
    def __init__(self, embedder, similarity_threshold: float = 0.93, *kwargs):
        super().__init__(*kwargs)
        self._embedder = embedder
        self._threshold = similarity_threshold

    def get(self, key: CacheKey) -> Optional[str]:
        hit = self.get_exact(key)
        if hit is not None: return hit
        query_emb = self._embedder(key.prompt)
        best_score, best_resp = 0.0, None
        with self._lock:
            for entry in self._store.values():
                if not self._is_fresh(entry) or entry.embedding is None: continue
                score = cosine(query_emb, entry.embedding)
                if score > best_score: best_score, best_resp = score, entry.response
            if best_score >= self._threshold: return best_resp
        return None

    def put_with_embedding(self, key: CacheKey, response: str) -> None:
        emb = self._embedder(key.prompt)
        self.put(key, response, embedding=emb)

0.93 这个阈值不是拍脑袋的。低于 0.9 开始出现「怎么重置密码」和「怎么暂停账户」答非所问的情况；高于 0.96 语义相似命中极少，缓存形同虚设。大部分团队在自己数据上测完会落在 0.92~0.94 这个区间。

线性扫描没问题。5000 条、1536 维的 embedding 做余弦扫描，单核 5ms 以内。如果规模到了 50000 条以上再考虑换 FAISS 或 HNSW，不要提前优化。

实际能省多少？

按 GPT-4o 公开定价（$2.50/M 输入，$10/M 输出）估算：一个客服 Copilot，每周 8000 次对话，平均每次 6 轮（约 48000 次调用），平均输入 1400 tokens，输出 280 tokens。单次约 $0.0063，不缓存每周约 $302。

加精确匹配层，命中率约 22%（重复问答、重试、刷新等），降到约 $236/周。

再加语义层（阈值 0.93），综合命中率到约 41%，降到约 $178/周。embedding 成本约 $4/周。净省约 $120/周——一周内回本，之后每周持续省。

这个缓存不管用的场景

确定性假设不成立时——如果你在用 temperature 0.9 追求创意输出，缓存是错误的工具。只缓存那些设计上是确定性的调用：分类、提取、结构化输出、FAQ 类查询。

底层知识变化时——如果缓存说「我们的退款窗口是 30 天」但你政策改成了 14 天，缓存会持续给出错误答案直到 TTL 过期。根据领域变化频率设 TTL：每周变化的支持内容设 1 小时，稳定产品文档设 24 小时，涉及用户状态的设 5 分钟。

还不理解监控的重要性——缓存之上你仍然需要 traces（调用链）、hit-rate 指标、缓存大小和淘汰率报警，以及看起来像省钱实际是正确性回归的故障模式。

集成进去有多简单？

整个接入代码一行：

response = cached_call(cache, key, lambda k: client.complete(...))

缓存实例放模块级别，embedder 是 embedding provider 的薄封装，call_llm 是你原来就在用的函数。如果你已经接了 GPTCache 或 LangChain 缓存而且跑得好，别动。如果每周账单四位数、在考虑换 vendor，先把这 100 行写了。

原文：The 100-Line LLM Cache That Pays For Itself in a Week