C.W.K.
Stream
Lesson 08 of 08 · published

합치기 — Production-Grade Loop

~14 min · production, gpu, logging, tqdm

Level 0Tensor 호기심
0 XP0/62 lessons0/13 achievements
0/120 XP to next level120 XP to go0% complete

한 곳에 모든 거 — copy 하고 적응할 loop

이 lesson 이 트랙의 모든 pattern 을 하나의 runnable training loop 로 조립. 한 번 신중히 읽기; 경력 동안 이 variant 에서 cribbing 할 거야.

recipe 안:

  • Device 선택 (CUDA → MPS → CPU fallback)
  • weight decay split 의 AdamW (bias / norm 엔 decay 없음)
  • warmup 의 cosine LR
  • Mixed-precision (Ampere+ / Apple Silicon 에 bfloat16)
  • Gradient clipping
  • running loss 의 tqdm 진행 bar
  • Per-epoch validation
  • Best-checkpoint 저장
  • NaN 보호

의도적으로 NOT 들어간 거 (각각 자체 트랙 / lesson 받음):

  • Distributed training (DDP / FSDP) — Track 7
  • torch.compile() — Track 7
  • Experiment tracking (W&B / MLflow) — Track 8

Code

전체 loop, 끝에서 끝까지·python
import math
import torch
import torch.nn as nn
from torch.amp import autocast
from tqdm import tqdm

# 1. Device
device = (
    "cuda" if torch.cuda.is_available()
    else "mps" if torch.backends.mps.is_available()
    else "cpu"
)

# 2. Model
model = MyModel().to(device)

# 3. Loss
criterion = nn.CrossEntropyLoss()

# 4. Optimizer with weight-decay split
decay, no_decay = [], []
for n, p in model.named_parameters():
    if not p.requires_grad: continue
    if p.dim() < 2 or any(k in n for k in ('bias', 'norm')):
        no_decay.append(p)
    else:
        decay.append(p)
optimizer = torch.optim.AdamW(
    [{'params': decay, 'weight_decay': 0.01},
     {'params': no_decay, 'weight_decay': 0.0}],
    lr=1e-4,
)

# 5. Scheduler — warmup + cosine
total_steps = len(train_loader) * num_epochs
warmup_steps = total_steps // 20    # 5% warmup
def lr_lambda(step):
    if step < warmup_steps:
        return step / warmup_steps
    progress = (step - warmup_steps) / max(1, total_steps - warmup_steps)
    return 0.5 * (1.0 + math.cos(math.pi * progress))
scheduler = torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
...계속: train + validate + checkpoint·python
best_val = float('inf')
global_step = 0

for epoch in range(num_epochs):
    # ---- TRAIN ----
    model.train()
    pbar = tqdm(train_loader, desc=f"epoch {epoch:02d}")
    running = 0.0

    for batch_x, batch_y in pbar:
        batch_x = batch_x.to(device, non_blocking=True)
        batch_y = batch_y.to(device, non_blocking=True)

        optimizer.zero_grad(set_to_none=True)

        with autocast(device_type=device, dtype=torch.bfloat16) if device != 'cpu' else nullcontext():
            output = model(batch_x)
            loss = criterion(output, batch_y)

        if not torch.isfinite(loss):
            print(f"step {global_step}: non-finite loss; skipping")
            continue

        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        scheduler.step()
        global_step += 1

        running = 0.95 * running + 0.05 * loss.item() if global_step > 1 else loss.item()
        pbar.set_postfix(loss=f"{running:.4f}", lr=f"{scheduler.get_last_lr()[0]:.2e}")

    # ---- VALIDATE ----
    model.eval()
    val_loss = 0.0
    val_n = 0
    with torch.inference_mode():
        for vx, vy in val_loader:
            vx, vy = vx.to(device), vy.to(device)
            val_loss += criterion(model(vx), vy).item() * vx.size(0)
            val_n += vx.size(0)
    val_loss /= val_n
    print(f"epoch {epoch:02d}  val_loss={val_loss:.4f}")

    # ---- CHECKPOINT ----
    if val_loss < best_val:
        best_val = val_loss
        torch.save({
            'epoch': epoch,
            'val_loss': val_loss,
            'model_state_dict': model.state_dict(),
            'optimizer_state_dict': optimizer.state_dict(),
            'scheduler_state_dict': scheduler.state_dict(),
        }, 'best.pt')
nullcontext shim — autocast 가 CPU 에서 no-op·python
from contextlib import nullcontext

# autocast doesn't apply on CPU. Use nullcontext as a no-op stand-in
# so the same training code runs cleanly on CPU for debugging.

device = "cpu"  # for example
amp_ctx = autocast(device_type=device, dtype=torch.bfloat16) if device != 'cpu' else nullcontext()

with amp_ctx:
    out = model(x)
    loss = criterion(out, y)

External links

Exercise

이 loop 를 좋아하는 작은 dataset (MNIST / CIFAR-10 / FashionMNIST) 에 적응. 5 epoch run. 확인: (1) val loss 가 epoch 들 위 감소, (2) best checkpoint 가 fresh model 에 깔끔히 로드, (3) bfloat16 autocast 가 hardware 에 실제로 engage (제거하고 시간 — 더 느려야).

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.