Gemma 4 と Nemotron 3 Nano Omni を、総合スコアではなく自分の画像100枚で採否判定する

# build_eval_set.py — 層化サンプリングで評価セットを作る
import json
import random
from collections import defaultdict
from pathlib import Path
 
random.seed(42)  # 再現性のため必ず固定する
 
CATALOG = Path("data/wallpapers.jsonl")   # {"id":..., "path":..., "label":...}
BOUNDARY = Path("data/boundary_ids.txt")  # 過去に指摘・迷いのあった画像ID
OUT = Path("data/eval_set.jsonl")
 
PER_CATEGORY = 3
BOUNDARY_QUOTA = 10
 
 
def load_catalog(path: Path) -> list[dict]:
    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
 
 
def stratified_sample(rows: list[dict], per_category: int) -> list[dict]:
    buckets: dict[str, list[dict]] = defaultdict(list)
    for row in rows:
        buckets[row["label"]].append(row)
 
    picked: list[dict] = []
    for label, items in sorted(buckets.items()):
        if len(items) < per_category:
            # 母数が足りないカテゴリこそ壊れやすい。黙って減らさず気づけるようにする
            print(f"[warn] {label}: only {len(items)} images (< {per_category})")
        picked.extend(random.sample(items, min(per_category, len(items))))
    return picked
 
 
def main() -> None:
    rows = load_catalog(CATALOG)
    boundary_ids = set(BOUNDARY.read_text().split())
 
    base = [r for r in rows if r["id"] not in boundary_ids]
    sampled = stratified_sample(base, PER_CATEGORY)
 
    boundary_rows = [r for r in rows if r["id"] in boundary_ids]
    sampled += random.sample(boundary_rows, min(BOUNDARY_QUOTA, len(boundary_rows)))
 
    with OUT.open("w") as f:
        for row in sampled:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
 
    print(f"eval set: {len(sampled)} images / {len(set(r['label'] for r in sampled))} categories")
 
 
if __name__ == "__main__":
    main()

# predictors.py — モデルごとの差し替え口を1つに揃える
import base64
import json
import os
import re
from pathlib import Path
 
import requests
 
LABELS = json.loads(Path("config/labels.json").read_text())  # 30カテゴリの正式名
PROMPT = (
    "次の画像を、以下のカテゴリのうち1つに分類してください。\n"
    "カテゴリ名以外の語・説明・記号は一切出力しないでください。\n\n"
    + "\n".join(f"- {label}" for label in LABELS)
)
 
 
def _encode(path: str) -> str:
    return base64.b64encode(Path(path).read_bytes()).decode()
 
 
def _normalize(raw: str) -> str:
    """生出力を正式カテゴリ名へ落とす。落とせなければ unknown を返す"""
    text = re.sub(r"[\s。、.:：\-*`]", "", raw)
    for label in LABELS:
        if label in text:
            return label
    return "unknown"  # 近いラベルへ寄せて救わない
 
 
def local_predict(path: str, model: str = "gemma-4-mm") -> str:
    """llama.cpp server / Ollama など OpenAI 互換エンドポイント向け"""
    res = requests.post(
        "http://127.0.0.1:8080/v1/chat/completions",
        json={
            "model": model,
            "temperature": 0,
            "max_tokens": 16,
            "messages": [{
                "role": "user",
                "content": [
                    {"type": "text", "text": PROMPT},
                    {"type": "image_url",
                     "image_url": {"url": f"data:image/jpeg;base64,{_encode(path)}"}},
                ],
            }],
        },
        timeout=120,
    )
    res.raise_for_status()
    return _normalize(res.json()["choices"][0]["message"]["content"])
 
 
def cloud_predict(path: str, model: str = "gemini-3.5-flash") -> str:
    res = requests.post(
        f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent",
        headers={"x-goog-api-key": os.environ["GEMINI_API_KEY"]},
        json={
            "contents": [{
                "parts": [
                    {"text": PROMPT},
                    {"inline_data": {"mime_type": "image/jpeg", "data": _encode(path)}},
                ]
            }],
            "generationConfig": {"temperature": 0, "maxOutputTokens": 16},
        },
        timeout=60,
    )
    res.raise_for_status()
    return _normalize(res.json()["candidates"][0]["content"]["parts"][0]["text"])

# evaluate.py — 予測を集めて一致率と混同行列を出す
import json
from collections import Counter, defaultdict
from pathlib import Path
from typing import Callable
 
EVAL_SET = Path("data/eval_set.jsonl")
 
 
def evaluate(predict: Callable[[str], str], name: str) -> dict:
    rows = [json.loads(line) for line in EVAL_SET.read_text().splitlines()]
 
    confusion: dict[tuple[str, str], int] = Counter()
    per_label_total: dict[str, int] = Counter()
    hits = 0
 
    for row in rows:
        truth = row["label"]
        pred = predict(row["path"])
        confusion[(truth, pred)] += 1
        per_label_total[truth] += 1
        if pred == truth:
            hits += 1
 
    accuracy = hits / len(rows)
 
    # 誤りだけを件数順に並べる。ここが読みどころ
    errors = sorted(
        ((t, p, n) for (t, p), n in confusion.items() if t != p),
        key=lambda x: x[2],
        reverse=True,
    )
 
    print(f"\n=== {name} ===")
    print(f"accuracy: {accuracy:.2%} ({hits}/{len(rows)})")
    print("top confusions:")
    for truth, pred, n in errors[:5]:
        share = n / per_label_total[truth]
        print(f"  {truth} -> {pred}: {n} ({share:.0%} of {truth})")
 
    # キーはタプルのままだと JSON に落ちないので、この時点で文字列へ寄せておく
    return {
        "name": name,
        "accuracy": accuracy,
        "confusion": {f"{t}\t{p}": n for (t, p), n in confusion.items()},
    }

# cost_weighted_error.py — 誤りのコストで重み付けした誤り率
import json
from pathlib import Path
 
# {(true, pred): cost}。書いていない組み合わせは既定値を使う
COST_MATRIX_PATH = Path("config/cost_matrix.json")
DEFAULT_COST = 1.0
 
 
def load_cost_matrix() -> dict[tuple[str, str], float]:
    raw = json.loads(COST_MATRIX_PATH.read_text())
    return {(item["true"], item["pred"]): item["cost"] for item in raw}
 
 
def cost_weighted_error(confusion: dict, costs: dict) -> float:
    """confusion: {(true, pred): count} を受け取り、加重誤り率を返す"""
    total = sum(confusion.values())
    if total == 0:
        return 0.0
 
    penalty = 0.0
    for (truth, pred), n in confusion.items():
        if truth == pred:
            continue
        penalty += costs.get((truth, pred), DEFAULT_COST) * n
 
    return penalty / total
 
 
def report(result: dict) -> None:
    costs = load_cost_matrix()
    # evaluate.py の返り値は JSON 化のためキーが文字列になっている前提で戻す
    confusion = {tuple(k.split("\t")): v for k, v in result["confusion"].items()}
    cwe = cost_weighted_error(confusion, costs)
 
    print(f"{result['name']}: accuracy={result['accuracy']:.2%} / cost-weighted error={cwe:.3f}")

# regression_check.py — ベースラインとの差でCIを落とす
import json
import sys
from pathlib import Path
 
BASELINE = Path("data/baseline.json")   # 採用時に確定させた基準値
CURRENT = Path("data/current.json")     # 今回の評価結果
 
# 一致率が上がっていても、加重誤りがこの幅を超えて悪化したら通さない
CWE_TOLERANCE = 0.03
 
 
def main() -> int:
    base = json.loads(BASELINE.read_text())
    now = json.loads(CURRENT.read_text())
 
    delta_acc = now["accuracy"] - base["accuracy"]
    delta_cwe = now["cost_weighted_error"] - base["cost_weighted_error"]
 
    print(f"accuracy: {base['accuracy']:.2%} -> {now['accuracy']:.2%} ({delta_acc:+.2%})")
    print(f"cost-weighted error: {base['cost_weighted_error']:.3f} -> "
          f"{now['cost_weighted_error']:.3f} ({delta_cwe:+.3f})")
 
    if delta_cwe > CWE_TOLERANCE:
        print(f"\nFAIL: 加重誤りが許容幅 {CWE_TOLERANCE} を超えて悪化しています。"
              f"一致率が改善していても採用しません。")
        return 1
 
    print("\nPASS")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())

# baseline_stamp.py — 測定結果に「何で測ったか」を焼き込む
import json
import os
from datetime import datetime, timezone
from pathlib import Path
 
import requests
 
BASELINE = Path("data/baseline.json")
 
 
def resolve_model_version(model: str) -> str:
    """最小の呼び出しを1回だけ投げ、実際に応答した版を取り出す"""
    res = requests.post(
        f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent",
        headers={"x-goog-api-key": os.environ["GEMINI_API_KEY"]},
        json={
            "contents": [{"parts": [{"text": "ping"}]}],
            "generationConfig": {"maxOutputTokens": 1},
        },
        timeout=30,
    )
    res.raise_for_status()
    # 別名で呼んでも、応答したのは実体の版。記録したいのはこちら
    return res.json().get("modelVersion", "unknown")
 
 
def stamp(result: dict, model: str) -> None:
    stamped = dict(result)
    stamped["model_alias"] = model
    stamped["model_version"] = resolve_model_version(model)
    stamped["measured_at"] = datetime.now(timezone.utc).isoformat()
    BASELINE.write_text(json.dumps(stamped, ensure_ascii=False, indent=2))
    print(f"baseline stamped: {stamped['model_alias']} -> {stamped['model_version']}")
 
 
def assert_same_ground(model: str) -> None:
    """回帰チェックの前に呼ぶ。土台が動いていたら比較させずに落とす"""
    base = json.loads(BASELINE.read_text())
    recorded = base.get("model_version")
    now = resolve_model_version(model)
    if now != recorded:
        raise SystemExit(
            f"baseline was measured on {recorded}, but {model} now resolves to {now}. "
            "同じ100枚で測り直してから比較してください。"
        )

# interval.py — 一致率の95%信頼区間（Wilson）と、必要枚数の逆算
from math import sqrt
 
Z = 1.96  # 95%
 
 
def wilson(hits: int, n: int, z: float = Z) -> tuple[float, float]:
    if n == 0:
        return (0.0, 0.0)
    p = hits / n
    denom = 1 + z * z / n
    center = (p + z * z / (2 * n)) / denom
    margin = z * sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / denom
    return (center - margin, center + margin)
 
 
def required_n(p: float, half_width: float, z: float = Z) -> int:
    """半幅 half_width（0.03 なら ±3ポイント）を得るのに要るおおよその枚数"""
    return int((z * z * p * (1 - p)) / (half_width * half_width)) + 1
 
 
if __name__ == "__main__":
    lo, hi = wilson(82, 100)
    print(f"82/100 -> 95% CI [{lo:.1%}, {hi:.1%}]")
    for hw in (0.05, 0.03, 0.02):
        print(f"±{hw:.0%} で測るには約 {required_n(0.82, hw)} 枚")