Gemini 3.5 Flash GA をどこから差し替えるか — ワークロード別の置き換え判定とモデルルーターによる段階導入の記録

# compare_models.py — 本番の代表タスクで新旧モデルを実測比較する
import json
import os
import statistics
import time
 
from google import genai
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
MODELS = ["gemini-2.5-flash", "gemini-3.5-flash"]
 
# 本番パイプラインから抜き出した代表タスク
# 各ケースは {"prompt": "...", "required_keys": ["title", "tags"]} の形式
with open("eval_cases.json", encoding="utf-8") as f:
    CASES = json.load(f)
 
 
def run_case(model: str, case: dict) -> dict:
    started = time.perf_counter()
    response = client.models.generate_content(
        model=model,
        contents=case["prompt"],
        config={
            "response_mime_type": "application/json",
            "temperature": 0.2,
        },
    )
    elapsed = time.perf_counter() - started
    usage = response.usage_metadata
    return {
        "latency": elapsed,
        "input_tokens": usage.prompt_token_count,
        "output_tokens": usage.candidates_token_count,
        "text": response.text,
    }
 
 
def is_valid(case: dict, text: str) -> bool:
    # 後段パーサーが要求する形式を、そのまま合否条件にする
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        return False
    return all(key in data for key in case["required_keys"])
 
 
for model in MODELS:
    latencies = []
    valid = 0
    in_tokens = 0
    out_tokens = 0
    for case in CASES:
        result = run_case(model, case)
        latencies.append(result["latency"])
        in_tokens += result["input_tokens"]
        out_tokens += result["output_tokens"]
        if is_valid(case, result["text"]):
            valid += 1
    print(
        f"{model}: p50={statistics.median(latencies):.2f}s "
        f"valid={valid}/{len(CASES)} "
        f"tokens(in/out)={in_tokens}/{out_tokens}"
    )

# decide.py — 実測サマリからワークロードの採否を機械的に決める
from dataclasses import dataclass
 
 
@dataclass
class Measured:
    workload: str
    valid_rate_old: float   # 0.0 - 1.0
    valid_rate_new: float
    p50_old: float          # 秒
    p50_new: float
    out_tokens_old: int
    out_tokens_new: int
    voice_visible: bool     # モデルの「声」がユーザーに直接見えるか
 
 
def decide(m: Measured) -> str:
    # 声が見えるワークロードは、数値が良くても自動 GO にしない
    if m.voice_visible:
        return "HOLD"
 
    valid_delta = (m.valid_rate_new - m.valid_rate_old) * 100        # ポイント差
    latency_delta = (m.p50_new - m.p50_old) / m.p50_old              # 比率
    token_delta = (m.out_tokens_new - m.out_tokens_old) / m.out_tokens_old
 
    if valid_delta < -2 or latency_delta > 0.10 or token_delta > 0.30:
        return "HOLD"
    if valid_delta < -0.5 or latency_delta > 0 or token_delta > 0.15:
        return "CANARY"
    return "GO"
 
 
if __name__ == "__main__":
    batch = Measured(
        workload="categorize",
        valid_rate_old=57 / 60,
        valid_rate_new=58 / 60,
        p50_old=1.82,
        p50_new=1.41,
        out_tokens_old=15890,
        out_tokens_new=17820,
        voice_visible=False,
    )
    print(batch.workload, decide(batch))  # -> categorize GO

# model_router.py — 環境変数で即時ロールバックできるモデルルーター
import os
import random
 
DEFAULT_ROUTES = {
    "categorize": "gemini-3.5-flash",    # 分類バッチ: 移行済み
    "metadata": "gemini-3.5-flash",      # メタデータ生成: カナリア 50% で試験中
    "review_reply": "gemini-2.5-flash",  # レビュー返信: 意図的に保留
    "agentic": "gemini-3.5-flash",       # 多段タスク: 移行済み
}
 
# 一部のトラフィックだけ新モデルへ流す比率(未指定は 100%)
CANARY_RATIO = {
    "metadata": 0.5,
}
 
FALLBACK_MODEL = "gemini-2.5-flash"
 
 
def resolve_model(workload: str) -> str:
    # 環境変数 MODEL_OVERRIDE_<WORKLOAD> が最優先(緊急ロールバック用)
    override = os.environ.get(f"MODEL_OVERRIDE_{workload.upper()}")
    if override:
        return override
    model = DEFAULT_ROUTES.get(workload, FALLBACK_MODEL)
    ratio = CANARY_RATIO.get(workload)
    if ratio is not None and random.random() >= ratio:
        return FALLBACK_MODEL
    return model

# schema_drift.py — 合否ではなく「想定外のキー」を記録する
import json
import logging
from collections import Counter
 
logger = logging.getLogger("schema_drift")
_unexpected: Counter = Counter()
 
 
def observe(workload: str, model: str, expected_keys: set[str], text: str) -> None:
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        logger.warning("drift/parse_error workload=%s model=%s", workload, model)
        return
    if not isinstance(data, dict):
        return
 
    extra = set(data) - expected_keys
    missing = expected_keys - set(data)
    for key in extra:
        _unexpected[(workload, model, key)] += 1
    if extra or missing:
        logger.info(
            "drift workload=%s model=%s extra=%s missing=%s",
            workload, model, sorted(extra), sorted(missing),
        )
 
 
def top_drift(limit: int = 10) -> list:
    return _unexpected.most_common(limit)

# generate_with_fallback — 主モデル失敗時に旧世代へ自動退避する
from model_router import FALLBACK_MODEL, resolve_model
 
 
def generate_with_fallback(client, workload, contents, config):
    primary = resolve_model(workload)
    last_error = None
    for model in (primary, FALLBACK_MODEL):
        try:
            return client.models.generate_content(
                model=model, contents=contents, config=config
            )
        except Exception as err:  # 本番では google.genai.errors の型で個別に捕捉する
            last_error = err
    raise last_error