예제 소스

eval_coords_game.py

#!/usr/bin/env python3
"""실전 국면 좌표 평가 (dataset_coords_game.py 의 val.jsonl).

  lookup    칸 기호 정답률 (hard = 모델이 실제로 이미 돌 있는 칸을 골랐던 좌표)
  row/col   한 줄 돌 위치 완전 일치율
  move*     빈 점·합법·금지 좌표 아님 비율 (legal), 휴리스틱 수와 일치율 (match)

  python scripts/eval_coords_game.py --data sft_data/gemma_coord_r1/val.jsonl \
      --model <gemma> --lora-dir vlm_models/lora_gemma4_heuristic_r3 --out logs/x/eval_r3.json
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import re
import sys
import time
from pathlib import Path

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

from vlm_server.dataset_baduk import row_line_indexed  # noqa: E402
from vlm_server.server import _MlxVlmModel, parse_move  # noqa: E402
from vlm_server.verify_coords import _parse_cell  # noqa: E402

N = 19


def _monitor():
    path = ROOT / "scripts" / "monitor_mariago_web_play.py"
    spec = importlib.util.spec_from_file_location("monitor_mariago_web_play", path)
    mod = importlib.util.module_from_spec(spec)
    sys.modules["monitor_mariago_web_play"] = mod
    spec.loader.exec_module(mod)
    return mod


mon = _monitor()


def parse_scan(text: str) -> dict | None:
    """마지막 완전한 {"X","O"} 객체를 읽는다.

    분석 채널과 본문에 JSON이 두 번 나오면 첫 `{`부터 마지막 `}`까지는
    한 객체가 아니므로, 중괄호가 중첩되지 않은 객체만 순서대로 파싱한다.
    """
    found = None
    for m in re.finditer(r"\{[^{}]*\}", text or ""):
        try:
            d = json.loads(m.group(0))
        except (ValueError, TypeError):
            continue
        if not isinstance(d, dict) or ("X" not in d and "O" not in d):
            continue
        try:
            found = {"X": sorted(int(x) for x in d.get("X", [])), "O": sorted(int(x) for x in d.get("O", []))}
        except (ValueError, TypeError):
            continue
    return found


def score(rec: dict, raw: str) -> dict:
    task, exp = rec["task"], rec["expect"]
    if task == "lookup":
        got = _parse_cell(raw)
        return {"ok": got == exp["cell"], "got": got, "hard": bool(exp.get("hard"))}
    if task in ("row", "col"):
        got = parse_scan(raw)
        return {"ok": got == {"X": sorted(exp["X"]), "O": sorted(exp["O"])}, "got": got}
    r, c, is_pass = parse_move(raw or "")
    board = [".XO".index(ch) for ch in rec["board"]]
    if is_pass or r is None or c is None:
        return {"ok": False, "legal": False, "match": False, "got": "pass" if is_pass else None, "why": "pass/none"}
    forbidden = {tuple(p) for p in exp.get("forbidden") or []}
    ok, why, *_ = mon.try_play(board, int(r), int(c), int(exp["side"]), None)
    legal = ok and (int(r), int(c)) not in forbidden
    if ok and not legal:
        why = "forbidden"
    match = (int(r), int(c)) == (exp["row"], exp["col"])
    v = board[int(r) * N + int(c)]
    kind = "empty" if v == 0 else ("own" if v == int(exp["side"]) else "opp")
    res = {"ok": legal, "legal": legal, "match": match, "got": [r, c], "why": why, "kind": kind}
    if exp.get("top"):
        res["top3"] = [int(r), int(c)] in [list(p) for p in exp["top"]]
    m = re.search(r'"line"\s*:\s*"([^"]*)"', raw or "")
    if m:
        line = m.group(1).strip()
        res["line_ok"] = line == row_line_indexed(bytes(board), int(r))
        cells = dict((t[:2], t[2:]) for t in line.split() if len(t) >= 3)
        res["consistent"] = cells.get(f"{int(c):02d}") == "."
    em = re.search(r'"empty"\s*:\s*(\[[^\]]*\])', raw or "")
    if em:
        try:
            got_empty = sorted(int(x) for x in json.loads(em.group(1)))
        except (ValueError, TypeError):
            got_empty = None
        true_empty = sorted(i for i in range(N) if board[int(r) * N + i] == 0)
        res["empty_ok"] = got_empty == true_empty
    return res


def summarize(results: list[dict]) -> dict:
    out: dict = {"n": len(results)}
    for task in ("lookup", "row", "col", "move", "move_retry", "kmove", "kmove_retry"):
        rs = [x for x in results if x["task"] == task]
        if not rs:
            continue
        d = {"n": len(rs), "acc": round(100 * sum(x["ok"] for x in rs) / len(rs), 1)}
        if task == "lookup":
            hard = [x for x in rs if x.get("hard")]
            if hard:
                d["hard_n"] = len(hard)
                d["hard_acc"] = round(100 * sum(x["ok"] for x in hard) / len(hard), 1)
        if "move" in task:
            d["match"] = round(100 * sum(x["match"] for x in rs) / len(rs), 1)
            if any("top3" in x for x in rs):
                d["top3"] = round(100 * sum(bool(x.get("top3")) for x in rs) / len(rs), 1)
            whys: dict[str, int] = {}
            for x in rs:
                if not x["ok"]:
                    whys[x.get("why") or "?"] = whys.get(x.get("why") or "?", 0) + 1
            d["fail_why"] = whys
        out[task] = d
    moves = [x for x in results if "move" in x["task"]]
    if moves:
        out["move_legal"] = round(100 * sum(x["ok"] for x in moves) / len(moves), 1)
        out["move_empty"] = round(100 * sum(x.get("kind") == "empty" for x in moves) / len(moves), 1)
        out["move_own"] = round(100 * sum(x.get("kind") == "own" for x in moves) / len(moves), 1)
        out["move_distinct"] = len({tuple(x["got"]) for x in moves if isinstance(x.get("got"), list)})
        lines = [x for x in moves if "line_ok" in x]
        if lines:
            out["line_ok"] = round(100 * sum(x["line_ok"] for x in lines) / len(lines), 1)
            out["consistent"] = round(100 * sum(x["consistent"] for x in lines) / len(lines), 1)
        empties = [x for x in moves if "empty_ok" in x]
        if empties:
            out["empty_ok"] = round(100 * sum(x["empty_ok"] for x in empties) / len(empties), 1)
    kmoves = [x for x in results if x["task"].startswith("kmove")]
    if kmoves:
        out["kata_top1"] = round(100 * sum(x["match"] for x in kmoves) / len(kmoves), 1)
        out["kata_top3"] = round(100 * sum(bool(x.get("top3")) for x in kmoves) / len(kmoves), 1)
    look = [x for x in results if x["task"] in ("lookup", "row", "col")]
    if look:
        out["read_acc"] = round(100 * sum(x["ok"] for x in look) / len(look), 1)
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data", required=True)
    ap.add_argument("--model", required=True)
    ap.add_argument("--lora-dir", default=None)
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--out", required=True)
    args = ap.parse_args()

    recs = [json.loads(x) for x in open(args.data, encoding="utf-8") if x.strip()]
    if args.limit:
        recs = recs[: args.limit]
    model = _MlxVlmModel(args.model, max_new_tokens=80, temperature=0.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"[eval-coord] load failed: {model._load_error}", flush=True)
        return 2
    results = []
    t0 = time.time()
    try:
        for i, rec in enumerate(recs, 1):
            try:
                raw = model.predict_raw(rec["sys"], rec["usr"], num_images=0, timeout_s=180, temperature=0.0)
            except Exception as exc:  # noqa: BLE001
                raw = f"ERR {exc}"
            res = score(rec, raw)
            res.update(task=rec["task"], raw=(raw or "")[:200])
            results.append(res)
            if i % 20 == 0:
                print(f"[eval-coord] {i}/{len(recs)} {summarize(results)}", flush=True)
    finally:
        model.stop()
    summary = summarize(results)
    summary.update(lora=args.lora_dir, data=args.data, sec=round(time.time() - t0))
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps({"summary": summary, "results": results}, ensure_ascii=False, indent=1) + "\n",
                   encoding="utf-8")
    print(f"[eval-coord] {json.dumps(summary, ensure_ascii=False)}", flush=True)
    return 0


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