その LLM Gateway はキャッシュについて嘘をついていないか？ 5 分で監査する

Gateway は、コードとモデルプロバイダーの間に入る。レスポンスの cached_tokens が小さくなっていれば、その分だけ料金も下がったと考えがちだ。しかし、upstream への実際のリクエストは見えない。Gateway がキャッシュヒットを返しながら、入力料金を満額請求している可能性もある。レスポンスには何の問題もないのに、裏ではまったくキャッシュされていないこともある。さらに、本番トラフィックの大半が通る streaming で usage メタデータが削られていれば、確認する手段すらない。

TL;DR

• Hacker News の注意喚起では、ある有名 Gateway 経由で DeepSeek V4 を呼ぶと、DeepSeek を直接呼んだ場合より cached token が 2-3 倍少なく返されたと報告されている。
• 実行可能な 1 本のスクリプトで、キャッシュが有効になるか、実際に料金が下がるか、token 数の整合性が取れているか、streaming でも usage メタデータが残るか、ネガティブコントロールが未キャッシュのままか、という 5 項目を確認できる。
• Synthorai Gateway 経由で監査したところ、deepseek-v4-flash は warm 時のキャッシュヒット率が 96%、1 リクエストあたりの料金が 72.3% 下がった。claude-opus-4-8 はそれぞれ 99.9% と 90.6% だった。
• cached_tokens > 0 なのに cold と warm の料金が同じなら、Gateway はヒットを報告しているだけで、料金には反映していない。

これは仮定の話ではない。Hacker News の注意喚起では、ある有名 Gateway 経由で DeepSeek V4 を呼ぶと、DeepSeek を直接呼んだ場合より cached token が 2–3 倍少なく返されたと報告されている。コメントには、Gateway 経由ではキャッシュ統計そのものが報告されていなかったことを示す請求明細も投稿された。Gateway 側は再現できず、調査中だと回答している。両者の認識が食い違うなら、自分で実行した計測結果を基準にするしかない。

多くの場合、悪意があるわけではない。API 変換時の欠落や、実装が終わっていないコードパスが原因だ。ただし、請求額への影響は変わらない。この記事では、実行可能な 1 本のスクリプトを使い、自動方式（DeepSeek）とマーカー方式（Claude）の両方の prompt cache を、任意の Gateway に対して監査する。この Gateway も対象にできる。5 分以内に比較用のスコアカードが出力される。

Gateway がキャッシュについて嘘をつく 4 つのパターン

特に危険なのは最初の 2 つだ。レスポンスだけを見ると、キャッシュは正常に動いているように見える。問題に気づくのは月末になる。

2 種類のキャッシュ方式を 1 つの監査で確認する

プロバイダーが提供するキャッシュ方式は 2 種類ある。Gateway はどちらも情報を欠落させずに通す必要がある。

• 自動方式（DeepSeek、GPT、Gemini、Qwen）：十分に長い prefix をプロバイダーが自動でキャッシュする。マーカーは不要。ヒット数は usage.prompt_tokens_details.cached_tokens に入る。
• マーカー方式（Anthropic Claude）：キャッシュ対象の範囲に cache_control を付ける。ヒット数は cache_read_input_tokens に入る。

スクリプトでは、この違いを薄い Lane adapter で吸収し、両方式に同じ 5 項目のチェックを実行する。全体は以下のとおりだ。2 つの lane と、すべてのチェックを実行する 1 つの 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

一意の prefix を使った cold リクエストでは、キャッシュヒットは必ず 0 になる。次に同じ prefix で呼ぶとヒットしなければならない。最も多い誤検知は、warm リクエストを 1 回実行しただけで「キャッシュされていない」と判断することだ。キャッシュは書き込み直後に参照できるとは限らない。このループでは 1 秒間隔で数回 polling するため、その不安定さを排除できる。キャッシュ対象となる最小サイズを超えた prompt で数回 warm リクエストを実行しても 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

このチェックで、見せかけのキャッシュを検出する。warm リクエストでは、実際の料金も下がらなければならない。DeepSeek では 1 リクエストあたりの合計料金が約 72% 下がった。cached input 自体の割引率はさらに大きいが、出力と未キャッシュ部分が含まれるため、合計で見ると割引率は下がる。Claude の cached input の read 料金は約 90% 引きになる。異常は明確だ。cached_tokens > 0 なのに cold と warm の料金が同じなら、Gateway はヒットを報告しているだけで、料金には反映していない。帳簿上は「動いている」キャッシュに対して、満額を支払っていることになる。

チェック 3：token 数の整合性は取れているか？

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

cached は prompt の合計に含まれ、残りが未キャッシュの入力として請求される。どちらも整合している。cached_tokens が prompt_tokens を超えている場合や、安定した prefix に対して未キャッシュ部分が不自然に大きい場合は、Gateway の集計に問題がある。API 変換のどこかで tokenization をやり直しているか、二重計上している可能性がある。

チェック 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"]

本番の chat は大半が streaming なので、ここが最も重要な経路になる。どちらの lane でも、キャッシュヒット情報と料金は streaming 後も保持された。cached_tokens と cost は最後の usage chunk に含まれるため、トラフィック量が最も多い経路も監査できる。注意すべきなのは、streaming で usage を落とす Gateway だ。token 出力が正常でも cached_tokens や cost がなければ、最も多く使う経路の実態を確認できない。（usage chunk を出力させるには stream_options={"include_usage": True} を指定する。）

チェック 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

リクエストごとに一意の prefix を送れば、ヒットは絶対に発生しない。どちらの lane でも、異なる prefix に対して cached=0 が返り、料金は満額だった。ここで「ヒット」が返るなら、キャッシュの報告は偽陽性であり、信用できない。ネガティブコントロールが正常だからこそ、チェック 1 と 2 の陽性結果に意味がある。

スコアカードの読み方

気づかないまま料金が増えるのは、2（ヒットと報告されているのに満額）と 1（正常なレスポンスの裏でキャッシュされていない）だ。請求対象のすべてのモデルで両方を実行する。

まとめ

LLM アプリのコスト削減では、キャッシュの効果が最も大きい。だからこそ、「キャッシュは動いている」と決めつけず、テストで確認する必要がある。請求対象の各モデルについて、チェック 1 とチェック 2 を CI に組み込む。割引率が想定範囲を下回ったら alert を出す。そうすれば、Gateway や upstream プロバイダーの挙動が変わっても、請求期間の終了を待たず、その日のうちにサイレントリグレッションを検出できる。また、どのような監査でも、キャッシュの故障を判断する前に warm read を polling すること。

数値の背景にある仕組み（prefill、KV cache、TTL）については、KV cache と TTL の仕組みを参照してほしい。プロバイダー別に実際に使えるキャッシュ実装パターンは、チュートリアルにまとめている。

FAQ

チェック 1 の warm リクエストが 0 でした。Gateway が嘘をついているのでしょうか？ まず 3 点を確認する。（1）prompt はプロバイダーのキャッシュ対象となる最小サイズを超えているか。多くは約 1,024 token で、DeepSeek はさらに細かい 64-token 単位で一致判定する。（2）warm read を数回 polling したか。次のリクエストですぐにキャッシュを参照できるとは限らない。（3）リクエスト間で prefix が byte 単位で同一か。先頭に timestamp やリクエスト単位の ID が入っていないか。この 3 点を確認した後で、Gateway を疑うべきだ。

「見せかけのキャッシュ」では、実際にどの程度の損失が出ますか？ 一部の料金で済んでいると思いながら、毎回通常の入力料金を満額支払うことになる。大きく安定した prefix を持つ高トラフィックの endpoint では、請求額が見積もりの数倍になる。alert の対象にすべきなのはチェック 2 だ。

ここでは、DeepSeek の割引率が Claude より低いのはなぜですか？ 計測対象が異なる。Claude の約 90% は、cached input の read に対する割引率だ。DeepSeek の約 72% は、1 リクエストあたりの合計料金の削減率になる。出力と未キャッシュ部分には通常料金がかかるため、全体の割引率は下がる。自分の prompt 構成で比較する際は、同じ指標同士を比べる必要がある。

GPT、Gemini、Qwen でも動きますか？ 動く。いずれも自動方式なので、model だけ変更して AutoLane をそのまま使える。MarkerLane が必要なのは Claude だけだ。どちらの方式でも、実行する 5 項目は同じになる。

CI に組み込むべきですか？ 組み込むべきだ。請求対象のすべてのモデルに対して、チェック 1 とチェック 2 を定期実行する。観測された割引率が想定範囲から外れたら alert を出す。常時監査を実行すれば、サイレントリグレッションを通知として受け取れる。

← ブログに戻る