최적화 알고리즘 (Optimization)¶
신경망 학습을 위한 파라미터 업데이트 전략. 학습 속도, 안정성, 최종 성능에 직접적인 영향을 미친다.
옵티마이저 선택¶
LLM/Transformer 학습 표준¶
| 모델 | 옵티마이저 | 설정 |
|---|---|---|
| GPT | AdamW | lr=1e-4~3e-4, β=(0.9, 0.95), wd=0.1 |
| BERT | Adam | lr=5e-5, β=(0.9, 0.999) |
| LLaMA | AdamW | lr=3e-4, β=(0.9, 0.95), wd=0.1 |
| ViT | AdamW | lr=1e-3, β=(0.9, 0.999), wd=0.3 |
AdamW 구현¶
import torch
class AdamW:
def __init__(self, params, lr=1e-3, betas=(0.9, 0.999), eps=1e-8, weight_decay=0.01):
self.params = list(params)
self.lr = lr
self.beta1, self.beta2 = betas
self.eps = eps
self.weight_decay = weight_decay
# 상태 초기화
self.m = [torch.zeros_like(p) for p in self.params] # 1차 모멘트
self.v = [torch.zeros_like(p) for p in self.params] # 2차 모멘트
self.t = 0
def step(self):
self.t += 1
for i, param in enumerate(self.params):
if param.grad is None:
continue
g = param.grad.data
# Weight decay (decoupled)
param.data -= self.lr * self.weight_decay * param.data
# 모멘트 업데이트
self.m[i] = self.beta1 * self.m[i] + (1 - self.beta1) * g
self.v[i] = self.beta2 * self.v[i] + (1 - self.beta2) * g ** 2
# 편향 보정
m_hat = self.m[i] / (1 - self.beta1 ** self.t)
v_hat = self.v[i] / (1 - self.beta2 ** self.t)
# 파라미터 업데이트
param.data -= self.lr * m_hat / (torch.sqrt(v_hat) + self.eps)
def zero_grad(self):
for param in self.params:
if param.grad is not None:
param.grad.zero_()
PyTorch 옵티마이저 사용¶
import torch.optim as optim
# AdamW (권장)
optimizer = optim.AdamW(
model.parameters(),
lr=1e-4,
betas=(0.9, 0.999),
eps=1e-8,
weight_decay=0.01
)
# 파라미터 그룹별 설정
optimizer = optim.AdamW([
{'params': model.encoder.parameters(), 'lr': 1e-5},
{'params': model.decoder.parameters(), 'lr': 1e-4},
], weight_decay=0.01)
# Weight decay 제외 설정
no_decay = ['bias', 'LayerNorm.weight', 'LayerNorm.bias']
optimizer_grouped_parameters = [
{
'params': [p for n, p in model.named_parameters()
if not any(nd in n for nd in no_decay)],
'weight_decay': 0.01
},
{
'params': [p for n, p in model.named_parameters()
if any(nd in n for nd in no_decay)],
'weight_decay': 0.0
}
]
optimizer = optim.AdamW(optimizer_grouped_parameters, lr=1e-4)
학습률 스케줄링¶
Warmup + Decay¶
LLM 학습의 표준.
def get_lr(step, warmup_steps, max_steps, max_lr, min_lr=0):
"""Linear warmup + Cosine decay"""
if step < warmup_steps:
# Linear warmup
return max_lr * step / warmup_steps
else:
# Cosine decay
progress = (step - warmup_steps) / (max_steps - warmup_steps)
return min_lr + 0.5 * (max_lr - min_lr) * (1 + np.cos(np.pi * progress))
# 시각화
steps = np.arange(10000)
lrs = [get_lr(s, warmup_steps=1000, max_steps=10000, max_lr=1e-4) for s in steps]
plt.plot(steps, lrs)
plt.xlabel('Step')
plt.ylabel('Learning Rate')
PyTorch 스케줄러¶
from torch.optim.lr_scheduler import (
CosineAnnealingLR,
CosineAnnealingWarmRestarts,
OneCycleLR,
LambdaLR
)
# Cosine Annealing
scheduler = CosineAnnealingLR(optimizer, T_max=num_epochs, eta_min=1e-6)
# OneCycleLR (단일 사이클)
scheduler = OneCycleLR(
optimizer,
max_lr=1e-3,
total_steps=num_epochs * len(train_loader),
pct_start=0.1, # Warmup 비율
anneal_strategy='cos'
)
# 커스텀 스케줄러 (LambdaLR)
def lr_lambda(current_step):
warmup_steps = 1000
if current_step < warmup_steps:
return current_step / warmup_steps
return max(0.0, 0.5 * (1 + np.cos(np.pi * (current_step - warmup_steps) / (total_steps - warmup_steps))))
scheduler = LambdaLR(optimizer, lr_lambda)
# 학습 루프
for epoch in range(num_epochs):
for batch in train_loader:
optimizer.zero_grad()
loss = model(batch)
loss.backward()
optimizer.step()
scheduler.step() # 스텝마다 호출
Transformers 라이브러리¶
from transformers import get_scheduler
scheduler = get_scheduler(
name="cosine", # linear, cosine, polynomial, constant_with_warmup
optimizer=optimizer,
num_warmup_steps=1000,
num_training_steps=total_steps
)
Gradient Clipping¶
기울기 폭발 방지.
# Norm Clipping (권장)
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
# Value Clipping
torch.nn.utils.clip_grad_value_(model.parameters(), clip_value=0.5)
# 학습 루프에서
loss.backward()
grad_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
# 기울기 norm 모니터링
if step % 100 == 0:
print(f"Gradient norm: {grad_norm:.4f}")
Gradient Accumulation¶
제한된 메모리에서 큰 배치 효과.
accumulation_steps = 4
optimizer.zero_grad()
for i, batch in enumerate(train_loader):
loss = model(batch) / accumulation_steps
loss.backward()
if (i + 1) % accumulation_steps == 0:
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
scheduler.step()
optimizer.zero_grad()
# 실제 배치 크기 = batch_size * accumulation_steps
Mixed Precision Training¶
FP16/BF16으로 메모리 절약 및 속도 향상.
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()
for batch in train_loader:
optimizer.zero_grad()
# Forward (FP16)
with autocast():
output = model(batch)
loss = criterion(output)
# Backward (FP32 gradients)
scaler.scale(loss).backward()
# Unscale and clip
scaler.unscale_(optimizer)
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
# Update
scaler.step(optimizer)
scaler.update()
# BF16 (Ampere GPU 이상)
with autocast(dtype=torch.bfloat16):
output = model(batch)
대규모 학습 최적화¶
LAMB (Large Batch Optimization)¶
대규모 배치에서 layer-wise 학습률 조정.
ZeRO (Zero Redundancy Optimizer)¶
분산 학습에서 메모리 최적화.
# DeepSpeed 사용
import deepspeed
model, optimizer, _, _ = deepspeed.initialize(
model=model,
model_parameters=model.parameters(),
config={
"train_batch_size": 32,
"optimizer": {
"type": "AdamW",
"params": {"lr": 1e-4, "weight_decay": 0.01}
},
"zero_optimization": {
"stage": 2 # 0, 1, 2, 3
},
"fp16": {"enabled": True}
}
)
FSDP (Fully Sharded Data Parallel)¶
from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
model = FSDP(
model,
auto_wrap_policy=transformer_auto_wrap_policy,
mixed_precision=MixedPrecision(
param_dtype=torch.bfloat16,
reduce_dtype=torch.bfloat16,
buffer_dtype=torch.bfloat16
)
)
학습 모니터링¶
class TrainingMonitor:
def __init__(self):
self.history = {
'loss': [],
'lr': [],
'grad_norm': []
}
def log(self, loss, lr, grad_norm):
self.history['loss'].append(loss)
self.history['lr'].append(lr)
self.history['grad_norm'].append(grad_norm)
def plot(self):
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
axes[0].plot(self.history['loss'])
axes[0].set_title('Loss')
axes[0].set_yscale('log')
axes[1].plot(self.history['lr'])
axes[1].set_title('Learning Rate')
axes[2].plot(self.history['grad_norm'])
axes[2].set_title('Gradient Norm')
plt.tight_layout()
# 사용
monitor = TrainingMonitor()
for step, batch in enumerate(train_loader):
loss = train_step(batch)
lr = scheduler.get_last_lr()[0]
grad_norm = compute_grad_norm(model)
monitor.log(loss.item(), lr, grad_norm)
if step % 100 == 0:
print(f"Step {step}: loss={loss.item():.4f}, lr={lr:.2e}, grad={grad_norm:.2f}")
완전한 학습 설정 예시¶
def create_optimizer_and_scheduler(model, config):
"""LLM 학습용 옵티마이저 및 스케줄러 설정"""
# Weight decay 제외 파라미터
no_decay = ['bias', 'LayerNorm.weight', 'LayerNorm.bias']
optimizer_grouped_parameters = [
{
'params': [p for n, p in model.named_parameters()
if not any(nd in n for nd in no_decay)],
'weight_decay': config.weight_decay
},
{
'params': [p for n, p in model.named_parameters()
if any(nd in n for nd in no_decay)],
'weight_decay': 0.0
}
]
optimizer = torch.optim.AdamW(
optimizer_grouped_parameters,
lr=config.learning_rate,
betas=(config.adam_beta1, config.adam_beta2),
eps=config.adam_epsilon
)
scheduler = get_scheduler(
name="cosine",
optimizer=optimizer,
num_warmup_steps=config.warmup_steps,
num_training_steps=config.total_steps
)
return optimizer, scheduler
# 설정 예시
class TrainingConfig:
learning_rate = 3e-4
weight_decay = 0.1
adam_beta1 = 0.9
adam_beta2 = 0.95
adam_epsilon = 1e-8
warmup_steps = 2000
total_steps = 100000
max_grad_norm = 1.0
gradient_accumulation_steps = 4
디버깅 가이드¶
Loss가 감소하지 않을 때¶
def diagnose_training_issues(model, train_loader, optimizer, criterion, num_batches=5):
"""학습 문제 진단"""
model.train()
issues = []
# 1. 단일 배치 과적합 테스트
print("=== 단일 배치 과적합 테스트 ===")
x, y = next(iter(train_loader))
initial_loss = None
for i in range(50):
optimizer.zero_grad()
output = model(x)
loss = criterion(output, y)
loss.backward()
optimizer.step()
if i == 0:
initial_loss = loss.item()
if i % 10 == 0:
print(f" Step {i}: Loss = {loss.item():.4f}")
final_loss = loss.item()
if final_loss > initial_loss * 0.5:
issues.append("단일 배치도 과적합 못함 - 모델 용량 또는 학습률 문제")
# 2. 학습률 확인
print("\n=== 학습률 확인 ===")
for i, param_group in enumerate(optimizer.param_groups):
lr = param_group['lr']
print(f" Group {i}: lr = {lr}")
if lr > 0.1:
issues.append(f"학습률이 너무 높을 수 있음: {lr}")
if lr < 1e-7:
issues.append(f"학습률이 너무 낮을 수 있음: {lr}")
# 3. 기울기 확인
print("\n=== 기울기 통계 ===")
grad_norms = []
for name, param in model.named_parameters():
if param.grad is not None:
norm = param.grad.norm().item()
grad_norms.append(norm)
if norm > 100:
issues.append(f"기울기 폭발: {name} = {norm:.2f}")
if norm < 1e-7:
issues.append(f"기울기 소실: {name} = {norm:.2e}")
avg_grad = sum(grad_norms) / len(grad_norms) if grad_norms else 0
print(f" 평균 기울기 norm: {avg_grad:.6f}")
# 4. 출력 확인
print("\n=== 모델 출력 확인 ===")
with torch.no_grad():
output = model(x)
print(f" Output shape: {output.shape}")
print(f" Output range: [{output.min():.4f}, {output.max():.4f}]")
if torch.isnan(output).any():
issues.append("출력에 NaN 발생!")
if torch.isinf(output).any():
issues.append("출력에 Inf 발생!")
# 결과 요약
print("\n=== 진단 결과 ===")
if issues:
for issue in issues:
print(f" [문제] {issue}")
else:
print(" 명확한 문제 없음. 더 오래 학습하거나 하이퍼파라미터 튜닝 필요")
return issues
# 사용
diagnose_training_issues(model, train_loader, optimizer, criterion)
학습률 찾기 (LR Range Test)¶
class LRFinder:
"""학습률 범위 테스트"""
def __init__(self, model, optimizer, criterion, device='cuda'):
self.model = model
self.optimizer = optimizer
self.criterion = criterion
self.device = device
# 초기 상태 저장
self.initial_state = {
'model': {k: v.clone() for k, v in model.state_dict().items()},
'optimizer': optimizer.state_dict()
}
def range_test(self, train_loader, start_lr=1e-7, end_lr=10, num_iter=100):
"""학습률 범위 테스트 수행"""
self.model.train()
# 학습률 스케줄
lr_mult = (end_lr / start_lr) ** (1 / num_iter)
lr = start_lr
lrs = []
losses = []
best_loss = float('inf')
iterator = iter(train_loader)
for i in range(num_iter):
try:
x, y = next(iterator)
except StopIteration:
iterator = iter(train_loader)
x, y = next(iterator)
x, y = x.to(self.device), y.to(self.device)
# 학습률 설정
for param_group in self.optimizer.param_groups:
param_group['lr'] = lr
# 학습 스텝
self.optimizer.zero_grad()
output = self.model(x)
loss = self.criterion(output, y)
loss.backward()
self.optimizer.step()
# 기록
lrs.append(lr)
losses.append(loss.item())
# 조기 종료 조건
if loss.item() < best_loss:
best_loss = loss.item()
if loss.item() > best_loss * 10:
print(f"Loss exploded at lr={lr:.2e}")
break
lr *= lr_mult
# 상태 복원
self.model.load_state_dict(self.initial_state['model'])
self.optimizer.load_state_dict(self.initial_state['optimizer'])
self.lrs = lrs
self.losses = losses
return lrs, losses
def plot(self):
"""결과 시각화"""
plt.figure(figsize=(10, 6))
plt.semilogx(self.lrs, self.losses)
plt.xlabel('Learning Rate')
plt.ylabel('Loss')
plt.title('LR Range Test')
# 최적 학습률 추천 (가장 가파른 하강 지점의 1/10)
min_idx = losses.index(min(losses))
suggested_lr = self.lrs[max(0, min_idx - 10)]
plt.axvline(x=suggested_lr, color='r', linestyle='--', label=f'Suggested: {suggested_lr:.2e}')
plt.legend()
plt.show()
return suggested_lr
# 사용
lr_finder = LRFinder(model, optimizer, criterion)
lrs, losses = lr_finder.range_test(train_loader)
suggested_lr = lr_finder.plot()
print(f"권장 학습률: {suggested_lr:.2e}")
학습 곡선 분석¶
class TrainingAnalyzer:
"""학습 곡선 분석 및 문제 감지"""
def __init__(self):
self.train_losses = []
self.val_losses = []
self.lrs = []
def update(self, train_loss, val_loss=None, lr=None):
self.train_losses.append(train_loss)
if val_loss is not None:
self.val_losses.append(val_loss)
if lr is not None:
self.lrs.append(lr)
def analyze(self, window=10):
"""학습 상태 분석"""
if len(self.train_losses) < window:
return "데이터 부족"
recent_train = self.train_losses[-window:]
trend = (recent_train[-1] - recent_train[0]) / window
analysis = []
# 1. 학습 진행 상태
if trend > 0:
analysis.append(f"[경고] Loss가 증가하는 추세 (trend: {trend:.6f})")
analysis.append(" → 학습률을 낮추거나 모델 확인")
elif abs(trend) < 1e-6:
analysis.append("[정보] Loss 정체 상태")
analysis.append(" → 학습률 감소 또는 조기 종료 고려")
else:
analysis.append(f"[정상] Loss 감소 중 (trend: {trend:.6f})")
# 2. 과적합 감지
if self.val_losses:
recent_val = self.val_losses[-window:]
val_trend = (recent_val[-1] - recent_val[0]) / window
gap = recent_val[-1] - recent_train[-1]
if val_trend > 0 and trend < 0:
analysis.append(f"[경고] 과적합 감지! Train↓ Val↑")
analysis.append(" → Dropout 증가, 데이터 증강, 조기 종료")
if gap > 0.1:
analysis.append(f"[정보] Train-Val gap: {gap:.4f}")
# 3. 진동 감지
if len(self.train_losses) > 20:
variance = torch.tensor(self.train_losses[-20:]).var().item()
if variance > 0.01:
analysis.append(f"[경고] Loss 진동 감지 (var: {variance:.4f})")
analysis.append(" → 학습률 감소 고려")
return '\n'.join(analysis)
def plot(self):
"""학습 곡선 시각화"""
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# Loss 곡선
axes[0].plot(self.train_losses, label='Train Loss')
if self.val_losses:
axes[0].plot(self.val_losses, label='Val Loss')
axes[0].set_xlabel('Epoch')
axes[0].set_ylabel('Loss')
axes[0].legend()
axes[0].set_title('Loss Curves')
# 학습률 곡선
if self.lrs:
axes[1].plot(self.lrs)
axes[1].set_xlabel('Epoch')
axes[1].set_ylabel('Learning Rate')
axes[1].set_yscale('log')
axes[1].set_title('Learning Rate Schedule')
plt.tight_layout()
plt.show()
# 사용
analyzer = TrainingAnalyzer()
for epoch in range(num_epochs):
train_loss = train_one_epoch(model, train_loader)
val_loss = evaluate(model, val_loader)
lr = scheduler.get_last_lr()[0]
analyzer.update(train_loss, val_loss, lr)
if epoch % 10 == 0:
print(analyzer.analyze())
analyzer.plot()
일반적인 문제와 해결¶
| 문제 | 증상 | 해결책 |
|---|---|---|
| Loss가 감소 안 함 | 초기부터 정체 | LR 증가, 초기화 확인, 데이터 확인 |
| Loss가 진동 | 오르락내리락 | LR 감소, 배치 크기 증가 |
| Loss가 폭발 | NaN 또는 급증 | LR 감소, Gradient Clipping, 초기화 |
| 과적합 | Train↓ Val↑ | Dropout, Weight Decay, 데이터 증강 |
| 과소적합 | 둘 다 높음 | 모델 용량↑, LR↑, 더 오래 학습 |