LLM Ops가 실제로 하는 일 — 재현, 오염, 체크포인트, 승격, 그리고 롤백

# run_manifest.py — 학습 시작 시 랭크 0에서 한 번 호출합니다
import hashlib
import importlib.metadata as md
import json
import os
import platform
import subprocess
import time


def _sh(cmd: str) -> str:
    try:
        return subprocess.check_output(cmd, shell=True, text=True,
                                       stderr=subprocess.DEVNULL).strip()
    except Exception:
        return "unknown"


def build_manifest(config: dict, data_snapshot_id: str) -> dict:
    cfg_bytes = json.dumps(config, sort_keys=True).encode()
    return {
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "code": {
            "commit": _sh("git rev-parse HEAD"),
            "dirty": _sh("git status --porcelain") != "",
            "remote": _sh("git config --get remote.origin.url"),
        },
        "config": {
            "sha256": hashlib.sha256(cfg_bytes).hexdigest()[:16],
            "values": config,
        },
        "data": {
            "snapshot_id": data_snapshot_id,      # 불변 스냅숏 식별자
            "manifest_sha256": _sh(f"sha256sum data/manifests/{data_snapshot_id}.jsonl | cut -d' ' -f1"),
        },
        "packages": {
            p: md.version(p)
            for p in ("torch", "transformers", "trl", "peft", "accelerate",
                      "deepspeed", "datasets")
            if _safe_version(p)
        },
        "hardware": {
            "gpu": _sh("nvidia-smi --query-gpu=name --format=csv,noheader | head -1"),
            "gpu_count": int(os.environ.get("WORLD_SIZE", "1")),
            "driver": _sh("nvidia-smi --query-gpu=driver_version --format=csv,noheader | head -1"),
            "python": platform.python_version(),
        },
        "scheduler": {
            "slurm_job_id": os.environ.get("SLURM_JOB_ID"),
            "nodelist": os.environ.get("SLURM_JOB_NODELIST"),
        },
        "seeds": {"global": config.get("seed"), "data_order": config.get("data_seed")},
    }


def _safe_version(p: str) -> bool:
    try:
        md.version(p)
        return True
    except md.PackageNotFoundError:
        return False

# 평가 세트와 학습 코퍼스의 n-그램 겹침을 러프하게 스캔합니다.
# 정밀한 도구가 아니라 "명백한 사고"를 잡는 1차 필터입니다.
from datasketch import MinHash, MinHashLSH   # pip install datasketch


def shingles(text: str, n: int = 13) -> set:
    toks = text.split()
    return {" ".join(toks[i:i + n]) for i in range(max(1, len(toks) - n + 1))}


def build_index(eval_docs, threshold: float = 0.8):
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    for i, doc in enumerate(eval_docs):
        m = MinHash(num_perm=128)
        for s in shingles(doc):
            m.update(s.encode())
        lsh.insert(f"eval-{i}", m)
    return lsh


def scan(train_docs, lsh) -> list:
    hits = []
    for j, doc in enumerate(train_docs):
        m = MinHash(num_perm=128)
        for s in shingles(doc):
            m.update(s.encode())
        found = lsh.query(m)
        if found:
            hits.append((j, found))
    return hits

import math


def optimal_interval_min(save_minutes: float, mtbf_hours: float) -> float:
    """Young/Daly 1차 근사: T = sqrt(2 * delta * M)"""
    return math.sqrt(2 * save_minutes * mtbf_hours * 60)


def wasted_fraction(interval_min: float, save_minutes: float, mtbf_hours: float) -> float:
    """저장에 쓰는 시간 + 장애로 버리는 평균 재계산 시간의 비율"""
    mtbf_min = mtbf_hours * 60
    return save_minutes / interval_min + (interval_min / 2) / mtbf_min


for gpus, mtbf in [(64, 120.0), (512, 15.0), (16384, 3.1)]:
    t = optimal_interval_min(save_minutes=5, mtbf_hours=mtbf)
    print(f"GPU {gpus:6}  MTBF {mtbf:6.1f}h  최적 주기 {t:6.1f}분  "
          f"손실률 {wasted_fraction(t, 5, mtbf):.1%}")
# GPU     64  MTBF  120.0h  최적 주기  268.3분  손실률   3.7%
# GPU    512  MTBF   15.0h  최적 주기   94.9분  손실률  10.5%
# GPU  16384  MTBF    3.1h  최적 주기   43.1분  손실률  23.2%

# 회귀 계층 예시. 평가 도구와 태스크 버전을 함께 고정합니다.
pip install "lm-eval==0.4.12"

lm_eval --model hf \
  --model_args pretrained=./out/sft-8b,dtype=bfloat16 \
  --tasks arc_challenge,hellaswag,gsm8k \
  --batch_size 16 \
  --seed 1234 \
  --output_path reports/$(date +%F)-sft-8b.json \
  --log_samples

# 승격 전 지문 대조 — 학습 산출물과 서빙 설정이 같은 것을 보고 있는지 확인합니다
import hashlib
import json
from transformers import AutoTokenizer, AutoConfig


def fingerprint(path: str) -> dict:
    tok = AutoTokenizer.from_pretrained(path)
    cfg = AutoConfig.from_pretrained(path)
    tmpl = tok.chat_template or ""
    return {
        "chat_template_sha": hashlib.sha256(tmpl.encode()).hexdigest()[:16],
        "vocab_size_tokenizer": len(tok),
        "vocab_size_config": cfg.vocab_size,
        "eos_token_id": tok.eos_token_id,
        "pad_token_id": tok.pad_token_id,
        "max_position_embeddings": getattr(cfg, "max_position_embeddings", None),
        "torch_dtype": str(getattr(cfg, "torch_dtype", None)),
    }


a, b = fingerprint("./out/sft-8b"), fingerprint("./serving/model")
diff = {k: (a[k], b[k]) for k in a if a[k] != b[k]}
print(json.dumps(diff, indent=2, ensure_ascii=False) if diff else "일치")
assert a["vocab_size_tokenizer"] == a["vocab_size_config"], "어휘 크기 불일치"