Gemini API キャッシュ戦略の運用ノート — Context Caching と Implicit Caching を本番で組み合わせて月額費用を抑える

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
SYSTEM_DOC = "ここに長い固定コンテンツが入ります..."  # 実際には数千トークン以上
 
# ❌ 悪い例: 変動する内容が先頭にある
def bad_prompt_structure(user_message: str) -> str:
    """ユーザーメッセージが先頭にあるため、Implicit Caching が効かない"""
    prompt = f"{user_message}\n\n以下のドキュメントを参照してください:\n{SYSTEM_DOC}"
    model = genai.GenerativeModel("gemini-2.0-flash")
    response = model.generate_content(prompt)
    return response.text
 
# ✅ 良い例: 固定コンテンツを先頭に、変動コンテンツを末尾に
def good_prompt_structure(user_message: str) -> str:
    """固定部分が先頭に来るため、Implicit Caching のヒット率が上がる"""
    prompt = f"""以下のドキュメントを参考に、ユーザーの質問に回答してください。
 
===ドキュメント開始===
{SYSTEM_DOC}
===ドキュメント終了===
 
ユーザーの質問: {user_message}"""
 
    model = genai.GenerativeModel("gemini-2.0-flash")
    response = model.generate_content(prompt)
 
    # キャッシュヒットを確認
    usage = response.usage_metadata
    cached = getattr(usage, "cached_content_token_count", 0)
    if cached > 0:
        print(f"🎯 Implicit Cache ヒット: {cached}トークン節約")
    else:
        print("💡 キャッシュミス（通常料金）")
 
    return response.text
 
# ✅ 最も効果的: system_instruction を使う
def use_system_instruction(user_message: str) -> str:
    """system_instruction に固定コンテンツを置くことで最もヒット率が高まる"""
    model = genai.GenerativeModel(
        "gemini-2.0-flash",
        system_instruction=SYSTEM_DOC,  # 固定コンテンツは system_instruction に
    )
    response = model.generate_content(user_message)
 
    usage = response.usage_metadata
    cached = getattr(usage, "cached_content_token_count", 0)
    print(f"入力: {usage.prompt_token_count} / キャッシュ: {cached}")
 
    return response.text

import google.generativeai as genai
from google.generativeai import caching
import datetime
import threading
from typing import Optional
import logging
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
logger = logging.getLogger(__name__)
 
class ProductionCacheStrategy:
    """
    本番環境向けのキャッシュ管理クラス。
    Context Caching と Implicit Caching を組み合わせてコストを最小化する。
    """
 
    def __init__(
        self,
        model_name: str = "models/gemini-2.5-pro",
        system_prompt: str = "",
        cache_ttl_hours: int = 12,
        min_token_threshold: int = 2048,  # Context Caching を使う最小トークン数
        cache_display_name: str = "prod-cache",
    ):
        self.model_name = model_name
        self.system_prompt = system_prompt
        self.cache_ttl_hours = cache_ttl_hours
        self.min_token_threshold = min_token_threshold
        self.cache_display_name = cache_display_name
        self._cache: Optional[caching.CachedContent] = None
        self._lock = threading.Lock()  # 複数スレッドからの同時アクセスを防ぐ
        self._request_count = 0
        self._cache_hit_count = 0
 
    def _estimate_tokens(self, text: str) -> int:
        """トークン数の概算（概算値。正確には model.count_tokens() を使うこと）"""
        return len(text) // 3  # 日英混在の場合の大まかな目安
 
    def _should_use_context_cache(self) -> bool:
        """Context Caching を使うべきか判定"""
        estimated = self._estimate_tokens(self.system_prompt)
        return estimated >= self.min_token_threshold
 
    def _get_or_create_cache(self) -> Optional[caching.CachedContent]:
        """キャッシュを取得（なければ作成）するスレッドセーフなメソッド"""
        with self._lock:
            # 既存キャッシュの有効期限を確認
            if self._cache is not None:
                now = datetime.datetime.now(datetime.timezone.utc)
                remaining_sec = (self._cache.expire_time - now).total_seconds()
                if remaining_sec > 300:  # 5 分以上残っていれば使用
                    return self._cache
                logger.info("キャッシュが期限切れ間近 — 再作成します")
 
            # Context Caching が適切か判定
            if not self._should_use_context_cache():
                logger.info("システムプロンプトが短いため Context Caching をスキップ（Implicit Caching で代替）")
                return None
 
            # まず既存キャッシュを名前で検索（プロセス再起動後の再利用）
            for existing in caching.CachedContent.list():
                if existing.display_name == self.cache_display_name:
                    now = datetime.datetime.now(datetime.timezone.utc)
                    remaining_h = (existing.expire_time - now).total_seconds() / 3600
                    if remaining_h > 0.5:
                        logger.info(f"既存キャッシュを再利用: {existing.name}（残り{remaining_h:.1f}h）")
                        self._cache = existing
                        return self._cache
 
            # 新規作成
            try:
                self._cache = caching.CachedContent.create(
                    model=self.model_name,
                    display_name=self.cache_display_name,
                    system_instruction=self.system_prompt,
                    ttl=datetime.timedelta(hours=self.cache_ttl_hours),
                )
                logger.info(f"新規キャッシュ作成: {self._cache.name}")
                return self._cache
            except Exception as e:
                logger.error(f"キャッシュ作成失敗（Implicit Caching にフォールバック）: {e}")
                return None
 
    def generate(self, user_message: str) -> dict:
        """
        キャッシュを活用してテキスト生成を実行する。
 
        Returns:
            dict: {"text", "cache_hit", "cached_tokens", "input_tokens", "output_tokens"}
        """
        self._request_count += 1
        cache = self._get_or_create_cache()
 
        try:
            if cache is not None:
                # Context Cache を明示的に参照するリクエスト
                model = genai.GenerativeModel.from_cached_content(cache)
                response = model.generate_content(user_message)
            else:
                # Context Cache なし（Implicit Caching が自動適用される可能性あり）
                # system_instruction を使うことで Implicit Caching のヒット率を上げる
                model = genai.GenerativeModel(
                    self.model_name,
                    system_instruction=self.system_prompt,
                )
                response = model.generate_content(user_message)
 
            usage = response.usage_metadata
            cached_tokens = getattr(usage, "cached_content_token_count", 0)
 
            if cached_tokens > 0:
                self._cache_hit_count += 1
 
            return {
                "text": response.text,
                "cache_hit": cached_tokens > 0,
                "cached_tokens": cached_tokens,
                "input_tokens": usage.prompt_token_count,
                "output_tokens": usage.candidates_token_count,
            }
        except Exception as e:
            logger.error(f"生成失敗: {e}")
            raise
 
    def get_stats(self) -> dict:
        """キャッシュのヒット統計を返す"""
        hit_rate = (
            self._cache_hit_count / self._request_count * 100
            if self._request_count > 0
            else 0
        )
        return {
            "total_requests": self._request_count,
            "cache_hits": self._cache_hit_count,
            "hit_rate_percent": round(hit_rate, 1),
        }
 
# 使用例
SYSTEM_PROMPT = """
[ここに 2,048 トークン以上の長いシステムプロンプトを配置する]
製品マニュアル、サポートポリシー、FAQ の標準回答など...
"""
 
strategy = ProductionCacheStrategy(
    model_name="models/gemini-2.5-pro",
    system_prompt=SYSTEM_PROMPT,
    cache_ttl_hours=12,
    cache_display_name="support-bot-v1",
)
 
result = strategy.generate("返品期限を教えてください")
print(f"回答: {result['text'][:200]}")
print(f"キャッシュヒット: {result['cache_hit']}")
print(f"キャッシュトークン: {result['cached_tokens']}")
print(f"統計: {strategy.get_stats()}")

import json
import pathlib
import hashlib
import google.generativeai as genai
 
CACHE_FILE = pathlib.Path("/tmp/.gemini_token_cache.json")
 
def count_tokens_cached(model_name: str, text: str) -> int:
    """count_tokens の結果をローカルに永続化して、同じテキストでは API を叩かない"""
    key = hashlib.sha256(f"{model_name}::{text}".encode("utf-8")).hexdigest()
 
    if CACHE_FILE.exists():
        try:
            store = json.loads(CACHE_FILE.read_text())
        except json.JSONDecodeError:
            store = {}
    else:
        store = {}
 
    if key in store:
        return store[key]
 
    model = genai.GenerativeModel(model_name)
    count = model.count_tokens(text).total_tokens
    store[key] = count
    CACHE_FILE.write_text(json.dumps(store))
    return count

def cleanup_test_caches() -> int:
    """CI 終了時に呼んで、テスト用キャッシュをまとめて削除"""
    deleted = 0
    for c in caching.CachedContent.list():
        if c.display_name.startswith("test-"):
            try:
                c.delete()
                deleted += 1
            except Exception as e:
                print(f"⚠️ {c.display_name} の削除に失敗: {e}")
    print(f"🧹 テストキャッシュ {deleted} 件削除")
    return deleted

import os
from datetime import date
 
def stable_display_name(base: str) -> str:
    """環境と日付を含む display_name を生成する"""
    env = os.environ.get("APP_ENV", "dev")
    # 月単位で固定にすると、毎月 1 回だけ新規作成される
    month_key = date.today().strftime("%Y-%m")
    return f"{env}-{base}-{month_key}"

import datetime
from google.generativeai import caching
 
def is_safely_usable(cache: caching.CachedContent, min_remaining_sec: int = 300) -> bool:
    """このキャッシュを使うのが安全か判定（race condition 回避）"""
    now = datetime.datetime.now(datetime.timezone.utc)
    remaining = (cache.expire_time - now).total_seconds()
    return remaining > min_remaining_sec

def cache_with_image(image_path: str, system_text: str) -> caching.CachedContent:
    """画像を含むコンテンツを Files API 経由でキャッシュする"""
    print("📤 Files API にアップロード中...")
    uploaded_file = genai.upload_file(
        path=image_path,
        display_name="brand-palette-reference",
    )
    print(f"✅ アップロード完了: {uploaded_file.uri}")
 
    return caching.CachedContent.create(
        model="models/gemini-2.5-pro",
        display_name="prod-brand-palette-2026-04",
        contents=[
            {
                "role": "user",
                "parts": [
                    {"text": system_text},
                    {"file_data": {"file_uri": uploaded_file.uri}},
                ],
            }
        ],
        ttl=datetime.timedelta(hours=6),
    )

def calculate_smart_ttl(hours_until_next_use: float, buffer_minutes: int = 30) -> int:
    """次の使用予定時刻に合わせて TTL を計算する"""
    # 0.5 時間刻みで切り上げ、最低 1 時間、最大 24 時間
    raw = hours_until_next_use + (buffer_minutes / 60)
    return max(1, min(24, int(raw + 0.999)))
 
# 使用例: バッチが 4 時間後に動く場合
ttl_hours = calculate_smart_ttl(4)
print(f"推奨 TTL: {ttl_hours}時間")  # 推奨 TTL: 5時間

def daily_budget_alert(monthly_budget_usd: float) -> float:
    """日次アラート閾値を計算する（月予算 ÷ 30 × 1.3）"""
    return (monthly_budget_usd / 30) * 1.3
 
# 月予算 $150 のアプリの場合
threshold = daily_budget_alert(150)
print(f"日次アラート閾値: ${threshold:.2f}")  # 日次アラート閾値: $6.50

from prometheus_client import Counter, Gauge, Histogram
 
# メトリクスの定義
cache_hits_total = Counter(
    "gemini_cache_hits_total",
    "キャッシュヒット数",
    ["cache_type"],  # context または implicit
)
cache_misses_total = Counter("gemini_cache_misses_total", "キャッシュミス数")
token_savings_total = Counter(
    "gemini_token_savings_total", "節約されたトークン数（推定）"
)
api_latency = Histogram(
    "gemini_api_latency_seconds",
    "API レイテンシ",
    buckets=[0.5, 1, 2, 5, 10, 30],
)
cache_hit_rate_gauge = Gauge("gemini_cache_hit_rate_percent", "現在のキャッシュヒット率")
 
class MonitoredCacheStrategy(ProductionCacheStrategy):
    """Prometheus メトリクスを記録するキャッシュ戦略（本番推奨）"""
 
    def generate(self, user_message: str) -> dict:
        import time
 
        start = time.time()
        result = super().generate(user_message)
        elapsed = time.time() - start
 
        # レイテンシを記録
        api_latency.observe(elapsed)
 
        if result["cache_hit"]:
            cache_hits_total.labels(cache_type="context_or_implicit").inc()
            # キャッシュトークンはフル料金より約 75% 割引 → 75% 分を節約として計上
            savings = int(result["cached_tokens"] * 0.75)
            token_savings_total.inc(savings)
        else:
            cache_misses_total.inc()
 
        stats = self.get_stats()
        cache_hit_rate_gauge.set(stats["hit_rate_percent"])
 
        return result