Gemini APIのレスポンスをCIで守る — スナップショットと意味的回帰の検知設計

# test_structure.py
import json
import pytest
from gemini_client import client
from google.genai import types
 
@pytest.mark.vcr  # APIレスポンスを録画再生（後述）
def test_sentiment_schema(snapshot):
    """JSON出力のスキーマが維持されているかを検証する"""
    resp = client.models.generate_content(
        model="gemini-2.5-flash",
        contents='テキスト「この製品はとても気に入りました」の感情をJSONで。'
                 'フィールド: sentiment, confidence, explanation',
        config=types.GenerateContentConfig(
            temperature=0,
            response_mime_type="application/json",
        ),
    )
    parsed = json.loads(resp.text)
 
    # 中身ではなく「形」だけをスナップショット
    schema = {
        "has_sentiment": "sentiment" in parsed,
        "sentiment_valid": parsed.get("sentiment") in ("positive", "negative", "neutral"),
        "confidence_is_float": isinstance(parsed.get("confidence"), float),
        "has_explanation": "explanation" in parsed,
        "finish_reason": str(resp.candidates[0].finish_reason),
    }
    assert schema == snapshot

pytest test_structure.py --snapshot-update   # 初回：スナップショット作成
pytest test_structure.py                      # 以後：差分検証

# semantic_snapshot.py
import numpy as np
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
 
def embed(text: str) -> np.ndarray:
    r = client.models.embed_content(
        model="gemini-embedding-2",
        contents=text,
        config=types.EmbedContentConfig(output_dimensionality=256),
    )
    v = np.array(r.embeddings[0].values, dtype=np.float32)
    return v / np.linalg.norm(v)  # 正規化してコサイン=内積に
 
def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b))

# test_semantic.py
import numpy as np
import pytest
from pathlib import Path
from gemini_client import classify_sentiment
from semantic_snapshot import embed, cosine
 
BASELINE = Path("baseline/sentiment_review.npy")
THRESHOLD = 0.88  # 経験的に決める（後述）
 
@pytest.mark.vcr
def test_sentiment_semantic_stability():
    current = classify_sentiment("配送は遅かったが、サポートの対応は丁寧だった。")
    cur_vec = embed(current)
 
    if not BASELINE.exists():
        BASELINE.parent.mkdir(exist_ok=True)
        np.save(BASELINE, cur_vec)
        pytest.skip("baselineを新規作成しました。次回から比較します。")
 
    base_vec = np.load(BASELINE)
    sim = cosine(cur_vec, base_vec)
    assert sim >= THRESHOLD, f"意味的にずれています: cosine={sim:.3f} < {THRESHOLD}"

# pytest.ini
[pytest]
addopts = --record-mode=none

# conftest.py — APIキーをカセットに残さない
import pytest
 
@pytest.fixture(scope="module")
def vcr_config():
    return {
        "filter_headers": [("x-goog-api-key", "REDACTED"),
                           ("authorization", "REDACTED")],
        "filter_query_parameters": [("key", "REDACTED")],
        "record_mode": "none",
    }

# 新しいテストを追加したとき：本物を1回だけ叩いて録画
pytest test_semantic.py --record-mode=once
git add tests/cassettes/   # カセットをコミット
# 以後のCI：ネットワークに出ずカセット再生のみ
pytest --record-mode=none

# calibrate_threshold.py — 自然なゆらぎの下限を測る
import numpy as np
from itertools import combinations
from gemini_client import classify_sentiment
from semantic_snapshot import embed, cosine
 
samples = [embed(classify_sentiment("配送は遅かったが、対応は丁寧だった。"))
           for _ in range(20)]
sims = [cosine(a, b) for a, b in combinations(samples, 2)]
print(f"min={min(sims):.3f}  mean={np.mean(sims):.3f}  p05={np.percentile(sims,5):.3f}")

def stable_assert(check, attempts=3, need_fail=2):
    fails = sum(0 if check() else 1 for _ in range(attempts))
    assert fails < need_fail, f"{attempts}回中{fails}回失敗：回帰の可能性が高い"

# Before：内容を完全一致で固定 → モデル更新のたびに無意味に赤くなる
def test_summary():
    out = summarize("...長文...")
    assert out == "この記事はGeminiのテスト手法を解説しています。"

# After：構造（第一層）と意味（第二層）を分けて監視する
@pytest.mark.vcr
def test_summary_structure(snapshot):
    out = summarize_json("...長文...")          # {"summary": str, "topics": [...]}
    parsed = json.loads(out)
    assert {
        "has_summary": isinstance(parsed.get("summary"), str),
        "topics_is_list": isinstance(parsed.get("topics"), list),
        "topics_nonempty": len(parsed.get("topics", [])) > 0,
    } == snapshot
 
@pytest.mark.vcr
def test_summary_semantic():
    out = summarize_json("...長文...")
    sim = cosine(embed(out), np.load("baseline/summary.npy"))
    assert sim >= 0.86

# conftest.py に追記 — カセットの録画日を検査する
import time
from pathlib import Path
import pytest
 
CASSETTE_MAX_AGE_DAYS = 45
 
def pytest_sessionfinish(session, exitstatus):
    """カセットが古くなっていたら警告として出す（テストは落とさない）"""
    stale = []
    now = time.time()
    for c in Path("tests/cassettes").rglob("*.yaml"):
        age_days = (now - c.stat().st_mtime) / 86400
        if age_days > CASSETTE_MAX_AGE_DAYS:
            stale.append((c.name, int(age_days)))
    if stale:
        print("\n[cassette] 鮮度切れのカセット:")
        for name, days in sorted(stale, key=lambda x: -x[1]):
            print(f"  - {name}: {days}日前の録画（上限 {CASSETTE_MAX_AGE_DAYS}日）")

# .github/workflows/live-smoke.yml
name: live-smoke
on:
  schedule:
    - cron: "0 21 * * 0"   # 毎週月曜 6:00 JST
  workflow_dispatch:
 
jobs:
  smoke:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements-dev.txt
      - name: 構造テストのみ本物のAPIで実行
        env:
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
        run: pytest test_structure.py --record-mode=none --disable-recording -q

# field_drift.py
import json
import numpy as np
from pathlib import Path
from semantic_snapshot import embed, cosine
 
BASE_DIR = Path("baseline/fields")
 
def _to_text(value) -> str:
    """listやdictも安定した文字列に落として比較可能にする"""
    if isinstance(value, str):
        return value
    return json.dumps(value, ensure_ascii=False, sort_keys=True)
 
def field_drift(payload: dict, name: str) -> dict[str, float]:
    """フィールドごとのコサイン類似度を返す。baseline未作成なら作って空を返す"""
    out: dict[str, float] = {}
    for key, value in payload.items():
        path = BASE_DIR / name / f"{key}.npy"
        vec = embed(_to_text(value))
        if not path.exists():
            path.parent.mkdir(parents=True, exist_ok=True)
            np.save(path, vec)
            continue
        out[key] = cosine(vec, np.load(path))
    return out
 
def format_report(drift: dict[str, float], threshold: float) -> str:
    rows = sorted(drift.items(), key=lambda kv: kv[1])
    lines = [f"{'field':<16}{'cosine':>8}  status"]
    for key, sim in rows:
        mark = "DRIFT" if sim < threshold else "ok"
        lines.append(f"{key:<16}{sim:>8.3f}  {mark}")
    return "\n".join(lines)

# test_semantic_fields.py
import json
import pytest
from gemini_client import classify_sentiment
from field_drift import field_drift, format_report
 
THRESHOLD = 0.88
 
@pytest.mark.vcr
def test_sentiment_fields_stable():
    payload = json.loads(
        classify_sentiment("配送は遅かったが、サポートの対応は丁寧だった。")
    )
    drift = field_drift(payload, name="sentiment_review")
    if not drift:
        pytest.skip("フィールドbaselineを新規作成しました。")
 
    worst_key = min(drift, key=drift.get)
    assert drift[worst_key] >= THRESHOLD, (
        f"\n最も劣化したフィールド: {worst_key} (cosine={drift[worst_key]:.3f})\n"
        + format_report(drift, THRESHOLD)
    )

pytest --snapshot-update
git add -p __snapshots__/ baseline/    # 差分を1つずつ目視してから載せる
git commit -m "test: update snapshots for prompt schema change (intentional)"