예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
run_gemma_read_loop.py
#!/usr/bin/env python3
"""Gemma 4 '읽고 고르기' 루프: 판을 안 보고 정해진 자리만 찍는 문제 교정.
답 형식을 {"row": r, "line": "<그 행 19칸 열번호+기호>", "col": c} 로 바꿔
행을 정한 뒤 그 행 내용을 옮겨 적고, 적은 줄에서 '.' 인 열을 고르게 한다
(서버 --board-repr aligned_read).
시험은 겜마 자신의 대국(수 품질 루프 set2, move_r1 10판) 국면으로만 한다. 그 판에서는
좋아하는 자리가 이미 자기 돌로 차 있어서 실전 첫 시도 합법률과 맞는다. 이 대국은 학습에 넣지 않는다.
세트 0 은 수 품질 루프 set2 를 그대로 쓴다. 회차마다
1) 대국 백 국면을 KataGo 최선수로 라벨 (읽고 고르기 형식, 20%는 금지 좌표 재시도형)
2) 실전 좌표 데이터(gemma_coordkeep_r2, 수 문항은 새 형식으로 변환) + coord_v3 200
3) 이어 학습 → 자기 대국 시험으로 전/후 비교 → 채택
4) 휴리스틱과 10판
python scripts/run_gemma_read_loop.py
상태: logs/gemma_read_loop_20261014/REPORT.md
"""
from __future__ import annotations
import argparse
import json
import os
import random
import re
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
from vlm_server.dataset_baduk import assistant_text, read_pick_answer, read_pick_answer2 # noqa: E402
from vlm_server.server import ( # noqa: E402
SYSTEM_PROMPT_ALIGNED,
SYSTEM_PROMPT_ALIGNED_READ,
SYSTEM_PROMPT_ALIGNED_READ2,
build_prompt,
)
OUT = ROOT / "logs" / "gemma_read_loop_20261014"
base.OUT = OUT
base.MAX_NEW_TOKENS = 160
PY, MODEL, PORT = base.PY, base.MODEL, base.PORT
log, run = base.log, base.run
START_LORA = "vlm_models/lora_gemma4_move_r1"
MOVE_LOOP = ROOT / "logs" / "gemma_move_loop_20261012"
COORD_LOOP = ROOT / "logs" / "gemma_coord_loop_20261011"
HOLDOUT = MOVE_LOOP / "set2"
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"]
TEST_READ = ROOT / "sft_data" / "gemma_read_test" / "val.jsonl"
TEST_ALIGNED = ROOT / "sft_data" / "gemma_read_test" / "val_aligned.jsonl"
COORD_VAL = ROOT / "sft_data" / "gemma_coord_r2" / "val.jsonl"
COORD_KEEP = ROOT / "sft_data" / "gemma_coordkeep_r2" / "train.jsonl"
MOVE_TASKS = ("move", "move_retry", "kmove", "kmove_retry")
_HINT_RE = re.compile(r"같은 실수를 반복하지 마세요:\n- (.+)\n")
def repr_of(lora: str) -> str:
p = ROOT / lora / "serve_board_repr.txt"
return p.read_text(encoding="utf-8").strip() if p.exists() else "aligned"
def convert(rec: dict, board_repr: str) -> dict:
"""수 문항을 다른 보드 프롬프트 형식으로 (금지 좌표·힌트 유지)."""
exp = rec["expect"]
cells = bytes(".XO".index(ch) for ch in rec["board"])
m = _HINT_RE.search(rec["usr"])
forbidden = [tuple(p) for p in exp.get("forbidden") or []]
user = build_prompt(cells, int(exp["side"]), forbidden=forbidden or None,
illegal_hint=m.group(1) if m else None, board_repr=board_repr)
system = {
"aligned_read": SYSTEM_PROMPT_ALIGNED_READ,
"aligned_read2": SYSTEM_PROMPT_ALIGNED_READ2,
}.get(board_repr, SYSTEM_PROMPT_ALIGNED)
if board_repr == "aligned_read2":
answer = read_pick_answer2(cells, exp["row"], exp["col"])
elif board_repr == "aligned_read":
answer = read_pick_answer(cells, exp["row"], exp["col"])
else:
answer = assistant_text((exp["row"], exp["col"]))
out = dict(rec)
out.update(sys=system, usr=user, messages=[
{"role": "user", "content": [{"type": "text", "text": f"{system}\n\n{user}"}]},
{"role": "assistant", "content": [{"type": "text", "text": answer}]},
])
return out
def jl(path: Path) -> list[str]:
return [x for x in path.read_text(encoding="utf-8").splitlines() if x.strip()]
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, 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")
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 읽고 고르기 루프 (행 → 그 행 내용 → 열)\n",
f"- 갱신: {datetime.now():%Y-%m-%d %H:%M}",
f"- 시작 어댑터: `{START_LORA}` · 현재: `{st.get('lora', START_LORA)}`",
"- 시험: 겜마 자기 대국(수 품질 루프 set2) 국면 — 학습에 넣지 않음",
"",
"## 대국 (휴리스틱 흑 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', 'aligned')} | {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## 시험 (자기 대국 국면)\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', '-')} | {e.get('consistent', '-')} | "
f"{e.get('kata_top3')} | {e.get('read_acc')} | {e.get('col_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_ALIGNED.exists():
return
cmd = [PY, "scripts/dataset_moves_katago.py", "--out", str(TEST_READ.parent), "--board-repr", "aligned_read",
"--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", "101"]
if run(cmd, OUT / "dataset.log") != 0:
raise SystemExit("test dataset failed")
recs = [json.loads(x) for x in jl(TEST_READ)]
TEST_ALIGNED.write_text("".join(json.dumps(convert(r, "aligned"), 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_read, --max-new-tokens 160). 대국 때는 --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()))
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"):
shutil.copy2(HOLDOUT / name, set0 / name)
s = base.play_stats(set0)
s.update(lora=START_LORA, board_repr="aligned", done=True, note="수 품질 루프 set2 재사용 (시험 국면 출처)")
st["sets"]["0"] = s
save_state(st)
write_report(st)
base.HOLD.write_text("gemma read loop\n", encoding="utf-8")
try:
build_test()
tdir0 = OUT / "train0"
record_eval(st, "기준(이전 형식)", START_LORA, TEST_ALIGNED, tdir0, "base_aligned", "aligned")
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_kread_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_read",
"--retry-frac", "0.2", "--max-positions", "2400", "--seed", str(131 + 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_read_r{k}"
if not t.get("trained"):
base.stop_gemma()
rng = random.Random(171 + k)
kl = jl(kdata / "train.jsonl")
rng.shuffle(kl)
lines = kl[: args.kmove_n]
for x in jl(COORD_KEEP):
r = json.loads(x)
lines.append(json.dumps(convert(r, "aligned_read"), 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_read_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"read 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_read\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_READ, tdir, "new", "aligned_read")
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"read 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"):
done = base.completed_games(set_dir / "play.log")
left = args.games - done
if left > 0:
base.BOARD_REPR = repr_of(st["lora"])
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"read 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"read set{k} stats {s}")
finally:
if not base.port_open(PORT):
try:
base.BOARD_REPR = repr_of(st["lora"])
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():
base.HOLD.unlink()
write_report(st)
log("read loop done")
return 0
if __name__ == "__main__":
raise SystemExit(main())