예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
run_gemma_coord_x10.py
#!/usr/bin/env python3
"""좌표 혼합 10에폭 학습 후 휴리스틱과 10국(150수).
채택된 lora_gemma4_coord_r2 에서 이어 받는다. 데이터는 좌표 루프 2회차와
같은 mix_gemma_coord_r2 (실전 좌표 + coord_v3 200 + 휴리스틱 수 400) 이다.
에폭이 끝날 때마다 중간 체크포인트를 지우고, 중단 후 다시 실행하면 끝난 에폭은 건너뛴다.
python scripts/run_gemma_coord_x10.py
상태: logs/gemma_coord_x10/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_x10"
HOLD = ROOT / "logs" / "mariago_web_watch" / "GEMMA_HOLD"
LORA_FILE = ROOT / "logs" / "mariago_web_watch" / "gemma_lora.txt"
MODEL = "********"
BASE_LORA = "vlm_models/lora_gemma4_coord_r2"
OUT_LORA = "********"
RESUME_STAGE = ROOT / "vlm_models" / "lora_gemma4_coord_x10_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 좌표 10에폭 후 휴리스틱 대국",
"",
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', 10)}",
"- 대국: 휴리스틱 흑 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 point_registry(lora: str) -> None:
LORA_FILE.write_text(lora + "\n", encoding="utf-8")
reg_path = ROOT / "scripts" / "models.json"
if not reg_path.exists():
return
reg = json.loads(reg_path.read_text(encoding="utf-8"))
reg.setdefault("models", {}).setdefault("gemma4_heuristic", {})["lora"] = lora
reg["models"]["gemma4_heuristic"]["note"] = "좌표 혼합 10에폭. 대국 때는 --no-score-guided."
reg_path.write_text(json.dumps(reg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--epochs", type=int, default=10)
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 x10\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:02d}"
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:
point_registry(OUT_LORA)
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:
point_registry(OUT_LORA)
if not port_open(PORT):
try:
start_gemma(OUT_LORA, OUT / "final" / "gemma4.jsonl")
except SystemExit as exc:
log(f"gemma restart failed: {exc}")
if HOLD.exists():
HOLD.unlink()
write_report(st)
log("coord x10 done" if finished else "coord x10 stopped")
if __name__ == "__main__":
raise SystemExit(main())