예제 소스

play_heuristic_vs_model.py

#!/usr/bin/env python3
"""휴리스틱(흑)과 모델(백)의 대국. 모델 서버의 수를 그대로 둔다."""

from __future__ import annotations

import argparse
import importlib.util
import random
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

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

SAFETY_PLIES = 1000
MOVE_TIMEOUT_S = 360.0


def _load(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"import failed: {path}")
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


mon = _load("monitor_mariago_web_play", ROOT / "scripts" / "monitor_mariago_web_play.py")
bridge = _load("mariago_ws_bridge", ROOT / "scripts" / "mariago_ws_bridge.py")
mdb = _load("match_db", ROOT / "scripts" / "match_db.py")

from rl_env.board import estimate_area_score  # noqa: E402
from vlm_server.cross_lessons import absorb_game, load_book, save_book  # noqa: E402


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def model_turn(port: int, board: list[int], epoch: int, ko):
    rejects: list[tuple[int, int]] = []
    last = ""
    for _ in range(2):
        resp = bridge.request_move_sync("127.0.0.1", port, board, mon.WHITE, epoch, rejects, MOVE_TIMEOUT_S)
        if resp.get("type") == "error":
            last = str(resp.get("error") or "error")
            continue
        if resp.get("pass"):
            return "pass", None, None, board, ko, "pass"
        row, col = int(resp["row"]), int(resp["col"])
        ok, reason, nxt, _cap, new_ko = mon.try_play(board, row, col, mon.WHITE, ko)
        if ok:
            return "move", row, col, nxt, new_ko, "ok"
        last = reason
        rejects.append((row, col))
    return "illegal", None, None, board, ko, last


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, required=True)
    ap.add_argument("--model-name", default="gemma4")
    ap.add_argument("--games", type=int, default=3)
    ap.add_argument("--start-id", type=int, default=1)
    ap.add_argument("--max-moves", type=int, default=0)
    ap.add_argument("--out-dir", default="logs/heuristic_gemma4")
    ap.add_argument("--seed", type=int, default=19)
    args = ap.parse_args()

    cap = SAFETY_PLIES if args.max_moves <= 0 else args.max_moves
    out = ROOT / args.out_dir
    out.mkdir(parents=True, exist_ok=True)
    rng = random.Random(args.seed)
    db = mdb.MatchDB(out / "matches.sqlite")
    db.start_run(
        now(),
        {
            "black": "heuristic",
            "white": args.model_name,
            "port": args.port,
            "games": args.games,
            "start_id": args.start_id,
            "max_moves": args.max_moves,
        },
        "heuristic_vs_" + args.model_name,
        None,
    )
    print(f"[match] heuristic(B) vs {args.model_name}(W:{args.port}) games={args.games} cap={cap}", flush=True)
    judge = None
    try:
        judge = mdb.FinalJudge(visits=64)
    except Exception as exc:
        print(f"[match] katago judge off: {exc}", flush=True)

    for game_id in range(args.start_id, args.start_id + args.games):
        board = mon.empty_board()
        ko = None
        passes = 0
        fails = 0
        issues: list[str] = []
        lesson_lines: list[str] = []
        ply = 0
        color_turn = 0
        end = "max_plies"
        db.start_game(game_id, game_id, now())
        print(f"[match] game {game_id} start", flush=True)
        while ply < cap:
            side_name = "B" if color_turn % 2 == 0 else "W"
            side = mon.BLACK if side_name == "B" else mon.WHITE
            t0 = time.time()
            if side_name == "B":
                row, col, is_pass = mon.heuristic_move(board, side, ko, rng)
                if is_pass:
                    kind, row, col, nxt, new_ko, detail = "pass", None, None, board, ko, "pass"
                else:
                    ok, reason, nxt, _cap, new_ko = mon.try_play(board, int(row), int(col), side, ko)
                    if ok:
                        kind, detail = "move", "ok"
                    else:
                        kind, nxt, new_ko, detail = "illegal", board, ko, reason
            else:
                kind, row, col, nxt, new_ko, detail = model_turn(args.port, board, game_id, ko)
            dt = time.time() - t0
            ko = new_ko
            db.add_move(
                game_id=game_id,
                ply=ply,
                color=side_name,
                kind=kind,
                board=board,
                ts=now(),
                row=row,
                col=col,
                dt=dt,
                attempt="first",
                server_rec={"model": "heuristic" if side_name == "B" else args.model_name, "detail": detail},
            )
            ply += 1
            if kind in ("move", "pass"):
                spot = "PASS" if kind == "pass" else f"P ({row},{col})"
                lesson_lines.append(f"{ply}. {side_name} {spot}")
            if kind == "illegal":
                fails += 1
                issues.append(f"{side_name}_illegal:{detail}")
                print(f"[match] g{game_id} ply {ply} {side_name} illegal {detail}", flush=True)
                if fails >= 3:
                    end = "error"
                    break
                continue
            fails = 0
            color_turn += 1
            if kind == "pass":
                passes += 1
                print(f"[match] g{game_id} ply {ply} {side_name} pass ({dt:.0f}s)", flush=True)
                if passes >= 2:
                    end = "double_pass"
                    break
                continue
            passes = 0
            board = nxt
            if ply % 20 == 0:
                print(f"[match] g{game_id} ply {ply} {side_name} ({row},{col}) {dt:.0f}s", flush=True)

        b_area, w_area, _dame = estimate_area_score(bytes(int(v) & 0xFF for v in board), 19)
        lead = None
        if judge is not None and end != "error":
            try:
                lead = judge.lead_black(board)
            except Exception as exc:
                print(f"[match] judge failed: {exc}", flush=True)
        if lead is None:
            margin = b_area - (w_area + 6.5)
            winner = "black" if margin > 0 else "white" if margin < 0 else "draw"
        else:
            winner = "black" if lead > 0 else "white" if lead < 0 else "draw"
        score = {"winner": winner, "black_score": b_area, "white_score": w_area + 6.5, "katago_lead_b": lead}
        db.finish_game(game_id, ts=now(), end_reason=end, plies=ply, score=score, issues=issues, katago_lead_b=lead)
        if end != "error" and winner in ("black", "white"):
            book = load_book(force=True)
            key = f"{out.name}:{game_id}"
            if absorb_game(book, lesson_lines, winner, key):
                save_book(book)
                print(
                    f"[match] lessons h={book['to_heuristic']} r5={book['to_r5']}",
                    flush=True,
                )
        print(f"[match] game {game_id} {end} plies={ply} winner={winner} area={b_area}-{w_area + 6.5:.1f} lead={lead}", flush=True)
    db.close()
    print(f"[match] done {out / 'matches.sqlite'}", flush=True)
    return 0


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