예제 소스
바둑 on-policy 학습: 대국 로그를 KataGo 라벨로 바꾸기 · 파일 169개
daemon_mix_pipeline.py
#!/usr/bin/env python3
"""혼합(좌표+수) 학습 ↔ 좌표 검증 루프 — 19:00까지.
시작점: vlm_models/lora_coord_qwen (locate 100% 통과본)
각 라운드:
1) coord_v3 + baduk_v3_aligned 혼합 jsonl 생성 (좌표 비중 점진 감소)
2) 직전 통과 어댑터에서 resume 학습 (낮은 lr, 제한 iters)
3) verify_coords — locate < 90% 이면 롤백·좌표 비중 상향 후 재시도
4) 통과면 다음 라운드 (수 비중↑)
상태: logs/vlm_eval_day/MIX_STATUS.md
로그: logs/vlm_eval_day/mix_pipeline.log
"""
from __future__ import annotations
import json
import os
import random
import subprocess
import time
from pathlib import Path
ROOT = Path("~/project")
PY = str(ROOT / ".venv" / "bin" / "python")
LOG = ROOT / "logs" / "vlm_eval_day" / "mix_pipeline.log"
STATUS = ROOT / "logs" / "vlm_eval_day" / "MIX_STATUS.md"
COORD_JSONL = ROOT / "sft_data" / "coord_v3" / "train.jsonl"
MOVE_JSONL = ROOT / "sft_data" / "baduk_v3_aligned" / "train.jsonl"
BASE_ADAPTER = ROOT / "vlm_models" / "lora_coord_qwen"
DEADLINE = (19, 0)
LOCATE_GATE = 90.0
MODEL = "********"
# round plans: (coord_n, move_n, max_samples, lr)
# 시간: max_samples≈1000 → ~1h @0.25it/s
ROUNDS = [
(2000, 800, 1000, "2e-5"),
(1800, 1200, 1000, "2e-5"),
(1500, 1500, 1200, "1.5e-5"),
(1200, 1800, 1200, "1.5e-5"),
(1000, 2000, 1200, "1e-5"),
(1000, 2500, 1500, "1e-5"),
(800, 2800, 1500, "1e-5"),
(800, 3000, 1500, "8e-6"),
]
def now() -> str:
return time.strftime("%Y-%m-%d %H:%M:%S")
def past_deadline() -> bool:
t = time.localtime()
return (t.tm_hour, t.tm_min) >= DEADLINE
def log(msg: str) -> None:
line = f"[{now()}] {msg}"
print(line, flush=True)
LOG.parent.mkdir(parents=True, exist_ok=True)
with LOG.open("a", encoding="utf-8") as f:
f.write(line + "\n")
def write_status(md: str) -> None:
STATUS.write_text(md, encoding="utf-8")
def run(cmd: list[str], log_path: Path) -> int:
log("RUN " + " ".join(cmd))
env = os.environ.copy()
env["HF_HOME"] = str(ROOT / "vlm_models")
env["HF_HUB_CACHE"] = str(ROOT / "vlm_models")
env["TOKENIZERS_PARALLELISM"] = "********"
env["PYTHONUNBUFFERED"] = "1"
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a", encoding="utf-8") as f:
f.write(f"\n===== {now()} =====\n")
f.flush()
p = subprocess.run(cmd, cwd=str(ROOT), env=env, stdout=f, stderr=subprocess.STDOUT)
log(f"EXIT {p.returncode}")
return int(p.returncode)
def locate_acc(summary: Path) -> float:
if not summary.exists():
return -1.0
d = json.loads(summary.read_text(encoding="utf-8"))
return float(d.get("by_kind", {}).get("locate", {}).get("acc", d.get("accuracy", -1)))
def build_mix(coord_n: int, move_n: int, out_path: Path, seed: int) -> int:
rng = random.Random(seed)
lines: list[str] = []
with COORD_JSONL.open(encoding="utf-8") as f:
cl = [x for x in f if x.strip()]
with MOVE_JSONL.open(encoding="utf-8") as f:
ml = [x for x in f if x.strip()]
rng.shuffle(cl)
rng.shuffle(ml)
lines.extend(cl[: min(coord_n, len(cl))])
lines.extend(ml[: min(move_n, len(ml))])
rng.shuffle(lines)
out_path.parent.mkdir(parents=True, exist_ok=True)
with out_path.open("w", encoding="utf-8") as f:
f.writelines(lines)
return len(lines)
def ensure_move_dataset() -> None:
if MOVE_JSONL.exists() and MOVE_JSONL.stat().st_size > 1_000_000:
# check board_format in first line / dataset_info
info = MOVE_JSONL.parent / "dataset_info.json"
if info.exists():
d = json.loads(info.read_text(encoding="utf-8"))
if d.get("board_format") == "aligned" and d.get("text_only"):
log("move dataset ready (aligned)")
return
log("rebuilding baduk_v3_aligned (left-aligned text-only)...")
run(
[
PY,
"-m",
"vlm_server.dataset_baduk",
"--sgf-dir",
"sgf_data",
"--out-dir",
"sft_data/baduk_v3_aligned",
"--max-games",
"4000",
"--moves-per-game",
"20",
"--board-format",
"aligned",
"--text-only",
"--no-render-png",
"--winner-only",
"--tactical-bias",
"--dedup",
"--val-split",
"0.05",
"--shuffle",
],
ROOT / "logs/vlm_eval_day/baduk_v3_aligned_build.log",
)
def ensure_coord_dataset() -> None:
if COORD_JSONL.exists() and sum(1 for _ in COORD_JSONL.open()) > 1000:
log("coord dataset ready")
return
run(
[
PY,
"-m",
"vlm_server.dataset_coords",
"--out-dir",
"sft_data/coord_v3",
"--n-samples",
"6000",
],
ROOT / "logs/vlm_eval_day/coord_dataset_build.log",
)
def main() -> int:
if os.fork() > 0:
return 0
os.setsid()
if os.fork() > 0:
os._exit(0)
os.chdir(ROOT)
os.umask(0)
with LOG.open("a", encoding="utf-8") as f:
os.dup2(f.fileno(), 1)
os.dup2(f.fileno(), 2)
dn = os.open(os.devnull, os.O_RDONLY)
os.dup2(dn, 0)
os.close(dn)
subprocess.run(["pkill", "-f", "daemon_day_pipeline"], capture_output=True)
subprocess.run(["pkill", "-f", "daemon_overnight"], capture_output=True)
write_status(
f"# Mix pipeline\n\n- started: {now()}\n- deadline: 19:00\n"
f"- base adapter: `lora_coord_qwen`\n- phase: init\n"
)
log("=== MIX TRAIN↔VERIFY PIPELINE START (until 19:00) ===")
if not (BASE_ADAPTER / "adapters.safetensors").exists():
log("FATAL: lora_coord_qwen missing")
os._exit(2)
ensure_coord_dataset()
ensure_move_dataset()
good_adapter = str(BASE_ADAPTER)
history: list[str] = []
round_i = 0
fail_streak = 0
while not past_deadline() and round_i < len(ROUNDS):
coord_n, move_n, max_samples, lr = ROUNDS[round_i]
# on fail streak, boost coord ratio
if fail_streak > 0:
coord_n = int(coord_n * 1.3)
move_n = max(400, int(move_n * 0.6))
lr = "1e-5"
round_i += 1
out_dir = ROOT / f"vlm_models/lora_mix_qwen_r{round_i}"
mix_path = ROOT / f"sft_data/mix_r{round_i}/train.jsonl"
n = build_mix(coord_n, move_n, mix_path, seed=200 + round_i)
log(
f"R{round_i}: mix lines={n} plan_coord={coord_n} plan_move={move_n} "
f"max_samples={max_samples} lr={lr} resume={good_adapter}"
)
write_status(
"\n".join(
[
"# Mix pipeline",
f"- updated: {now()}",
f"- round: **{round_i}** / train",
f"- mix: coord~{coord_n} + move~{move_n} (file {n} lines)",
f"- resume: `{good_adapter}`",
f"- lr: {lr}, max_samples: {max_samples}",
"",
*history[-12:],
"",
]
)
)
if past_deadline():
break
rc = run(
[
PY,
"-m",
"vlm_server.train_lora",
"--dataset-jsonl",
str(mix_path),
"--model",
MODEL,
"--output-dir",
str(out_dir),
"--resume-from",
good_adapter,
"--epochs",
"1",
"--batch-size",
"1",
"--grad-accum",
"8",
"--lr",
lr,
"--lora-rank",
"16",
"--lora-alpha",
"32",
"--max-seq-length",
"4096",
"--max-samples",
str(max_samples),
"--grad-checkpoint",
],
ROOT / f"logs/vlm_eval_day/mix_train_r{round_i}.log",
)
if rc != 0:
history.append(f"- R{round_i} train FAILED rc={rc}")
fail_streak += 1
continue
if past_deadline():
# still try a quick verify if time
pass
sum_path = ROOT / f"logs/vlm_eval_day/coord_mix_r{round_i}_summary.json"
run(
[
PY,
"-m",
"vlm_server.verify_coords",
"--model",
MODEL,
"--lora-dir",
str(out_dir),
"--board-format",
"aligned",
"--trials",
"48",
"--out-jsonl",
str(ROOT / f"logs/vlm_eval_day/coord_mix_r{round_i}.jsonl"),
"--out-summary",
str(sum_path),
],
ROOT / "logs/vlm_eval_day/coord_verify_mix.log",
)
acc = locate_acc(sum_path)
log(f"R{round_i} locate={acc}% gate={LOCATE_GATE}")
if acc >= LOCATE_GATE:
good_adapter = str(out_dir)
fail_streak = 0
history.append(f"- R{round_i} PASS locate **{acc}%** → keep `{out_dir.name}`")
else:
fail_streak += 1
history.append(
f"- R{round_i} FAIL locate **{acc}%** → rollback to `{Path(good_adapter).name}`"
)
if fail_streak >= 3:
history.append("- fail_streak>=3 → stop (protect coord LoRA)")
log("too many fails, stopping")
break
write_status(
"\n".join(
[
"# Mix pipeline",
f"- updated: {now()}",
f"- last round: {round_i}",
f"- last locate: **{acc}%**",
f"- best adapter: `{good_adapter}`",
f"- fail_streak: {fail_streak}",
"",
"## history",
*history,
"",
"Logs: `mix_pipeline.log`, `mix_train_r*.log`, `coord_verify_mix.log`",
"",
]
)
)
write_status(
(STATUS.read_text(encoding="utf-8") if STATUS.exists() else "")
+ f"\n\n## finished\n- {now()}\n- best: `{good_adapter}`\n"
)
log(f"=== MIX PIPELINE END best={good_adapter} ===")
os._exit(0)
if __name__ == "__main__":
raise SystemExit(main())