Weights & Biases(W&B) 실험 관리 실전 가이드: 실험 추적부터 모델 레지스트리와 프로덕션 모니터링까지

┌─────────────────────────────────────────────────────────────────┐
│                    W&B Platform Architecture                     │
├─────────────────────────────────────────────────────────────────┤
│                                                                   │
│  ┌──────────────┐   ┌──────────────┐   ┌──────────────────────┐  │
│  │  Experiment   │   │   Sweeps     │   │     Artifacts        │  │
│  │  Tracking     │──▶│ Hyperparams  │──▶│  Data/Model Version  │  │
│  │  (wandb.log)  │   │  Tuning      │   │  Control             │  │
│  └──────┬───────┘   └──────────────┘   └──────────┬───────────┘  │
│         │                                          │              │
│         ▼                                          ▼              │
│  ┌──────────────┐   ┌──────────────┐   ┌──────────────────────┐  │
│  │   Tables      │   │   Reports    │   │   Model Registry     │  │
│  │  Data Viz     │   │  Team Collab │   │  Production Staging  │  │
│  │  Analysis     │   │  Sharing     │   │  Deployment          │  │
│  └──────────────┘   └──────────────┘   └──────────┬───────────┘  │
│                                                    │              │
│                                          ┌─────────▼──────────┐  │
│                                          │   W&B Launch /      │  │
│                                          │   Webhooks / CI/CD  │  │
│                                          │   Automation        │  │
│                                          └────────────────────┘  │
│                                                                   │
│  Infrastructure: Cloud (SaaS) / Self-Hosted / Dedicated Cloud    │
└─────────────────────────────────────────────────────────────────┘

import wandb
import torch
import torch.nn as nn
from torch.utils.data import DataLoader

# 1. W&B 초기화 - 프로젝트와 실험 이름 지정
wandb.init(
    project="image-classifier",
    name="resnet50-lr0.001-batch32",
    config={
        "architecture": "ResNet50",
        "learning_rate": 0.001,
        "batch_size": 32,
        "epochs": 100,
        "optimizer": "AdamW",
        "weight_decay": 1e-4,
        "scheduler": "CosineAnnealingLR",
        "dataset": "CIFAR-100",
        "augmentation": "RandAugment",
    },
    tags=["baseline", "resnet", "cifar100"],
    notes="ResNet50 baseline with CosineAnnealing scheduler",
)

config = wandb.config  # config 객체로 하이퍼파라미터 접근

# 2. 모델, 옵티마이저, 스케줄러 설정
model = build_model(config.architecture)
optimizer = torch.optim.AdamW(
    model.parameters(),
    lr=config.learning_rate,
    weight_decay=config.weight_decay
)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
    optimizer, T_max=config.epochs
)

# 3. W&B에서 모델 gradient 추적 (선택 사항)
wandb.watch(model, log="all", log_freq=100)

# 4. 학습 루프 - 메트릭 로깅
for epoch in range(config.epochs):
    model.train()
    train_loss = 0.0
    correct = 0
    total = 0

    for batch_idx, (images, labels) in enumerate(train_loader):
        images, labels = images.cuda(), labels.cuda()
        outputs = model(images)
        loss = criterion(outputs, labels)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        train_loss += loss.item()
        _, predicted = outputs.max(1)
        total += labels.size(0)
        correct += predicted.eq(labels).sum().item()

        # 배치 단위 로깅 (선택적)
        if batch_idx % 50 == 0:
            wandb.log({
                "batch/loss": loss.item(),
                "batch/step": epoch * len(train_loader) + batch_idx,
            })

    # 에포크 단위 메트릭 로깅
    train_acc = 100.0 * correct / total
    val_loss, val_acc = evaluate(model, val_loader)

    wandb.log({
        "epoch": epoch,
        "train/loss": train_loss / len(train_loader),
        "train/accuracy": train_acc,
        "val/loss": val_loss,
        "val/accuracy": val_acc,
        "learning_rate": scheduler.get_last_lr()[0],
        "gpu/memory_allocated_gb": torch.cuda.memory_allocated() / 1e9,
        "gpu/utilization": torch.cuda.utilization(),
    })

    scheduler.step()

    # 최고 성능 모델 체크포인트 저장
    if val_acc > best_val_acc:
        best_val_acc = val_acc
        torch.save(model.state_dict(), "best_model.pth")
        wandb.save("best_model.pth")
        wandb.run.summary["best_val_accuracy"] = best_val_acc

# 5. 실험 종료
wandb.finish()

# Hugging Face Transformers 통합 예시
from transformers import TrainingArguments, Trainer

training_args = TrainingArguments(
    output_dir="./results",
    report_to="wandb",           # W&B 자동 연동
    run_name="bert-finetune-v1",
    logging_steps=10,
    evaluation_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="eval_f1",
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    compute_metrics=compute_metrics,
)

trainer.train()  # 자동으로 W&B에 메트릭 로깅

import wandb

# 1. Sweep 설정 정의
sweep_config = {
    "method": "bayes",  # grid, random, bayes 중 선택
    "metric": {
        "name": "val/accuracy",
        "goal": "maximize"
    },
    "parameters": {
        "learning_rate": {
            "distribution": "log_uniform_values",
            "min": 1e-5,
            "max": 1e-2,
        },
        "batch_size": {
            "values": [16, 32, 64, 128]
        },
        "optimizer": {
            "values": ["adam", "adamw", "sgd"]
        },
        "weight_decay": {
            "distribution": "log_uniform_values",
            "min": 1e-6,
            "max": 1e-2,
        },
        "dropout": {
            "distribution": "uniform",
            "min": 0.1,
            "max": 0.5,
        },
        "architecture": {
            "values": ["resnet18", "resnet34", "resnet50"]
        },
    },
    # Early Termination: 성능이 낮은 실험 자동 중단
    "early_terminate": {
        "type": "hyperband",
        "min_iter": 5,       # 최소 5 에포크 후 평가
        "eta": 3,            # 상위 1/3만 계속 진행
        "max_iter": 50,
    },
}

# 2. Sweep 생성
sweep_id = wandb.sweep(
    sweep=sweep_config,
    project="image-classifier"
)

# 3. 학습 함수 정의
def train():
    wandb.init()
    config = wandb.config

    model = build_model(config.architecture, dropout=config.dropout)
    optimizer = get_optimizer(
        config.optimizer, model, config.learning_rate, config.weight_decay
    )

    for epoch in range(50):
        train_loss = train_one_epoch(model, train_loader, optimizer)
        val_loss, val_acc = evaluate(model, val_loader)

        wandb.log({
            "epoch": epoch,
            "train/loss": train_loss,
            "val/loss": val_loss,
            "val/accuracy": val_acc,
        })

    wandb.finish()

# 4. Sweep Agent 실행 - 최대 50회 실험
wandb.agent(sweep_id, function=train, count=50)

# 터미널 1 (GPU 0)
CUDA_VISIBLE_DEVICES=0 wandb agent your-entity/image-classifier/sweep_id

# 터미널 2 (GPU 1)
CUDA_VISIBLE_DEVICES=1 wandb agent your-entity/image-classifier/sweep_id

# 터미널 3 (다른 머신)
CUDA_VISIBLE_DEVICES=0 wandb agent your-entity/image-classifier/sweep_id

import wandb

# === 데이터셋 업로드 ===
run = wandb.init(project="image-classifier", job_type="data-pipeline")

# Artifact 생성
dataset_artifact = wandb.Artifact(
    name="cifar100-processed",
    type="dataset",
    description="CIFAR-100 전처리 완료 데이터셋 (정규화, 증강 적용)",
    metadata={
        "num_classes": 100,
        "train_size": 50000,
        "val_size": 10000,
        "preprocessing": "normalize_mean_std",
        "augmentation": "randaugment_n2_m14",
        "source": "torchvision.datasets.CIFAR100",
    },
)

# 디렉토리 또는 파일 추가
dataset_artifact.add_dir("./data/processed/cifar100")
dataset_artifact.add_file("./data/metadata.json")

# 업로드 - 자동으로 v0, v1, v2... 버전 관리
run.log_artifact(dataset_artifact)
run.finish()

# === 학습 시 데이터셋 다운로드 ===
run = wandb.init(project="image-classifier", job_type="training")

# 특정 버전 또는 최신 버전 사용
dataset = run.use_artifact("cifar100-processed:latest")
data_dir = dataset.download()  # 로컬에 캐시된 경로 반환

# 학습 진행...
train_loader = create_dataloader(data_dir, split="train")

# === 모델 체크포인트 저장 ===
model_artifact = wandb.Artifact(
    name="resnet50-cifar100",
    type="model",
    description="ResNet50 trained on CIFAR-100",
    metadata={
        "val_accuracy": best_val_acc,
        "architecture": "resnet50",
        "framework": "pytorch",
        "training_run": wandb.run.id,
    },
)

model_artifact.add_file("best_model.pth")
model_artifact.add_file("model_config.json")
run.log_artifact(model_artifact)
run.finish()

import wandb

# 1. 학습 완료 후 모델을 Registry에 등록
run = wandb.init(project="image-classifier", job_type="register-model")

# 기존 모델 Artifact 참조
model_artifact = run.use_artifact("resnet50-cifar100:v5")

# Registry에 링크 - "Registered Model" 생성 또는 기존 모델에 새 버전 추가
run.link_artifact(
    artifact=model_artifact,
    target_path="model-registry/image-classifier-prod",
    aliases=["candidate", "v2.1.0"],
)
run.finish()

# 2. 프로덕션 배포 시 Registry에서 모델 가져오기
run = wandb.init(project="image-classifier", job_type="deployment")

# Registry에서 특정 alias로 모델 로드
model_artifact = run.use_artifact(
    "model-registry/image-classifier-prod:production"
)
model_dir = model_artifact.download()

# 모델 로드 및 서빙
model = load_model(model_dir)
run.finish()

# 3. Webhook을 통한 자동화 - Registry 이벤트에 반응
# W&B UI에서 설정하거나 API로 구성
# 새 모델이 "production" alias를 받으면 자동으로 배포 파이프라인 트리거

import wandb

run = wandb.init(project="image-classifier", job_type="evaluation")

# 예측 결과 테이블 생성
columns = ["image", "true_label", "predicted_label", "confidence", "correct"]
prediction_table = wandb.Table(columns=columns)

model.eval()
with torch.no_grad():
    for images, labels in test_loader:
        outputs = model(images.cuda())
        probs = torch.softmax(outputs, dim=1)
        confidences, predictions = probs.max(dim=1)

        for i in range(len(images)):
            is_correct = predictions[i].item() == labels[i].item()
            prediction_table.add_data(
                wandb.Image(images[i]),                    # 이미지
                class_names[labels[i].item()],             # 실제 라벨
                class_names[predictions[i].item()],        # 예측 라벨
                confidences[i].item(),                     # 신뢰도
                is_correct,                                # 정오답 여부
            )

# 테이블 로깅
run.log({"predictions": prediction_table})

# 오분류 분석 테이블
error_table = wandb.Table(
    columns=["image", "true", "predicted", "confidence", "loss"]
)
# ... 오분류 케이스만 필터링하여 추가

run.log({"error_analysis": error_table})
run.finish()

organization/
├── project-name-research/     # 탐색적 실험
├── project-name-staging/      # 배포 후보 모델 검증
└── project-name-production/   # 프로덕션 모니터링

{모델아키텍처}-{데이터셋}-{핵심변경사항}-{날짜}
예: resnet50-cifar100-cosine_lr-20260308

# .github/workflows/model-training.yml
name: Model Training and Evaluation Pipeline

on:
  push:
    paths:
      - 'src/models/**'
      - 'configs/**'
  workflow_dispatch:
    inputs:
      experiment_name:
        description: 'Experiment name'
        required: true

env:
  WANDB_API_KEY: ${{ secrets.WANDB_API_KEY }}
  WANDB_PROJECT: image-classifier
  WANDB_ENTITY: your-team

jobs:
  train:
    runs-on: [self-hosted, gpu]
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Train model
        run: |
          python src/train.py \
            --config configs/production.yaml \
            --wandb-run-name "ci-${{ github.sha }}"

      - name: Evaluate model
        run: |
          python src/evaluate.py \
            --model-artifact "resnet50-cifar100:latest" \
            --threshold-accuracy 0.92

      - name: Register model to W&B Registry
        if: success()
        run: |
          python src/register_model.py \
            --artifact "resnet50-cifar100:latest" \
            --registry "image-classifier-prod" \
            --alias "candidate"

  deploy:
    needs: train
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Promote model to production
        run: |
          python src/promote_model.py \
            --registry "image-classifier-prod" \
            --from-alias "candidate" \
            --to-alias "production"

# W&B Webhook 수신 서버 예시 (FastAPI)
from fastapi import FastAPI, Request
import subprocess

app = FastAPI()

@app.post("/wandb-webhook")
async def handle_wandb_webhook(request: Request):
    payload = await request.json()
    event_type = payload.get("event_type")
    artifact_alias = payload.get("artifact_alias", "")

    if event_type == "ADD_ARTIFACT_ALIAS" and "production" in artifact_alias:
        # 프로덕션 배포 파이프라인 트리거
        artifact_path = payload["artifact_version_string"]
        subprocess.run([
            "kubectl", "set", "image",
            "deployment/model-server",
            f"model-server=model-image:{artifact_path}",
        ])
        # Slack 알림
        notify_slack(
            f"New production model deployed: {artifact_path}"
        )

    return {"status": "ok"}

# 오프라인 모드로 실행 후 나중에 동기화
import os
os.environ["WANDB_MODE"] = "offline"

wandb.init(project="my-project")
# ... 학습 진행 ...
wandb.finish()

# 나중에 인터넷 연결 시 수동 동기화
# 터미널에서 실행: wandb sync ./wandb/offline-run-*

# 참조 Artifact 사용 - 실제 파일을 업로드하지 않고 경로만 기록
artifact = wandb.Artifact("large-dataset", type="dataset")
artifact.add_reference(
    "s3://my-bucket/datasets/imagenet/",  # S3, GCS, Azure Blob 등 지원
    name="imagenet"
)
run.log_artifact(artifact)

# 배치 단위로 테이블 로깅
BATCH_SIZE = 1000
table = wandb.Table(columns=["id", "prediction", "label"])

for i in range(0, len(predictions), BATCH_SIZE):
    batch = predictions[i:i + BATCH_SIZE]
    for pred in batch:
        table.add_data(pred["id"], pred["prediction"], pred["label"])

run.log({"predictions": table})

# 기존 sweep 재개
wandb agent your-entity/project/existing_sweep_id

# W&B Alert 설정 - 학습 이상 감지 시 알림
import wandb

wandb.init(project="production-monitor")

# 학습 중 이상 감지 시 Slack/Email 알림
if val_loss > threshold:
    wandb.alert(
        title="Training Anomaly Detected",
        text=f"Validation loss ({val_loss:.4f}) exceeded threshold ({threshold})",
        level=wandb.AlertLevel.WARN,
    )

# NaN 감지
if torch.isnan(loss):
    wandb.alert(
        title="NaN Loss Detected",
        text="Training diverged - NaN loss detected",
        level=wandb.AlertLevel.ERROR,
    )
    wandb.finish(exit_code=1)

# 오래된 Artifact 버전 자동 정리 스크립트
import wandb

api = wandb.Api()
artifact_type = api.artifact_type("model", "your-entity/your-project")

for artifact_collection in artifact_type.collections():
    versions = artifact_collection.versions()
    # 최신 5개 버전만 유지
    for version in versions[5:]:
        if "production" not in version.aliases and "staging" not in version.aliases:
            version.delete()
            print(f"Deleted: {version.name}")

import sys
import torch
import wandb

def init_wandb_with_env(project, name, config):
    """환경 정보를 자동으로 포함하는 wandb.init 래퍼"""
    env_config = {
        "python_version": sys.version,
        "pytorch_version": torch.__version__,
        "cuda_version": torch.version.cuda,
        "cudnn_version": torch.backends.cudnn.version(),
        "gpu_name": torch.cuda.get_device_name(0) if torch.cuda.is_available() else "N/A",
        "num_gpus": torch.cuda.device_count(),
    }
    config.update(env_config)

    return wandb.init(
        project=project,
        name=name,
        config=config,
        save_code=True,  # 실행한 코드 자동 저장
    )