你的 LLM 网关在缓存上撒谎了吗？5 分钟审计

网关位于你的代码和模型供应商之间。你从响应里读回 cached_tokens，看到一个更小的数字，于是相信省下的钱是真实的。但你永远看不到上游那次调用。网关可以报告一次缓存命中，却仍按全额输入费率计费。它可能在一个看起来完美干净的响应背后根本没有缓存。它可能在流式传输（也就是你大部分生产流量所走的路径）上剥离 usage 元数据，让你两头都无从判断。

这不是假设。一篇 Hacker News PSA 报告称，把 DeepSeek V4 通过一个流行的网关路由，返回的缓存 token 比直接调用 DeepSeek 少 2–3 倍；一位评论者贴出账单，显示缓存统计根本没有通过网关上报。该网关团队回复说他们无法复现，正在调查。这种分歧正是关键所在。当两方对你的缓存是否生效都无法达成一致时，唯一的仲裁就是你自己跑过的一次测量。

通常这不是恶意，而是一处翻译落差或一条未完成的代码路径。但无论哪种情况，对你账单的影响都是一样的。本文给出一个可直接运行的脚本，用它来对任意网关（包括本网关）审计两种风格的提示缓存：自动缓存（DeepSeek）和基于标记的缓存（Claude）。它会在五分钟内打印一张并排的评分卡。

网关在缓存上撒谎的四种方式

最危险的是前两种：响应看起来像缓存在正常工作。你直到月底才会发现真相。

两种缓存机制，一次审计

供应商以两种形态暴露缓存，而一个真正的网关必须忠实地把两者都透传：

• 自动（DeepSeek、GPT、Gemini、Qwen）：供应商会自行缓存任何足够长的前缀。没有标记。命中出现在 usage.prompt_tokens_details.cached_tokens 中。
• 基于标记（Anthropic Claude）：你用 cache_control 标记可缓存的片段。命中出现为 cache_read_input_tokens。

脚本用一个轻量的 Lane 适配器把这种差异藏起来，然后对两者跑全部五项检查。下面就是全部内容：两条 lane 和一个执行每项检查的 audit()。

import os, time, uuid
from openai import OpenAI
from anthropic import Anthropic

KEY  = os.environ["GATEWAY_KEY"]
oai  = OpenAI(api_key=KEY,    base_url="https://synthorai.io/v1")   # auto lane
anth = Anthropic(api_key=KEY, base_url="https://synthorai.io/")     # marker lane

class AutoLane:      # DeepSeek / GPT / Gemini / Qwen: provider caches automatically
    mode = "auto"
    def __init__(self, model): self.model = model
    def call(self, sys, q, stream=False):
        if stream:
            cached = cost = None
            s = oai.chat.completions.create(model=self.model, max_tokens=48, stream=True,
                stream_options={"include_usage": True},
                messages=[{"role":"system","content":sys},{"role":"user","content":q}])
            for ev in s:
                if ev.usage:
                    d = ev.usage.prompt_tokens_details
                    cached, cost = (d.cached_tokens if d else None), getattr(ev.usage,"cost",None)
            return {"cached": cached or 0, "cost": cost, "prompt_total": None}
        u = oai.chat.completions.create(model=self.model, max_tokens=48,
            messages=[{"role":"system","content":sys},{"role":"user","content":q}]).usage
        cached = u.prompt_tokens_details.cached_tokens if u.prompt_tokens_details else 0
        return {"cached": cached or 0, "cost": u.cost, "prompt_total": u.prompt_tokens}

class MarkerLane:    # Anthropic Claude: explicit cache_control markers
    mode = "marker"
    def __init__(self, model): self.model = model
    def call(self, sys, q, stream=False):
        block = {"type":"text","text":sys,"cache_control":{"type":"ephemeral"}}
        if stream:
            with anth.messages.stream(model=self.model, max_tokens=48, system=[block],
                    messages=[{"role":"user","content":q}]) as s:
                for _ in s.text_stream: pass
                u = s.get_final_message().usage.model_dump()
            return {"cached": u.get("cache_read_input_tokens") or 0,
                    "cost": u.get("cost"), "prompt_total": None}
        u = anth.messages.create(model=self.model, max_tokens=48, system=[block],
            messages=[{"role":"user","content":q}]).usage.model_dump()
        read, created = u.get("cache_read_input_tokens",0), u.get("cache_creation_input_tokens",0)
        return {"cached": read, "cost": u.get("cost"),
                "prompt_total": u.get("input_tokens",0) + read + created}

def audit(lane, long_prompt):
    SYS = f"[audit {uuid.uuid4().hex}]\n\n" + long_prompt    # unique => guaranteed cold start
    r = {"lane": lane.model, "mode": lane.mode}

    # CHECK 1: cache engages. Cold misses; a repeat should hit. A cache can
    # take a moment to become readable, so poll the warm read (sleep 1s between
    # attempts) before concluding "no cache".
    cold = lane.call(SYS, "Q1")
    warm = cold
    for i in range(4):
        warm = lane.call(SYS, f"warm {i}")
        if warm["cached"] > 0: break
        time.sleep(1.0)
    r["cold"], r["warm"] = cold, warm
    r["check1"] = cold["cached"] == 0 and warm["cached"] > 0

    # CHECK 2: cost reflects the discount (catches "cache theater").
    disc = (1 - warm["cost"]/cold["cost"])*100 if cold["cost"] and warm["cost"] else None
    r["discount"], r["check2"] = disc, (disc is not None and disc > 30)

    # CHECK 3: token accounting. cached fits inside the prompt total.
    r["check3"] = warm["prompt_total"] is None or warm["cached"] <= warm["prompt_total"]

    # CHECK 4: streaming preserves usage metadata (cache count AND cost).
    st = lane.call(SYS, "stream", stream=True)
    r["stream_cached"], r["stream_cost"] = st["cached"] > 0, st["cost"] is not None
    r["check4"] = r["stream_cached"] and r["stream_cost"]

    # CHECK 5: negative control. a unique prefix must always miss.
    n1 = lane.call(f"[uniq {uuid.uuid4().hex}]\n\n"+long_prompt, "x")
    n2 = lane.call(f"[uniq {uuid.uuid4().hex}]\n\n"+long_prompt, "y")
    r["check5"] = n1["cached"] == 0 and n2["cached"] == 0
    return r

# Any long, STABLE text works as the cacheable prefix: a system prompt, tool
# schemas, or a retrieved document. It only needs to clear the provider's
# minimum cacheable size (see Check 1). Load yours however you like.
LONG_SYSTEM_PROMPT = open("system_prompt.txt").read()   # ~8K+ tokens

for lane in [AutoLane("deepseek-v4-flash"), MarkerLane("claude-opus-4-8")]:
    print(audit(lane, LONG_SYSTEM_PROMPT))

本文余下部分逐一讲解每项检查：实现它的代码行、两条 lane 各自返回了什么，以及如何解读结果。

检查 1：缓存是否生效？

cold = lane.call(SYS, "Q1")
warm = cold
for i in range(4):                       # poll: a cache may take a beat to be readable
    warm = lane.call(SYS, f"warm {i}")
    if warm["cached"] > 0: break
    time.sleep(1.0)
r["check1"] = cold["cached"] == 0 and warm["cached"] > 0

对一个唯一前缀的冷调用必须什么都不缓存；重复调用则必须命中。最常见的单一误报是在一次温调用之后就宣称「无缓存」，因为缓存并不总是立即可读。这个循环带 1 秒间隔轮询几次，消除了这种偶发性。如果你的提示已经超过尺寸下限（大多数供应商约 1,024 token；DeepSeek 以更细的 64 token 粒度对齐），却在几次温调用后仍然得到 0，那么缓存确实没有生效。

检查 2：成本是否反映了折扣？

disc = (1 - warm["cost"]/cold["cost"])*100 if cold["cost"] and warm["cost"] else None
r["check2"] = disc is not None and disc > 30

这是抓出缓存表演的那项检查。温调用的成本必须真的下降。DeepSeek 的每次调用总额下降了约 72%（缓存输入的折扣更陡峭；输出和未缓存的剩余部分稀释了这个表面数字）。Claude 的缓存读取打了约 1 折。失败信号一目了然：cached_tokens > 0 却伴随完全相同的冷、温成本，意味着网关报告了一次它并未据此定价的命中。你正在为一个纸面上「能用」的缓存付全价。

检查 3：token 计数对得上吗？

r["check3"] = warm["prompt_total"] is None or warm["cached"] <= warm["prompt_total"]

cached 必须落在 prompt total 之内，剩余部分作为未缓存输入计费。两者都对得上。如果 cached_tokens 超过了 prompt_tokens，或者对一个稳定前缀来说未缓存的剩余部分大得不合常理，那么网关在记账上出了错：在翻译过程的某处重新分词或重复计数了。

检查 4：流式是否保留了元数据？

st = lane.call(SYS, "stream", stream=True)
r["stream_cached"], r["stream_cost"] = st["cached"] > 0, st["cost"] is not None
r["check4"] = r["stream_cached"] and r["stream_cost"]

大多数生产环境的聊天都是流式的，所以这是最要紧的路径。在两条 lane 上，缓存命中信号和成本都挺过了流式传输。cached_tokens 和 cost 都在最后一个 usage 块里返回，所以你流量最大的路径仍然可审计。要警惕的失败模式是网关在流式上丢弃 usage：干净的 token 输出却没有 cached_tokens 或 cost，意味着你在跑得最多的路径上两眼一抹黑。（要传 stream_options={"include_usage": True}，usage 块才会被发出。）

检查 5：阴性对照

n1 = lane.call(f"[uniq {uuid.uuid4().hex}]\n\n"+long_prompt, "x")
n2 = lane.call(f"[uniq {uuid.uuid4().hex}]\n\n"+long_prompt, "y")
r["check5"] = n1["cached"] == 0 and n2["cached"] == 0

每次调用都发送一个唯一前缀；它绝不能命中。两条 lane 都正确地对不同前缀报告了 cached=0 并按全价计费。这里若出现「命中」，就会让缓存报告成为一个你永远无法信任的假阳性。正是这个干净的阴性对照，才让检查 1–2 中的阳性结果在一开始就具有意义。

解读你的评分卡

会悄悄花钱的是 2（为一次被报告的命中付全价）和 1（干净响应背后毫无缓存）。对你每个要计费的模型都把这两项跑一遍。

结语

缓存是 LLM 应用里杠杆率最高的成本调节项，这恰恰是「缓存在工作」值得测试、而不该想当然的原因。把检查 1 + 检查 2 接入 CI，对你每个要计费的模型都跑，并在折扣偏离你预期区间时告警，那么当网关或上游供应商改变行为的当天，你就能抓到一次静默回归，而不是等到账单周期结束。而无论你的审计怎么做，在你判定缓存坏掉之前，先轮询温读取。

想了解这些数字背后的机制（预填充、KV 缓存、TTL），从 KV 缓存与 TTL 如何工作 开始。想看各供应商可用的缓存模式，请参阅教程。

常见问题

我的检查 1 在温调用上显示 0。是我的网关在撒谎吗？ 先检查三件事。(1) 你的提示是否超过供应商的最小可缓存尺寸（大多数约 1,024 token；DeepSeek 以更细的 64 token 粒度对齐）？(2) 你轮询了温读取几次吗？缓存并不总是在紧接着的下一次调用就变得可读。(3) 各次调用之间前缀是否逐字节相同，前面没有时间戳或每请求 ID？只有在这三点都满足之后，你才应该怀疑网关。

「缓存表演」在实际中会让我付出什么代价？ 你在每次调用上都付全额输入费率，却以为只付了一个零头。在一个带大型稳定前缀的高流量端点上，那就意味着你的账单是你模型估算的好几倍。检查 2 是值得告警的那一项。

为什么这里 DeepSeek 的折扣比 Claude 低？ 测量的东西不一样。Claude 的约 90% 是缓存输入上的读取折扣。DeepSeek 的约 72% 是每次调用总额的下降，其中输出和未缓存的剩余部分按全价计费，稀释了表面数字。请按你自己的提示形态拿同类对比同类。

这对 GPT、Gemini、Qwen 也适用吗？ 适用。它们全都是自动缓存，所以原封不动地使用 AutoLane，只换一个不同的 model。只有 Claude 需要 MarkerLane。两种方式都跑同样的五项检查。

这该放进 CI 吗？ 该。对你每个要计费的模型，定期跑检查 1 + 检查 2，并在观测到的折扣偏出你的预期区间时告警。一个常驻的审计能把一次静默回归变成一条通知。

← 返回博客