[DL][Implementation] Numpy을 사용하여 CNN 구현하기

이번 포스팅에서는 numpy을 사용하여 간단한 cnn을 구현해본다. 모든 코드는 아래 repository에서 가져왔다.

https://github.com/bohyunshin/deep-learning-with-pure-numpy

GitHub - bohyunshin/deep-learning-with-pure-numpy

channel이 없는 2차원 데이터 (이미지)를 받아서 convolution 연산을 하고 max pooling, relu을 거쳐서 최종 클래스의 개수만큼의 logits을 반환하는 neural network을 구성해보자.

Architecture

 

전체적인 아키텍쳐는 위와 같다.

1. (1000, 1, 15, 15) 차원의 데이터를 받아서 convolution 연산을 수행한다.
2. max pooling 연산을 통해 (1000, 1, 5, 5) 차원의 데이터로 변환한다.
3. relu 연산을 한다.
4. flatten하여 (1000, 25) 차원의 데이터로 변환한다.
5. FC을 통해 (1000, 9) 차원의 데이터로 변환한다.
6. softmax 연산을 통해 row sum이 1이 되는 행렬을 도출한다.

Numpy code

코드는 여기서 확인할 수 있다.

 

from nn.base import BaseNeuralNet
from modules.linear import Linear
from modules.convolution import Convolution
from tools.activations import MaxPooling, Softmax, Relu


class SingleCNN(BaseNeuralNet):
    def __init__(self, input_dim, output_dim: int, kernel_dim: tuple , padding: str, pooling_size: int):
        super().__init__()
        self.cnn = Convolution(
            input_dim=input_dim,
            kernel_dim=kernel_dim,
            padding = padding
        )
        n, _, _ = input_dim
        self.n = n
        self.max_pooling = MaxPooling(input_dim=(n, self.cnn.h_out, self.cnn.w_out),
                                      k=pooling_size)
        self.softmax = Softmax()
        self.relu = Relu()

        h_out = self.max_pooling.h_out
        w_out = self.max_pooling.w_out
        self.fc = Linear(input_dim=h_out*w_out, output_dim=output_dim)

        self.gradient_step_layers += [self.cnn, self.fc]

    def forward(self, x):
        n, h_in, w_in = x.shape
        x = self.cnn.forward(x) # (n, h_out, w_out)
        x = self.max_pooling.forward(x) # (n, h_out // k, w_out // k)
        x = self.relu.forward(x)
        x = self.fc.forward(x.reshape(n,-1)) # (n, h_out*w_out) -> (n, out_dim)
        x = self.softmax.forward(x)
        return x

    def backward(self, dx_out):
        """
        params
        ------
        dx_out: np.ndarray (n, n_label)
            Upstream gradient from loss function
        """
        n, _ = dx_out.shape
        dx_out = self.softmax.backward(dx_out) # (n, n_label)
        dx_out = self.fc.backward(dx_out) # (n, h_in)
        dx_out = self.relu.backward(dx_out)
        dx_out = self.max_pooling.backward(dx_out.reshape(n, self.max_pooling.h_out, -1)) # (n, h_in, w_in)
        dx_out = self.cnn.backward(dx_out) # (n, h_in, w_in)
        return dx_out

 

single cnn을 구성하는 각 component (convolution, max pooling, relu, fc, softmax)는 모두 `forward` 메써드와 `backward` 메써드를 가진다. 여기서 짚고 넘어가야하는 점은 `backward` 메써드의 최초 input인 `dx_out` 이다. 이 `dx_out`은 loss에 대한 예측값의 gradient이다. regression 이든, classification 이든, 모델의 최종 출력값으로 $\hat{y}$가 나오고 이 값이 loss의 input으로 들어간다. 바로 이 loss 값에 대한 $\hat{y}$의 gradient인 $\dfrac{\partial L}{\partial \hat{y}}$가 `backward` 메써드의 최초 input 값인 `dx_out`으로 들어가는 것이다.

Comparison with pytorch

numpy 코드를 올바르게 짰는지 확인하고 싶어서 pytorch 결과와 비교해보고자 한다. 코드는 여기서 확인할 수 있다. 확인 과정은 아래와 같다.

1. torch cnn 모델을 초기화한다.
2. torch cnn 모델의 가중치를 numpy cnn 모델의 가중치로 강제 정의한다. (코드)
3. 전체 데이터 개수는 1000개, batch_size는 32개로 하여 전체 큰 epoch을 돌고 각 epoch 별로는 mini-batch gradient descent을 진행한다.
4. 각 epoch을 돌 때마다 numpy 모델의 loss와 pytorch 모델의 loss을 확인한다. (코드)
5. 전체 epoch가 모두 종료되고 torch, numpy 모델의 가중치를 확인한다. (코드)
   • convolution의 kernel, bias 값
   • FC의 weight, bias 값

코드로 살펴보자.

 

mport numpy as np
import torch
from torch import nn, optim
from torch.utils.data import DataLoader, Dataset
import sys
import os
sys.path.append(os.path.join(os.getcwd(), "src"))
sys.path.append(os.path.join(os.getcwd(), "test"))

from nn.single_cnn import SingleCNN
from nn.mlp import MultipleLayerPerceptron
from loss.classification import CrossEntropyLoss
from loss.regression import MeanSquaredError
from torch_model import TorchMLP, TorchCNN
from tools.utils import one_hot_vector
from data.data import NumpyDataset, NumpyDataLoader


def test_single_cnn_dummy_data_same_as_torch():
    n = 1000
    h_in, w_in = 15, 15
    n_channel = 1
    imgs = torch.randn((n, n_channel, h_in, w_in))
    output_dim = 9
    y = np.eye(output_dim)[np.random.choice(output_dim, n)]
    kernel_dim = (3,3)
    padding = "same"
    pooling_size = 3
    epoch = 1
    lr = 0.1
    batch_size = 32
    n_batch = n // batch_size + (n % batch_size >= 1)

    # use numpy dataloader
    dataset = NumpyDataset(imgs, y)
    dataloader = NumpyDataLoader(dataset, batch_size=batch_size, shuffle=False)

    # set numpy model
    cnn = SingleCNN(input_dim=(n, h_in, w_in),
                    output_dim=output_dim,
                    kernel_dim=kernel_dim,
                    padding=padding,
                    pooling_size=pooling_size)
    ce_loss = CrossEntropyLoss()

    # set torch model
    model = TorchCNN(h_in, w_in, output_dim, kernel_dim, pooling_size)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(model.parameters(), lr=lr)

    # set numpy model weight as torch model weight in advance
    cnn.cnn.kernel = model.conv.weight.data.squeeze(0,1).detach().numpy().copy()
    cnn.cnn.b = model.conv.bias.data.detach().numpy().copy()
    cnn.fc.w = model.fc.weight.data.detach().numpy().copy().T
    cnn.fc.b = model.fc.bias.data.detach().numpy().copy()

    for _ in range(epoch):

        running_loss_np = 0.0
        running_loss_pt = 0.0

        for data in dataloader:
            X_train, y_train = data
            y_train = torch.tensor(torch.from_numpy(y_train.copy()))

            # torch implementation
            optimizer.zero_grad()
            y_pred = model(X_train)
            y_pred.retain_grad()
            loss = criterion(y_pred, y_train)
            loss.backward()
            optimizer.step()

            running_loss_pt += loss.item()

            # numpy implementation
            X_train = X_train.squeeze(1).detach().numpy() # no channel
            y_train = y_train.detach().numpy()
            y_pred_prob = cnn.forward(X_train)  # not logits, probability prediction
            loss = ce_loss.forward(y_train, y_pred_prob)
            running_loss_np += loss.item()

            dx_out = ce_loss.backward(y_train, y_pred_prob)
            cnn.backward(dx_out)
            cnn.step(lr)

        running_loss_pt /= n_batch
        running_loss_np /= n_batch

        # check loss at every epoch
        np.testing.assert_almost_equal(running_loss_pt, running_loss_np)


    # fc layer check
    np.testing.assert_array_almost_equal(model.fc.bias.detach().numpy(), cnn.fc.b)
    np.testing.assert_array_almost_equal(model.fc.weight.detach().numpy(), cnn.fc.w.T)
    # conv layer check
    np.testing.assert_array_almost_equal(model.conv.weight.squeeze((0, 1)).detach().numpy(), cnn.cnn.kernel)
    np.testing.assert_array_almost_equal(model.conv.bias.detach().numpy(), cnn.cnn.b)

 

Issue history

batch gradient descent 기준으로 코드를 짜고 backpropagation까지 했을 때, numpy으로 구현된 모델의 가중치와 torch 모델의 가중치가 동일해서 mini-batch gradient descent 기준으로 코드를 짜도 결과가 크게 달라지지 않을 것이라 생각했다. 근데, mini-batch gradient descent로 코드를 짜고 돌려보니 첫번째 mini-batch 에서만 가중치 및 gradient가 동일하고 그 이후 mini-batch에서부터는 convolution의 kernel 가중치 및 gradient가 달라지는게 아닌가! 이상한 점은 fc의 가중치, bias, convolution의 bias는 mini-batch가 진행됨에 따라서 값이 달라지지 않았는데 convolution의 kernel만 달라졌다. 그냥 포기하고 known issue로 가져갈까도 생각했는데..

 

일주일 넘게 탐구한 끝에 max pooling의 backpropagation을 하는 과정에서 max value index 캐싱이 잘못 됐음을 발견했다. 현재 코드 상으로 캐싱 리스트가 `forward` 메써드가 호출될 때 생성이 되는데 그 이전에는 `__init__` 메써드에 있어서 여기서 쌓인 max value index가 계속 쓰이는 문제가 있었던 것이다. 이러한 이유 때문에 첫번째 mini-batch에서만 결과가 올바르게 나왔던 것이다.

Conclusion

pytorch의 결과와 비교하기 위해서 torch의 코드 방식도 익힐 수 있어서 일석이조인 공부였다. 특히, torch의 backpropagation 결과와 내가 구현한 결과가 동일함을 확인하니 수학적으로, 프로그램적으로 틀린 것이 하나도 없었음을 확인받은 것 같아서 안심이 됐다. 앞으로 이렇게 test 코드를 통해 확인하는 습관을 기르면 좋을 것 같다는 생각이 들었다.