예제 소스

run_gemma_coord_x100.py

#!/usr/bin/env python3
"""좌표 혼합 100에폭 학습 후 휴리스틱과 10국(150수).

10에폭 어댑터 lora_gemma4_coord_x10 에서 이어 받는다. 데이터는 좌표 루프 2회차와
같은 mix_gemma_coord_r2 (실전 좌표 + coord_v3 200 + 휴리스틱 수 400) 이다.
에폭이 끝날 때마다 중간 체크포인트를 지우고, 중단 후 다시 실행하면 끝난 에폭은 건너뛴다.
대국이 끝나면 학습 중에 멈춰 둔 9341 서버는 워치독이 gemma_lora.txt 로 다시 띄운다.

  python scripts/run_gemma_coord_x100.py
  상태: logs/gemma_coord_x100/REPORT.md
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import signal
import socket
import sqlite3
import subprocess
import time
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
PY = str(ROOT / ".venv" / "bin" / "python")
OUT = ROOT / "logs" / "gemma_coord_x100"
HOLD = ROOT / "logs" / "mariago_web_watch" / "GEMMA_HOLD"
MODEL = "********"
BASE_LORA = "********"
OUT_LORA = "********"
RESUME_STAGE = ROOT / "vlm_models" / "lora_gemma4_coord_x100_resume"
MIX = ROOT / "sft_data" / "mix_gemma_coord_r2" / "train.jsonl"
PORT = 9341


def log(msg: str) -> None:
    line = f"[{datetime.now():%m-%d %H:%M:%S}] {msg}"
    print(line, flush=True)
    OUT.mkdir(parents=True, exist_ok=True)
    with (OUT / "runner.log").open("a", encoding="utf-8") as f:
        f.write(line + "\n")


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


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


def gemma_pids() -> list[int]:
    try:
        out = subprocess.check_output(["pgrep", "-f", rf"vlm_server serve.*--port {PORT}"], text=True)
    except subprocess.CalledProcessError:
        return []
    return [int(x) for x in out.split()]


def stop_gemma() -> None:
    pids = gemma_pids()
    if not pids:
        return
    log(f"stop gemma {pids}")
    for pid in pids:
        try:
            os.kill(pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
    deadline = time.time() + 40
    while time.time() < deadline and (port_open(PORT) or gemma_pids()):
        time.sleep(1)
    for pid in gemma_pids():
        try:
            os.kill(pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
    time.sleep(2)


def start_gemma(lora: str, jsonl: Path) -> None:
    stop_gemma()
    jsonl.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        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(jsonl),
    ]
    logf = open(OUT / "gemma_serve.stdout", "ab")
    log(f"start gemma lora={lora}")
    subprocess.Popen(cmd, cwd=ROOT, stdout=logf, stderr=subprocess.STDOUT, start_new_session=True)
    deadline = time.time() + 600
    while time.time() < deadline:
        if port_open(PORT):
            log("gemma listening")
            return
        time.sleep(5)
    raise SystemExit("gemma did not listen")


def state() -> dict:
    p = OUT / "state.json"
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}


def save_state(st: dict) -> None:
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / "state.json").write_text(json.dumps(st, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def prune_checkpoints(directory: Path) -> None:
    for path in directory.glob("*_adapters.safetensors"):
        if path.name[0].isdigit():
            path.unlink()
            log(f"pruned {path.name}")


def publish_resume(src: Path) -> None:
    weights = src / "adapters.safetensors"
    cfg = src / "adapter_config.json"
    if not weights.exists() or not cfg.exists():
        raise SystemExit(f"adapter incomplete: {src}")
    RESUME_STAGE.mkdir(parents=True, exist_ok=True)
    shutil.copy2(weights, RESUME_STAGE / "adapters.safetensors")
    shutil.copy2(cfg, RESUME_STAGE / "adapter_config.json")


def train_epoch(epoch: int, resume_from: Path, log_path: Path) -> int:
    cmd = [
        PY, "-m", "vlm_server.train_lora",
        "--dataset-jsonl", str(MIX),
        "--model", MODEL,
        "--output-dir", OUT_LORA,
        "--epochs", "1",
        "--batch-size", "1",
        "--grad-accum", "8",
        "--lr", "1e-5",
        "--lora-rank", "16",
        "--lora-alpha", "32",
        "--max-seq-length", "4096",
        "--max-samples", "2600",
        "--grad-checkpoint",
        "--resume-from", str(resume_from),
    ]
    return run(cmd, log_path)


def edge_line(row: int, col: int) -> int:
    return min(row, col, 18 - row, 18 - col) + 1


def completed_games(log_path: Path) -> int:
    if not log_path.exists():
        return 0
    return sum(1 for ln in log_path.read_text(encoding="utf-8", errors="replace").splitlines()
               if ln.startswith("[match] game ") and " plies=" in ln)


def match_stats(set_dir: Path) -> dict:
    moves = []
    jl = set_dir / "gemma4.jsonl"
    if jl.exists():
        for ln in jl.open(encoding="utf-8", errors="replace"):
            try:
                rec = json.loads(ln)
            except json.JSONDecodeError:
                continue
            if rec.get("event") == "move":
                moves.append(rec)
    n = len(moves) or 1
    st = {
        "gemma_moves": len(moves),
        "first_try_pct": round(100 * sum(1 for r in moves if not r.get("retry_count") and not r.get("fallback_used")) / n, 1),
        "fallback_pct": round(100 * sum(1 for r in moves if r.get("fallback_used")) / n, 1),
        "attempts_per_move": round(sum(len(r.get("vlm_seeds") or []) for r in moves) / n, 2),
        "occupied_pct": round(100 * sum(1 for r in moves if r.get("illegal_reason") == "occupied") / n, 1),
    }
    db = set_dir / "matches.sqlite"
    if db.exists():
        con = sqlite3.connect(db)
        games = con.execute(
            "select game_id, end_reason, plies, winner, katago_lead_b from games where finished_at is not null"
        ).fetchall()
        rows = con.execute(
            "select ply, row, col from moves where color='W' and kind='move' and row is not null"
        ).fetchall()
        con.close()
        leads = [g[4] for g in games if g[4] is not None]
        lines: dict[str, int] = {}
        early: dict[str, int] = {}
        for ply, row, col in rows:
            key = str(edge_line(int(row), int(col)))
            lines[key] = lines.get(key, 0) + 1
            if int(ply) < 40:
                early[key] = early.get(key, 0) + 1
        white_n = sum(lines.values()) or 1
        early_n = sum(early.values()) or 1
        st.update(
            games=len(games),
            gemma_wins=sum(1 for g in games if g[3] == "white"),
            heuristic_wins=sum(1 for g in games if g[3] == "black"),
            avg_lead_b=round(sum(leads) / len(leads), 1) if leads else None,
            leads=[round(x, 2) for x in leads],
            line1=lines.get("1", 0),
            line1_rate=round(lines.get("1", 0) / white_n, 3),
            early40_line1_rate=round(early.get("1", 0) / early_n, 3) if early else None,
            lines=lines,
        )
    return st


def write_report(st: dict) -> None:
    match = st.get("match") or {}
    lines = [
        "# Gemma 4 좌표 100에폭 후 휴리스틱 대국",
        "",
        f"- 갱신: {datetime.now():%Y-%m-%d %H:%M}",
        f"- 시작 어댑터: `{BASE_LORA}`",
        f"- 학습 어댑터: `{OUT_LORA}`",
        f"- 데이터: `{MIX.relative_to(ROOT)}` 2600샘플, 학습률 1e-5, 에폭 {st.get('epochs_done', 0)}/{st.get('epochs', 100)}",
        "- 대국: 휴리스틱 흑 vs Gemma 백, 150수, 점수 유도 없음",
        "",
        "## 대국",
        "",
        f"- 판: {match.get('games', '-')}",
        f"- Gemma 승: {match.get('gemma_wins', '-')}",
        f"- 휴리스틱 승: {match.get('heuristic_wins', '-')}",
        f"- 평균 KataGo 흑 우세: {match.get('avg_lead_b', '-')}",
        f"- 흑 우세 목록: {match.get('leads', '-')}",
        f"- 첫 시도 합법: {match.get('first_try_pct', '-')}%",
        f"- 서버 대체: {match.get('fallback_pct', '-')}%",
        f"- 수당 시도: {match.get('attempts_per_move', '-')}",
        f"- 이미 돌 있는 칸: {match.get('occupied_pct', '-')}%",
        f"- 백 1선: {match.get('line1', '-')} ({match.get('line1_rate', '-')})",
        f"- 초반 40수 1선: {match.get('early40_line1_rate', '-')}",
        f"- 선 분포: {match.get('lines', '-')}",
        "",
    ]
    (OUT / "REPORT.md").write_text("\n".join(lines), encoding="utf-8")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--epochs", type=int, default=100)
    ap.add_argument("--games", type=int, default=10)
    ap.add_argument("--max-moves", type=int, default=150)
    args = ap.parse_args()
    if not MIX.exists():
        raise SystemExit(f"missing mix {MIX}")
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / "runner.pid").write_text(str(os.getpid()))
    st = state()
    st.setdefault("epochs_done", 0)
    st["epochs"] = args.epochs
    st["base"] = BASE_LORA
    st["lora"] = OUT_LORA
    save_state(st)
    HOLD.write_text("gemma coord x100\n", encoding="utf-8")
    stop_gemma()
    finished = False
    try:
        while st["epochs_done"] < args.epochs:
            epoch = st["epochs_done"] + 1
            resume = Path(BASE_LORA) if epoch == 1 else RESUME_STAGE
            if epoch > 1 and not (RESUME_STAGE / "adapters.safetensors").exists():
                raise SystemExit(f"resume adapter missing for epoch {epoch}")
            tdir = OUT / f"epoch{epoch:03d}"
            t0 = time.time()
            log(f"epoch {epoch}/{args.epochs} resume={resume}")
            rc = train_epoch(epoch, resume, tdir / "train.log")
            if rc != 0:
                log(f"epoch {epoch} failed rc={rc}")
                return rc
            out_dir = ROOT / OUT_LORA
            prune_checkpoints(out_dir)
            publish_resume(out_dir)
            st["epochs_done"] = epoch
            st[f"epoch{epoch}_hours"] = round((time.time() - t0) / 3600, 2)
            save_state(st)
            write_report(st)
            log(f"epoch {epoch} done {st[f'epoch{epoch}_hours']}h")

        set_dir = OUT / "match150"
        if not st.get("match", {}).get("done"):
            done = completed_games(set_dir / "play.log")
            left = args.games - done
            if left > 0:
                start_gemma(OUT_LORA, set_dir / "gemma4.jsonl")
                rc = run([
                    PY, "scripts/play_heuristic_vs_model.py",
                    "--port", str(PORT),
                    "--model-name", Path(OUT_LORA).name,
                    "--games", str(left),
                    "--start-id", str(done + 1),
                    "--max-moves", str(args.max_moves),
                    "--out-dir", str(set_dir.relative_to(ROOT)),
                ], set_dir / "play.log")
                if rc != 0 or completed_games(set_dir / "play.log") < args.games:
                    log(f"match stopped rc={rc}")
                    return 1
            stats = match_stats(set_dir)
            stats["done"] = True
            stats["lora"] = OUT_LORA
            st["match"] = stats
            save_state(st)
            write_report(st)
            log(f"match {stats}")
        finished = True
        return 0
    finally:
        if finished:
            stop_gemma()
            if HOLD.exists():
                HOLD.unlink()
            log("released gemma port for the watchdog")
        write_report(st)
        log("coord x100 done" if finished else "coord x100 stopped")


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