Callback 함수란
- 특정 이벤트가 발생할 때 실행되는 함수
- 모델 성능이 더 진전 없을 때, 학습 중단.
- 에폭 마칠 때마다, 중간 결과값 저장
1. EarlyStopping
성능 진전이 없어서 조기종료
성능 평가: 최고값 + delta
# monitor : val acc
if val_acc > best_accuracy + min_delta: # 최고값 + delta 를 넘어라
counter = 0 # MAX달성시(갱신하면) 초기화
else:
counter += 1
# Early Stopping
if counter >= patience:
break
2. 모델 저장하기
- 파라미터만 저장
- 어떤 모델이었는지 기억하고 있어야 함.
- 사용했던 모델 가져와서 load
- 저장:
torch.save(model.state_dict(), 'model.pth') - 로드:
model = MyModel()
model.load_state_dict(torch.load('model.pth')) - 파라미터 + 체크포인트 저장
- 어떤 모델이었는지 기억하고 있어야
- 체크포인트는 딕셔너리 형태로 저장
- 저장:
torch.save({ 'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'loss': loss,
}, 'checkpoint.pth') - 로드:
- from model import MyModel
model = MyModel()
checkpoint = torch.load('checkpoint.pth')
model.load_state_dict(checkpoint['model_state_dict'])
optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
epoch = checkpoint['epoch']
loss = checkpoint['loss'] - 모델 저장
- 저장:
torch.save(model, 'full_model.pth') - 로드:
model = torch.load('full_model.pth')
3. Learning Rate 스케줄
Learning Rate(weight이동간격) Schedule(바꿔주자)
Callbacks API 종류
1) StepLR:
- scheduler = lr_scheduler.StepLR(optimizer, step_size=30, gamma=0.1)
- 특정 에폭(30)마다 LR을 낮춘다.
2) MultiStepLR:
- scheduler = lr_scheduler.MultiStepLR(optimizer, milestones=[30,80], gamma=0.1)
- 몇몇 에폭마다 LR을 낮춘다.
3) ReduceLROnPlateau:
- scheduler = lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', patience=10, factor=0.1)
- 고원(Plateau)을 만나면, 10에폭동안 개선 없으면, LR을 낮춘다.
- global min 에서 '진동'하는 것을 멈추기 위해 사용한다.
4) CosineAnnealingLR
- scheduler = lr_scheduler.CosineAnnealingLR(optimizer, T_max=200)
- 코사인 주기별로 LR을 낮춘다.
4. Tensorboard : 모델 학습 결과 시각화
SummaryWriter 기본 기능:
- add_scalar: 단일 값 기록
- add_scalars: 여러 값을 한 그래프에 기록
from torch.utils.tensorboard import SummaryWriter
# 1. 기본적인 사용
writer = SummaryWriter('runs/ex_1') # 로그 저장 경로
# 2. 학습 루프에서 사용 예시
def train(model, train_loader, criterion, optimizer, epoch):
model.train()
for batch_idx, (data, target) in enumerate(train_loader):
output = model(data)
loss = criterion(output, target)
# 매 배치마다 기록
writer.add_scalar('Loss/train_step',
loss.item(),
epoch * len(train_loader) + batch_idx)
# 사용 후 반드시 닫기
writer.close()
실습
import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader
from torch.utils.tensorboard import SummaryWriter
import os
from datetime import datetime
from tqdm.notebook import tqdm
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# 데이터 증강 및 전처리 정의
transform_train = transforms.Compose([
transforms.RandomCrop(32, padding=4), # 랜덤 크롭
transforms.RandomHorizontalFlip(), # 좌우 뒤집기
transforms.AutoAugment(), # 자동 증강
transforms.ToTensor(), # 텐서 변환
transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)) # 정규화
])
# testset은 증강을 하지 않는다.
# 일반적으로 훈련과 검증 데이터를 나눈 뒤에 훈련데이터에만 증강을 한다. (증강의 효과를 확인하기 위해)
transform_test = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010))
])
# CIFAR10 데이터셋 로드
trainset = torchvision.datasets.CIFAR10(root='./data', train=True, download=True, transform=transform_train)
testset = torchvision.datasets.CIFAR10(root='./data', train=False, download=True, transform=transform_test)
trainloader = DataLoader(trainset, batch_size=128, shuffle=True, num_workers=2)
testloader = DataLoader(testset, batch_size=128, shuffle=False, num_workers=2)
# CNN 모델 정의
class SimpleCNN(nn.Module):
def __init__(self):
super(SimpleCNN, self).__init__()
self.backbone = nn.Sequential(
nn.Conv2d(3, 64, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2),
nn.Conv2d(64, 128, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2),
nn.Conv2d(128, 256, kernel_size=3, padding=1),
nn.ReLU(inplace=True),
nn.MaxPool2d(kernel_size=2, stride=2)
)
self.head = nn.Sequential(
nn.Linear(256 * 4 * 4, 512),
nn.ReLU(inplace=True),
nn.Dropout(0.5),
nn.Linear(512, 10)
)
def forward(self, x):
x = self.backbone(x)
x = x.view(x.size(0), -1)
x = self.head(x)
return x
# 학습 파라미터
EPOCHS = 2
# 모델, 손실함수, 옵티마이저 설정
model = SimpleCNN().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
#주요 콜백 선언
# Learning Rate Scheduler - 3에폭 동안 개선 없으면 낮춘다.
lr_scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', patience=3, factor=0.1, verbose=True)
# TensorBoard : 모델 학습 결과 시각화
current_time = datetime.now().strftime("%Y%m%d-%H%M%S")
log_dir = f'runs/{current_time}'
writer = SummaryWriter(log_dir)
checkpoint_dir = './checkpoints' # 체크포인트 디렉토리
os.makedirs(checkpoint_dir, exist_ok=True)
# EarlyStopping 조기종료 - 5번까지 (best_accuracy + min_delta) 를 비교
best_accuracy = 0
patience = 5
counter = 0
# 학습 함수
def train(epoch):
model.train()
train_loss = 0
correct = 0
total = 0
for batch_idx, (inputs, targets) in tqdm(enumerate(trainloader), total=len(trainloader)):
inputs, targets = inputs.to(device), targets.to(device)
optimizer.zero_grad()
outputs = model(inputs)
loss = criterion(outputs, targets)
loss.backward()
optimizer.step()
train_loss += loss.item()
_, predicted = outputs.max(1)
total += targets.size(0)
correct += predicted.eq(targets).sum().item()
accuracy = 100 * correct / total
# Tensorboard
writer.add_scalar('Training Loss', train_loss/len(trainloader), epoch)
writer.add_scalar('Training Accuracy', accuracy, epoch)
return train_loss/len(trainloader), accuracy
# 검증 함수
def evaluate(epoch):
model.eval()
val_loss = 0
correct = 0
total = 0
with torch.no_grad():
for batch_idx, (inputs, targets) in enumerate(testloader):
inputs, targets = inputs.to(device), targets.to(device)
outputs = model(inputs)
loss = criterion(outputs, targets)
val_loss += loss.item()
_, predicted = outputs.max(1)
total += targets.size(0)
correct += predicted.eq(targets).sum().item()
accuracy = 100 * correct / total
# Tensorboard
writer.add_scalar('Validation Loss', val_loss/len(testloader), epoch)
writer.add_scalar('Validation Accuracy', accuracy, epoch)
return val_loss/len(testloader), accuracy
# 메인 학습 루프
for epoch in range(EPOCHS):
print(f'\nEpoch: {epoch+1}/{EPOCHS}')
train_loss, train_acc = train(epoch)
val_loss, val_acc = evaluate(epoch)
# Learning Rate Scheduler
lr_scheduler.step(val_loss)
# lr_scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode='min', patience=3, factor=0.1, verbose=True)
# Early Stopping & Model Save
if val_acc > best_accuracy:
best_accuracy = val_acc
torch.save({
'epoch': epoch,
'model_state_dict': model.state_dict(),
'optimizer_state_dict': optimizer.state_dict(),
'accuracy': best_accuracy
}, os.path.join(checkpoint_dir, 'best_model.pth'))
counter = 0 # 조기종료 변수 초기화
else:
counter += 1
# Early Stopping
if counter >= patience:
print(f'Early stopping triggered at epoch {epoch+1}')
break
# best_accuracy = 0
# patience = 5
# counter = 0
# 학습종료
writer.close() # TensorBoard 종료
print(f"최종 최고 정확도: {best_accuracy:.2f}%")
print(f"모델 체크포인트 저장 경로: {checkpoint_dir}/best_model.pth")
print(f"TensorBoard 로그 경로: {log_dir}")
#tensorboard
%load_ext tensorboard
%tensorboard --logdir=runs
Tags:
AI개발_교육
