과적합(Overfitting) 방지 기법

 

과적합 방지 기법

 

과적합

 

모델이 훈련 데이터에 너무 과하게 적합(overfit)되어, 새로운 데이터(검증/테스트 데이터)에 대해 일반화 성능이 낮아지는 현상

→ 훈련 데이터에선 높은 정확도를 보이지만, 실제 사용 환경에서는 성능이 떨어지는 문제가 발생

 

 

 

정규화와 드롭아웃

 

정규화 (Normalization)

 

데이터 분포를 일정한 범위로 조정, 학습 안정성과 수렴 속도를 향상시키고 과적합을 방지

 

배치 정규화 (Batch Normalization)

→ 각 미니배치에 대해 평균과 분산을 정규화

→ 학습 속도를 높이고, 과적합을 방지하는 데 도움

레이어 정규화 (Layer Normalization)

→ 각 레이어별로 정규화 수행 (RNN 등 순차 모델에서 자주 사용)

 

드롭아웃 (Dropout)

 

학습 중 임의로 일부 뉴런을 비활성화(drop) 하여 특정 뉴런에 의존하는 것을 방지하고 일반화 성능을 높임

학습 중에만 적용됨 (추론 시엔 적용 X)

일반적으로 fully-connected layer 이후에 적용

 

 

조기 종료와 데이터 증강

 

조기 종료 (Early Stopping)

검증 데이터의 손실(loss)이 더 이상 감소하지 않을 경우 학습 중단

너무 오래 학습하여 생기는 과적합 방지

학습 과정에서 검증 손실이 일정 에포크 동안 감소하지 않으면 학습을 중단함

 

데이터 증강 (Data Augmentation)

데이터의 다양성을 인위적으로 증가시켜 일반화 성능 향상

주로 이미지, 음성, 텍스트 데이터에서 사용

 

예:

• 이미지 → 회전, 확대/축소, 색상 조정, 좌우 반전 등
• 텍스트 → 동의어 대체, 순서 변경 등

 

범주	기법	설명
데이터	데이터 증강, 데이터 양 늘리기	일반화 성능 향상
모델 구조	모델 단순화, 드롭아웃, 정규화	복잡도 감소 및 안정화
학습 방법	조기 종료(Early Stopping), 정규화 손실 적용	학습 과정을 제어하여 일반화 향상
기타	앙상블, 교차 검증 등	다양한 모델을 결합하거나 평가 방식 개선

 

 

 

과적합 방지 기법 PyTorch 실습

 

• 데이터 증강
• 배치 정규화(BatchNorm)
• 드롭아웃(Dropout)
• 조기 종료(EarlyStopping)

 

PyTorch 및 필요한 라이브러리 임포트

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
import numpy as np

 

 

데이터셋 로드 및 전처리 (데이터 증강 포함)

# 학습용 데이터: 데이터 증강 적용
transform_train = transforms.Compose([
    transforms.RandomHorizontalFlip(),
    transforms.RandomCrop(32, padding=4),
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

# 테스트용 데이터: 정규화만 적용
transform_test = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])

# CIFAR-10 데이터셋
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=64, shuffle=True)
testloader = DataLoader(testset, batch_size=64, shuffle=False)

 

 

모델 정의 (드롭아웃 + 배치 정규화 적용)

# Residual Block 정의
class Block(nn.Module):
    def __init__(self, in_ch, out_ch, stride=1):
        """
        Residual Block (잔차 블록)
        - 두 개의 3x3 컨볼루션 레이어(conv1, conv2)와 Batch Normalization을 포함
        - 입력이 shortcut을 통해 출력에 더해지는 구조

        Args:
        - in_ch (int): 입력 채널 수
        - out_ch (int): 출력 채널 수
        - stride (int, optional): 첫 번째 컨볼루션의 stride (기본값: 1)
        """
        super(Block, self).__init__()
        # 첫 번째 3x3 컨볼루션 레이어 (stride 적용 가능)
        self.conv1 = nn.Conv2d(
            in_ch, out_ch, kernel_size=3, stride=stride, padding=1, bias=False
        )
        self.bn1 = nn.BatchNorm2d(out_ch)  # 배치 정규화
        # 두 번째 3x3 컨볼루션 레이어 (채널 수 유지, stride=1)
        self.conv2 = nn.Conv2d(
            out_ch, out_ch, kernel_size=3, stride=1, padding=1, bias=False
        )
        self.bn2 = nn.BatchNorm2d(out_ch)  # 배치 정규화

        # 입력과 출력 채녈이 다를 경우 1x1 Conv를 사용하여 차원 맞추기
        self.skip_connection = nn.Sequential()
        if stride != 1 or in_ch != out_ch:
            self.skip_connection = nn.Sequential(
                nn.Conv2d(in_ch, out_ch, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(out_ch),  # 배치 정규화
            )

    def forward(self, x):
        """
        Forward 함수 (잔차 학습 적용)
        - 입력 데이터 (x)를 컨볼루션과 배치 정규화를 거쳐 변환
        - Shortcut Connection을 통해 원래 입력(x)을 출력에 더함
        """
        # 첫 번째 컨볼루션 + ReLU 활성화 함수
        output = F.relu(self.bn1(self.conv1(x)))
        # 두 번째 컨볼루션 후 배치 정규화
        output = self.bn2(self.conv2(output))
        # shortcut 경로 출력과 현재 블록의 출력 더하기
        output += self.skip_connection(x)  # Residual 연결
        # 최종 ReLU 활성화 함수 적용
        output = F.relu(output)
        return output


# ResNet 모델 정의
class CustomResNet(nn.Module):
    def __init__(self, block, layers, num_classes=10):
        """
        Custom ResNet 모델
        - 첫 번째 컨볼루션 레이어와 배치 정규화 적용
        - Residual Block을 쌓아 네트워크 구성
        - 최종적으로 Fully Connected Layer를 통해 분류 수행

        Args:
        - block (nn.Module): Residual Block 클래스
        - layers (list): 각 단계에서 사용할 Residual Block의 개수
        - num_classes (int, optional): 최종 분류할 클래스 개수 (기본값: 10)
        """
        super(CustomResNet, self).__init__()

        self.initial_channels = 64  # 첫 번째 레이어의 입력 채널 수 정의

        # 첫 번째 컨볼루션 레이어 (입력: 3채널 이미지)
        self.conv1 = nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(64)  # 배치 정규화

        # ResNet의 각 레이어 생성 (Residual Block을 여러 개 쌓음)
        self.layer1 = self._create_layer(block, 64, layers[0], stride=1)
        self.layer2 = self._create_layer(block, 128, layers[1], stride=2)
        self.layer3 = self._create_layer(block, 256, layers[2], stride=2)
        self.layer4 = self._create_layer(block, 512, layers[3], stride=2)

        # 평균 풀링 레이어 (AdaptiveAvgPool2d: 입력 크기에 관계없이 1x1로 변환)
        self.avgpool = nn.AdaptiveAvgPool2d((1, 1))

        # 드롭아웃 추가
        self.dropout = nn.Dropout(p=0.5)
        # 최종 완전 연결 레이어 (출력: num_classes)
        self.fc = nn.Linear(512, num_classes)

    # ResNet의 각 레이어를 생성하는 함수
    def _create_layer(self, block, out_ch, num_layers, stride):
        """
        Residual Block을 여러 개 쌓아 하나의 레이어를 구성하는 함수

        Args:
        - block (nn.Module): Residual Block 클래스
        - out_ch (int): 출력 채널 수
        - num_layers (int): 해당 레이어에서 사용할 Residual Block 개수
        - stride (int): 첫 번째 블록에서 적용할 stride 값

        Returns:
        - nn.Sequential: 구성된 Residual Layer
        """
        layers = []
        # 첫 번째 블록은 stride를 받을 수 있음
        layers.append(block(self.initial_channels, out_ch, stride))
        self.initial_channels = out_ch  # 다음 불록을 위해 채널 수 업데이트

        # 나머지 블록들은 기본 stride를 사용
        for _ in range(1, num_layers):
            layers.append(block(out_ch, out_ch))

        return nn.Sequential(*layers)

    def forward(self, x):
        """
        Forward 함수 (ResNet 모델의 순전파 과정)
        - 첫 번째 컨볼루션을 거친 후, 여러 개의 Residual Block을 순차적으로 통과
        - 마지막으로 Fully Connected Layer를 거쳐 최종 클래스 확률을 출력
        """
        # 첫 번째 컨볼루션 + ReLU 활성화 함수
        x = F.relu(self.bn1(self.conv1(x)))
        # 각 레이어를 순차적으로 통과
        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)
        # 평균 풀링 및 텐서의 차원 축소
        x = self.avgpool(x)
        x = torch.flatten(x, 1)  # 1차원 벡터로 변환
        x = self.dropout(x)  # 드롭아웃 적용
        # 최종 완전 연결 레이어를 통해 클래스별 예측값 출력
        x = self.fc(x)
        return x


# ResNet-18 모델 생성 (각 레이어의 블록 수 : [2, 2, 2, 2])
# model = CustomResNet(Block, [2, 2, 2, 2], num_classes=10)

 

 

손실 함수, 옵티마이저, 디바이스 설정

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
model = model.to(device)

 

 

조기 종료 클래스

import numpy as np
import torch


class EarlyStopping:
    def __init__(self, patience=5, verbose=False, delta=0.0, path="checkpoint.pth"):
        """
        조기 종료 클래스
        Args:
            patience (int): 개선되지 않아도 참을 epoch 수
            verbose (bool): 개선 시 출력 여부
            delta (float): 개선으로 간주할 최소 변화량
            path (str): 모델 저장 경로
        """
        self.patience = patience
        self.verbose = verbose
        self.delta = delta
        self.path = path

        self.counter = 0
        self.best_score = None
        self.early_stop = False
        self.val_loss_min = np.inf

    def __call__(self, val_loss, model):
        score = -val_loss

        if self.best_score is None:
            self.best_score = score
            self._save_checkpoint(val_loss, model)

        elif score < self.best_score + self.delta:
            self.counter += 1
            if self.verbose:
                print(f"⏳ EarlyStopping counter: {self.counter}/{self.patience}")
            if self.counter >= self.patience:
                self.early_stop = True

        else:
            self.best_score = score
            self._save_checkpoint(val_loss, model)
            self.counter = 0

    def _save_checkpoint(self, val_loss, model):
        """최고 성능 모델 저장"""
        if self.verbose:
            print(f"Validation loss improved. Saving model to {self.path}")
        torch.save(model.state_dict(), self.path)
        self.val_loss_min = val_loss

 

 

학습 함수 (조기 종료 포함)

for epoch in range(num_epochs):
        model.train()
        running_loss, correct, total = 0.0, 0, 0

        for inputs, labels in trainloader:
            inputs, labels = inputs.to(device), labels.to(device)

            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            running_loss += loss.item()
            _, predicted = torch.max(outputs, 1)
            correct += (predicted == labels).sum().item()
            total += labels.size(0)

        train_loss = running_loss / len(trainloader)
        train_acc = 100 * correct / total
        train_losses.append(train_loss)
        train_accuracies.append(train_acc)

        # 검증 손실 계산
        model.eval()
        val_loss = 0.0
        with torch.no_grad():
            for inputs, labels in testloader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)
                val_loss += loss.item()
        val_loss /= len(testloader)
        val_losses.append(val_loss)

        print(
            f"[Epoch {epoch+1}] Train Loss: {train_loss:.4f}, Accuracy: {train_acc:.2f}%, Val Loss: {val_loss:.4f}"
        )
        # 조기 종료 체크
        early_stopping(val_loss, model)
        if early_stopping.early_stop:
            print("Early stopping")
            break
    # 가장 좋은 모델 불러오기
    model.load_state_dict(torch.load("checkpoint.pth"))

 

 

모델 평가

model.eval()
correct = 0
total = 0

with torch.no_grad():
    for images, labels in testloader:
        images, labels = images.to(device), labels.to(device)
        outputs = model(images)
        _, predicted = torch.max(outputs, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()

accuracy = 100 * correct / total
print(f"Test Accuracy: {accuracy:.2f}%")

"""
Test Accuracy: 89.78%
"""