Gemini API 本番環境のオブザーバビリティ — ログ・監視・コスト追跡の実践パターン

from collections import defaultdict
from dataclasses import dataclass, field
import threading
import time
 
@dataclass
class RequestMetric:
    timestamp: float
    model: str
    latency: float
    input_tokens: int
    output_tokens: int
    success: bool
    error_type: str = ""
 
class MetricsCollector:
    """Gemini APIメトリクス収集・集約クラス"""
 
    def __init__(self):
        self._metrics: list[RequestMetric] = []
        self._lock = threading.Lock()
 
    def record_request(self, model: str, latency: float,
                       input_tokens: int, output_tokens: int,
                       success: bool, error_type: str = ""):
        metric = RequestMetric(
            timestamp=time.time(),
            model=model,
            latency=latency,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            success=success,
            error_type=error_type
        )
        with self._lock:
            self._metrics.append(metric)
 
    def get_summary(self, window_minutes: int = 60) -> dict:
        """指定期間のメトリクスサマリーを返す"""
        cutoff = time.time() - (window_minutes * 60)
        with self._lock:
            recent = [m for m in self._metrics if m.timestamp > cutoff]
 
        if not recent:
            return {"period_minutes": window_minutes, "total_requests": 0}
 
        latencies = sorted([m.latency for m in recent if m.success])
        total_input = sum(m.input_tokens for m in recent)
        total_output = sum(m.output_tokens for m in recent)
        errors = [m for m in recent if not m.success]
 
        # モデル別の集計
        by_model = defaultdict(lambda: {"count": 0, "tokens": 0})
        for m in recent:
            by_model[m.model]["count"] += 1
            by_model[m.model]["tokens"] += m.input_tokens + m.output_tokens
 
        return {
            "period_minutes": window_minutes,
            "total_requests": len(recent),
            "success_rate": round((len(recent) - len(errors)) / len(recent) * 100, 2),
            "latency": {
                "p50": round(latencies[len(latencies) // 2] * 1000, 2) if latencies else 0,
                "p95": round(latencies[int(len(latencies) * 0.95)] * 1000, 2) if latencies else 0,
                "p99": round(latencies[int(len(latencies) * 0.99)] * 1000, 2) if latencies else 0,
            },
            "tokens": {
                "total_input": total_input,
                "total_output": total_output,
                "total": total_input + total_output,
            },
            "errors": {
                "count": len(errors),
                "by_type": dict(defaultdict(int, {
                    e.error_type: sum(1 for x in errors if x.error_type == e.error_type)
                    for e in errors
                }))
            },
            "by_model": dict(by_model),
        }

from prometheus_client import Counter, Histogram, Gauge, start_http_server
 
# Prometheusメトリクス定義
gemini_requests_total = Counter(
    "gemini_api_requests_total",
    "Gemini APIリクエスト総数",
    ["model", "status"]
)
gemini_latency = Histogram(
    "gemini_api_latency_seconds",
    "Gemini APIレイテンシ（秒）",
    ["model"],
    buckets=[0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]
)
gemini_tokens = Counter(
    "gemini_api_tokens_total",
    "Gemini APIトークン消費量",
    ["model", "direction"]  # direction: input / output
)
gemini_estimated_cost = Counter(
    "gemini_api_estimated_cost_usd",
    "Gemini API推定コスト（USD）",
    ["model"]
)
 
# メトリクスサーバー起動（ポート9090）
# start_http_server(9090)

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class ModelPricing:
    """モデル別料金（USD / 1Mトークン）"""
    input_per_million: float
    output_per_million: float
    context_cache_per_million: Optional[float] = None
 
# 2026年3月時点のGemini API料金（公式料金ページを定期確認すること）
PRICING = {
    "gemini-2.5-pro": ModelPricing(
        input_per_million=1.25,
        output_per_million=10.00,
        context_cache_per_million=0.3125
    ),
    "gemini-2.5-flash": ModelPricing(
        input_per_million=0.15,
        output_per_million=0.60,
        context_cache_per_million=0.0375
    ),
    "gemini-2.0-flash": ModelPricing(
        input_per_million=0.10,
        output_per_million=0.40
    ),
    "gemini-2.0-flash-lite": ModelPricing(
        input_per_million=0.075,
        output_per_million=0.30
    ),
}
 
class CostTracker:
    """リアルタイムコスト追跡"""
 
    def __init__(self):
        self._costs: list[dict] = []
        self._lock = threading.Lock()
 
    def calculate_cost(self, model: str, input_tokens: int,
                       output_tokens: int,
                       cached_tokens: int = 0) -> float:
        """リクエストごとのコストを計算"""
        pricing = PRICING.get(model)
        if not pricing:
            logger.warning("unknown_model_pricing", model=model)
            return 0.0
 
        input_cost = (input_tokens / 1_000_000) * pricing.input_per_million
        output_cost = (output_tokens / 1_000_000) * pricing.output_per_million
 
        # コンテキストキャッシュ利用時の割引
        cache_discount = 0.0
        if cached_tokens > 0 and pricing.context_cache_per_million:
            full_price = (cached_tokens / 1_000_000) * pricing.input_per_million
            cache_price = (cached_tokens / 1_000_000) * pricing.context_cache_per_million
            cache_discount = full_price - cache_price
 
        total = input_cost + output_cost - cache_discount
 
        cost_entry = {
            "timestamp": time.time(),
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cached_tokens": cached_tokens,
            "cost_usd": round(total, 6),
            "cache_savings_usd": round(cache_discount, 6)
        }
 
        with self._lock:
            self._costs.append(cost_entry)
 
        return total
 
    def get_daily_summary(self) -> dict:
        """本日のコストサマリーを返す"""
        import datetime
        today_start = datetime.datetime.now().replace(
            hour=0, minute=0, second=0
        ).timestamp()
 
        with self._lock:
            today = [c for c in self._costs if c["timestamp"] > today_start]
 
        total_cost = sum(c["cost_usd"] for c in today)
        by_model = defaultdict(float)
        for c in today:
            by_model[c["model"]] += c["cost_usd"]
 
        return {
            "date": datetime.date.today().isoformat(),
            "total_cost_usd": round(total_cost, 4),
            "total_requests": len(today),
            "by_model": {k: round(v, 4) for k, v in by_model.items()},
            "total_savings_from_cache": round(
                sum(c["cache_savings_usd"] for c in today), 4
            )
        }

from collections import deque
import statistics
 
class LatencyMonitor:
    """レイテンシ異常検知モニター"""
 
    def __init__(self, window_size: int = 100,
                 static_threshold_ms: float = 10000,
                 sigma_multiplier: float = 3.0):
        self._window = deque(maxlen=window_size)
        self.static_threshold_ms = static_threshold_ms
        self.sigma_multiplier = sigma_multiplier
 
    def record_and_check(self, latency_ms: float, model: str) -> dict:
        """レイテンシを記録し、異常かどうかを判定する"""
        self._window.append(latency_ms)
 
        alert = {"is_anomaly": False, "reason": ""}
 
        # 静的閾値チェック
        if latency_ms > self.static_threshold_ms:
            alert = {
                "is_anomaly": True,
                "reason": f"レイテンシ {latency_ms:.0f}ms が"
                          f"閾値 {self.static_threshold_ms}ms を超過"
            }
 
        # 動的異常検知（σルール: 平均 + N×標準偏差を超えたら異常）
        if len(self._window) >= 20:
            mean = statistics.mean(self._window)
            stdev = statistics.stdev(self._window)
            dynamic_threshold = mean + (self.sigma_multiplier * stdev)
 
            if latency_ms > dynamic_threshold and not alert["is_anomaly"]:
                alert = {
                    "is_anomaly": True,
                    "reason": f"レイテンシ {latency_ms:.0f}ms が"
                              f"動的閾値 {dynamic_threshold:.0f}ms を超過"
                              f"（平均={mean:.0f}, σ={stdev:.0f}）"
                }
 
        if alert["is_anomaly"]:
            logger.warning("latency_anomaly_detected",
                           model=model,
                           latency_ms=latency_ms,
                           reason=alert["reason"])
 
        return alert

import json
from urllib.request import Request, urlopen
 
def send_slack_alert(webhook_url: str, alert_data: dict):
    """Slackにアラートを送信する"""
    message = {
        "text": ":rotating_light: Gemini API レイテンシ異常検知",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*モデル*: `{alert_data.get('model', 'unknown')}`\n"
                            f"*レイテンシ*: {alert_data.get('latency_ms', 0):.0f}ms\n"
                            f"*理由*: {alert_data.get('reason', '')}"
                }
            }
        ]
    }
    req = Request(webhook_url,
                  data=json.dumps(message).encode(),
                  headers={"Content-Type": "application/json"})
    urlopen(req)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
 
# OTel初期化
resource = Resource.create({"service.name": "gemini-api-service"})
provider = TracerProvider(resource=resource)
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
 
tracer = trace.get_tracer("gemini.api")
 
class TracedGeminiClient(GeminiObservableClient):
    """OpenTelemetryトレーシング付きGeminiクライアント"""
 
    def generate_with_trace(self, prompt: str,
                            model: str = None, config: dict = None):
        model_name = model or self.default_model
 
        with tracer.start_as_current_span("gemini.generate") as span:
            span.set_attribute("gemini.model", model_name)
            span.set_attribute("gemini.prompt_length", len(prompt))
 
            try:
                response = self.generate(
                    prompt=prompt,
                    model=model_name,
                    config=config,
                    trace_id=format(span.get_span_context().trace_id, "032x")
                )
 
                span.set_attribute("gemini.input_tokens",
                                   response.usage_metadata.prompt_token_count)
                span.set_attribute("gemini.output_tokens",
                                   response.usage_metadata.candidates_token_count)
                span.set_status(trace.StatusCode.OK)
                return response
 
            except Exception as e:
                span.set_status(trace.StatusCode.ERROR, str(e))
                span.record_exception(e)
                raise
 
    def rag_pipeline(self, query: str, context_docs: list[str]):
        """RAGパイプラインをトレース付きで実行"""
        with tracer.start_as_current_span("rag.pipeline") as parent:
            parent.set_attribute("rag.query", query)
            parent.set_attribute("rag.context_count", len(context_docs))
 
            # Step 1: クエリ拡張
            with tracer.start_as_current_span("rag.query_expansion"):
                expanded = self.generate_with_trace(
                    f"検索クエリを拡張してください: {query}",
                    model="gemini-2.5-flash"
                )
 
            # Step 2: コンテキスト付きで回答生成
            context = "\n---\n".join(context_docs)
            with tracer.start_as_current_span("rag.answer_generation"):
                answer = self.generate_with_trace(
                    f"コンテキスト:\n{context}\n\n質問: {query}",
                    model="gemini-2.5-pro"
                )
 
            return answer

import re
 
def sanitize_prompt(text: str) -> str:
    """プロンプト内の個人情報をマスクする"""
    # メールアドレス
    text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[EMAIL]', text)
    # 電話番号（日本）
    text = re.sub(r'0\d{1,4}-?\d{1,4}-?\d{4}', '[PHONE]', text)
    # クレジットカード番号
    text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', text)
    return text

class AdaptiveModelSelector:
    """メトリクスベースの自動モデル選択"""
 
    def __init__(self, latency_monitor: LatencyMonitor,
                 primary: str = "gemini-2.5-pro",
                 fallback: str = "gemini-2.5-flash",
                 latency_threshold_ms: float = 8000):
        self.latency_monitor = latency_monitor
        self.primary = primary
        self.fallback = fallback
        self.threshold = latency_threshold_ms
        self._use_fallback = False
 
    def select_model(self) -> str:
        """現在のレイテンシ状況に応じたモデルを返す"""
        if self._use_fallback:
            # フォールバック中は定期的にプライマリに戻す試行
            return self.primary if time.time() % 60 < 5 else self.fallback
        return self.primary
 
    def on_response(self, model: str, latency_ms: float):
        """レスポンス受信時にフォールバック判定"""
        result = self.latency_monitor.record_and_check(latency_ms, model)
        if result["is_anomaly"] and model == self.primary:
            self._use_fallback = True
            logger.warning("model_fallback_activated",
                           primary=self.primary,
                           fallback=self.fallback)