Gemini API × Sentry でLLMエラー追跡とプロンプト失敗を観測する本番運用パイプライン

pip install "sentry-sdk[fastapi]==2.15.0" "google-genai==1.5.0"

import os
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
 
def gemini_traces_sampler(sampling_context: dict) -> float:
    """Gemini 関連スパンは個別にサンプルレートを下げる。
 
    通常リクエストは 5%、エラーが疑われるスパンは 100% 採取する。
    """
    op = sampling_context.get("transaction_context", {}).get("op", "")
    if op == "gemini.generate":
        return 0.05
    if op == "gemini.generate.failed":
        return 1.0
    return 0.1
 
sentry_sdk.init(
    dsn=os.environ["SENTRY_DSN"],
    environment=os.environ.get("APP_ENV", "production"),
    release=os.environ.get("APP_VERSION", "dev"),
    traces_sampler=gemini_traces_sampler,
    profiles_sample_rate=0.0,
    send_default_pii=False,
    integrations=[FastApiIntegration()],
)

import hashlib
import time
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Any, Optional
 
import sentry_sdk
from google import genai
from google.genai import types as genai_types
 
@dataclass
class GeminiCallResult:
    text: str
    finish_reason: str
    block_reason: Optional[str]
    prompt_tokens: int
    candidates_tokens: int
    latency_ms: float
    model: str
    prompt_hash: str
 
class ObservedGeminiClient:
    """Gemini API の薄いラッパー。
    すべてのコールを Sentry の span として記録し、
    失敗パターンを別イベントとして送信する。"""
 
    SAFE_FINISH = {"STOP", "MAX_TOKENS"}
 
    def __init__(self, api_key: str, default_model: str = "gemini-2.5-flash"):
        self.client = genai.Client(api_key=api_key)
        self.default_model = default_model
 
    def _prompt_hash(self, prompt: str) -> str:
        return hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:12]
 
    def generate(
        self,
        prompt: str,
        *,
        model: Optional[str] = None,
        config: Optional[genai_types.GenerateContentConfig] = None,
        user_id_hash: Optional[str] = None,
    ) -> GeminiCallResult:
        model = model or self.default_model
        prompt_hash = self._prompt_hash(prompt)
 
        with sentry_sdk.start_span(
            op="gemini.generate",
            description=f"{model} / {prompt_hash}",
        ) as span:
            span.set_tag("gemini.model", model)
            span.set_tag("gemini.prompt_hash", prompt_hash)
            if user_id_hash:
                span.set_tag("user.id_hash", user_id_hash)
 
            t0 = time.perf_counter()
            try:
                resp = self.client.models.generate_content(
                    model=model,
                    contents=prompt,
                    config=config,
                )
            except Exception as exc:
                latency_ms = (time.perf_counter() - t0) * 1000
                span.set_status("internal_error")
                self._capture_transport_error(exc, model, prompt_hash, latency_ms)
                raise
            latency_ms = (time.perf_counter() - t0) * 1000
 
            result = self._parse(resp, model, prompt_hash, latency_ms)
            span.set_data("gemini.finish_reason", result.finish_reason)
            span.set_data("gemini.tokens.prompt", result.prompt_tokens)
            span.set_data("gemini.tokens.candidates", result.candidates_tokens)
            span.set_measurement("latency_ms", latency_ms, "millisecond")
 
            self._inspect(result)
            return result

    def _parse(
        self,
        resp: Any,
        model: str,
        prompt_hash: str,
        latency_ms: float,
    ) -> GeminiCallResult:
        # 注意: candidates が空の場合がある（block_reason で弾かれた等）
        candidates = getattr(resp, "candidates", []) or []
        candidate = candidates[0] if candidates else None
        finish_reason = (
            str(candidate.finish_reason).split(".")[-1]
            if candidate and candidate.finish_reason
            else "UNKNOWN"
        )
        text = (resp.text or "") if hasattr(resp, "text") else ""
 
        block_reason = None
        pf = getattr(resp, "prompt_feedback", None)
        if pf and getattr(pf, "block_reason", None):
            block_reason = str(pf.block_reason).split(".")[-1]
 
        usage = getattr(resp, "usage_metadata", None)
        prompt_tokens = getattr(usage, "prompt_token_count", 0) if usage else 0
        candidates_tokens = (
            getattr(usage, "candidates_token_count", 0) if usage else 0
        )
 
        return GeminiCallResult(
            text=text,
            finish_reason=finish_reason,
            block_reason=block_reason,
            prompt_tokens=prompt_tokens,
            candidates_tokens=candidates_tokens,
            latency_ms=latency_ms,
            model=model,
            prompt_hash=prompt_hash,
        )
 
    def _inspect(self, r: GeminiCallResult) -> None:
        """成功した失敗パターンを検出し Sentry に通知。"""
        if r.block_reason:
            self._capture_prompt_block(r)
            return
        if r.finish_reason == "SAFETY":
            self._capture_safety_block(r)
            return
        if r.finish_reason == "RECITATION":
            self._capture_recitation_block(r)
            return
        if r.finish_reason not in self.SAFE_FINISH:
            self._capture_unknown_finish(r)
            return
        if not r.text.strip():
            self._capture_empty_response(r)
            return
        if r.candidates_tokens > 0 and r.candidates_tokens < 16:
            # 極端に短いレスポンスは品質劣化のサインになる
            self._capture_thin_response(r)
 
    def _capture_safety_block(self, r: GeminiCallResult) -> None:
        with sentry_sdk.push_scope() as scope:
            scope.set_tag("gemini.failure", "safety_block")
            scope.set_tag("gemini.model", r.model)
            scope.set_tag("gemini.prompt_hash", r.prompt_hash)
            scope.set_extra("tokens_prompt", r.prompt_tokens)
            scope.set_extra("latency_ms", r.latency_ms)
            scope.set_level("warning")
            sentry_sdk.capture_message(
                f"Gemini SAFETY block: {r.prompt_hash}",
                level="warning",
            )

import re
 
EMAIL_RE = re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}")
PHONE_RE = re.compile(r"\+?\d[\d\-\s\(\)]{7,}\d")
CC_RE = re.compile(r"\b(?:\d[ -]*?){13,19}\b")
 
def _scrub(text: str) -> str:
    text = EMAIL_RE.sub("[email]", text)
    text = PHONE_RE.sub("[phone]", text)
    text = CC_RE.sub("[cc]", text)
    return text
 
def before_send(event: dict, hint: dict) -> Optional[dict]:
    # 1) message を必ずスクラブ
    if event.get("message"):
        event["message"] = _scrub(event["message"])
    # 2) breadcrumbs から本文っぽい巨大文字列を落とす
    for crumb in event.get("breadcrumbs", {}).get("values", []) or []:
        data = crumb.get("data") or {}
        for k, v in list(data.items()):
            if isinstance(v, str) and len(v) > 200:
                data[k] = f"<redacted: {len(v)} chars>"
            elif isinstance(v, str):
                data[k] = _scrub(v)
    # 3) extra から prompt キーを問答無用で消す
    extra = event.get("extra") or {}
    for k in list(extra.keys()):
        if "prompt" in k.lower() or "input" in k.lower():
            extra[k] = "<scrubbed>"
    # 4) 環境変数経由のシークレットを消す
    env_keys = ("API_KEY", "TOKEN", "SECRET", "PASSWORD")
    request = event.get("request") or {}
    env = request.get("env") or {}
    for k in list(env.keys()):
        if any(s in k.upper() for s in env_keys):
            env[k] = "<scrubbed>"
    return event

def before_send_transaction(event, hint):
    tx = event.get("transaction", "")
    if tx.startswith("GET /healthz"):
        return None
    return event

scope.fingerprint = ["gemini-safety-block", r.model]

def report_eval_summary(pass_rate: float, sample_size: int, dataset: str):
    with sentry_sdk.push_scope() as scope:
        scope.set_tag("eval.dataset", dataset)
        scope.set_extra("sample_size", sample_size)
        scope.set_extra("pass_rate", pass_rate)
        if pass_rate < 0.85:
            scope.set_level("error")
            sentry_sdk.capture_message(
                f"Eval regression: {dataset} pass_rate={pass_rate:.2%}",
                level="error",
            )
        elif pass_rate < 0.92:
            scope.set_level("warning")
            sentry_sdk.capture_message(
                f"Eval warning: {dataset} pass_rate={pass_rate:.2%}",
                level="warning",
            )