Pytorch 6 : Implement CNN

 

오늘은 대표적인 CNN인 ResNet을 구현해보도록 하자.

ResNet은 마이크로소프트에서 개발한 모델로 'Deep Residual learning for Image recognition'이라 하는 논문에서 발표된 모델이다.

이 모델의 가장 핵심적인 부분은 skip-connection이 적용된 residual block인데 skip connection은 다음을 의미한다.

 

 

Skip connection에 대한 자세한 설명은 다음 글을 확인하자.

https://kyteris0624.tistory.com/40

Deep dive into Deep learning part 15. CNN(3)

 

 

ResNet을 구현한 코드를 살펴보기 전에 먼저 'block'이란 용어부터 정의하고 넘어가자.

block이란 결국 layer들의 묶음이다. CNN에서는 여러 개의 convolution layer, pooling layer, batch-normalization layer 등을 묶어서 하나의 block으로 정의한 후, 이들을 여러 개 쌓은 구조로 Network를 설계한다.

ResNet은 여러 개의 'residual block'으로 이뤄져 있다.

 

하지만 이렇게 계속해서 block을 쌓기만 한다면 Network의 깊이가 깊어질수록 파라미터의 개수는 기하급수적으로 커질 것이다. 그렇다면 파라미터가 기하급수적으로 커지는 것을 방지하면서 Network의 사이즈를 키워나가기 위해선 어떤 방법이 있을까?

이를 위해 도입한 것이 bottleneck block (병목 블록)이다.

CNN에서 bottleneck block의 핵심은 $1 \times 1$ convolution layer이다. 

 

위 그림을 보면 우측은 'bottleneck block', 좌측은 일반적인 block이다. (데이터가 위에서 아래로 흘러가고 있는 방향이다)

 

bottleneck block은 $3 \times 3$ convolution layer에 $1 \times 1$ convolution layer를 앞뒤로 이어붙였다.

처음에 $1 \times 1$ convolution을 통해서 채널을 압축 ($256 \rightarrow 64$) 하고, $3 \times 3$을 통해서 Feature를 뽑아낸 이후, 다시 채널의 수 ($64 \rightarrow 256$)를 늘려주는 것이다.

$1 \times 1$ convolution layer는 이미지의 크기는 유지하면서 채널의 수 (깊이)를 조정하는 것이 가능하기 때문에 이를 활용해 우리는 파라미터의 수를 줄일 수 있다.

bottleneck block은 basic보다 layer를 한 개 더 쌓으면서 기존보다 파라미터의 수 (계산량)을 줄일 수 있기 때문에 이 당시에 상당히 혁신적인 아이디어였다. 

 

skip connection과 bottleneck block등을 활용해서 ResNet은 다음과 같은 'Residual block'으로 이뤄져 있다.

 

 

이 중 (a)가 ResNet의 residual block이고 1년 뒤 동일한 저자들이 실험을 통해 더 나은 residual block을 제안하였고 이것이 (e)의 구조이다. 현재 쓰이는 ResNet의 residual block은 대다수가 (e)로 구현되어져 있다.

여기서 'weight'라고 되어져 있는 부분이 convolution layer라고 보면 된다.

 

(a)와 (e)의 차이점은 우선 CONV -> BN -> RELU를 BN -> RELU -> CONV로 순서를 바꾸어줌과, skip connection에 ReLU를 없애주었다.

실제로 (a)와 (e) 사이에는 무시할 수 없는 성능 차이가 존재한다.

 

 

이제 본격적으로 ResNet을 구현해보자. 이는 ResNet 18을 기준으로 한다.

 

import torch
import torch.nn as nn
import torch.nn.functional as F

class BasicBlock(nn.Module):
	expansion = 1
    
    def __init__(self, in_channels, out_channels, stride = 1, downsample = False):
    	super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size = 3, stride = stride,
        padding = 1, bias = False) # 3 * 3 convolution layer
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, stride = 1, 
        padding = 1, bias = False) # 3 * 3 convolution layer
        self.bn2 = nn.BatchNorm2d(out_channels)
        self.relu = nn.ReLU(inplace = True)
        
       if downsample:
           conv = nn.Conv2d(in_channels, out_channels, kernel_size = 1,
           stride = stride, bias = False)
           bn = nn.BatchNorm2d(out_channels)
           downsample = nn.Sequential(conv, bn)
       else:
           downsample = None
       self.downsample = downsample
       
   def forward(self, x):
       i = x
       x = self.conv1(x)
       x = self.bn1(x)
       x = self.relu(x)
       x = self.conv2(x)
       x = self.bn2(x)
       
       if self.downsample is not None:
           i = self.downsample(i)
       
       x += I
       x = self.relu(x)
       
       return x

 

 

여기서 아래 부분의 x += I가 skip connection이 적용되는 부분이다. 

downsample이란 

다음은 bottleneck block이다.

이 block은 사실 ResNet18, 20, 32 등에서는 잘 쓰이지 않으며 주로 50, 101, 152 등에서 쓰인다는 점 참고 바란다.

 

class Bottleneck(nn.Module):
	expansion =4
    
    def __init___(self, in_channels, out_channels, stride = 1, downsample= False):
    	super().__init__()
        self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size = 1, stride = 1,
        bias = False) # 1 * 1 convolution layer
        self.bn1 = nn.BatchNorm2d(out_channels)
        self.conv2 = nn.Conv2d(out_channels, kernel_size = 3, stride = stride, padding = 1,
        bias = False) # 3 * 3 convolution layer
        self.bn2 = nn.BatchNorm2d(out_channels)
        self.conv3 = nn.Conv2d(out_channels, self.expansion * out_channels, kernel_size = 1,
        stride = 1, bias = False)
        self.bn3 = nn.BatchNorm2d(self.expansion * out_channels)
        self.relu = nn.ReLU(inplace = True)
        
        if downsample:
        	conv = nn.Conv2d(in_channels, self.expansion * out_channels, kernel_size =1,
            stride = strirde, bias = False)
            bn = nn.BatchNorm2d(self.expansion * out_channels)
            downsample = nn.Sequential(conv, bn)
        else:
        	downsample = None
        self.downsample = downsample

 

forward 연산을 위한 메소드는 첫 번째 코드와 동일하다.

bottleneck block은 위 그림에서 설명하였듯이, $1 \times 1$ conv 2개와 $3 \times 3$ conv를 가진다.

 

이제 이를 활용해서 모델을 구현하자.

 

class ResNet(nn.Module):
	def __init__(self, config, output_dim, zero_init_residual =False):
    	super().__init__()
        
        block, n_blocks, channels = config 
        self.in_channels = channels[0]
        assert len(n_blocks) == len(channels) == 4
        
        self.conv1 = nn.Conv2d(3, self.in_channels, kernel_size = 7, stride = 2, 
        padding = 3, bias = False)
        self.bn1 = nn.BatchNorm2d(self.in_channels)
        self.relu = nn.ReLU(inplace = True)
        self.maxpool = nn.MaxPool2d(kernel_size = 3, stride =2, padding = 1)
        
        self.layer1 =self.get_resnet_layer(block, n_blocks[0], channels[0])
        self.layer2 = self.get_resnet_layer(block, n_blocks[1], channels[1], stride = 2)
        self.layer3 = self.get_resnet_layer(block, n_blocks[2], channels[2], stride = 2)
        self.layer4 = self.get_resnet_layer(block, n_blocks[3], channels[3], stride = 3)
        
        self.avgpool = nn.AdaptiveAvgPool2d((1, 1))
        self.fc = nn.Linear(self.in_channels, output_dim)
        
    def get_resnet_layer(self, block, n_blocks, channels, stride =1):
    	layers = []
        if self.in_channels != block.expansion * channels:
        	downsample = True
        else:
        	downsample = False
            
        layers.append(block(self.in_channels, channels, stride, downsample))
        for i in range(1, n_blocks):
        	layers.append(block(block.expansion * channels, channels))
            
        self.in_channels = block.expansion * channels
        return nn.Sequential(*layers)

 

 

forward 메소드는 생략하였다.

여기서 get_resnet_layer 메소드는 블록을 추가하기 위한 함수이다.

n_blocks만큼 추가해준다. 당연히 이때 블록은 basic block이거나 bottleneck block일 것이다.

 

from collections import namedtuple

ResNetConfig = namedtuple('ResNetConfig', ['block', 'n_blocks', 'channels'])

resnet18_config = ResNetConfig(block = BasicBlock, n_blocks = [2, 2, 2, 2], channels = [64, 128, 256, 512])
resnet50_config = ResNetConfig(block = Bottleneck, n_blocks = [3, 4, 6, 3], channels = [64, 128, 256, 512])

 

여기서 'namedtuple'이란 파이썬의 자료형 중 하나이며 튜플의 성질을 갖고 있으면서 'key, value'로 데이터에 접근할 수 있는 특수한 자료형이다.

다음과 같이 사용할 수 있다.

 

Student = namedtuple('Student', ['name', 'age', 'DOB'])
S = Student('홍길동', '19', '187')
print(S[1]) # 19
print(S.name) # 홍길동

 

우리가 구현한 ResNet 클래스에서 layer1, layer2, layer3, layer4에 namedtuple에서의  n_blocks와 channels가 들어가게 된다.

 

 

하지만 사실 이렇게 우리가 직접 구현하는 것은 공부에는 의미가 있겠지만 실제 사용할 때는 이렇게 할 필요가 없다.

다음과 같이 이미 파이토치에 구현된 모델을 가져와 사용하면 된다.

 

import torchvision.models as models

model = models.ResNet50(pretrained = True)