Gemini API 本番パフォーマンスチューニング — レイテンシ・スループット・コストの3軸最適化戦略

import asyncio
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
async def resilient_stream(
    model: str,
    prompt: str,
    timeout_seconds: float = 30.0,
    max_retries: int = 2,
):
    """タイムアウト・リトライ付きのストリーミング生成"""
    last_error = None
 
    for attempt in range(max_retries + 1):
        try:
            buffer = []
            async for chunk in await client.aio.models.generate_content_stream(
                model=model,
                contents=prompt,
                config=types.GenerateContentConfig(
                    temperature=0.7,
                    max_output_tokens=2048,
                ),
            ):
                if chunk.text:
                    buffer.append(chunk.text)
                    yield chunk.text  # クライアントへ即座に送信
 
            return  # 成功したらループを抜ける
 
        except Exception as e:
            last_error = e
            if attempt < max_retries:
                # 指数バックオフ: 1秒, 2秒, 4秒...
                wait = 2 ** attempt
                print(f"⚠ Attempt {attempt+1} failed: {e}. Retry in {wait}s")
                await asyncio.sleep(wait)
            else:
                raise RuntimeError(
                    f"Streaming failed after {max_retries+1} attempts: {last_error}"
                )
 
# FastAPI との統合例
# from fastapi import FastAPI
# from fastapi.responses import StreamingResponse
# app = FastAPI()
#
# @app.get("/chat")
# async def chat(q: str):
#     return StreamingResponse(
#         resilient_stream("gemini-2.5-flash", q),
#         media_type="text/plain",
#     )

from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class AdaptiveInference:
    """リクエストの性質に応じて推論ティアを自動選択する"""
 
    def __init__(self):
        self.model = "gemini-2.5-flash"
 
    async def generate(
        self,
        prompt: str,
        priority: str = "auto",
        user_facing: bool = True,
    ) -> str:
        # ティアの自動判定
        if priority == "auto":
            tier = self._select_tier(prompt, user_facing)
        else:
            tier = priority
 
        config_kwargs = {
            "temperature": 0.7,
            "max_output_tokens": 2048,
        }
 
        # Flex / Priority の設定
        # service_tier を httpOptions で渡す
        # （2026年4月時点の Python SDK では
        #   GenerateContentConfig に service_tier を追加）
        if tier == "flex":
            config_kwargs["http_options"] = {"headers": {"X-Goog-Api-Service-Tier": "flex"}}
        elif tier == "priority":
            config_kwargs["http_options"] = {"headers": {"X-Goog-Api-Service-Tier": "priority"}}
 
        response = await client.aio.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(**config_kwargs),
        )
        return response.text
 
    def _select_tier(self, prompt: str, user_facing: bool) -> str:
        """リクエスト特性からティアを判定"""
        # ユーザーが画面の前で待っている → Standard or Priority
        if user_facing:
            # 短いプロンプトは Standard で十分速い
            if len(prompt) < 500:
                return "standard"
            # 長文処理でも対話型なら Priority で低レイテンシを保証
            return "priority"
 
        # バックグラウンド処理 → Flex でコスト50%削減
        return "flex"
 
# 使用例
engine = AdaptiveInference()
 
# ユーザー対話: Standard で即応答
# answer = await engine.generate("この関数のバグを教えて", user_facing=True)
 
# バッチレポート生成: Flex で50%コスト削減
# report = await engine.generate(long_document, user_facing=False)
 
# 決済確認メッセージ: Priority で最高信頼性
# confirmation = await engine.generate(payment_prompt, priority="priority")

from google import genai
from google.genai import types
import hashlib
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class SmartCacheManager:
    """Context Caching の損益分岐点を考慮した自動管理"""
 
    # キャッシュストレージ: $1.00 / 1M tokens / hour（2.5 Flash の場合）
    # 通常入力: $0.15 / 1M tokens
    # キャッシュ入力: $0.0375 / 1M tokens（75%割引）
    # 損益分岐: 1時間あたり N 回以上同じコンテキストを送るなら得
    # N = storage_cost / (normal_cost - cached_cost) per token
    # N ≈ $1.00 / ($0.15 - $0.0375) = 8.9 回/時間
 
    BREAK_EVEN_REQUESTS_PER_HOUR = 9  # 1時間に9回以上で元が取れる
 
    def __init__(self):
        self._cache_registry: dict[str, str] = {}  # hash -> cache_name
        self._request_counts: dict[str, list[float]] = {}  # hash -> timestamps
 
    def _content_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()[:16]
 
    def should_cache(self, content: str, expected_rpm: float) -> bool:
        """キャッシュすべきかどうかを判定"""
        token_estimate = len(content) / 4  # 粗い推定
        if token_estimate < 32768:
            return False  # 最低トークン数未満
        return expected_rpm * 60 >= self.BREAK_EVEN_REQUESTS_PER_HOUR
 
    async def get_or_create_cache(
        self,
        system_content: str,
        model: str = "gemini-2.5-flash",
        ttl_minutes: int = 60,
    ) -> str | None:
        """キャッシュを取得、なければ作成。損益分岐に達しない場合はNone"""
        content_hash = self._content_hash(system_content)
 
        # 既存キャッシュがあればそのまま返す
        if content_hash in self._cache_registry:
            return self._cache_registry[content_hash]
 
        # 新規キャッシュ作成
        cache = await client.aio.caches.create(
            model=model,
            config=types.CreateCachedContentConfig(
                contents=[
                    types.Content(
                        role="user",
                        parts=[types.Part(text=system_content)],
                    )
                ],
                ttl=f"{ttl_minutes * 60}s",
                display_name=f"ctx-{content_hash}",
            ),
        )
        self._cache_registry[content_hash] = cache.name
        return cache.name
 
    async def generate_with_cache(
        self,
        cache_name: str,
        user_prompt: str,
        model: str = "gemini-2.5-flash",
    ) -> str:
        """キャッシュを使った生成"""
        response = await client.aio.models.generate_content(
            model=model,
            contents=user_prompt,
            config=types.GenerateContentConfig(
                cached_content=cache_name,
                temperature=0.7,
            ),
        )
        return response.text
 
# 使用例: 大量のカスタマーサポート問い合わせを同じナレッジベースで処理
# manager = SmartCacheManager()
# cache = await manager.get_or_create_cache(knowledge_base_text)
# if cache:
#     answer = await manager.generate_with_cache(cache, user_question)
# else:
#     # キャッシュ不要（リクエスト頻度が低い場合）→ 通常リクエスト
#     answer = await client.aio.models.generate_content(...)

import re
from dataclasses import dataclass
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
@dataclass
class RoutingDecision:
    model: str
    reason: str
    estimated_cost_multiplier: float  # Flash基準の倍率
 
class ModelRouter:
    """タスク複雑度に基づくモデルルーティング"""
 
    COMPLEXITY_SIGNALS = {
        "high": [
            r"(?:分析|analyze|compare|比較).*(?:詳細|detail|comprehensive)",
            r"(?:設計|design|architect).*(?:システム|system|production)",
            r"(?:リファクタ|refactor|最適化|optimize).*(?:コード|code)",
            r"(?:数学|math|証明|prove|theorem)",
        ],
        "low": [
            r"(?:翻訳|translate|要約|summar)",
            r"(?:フォーマット|format|変換|convert)",
            r"(?:リスト|list|列挙|enumerate)",
            r"(?:簡単|simple|短く|brief)",
        ],
    }
 
    def route(self, prompt: str, context_tokens: int = 0) -> RoutingDecision:
        complexity = self._estimate_complexity(prompt, context_tokens)
 
        if complexity >= 0.7:
            return RoutingDecision(
                model="gemini-2.5-pro",
                reason=f"High complexity ({complexity:.2f}): Pro model needed",
                estimated_cost_multiplier=8.0,  # Proの入力コストはFlashの約8倍
            )
        elif complexity >= 0.4:
            return RoutingDecision(
                model="gemini-2.5-flash",
                reason=f"Medium complexity ({complexity:.2f}): Flash sufficient",
                estimated_cost_multiplier=1.0,
            )
        else:
            return RoutingDecision(
                model="gemini-2.5-flash",
                reason=f"Low complexity ({complexity:.2f}): Flash with lower tokens",
                estimated_cost_multiplier=1.0,
            )
 
    def _estimate_complexity(self, prompt: str, context_tokens: int) -> float:
        score = 0.5  # ベースライン
 
        # 高複雑度シグナルの検出
        for pattern in self.COMPLEXITY_SIGNALS["high"]:
            if re.search(pattern, prompt, re.IGNORECASE):
                score += 0.15
 
        # 低複雑度シグナルの検出
        for pattern in self.COMPLEXITY_SIGNALS["low"]:
            if re.search(pattern, prompt, re.IGNORECASE):
                score -= 0.15
 
        # コンテキスト長が長い → 複雑なタスクの可能性
        if context_tokens > 50000:
            score += 0.1
        elif context_tokens > 100000:
            score += 0.2
 
        # プロンプト自体が長い → 複雑な指示の可能性
        if len(prompt) > 2000:
            score += 0.1
 
        return max(0.0, min(1.0, score))
 
# 使用例
router = ModelRouter()
 
# 簡単なタスク → Flash
decision = router.route("この英文を日本語に翻訳してください: Hello world")
print(f"Model: {decision.model}, Reason: {decision.reason}")
# 出力: Model: gemini-2.5-flash, Reason: Low complexity (0.35): Flash with lower tokens
 
# 複雑なタスク → Pro
decision = router.route("このマイクロサービスアーキテクチャを分析して、ボトルネックを特定し、詳細な改善案を設計してください", context_tokens=80000)
print(f"Model: {decision.model}, Reason: {decision.reason}")
# 出力: Model: gemini-2.5-pro, Reason: High complexity (0.85): Pro model needed

import asyncio
import time
from collections import deque
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class AdaptiveRateLimiter:
    """429エラーに応じてレートを自動調整するリミッター"""
 
    def __init__(self, initial_rpm: int = 1800):
        self.rpm = initial_rpm
        self.min_interval = 60.0 / initial_rpm
        self._timestamps: deque[float] = deque()
        self._lock = asyncio.Lock()
        self._consecutive_429 = 0
 
    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
 
            # 1分以上前のタイムスタンプを除去
            while self._timestamps and now - self._timestamps[0] > 60:
                self._timestamps.popleft()
 
            # レート制限内か確認
            if len(self._timestamps) >= self.rpm:
                wait = 60.0 - (now - self._timestamps[0]) + 0.1
                await asyncio.sleep(wait)
 
            self._timestamps.append(time.monotonic())
 
    def on_429(self):
        """429エラー時にレートを下げる"""
        self._consecutive_429 += 1
        reduction = min(0.5, 0.1 * self._consecutive_429)  # 最大50%削減
        self.rpm = max(10, int(self.rpm * (1 - reduction)))
        self.min_interval = 60.0 / self.rpm
        print(f"⚠ Rate reduced to {self.rpm} RPM (429 count: {self._consecutive_429})")
 
    def on_success(self):
        """成功時にレートを徐々に回復"""
        if self._consecutive_429 > 0:
            self._consecutive_429 = max(0, self._consecutive_429 - 1)
            self.rpm = min(1800, int(self.rpm * 1.05))
            self.min_interval = 60.0 / self.rpm
 
async def batch_generate(
    prompts: list[str],
    model: str = "gemini-2.5-flash",
    concurrency: int = 10,
    rate_limiter: AdaptiveRateLimiter | None = None,
) -> list[str]:
    """適応型レート制御付きバッチ生成"""
    if rate_limiter is None:
        rate_limiter = AdaptiveRateLimiter()
 
    semaphore = asyncio.Semaphore(concurrency)
    results: list[str | None] = [None] * len(prompts)
 
    async def process_one(index: int, prompt: str):
        async with semaphore:
            await rate_limiter.acquire()
            for attempt in range(3):
                try:
                    response = await client.aio.models.generate_content(
                        model=model,
                        contents=prompt,
                        config=types.GenerateContentConfig(
                            temperature=0.7,
                            max_output_tokens=1024,
                        ),
                    )
                    rate_limiter.on_success()
                    results[index] = response.text
                    return
                except Exception as e:
                    error_str = str(e)
                    if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
                        rate_limiter.on_429()
                        await asyncio.sleep(2 ** attempt * 2)
                    else:
                        if attempt == 2:
                            results[index] = f"ERROR: {e}"
                            return
                        await asyncio.sleep(2 ** attempt)
 
    tasks = [process_one(i, p) for i, p in enumerate(prompts)]
    await asyncio.gather(*tasks)
    return results
 
# 使用例: 1000件の商品説明を一括生成
# prompts = [f"商品名「{name}」の魅力的な説明文を50字で" for name in product_names]
# results = await batch_generate(prompts, concurrency=15)
# 期待される処理時間: Tier 2（2000 RPM）で約30秒

from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
async def generate_with_thinking_budget(
    prompt: str,
    task_type: str = "general",
) -> str:
    """タスクタイプに応じてThinking Budgetを最適化"""
 
    # タスク別のThinking Budget設定
    budget_map = {
        "simple_qa": 256,       # 簡単な質問: 最小限の推論
        "translation": 512,     # 翻訳: 軽い推論
        "summarization": 1024,  # 要約: 中程度の推論
        "coding": 4096,         # コーディング: 十分な推論
        "analysis": 8192,       # 分析: 深い推論
        "math": 16384,          # 数学・証明: 最大限の推論
        "general": 2048,        # デフォルト
    }
 
    budget = budget_map.get(task_type, budget_map["general"])
 
    response = await client.aio.models.generate_content(
        model="gemini-2.5-pro",
        contents=prompt,
        config=types.GenerateContentConfig(
            thinking_config=types.ThinkingConfig(
                thinking_budget=budget,
            ),
            temperature=0.7,
        ),
    )
 
    # 実際に使用された思考トークン数をログ出力
    if hasattr(response, "usage_metadata"):
        meta = response.usage_metadata
        thinking_tokens = getattr(meta, "thinking_tokens", "N/A")
        print(f"💭 Thinking tokens used: {thinking_tokens} / budget: {budget}")
 
    return response.text
 
# 使用例: 同じモデルでもタスクに応じてコストが変わる
# 簡単な質問（256トークン推論）→ 約$0.001
# answer = await generate_with_thinking_budget("Pythonのリスト内包表記とは？", "simple_qa")
 
# 数学の証明（16384トークン推論）→ 約$0.05
# proof = await generate_with_thinking_budget("フェルマーの小定理を証明して", "math")

import time
import json
from dataclasses import dataclass, field, asdict
from collections import defaultdict
 
@dataclass
class RequestMetrics:
    model: str = ""
    tier: str = "standard"
    ttft_ms: float = 0
    total_ms: float = 0
    input_tokens: int = 0
    output_tokens: int = 0
    thinking_tokens: int = 0
    cached_tokens: int = 0
    status: str = "success"
    estimated_cost_usd: float = 0
 
class PerformanceMonitor:
    """3軸メトリクスの収集と分析"""
 
    # 2026年4月時点の価格（1Mトークンあたり、USD）
    PRICING = {
        "gemini-2.5-flash": {"input": 0.15, "output": 0.60, "cached": 0.0375, "thinking": 0.15},
        "gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached": 0.3125, "thinking": 1.25},
    }
 
    def __init__(self):
        self.metrics: list[RequestMetrics] = []
 
    def record(self, m: RequestMetrics):
        m.estimated_cost_usd = self._calc_cost(m)
        self.metrics.append(m)
 
    def _calc_cost(self, m: RequestMetrics) -> float:
        prices = self.PRICING.get(m.model, self.PRICING["gemini-2.5-flash"])
        tier_mult = {"standard": 1.0, "flex": 0.5, "priority": 1.875}.get(m.tier, 1.0)
 
        normal_input = m.input_tokens - m.cached_tokens
        cost = (
            (normal_input / 1_000_000) * prices["input"] * tier_mult
            + (m.cached_tokens / 1_000_000) * prices["cached"] * tier_mult
            + (m.output_tokens / 1_000_000) * prices["output"] * tier_mult
            + (m.thinking_tokens / 1_000_000) * prices["thinking"] * tier_mult
        )
        return round(cost, 6)
 
    def summary(self, last_n: int = 100) -> dict:
        recent = self.metrics[-last_n:]
        if not recent:
            return {}
 
        successful = [m for m in recent if m.status == "success"]
        return {
            "total_requests": len(recent),
            "success_rate": f"{len(successful)/len(recent)*100:.1f}%",
            "avg_ttft_ms": f"{sum(m.ttft_ms for m in successful)/max(len(successful),1):.0f}",
            "p95_ttft_ms": f"{sorted(m.ttft_ms for m in successful)[int(len(successful)*0.95)]:.0f}" if successful else "N/A",
            "avg_total_ms": f"{sum(m.total_ms for m in successful)/max(len(successful),1):.0f}",
            "total_cost_usd": f"${sum(m.estimated_cost_usd for m in recent):.4f}",
            "model_distribution": dict(
                defaultdict(int, {m.model: 1 for m in recent})
            ),
        }
 
    def export_jsonl(self, filepath: str):
        with open(filepath, "w") as f:
            for m in self.metrics:
                f.write(json.dumps(asdict(m)) + "\n")
 
# 使用例
# monitor = PerformanceMonitor()
# metrics = RequestMetrics(model="gemini-2.5-flash", ttft_ms=380, ...)
# monitor.record(metrics)
# print(monitor.summary())