[深さ学習論文レビュー]1AlexNet

AlexNet

Architecture

ReLU Non-Linearity

のアクティブ化機能として使用していた従来のモデルとは異なり、ReLUを使用している.

Training on Multiple GPUs

Local Response Normalization

ReLUはSigmoid型関数とは異なり,入力正規化は必須ではない.しかし,AlexNetはローカル応答正規化を用い,誤差を約1−2%低減した.

Overlapping Pooling

Reducing Overfitting

Data Augmentation

Dropout

Details of Learning

Results

コード実装

import numpy as np
import pandas as pd
import tensorflow as tf

# define model parameters
NUM_EPOCHS = 1000
NUM_CLASSES = 10
IMAGE_SIZE = 227
BATCH_SIZE = 16

from tensorflow.keras.utils import to_categorical
from sklearn.model_selection import train_test_split
from tensorflow.keras.datasets import cifar10

def zero_one_scaler(image):
    return image/255.0

def get_preprocessed_ohe(images, labels, pre_func=None):
    # preprocessing 함수가 입력되면 이를 이용하여 image array를 scaling 적용.
    if pre_func is not None:
        images = pre_func(images)
    # OHE 적용    
    oh_labels = to_categorical(labels)
    return images, oh_labels

# 학습/검증/테스트 데이터 세트에 전처리 및 OHE 적용한 뒤 반환 
def get_train_valid_test_set(train_images, train_labels, test_images, test_labels, valid_size=0.15, random_state=2021):
    # 학습 및 테스트 데이터 세트를  0 ~ 1사이값 float32로 변경 및 OHE 적용. 
    train_images, train_oh_labels = get_preprocessed_ohe(train_images, train_labels)
    test_images, test_oh_labels = get_preprocessed_ohe(test_images, test_labels)
    
    # 학습 데이터를 검증 데이터 세트로 다시 분리
    tr_images, val_images, tr_oh_labels, val_oh_labels = train_test_split(train_images, train_oh_labels, test_size=valid_size, random_state=random_state)
    
    return (tr_images, tr_oh_labels), (val_images, val_oh_labels), (test_images, test_oh_labels )

# CIFAR10 데이터 재 로딩 및 Scaling/OHE 전처리 적용하여 학습/검증/데이터 세트 생성. 
(train_images, train_labels), (test_images, test_labels) = cifar10.load_data()
print(train_images.shape, train_labels.shape, test_images.shape, test_labels.shape)
(train_images, train_labels), (val_images, val_labels), (test_images, test_labels) = \
    get_train_valid_test_set(train_images, train_labels, test_images, test_labels, valid_size=0.2, random_state=2021)

print(train_images.shape, train_labels.shape, val_images.shape, val_labels.shape, test_images.shape, test_labels.shape)

from tensorflow.keras.utils import Sequence
import cv2
import sklearn

# 입력 인자 images_array labels는 모두 numpy array로 들어옴. 
# 인자로 입력되는 images_array는 전체 32x32 image array임. 
class CIFAR_Dataset(Sequence):
    def __init__(self, images_array, labels, batch_size=BATCH_SIZE, augmentor=None, shuffle=False, pre_func=None):
        '''
        파라미터 설명
        images_array: 원본 32x32 만큼의 image 배열값. 
        labels: 해당 image의 label들
        batch_size: __getitem__(self, index) 호출 시 마다 가져올 데이터 batch 건수
        augmentor: albumentations 객체
        shuffle: 학습 데이터의 경우 epoch 종료시마다 데이터를 섞을지 여부
        '''
        # 객체 생성 인자로 들어온 값을 객체 내부 변수로 할당. 
        # 인자로 입력되는 images_array는 전체 32x32 image array임.
        self.images_array = images_array
        self.labels = labels
        self.batch_size = batch_size
        self.augmentor = augmentor
        self.pre_func = pre_func
        # train data의 경우 
        self.shuffle = shuffle
        if self.shuffle:
            # 객체 생성시에 한번 데이터를 섞음. 
            #self.on_epoch_end()
            pass
    
    # Sequence를 상속받은 Dataset은 batch_size 단위로 입력된 데이터를 처리함. 
    # __len__()은 전체 데이터 건수가 주어졌을 때 batch_size단위로 몇번 데이터를 반환하는지 나타남
    def __len__(self):
        # batch_size단위로 데이터를 몇번 가져와야하는지 계산하기 위해 전체 데이터 건수를 batch_size로 나누되, 정수로 정확히 나눠지지 않을 경우 1회를 더한다. 
        return int(np.ceil(len(self.labels) / self.batch_size))
    
    # batch_size 단위로 image_array, label_array 데이터를 가져와서 변환한 뒤 다시 반환함
    # 인자로 몇번째 batch 인지를 나타내는 index를 입력하면 해당 순서에 해당하는 batch_size 만큼의 데이타를 가공하여 반환
    # batch_size 갯수만큼 변환된 image_array와 label_array 반환. 
    def __getitem__(self, index):
        # index는 몇번째 batch인지를 나타냄. 
        # batch_size만큼 순차적으로 데이터를 가져오려면 array에서 index*self.batch_size:(index+1)*self.batch_size 만큼의 연속 데이터를 가져오면 됨
        # 32x32 image array를 self.batch_size만큼 가져옴. 
        images_fetch = self.images_array[index*self.batch_size:(index+1)*self.batch_size]
        if self.labels is not None:
            label_batch = self.labels[index*self.batch_size:(index+1)*self.batch_size]
        
        # 만일 객체 생성 인자로 albumentation으로 만든 augmentor가 주어진다면 아래와 같이 augmentor를 이용하여 image 변환
        # albumentations은 개별 image만 변환할 수 있으므로 batch_size만큼 할당된 image_name_batch를 한 건씩 iteration하면서 변환 수행. 
        # 변환된 image 배열값을 담을 image_batch 선언. image_batch 배열은 float32 로 설정. 
        image_batch = np.zeros((images_fetch.shape[0], IMAGE_SIZE, IMAGE_SIZE, 3), dtype='float32')
        
        # batch_size에 담긴 건수만큼 iteration 하면서 opencv image load -> image augmentation 변환(augmentor가 not None일 경우)-> image_batch에 담음. 
        for image_index in range(images_fetch.shape[0]):
            #image = cv2.cvtColor(cv2.imread(image_name_batch[image_index]), cv2.COLOR_BGR2RGB)
            # 원본 image를 IMAGE_SIZE x IMAGE_SIZE 크기로 변환
            image = cv2.resize(images_fetch[image_index], (IMAGE_SIZE, IMAGE_SIZE))
            # 만약 augmentor가 주어졌다면 이를 적용. 
            if self.augmentor is not None:
                image = self.augmentor(image=image)['image']
                
            # 만약 scaling 함수가 입력되었다면 이를 적용하여 scaling 수행. 
            if self.pre_func is not None:
                image = self.pre_func(image)
            
            # image_batch에 순차적으로 변환된 image를 담음.               
            image_batch[image_index] = image
        
        return image_batch, label_batch
    
    # epoch가 한번 수행이 완료 될 때마다 모델의 fit()에서 호출됨. 
    def on_epoch_end(self):
        if(self.shuffle):
            #print('epoch end')
            # 원본 image배열과 label를 쌍을 맞춰서 섞어준다. scikt learn의 utils.shuffle에서 해당 기능 제공
            self.images_array, self.labels = sklearn.utils.shuffle(self.images_array, self.labels)
        else:
            pass
            
def zero_one_scaler(image):
    return image/255.0

train_ds = CIFAR_Dataset(train_images, train_labels, batch_size=BATCH_SIZE, augmentor=None, shuffle=True, pre_func=zero_one_scaler)
val_ds = CIFAR_Dataset(val_images, val_labels, batch_size=BATCH_SIZE, augmentor=None, shuffle=False, pre_func=zero_one_scaler)

from matplotlib import pyplot
from keras.datasets import cifar10
# summarize loaded dataset
print('Train: X=%s, y=%s' % (train_ds[0][0][0].shape, train_ds[0][1][0].shape))
print('Test: X=%s, y=%s' % (val_ds[0][0][0].shape, val_ds[0][1][0].shape))
# plot first few images
for i in range(9):
    # define subplot
    pyplot.subplot(330+1+i)
    # plot raw pixel data
    pyplot.imshow(train_ds[0][0][i])
# show the figure
pyplot.show()

from tensorflow.keras.layers import *

class AlexNet(tf.keras.Model):
    def __init__(self, num_classes=NUM_CLASSES):
        super().__init__()
#         self.weight = tf.keras.initializers.RandomNormal(mean=0., stddev=1.)
#         self.bias = tf.keras.initializers.Constant(1.)
        self.lrn = lambda x: tf.nn.local_response_normalization(x, depth_radius=5, alpha=0.0001, beta=0.75, bias=2)
        self.regularizer = tf.keras.regularizers.l2(l2=0.0005)
        self.net = tf.keras.Sequential([
            Conv2D(
                filters=96, 
                kernel_size=11, 
                strides=4, 
#                 kernel_initializer=self.weight, 
                kernel_regularizer=self.regularizer
            ),
            ReLU(),
            Lambda(self.lrn),
            MaxPool2D(pool_size=3, strides=2),
            Conv2D(
                filters=256, 
                kernel_size=5, 
                padding='same',
#                 kernel_initializer=self.weight, 
#                 bias_initializer=self.bias, 
                kernel_regularizer=self.regularizer
            ),
            ReLU(),
            Lambda(self.lrn),
            MaxPool2D(pool_size=3, strides=2),
            Conv2D(
                filters=384, 
                kernel_size=3,
                padding='same',
#                 kernel_initializer=self.weight,
#                 bias_initializer='self.bias',
                kernel_regularizer=self.regularizer
            ),
            ReLU(),
            Conv2D(
                filters=384, 
                kernel_size=3, 
                padding='same',
#                 kernel_initializer=self.weight, 
#                 bias_initializer=self.bias, 
                kernel_regularizer=self.regularizer
            ),
            ReLU(),
            Conv2D(
                filters=256, 
                kernel_size=3, 
                padding='same',
#                 kernel_initializer=self.weight, 
#                 bias_initializer=self.bias,
                kernel_regularizer=self.regularizer
            ),
            ReLU(),
            MaxPool2D(pool_size=3, strides=2),
            
            Reshape((-1, 256 * 6 * 6)),
            
            Dropout(0.5),
            Dense(4096),
            ReLU(),

            Dropout(0.5),
            Dense(4096),
            ReLU(),

            Dense(num_classes, activation='softmax'),
        ])
        
    def call(self, inputs):
        return tf.reshape(self.net(inputs), (-1, 10))

from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping

rlr_cb = ReduceLROnPlateau(monitor='val_loss', factor=0.001, patience=5, mode='min', verbose=1)
ely_cb = EarlyStopping(monitor='val_loss', patience=15, mode='min', verbose=1)

tb_path = './tensorboard'
tensorboard_cb = tf.keras.callbacks.TensorBoard(log_dir=tb_path)

alexnet = AlexNet(num_classes=NUM_CLASSES)

# alexnet = tf.keras.models.load_model("alexnet")

alexnet.compile(
    optimizer=tf.keras.optimizers.SGD(0.0001, momentum=0.9),
#     optimizer=tf.keras.optimizers.Adam(0.0001),
    loss=tf.keras.losses.CategoricalCrossentropy(),
    metrics=['accuracy'],
    run_eagerly=True
)

alexnet.fit(train_ds, verbose=1, validation_data=val_ds, epochs=NUM_EPOCHS, callbacks=[tensorboard_cb, rlr_cb, ely_cb])

test_ds = CIFAR_Dataset(test_images, test_labels, batch_size=BATCH_SIZE, augmentor=None, shuffle=False, pre_func=zero_one_scaler)
alexnet.evaluate(test_ds)

alexnet.save("alexnet")

結果

その後test datasetを評価した結果,損失は1.1369,精度は0.7797であった.
オレンジ色はtrain精度、青は検証精度です.

総括と討論

参考資料