예제 소스

run_heuristic_gemma4.py

#!/usr/bin/env python3
"""휴리스틱이 둔 수로 Gemma 4를 학습한 뒤, 계가 없이 휴리스틱과 3국을 둔다."""

from __future__ import annotations

import json
import socket
import subprocess
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PY = ROOT / ".venv" / "bin" / "python"
OUT = ROOT / "logs" / "heuristic_gemma4_20261009"
MODEL = "********"
LORA = "********"
PORT = 9341


def log(msg: str) -> None:
    print(msg, flush=True)


def run(cmd: list[str], log_path: Path) -> int:
    log("[runner] " + " ".join(cmd))
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("ab") as fh:
        rc = subprocess.call(cmd, cwd=ROOT, stdout=fh, stderr=subprocess.STDOUT)
    log(f"[runner] rc={rc} log={log_path}")
    return rc


def port_open(port: int) -> bool:
    sock = socket.socket()
    sock.settimeout(1)
    try:
        sock.connect(("127.0.0.1", port))
        return True
    except OSError:
        return False
    finally:
        sock.close()


def wait_port(port: int, timeout_s: float) -> bool:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        if port_open(port):
            return True
        time.sleep(5)
    return False


def main() -> int:
    OUT.mkdir(parents=True, exist_ok=True)
    if port_open(PORT):
        raise SystemExit(f"port {PORT} is already in use")
    rc = run(
        [
            str(PY),
            "scripts/dataset_heuristic_teacher.py",
            "--source",
            "logs/gemma4_r5_20261009/matches.sqlite:BW",
            "--source",
            "logs/heuristic_r5_eye_20261008/matches.sqlite:W",
            "--out",
            "sft_data/heuristic_gemma4",
        ],
        OUT / "dataset.log",
    )
    if rc != 0:
        return rc
    rc = run(
        [
            str(PY),
            "scripts/train_ladder_lora.py",
            "--shape",
            "onpolicy",
            "--data-name",
            "heuristic_gemma4",
            "--log-name",
            "gemma4_heuristic",
            "--model",
            MODEL,
            "--base",
            "-",
            "--out",
            LORA,
            "--shape-n",
            "2000",
            "--skip-base-eval",
        ],
        OUT / "train.log",
    )
    if rc != 0:
        return rc
    cmd = [
        str(PY),
        "-m",
        "vlm_server",
        "serve",
        "--host",
        "127.0.0.1",
        "--port",
        str(PORT),
        "--model",
        MODEL,
        "--lora-dir",
        LORA,
        "--text-only-move",
        "--board-repr",
        "aligned",
        "--no-score-guided",
        "--illegal-retry-max",
        "16",
        "--illegal-retry-budget-s",
        "240",
        "--log-jsonl",
        str(OUT / "gemma4.jsonl"),
    ]
    serve_log = open(OUT / "gemma_serve.stdout", "ab")
    log(f"[runner] start gemma4 without score guide on {PORT}")
    proc = subprocess.Popen(cmd, cwd=ROOT, stdout=serve_log, stderr=subprocess.STDOUT, start_new_session=True)
    if not wait_port(PORT, 600):
        proc.terminate()
        raise SystemExit("gemma4 did not listen")
    log(f"[runner] gemma4 listening pid={proc.pid}")
    reg_path = ROOT / "scripts" / "models.json"
    reg = json.loads(reg_path.read_text(encoding="utf-8"))
    reg["models"]["gemma4_heuristic"] = {
        "port": PORT,
        "model": MODEL,
        "lora": LORA,
        "note": "휴리스틱 착수 학습. 대국 때는 --no-score-guided.",
    }
    reg_path.write_text(json.dumps(reg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    rc = run(
        [
            str(PY),
            "scripts/play_heuristic_vs_model.py",
            "--port",
            str(PORT),
            "--model-name",
            "gemma4_heuristic",
            "--games",
            "3",
            "--max-moves",
            "0",
            "--out-dir",
            str(OUT.relative_to(ROOT)),
        ],
        OUT / "play.log",
    )
    log(f"[runner] match rc={rc}")
    return rc


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