Gemini API のキャッシュ戦略で月3万円を6,000円にした：Context Caching・Implicit Caching の本番設計

# ❌ 良くない順序：変化する部分が先頭にある
BAD_PROMPT_ORDER = """
ユーザーID: {user_id}
タイムスタンプ: {timestamp}
 
[ここに長大なシステムコンテキスト]
"""
 
# ✅ 良い順序：固定部分を先頭にまとめる
GOOD_PROMPT_ORDER = """
[ここに長大なシステムコンテキスト]
 
ユーザーID: {user_id}
タイムスタンプ: {timestamp}
"""

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro")
 
# 固定コンテキストを先頭に、変化する部分を末尾に
FIXED_CONTEXT = """
あなたはPythonの専門家です。以下のコーディング規約に従ってコードをレビューしてください。
 
[コーディング規約 - 30,000トークン相当]
1. 変数名はsnake_caseを使用する
2. 関数は単一責任の原則に従う
3. 型ヒントを必ず記述する
... (大量の規約が続く)
"""
 
def review_code_with_implicit_cache(user_code: str) -> dict:
    """コードレビュー - Implicit Cachingが効くよう設計"""
    # 固定部分を先頭に、変化するコードを末尾に配置
    full_prompt = f"{FIXED_CONTEXT}\n\n以下のコードをレビューしてください：\n\n```python\n{user_code}\n```"
    
    response = model.generate_content(
        full_prompt,
        generation_config={"temperature": 0}
    )
    
    usage = response.usage_metadata
    hit_rate = 0
    if usage.prompt_token_count > 0:
        hit_rate = usage.cached_content_token_count / usage.prompt_token_count * 100
    
    return {
        "review": response.text,
        "total_input_tokens": usage.prompt_token_count,
        "cached_tokens": usage.cached_content_token_count,
        "cache_hit_rate": f"{hit_rate:.1f}%",
        "billing_tokens": usage.prompt_token_count - usage.cached_content_token_count,
    }
 
# 連続したリクエストでImplicit Cachingの効果を確認
code_samples = [
    "def get_user(id):\n    return db.query(f'SELECT * FROM users WHERE id={id}')",
    "class UserManager:\n    def __init__(self):\n        self.users = []",
    "import os\nAPI_KEY = os.getenv('API_KEY')",
]
 
for i, code in enumerate(code_samples):
    result = review_code_with_implicit_cache(code)
    print(f"\n[リクエスト {i+1}]")
    print(f"  総入力トークン: {result['total_input_tokens']}")
    print(f"  キャッシュヒット: {result['cached_tokens']} トークン（{result['cache_hit_rate']}）")
    print(f"  課金対象トークン: {result['billing_tokens']}")

# ❌ キャッシュが作られたモデルと違うバリアントを使っている
cache = caching.CachedContent.create(
    model="models/gemini-2.5-pro",  # キャッシュ作成時
    ...
)
# 別のコードで
model = genai.GenerativeModel("gemini-2.5-pro-latest")  # ← バリアントが違う！

# ❌ タイムスタンプや動的な値をシステムプロンプトに含めている
system_prompt = f"""
現在時刻: {datetime.now().isoformat()}  # ← 毎回変わる！
あなたはサポートAIです...
"""

def safe_cache_request(cache_name: str, user_message: str, fallback_content: str):
    """TTL切れを安全にハンドリングする"""
    try:
        cache = caching.CachedContent.get(cache_name)
        model = genai.GenerativeModel.from_cached_content(cached_content=cache)
        return model.generate_content(user_message)
    except Exception as e:
        if "INVALID_ARGUMENT" in str(e) or "not found" in str(e).lower():
            print("⚠️  キャッシュ期限切れ。再作成します...")
            cache = create_context_cache(fallback_content)
            model = genai.GenerativeModel.from_cached_content(cached_content=cache)
            return model.generate_content(user_message)
        raise

import json
import time
from dataclasses import dataclass, field
from collections import defaultdict
 
@dataclass
class CacheMetrics:
    total_requests: int = 0
    total_cached_tokens: int = 0
    total_input_tokens: int = 0
    total_cost_usd: float = 0.0
    estimated_savings_usd: float = 0.0
    hourly_stats: dict = field(default_factory=lambda: defaultdict(lambda: {
        "requests": 0, "cached_tokens": 0, "input_tokens": 0
    }))
 
# Gemini 2.5 Proの料金（2026年5月時点）
PRICING = {
    "input_per_token": 0.00000125,        # $1.25 / 1M tokens
    "cached_input_per_token": 0.000000313, # $0.3125 / 1M tokens（約25%）
    "output_per_token": 0.000005,          # $5 / 1M tokens
}
 
metrics = CacheMetrics()
 
def track_and_generate(model, prompt: str) -> str:
    """メトリクスを収集しながらGeminiを呼ぶ"""
    response = model.generate_content(prompt)
    usage = response.usage_metadata
    
    # メトリクス更新
    cached = usage.cached_content_token_count or 0
    total_input = usage.prompt_token_count or 0
    non_cached_input = total_input - cached
    
    # コスト計算
    actual_cost = (
        non_cached_input * PRICING["input_per_token"] +
        cached * PRICING["cached_input_per_token"] +
        (usage.candidates_token_count or 0) * PRICING["output_per_token"]
    )
    hypothetical_cost = (
        total_input * PRICING["input_per_token"] +
        (usage.candidates_token_count or 0) * PRICING["output_per_token"]
    )
    savings = hypothetical_cost - actual_cost
    
    metrics.total_requests += 1
    metrics.total_cached_tokens += cached
    metrics.total_input_tokens += total_input
    metrics.total_cost_usd += actual_cost
    metrics.estimated_savings_usd += savings
    
    hour_key = time.strftime("%Y-%m-%d %H:00")
    metrics.hourly_stats[hour_key]["requests"] += 1
    metrics.hourly_stats[hour_key]["cached_tokens"] += cached
    
    return response.text
 
def print_cost_report():
    """コスト削減レポートを出力"""
    if metrics.total_requests == 0:
        return
    
    hit_rate = metrics.total_cached_tokens / max(metrics.total_input_tokens, 1) * 100
    
    print("\n=== Gemini API キャッシュ効果レポート ===")
    print(f"総リクエスト数: {metrics.total_requests:,}")
    print(f"キャッシュヒット率: {hit_rate:.1f}%")
    print(f"実際のAPI費用: ${metrics.total_cost_usd:.4f}")
    print(f"推定節約額: ${metrics.estimated_savings_usd:.4f}")
    print(f"節約率: {metrics.estimated_savings_usd / max(metrics.total_cost_usd + metrics.estimated_savings_usd, 0.0001) * 100:.1f}%")