예제 소스
SGF 기보로 학습하는 바둑 LLM 인공지능 · 파일 169개
daemon_day_pipeline.py
#!/usr/bin/env python3
"""Day pipeline until 19:00 KST — train ↔ coord verify loop (no confirmation).
Phases:
1) Re-verify Qwen base with left-aligned board (col-1 fix)
2) Build coord_v3 SFT + train lora_coord_qwen
3) Verify coord LoRA (gate locate>=90)
4) If pass: mix coord+move short train, verify again; optionally longer move mix
5) Repeat improvements until 19:00 or success marker
Status: logs/vlm_eval_day/DAY_STATUS.md
Log: logs/vlm_eval_day/day_pipeline.log
"""
from __future__ import annotations
import json
import os
import subprocess
import time
from pathlib import Path
ROOT = Path("~/project")
PY = str(ROOT / ".venv" / "bin" / "python")
LOG = ROOT / "logs" / "vlm_eval_day" / "day_pipeline.log"
STATUS = ROOT / "logs" / "vlm_eval_day" / "DAY_STATUS.md"
DEADLINE_H, DEADLINE_M = 19, 0
LOCATE_GATE = 90.0
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_H, DEADLINE_M)
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(body: str) -> None:
STATUS.write_text(body, 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 mix_jsonl(paths: list[Path], out: Path, max_lines: int, seed: int = 1) -> int:
import random
rng = random.Random(seed)
lines: list[str] = []
for p in paths:
if not p.exists():
continue
with p.open(encoding="utf-8") as f:
lines.extend(x for x in f if x.strip())
rng.shuffle(lines)
lines = lines[:max_lines]
out.parent.mkdir(parents=True, exist_ok=True)
with out.open("w", encoding="utf-8") as f:
f.writelines(lines)
return len(lines)
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)
# stop leftover overnight
subprocess.run(["pkill", "-f", "daemon_overnight_pipeline"], capture_output=True)
write_status(f"# Day pipeline\n\n- started: {now()}\n- deadline: 19:00\n- phase: start\n")
log("=== DAY PIPELINE START (until 19:00) ===")
# --- Phase 1: base re-verify with left-align fix ---
if past_deadline():
log("already past deadline")
os._exit(0)
write_status(f"# Day pipeline\n\n- {now()} phase: verify_base_leftalign\n")
base_sum = ROOT / "logs/vlm_eval_day/coord_qwen_base_left_summary.json"
run(
[
PY,
"-m",
"vlm_server.verify_coords",
"--model",
"********",
"--board-format",
"aligned",
"--trials",
"48",
"--out-jsonl",
str(ROOT / "logs/vlm_eval_day/coord_qwen_base_left.jsonl"),
"--out-summary",
str(base_sum),
],
ROOT / "logs/vlm_eval_day/coord_verify_day.log",
)
acc_base = locate_acc(base_sum)
log(f"base locate after left-align fix: {acc_base}%")
# --- Phase 2: coord dataset + train ---
if past_deadline():
os._exit(0)
write_status(
f"# Day pipeline\n\n- base locate: {acc_base}%\n- {now()} phase: build_coord_dataset\n"
)
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",
)
if past_deadline():
os._exit(0)
write_status(f"# Day pipeline\n\n- {now()} phase: train_coord_lora\n")
run(
[
PY,
"-m",
"vlm_server.train_lora",
"--dataset-jsonl",
"sft_data/coord_v3/train.jsonl",
"--val-jsonl",
"sft_data/coord_v3/val.jsonl",
"--model",
"********",
"--output-dir",
"vlm_models/lora_coord_qwen",
"--epochs",
"1",
"--batch-size",
"1",
"--grad-accum",
"8",
"--lr",
"5e-5",
"--lora-rank",
"16",
"--lora-alpha",
"32",
"--max-seq-length",
"4096",
"--max-samples",
"3000",
"--grad-checkpoint",
],
ROOT / "logs/vlm_eval_day/coord_lora_train.log",
)
# --- Phase 3: verify coord LoRA ---
if past_deadline():
os._exit(0)
write_status(f"# Day pipeline\n\n- {now()} phase: verify_coord_lora\n")
coord_sum = ROOT / "logs/vlm_eval_day/coord_lora_qwen_summary.json"
run(
[
PY,
"-m",
"vlm_server.verify_coords",
"--model",
"********",
"--lora-dir",
"vlm_models/lora_coord_qwen",
"--board-format",
"aligned",
"--trials",
"64",
"--out-jsonl",
str(ROOT / "logs/vlm_eval_day/coord_lora_qwen.jsonl"),
"--out-summary",
str(coord_sum),
],
ROOT / "logs/vlm_eval_day/coord_verify_day.log",
)
acc_coord = locate_acc(coord_sum)
log(f"coord LoRA locate: {acc_coord}% (gate {LOCATE_GATE})")
write_status(
"\n".join(
[
"# Day pipeline",
f"- updated: {now()}",
f"- base locate (left-align): **{acc_base}%**",
f"- coord LoRA locate: **{acc_coord}%**",
f"- gate: {'PASS' if acc_coord >= LOCATE_GATE else 'FAIL'}",
"",
]
)
)
if acc_coord < LOCATE_GATE:
# one more round: more hard samples + lower lr resume
if past_deadline():
os._exit(3)
log("coord gate fail — extra hard-focused train")
run(
[
PY,
"-m",
"vlm_server.dataset_coords",
"--out-dir",
"sft_data/coord_v3_hard",
"--n-samples",
"4000",
"--seed",
"99",
],
ROOT / "logs/vlm_eval_day/coord_dataset_build.log",
)
run(
[
PY,
"-m",
"vlm_server.train_lora",
"--dataset-jsonl",
"sft_data/coord_v3_hard/train.jsonl",
"--model",
"********",
"--output-dir",
"vlm_models/lora_coord_qwen",
"--resume-from",
"vlm_models/lora_coord_qwen/adapters.safetensors",
"--epochs",
"1",
"--batch-size",
"1",
"--grad-accum",
"8",
"--lr",
"2e-5",
"--lora-rank",
"16",
"--lora-alpha",
"32",
"--max-seq-length",
"4096",
"--max-samples",
"2500",
"--grad-checkpoint",
],
ROOT / "logs/vlm_eval_day/coord_lora_train.log",
)
run(
[
PY,
"-m",
"vlm_server.verify_coords",
"--model",
"********",
"--lora-dir",
"vlm_models/lora_coord_qwen",
"--board-format",
"aligned",
"--trials",
"64",
"--out-jsonl",
str(ROOT / "logs/vlm_eval_day/coord_lora_qwen_r2.jsonl"),
"--out-summary",
str(ROOT / "logs/vlm_eval_day/coord_lora_qwen_r2_summary.json"),
],
ROOT / "logs/vlm_eval_day/coord_verify_day.log",
)
acc_coord = locate_acc(ROOT / "logs/vlm_eval_day/coord_lora_qwen_r2_summary.json")
log(f"coord LoRA r2 locate: {acc_coord}%")
if acc_coord < LOCATE_GATE:
write_status(
f"# Day STOP — coord gate fail\n\n- {now()}\n- coord locate {acc_coord}% < {LOCATE_GATE}\n"
"- skipped move training (would collapse coords like smoke LoRA)\n"
)
log("STOP: coord gate still failing — no move train")
os._exit(3)
# --- Phase 4: careful mix train (coord-heavy) ---
round_i = 0
while not past_deadline():
round_i += 1
write_status(
f"# Day pipeline\n\n- coord gate PASS ({acc_coord}%)\n"
f"- {now()} phase: mix_train_r{round_i}\n"
)
mix_dir = ROOT / f"sft_data/mix_coord_move_r{round_i}"
mix_train = mix_dir / "train.jsonl"
# 60% coord + 40% move, capped samples growing
n_coord = 2500
n_move = 1500 + 500 * (round_i - 1)
# take lines
import random
rng = random.Random(100 + round_i)
lines = []
with (ROOT / "sft_data/coord_v3/train.jsonl").open(encoding="utf-8") as f:
cl = [x for x in f if x.strip()]
rng.shuffle(cl)
lines.extend(cl[:n_coord])
with (ROOT / "sft_data/baduk_v3/train.jsonl").open(encoding="utf-8") as f:
ml = [x for x in f if x.strip()]
rng.shuffle(ml)
lines.extend(ml[:n_move])
rng.shuffle(lines)
mix_dir.mkdir(parents=True, exist_ok=True)
with mix_train.open("w", encoding="utf-8") as f:
f.writelines(lines)
log(f"mix r{round_i}: {len(lines)} samples (coord~{n_coord} move~{n_move})")
out_dir = ROOT / f"vlm_models/lora_mix_qwen_r{round_i}"
resume = str(ROOT / "vlm_models/lora_coord_qwen/adapters.safetensors")
if round_i > 1:
prev = ROOT / f"vlm_models/lora_mix_qwen_r{round_i-1}/adapters.safetensors"
if prev.exists():
resume = str(prev)
run(
[
PY,
"-m",
"vlm_server.train_lora",
"--dataset-jsonl",
str(mix_train),
"--model",
"********",
"--output-dir",
str(out_dir),
"--resume-from",
resume,
"--epochs",
"1",
"--batch-size",
"1",
"--grad-accum",
"8",
"--lr",
"2e-5",
"--lora-rank",
"16",
"--lora-alpha",
"32",
"--max-seq-length",
"4096",
"--max-samples",
str(min(4000, len(lines))),
"--grad-checkpoint",
],
ROOT / f"logs/vlm_eval_day/mix_train_r{round_i}.log",
)
if past_deadline():
break
mix_sum = ROOT / f"logs/vlm_eval_day/coord_mix_r{round_i}_summary.json"
run(
[
PY,
"-m",
"vlm_server.verify_coords",
"--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(mix_sum),
],
ROOT / "logs/vlm_eval_day/coord_verify_day.log",
)
acc_mix = locate_acc(mix_sum)
log(f"mix r{round_i} locate: {acc_mix}%")
write_status(
"\n".join(
[
"# Day pipeline",
f"- updated: {now()}",
f"- base locate: {acc_base}%",
f"- coord LoRA: {acc_coord}%",
f"- mix r{round_i} locate: **{acc_mix}%**",
f"- out: `{out_dir}`",
"",
"If locate stays >=90, evening can serve with this LoRA.",
"",
]
)
)
if acc_mix < LOCATE_GATE:
log(f"mix r{round_i} collapsed coords — stop mixing, keep coord LoRA")
write_status(
f"# Day STOP — mix collapsed coords\n\n- mix r{round_i} locate {acc_mix}%\n"
f"- keep using `vlm_models/lora_coord_qwen`\n- {now()}\n"
)
break
# success continue until deadline
if round_i >= 3:
log("completed 3 mix rounds with coords intact")
break
write_status(
(STATUS.read_text(encoding="utf-8") if STATUS.exists() else "")
+ f"\n\n## finished\n- {now()}\n- see day_pipeline.log\n"
)
log("=== DAY PIPELINE END ===")
os._exit(0)
if __name__ == "__main__":
raise SystemExit(main())