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}"
    )

# 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

# 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