예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
train_ladder_lora.py
#!/usr/bin/env python3
"""모양(축·호구) LoRA 추가 학습 파이프라인.
1) 혼합 데이터: <shape>_v1 + (옵션) 이전 모양 데이터 일부 + 기존 수 예측(baduk_v3_aligned) + 좌표(coord_v3) — 잊어버림 방지
2) (옵션) 운영 VLM(com.mariago.vlm-serve) 일시 중지 — 메모리 확보
3) 학습 전 평가(베이스 어댑터)
4) train_lora --resume-from <base>
5) verify_coords 좌표 게이트 + 학습 후 평가
6) 운영 VLM 재기동 (어댑터 교체는 하지 않음 — 결과 보고 수동 결정)
python scripts/train_ladder_lora.py --stop-vlm # 축
python scripts/train_ladder_lora.py --shape hogu --stop-vlm \
--base vlm_models/lora_ladder_qwen_v1 --out vlm_models/lora_hogu_qwen_v1 --retain ladder_v1:300
python scripts/train_ladder_lora.py --shape onpolicy --stop-vlm --shape-n 1200 # 대국 국면 KataGo 증류
상태: logs/<shape>/TRAIN_STATUS.md
"""
from __future__ import annotations
import argparse
import json
import os
import random
import subprocess
import time
from datetime import datetime
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
PY = str(ROOT / ".venv/bin/python")
MODEL = "********"
VLM_LABEL = "com.mariago.vlm-serve"
VLM_PLIST = Path.home() / "Library/LaunchAgents" / f"{VLM_LABEL}.plist"
KINDS = {
"ladder": ("flee_dead", "flee_alive", "chase_work", "chase_broken"),
"hogu": ("peep", "entry", "fill"),
"onpolicy": ("onpolicy_move", "onpolicy_pass"),
}
def now() -> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def run(cmd: list[str], log: Path) -> int:
log.parent.mkdir(parents=True, exist_ok=True)
with log.open("a", encoding="utf-8") as f:
f.write(f"\n=== {now()} {' '.join(cmd)}\n")
f.flush()
return subprocess.run(cmd, cwd=ROOT, stdout=f, stderr=subprocess.STDOUT).returncode
def take(p: Path, n: int, rng: random.Random) -> list[str]:
lines = [x for x in p.open(encoding="utf-8") if x.strip()]
rng.shuffle(lines)
return lines[:n]
def vlm(action: str) -> None:
gui = f"gui/{os.getuid()}"
if action == "stop":
subprocess.run(["launchctl", "bootout", gui, str(VLM_PLIST)], capture_output=True)
else:
subprocess.run(["launchctl", "bootstrap", gui, str(VLM_PLIST)], capture_output=True)
subprocess.run(["launchctl", "enable", f"{gui}/{VLM_LABEL}"], capture_output=True)
def locate_acc(p: Path) -> float:
if not p.exists():
return -1.0
d = json.loads(p.read_text(encoding="utf-8"))
return float(d.get("by_kind", {}).get("locate", {}).get("acc", d.get("accuracy", -1)))
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--shape", choices=sorted(KINDS), default="ladder")
ap.add_argument("--base", default="vlm_models/lora_mix_qwen_r5")
ap.add_argument("--out", default=None, help="기본 vlm_models/lora_<shape>_qwen_v1")
ap.add_argument("--shape-n", type=int, default=900)
ap.add_argument("--retain", action="append", default=[], help="이전 모양 데이터 유지: 'ladder_v1:300' (반복 가능)")
ap.add_argument("--move-n", type=int, default=400)
ap.add_argument("--coord-n", type=int, default=200)
ap.add_argument("--lr", default="1e-5")
ap.add_argument("--eval-limit", type=int, default=120)
ap.add_argument("--skip-base-eval", action="store_true")
ap.add_argument("--stop-vlm", action="store_true", help="학습 동안 운영 VLM 중지 후 끝나면 재기동")
ap.add_argument("--model", default=MODEL)
ap.add_argument("--data-name", default="", help="sft_data 디렉터리. 비우면 <shape>_v1")
ap.add_argument("--log-name", default="", help="logs 하위 이름. 비우면 shape")
a = ap.parse_args()
os.chdir(ROOT)
shape = a.shape
data_name = a.data_name or f"{shape}_v1"
log_name = a.log_name or shape
model = a.model
out = a.out or f"vlm_models/lora_{shape}_qwen_v1"
logdir = ROOT / f"logs/{log_name}"
status_md = logdir / "TRAIN_STATUS.md"
val = f"sft_data/{data_name}/val.jsonl"
def status(lines: list[str]) -> None:
logdir.mkdir(parents=True, exist_ok=True)
status_md.write_text(f"# {shape} LoRA\n\n" + "\n".join(lines) + "\n", encoding="utf-8")
print("\n".join(lines[-3:]), flush=True)
rng = random.Random(41)
parts = [(data_name, a.shape_n)]
for r in a.retain:
name, n = r.split(":")
parts.append((name, int(n)))
lines: list[str] = []
for name, n in parts:
lines += take(ROOT / f"sft_data/{name}/train.jsonl", n, rng)
lines += take(ROOT / "sft_data/baduk_v3_aligned/train.jsonl", a.move_n, rng)
lines += take(ROOT / "sft_data/coord_v3/train.jsonl", a.coord_n, rng)
rng.shuffle(lines)
mix = ROOT / f"sft_data/mix_{data_name}/train.jsonl"
mix.parent.mkdir(parents=True, exist_ok=True)
mix.write_text("".join(lines), encoding="utf-8")
desc = " / ".join(f"{nm} {n}" for nm, n in parts) + f" / move {a.move_n} / coord {a.coord_n}"
hist = [f"- start: {now()}", f"- base: `{a.base}` → out: `{out}`",
f"- mix: {len(lines)} ({desc}) → `{mix.relative_to(ROOT)}`"]
status(hist)
def evaluate(lora: str, name: str) -> None:
run([PY, "scripts/eval_ladder_vlm.py", "--data", val, "--lora-dir", lora,
"--limit", str(a.eval_limit), "--out", f"logs/{log_name}/{name}.json"], logdir / "eval.log")
if a.stop_vlm:
vlm("stop")
hist.append(f"- {now()} 운영 VLM 중지")
status(hist)
try:
if not a.skip_base_eval:
hist.append(f"- {now()} 베이스 평가 중…")
status(hist)
evaluate(a.base, "eval_base")
hist[-1] = f"- {now()} 베이스 평가 → `logs/{shape}/eval_base.json`"
status(hist)
hist.append(f"- {now()} 학습 중 (log: logs/{log_name}/train_v1.log)")
status(hist)
t0 = time.time()
cmd = [PY, "-m", "vlm_server.train_lora", "--dataset-jsonl", str(mix), "--model", model,
"--output-dir", out, "--epochs", "1", "--batch-size", "1",
"--grad-accum", "8", "--lr", a.lr, "--lora-rank", "16", "--lora-alpha", "32",
"--max-seq-length", "4096", "--max-samples", str(len(lines)), "--grad-checkpoint",
"--val-jsonl", val]
if a.base and a.base != "-":
cmd.extend(["--resume-from", a.base])
rc = run(cmd, logdir / "train_v1.log")
hist[-1] = f"- {now()} 학습 rc={rc} ({(time.time() - t0) / 3600:.1f}h)"
status(hist)
if rc != 0:
return rc
summ = logdir / "coord_v1_summary.json"
run(
[PY, "-m", "vlm_server.verify_coords", "--model", model, "--lora-dir", out,
"--board-format", "aligned", "--trials", "48",
"--out-jsonl", str(logdir / "coord_v1.jsonl"), "--out-summary", str(summ)],
logdir / "coord.log",
)
hist.append(f"- {now()} 좌표 locate **{locate_acc(summ)}%** (게이트 90%)")
status(hist)
evaluate(out, "eval_v1")
hist.append(f"- {now()} 학습 후 평가 → `logs/{log_name}/eval_v1.json`")
for name in ("eval_base", "eval_v1"):
p = logdir / f"{name}.json"
if p.exists():
s = json.loads(p.read_text(encoding="utf-8"))["summary"]
row = ", ".join(
f"{k}: " + "/".join(f"{m}={v}" for m, v in s[k].items() if m != "n")
for k in KINDS[shape] if k in s
)
hist.append(f" - {name}: {row}")
status(hist)
finally:
if a.stop_vlm:
vlm("start")
hist.append(f"- {now()} 운영 VLM 재기동 (어댑터 그대로)")
status(hist)
hist.append(f"- done: {now()}")
status(hist)
return 0
if __name__ == "__main__":
raise SystemExit(main())