LLM响应缓存实战：为OpenAI兼容API构建精确匹配缓存

在真实的生产应用里，有相当比例的大模型请求其实是完全重复的——相同的系统提示词、相同的用户问题、相同的参数，仅仅因为页面重新渲染、测试套件重跑、定时任务重试，或者两个用户问了同一个FAQ而被再次发送。

OpenAI兼容网关运营者Aman Kumar观察到了这个流量规律：每一次重复请求都在消耗延迟，并且根据计费方式的不同，还在消耗金钱或配额。而在客户端加一层缓存，就能解决其中大部分问题。本文梳理他实际使用的方案：缓存键里该放什么、数据存在哪、如何处理流式响应，以及哪些场景根本不该缓存。

Caching LLM Responses 原文截图

客户端缓存 ≠ 服务商的 Prompt Caching

不少人会把这件事和服务商提供的"prompt caching"混为一谈。一些厂商会对长而重复的前缀（比如庞大的系统提示词）按折扣计费，这确实有用，但那是另一个功能：模型仍然要跑，你仍然要等待生成，请求仍然会发出去。

客户端响应缓存则是彻底跳过请求。如果输入完全相同，而你也接受一个完全相同的回答，那么存储的答案可以在毫秒级返回。两者可以叠加使用——服务商缓存处理那些只在末尾不同的请求，你自己的缓存处理那些完全不变的请求。理解这个区别，是避免重复投入的前提。

服务商侧的 Prompt Caching（如 OpenAI、Anthropic 的实现）本质上是在模型推理的 KV Cache 层面做的优化：对于前缀相同的请求，推理引擎可以复用已计算过的注意力键值对，从而减少重复计算量，进而给出 token 折扣（通常为 50%–75% 的输入 token 费用减免）。但这项优化对调用方是透明的——请求仍然要发送、网络往返仍然存在、模型仍然要运行剩余部分的推理、延迟仍然在秒级。

客户端响应缓存则工作在完全不同的层面：它在请求离开客户端之前就做出决策，匹配成功时延迟可以低至个位数毫秒。两者的适用粒度也不同：KV Cache 擅长处理"长系统提示词 + 不同用户问题"这类前缀重复的场景，而客户端缓存只对完全相同的输入有效。正因如此，两者叠加使用能覆盖更广的重复模式。

缓存键必须包含哪些字段

大多数自制缓存出错的根源都在这里。如果缓存键漏掉了任何会影响输出的因素，你就会返回错误答案。按照作者的做法，键是以下内容的哈希值：

base URL（这样 staging 和 production、或两个不同服务商永远不会共享条目）
model ID，与发送时完全一致
完整的 messages 数组，包含系统提示词
temperature、top_p、max_tokens、stop，以及若使用了 seed
tools、tool_choice 和 response_format
一个版本字符串，每次改动提示词或后处理逻辑时递增

一个关键的安全细节：API key 绝不能以明文进入缓存键，也不该存储在缓存的任何地方。如果需要按租户隔离，应改用租户 ID。

序列化时必须排序键名，这样 {"a":1,"b":2} 和 {"b":2,"a":1} 才能哈希到同一个值：

import hashlib, json

CACHE_VERSION = "v3"

def cache_key(base_url: str, payload: dict) -> str:
    material = {
        "v": CACHE_VERSION,
        "base_url": base_url.rstrip("/"),
        **payload,
    }
    blob = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

一个最小可用的 SQLite 缓存

对于单进程或小型服务，SQLite 就足够了。它能在重启后存活，也不需要额外基础设施。

import sqlite3, time
from openai import OpenAI

client = OpenAI(base_url=BASE_URL, api_key="sk-your-key")

db = sqlite3.connect("llm_cache.sqlite3", check_same_thread=False)
db.execute("""CREATE TABLE IF NOT EXISTS cache (
    k TEXT PRIMARY KEY, body TEXT NOT NULL, created REAL NOT NULL)""")

TTL_SECONDS = 7 * 24 * 3600

def cached_chat(**payload) -> dict:
    payload.pop("stream", None)
    k = cache_key(BASE_URL, payload)
    row = db.execute("SELECT body, created FROM cache WHERE k=?", (k,)).fetchone()
    if row and time.time() - row[1] < TTL_SECONDS:
        return json.loads(row[0])
    resp = client.chat.completions.create(**payload)
    data = resp.model_dump()
    choice = data["choices"][0]
    if choice.get("finish_reason") in ("stop", "tool_calls"):
        db.execute("INSERT OR REPLACE INTO cache VALUES (?,?,?)",
                   (k, json.dumps(data), time.time()))
        db.commit()
    return data

这里有两个容易被忽略但很重要的细节。

其一，只存储 finish_reason 为 stop 或 tool_calls 的响应。如果答案是因为 length 被截断，或过程中出了问题，就不该把它冻结在缓存里长达一周。其二，错误永远不缓存。429 或 5xx 会在写入之前就抛出异常，于是下一次调用会重新尝试——这正是我们想要的行为。

流式响应也能缓存

流式输出和缓存看似矛盾，其实不然。未命中时，你照常把内容流式推给用户，同时收集所有分片；当流干净地结束后，拼接文本并存储。命中时，既可以一次性返回全文，也可以按 UI 的期望切成小片"重放"出流式效果。

def cached_stream(**payload):
    payload.pop("stream", None)
    k = cache_key(BASE_URL, payload)
    row = db.execute("SELECT body FROM cache WHERE k=?", (k,)).fetchone()
    if row:
        text = json.loads(row[0])["choices"][0]["message"]["content"]
        for i in range(0, len(text), 40):
            yield text[i:i+40]
        return
    parts, finish = [], None
    for chunk in client.chat.completions.create(stream=True, **payload):
        if not chunk.choices:
            continue
        delta = chunk.choices[0].delta.content or ""
        finish = chunk.choices[0].finish_reason or finish
        parts.append(delta)
        yield delta
    if finish == "stop":
        body = {"choices": [{"message": {"role": "assistant",
                "content": "".join(parts)}, "finish_reason": "stop"}]}
        db.execute("INSERT OR REPLACE INTO cache VALUES (?,?,?)",
                   (k, json.dumps(body), time.time()))
        db.commit()

巧妙之处在于：如果用户中途断开连接，生成器会被关闭，finish 永远不会变成 stop，于是任何不完整的内容都不会被写入缓存。

温度与"这到底该不该缓存"

缓存每次都返回同一个答案。当你本来就期望同样答案时，这完美无缺；而当"多样性"才是重点时，它就是错的。

适合缓存的场景：

temperature=0 下的分类、打标签、信息抽取和路由
不会变化的文档摘要（用内容哈希作键，而非文件名）
FAQ 式回答和帮助中心机器人
批处理管线里类似"描述这个条目"的任务
测试套件和 CI 运行，这类重复调用纯属浪费

不适合缓存的场景：

用户点击"重新生成"期待新内容的创意生成
依赖当前时间、实时数据或请求时才获取的工具结果的任务
多轮对话——完整历史几乎不可能原样重复（缓存不会造成损害，只是命不中）

作者给出的经验法则：当 temperature 为 0 且输入完全包含在请求内时，默认缓存；其余情况一律显式按需开启。

扩容时切换到 Redis

一旦有了多个 worker，按进程存放的 SQLite 文件就无法共享了。此时把存储换成 Redis 的 SET key value EX ttl 即可，键函数保持不变。注意让值保持紧凑（只存你会读回的字段），并设置内存上限与 LRU 淘汰策略，防止一波独特提示词把机器塞满。

这个阶段还值得加上请求合并（request coalescing）：如果十个 worker 在同一时刻命不中同一个键，应该只有一个去调用 API，其余等待它的结果。一个短暂的 Redis 锁（SET lock:<key> 1 NX EX 30）对大多数应用就够了。

请求合并（request coalescing） 本质上是一种"只让一个请求去做重复工作，其余请求等待并复用其结果"的并发控制模式，也常被称为 request deduplication 或 cache stampede 防护。在高并发场景下，当缓存条目恰好过期（或第一次被写入之前），可能有数十个 worker 在毫秒级内同时发现缓存未命中，随即各自向上游发出完全相同的 API 请求——这就是"缓存击穿"（cache stampede）。其代价不只是多余的费用，还会因并发请求量激增而触发上游限流（429）。

使用 Redis 的 SET lock:<key> 1 NX EX 30 可以实现一个轻量级分布式锁：NX（Not eXists）保证只有第一个执行该命令的 worker 能拿到锁，其余 worker 发现锁已存在后进入短暂轮询或等待；EX 30 则设置 30 秒的自动过期，防止持锁进程崩溃后锁永远无法释放（死锁）。对于大多数 LLM 网关场景，这个粒度已经足够——LLM 推理的耗时通常在数秒到数十秒之间，远小于 30 秒的锁超时。

如何衡量缓存是否值得

对每个功能记录三个数字：命中数、未命中数，以及各自的延迟。按作者经验，内部工具和批处理任务往往命中率很高，而开放式聊天命中极少。如果某个功能一周后命中率仍接近零，就把那里的缓存移除——它只是多余的代码。

同样要警惕过期答案。如果你改了提示词却忘了递增 CACHE_VERSION，用户在整个 TTL 周期内都会拿到旧风格的回复。因此作者的做法是：在改动提示词的同一个 commit 里顺手递增版本号，并把 TTL 控制在天级而非月级。

与按量计费、包月套餐的关系

在按 token 计费下，缓存直接省钱。而在"每日请求额度"的包月套餐下，它省的是额度——对于那些本会在一天结束前就撞上限额的 agent 或批处理任务，这一点尤其重要。无论哪种方式，你没发出的那个请求，也是最快的那个请求。

需要说明的是，原文作者构建的是一个 OpenAI 兼容网关产品，示例中的 base URL 来自该产品。但上述所有方法对任何 OpenAI 兼容端点都同样适用——只需把 BASE_URL、key 和 model ID 换成你的服务商提供的即可。