[Project] Classification Model 구현해보기 (ResNet)

Classification 모델을 구현해보자 먼저 Train과 Test 구조를 정의

 

Import

import torch
import torch.nn as nn
import torch.nn.functional as F
import torchvision
import torchvision.transforms as transforms
import torch.optim as optim

import os
import matplotlib.pyplot as plt

 

torch.nn as nn

 

• 신경망 레이어와 모델 구축을 위한 모듈.
• 예: nn.Conv2d, nn.Linear, nn.BatchNorm2d 등을 사용해 레이어 정의.

 

torch.nn.functional as F

 

• 활성화 함수나 pooling 같은 연산을 함수형(functional)으로 제공.
• 예: F.relu(x), F.max_pool2d(x, 2).

 

 

torch.optim as optim

- SGD, Adam, RMSprop 등 다양한 Optimizer 제공.

 

torchvision

- CIFAR-10, ImageNet 등 유명 데이터셋과 미리 정의된 모델들(ResNet, VGG 등) 제공.

 

def accuracy(output, target, topk=(1, )):
    with torch.no_grad():
        maxk = max(topk)
        batch_size = target.size(0)

        _, pred = output.topk(maxk, 1, True, True)
        pred = pred.t()
        correct = pred.eq(target.view(1, -1).expand_as(pred))

        acc = []
        num_cor = []
        for k in topk:
            correct_k = correct[:k].view(-1).float().sum(0, keepdim=True)
            num_cor.append(correct_k.clone())
            acc.append(correct_k.mul_(1 / batch_size))
    return acc, num_cor

 

top-k 분류 정확도를 계산

• topk=(1,) → 일반적인 top-1 정확도.
• topk=(1,5) → top-1 정확도와 top-5 정확도를 동시에 계산 가능.

def initialize_weights(module):
    if isinstance(module, nn.Conv2d):
        nn.init.kaiming_normal_(module.weight.data, mode='fan_out')
    elif isinstance(module, nn.BatchNorm2d):
        module.weight.data.fill_(1)
        module.bias.data.zero_()
    elif isinstance(module, nn.Linear):
        module.bias.data.zero_()

 

만약 레이어가 합성곱(Conv2D) 레이어라면 He(Kaiming) 초기화 [각 층의 가중치를 입력 노드 수의 역수에 비례하는 분산을 가진 분포에서 무작위로 선택하여 초기화]

https://resultofeffort.tistory.com/114

[Deep learning] 가중치 초기화(weight initialization) (feat. Xavier, He,normal, uniform)

 

fan_out 모드: 출력 채널 기준으로 분산을 맞춤.

만약 레이어가 배치 정규화(BatchNorm2D) 라면 배치 정규화 레이어의 scale(γ)은 1로, shift(β)는 0으로 초기화. = 정규화된 출력 그대로 weight=1, bias=0.

만약 레이어가 완전연결층(FC, Linear) 라면 bias 항을 0으로 초기화.

 

정규화된 이미지를 다시 원래 값으로 되돌리는 함수

 

 

def inverse_normalize(tensor, mean=(0.4914, 0.4822, 0.4465), std=(0.2023, 0.1994, 0.2010)):
    for t, m, s in zip(tensor, mean, std):
        t.mul_(s).add_(m)
    return tensor

 

 

• tensor: 정규화된 이미지 텐서 (C, H, W 형태).
• mean: 정규화 시 사용한 평균값 (CIFAR-10 채널 평균).
• std: 정규화 시 사용한 표준편차 (CIFAR-10 채널 표준편차).

일반적으로 학습 전에 이미지를 Normalize(mean, std)로 변환함

이렇게 되면 픽셀 값이 -1~1 근처의 값으로 바뀌어서 시각화할 때 원본 이미지와 다르게 보임 따라서 다시 원래 값으로 돌려줄 필요가 있기 때문에 정의

 

Train

def train(epochs):
    best_acc = 0.0
    print('[*] start training')
    for epoch in range(1, epochs):
        model.train() #입력과 출력이 다를 댸
        for step, (data, targets) in enumerate(trainloader): # trainloader에서 데이터(batch) 꺼내오기.
            data = data.to(device, dtype=torch.float)
            targets = targets.to(device)
            optimizer.zero_grad() # optimizer.zero_grad() → 이전 배치에서 계산된 gradient 초기화.

            outputs = model(data) # 모델에 입력 넣고 예측값 outputs
            loss = nn.CrossEntropyLoss(reduction='mean')(outputs, targets) # 손실 함수

            loss.backward() # gradient 계산 (backpropagation).
            optimizer.step() # Adam optimizer로 가중치 업데이트.

            loss = loss.item()
            acc, _ = accuracy(outputs, targets)
            acc = acc[0].item()

            if step % 10 == 0: # 중간 학습 상황 출력.
                print(f'[Epoch {epoch}/{epochs}, Step {step}/{len(trainloader)}] Loss {loss:.4f}, Accuracy {acc:.4f}')
        scheduler.step() # 에폭이 끝날 때마다 학습률(learning rate) 조정


        model.eval() # 평가 모드 전환 (Dropout 비활성화, BatchNorm 고정).
        total_cor = 0
        total_samples = 0

        with torch.no_grad():
            for step, (data, targets) in enumerate(testloader):
                data = data.to(device, dtype=torch.float)
                targets = targets.to(device)
                outputs = model(data)
                _, num_cor = accuracy(outputs, targets)
                num_cor = num_cor[0].item()
                total_samples += data.size(0)
                total_cor += num_cor
            acc = total_cor / total_samples
            print(f'Epoch {epoch} : Test Accuracy {acc:.4f}')

        if acc > best_acc: # 모델 저장 새로운 최고 정확도 달성 시 → 모델 저장.
            print('[*] model saving...')
            state = {
                'model': model.state_dict(),#모델 가중치만 저장
                'acc': acc,
                'epoch': epoch,
            }
            if not os.path.isdir('ckpt_0'):
                os.mkdir('ckpt_0')

            path = f'ckpt_0/model_{model.__class__.__name__}_state_{epoch:03d}_{acc:.4f}.st'
            torch.save(state, path)
            best_acc = acc
    print(f'Best Test Accuracy {best_acc:.4f}')

 

 

 

best_acc = 최고 test accuracy 저장

model.train() → 학습 모드 전환. (Dropout, BatchNorm 등 학습용 동작 활성화)

 

def test(ckpt_path):
    print(f'[*] load {ckpt_path}')
    model.eval() #평가 모드 전환
    state_dict = torch.load(ckpt_path) # 학습된 모델 checkpoint 파일 불러오기.
    model.load_state_dict(state_dict['model'], strict=True) # 저장된 가중치 불러오기.
    # (strict=True → 저장된 키와 모델 구조가 정확히 일치해야 함)

    total_cor = 0
    total_samples = 0
    with torch.no_grad():
        for step, (data, targets) in enumerate(testloader):
            data = data.to(device, dtype=torch.float)
            targets = targets.to(device)
            outputs = model(data)
            _, num_cor = accuracy(outputs, targets)
            num_cor = num_cor[0].item()
            total_samples += data.size(0)
            total_cor += num_cor
        acc = total_cor / total_samples # 전체 정답 수 ÷ 전체 샘플 수 → 최종 Test Accuracy 출력.
        print(f'Test Accuracy {acc:.4f} of Loaded Model {model.__class__.__name__}')

        # Visualize
        images = []
        pred_classes = []
        labels = []
        pred = outputs.topk(1, dim=1, largest=True, sorted=True) # 각 샘플의 예측 클래스(Top-1) 뽑기
        fig, axes = plt.subplots(3, 3, figsize=(15, 5))  # 3 row, 3 columns
        axes = axes.flatten()
        for k in range(9):  # check only the first 9 images
            images.append(inverse_normalize(data[k, :, :, :]).detach().cpu().permute(1, 2, 0).numpy())
            pred_classes.append(classes[pred[1][k].item()])
            labels.append(classes[targets[k].item()])
        for k, image in enumerate(images):
            axes[k].imshow(image)
            axes[k].axis('off')
            axes[k].set_title(f'label: {labels[k]}, pred: {pred_classes[k]}', fontsize=10)
        plt.tight_layout()
        plt.show()

 

• 저장된 모델 checkpoint 로드
• 전체 test dataset에 대해 정확도 계산
• 일부 이미지를 예측 결과 vs 실제 라벨로 시각화

Main - train

if __name__ == '__main__':
    # 학습에 사용할 디바이스 설정 (CUDA가 있으면 GPU, 없으면 CPU)
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    print(device)

    # --------------------
    # 1. 데이터 준비
    # --------------------
    print('[*] preparing data')
    # 학습 데이터 변환 (데이터 증강 포함)
    transform_train = transforms.Compose([
        transforms.RandomCrop(32, padding=4),         # 랜덤 크롭 (여백 포함)
        transforms.RandomHorizontalFlip(),            # 랜덤 좌우 반전
        transforms.ToTensor(),                        # 텐서 변환
        transforms.Normalize((0.4914, 0.4822, 0.4465),# 채널별 평균 정규화
                             (0.2023, 0.1994, 0.2010))# 채널별 표준편차 정규화
    ])

    # 테스트 데이터 변환 (증강 X, 정규화만 적용)
    transform_test = transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize((0.4914, 0.4822, 0.4465),
                             (0.2023, 0.1994, 0.2010))
    ])

    # CIFAR-10 학습 데이터셋 (5만 장)
    trainset = torchvision.datasets.CIFAR10(
        root='./data', train=True, download=True, transform=transform_train)
    # 학습 데이터 로더
    trainloader = torch.utils.data.DataLoader(
        trainset, batch_size=128, shuffle=True, num_workers=2)

    # CIFAR-10 테스트 데이터셋 (1만 장)
    testset = torchvision.datasets.CIFAR10(
        root='./data', train=False, download=True, transform=transform_test)
    # 테스트 데이터 로더
    testloader = torch.utils.data.DataLoader(
        testset, batch_size=100, shuffle=False, num_workers=2)

    # CIFAR-10 클래스 이름
    classes = ('plane', 'car', 'bird', 'cat', 'deer',
               'dog', 'frog', 'horse', 'ship', 'truck')

    # --------------------
    # 2. 모델 정의
    # --------------------
    print('[*] building model')
    # model = ToyNetwork()   # 간단한 CNN (연습용)
    model = ResNet50()       # 실제 학습에 사용할 ResNet50 모델
    model.to(device)         # GPU/CPU 디바이스에 올리기

    # --------------------
    # 3. 손실 함수
    # --------------------
    criterion = nn.CrossEntropyLoss(reduction='mean') # 다중 분류용 손실 함수

    # --------------------
    # 4. 최적화 도구 & 학습률 스케줄러
    # --------------------
    epochs = 100
    params = model.parameters()
    optimizer = optim.Adam(params, lr=1e-3) # Adam Optimizer (lr=0.001)
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
        optimizer, T_max=epochs)             # Cosine Annealing 학습률 스케줄러

    # --------------------
    # 5. 학습 시작
    # --------------------
    train(epochs)  # 학습 루프 실행

 

 

Test

    directory = './ckpt_0'
    ckpt_list = os.listdir(directory) # ckpt_0 폴더 안의 모든 파일 목록 불러오기.
    ckpt_list = [f for f in ckpt_list if os.path.isfile(os.path.join(directory, f)) and model.__class__.__name__ in f]
    # 리스트 컴프리헨션으로 필터링 파일 이름에 현재 모델 이름 포함된 것만 선택
    ckpt_list.sort()
    ckpt_path = os.path.join(directory, ckpt_list[-1]) # 가장 최신 checkpoint 파일 경로 얻기.
    print(ckpt_path)
    test(ckpt_path=ckpt_path)

 

 

Model ( ResNet50 )

2025.09.05 - [Computer Vision1/Paper reviews] - [ILSVRC 논문 정리해 보기] VGGNet, GoogleNet, ResNet

[ILSVRC 논문 정리해 보기] VGGNet, GoogleNet, ResNet

 

ResNet을 구성하는 기본 단위 블록

class BasicBlock(nn.Module):
    expansion = 1

    def __init__(self, in_planes, planes, stride=1):
        super(BasicBlock, self).__init__()
        self.conv1 = nn.Conv2d(
            in_planes, planes, kernel_size=3, stride=stride, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(planes)
        self.conv2 = nn.Conv2d(planes, planes, kernel_size=3,
                               stride=1, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(planes)

        self.shortcut = nn.Sequential()
        if stride != 1 or in_planes != self.expansion*planes:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_planes, self.expansion*planes,
                          kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(self.expansion*planes)
            )

    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))
        out += self.shortcut(x)
        out = F.relu(out)
        return out


class Bottleneck(nn.Module):
    expansion = 4

    def __init__(self, in_planes, planes, stride=1):
        super(Bottleneck, self).__init__()
        self.conv1 = nn.Conv2d(in_planes, planes, kernel_size=1, bias=False)
        self.bn1 = nn.BatchNorm2d(planes)
        self.conv2 = nn.Conv2d(planes, planes, kernel_size=3,
                               stride=stride, padding=1, bias=False)
        self.bn2 = nn.BatchNorm2d(planes)
        self.conv3 = nn.Conv2d(planes, self.expansion *
                               planes, kernel_size=1, bias=False)
        self.bn3 = nn.BatchNorm2d(self.expansion*planes)

        self.shortcut = nn.Sequential()
        if stride != 1 or in_planes != self.expansion*planes:
            self.shortcut = nn.Sequential(
                nn.Conv2d(in_planes, self.expansion*planes,
                          kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(self.expansion*planes)
            )

    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = F.relu(self.bn2(self.conv2(out)))
        out = self.bn3(self.conv3(out))
        out += self.shortcut(x)
        out = F.relu(out)
        return out

 

- BasicBlock

class BasicBlock(nn.Module):
    expansion = 1

• conv1: 3×3 Conv, stride는 입력에 따라 다름.
• bn1: BatchNorm → 학습 안정화.
• conv2: 또 다른 3×3 Conv.
• bn2: BatchNorm.
• shortcut: 잔차 연결(residual connection).
  • 기본은 단순히 identity 연결.
  • 하지만 stride가 바뀌거나(in_planes ≠ planes) 차원이 달라지면 →
    1×1 Conv + BatchNorm으로 차원 맞춰줌.

 

forward()

out = F.relu(self.bn1(self.conv1(x)))
out = self.bn2(self.conv2(out))
out += self.shortcut(x)     # 입력 + 변환된 출력 → skip connection
out = F.relu(out)

 

• 입력을 그대로 더하는 skip connection(잔차 연결)

- Bottleneck

class Bottleneck(nn.Module):
    expansion = 4

 

• ResNet-50 이상에서 사용하는 블록.
• expansion=4 → 출력 채널 수가 입력보다 4배 확장.
• conv1: 1×1 Conv → 채널 축소 (차원 줄여 연산량 감소).
• conv2: 3×3 Conv → 실제 feature 추출.
• conv3: 1×1 Conv → 채널 확장 (planes → planes×4).
• 각 conv 뒤에 BatchNorm.
• shortcut: 마찬가지로 stride나 채널이 다르면 1×1 Conv로 매핑.

forward()

out = F.relu(self.bn1(self.conv1(x)))  # 채널 축소
out = F.relu(self.bn2(self.conv2(out)))# 3×3 conv
out = self.bn3(self.conv3(out))        # 채널 확장
out += self.shortcut(x)                # skip connection
out = F.relu(out)

 

 

 

• BasicBlock: 2개의 3×3 Conv → ResNet-18, ResNet-34에 사용.
• Bottleneck: 1×1 → 3×3 → 1×1 Conv 구조 → ResNet-50 이상에서 사용.

class ResNet(nn.Module):
    def __init__(self, block, num_blocks, num_classes=10):
        super(ResNet, self).__init__()
        self.in_planes = 64

        self.conv1 = nn.Conv2d(3, 64, kernel_size=3,
                               stride=1, padding=1, bias=False)
        self.bn1 = nn.BatchNorm2d(64)
        self.layer1 = self._make_layer(block, 64, num_blocks[0], stride=1)
        self.layer2 = self._make_layer(block, 128, num_blocks[1], stride=2)
        self.layer3 = self._make_layer(block, 256, num_blocks[2], stride=2)
        self.layer4 = self._make_layer(block, 512, num_blocks[3], stride=2)
        self.linear = nn.Linear(512*block.expansion, num_classes)

    def _make_layer(self, block, planes, num_blocks, stride):
        strides = [stride] + [1]*(num_blocks-1)
        layers = []
        for stride in strides:
            layers.append(block(self.in_planes, planes, stride))
            self.in_planes = planes * block.expansion
        return nn.Sequential(*layers)

    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.layer1(out)
        out = self.layer2(out)
        out = self.layer3(out)
        out = self.layer4(out)
        out = F.avg_pool2d(out, 4)
        out = out.view(out.size(0), -1)
        out = self.linear(out)
        return out


def ResNet18():
    return ResNet(BasicBlock, [2, 2, 2, 2])


def ResNet34():
    return ResNet(BasicBlock, [3, 4, 6, 3])


def ResNet50():
    return ResNet(Bottleneck, [3, 4, 6, 3])


def ResNet101():
    return ResNet(Bottleneck, [3, 4, 23, 3])


def ResNet152():
    return ResNet(Bottleneck, [3, 8, 36, 3])


def test():
    net = ResNet18()
    y = net(torch.randn(1, 3, 32, 32))
    print(y.size())

 

class ResNet(nn.Module):

def __init__(self, block, num_blocks, num_classes=10):

super(ResNet, self).__init__()

self.in_planes = 64

• block: 어떤 블록을 쓸지 (BasicBlock or Bottleneck).
• num_blocks: 각 레이어에 몇 개의 블록을 쌓을지 지정.
• num_classes: 분류할 클래스 수 (CIFAR-10 → 10).

수치 표현과 데이터 전처리

• 딥러닝 프레임워크의 기본 자료형은 float32 (32비트 부동소수점).
• 일부 최적화/압축 기법으로 int4, float16 같은 quantization을 적용할 수 있음.
• 데이터 정규화(Normalize):
  • 각 채널(R, G, B)의 평균과 표준편차로 정규화.
  • 보통 ImageNet의 통계값(예: 평균 [0.4914, 0.4822, 0.4465], 표준편차 [0.2023, 0.1994, 0.2010])을 많이 사용.
• 학습(train) 때는 데이터 어그멘테이션(RandomCrop, Flip 등)을 적용하지만, 테스트(test) 시에는 동일한 Normalize만 적용.

데이터셋 & 데이터로더

• 데이터셋 클래스 작성법:
  • torch.utils.data.Dataset 상속.
  • 필수 구현 메서드:
    • __init__: 경로/데이터/라벨 로드 및 초기화.
    • __getitem__: 인덱스로 데이터와 레이블을 반환.
    • __len__: 전체 데이터 크기 반환.
• 데이터로더(DataLoader):
  • 배치 단위로 데이터를 꺼내 학습 파이프라인에 전달.
  • 인자: dataset, batch_size, shuffle, num_workers.
  • num_workers: CPU가 GPU 학습 중에도 병렬로 데이터를 전처리해서 성능 향상.

모델 구조 (PyTorch)

• 커스텀 네트워크는 nn.Module 상속.
  • __init__에 레이어 정의.
  • forward()에서 순전파 정의 (연산 그래프 구성).
• 학습 가능한 텐서(requires_grad=True)는 **파라미터(필터, 가중치)**이고, 입력 데이터는 보통 requires_grad=False.
• Convolution Layer 예시:
  • Conv2d(3, 32, kernel_size=7) → 파라미터 크기: (32, 3, 7, 7).
• 1x1 Convolution은 Fully Connected Layer와 동일한 효과.

미분 그래프(Autograd)

• PyTorch 핵심: 자동 미분(Autograd).
• 각 텐서는 다음 3개 멤버를 가짐:
  1. requires_grad: 학습 여부 (True/False).
  2. grad_fn: 어떤 연산을 거쳤는지(미분 함수 추적).
  3. grad: 역전파 시 계산된 기울기 값.
• Forward Pass = 그래프 구성, Backward Pass = 기울기 전파 및 파라미터 업데이트.

학습 과정

1. 데이터 로더에서 배치 꺼냄.
2. 모델 forward → 출력값 계산.
3. 손실함수(loss) 계산 (CrossEntropyLoss 등).
4. loss.backward() → gradient 계산.
5. 옵티마이저(Adam, SGD)가 파라미터 업데이트.
6. 루프: (배치 → 옵티마이저 step) × (에폭 수 만큼 반복).

학습 & 추론 모드

• model.train(): 드롭아웃/배치정규화가 학습 모드로 동작.
• model.eval(): 추론 모드 (드롭아웃 비활성화, 배치정규화는 EMA 값 사용).

학습률(Learning Rate) 제어

• 기본: 일정한 lr (예: 0.001).
• 고급 기법:
  • Cosine Annealing: 코사인 함수 곡선 형태로 lr 감소.
  • Warm-up: 초기에 작은 lr로 시작해 점차 증가시킨 뒤 감소.

결과 시각화

 

모델의 더 자세한 정보는 이전에 정리한 포스팅을 참고하면 좋겠다.

2025.09.05 - [Computer Vision1/Paper reviews] - [ILSVRC 논문 정리해 보기] VGGNet, GoogleNet, ResNet

[ILSVRC 논문 정리해 보기] VGGNet, GoogleNet, ResNet