你的 LLM Gateway 会谎报缓存吗？5 分钟审计

Gateway 位于你的代码和模型提供商之间。响应里返回了 cached_tokens，数值也变小了，于是你自然会认为省下的钱真实有效。但你看不到上游调用。Gateway 完全可能报告缓存命中，却仍按完整输入费率计费；也可能根本没有缓存，只是返回了看起来完全正常的响应；还可能在 streaming 路径上丢掉 usage 元数据，让你无从判断，而大部分生产流量恰恰走的就是这条路径。

TL;DR

• Hacker News 上的一篇 PSA 称，通过某个热门 Gateway 调用 DeepSeek V4 时，返回的缓存 token 比直接调用 DeepSeek 少 2-3 倍。
• 一个可直接运行的脚本完成五项检查：缓存是否生效、成本是否实际下降、token 数量能否对上、streaming 是否保留 usage 元数据，以及负对照是否始终未命中。
• 通过 Synthorai Gateway 审计时，deepseek-v4-flash 的热缓存命中率为 96%，单次调用成本下降 72.3%；claude-opus-4-8 分别为 99.9% 和 90.6%。
• 如果 cached_tokens > 0，但冷调用和热调用成本完全相同，说明 Gateway 虽然报告了命中，却没有按缓存价格计费。

这不是假设。Hacker News PSA 称，通过某个热门 Gateway 调用 DeepSeek V4 时，返回的缓存 token 比直接调用 DeepSeek 少 2–3 倍；还有一位评论者贴出了账单，显示 Gateway 完全没有上报缓存统计。Gateway 团队回复称无法复现，正在调查。这个分歧本身就说明了问题：如果双方连缓存是否生效都无法达成一致，唯一可靠的判断依据就是你自己跑出来的数据。

通常这不是恶意行为，更可能是协议转换不完整，或者某条代码路径还没实现完。但无论原因是什么，对账单的影响都一样。本文提供一个可直接运行的脚本，用来审计任意 Gateway 的两种 prompt 缓存机制：自动缓存（DeepSeek）和标记式缓存（Claude），也包括当前这个 Gateway。不到五分钟，就能打印出并排对比的检查结果。

Gateway 谎报缓存的四种方式

前两种最危险：响应看起来像是缓存正常工作，直到月底你才会从账单里发现问题。

两种缓存机制，一套审计方法

提供商对外提供的缓存机制主要有两种，真正可靠的 Gateway 必须原样支持这两种方式：

• 自动缓存（DeepSeek、GPT、Gemini、Qwen）：只要前缀足够长，提供商就会自动缓存，不需要添加标记。命中数出现在 usage.prompt_tokens_details.cached_tokens。
• 标记式缓存（Anthropic Claude）：使用 cache_control 标记可缓存的内容区间。命中数以 cache_read_input_tokens 返回。

脚本通过一个很薄的 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))

下文会逐项说明每个检查：对应的实现代码、两个通道返回的结果，以及如何解读。

检查 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 秒轮询一次，最多尝试几次，可以避免这种不稳定因素。如果 prompt 已经超过最低缓存长度，多次热调用后仍为 0，缓存才是真的没有生效。大多数提供商的最低长度约为 1,024 个 token；DeepSeek 会以更细的 64 个 token 为匹配粒度。

检查 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 的缓存读取价格约低 90%。失败信号非常明确：如果 cached_tokens > 0，但冷调用和热调用的成本完全相同，说明 Gateway 报告了命中，却没有按缓存价格计费。缓存只是在数据上“生效”，你付的仍然是全价。

检查 3：token 数量能否对上？

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

cached 必须包含在 prompt token 总数内，剩余部分则按未缓存输入计费。两个通道的数据都能对上。如果 cached_tokens 超过 prompt_tokens，或者对于稳定前缀而言，未缓存部分大得不合理，就说明 Gateway 的计数有问题：可能在协议转换过程中重新进行了 token 化，也可能重复计数。

检查 4：streaming 是否保留元数据？

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"]

大多数生产环境中的聊天请求都使用 streaming，因此这条路径最关键。两个通道在 streaming 过程中都保留了缓存命中信息和成本数据。cached_tokens 与 cost 会在最后一个 usage 数据块中返回，因此流量最高的路径仍然可以审计。需要警惕的是 Gateway 在 streaming 时丢弃 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

每次调用都发送唯一前缀，必须始终无法命中。对于不同前缀，两个通道都正确返回了 cached=0，并按全价计费。如果这里出现“命中”，缓存报告就存在假阳性，完全不值得信任。负对照结果正常，检查 1–2 的正向结果才有意义。

如何解读检查结果

会在不知不觉中增加成本的是 2（报告命中却按全价计费）和 1（响应正常但完全没有缓存）。每个实际计费的模型都应该执行这两项检查。

结语

在 LLM 应用中，缓存对成本的影响最大。正因如此，“缓存正常工作”不能靠假设，必须通过测试验证。把检查 1 和检查 2 接入 CI，覆盖每个实际计费的模型；当折扣偏离预期区间时立即告警。这样，Gateway 或上游提供商一旦改变行为，你当天就能发现静默回归，而不是等到账单周期结束。无论审计怎么实现，在判定缓存失效前，都要先轮询热缓存读取。

如果想了解这些数字背后的机制，包括 prefill、KV 缓存和 TTL，可以先读 KV 缓存和 TTL 的工作原理。各提供商的缓存实现示例见这篇教程。

常见问题

检查 1 的热调用结果为 0。我的 Gateway 在谎报吗？ 先确认三点。（1）prompt 是否超过提供商要求的最低缓存长度？大多数约为 1,024 个 token；DeepSeek 采用更细的 64-token 匹配粒度。（2）是否对热缓存读取进行了多次轮询？缓存不一定会在下一次调用时立即可读。（3）两次调用的前缀是否逐字节完全一致，开头有没有时间戳或单次请求 ID？只有这三点都确认无误后，才应该怀疑 Gateway。

“缓存假象”实际会让我多花多少钱？ 你以为支付的是折扣价，实际上每次调用都按完整输入费率计费。对于请求量大、稳定前缀又很长的 endpoint，实际账单可能是预估的数倍。应该重点为检查 2 配置告警。

为什么这里 DeepSeek 的折扣低于 Claude？ 两者衡量的指标不同。Claude 的约 90% 指缓存输入的读取折扣。DeepSeek 的约 72% 指单次调用总成本的降幅，其中输出和未缓存部分仍按全价计费，因此整体降幅会被拉低。应根据自己的 prompt 结构，用相同口径进行比较。

GPT、Gemini、Qwen 也适用吗？ 适用。它们都采用自动缓存，只需更换 model，无需修改 AutoLane。只有 Claude 需要使用 MarkerLane。无论采用哪种机制，五项检查都相同。

应该把它放进 CI 吗？ 应该。定期对每个实际计费的模型运行检查 1 和检查 2；当实际折扣偏离预期区间时告警。持续审计可以把静默回归变成即时通知。

← 返回博客