예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
run_gemma_coord_loop.py
#!/usr/bin/env python3
"""Gemma 4 실전 좌표 보정 루프: 휴리스틱과 10판 → 좌표 학습 → 10판 → … (중단 후 다시 실행하면 이어서).
세트마다 대국 기록과 서버 jsonl(틀린 좌표)로 dataset_coords_game 데이터를 만들고,
이전 어댑터에 이어 학습한다. 같은 val 로 전/후를 비교해 나아졌을 때만 채택하고,
나아지지 않으면 학습률을 올리고 읽기 과제 비중을 늘려 다시 한다.
python scripts/run_gemma_coord_loop.py # 10판 ×3세트, 학습 2회
상태: logs/gemma_coord_loop_20261011/REPORT.md
"""
from __future__ import annotations
import argparse
import json
import os
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_loop_20261011"
HOLD = ROOT / "logs" / "mariago_web_watch" / "GEMMA_HOLD"
LORA_FILE = ROOT / "logs" / "mariago_web_watch" / "gemma_lora.txt"
MODEL = "********"
START_LORA = "********"
PORT = 9341
BOARD_REPR = "aligned"
MAX_NEW_TOKENS = 96
SEED_SOURCES = [ # 첫 학습에 보탤 이전 대국 (r3 어댑터·같은 서버 설정)
("logs/heuristic_gemma4_r3_20261010/matches.sqlite", "logs/heuristic_gemma4_r3_20261010/gemma4.jsonl"),
("logs/heuristic_gemma4_150x5/matches.sqlite", ""),
]
MIX_DEFAULT = "lookup:0.35,row:0.15,col:0.10,move:0.15,move_retry:0.25"
MIX_READ = "lookup:0.40,row:0.20,col:0.15,move:0.10,move_retry:0.15"
def log(msg: str) -> None:
line = f"[{datetime.now():%m-%d %H:%M:%S}] {msg}"
print(line, flush=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", BOARD_REPR, "--no-score-guided",
"--illegal-retry-max", "16", "--illegal-retry-budget-s", "240",
"--max-new-tokens", str(MAX_NEW_TOKENS),
"--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 / "state.json").write_text(json.dumps(st, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
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 play_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:
r = json.loads(ln)
except json.JSONDecodeError:
continue
if r.get("event") == "move":
moves.append(r)
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()
con.close()
leads = [g[4] for g in games if g[4] is not None]
st.update(
games=len(games),
gemma_wins=sum(1 for g in games if g[3] == "white"),
avg_lead_b=round(sum(leads) / len(leads), 1) if leads else None,
)
return st
def eval_adapter(lora: str, val: Path, out: Path) -> dict:
if not out.exists():
run([PY, "scripts/eval_coords_game.py", "--data", str(val), "--model", MODEL,
"--lora-dir", lora, "--out", str(out)], OUT / "eval.log")
if not out.exists():
return {}
return json.loads(out.read_text(encoding="utf-8"))["summary"]
def better(new: dict, old: dict) -> bool:
if not new:
return False
nl, ol = new.get("move_legal", 0), old.get("move_legal", 0)
nr, orr = new.get("read_acc", 0), old.get("read_acc", 0)
return (nl > ol and nr >= orr - 5) or (nl >= ol and nr > orr + 5)
def write_report(st: dict) -> None:
lines = [f"# Gemma 4 실전 좌표 보정 루프\n", f"- 갱신: {datetime.now():%Y-%m-%d %H:%M}",
f"- 시작 어댑터: `{START_LORA}`", f"- 현재 어댑터: `{st.get('lora', START_LORA)}`", ""]
lines.append("## 대국 (휴리스틱 흑 vs Gemma 백, 150수)\n")
lines.append("| 세트 | 어댑터 | 판 | Gemma 승 | 평균 KataGo 흑 우세 | 첫 시도 합법 | 서버 대체 | 수당 시도 | 이미 돌 있는 칸 |")
lines.append("|---|---|---|---|---|---|---|---|---|")
for k, s in sorted(st.get("sets", {}).items(), key=lambda x: int(x[0])):
lines.append(f"| {k} | `{Path(s['lora']).name}` | {s.get('games', '-')} | {s.get('gemma_wins', '-')} | "
f"{s.get('avg_lead_b', '-')} | {s.get('first_try_pct')}% | {s.get('fallback_pct')}% | "
f"{s.get('attempts_per_move')} | {s.get('occupied_pct')}% |")
lines.append("\n## 학습 · 좌표 평가 (실전 국면 val)\n")
lines.append("| 회차 | 학습률 | 데이터 | 이전 합법수/읽기 | 새 합법수/읽기 | 채택 |")
lines.append("|---|---|---|---|---|---|")
for k, t in sorted(st.get("trains", {}).items(), key=lambda x: int(x[0])):
o, n = t.get("eval_old", {}), t.get("eval_new", {})
lines.append(f"| {k} | {t.get('lr')} | {t.get('mix_n')} | {o.get('move_legal')}% / {o.get('read_acc')}% | "
f"{n.get('move_legal')}% / {n.get('read_acc')}% | {'예' if t.get('adopted') else '아니오'} |")
(OUT / "REPORT.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--rounds", type=int, default=2, help="학습 횟수 (대국 세트는 +1)")
ap.add_argument("--games", type=int, default=10)
ap.add_argument("--max-moves", type=int, default=150)
ap.add_argument("--n", type=int, default=2000, help="회차당 실전 좌표 샘플 수")
args = ap.parse_args()
OUT.mkdir(parents=True, exist_ok=True)
(OUT / "runner.pid").write_text(str(os.getpid()))
st = state()
st.setdefault("lora", START_LORA)
st.setdefault("lr", "1e-5")
st.setdefault("mix", MIX_DEFAULT)
st.setdefault("sets", {})
st.setdefault("trains", {})
save_state(st)
HOLD.parent.mkdir(parents=True, exist_ok=True)
HOLD.write_text("gemma coord loop\n", encoding="utf-8")
try:
for k in range(args.rounds + 1):
set_dir = OUT / f"set{k}"
key = str(k)
if key not in st["sets"] or not st["sets"][key].get("done"):
done = completed_games(set_dir / "play.log")
left = args.games - done
if left > 0:
start_gemma(st["lora"], set_dir / "gemma4.jsonl")
rc = run([PY, "scripts/play_heuristic_vs_model.py", "--port", str(PORT),
"--model-name", Path(st["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"set{k} play stopped rc={rc}")
return 1
s = play_stats(set_dir)
s.update(lora=st["lora"], done=True)
st["sets"][key] = s
save_state(st)
write_report(st)
log(f"set{k} stats {s}")
if k == args.rounds:
break
t = st["trains"].setdefault(str(k + 1), {})
data = f"sft_data/gemma_coord_r{k + 1}"
new_lora = f"vlm_models/lora_gemma4_coord_r{k + 1}"
tdir = OUT / f"train{k + 1}"
if not (ROOT / data / "summary.json").exists():
cmd = [PY, "scripts/dataset_coords_game.py", "--out", data, "--n", str(args.n),
"--mix", st["mix"], "--seed", str(29 + k)]
for j in range(k + 1):
cmd += ["--source", str(OUT / f"set{j}" / "matches.sqlite"),
"--jsonl", str(OUT / f"set{j}" / "gemma4.jsonl")]
for db, jl in SEED_SOURCES:
cmd += ["--source", db] + (["--jsonl", jl] if jl else [])
if run(cmd, tdir / "dataset.log") != 0:
return 1
val = ROOT / data / "val.jsonl"
if not t.get("trained"):
stop_gemma()
mix = ROOT / f"sft_data/mix_gemma_coord_r{k + 1}" / "train.jsonl"
mix.parent.mkdir(parents=True, exist_ok=True)
lines = (ROOT / data / "train.jsonl").read_text(encoding="utf-8").splitlines()
import random as _r
rng = _r.Random(41 + k)
for extra, n in (("sft_data/coord_v3/train.jsonl", 200), ("sft_data/heuristic_gemma4_r3/train.jsonl", 400)):
pool = [x for x in (ROOT / extra).read_text(encoding="utf-8").splitlines() if x.strip()]
rng.shuffle(pool)
lines += pool[:n]
rng.shuffle(lines)
mix.write_text("\n".join(lines) + "\n", encoding="utf-8")
t.update(lr=st["lr"], mix_n=len(lines), base=st["lora"], started=f"{datetime.now():%m-%d %H:%M}")
save_state(st)
t0 = time.time()
rc = run([PY, "-m", "vlm_server.train_lora", "--dataset-jsonl", str(mix), "--model", MODEL,
"--output-dir", new_lora, "--epochs", "1", "--batch-size", "1", "--grad-accum", "8",
"--lr", st["lr"], "--lora-rank", "16", "--lora-alpha", "32", "--max-seq-length", "4096",
"--max-samples", str(len(lines)), "--grad-checkpoint", "--val-jsonl", str(val),
"--resume-from", st["lora"]], tdir / "train.log")
log(f"train{k + 1} rc={rc} {(time.time() - t0) / 3600:.1f}h")
if rc != 0:
return rc
t["trained"] = True
t["hours"] = round((time.time() - t0) / 3600, 2)
save_state(st)
stop_gemma()
t["eval_old"] = eval_adapter(t.get("base", st["lora"]), val, tdir / "eval_old.json")
t["eval_new"] = eval_adapter(new_lora, val, tdir / "eval_new.json")
if "adopted" not in t:
t["adopted"] = better(t["eval_new"], t["eval_old"])
if t["adopted"]:
st["lora"] = new_lora
st["lr"], st["mix"] = "1e-5", MIX_DEFAULT
LORA_FILE.write_text(new_lora + "\n", encoding="utf-8")
reg_path = ROOT / "scripts" / "models.json"
reg = json.loads(reg_path.read_text(encoding="utf-8"))
reg.setdefault("models", {})["gemma4_heuristic"]["lora"] = new_lora
reg["models"]["gemma4_heuristic"]["note"] = "실전 국면 좌표 보정 루프 채택 어댑터. 대국 때는 --no-score-guided."
reg_path.write_text(json.dumps(reg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
else:
st["lr"], st["mix"] = "2e-5", MIX_READ
log(f"train{k + 1} old={t['eval_old']} new={t['eval_new']} adopted={t['adopted']}")
save_state(st)
write_report(st)
finally:
if not port_open(PORT):
try:
start_gemma(st["lora"], OUT / "final" / "gemma4.jsonl")
except SystemExit as exc:
log(f"gemma restart failed: {exc}")
if HOLD.exists():
HOLD.unlink()
write_report(st)
log("loop done")
return 0
if __name__ == "__main__":
raise SystemExit(main())