예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
run_gemma_read2_loop.py
#!/usr/bin/env python3
"""Gemma 4 읽고 고르기 3회차: 빈 열을 먼저 적고 그 안에서만 고르게 한다.
2회차는 행을 맞게 옮겨 적고도(99%) 그 줄의 돌 있는 열을 골랐다. 좋아하는 자리가
10행에 몰려 있었고, 시험 국면은 그 자리가 비어 있어서 100%가 나왔다.
이번 시험은 2회차 자신의 대국(읽고 고르기 루프 set2)이다. 이 대국은 학습에 넣지 않는다.
답은 row → line → empty(그 행의 빈 열 전부) → col 이고, col 은 empty 안에서만 고른다.
좋아하는 자리가 이미 돌로 찬 국면은 학습 데이터에 한 번 더 넣는다.
다른 세션의 좌표 반복 학습(run_gemma_x10_nolimit_loop.py)이 그래픽 장치를 쓰는 동안은
시작하지 않고 기다린다. 그 작업은 logs/gemma_x10_nolimit/STOP 이 생기면 이번 단계 끝에 선다.
python scripts/run_gemma_read2_loop.py
상태: logs/gemma_read_loop_20261020/REPORT.md
"""
from __future__ import annotations
import argparse
import json
import os
import random
import subprocess
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
from scripts.run_gemma_read_loop import MOVE_TASKS, convert, jl, repr_of # noqa: E402
OUT = ROOT / "logs" / "gemma_read_loop_20261020"
base.OUT = OUT
PY, MODEL, PORT = base.PY, base.MODEL, base.PORT
log, run = base.log, base.run
START_LORA = "vlm_models/lora_gemma4_read_r2"
READ_LOOP = ROOT / "logs" / "gemma_read_loop_20261014"
MOVE_LOOP = ROOT / "logs" / "gemma_move_loop_20261012"
COORD_LOOP = ROOT / "logs" / "gemma_coord_loop_20261011"
HOLDOUT = READ_LOOP / "set2" # 2회차 자신의 대국. 학습 금지
BASE_SOURCES = [
COORD_LOOP / "set0", COORD_LOOP / "set1", COORD_LOOP / "set2", MOVE_LOOP / "set1",
ROOT / "logs" / "gemma_coord_x10" / "match150", ROOT / "logs" / "heuristic_gemma4_r3_20261010",
READ_LOOP / "set1",
]
TEST_DIR = ROOT / "sft_data" / "gemma_read2_test"
TEST_NEW = TEST_DIR / "val.jsonl"
TEST_OLD = TEST_DIR / "val_prev.jsonl"
COORD_VAL = ROOT / "sft_data" / "gemma_coord_r2" / "val.jsonl"
COORD_KEEP = ROOT / "sft_data" / "gemma_coordkeep_r2" / "train.jsonl"
# 2회차가 실전에서 반복해서 고른 자리. 이 자리가 이미 찬 국면을 더 학습한다.
FAV = [(10, 2), (10, 14), (10, 1), (10, 4), (10, 16), (10, 3), (10, 5), (10, 13)]
OTHER_JOB = "run_gemma_x10_nolimit_loop.py"
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 wait_for_gpu() -> None:
while True:
try:
out = subprocess.check_output(["pgrep", "-f", OTHER_JOB], text=True)
except subprocess.CalledProcessError:
return
log(f"다른 학습이 장치를 사용 중({out.strip().replace(chr(10), ',')}) — 60초 후 다시 확인")
time.sleep(60)
def taken_favorites(line: str) -> int:
board = json.loads(line)["board"]
return sum(board[r * 19 + c] != "." for r, c in FAV)
def evaluate(lora: str, test: Path, tdir: Path, tag: str) -> dict:
t = base.eval_adapter(lora, test, tdir / f"eval_test_{tag}.json")
c = base.eval_adapter(lora, COORD_VAL, tdir / f"eval_coord_{tag}.json")
keys = ("move_legal", "move_empty", "move_own", "move_distinct", "kata_top1", "kata_top3",
"line_ok", "consistent", "empty_ok")
out = {k: t.get(k) for k in keys}
out["read_acc"] = c.get("read_acc")
out["col_acc"] = (c.get("col") or {}).get("acc")
return out
def better(new: dict, old: dict) -> bool:
g = lambda d, k: d.get(k) or 0.0 # noqa: E731
return g(new, "move_legal") >= g(old, "move_legal") + 5 and g(new, "read_acc") >= g(old, "read_acc") - 5
def write_report(st: dict) -> None:
L = [
"# Gemma 4 읽고 고르기 3회차 (빈 열 목록 다음 그 안에서 고르기)\n",
f"- 갱신: {datetime.now():%Y-%m-%d %H:%M}",
f"- 시작 어댑터: `{START_LORA}` · 현재: `{st.get('lora', START_LORA)}`",
"- 시험: 2회차 자신의 대국(읽고 고르기 루프 set2) — 학습에 넣지 않음",
"- 좋아하는 자리(10행)가 이미 찬 국면은 학습에 한 번 더 넣음",
"",
"## 대국 (휴리스틱 흑 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('board_repr', '-')} | {s.get('games', '-')} | "
f"{s.get('gemma_wins', '-')} | {s.get('avg_lead_b', '-')} | {s.get('first_try_pct')}% | "
f"{s.get('fallback_pct')}% | {s.get('attempts_per_move')} | {s.get('occupied_pct')}% |")
L += ["\n## 시험 (2회차 자기 대국 국면)\n",
"| 단계 | 어댑터 | 형식 | 첫 수 합법 | 빈칸 | 자기 돌 위 | 서로 다른 자리 | 줄 정확 | 적은 줄의 빈칸 | 빈 열 목록 정확 | KataGo top3 | 좌표 읽기 |",
"|---|---|---|---|---|---|---|---|---|---|---|---|"]
for name, e in st.get("evals", {}).items():
L.append(f"| {name} | `{Path(e['lora']).name}` | {e['repr']} | {e.get('move_legal')} | {e.get('move_empty')} | "
f"{e.get('move_own')} | {e.get('move_distinct')} | {e.get('line_ok', '-')} | "
f"{e.get('consistent', '-')} | {e.get('empty_ok', '-')} | {e.get('kata_top3')} | {e.get('read_acc')} |")
L += ["\n## 학습\n", "| 회차 | 학습률 | 데이터 | 시간 | 채택 |", "|---|---|---|---|---|"]
for k, t in sorted(st.get("trains", {}).items(), key=lambda x: int(x[0])):
L.append(f"| {k} | {t.get('lr')} | {t.get('mix_n')} | {t.get('hours', '-')}h | "
f"{'예' if t.get('adopted') else ('아니오' if 'adopted' in t else '-')} |")
(OUT / "REPORT.md").write_text("\n".join(L) + "\n", encoding="utf-8")
def build_test() -> None:
if TEST_OLD.exists():
return
cmd = [PY, "scripts/dataset_moves_katago.py", "--out", str(TEST_DIR), "--board-repr", "aligned_read2",
"--source", str(HOLDOUT / "matches.sqlite"), "--max-positions", "0",
"--holdout-source", str(HOLDOUT / "matches.sqlite"), "--jsonl", str(HOLDOUT / "gemma4.jsonl"),
"--val-n", "150", "--retry-frac", "0.2", "--seed", "201"]
if run(cmd, OUT / "dataset.log") != 0:
raise SystemExit("test dataset failed")
recs = [json.loads(x) for x in jl(TEST_NEW)]
TEST_OLD.write_text("".join(json.dumps(convert(r, "aligned_read"), ensure_ascii=False) + "\n" for r in recs),
encoding="utf-8")
def record_eval(st: dict, name: str, lora: str, test: Path, tdir: Path, tag: str, board_repr: str) -> dict:
if name not in st.setdefault("evals", {}):
base.stop_gemma()
e = evaluate(lora, test, tdir, tag)
e.update(lora=lora, repr=board_repr)
st["evals"][name] = e
save_state(st)
write_report(st)
log(f"eval {name} {e}")
return st["evals"][name]
def adopt(new_lora: str) -> None:
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"] = (
"빈 열 목록 다음 고르기(--board-repr aligned_read2, --max-new-tokens 192). 대국 때는 --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("--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=1700)
ap.add_argument("--lr", default="2e-5")
args = ap.parse_args()
OUT.mkdir(parents=True, exist_ok=True)
(OUT / "runner.pid").write_text(str(os.getpid()))
wait_for_gpu()
st = state()
st.setdefault("lora", START_LORA)
st.setdefault("lr", args.lr)
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"):
src = HOLDOUT / name
if src.exists():
(set0 / name).write_bytes(src.read_bytes())
s = base.play_stats(set0)
s.update(lora=START_LORA, board_repr="aligned_read", done=True, note="2회차 대국 재사용 (이번 시험 국면)")
st["sets"]["0"] = s
save_state(st)
write_report(st)
base.HOLD.write_text("gemma read2 loop\n", encoding="utf-8")
try:
build_test()
record_eval(st, "기준(2회차 형식)", START_LORA, TEST_OLD, OUT / "train0", "base", "aligned_read")
for k in range(1, args.rounds + 1):
wait_for_gpu()
t = st["trains"].setdefault(str(k), {})
tdir = OUT / f"train{k}"
kdata = ROOT / f"sft_data/gemma_kread2_r{k}"
sources = BASE_SOURCES + [OUT / f"set{j}" for j in range(1, k)]
if not (kdata / "summary.json").exists():
cmd = [PY, "scripts/dataset_moves_katago.py", "--out", str(kdata), "--board-repr", "aligned_read2",
"--retry-frac", "0.2", "--max-positions", "2400", "--seed", str(231 + k)]
for sd in sources:
cmd += ["--source", str(sd / "matches.sqlite")]
if (sd / "gemma4.jsonl").exists():
cmd += ["--jsonl", str(sd / "gemma4.jsonl")]
if run(cmd, tdir / "dataset.log") != 0:
return 1
new_lora = f"vlm_models/lora_gemma4_read2_r{k}"
if not t.get("trained"):
base.stop_gemma()
rng = random.Random(271 + k)
kl = jl(kdata / "train.jsonl")
rng.shuffle(kl)
lines = kl[: args.kmove_n]
lines += [x for x in lines if taken_favorites(x) >= 2]
for x in jl(COORD_KEEP):
r = json.loads(x)
lines.append(json.dumps(convert(r, "aligned_read2"), ensure_ascii=False)
if r.get("task") in MOVE_TASKS else x)
cv3 = jl(ROOT / "sft_data/coord_v3/train.jsonl")
rng.shuffle(cv3)
lines += cv3[:200]
rng.shuffle(lines)
mix = ROOT / f"sft_data/mix_gemma_read2_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)
write_report(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(kdata / "val.jsonl"), "--resume-from", st["lora"]], tdir / "train.log")
log(f"read2 train{k} rc={rc} {(time.time() - t0) / 3600:.1f}h")
if rc != 0:
return rc
(ROOT / new_lora / "serve_board_repr.txt").write_text("aligned_read2\n", encoding="utf-8")
t["trained"] = True
t["hours"] = round((time.time() - t0) / 3600, 2)
save_state(st)
new = record_eval(st, f"학습 {k}회차 후", new_lora, TEST_NEW, tdir, "new", "aligned_read2")
if "adopted" not in t:
ref = max((e for e in st["evals"].values() if e["lora"] == st["lora"]),
key=lambda e: e.get("move_legal") or 0)
t["adopted"] = better(new, ref)
if t["adopted"]:
st["lora"] = new_lora
adopt(new_lora)
log(f"read2 train{k} adopted={t['adopted']} ref={ref.get('move_legal')} new={new.get('move_legal')}")
save_state(st)
write_report(st)
set_dir = OUT / f"set{k}"
if not st["sets"].get(str(k), {}).get("done"):
wait_for_gpu()
done = base.completed_games(set_dir / "play.log")
left = args.games - done
if left > 0:
base.BOARD_REPR = repr_of(st["lora"])
base.MAX_NEW_TOKENS = 192 if base.BOARD_REPR == "aligned_read2" else 160
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"read2 set{k} play stopped rc={rc}")
return 1
s = base.play_stats(set_dir)
s.update(lora=st["lora"], board_repr=repr_of(st["lora"]), done=True)
st["sets"][str(k)] = s
save_state(st)
write_report(st)
log(f"read2 set{k} stats {s}")
finally:
if not base.port_open(PORT):
try:
base.BOARD_REPR = repr_of(st["lora"])
base.MAX_NEW_TOKENS = 192 if base.BOARD_REPR == "aligned_read2" else 160
base.start_gemma(st["lora"], ROOT / "logs" / "mariago_web_watch" / "gemma.jsonl")
except SystemExit as exc:
log(f"gemma restart failed: {exc}")
if base.HOLD.exists() and base.HOLD.read_text(encoding="utf-8").startswith("gemma read2"):
base.HOLD.unlink()
write_report(st)
log("read2 loop done")
return 0
if __name__ == "__main__":
raise SystemExit(main())