인스톨! 파이토치 12강 | PyTorch 완전한 학습 파이프라인: 실전 템플릿 코드
2025. 12. 25. 00:52
728x90
반응형
[AI 인공지능 머신러닝 딥러닝/Python | PyTorch] - 인스톨! 파이토치 강의 소개
인스톨! 파이토치 강의 소개
혁펜하임 PyTorch 강의 오리엔테이션 요약혁펜하임 채널의 '[PyTorch] 0강. 오리엔테이션' 영상은 채널 5주년 기념으로 '인스톨! 파이토치' 강의를 소개하는 내용입니다. 강의자는 최근 출간한 '이론
inner-game.tistory.com
딥러닝 모델을 실전에서 학습시키려면 단순히 모델을 정의하고 학습 루프를 돌리는 것 이상의 작업이 필요합니다. Validation set을 활용한 모델 평가, 학습률 스케줄러를 통한 성능 최적화, 체크포인트 저장 등 다양한 기법이 필요합니다. 이번 강의에서는 지금까지 배운 모든 내용을 통합한 완성도 높은 코드 템플릿을 제공합니다.[^1]
전체 파이프라인 구조
딥러닝 학습 파이프라인은 크게 다음과 같은 단계로 구성됩니다:[^2][^1]
- 데이터 준비: Train/Validation/Test 분할 및 DataLoader 생성
- 모델 정의: 네트워크 아키텍처 구성
- 손실 함수 및 옵티마이저 설정: 학습 방법 정의
- 학습률 스케줄러 설정: 동적 학습률 조정
- 학습 루프: Training과 Validation 반복
- 모델 평가: Test 데이터로 최종 성능 평가
- 모델 저장: 최적의 모델 체크포인트 저장
완전한 학습 템플릿 코드
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, random_split
from torchvision import datasets, transforms, models
import copy
import time
# ==================== 1. 설정 및 하이퍼파라미터 ====================
class Config:
# 데이터 관련
data_dir = './data'
batch_size = 32
num_workers = 4
# 모델 관련
num_classes = 10
pretrained = True
# 학습 관련
num_epochs = 50
learning_rate = 0.001
momentum = 0.9
weight_decay = 1e-4
# 데이터 분할 비율
train_ratio = 0.7
val_ratio = 0.15
test_ratio = 0.15
# 기타
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
save_path = './best_model.pth'
seed = 42
config = Config()
# ==================== 2. 재현성을 위한 시드 고정 ====================
def set_seed(seed):
torch.manual_seed(seed)
torch.cuda.manual_seed_all(seed)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
import numpy as np
import random
np.random.seed(seed)
random.seed(seed)
set_seed(config.seed)
# ==================== 3. 데이터 준비 ====================
# Train transform (데이터 증강 포함)
train_transform = transforms.Compose([
transforms.Resize(256),
transforms.RandomResizedCrop(224),
transforms.RandomHorizontalFlip(p=0.5),
transforms.RandomRotation(degrees=15),
transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
])
# Validation/Test transform (증강 없음)
val_test_transform = transforms.Compose([
transforms.Resize(256),
transforms.CenterCrop(224),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
])
# 전체 데이터셋 로드
full_dataset = datasets.ImageFolder(root=config.data_dir)
# Train/Val/Test 분할
train_size = int(config.train_ratio * len(full_dataset))
val_size = int(config.val_ratio * len(full_dataset))
test_size = len(full_dataset) - train_size - val_size
train_dataset, val_dataset, test_dataset = random_split(
full_dataset,
[train_size, val_size, test_size],
generator=torch.Generator().manual_seed(config.seed)
)
# Transform 적용
train_dataset.dataset.transform = train_transform
val_dataset.dataset.transform = val_test_transform
test_dataset.dataset.transform = val_test_transform
# DataLoader 생성
train_loader = DataLoader(
train_dataset,
batch_size=config.batch_size,
shuffle=True,
num_workers=config.num_workers,
pin_memory=True
)
val_loader = DataLoader(
val_dataset,
batch_size=config.batch_size,
shuffle=False,
num_workers=config.num_workers,
pin_memory=True
)
test_loader = DataLoader(
test_dataset,
batch_size=config.batch_size,
shuffle=False,
num_workers=config.num_workers,
pin_memory=True
)
# ==================== 4. 모델 정의 ====================
def build_model(num_classes, pretrained=True):
"""사전 학습된 모델 로드 및 수정"""
model = models.resnet50(pretrained=pretrained)
# 마지막 FC 레이어 교체
num_features = model.fc.in_features
model.fc = nn.Linear(num_features, num_classes)
return model
model = build_model(config.num_classes, config.pretrained)
model = model.to(config.device)
# ==================== 5. 손실 함수 및 옵티마이저 ====================
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(
model.parameters(),
lr=config.learning_rate,
momentum=config.momentum,
weight_decay=config.weight_decay
)
# ==================== 6. 학습률 스케줄러 ====================
# StepLR: 일정 epoch마다 lr을 gamma배 감소
scheduler = optim.lr_scheduler.StepLR(
optimizer,
step_size=10, # 10 epoch마다
gamma=0.1 # lr을 0.1배로 감소
)
# 다른 스케줄러 옵션들:
# ReduceLROnPlateau: validation loss가 개선되지 않을 때 lr 감소
# scheduler = optim.lr_scheduler.ReduceLROnPlateau(
# optimizer, mode='min', factor=0.1, patience=5, verbose=True
# )
# CosineAnnealingLR: cosine 함수 형태로 lr 감소
# scheduler = optim.lr_scheduler.CosineAnnealingLR(
# optimizer, T_max=config.num_epochs
# )
# ==================== 7. 학습 함수 ====================
def train_one_epoch(model, dataloader, criterion, optimizer, device):
"""1 epoch 학습"""
model.train()
running_loss = 0.0
running_corrects = 0
for inputs, labels in dataloader:
inputs = inputs.to(device)
labels = labels.to(device)
# Forward
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, labels)
_, preds = torch.max(outputs, 1)
# Backward
loss.backward()
optimizer.step()
# 통계
running_loss += loss.item() * inputs.size(0)
running_corrects += torch.sum(preds == labels.data)
epoch_loss = running_loss / len(dataloader.dataset)
epoch_acc = running_corrects.double() / len(dataloader.dataset)
return epoch_loss, epoch_acc
# ==================== 8. 검증 함수 ====================
def validate(model, dataloader, criterion, device):
"""Validation/Test 평가"""
model.eval()
running_loss = 0.0
running_corrects = 0
with torch.no_grad():
for inputs, labels in dataloader:
inputs = inputs.to(device)
labels = labels.to(device)
outputs = model(inputs)
loss = criterion(outputs, labels)
_, preds = torch.max(outputs, 1)
running_loss += loss.item() * inputs.size(0)
running_corrects += torch.sum(preds == labels.data)
epoch_loss = running_loss / len(dataloader.dataset)
epoch_acc = running_corrects.double() / len(dataloader.dataset)
return epoch_loss, epoch_acc
# ==================== 9. 전체 학습 루프 ====================
def train_model(model, train_loader, val_loader, criterion, optimizer,
scheduler, num_epochs, device, save_path):
"""전체 학습 파이프라인"""
since = time.time()
best_model_wts = copy.deepcopy(model.state_dict())
best_acc = 0.0
# 학습 히스토리 저장
history = {
'train_loss': [], 'train_acc': [],
'val_loss': [], 'val_acc': []
}
for epoch in range(num_epochs):
print(f'Epoch {epoch+1}/{num_epochs}')
print('-' * 60)
# 학습
train_loss, train_acc = train_one_epoch(
model, train_loader, criterion, optimizer, device
)
# 검증
val_loss, val_acc = validate(
model, val_loader, criterion, device
)
# 학습률 스케줄러 업데이트
scheduler.step()
# ReduceLROnPlateau 사용 시:
# scheduler.step(val_loss)
# 히스토리 저장
history['train_loss'].append(train_loss)
history['train_acc'].append(train_acc.item())
history['val_loss'].append(val_loss)
history['val_acc'].append(val_acc.item())
# 현재 학습률 출력
current_lr = optimizer.param_groups[^0]['lr']
print(f'Train Loss: {train_loss:.4f} Acc: {train_acc:.4f}')
print(f'Val Loss: {val_loss:.4f} Acc: {val_acc:.4f}')
print(f'Learning Rate: {current_lr:.6f}')
# 최고 성능 모델 저장
if val_acc > best_acc:
best_acc = val_acc
best_model_wts = copy.deepcopy(model.state_dict())
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'best_acc': best_acc,
}, save_path)
print(f'✓ Best model saved! (Acc: {best_acc:.4f})')
print()
time_elapsed = time.time() - since
print(f'Training complete in {time_elapsed//60:.0f}m {time_elapsed%60:.0f}s')
print(f'Best Validation Acc: {best_acc:.4f}')
# 최고 성능 모델 로드
model.load_state_dict(best_model_wts)
return model, history
# ==================== 10. 학습 실행 ====================
trained_model, history = train_model(
model, train_loader, val_loader, criterion, optimizer,
scheduler, config.num_epochs, config.device, config.save_path
)
# ==================== 11. 테스트 평가 ====================
print("=" * 60)
print("Final Test Evaluation")
print("=" * 60)
test_loss, test_acc = validate(
trained_model, test_loader, criterion, config.device
)
print(f'Test Loss: {test_loss:.4f}')
print(f'Test Acc: {test_acc:.4f}')
# ==================== 12. 학습 결과 시각화 ====================
import matplotlib.pyplot as plt
def plot_history(history):
"""학습 히스토리 시각화"""
fig, axes = plt.subplots(1, 2, figsize=(15, 5))
# Loss 그래프
axes[^0].plot(history['train_loss'], label='Train Loss')
axes[^0].plot(history['val_loss'], label='Val Loss')
axes[^0].set_xlabel('Epoch')
axes[^0].set_ylabel('Loss')
axes[^0].legend()
axes[^0].set_title('Training and Validation Loss')
axes[^0].grid(True)
# Accuracy 그래프
axes[^1].plot(history['train_acc'], label='Train Acc')
axes[^1].plot(history['val_acc'], label='Val Acc')
axes[^1].set_xlabel('Epoch')
axes[^1].set_ylabel('Accuracy')
axes[^1].legend()
axes[^1].set_title('Training and Validation Accuracy')
axes[^1].grid(True)
plt.tight_layout()
plt.savefig('training_history.png')
plt.show()
plot_history(history)
# ==================== 13. 저장된 모델 로드 ====================
def load_checkpoint(model, checkpoint_path):
"""저장된 체크포인트 로드"""
checkpoint = torch.load(checkpoint_path)
model.load_state_dict(checkpoint['model_state_dict'])
epoch = checkpoint['epoch']
best_acc = checkpoint['best_acc']
print(f'Loaded checkpoint from epoch {epoch} (Acc: {best_acc:.4f})')
return model
# 사용 예시
# loaded_model = build_model(config.num_classes, pretrained=False)
# loaded_model = load_checkpoint(loaded_model, config.save_path)학습률 스케줄러 심화
학습률 스케줄러는 학습 과정에서 학습률을 동적으로 조정하여 성능을 향상시킵니다. PyTorch는 다양한 스케줄러를 제공합니다.[^3][^4]
주요 스케줄러 종류
1. StepLR: 일정 epoch마다 학습률 감소[^5]
scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)
# 10 epoch마다 lr을 0.1배로 감소2. MultiStepLR: 지정한 epoch들에서 학습률 감소[^5]
scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=[30, 60, 90], gamma=0.1)
# 30, 60, 90 epoch에서 lr을 0.1배로 감소3. ReduceLROnPlateau: 성능 개선이 멈추면 학습률 감소[^4][^3]
scheduler = optim.lr_scheduler.ReduceLROnPlateau(
optimizer, mode='min', factor=0.1, patience=5, verbose=True
)
# validation loss가 5 epoch 동안 개선되지 않으면 lr을 0.1배로 감소4. CosineAnnealingLR: Cosine 함수 형태로 학습률 변화[^5]
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
# 50 epoch 동안 cosine 형태로 lr 감소5. ExponentialLR: 지수적으로 학습률 감소[^5]
scheduler = optim.lr_scheduler.ExponentialLR(optimizer, gamma=0.95)
# 매 epoch마다 lr = lr * 0.95스케줄러 사용 시 주의사항
- optimizer.step() 후에 scheduler.step() 호출[^4]
- ReduceLROnPlateau는 validation metric을 인자로 전달
- epoch마다 또는 batch마다 호출 가능 (스케줄러에 따라 다름)
Validation Set의 중요성
Validation set은 모델의 일반화 성능을 평가하고 과적합을 방지하는 데 필수적입니다:[^6][^1]
- 조기 종료(Early Stopping): validation loss가 증가하면 학습 중단
- 하이퍼파라미터 튜닝: 최적의 설정 찾기
- 모델 선택: 여러 모델 중 가장 좋은 성능의 모델 선택
일반적으로 Train:Validation:Test = 70:15:15 또는 80:10:10 비율로 분할합니다.[^7]
체크포인트 저장 전략
# 최고 성능 모델만 저장
if val_acc > best_acc:
best_acc = val_acc
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'scheduler_state_dict': scheduler.state_dict(),
'best_acc': best_acc,
}, 'best_model.pth')
# 일정 epoch마다 저장
if (epoch + 1) % 10 == 0:
torch.save({...}, f'checkpoint_epoch_{epoch+1}.pth')결론
이 템플릿 코드는 실전에서 바로 활용 가능한 완전한 학습 파이프라인을 제공합니다. Train/Validation/Test 분할, 데이터 증강, 학습률 스케줄러, 체크포인트 저장 등 모든 필수 요소가 포함되어 있습니다. 자신의 데이터와 모델에 맞게 config 부분만 수정하면 즉시 사용할 수 있습니다. 이 코드를 기반으로 다양한 딥러닝 프로젝트에 도전해보세요!
728x90
반응형
