예제 소스

eval_ladder_vlm.py

#!/usr/bin/env python3
"""축·호구 검증셋으로 VLM(LoRA) 수 예측을 채점.

  python scripts/eval_ladder_vlm.py --lora-dir vlm_models/lora_mix_qwen_r5 --limit 120 \
      --out logs/ladder/eval_r5.json
  python scripts/eval_ladder_vlm.py --data sft_data/hogu_v1/val.jsonl --lora-dir ... --out logs/hogu/eval.json

지표 (종류별)
  축  flee_dead    escape_rate   : 잡히는 축에서 도망친 비율 (낮을수록 좋음)
      chase_broken chase_rate    : 깨진 축을 쫓은 비율 (낮을수록 좋음)
  호구 peep/entry/fill on_point_rate : 호구점(메움·진입·들여다보기 잇기)에 둔 비율 (낮을수록 좋음)
  전체 katago_match : KataGo 최선과 같은 수 비율, no_move : 파싱 실패·패스
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from collections import defaultdict
from pathlib import Path

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

from vlm_server.server import _MlxVlmModel, parse_move  # noqa: E402

DEFAULT_MODEL = "********"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--model", default=DEFAULT_MODEL)
    ap.add_argument("--lora-dir", default=None)
    ap.add_argument("--data", default="sft_data/ladder_v1/val.jsonl")
    ap.add_argument("--limit", type=int, default=150)
    ap.add_argument("--out", default=None)
    a = ap.parse_args()

    rows = [json.loads(l) for l in open(ROOT / a.data, encoding="utf-8") if l.strip()][: a.limit]
    m = _MlxVlmModel(a.model, max_new_tokens=32, temperature=0.0, lora_dir=a.lora_dir)
    t0 = time.time()
    m.start(wait_load=True)
    print(f"loaded in {time.time() - t0:.0f}s  lora={a.lora_dir}", flush=True)

    agg: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
    details = []
    for i, row in enumerate(rows):
        meta = row.get("ladder") or row.get("hogu") or row["onpolicy"]
        is_hogu = "hogu" in row
        is_onpolicy = "onpolicy" in row
        kind = meta["kind"]
        text = row["messages"][0]["content"][0]["text"]
        system, _, user = text.partition("\n\n")
        try:
            out = m.predict_raw(system, user, max_tokens=32, temperature=0.0, timeout_s=180)
        except Exception as e:  # noqa: BLE001
            out = f"ERR {e}"
        r, c, is_pass = parse_move(out)
        pred = None if (is_pass or r is None or c is None) else [r, c]
        target = [row["target_row"], row["target_col"]]
        g = agg[kind]
        g["n"] += 1
        g["katago_match"] += int(pred == target)
        g["no_move"] += int(pred is None)
        lm = meta.get("ladder_move")
        if is_onpolicy:
            g["top3"] += int(pred is not None and pred in meta.get("top", []))
        elif is_hogu:
            g["on_point"] += int(pred is not None and pred in meta["points"])
        elif kind.startswith("flee"):
            g["escape"] += int(pred is not None and pred == lm)
        else:
            g["chase"] += int(pred is not None and pred in meta.get("chase_points", []))
        details.append({"i": i, "kind": kind, "pred": pred, "target": target, "ladder_move": lm, "raw": out[:60]})
        if (i + 1) % 20 == 0:
            print(f"  {i + 1}/{len(rows)} {time.time() - t0:.0f}s", flush=True)

    summary: dict = {"lora": a.lora_dir, "n": len(rows)}
    for kind, g in sorted(agg.items()):
        n = max(1, g["n"])
        s = {"n": g["n"], "katago_match": round(g["katago_match"] / n, 3), "no_move": round(g["no_move"] / n, 3)}
        if kind.startswith("onpolicy"):
            s["top3_rate"] = round(g["top3"] / n, 3)
        elif kind in ("peep", "entry", "fill"):
            s["on_point_rate"] = round(g["on_point"] / n, 3)
        elif kind.startswith("flee"):
            s["escape_rate"] = round(g["escape"] / n, 3)
        else:
            s["chase_rate"] = round(g["chase"] / n, 3)
        summary[kind] = s
    print(json.dumps(summary, ensure_ascii=False, indent=2))
    if a.out:
        p = ROOT / a.out
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps({"summary": summary, "details": details}, ensure_ascii=False, indent=2), encoding="utf-8")
    m.stop()
    return 0


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