あなたの LLM ゲートウェイはキャッシュについて嘘をついていないか？5分でできる監査

ゲートウェイは、あなたのコードとモデルプロバイダーの間に位置する。レスポンスから cached_tokens を読み取り、より小さな数字を目にして、節約された金額が本物だと信じる。だが、あなたは上流への呼び出しを決して見ることはない。ゲートウェイはキャッシュヒットを報告しながら、なお満額の入力レートで課金できてしまう。完璧にきれいなレスポンスの裏で、まったくキャッシュしていないこともある。本番トラフィックの大半が通る経路であるストリーミングで usage メタデータを削ぎ落とし、どちらとも判断できなくしてしまうこともある。

これは仮定の話ではない。ある Hacker News の PSA は、DeepSeek V4 を人気のゲートウェイ経由でルーティングすると、DeepSeek を直接呼び出した場合より キャッシュトークンが 2〜3 倍少なく 返ってきたと報告した。あるコメント投稿者は、キャッシュ統計がゲートウェイを通じてまったく報告されていないことを示す請求書を公開した。ゲートウェイのチームは再現できず、調査中だと返答した。この食い違いこそが核心だ。キャッシュが機能しているかどうかについて二者が合意できないとき、唯一の決め手は、あなた自身が実行した測定だけである。

たいていこれは悪意ではない。翻訳のギャップか、未完成のコードパスだ。だが、どちらにせよ請求書への影響は同じである。本記事は、実行可能な1本のスクリプトであり、自動（DeepSeek）とマーカー方式（Claude）という2つのスタイルのプロンプトキャッシュを、本ゲートウェイを含む任意のゲートウェイに対して監査する。5分以内に、横並びのスコアカードを出力する。

ゲートウェイがキャッシュについて嘘をつく4つの方法

危険なのは最初の2つだ。レスポンスがキャッシュ機能しているように見える。あなたが気づくのは月末である。

2つのキャッシュ機構、1回の監査

プロバイダーはキャッシュを2つの形で提供し、本物のゲートウェイはその両方を忠実に透過させなければならない。

• 自動（DeepSeek、GPT、Gemini、Qwen）：プロバイダーが十分に長いプレフィックスを自前でキャッシュする。マーカーはない。ヒットは usage.prompt_tokens_details.cached_tokens に現れる。
• マーカー方式（Anthropic Claude）：キャッシュ可能な範囲を cache_control でタグ付けする。ヒットは cache_read_input_tokens として現れる。

スクリプトは、その差を薄い Lane アダプターの裏に隠し、両方に対して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

ユニークなプレフィックスへのコールドな呼び出しは何もキャッシュしてはならない。繰り返しはヒットしなければならない。最もよくある単一の誤報は、1回のウォーム呼び出しの後に「キャッシュなし」と断じてしまうことだ。キャッシュは必ずしも即座に読めるようになるわけではないからである。このループは1秒の間隔を挟んで数回ポーリングし、その不安定さを取り除く。サイズの下限（多くのプロバイダーで約 1,024 トークン。DeepSeek はより細かい 64 でそろえる）を超えたプロンプトで、数回のウォーム呼び出しの後もなお 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 の1回あたりの合計は約 72% 下がった（キャッシュされた入力はより急峻に割引され、出力とキャッシュされていない残りが見出しの数字を薄める）。Claude のキャッシュ読み取りは約 90% オフだ。失敗のシグナルは明白である。cached_tokens > 0 でありながら、コールドとウォームのコストがまったく同じなら、それはゲートウェイが価格に反映していないヒットを報告しているということだ。あなたは紙の上で「機能する」キャッシュに満額を支払っている。

チェック 3：トークン数は辻褄が合うか？

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 を落とすゲートウェイだ。cached_tokens も cost もないきれいなトークン出力は、あなたが最も多く走らせる経路で目隠し状態であることを意味する。（usage チャンクがそもそも発行されるよう、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

毎回ユニークなプレフィックスを送る。それは決してヒットしてはならない。両 lane とも、異なるプレフィックスに対して cached=0 を正しく報告し、満額で課金した。ここで「ヒット」が出れば、キャッシュ報告は二度と信頼できない偽陽性になってしまう。このきれいな陰性対照こそが、そもそもチェック 1〜2 における陽性の結果に意味を与えるものなのだ。

あなたのスコアカードの読み方

静かにお金を奪うのは 2（報告されたヒットに対して満額を支払う）と 1（きれいなレスポンスの裏でキャッシュなし）だ。課金対象のすべてのモデルで、この両方を実行すること。

結び

キャッシュは LLM アプリにおいて最もレバレッジの高いコストレバーであり、まさにそれゆえ「キャッシュは機能している」は思い込みではなくテストに値する。チェック 1 + チェック 2 を、課金対象の各モデルに対して CI に組み込み、割引が想定の帯域を下回ったらアラートを出せば、ゲートウェイや上流プロバイダーが挙動を変えたその日にサイレントなリグレッションを捕まえられる。請求サイクルの終わりにではなく。そして監査が何をするにせよ、キャッシュが壊れていると断ずる前に、ウォーム読み取りをポーリングせよ。

これらの数字の背後にある仕組み（プリフィル、KV キャッシュ、TTL）については、KV キャッシュと TTL の仕組み から始めるとよい。プロバイダーごとの実用的なキャッシュパターンは、チュートリアルを参照。

FAQ

チェック 1 がウォーム呼び出しで 0 を示す。私のゲートウェイは嘘をついている？ まず3つを確認しよう。(1) あなたのプロンプトはプロバイダーの最小キャッシュ可能サイズ（多くは約 1,024 トークン。DeepSeek はより細かい 64 トークン粒度でそろえる）を超えているか？(2) ウォーム読み取りを数回ポーリングしたか？キャッシュは必ずしも次の呼び出しで読めるようになるとは限らない。(3) プレフィックスは呼び出し間でバイト単位まで同一で、先頭にタイムスタンプやリクエストごとの ID がないか？この3つがすべて満たされて初めて、ゲートウェイを疑うべきだ。

「キャッシュ・シアター」は実際にいくらの損になる？ ごく一部しか払っていないと信じながら、毎回の呼び出しで満額の入力レートを支払うことになる。大きく安定したプレフィックスを持つ高ボリュームのエンドポイントでは、それはあなたの見積もりの数倍の請求額を意味する。アラートを出すべきはチェック 2 だ。

なぜここでは DeepSeek の割引が Claude より低いのか？ 測っているものが違う。Claude の約 90% は、キャッシュされた入力に対する読み取り割引だ。DeepSeek の約 72% は1回あたりの合計の削減であり、そこでは出力とキャッシュされていない残りが満額で課金され、見出しの数字を薄める。あなた自身のプロンプト形状について、同種のものを同種と比較すること。

これは GPT、Gemini、Qwen でも動くか？ 動く。それらはすべて自動なので、model を変えるだけで AutoLane をそのまま使う。MarkerLane が必要なのは Claude だけだ。どちらの場合も同じ5つのチェックである。

これは CI に置くべきか？ 置くべきだ。課金対象のすべてのモデルに対して、チェック 1 + チェック 2 をスケジュール実行し、観測された割引が想定の帯域から外れたらアラートを出す。常設の監査は、サイレントなリグレッションを通知へと変える。

← ブログに戻る