필사 모드: 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"], "語彙サイズ不一致"