Deep Learning 진행 과정 (예제: pizza)

import torch from torch import nn, optim # PyTorch의 신경망 및 최적화 함수 from torchvision import datasets, transforms, models # 이미지 데이터셋과 모델 관련 기능 from torch.utils.data import DataLoader # 데이터 로딩 기능 from sklearn.metrics import confusion_matrix, accuracy_score # 혼동 행렬 및 정확도 계산 import seaborn as sns # 데이터 시각화 라이브러리 import matplotlib.pyplot as plt # 플롯을 그리기 위한 Matplotlib import numpy as np # NumPy 사용 import random # 난수 생성

# 시드 고정 함수 def set_seed(seed=42): random.seed(seed) # Python 기본 랜덤 시드 고정 np.random.seed(seed) # NumPy 랜덤 시드 고정 torch.manual_seed(seed) # PyTorch CPU 시드 고정 torch.cuda.manual_seed(seed) # PyTorch GPU 시드 고정 (CUDA) torch.cuda.manual_seed_all(seed) # PyTorch 멀티-GPU 시드 고정 (CUDA) torch.backends.cudnn.deterministic = True # CUDNN에서 난수 생성기가 동일한 결과를 보장 torch.backends.cudnn.benchmark = False # CUDNN 최적화 기능 비활성화 # 시드 고정 set_seed(42) # 배치 크기 설정 (배치 크기를 32로 조정) BATCH_SIZE = 32

# GPU가 사용 가능한지 확인하고 GPU로 설정 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # GPU 사용 가능하면 CUDA 사용 print(f"Using device: {device}")

# 데이터 증강 및 전처리 (학습 데이터에 다양한 변형 적용) train_transform = transforms.Compose([ transforms.RandomResizedCrop(224), # 랜덤하게 이미지 크롭 transforms.RandomHorizontalFlip(), # 좌우 반전 transforms.RandomRotation(10), # 이미지를 랜덤하게 회전 transforms.ColorJitter(), # 색상 변형 transforms.ToTensor(), # 이미지를 텐서로 변환 transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) # 이미지 정규화 ]) # 테스트 데이터는 크기 조정 및 정규화만 적용 test_transform = transforms.Compose([ transforms.Resize(256), # 이미지 크기를 256으로 조정 transforms.CenterCrop(224), # 중앙 크롭 transforms.ToTensor(), # 이미지를 텐서로 변환 transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]) # 정규화 ])

# DataLoader로 변환 (데이터셋을 배치 단위로 나누어 로드) train_loader = DataLoader(train_data, batch_size=BATCH_SIZE, shuffle=True) # 학습 데이터 로더 (shuffle 활성화) test_loader = DataLoader(test_data, batch_size=BATCH_SIZE, shuffle=False) # 테스트 데이터 로더

# EfficientNet-B0 모델 사용 (사전 학습된 모델 활용) model = models.efficientnet_b0(pretrained=True)

# 모델의 출력 레이어 변경 (사전 학습된 레이어는 동결되고, 마지막 레이어만 학습) num_features = model.classifier[1].in_features # 마지막 레이어의 입력 크기 model.classifier[1] = nn.Linear(num_features, len(train_data.classes)) # 출력 클래스 수로 레이어 변경

# 모델을 GPU로 이동 model = model.to(device)

# 클래스 불균형 처리 (클래스 가중치 적용) class_counts = torch.bincount(torch.tensor(train_data.targets)) # 각 클래스의 샘플 수 계산 class_weights = 1.0 / class_counts.float() # 클래스 가중치 계산 class_weights = class_weights.to(device) # 가중치를 GPU로 이동

# 손실 함수 및 옵티마이저 설정 (Adam 사용) loss_fn = nn.CrossEntropyLoss(weight=class_weights) # 클래스 가중치를 적용한 손실 함수 optimizer = optim.Adam(model.parameters(), lr=0.001) # Adam 옵티마이저 설정

# 학습률 스케줄러 (에포크마다 학습률을 줄여주는 StepLR 사용) scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=7, gamma=0.1) # 7 에포크마다 학습률 감소

# 학습 속도 개선을 위한 AMP(Mixed Precision Training) 사용 scaler = torch.cuda.amp.GradScaler() # 혼합 정밀도 학습을 위한 Scaler

# Early Stopping 설정 class EarlyStopping: def __init__(self, patience=5, delta=0): self.patience = patience # 몇 에포크 동안 개선이 없으면 학습 중단 self.delta = delta # 개선을 판단하는 기준 self.best_loss = None # 가장 낮은 손실 값 저장 self.counter = 0 # 개선이 없을 때 카운터 증가 self.early_stop = False # 학습 중단 여부 def __call__(self, val_loss): if self.best_loss is None: # 첫 에포크에서 손실 값을 설정 self.best_loss = val_loss elif val_loss > self.best_loss - self.delta: # 개선되지 않은 경우 self.counter += 1 # 카운터 증가 print(f"EarlyStopping counter: {self.counter} out of {self.patience}") if self.counter >= self.patience: # 개선이 없으면 학습 중단 self.early_stop = True else: # 개선된 경우 self.best_loss = val_loss self.counter = 0 # 카운터 초기화

class EarlyStopper(object): def __init__(self, num_trials, save_path): self.num_trials = num_trials self.trial_counter = 0 self.best_loss = np.inf self.save_path = save_path def is_continuable(self, model, loss): if loss < self.best_loss: # 현재 loss가 최고 loss보다 더 낮은 경우 self.best_loss = loss # 최고 loss를 현재 loss로 업데이트 self.trial_counter = 0 # 초기화 torch.save(model, self.save_path) # 최고 loss를 갖은 모델 저장 return True elif self.trial_counter + 1 < self.num_trials: # 현재 loss가 최고 loss보다 작은 경우 & max 시도횟수보다 현재 시도횟수가 작은 경우 self.trial_counter += 1 # 기존 시도횟수 + 1 return True else: # 현재 정확도가 최고 정확도보다 작은 경우 & 현재 시도횟수가 max 시도횟수보다 큰 경우 return False def get_best_model(self, device): return torch.load(self.save_path).to(device)

def train_batch(train_loader): for inputs, labels in train_loader: # 배치별 학습 inputs, labels = inputs.to(device), labels.to(device) # 데이터를 GPU로 이동 optimizer.zero_grad() # 옵티마이저의 기울기 초기화 with torch.cuda.amp.autocast(): # 혼합 정밀도 연산 (AMP) outputs = model(inputs) # 모델의 예측값 loss = loss_fn(outputs, labels) # 손실 계산 scaler.scale(loss).backward() # 손실의 역전파 scaler.step(optimizer) # 옵티마이저 스텝 scaler.update() # Scaler 업데이트 running_loss += loss.item() * inputs.size(0) # 배치 손실을 누적 _, preds = torch.max(outputs, 1) # 예측값을 레이블로 변환 correct += (preds == labels).sum().item() # 맞은 예측 개수 누적 total += labels.size(0) # 총 샘플 개수 누적 # 학습 함수 (학습 과정을 추적 및 출력) def train_model(model, train_loader, test_loader, loss_fn, optimizer, scheduler, num_epochs=20, patience=5): early_stopping = EarlyStopping(patience=patience, delta=0.001) # Early Stopping 설정 for epoch in range(num_epochs): # 지정된 에포크 수만큼 반복 model.train() # 모델을 학습 모드로 전환 running_loss = 0.0 # 누적 손실 correct = 0 # 맞은 예측 개수 total = 0 # 총 샘플 개수 train_batch(train_loader) scheduler.step() # 학습률 스케줄러 업데이트 epoch_loss = running_loss / len(train_loader.dataset) # 에포크당 평균 손실 계산 epoch_acc = correct / total * 100 # 에포크당 정확도 계산 print(f"Epoch {epoch+1}/{num_epochs}, Loss: {epoch_loss:.4f}, Accuracy: {epoch_acc:.2f}%") # 평가를 통해 Early Stopping 적용 val_loss = evaluate_model(model, test_loader, loss_fn) # 평가 함수 호출 early_stopping(val_loss) # Early Stopping 체크 if early_stopping.early_stop: # Early Stopping 조건 충족 시 print("Early stopping") # 학습 중단 메시지 출력 break

# 평가 함수 (손실 값 반환) def evaluate_model(model, test_loader, loss_fn): model.eval() # 모델을 평가 모드로 전환 all_labels = [] # 실제 레이블 저장 리스트 all_preds = [] # 예측 레이블 저장 리스트 running_loss = 0.0 # 손실 초기화 with torch.no_grad(): # 평가 중에는 기울기 계산 안함 for inputs, labels in test_loader: inputs, labels = inputs.to(device), labels.to(device) # 데이터를 GPU로 이동 outputs = model(inputs) # 모델 예측 loss = loss_fn(outputs, labels) # 손실 계산 running_loss += loss.item() * inputs.size(0) # 손실 누적 _, preds = torch.max(outputs, 1) # 예측값을 레이블로 변환 all_labels.extend(labels.cpu().numpy()) # 실제 레이블 저장 all_preds.extend(preds.cpu().numpy()) # 예측 레이블 저장 val_loss = running_loss / len(test_loader.dataset) # 검증 손실 계산 acc = accuracy_score(all_labels, all_preds) # 정확도 계산 cm = confusion_matrix(all_labels, all_preds) # 혼동 행렬 계산 print(f"Validation Loss: {val_loss:.4f}, Test Accuracy: {acc * 100:.2f}%") # 검증 손실 및 정확도 출력 # 혼동 행렬 시각화 plt.figure(figsize=(8, 6)) # 플롯 크기 설정 sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=train_data.classes, yticklabels=train_data.classes) # 혼동 행렬 시각화 plt.xlabel('Predicted') # x축 라벨 plt.ylabel('Actual') # y축 라벨 plt.title('Confusion Matrix') # 제목 설정 plt.show() # 플롯 출력 return val_loss # 검증 손실 반환

# 모델 학습 train_model(model, train_loader, test_loader, loss_fn, optimizer, scheduler, num_epochs=20, patience=5) # 모델 학습 호출