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()

# 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})")
 
    return {"name": name, "accuracy": accuracy, "confusion": dict(confusion)}

# 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())