예제 소스

diag_gemma_gap.py

#!/usr/bin/env python3
"""시험(val) 수 합법률과 실전 첫 시도 합법률 차이 진단.

  stats   모델 없이 실전 첫 시도 좌표 분석 (직전 자기 수 반복, 점유 칸 종류, 돌 수 분포)
  replay  같은 실전 국면을 로컬 모델로 다시 물어본다 (서버 경로 predict_move, 온도별)

  python scripts/diag_gemma_gap.py stats --set logs/gemma_move_loop_20261012/set2
  python scripts/diag_gemma_gap.py replay --set logs/gemma_move_loop_20261012/set2 \
      --lora-dir vlm_models/lora_gemma4_move_r1 --n 120 --out logs/gemma_gap/replay.json
"""

from __future__ import annotations

import argparse
import json
import random
import sqlite3
import sys
import time
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from scripts.dataset_coords_game import board_hash  # noqa: E402

N = 19
MODEL = "********"


def load_game_positions(set_dir: Path) -> list[dict]:
    """백(겜마) 차례 국면 + 서버 기록(첫 시도 좌표) 연결."""
    con = sqlite3.connect(set_dir / "matches.sqlite")
    rows = con.execute(
        "select game_id, ply, color, board, row, col, kind from moves where board is not null order by game_id, ply"
    ).fetchall()
    con.close()
    picks: dict[str, list[dict]] = {}
    for ln in (set_dir / "gemma4.jsonl").open(encoding="utf-8", errors="replace"):
        try:
            r = json.loads(ln)
        except json.JSONDecodeError:
            continue
        if r.get("event") == "move" and r.get("board_hash"):
            picks.setdefault(r["board_hash"], []).append(r)
    out = []
    last_own: dict = {}
    last_opp: dict = {}
    for gid, ply, color, text, row, col, kind in rows:
        if not text or len(text) != N * N:
            continue
        cells = bytes(".XO".index(ch) for ch in text)
        if color == "W":
            h = board_hash(cells, 2)
            rec = (picks.get(h) or [None])[0]
            out.append({"gid": gid, "ply": ply, "cells": cells, "rec": rec,
                        "last_own": last_own.get(gid), "last_opp": last_opp.get(gid)})
            if kind == "move" and row is not None:
                last_own[gid] = (int(row), int(col))
        elif kind == "move" and row is not None:
            last_opp[gid] = (int(row), int(col))
    return out


def cmd_stats(args) -> int:
    pos = load_game_positions(Path(args.set))
    joined = [p for p in pos if p["rec"] and p["rec"].get("vlm_seeds")]
    print(f"백 차례 {len(pos)}, 서버 기록 연결 {len(joined)}")
    kinds = Counter()
    stones_bins = Counter()
    first_rows = Counter()
    for p in joined:
        r, c = p["rec"]["vlm_seeds"][0]
        v = p["cells"][r * N + c]
        n = sum(1 for x in p["cells"] if x)
        b = f"{(n // 20) * 20:3d}-{(n // 20) * 20 + 19}"
        k = "empty" if v == 0 else ("own" if v == 2 else "opp")
        if (r, c) == p["last_own"]:
            k += "/직전내수"
        elif (r, c) == p["last_opp"]:
            k += "/직전상대수"
        kinds[k] += 1
        stones_bins[(b, v == 0)] += 1
        first_rows[r] += 1
    tot = len(joined) or 1
    print("첫 시도 좌표 종류:")
    for k, v in kinds.most_common():
        print(f"  {k:14s} {v:4d} ({100 * v / tot:.1f}%)")
    print("돌 수 구간별 첫 시도 빈칸 비율:")
    for b in sorted({b for b, _ in stones_bins}):
        e, o = stones_bins[(b, True)], stones_bins[(b, False)]
        print(f"  {b}: {e}/{e + o} = {100 * e / max(1, e + o):.1f}%   무작위 기준 {100 * (1 - int(b.split('-')[0]) / 361):.0f}%대")
    print("첫 시도 행 분포:", sorted(first_rows.items()))
    seeds_all = Counter(tuple(s) for p in joined for s in p["rec"]["vlm_seeds"])
    print("자주 고른 좌표 상위 15:", seeds_all.most_common(15))
    return 0


def cmd_replay(args) -> int:
    from vlm_server.server import _MlxVlmModel

    pos = [p for p in load_game_positions(Path(args.set)) if p["rec"] and p["rec"].get("vlm_seeds")]
    rng = random.Random(args.seed)
    rng.shuffle(pos)
    pos = pos[: args.n]
    temps = [float(t) for t in args.temps.split(",")]
    model = _MlxVlmModel(MODEL, max_new_tokens=args.max_new_tokens, temperature=temps[0],
                         lora_dir=args.lora_dir or None, text_only_moves=True, board_repr="aligned")
    model.start(wait_load=True, load_timeout_s=600.0)
    if model._load_error is not None:
        print(f"load failed: {model._load_error}")
        return 2
    results = []
    t0 = time.time()
    try:
        for i, p in enumerate(pos, 1):
            game_first = tuple(p["rec"]["vlm_seeds"][0])
            row = {"gid": p["gid"], "ply": p["ply"], "stones": sum(1 for x in p["cells"] if x),
                   "game_first": list(game_first),
                   "game_first_empty": p["cells"][game_first[0] * N + game_first[1]] == 0}
            for t in temps:
                model.temperature = t
                for k in range(args.repeats if t > 0 else 1):
                    r, c, is_pass = model.predict_move(None, p["cells"], 2, timeout_s=180)
                    empty = (not is_pass and r is not None and c is not None
                             and p["cells"][int(r) * N + int(c)] == 0)
                    row.setdefault(f"t{t}", []).append(
                        {"pick": None if r is None else [int(r), int(c)], "pass": bool(is_pass), "empty": bool(empty),
                         "same_as_game": r is not None and (int(r), int(c)) == game_first})
            results.append(row)
            if i % 10 == 0:
                print(f"[replay] {i}/{len(pos)} {json.dumps(summarize(results, temps), ensure_ascii=False)}", flush=True)
    finally:
        model.stop()
    summ = summarize(results, temps)
    summ.update(n=len(results), lora=args.lora_dir, set=args.set, sec=round(time.time() - t0))
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps({"summary": summ, "results": results}, ensure_ascii=False, indent=1) + "\n",
                   encoding="utf-8")
    print(json.dumps(summ, ensure_ascii=False))
    return 0


def summarize(results: list[dict], temps: list[float]) -> dict:
    n = len(results) or 1
    s = {"game_first_empty": round(100 * sum(r["game_first_empty"] for r in results) / n, 1)}
    for t in temps:
        picks = [x for r in results for x in r.get(f"t{t}", [])]
        m = len(picks) or 1
        s[f"t{t}_empty"] = round(100 * sum(x["empty"] for x in picks) / m, 1)
        s[f"t{t}_same_as_game"] = round(100 * sum(x["same_as_game"] for x in picks) / m, 1)
    return s


def main() -> int:
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd", required=True)
    a = sub.add_parser("stats")
    a.add_argument("--set", required=True)
    b = sub.add_parser("replay")
    b.add_argument("--set", required=True)
    b.add_argument("--lora-dir", default="vlm_models/lora_gemma4_move_r1")
    b.add_argument("--n", type=int, default=120)
    b.add_argument("--temps", default="0.0,0.2")
    b.add_argument("--repeats", type=int, default=1)
    b.add_argument("--max-new-tokens", type=int, default=64)
    b.add_argument("--seed", type=int, default=7)
    b.add_argument("--out", required=True)
    args = ap.parse_args()
    return cmd_stats(args) if args.cmd == "stats" else cmd_replay(args)


if __name__ == "__main__":
    raise SystemExit(main())