Claude キャッシュ破壊の90%がサーバー側原因—Anthropic内部分析が流出

import anthropic

client = anthropic.Anthropic()

# プロンプトキャッシュを有効にしたリクエスト例
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "あなたは優秀なコードレビュアーです。以下のコーディング規約に従って...",
        },
        {
            "type": "text",
            "text": "<長大な規約ドキュメント>...",  # 数千〜数万トークン
            "cache_control": {"type": "ephemeral"},  # ここにキャッシュ指定
        },
    ],
    messages=[
        {
            "role": "user",
            "content": "このPythonコードをレビューしてください: ..."
        }
    ],
)

# キャッシュヒット率の確認
usage = response.usage
print(f"入力トークン: {usage.input_tokens}")
print(f"キャッシュ生成トークン: {usage.cache_creation_input_tokens}")
print(f"キャッシュ読み出しトークン: {usage.cache_read_input_tokens}")

# キャッシュ効果のコスト計算スクリプト
def calculate_cache_cost(
    system_tokens: int,
    user_tokens: int,
    requests_per_day: int,
    hit_rate: float,
    model: str = "claude-opus-4-5"
) -> dict:
    """
    hit_rate: 0.0〜1.0（キャッシュヒット率）
    """
    # claude-opus-4-5の料金（$/ 1Mトークン）
    pricing = {
        "input": 15.00,
        "cache_create": 18.75,
        "cache_read": 1.50,
    }

    # 初回リクエスト（キャッシュ生成）のコスト
    first_request_cost = (
        (system_tokens * pricing["cache_create"]) +
        (user_tokens * pricing["input"])
    ) / 1_000_000

    # キャッシュヒット時のコスト
    hit_cost = (
        (system_tokens * pricing["cache_read"]) +
        (user_tokens * pricing["input"])
    ) / 1_000_000

    # キャッシュミス時のコスト（フル課金）
    miss_cost = (
        (system_tokens * pricing["input"]) +
        (user_tokens * pricing["input"])
    ) / 1_000_000

    # 1日あたりのコスト計算
    daily_cost_with_cache = (
        first_request_cost +
        (requests_per_day - 1) * (
            hit_rate * hit_cost +
            (1 - hit_rate) * miss_cost
        )
    )
    daily_cost_no_cache = requests_per_day * miss_cost

    return {
        "daily_cost_with_ideal_cache": first_request_cost + (requests_per_day - 1) * hit_cost,
        "daily_cost_with_actual_cache": daily_cost_with_cache,
        "daily_cost_no_cache": daily_cost_no_cache,
        "monthly_savings": (daily_cost_no_cache - daily_cost_with_cache) * 30,
    }

# 実行例
result = calculate_cache_cost(
    system_tokens=10_000,
    user_tokens=500,
    requests_per_day=100,
    hit_rate=0.5,  # 実測のキャッシュヒット率50%
)
print(f"キャッシュなし: ${result['daily_cost_no_cache']:.2f}/日")
print(f"キャッシュあり（50%ヒット）: ${result['daily_cost_with_actual_cache']:.2f}/日")
print(f"月間節約額: ${result['monthly_savings']:.2f}")

import anthropic
from dataclasses import dataclass, field
from typing import List

@dataclass
class CacheMetrics:
    total_requests: int = 0
    cache_hits: int = 0
    cache_misses: int = 0
    total_input_tokens: int = 0
    total_cache_read_tokens: int = 0
    total_cache_create_tokens: int = 0

    def record(self, usage: anthropic.types.Usage):
        self.total_requests += 1
        self.total_input_tokens += usage.input_tokens
        
        cache_read = getattr(usage, 'cache_read_input_tokens', 0) or 0
        cache_create = getattr(usage, 'cache_creation_input_tokens', 0) or 0
        
        self.total_cache_read_tokens += cache_read
        self.total_cache_create_tokens += cache_create
        
        if cache_read > 0:
            self.cache_hits += 1
        else:
            self.cache_misses += 1

    @property
    def hit_rate(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.cache_hits / self.total_requests

    def report(self):
        print(f"総リクエスト数: {self.total_requests}")
        print(f"キャッシュヒット: {self.cache_hits} ({self.hit_rate:.1%})")
        print(f"キャッシュミス: {self.cache_misses}")
        print(f"キャッシュ読み出しトークン計: {self.total_cache_read_tokens:,}")
        print(f"キャッシュ生成トークン計: {self.total_cache_create_tokens:,}")

# 使用例
metrics = CacheMetrics()
client = anthropic.Anthropic()

for query in user_queries:
    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system=[
            {"type": "text", "text": large_system_prompt,
             "cache_control": {"type": "ephemeral"}}
        ],
        messages=[{"role": "user", "content": query}]
    )
    metrics.record(response.usage)

metrics.report()

# 悪い例：変化する情報をキャッシュ対象にしている
bad_system = [
    {
        "type": "text",
        "text": f"現在時刻: {datetime.now()}。あなたはアシスタントです...",
        "cache_control": {"type": "ephemeral"},  # タイムスタンプが変わるためキャッシュは毎回ミス
    }
]

# 良い例：不変のコンテンツのみキャッシュ対象にする
good_system = [
    {
        "type": "text",
        "text": "あなたは優秀なアシスタントです。以下の製品マニュアルを参照して回答してください。\n\n" + large_manual_text,
        "cache_control": {"type": "ephemeral"},  # マニュアルは変化しないのでキャッシュ効果が高い
    }
]

# 長い会話履歴のキャッシュ例
messages = [
    {"role": "user", "content": "最初の質問"},
    {"role": "assistant", "content": "最初の回答"},
    # ...多数のターン...
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "ここまでの会話を踏まえて...",
                "cache_control": {"type": "ephemeral"},  # 会話履歴全体をキャッシュ
            }
        ]
    }
]