예제 소스

daemon_overnight_pipeline.py

#!/usr/bin/env python3
"""Overnight pipeline (no user confirmation):

1) Wait for Qwen smoke LoRA train to finish
2) Coordinate grounding verify: Qwen base, then smoke LoRA
3) If base locate accuracy >= 90%% → start full baduk_v3 Qwen LoRA train
4) Otherwise write FAIL note and stop (do not waste a night on broken coords)

Logs: logs/vlm_eval_day/overnight.log
Status: logs/vlm_eval_day/OVERNIGHT_STATUS.md
"""
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" / "overnight.log"
STATUS = ROOT / "logs" / "vlm_eval_day" / "OVERNIGHT_STATUS.md"
SMOKE_ADAPTER = ROOT / "vlm_models" / "lora_baduk_v3_qwen_smoke" / "adapters.safetensors"
FULL_OUT = ROOT / "vlm_models" / "lora_baduk_v3_qwen"
LOCATE_GATE = 90.0


def log(msg: str) -> None:
    line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {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 pgrep(pattern: str) -> bool:
    r = subprocess.run(["pgrep", "-f", pattern], capture_output=True)
    return r.returncode == 0


def wait_until(pred, label: str, timeout_s: float, poll_s: float = 20.0) -> bool:
    t0 = time.time()
    while time.time() - t0 < timeout_s:
        if pred():
            return True
        log(f"waiting: {label} ({int(time.time() - t0)}s)")
        time.sleep(poll_s)
    return pred()


def smoke_done() -> bool:
    if pgrep(r"mlx_vlm\.lora.*lora_baduk_v3_qwen_smoke"):
        return False
    if pgrep(r"train_lora.*lora_baduk_v3_qwen_smoke"):
        return False
    return SMOKE_ADAPTER.exists() and SMOKE_ADAPTER.stat().st_size > 1_000_000


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===== {time.strftime('%Y-%m-%d %H:%M:%S')} =====\n")
        f.flush()
        p = subprocess.run(cmd, cwd=str(ROOT), env=env, stdout=f, stderr=subprocess.STDOUT)
    log(f"EXIT {p.returncode}: {' '.join(cmd[:6])}...")
    return int(p.returncode)


def locate_acc(summary_path: Path) -> float:
    if not summary_path.exists():
        return -1.0
    data = json.loads(summary_path.read_text(encoding="utf-8"))
    loc = data.get("by_kind", {}).get("locate", {})
    return float(loc.get("acc", data.get("accuracy", -1.0)))


def main() -> int:
    # double-fork
    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)

    write_status(
        f"# Overnight status\n\n- started: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
        "- phase: waiting_smoke\n"
    )
    log("=== OVERNIGHT PIPELINE START ===")

    # Kill older coord-only waiter if still around (avoid GPU fight later)
    subprocess.run(["pkill", "-f", "daemon_coord_verify_after_smoke.py"], capture_output=True)

    if not wait_until(smoke_done, "smoke train finish", timeout_s=4 * 3600):
        log("FAIL: smoke train timeout")
        write_status("# Overnight FAIL\n\nSmoke train did not finish in time.\n")
        os._exit(1)
    log("smoke train finished")
    write_status(
        f"# Overnight status\n\n- smoke: DONE @ {time.strftime('%H:%M:%S')}\n"
        "- phase: coord_verify\n"
    )

    # brief pause so Metal releases
    time.sleep(10)

    base_sum = ROOT / "logs" / "vlm_eval_day" / "coord_qwen_base_summary.json"
    smoke_sum = ROOT / "logs" / "vlm_eval_day" / "coord_qwen_smoke_summary.json"
    coord_log = ROOT / "logs" / "vlm_eval_day" / "coord_verify_run.log"

    rc_base = run(
        [
            PY,
            "-m",
            "vlm_server.verify_coords",
            "--model",
            "********",
            "--board-format",
            "aligned",
            "--trials",
            "48",
            "--out-jsonl",
            str(ROOT / "logs" / "vlm_eval_day" / "coord_qwen_base.jsonl"),
            "--out-summary",
            str(base_sum),
        ],
        coord_log,
    )
    time.sleep(5)
    rc_smoke = run(
        [
            PY,
            "-m",
            "vlm_server.verify_coords",
            "--model",
            "********",
            "--lora-dir",
            "********",
            "--board-format",
            "aligned",
            "--trials",
            "48",
            "--out-jsonl",
            str(ROOT / "logs" / "vlm_eval_day" / "coord_qwen_smoke.jsonl"),
            "--out-summary",
            str(smoke_sum),
        ],
        coord_log,
    )

    acc_base = locate_acc(base_sum)
    acc_smoke = locate_acc(smoke_sum)
    log(f"coord locate acc: base={acc_base} smoke={acc_smoke} (gate>={LOCATE_GATE})")

    gate_ok = acc_base >= LOCATE_GATE
    write_status(
        "\n".join(
            [
                "# Overnight status",
                "",
                f"- updated: {time.strftime('%Y-%m-%d %H:%M:%S')}",
                f"- smoke train: DONE",
                f"- coord base locate: **{acc_base}%** (rc={rc_base})",
                f"- coord smoke locate: **{acc_smoke}%** (rc={rc_smoke})",
                f"- gate (>= {LOCATE_GATE}% on **base**): {'PASS' if gate_ok else 'FAIL'}",
                f"- phase: {'full_train' if gate_ok else 'stopped'}",
                "",
                "Details: `coord_qwen_base_summary.json`, `coord_qwen_smoke_summary.json`",
                "Logs: `overnight.log`, `coord_verify_run.log`, `qwen_full_train.log`",
                "",
            ]
        )
    )

    if not gate_ok:
        log("GATE FAIL — skip full train")
        os._exit(3)

    log("GATE PASS — starting full Qwen LoRA train on baduk_v3")
    FULL_OUT.mkdir(parents=True, exist_ok=True)
    full_log = ROOT / "logs" / "vlm_eval_day" / "qwen_full_train.log"
    # Do not block forever in this process: launch train as child and wait
    rc_full = run(
        [
            PY,
            "-m",
            "vlm_server.train_lora",
            "--dataset-jsonl",
            "sft_data/baduk_v3/train.jsonl",
            "--val-jsonl",
            "sft_data/baduk_v3/val.jsonl",
            "--model",
            "********",
            "--output-dir",
            "vlm_models/lora_baduk_v3_qwen",
            "--epochs",
            "2",
            "--batch-size",
            "1",
            "--grad-accum",
            "16",
            "--lr",
            "5e-5",
            "--lora-rank",
            "16",
            "--lora-alpha",
            "32",
            "--max-seq-length",
            "4096",
            "--grad-checkpoint",
        ],
        full_log,
    )
    log(f"full train exit={rc_full}")
    write_status(
        "\n".join(
            [
                "# Overnight DONE",
                "",
                f"- finished: {time.strftime('%Y-%m-%d %H:%M:%S')}",
                f"- coord base locate: {acc_base}%",
                f"- coord smoke locate: {acc_smoke}%",
                f"- full train exit: {rc_full}",
                f"- adapter: `vlm_models/lora_baduk_v3_qwen/`",
                "",
                "Morning: check OVERNIGHT_STATUS.md + qwen_full_train.log",
                "",
            ]
        )
    )
    os._exit(0 if rc_full == 0 else 4)


if __name__ == "__main__":
    raise SystemExit(main())