예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
run_gemma_move_loop.py
#!/usr/bin/env python3
"""Gemma 4 수 품질 루프: 좌표 보정 뒤 자기 수가 약해 지는 문제를 KataGo 최선수 학습으로 보완.
세트 0 은 좌표 루프 마지막 세트(coord_r2 10판)를 그대로 쓴다. 회차마다
1) 지금까지 대국의 백 국면을 KataGo 최선수로 라벨(서버 수 프롬프트, 30%는 금지 좌표 재시도형)
2) 실전 좌표 데이터를 계속 섞음(좌표를 잊지 않게) + coord_v3 일부
3) 이전 어댑터에 이어 학습 → 같은 val 로 전/후 비교(KataGo top3·합법수·읽기)
4) 나아졌으면 채택, 아니면 학습률을 올려 다시
5) 휴리스틱과 10판
python scripts/run_gemma_move_loop.py
상태: logs/gemma_move_loop_20261012/REPORT.md
"""
from __future__ import annotations
import argparse
import json
import os
import random
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
import scripts.run_gemma_coord_loop as base # noqa: E402
OUT = ROOT / "logs" / "gemma_move_loop_20261012"
base.OUT = OUT
PY, MODEL, PORT = base.PY, base.MODEL, base.PORT
COORD_LOOP = ROOT / "logs" / "gemma_coord_loop_20261011"
START_LORA = "vlm_models/lora_gemma4_coord_r2"
OLD_SETS = [COORD_LOOP / f"set{k}" for k in (0, 1)] # 다른 어댑터로 둔 대국이지만 국면 라벨용으로는 유효
COORD_VAL = ROOT / "sft_data" / "gemma_coord_r2" / "val.jsonl"
log, run = base.log, base.run
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 evaluate(lora: str, kval: Path, tdir: Path, tag: str) -> dict:
k = base.eval_adapter(lora, kval, tdir / f"eval_kmove_{tag}.json")
c = base.eval_adapter(lora, COORD_VAL, tdir / f"eval_coord_{tag}.json")
return {
"kata_top1": k.get("kata_top1"),
"kata_top3": k.get("kata_top3"),
"kmove_legal": k.get("move_legal"),
"coord_move_legal": c.get("move_legal"),
"read_acc": c.get("read_acc"),
}
def better(new: dict, old: dict) -> bool:
if not new or new.get("kata_top3") is None:
return False
g = lambda d, k: d.get(k) or 0.0 # noqa: E731
return (
g(new, "kata_top3") > g(old, "kata_top3")
and g(new, "kmove_legal") >= g(old, "kmove_legal") - 5
and g(new, "read_acc") >= g(old, "read_acc") - 5
)
def write_report(st: dict) -> None:
L = [
"# Gemma 4 수 품질 루프 (KataGo 최선수 + 실전 좌표 유지)\n",
f"- 갱신: {datetime.now():%Y-%m-%d %H:%M}",
f"- 시작 어댑터: `{START_LORA}` · 현재: `{st.get('lora', START_LORA)}`",
"",
"## 대국 (휴리스틱 흑 vs Gemma 백, 150수)\n",
"| 세트 | 어댑터 | 판 | Gemma 승 | 평균 KataGo 흑 우세 | 첫 시도 합법 | 서버 대체(무작위) | 수당 시도 |",
"|---|---|---|---|---|---|---|---|",
]
for k, s in sorted(st.get("sets", {}).items(), key=lambda x: int(x[0])):
L.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')} |")
L += ["\n## 학습 · 평가 (같은 val 로 전/후)\n",
"| 회차 | 학습률 | 데이터 | KataGo top1/top3 (전→후) | 수 합법 (전→후) | 좌표 읽기 (전→후) | 채택 |",
"|---|---|---|---|---|---|---|"]
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", {})
L.append(f"| {k} | {t.get('lr')} | {t.get('mix_n')} | {o.get('kata_top1')}/{o.get('kata_top3')} → "
f"{n.get('kata_top1')}/{n.get('kata_top3')} | {o.get('kmove_legal')} → {n.get('kmove_legal')} | "
f"{o.get('read_acc')} → {n.get('read_acc')} | {'예' if t.get('adopted') else '아니오'} |")
(OUT / "REPORT.md").write_text("\n".join(L) + "\n", encoding="utf-8")
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--rounds", type=int, default=2)
ap.add_argument("--games", type=int, default=10)
ap.add_argument("--max-moves", type=int, default=150)
ap.add_argument("--kmove-n", type=int, default=1500)
ap.add_argument("--coord-n", type=int, default=700)
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("sets", {})
st.setdefault("trains", {})
save_state(st)
set0 = OUT / "set0"
if "0" not in st["sets"]:
set0.mkdir(parents=True, exist_ok=True)
for name in ("matches.sqlite", "gemma4.jsonl", "play.log"):
shutil.copy2(COORD_LOOP / "set2" / name, set0 / name)
s = base.play_stats(set0)
s.update(lora=START_LORA, done=True, note="좌표 루프 set2 재사용")
st["sets"]["0"] = s
save_state(st)
write_report(st)
base.HOLD.write_text("gemma move loop\n", encoding="utf-8")
try:
for k in range(1, args.rounds + 1):
t = st["trains"].setdefault(str(k), {})
tdir = OUT / f"train{k}"
kdata = ROOT / f"sft_data/gemma_kmove_r{k}"
cdata = ROOT / f"sft_data/gemma_coordkeep_r{k}"
sets = OLD_SETS + [OUT / f"set{j}" for j in range(k)]
if not (kdata / "summary.json").exists():
cmd = [PY, "scripts/dataset_moves_katago.py", "--out", str(kdata), "--seed", str(31 + k)]
for sd in sets:
cmd += ["--source", str(sd / "matches.sqlite"), "--jsonl", str(sd / "gemma4.jsonl")]
if run(cmd, tdir / "dataset.log") != 0:
return 1
if not (cdata / "summary.json").exists():
cmd = [PY, "scripts/dataset_coords_game.py", "--out", str(cdata), "--n", str(args.coord_n),
"--val-n", "20", "--seed", str(61 + k)]
for sd in sets:
cmd += ["--source", str(sd / "matches.sqlite"), "--jsonl", str(sd / "gemma4.jsonl")]
if run(cmd, tdir / "dataset.log") != 0:
return 1
kval = kdata / "val.jsonl"
if not t.get("trained"):
base.stop_gemma()
rng = random.Random(71 + k)
kl = [x for x in (kdata / "train.jsonl").read_text(encoding="utf-8").splitlines() if x.strip()]
rng.shuffle(kl)
lines = kl[: args.kmove_n]
lines += [x for x in (cdata / "train.jsonl").read_text(encoding="utf-8").splitlines() if x.strip()]
cv3 = [x for x in (ROOT / "sft_data/coord_v3/train.jsonl").read_text(encoding="utf-8").splitlines() if x.strip()]
rng.shuffle(cv3)
lines += cv3[:200]
rng.shuffle(lines)
mix = ROOT / f"sft_data/mix_gemma_move_r{k}/train.jsonl"
mix.parent.mkdir(parents=True, exist_ok=True)
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()
new_lora = f"vlm_models/lora_gemma4_move_r{k}"
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(kval),
"--resume-from", st["lora"]], tdir / "train.log")
log(f"move train{k} 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)
new_lora = f"vlm_models/lora_gemma4_move_r{k}"
base.stop_gemma()
t["eval_old"] = evaluate(t.get("base", st["lora"]), kval, tdir, "old")
t["eval_new"] = evaluate(new_lora, kval, tdir, "new")
if "adopted" not in t:
t["adopted"] = better(t["eval_new"], t["eval_old"])
if t["adopted"]:
st["lora"], st["lr"] = new_lora, "1e-5"
base.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["models"]["gemma4_heuristic"]["lora"] = new_lora
reg["models"]["gemma4_heuristic"]["note"] = "좌표 보정 + KataGo 수 품질 루프 채택 어댑터. 대국 때는 --no-score-guided."
reg_path.write_text(json.dumps(reg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
else:
st["lr"] = "2e-5"
log(f"move train{k} old={t['eval_old']} new={t['eval_new']} adopted={t['adopted']}")
save_state(st)
write_report(st)
set_dir = OUT / f"set{k}"
if not st["sets"].get(str(k), {}).get("done"):
done = base.completed_games(set_dir / "play.log")
left = args.games - done
if left > 0:
base.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 base.completed_games(set_dir / "play.log") < args.games:
log(f"move set{k} play stopped rc={rc}")
return 1
s = base.play_stats(set_dir)
s.update(lora=st["lora"], done=True)
st["sets"][str(k)] = s
save_state(st)
write_report(st)
log(f"move set{k} stats {s}")
finally:
if not base.port_open(PORT):
try:
base.start_gemma(st["lora"], OUT / "final" / "gemma4.jsonl")
except SystemExit as exc:
log(f"gemma restart failed: {exc}")
if base.HOLD.exists():
base.HOLD.unlink()
write_report(st)
log("move loop done")
return 0
if __name__ == "__main__":
raise SystemExit(main())