예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
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())