11-4. Tensorflow를 이용한 CNN의 구현

김성훈 교수님의 강의내용을 정리한 내용입니다.

출처 : http://hunkim.github.io/ml/

모두를 위한 머신러닝/딥러닝 강의

---

CNN 구조 예시

컨볼루션 연산을 테스트 하기 위한 입력 이미지를 만들었다.

 

모든 요소가 1이고 2x2크기를 갖는 필터를 컨볼루션 연산을 수행하면

2x2크기의 출력값이 나오게 된다.

 

Padding을 적용하게 되면 입력 크기와 출력 크기가 동일하게 된다.

 

여러개의 필터를 사용하여 출력 결과의 갯수를 늘릴 수 있다.

 

최대 풀링을 적용하여 값을 출력한다.

 

 

MNIST 이미지에서 컨볼루션 연산을 수행해보자.

5개의 필터를 적용하여 나온 출력 결과들이다.

 

최대 풀링을 적용한다면 위와 같은 결과가 나오게 된다.

 

---

간단한 CNN 네트워크를 구성해보자.

첫번째 레이어의 구성은 위의 그림과 같다.

3x3 크기의 필터를 이용해 32개의 출력값을 얻기위해 W1의 값을 [3, 3, 1, 32]로 설정하였다.

 

첫번째 레이어와 동일한 구조이다.

입력값과 출력값의 크기에 유의하여 설정하도록 하자.

 

CNN 네트워크의 말단 부분인 FC 레이어를 구성하는 코드이다.

입체적인 출력값을 FC 레이어는 받지 못하기 때문에 reshape를 통해 한줄로 재구성하는 과정을 거친다.

 

학습 및 테스트 결과 98%라는 높은 결과가 나왔다.

 

CNN 구조를 더욱 깊게 하면 어떤 결과가 나오게 될까?

 

단순히 똑같은 구조의 코드를 더 추가하면 간단하게 완성할 수 있다.

 

최종 결과는 99%로 놀라운 결과가 나오게 되었다.

 

---

파이썬에서 제공하는 Class 구조를 사용한다면 복잡한 코드를 단순하게 만들 수 있다.

네트워크의 구성, 예측, 정확도 측정, 학습 등 여러 기능들을 한 Class로 묶어 만들었다.

 

 

Tensorflow에서는 high level의 api를 제공하는데,

이를 이용하면 아래 그림과 같이 번거로운 변수 설정을 한번에 수행할 수 있다.

레이어 구축을 좀 더 쉽고 간결하게 할 수 있다.

앙상블 구조

앙상블 구조를 이용하여 분류 작업을 수행해보자.

models라는 list에 모델들을 추가한 후 각각의 모델들을 학습시킨다.

각 모델들의 결과값을 합하여 최종 결과값을 얻어낸다.

7개의 다른 모델들의 정확도 보다 더 높은 정확도를 얻을 수 있었다.

---

위의 코드에서 높은 버전의 텐서플로에서 동작하지 않는 코드들을 수정하였습니다.

 

코드

# Lab 11 MNIST and Deep learning CNN
# https://www.tensorflow.org/tutorials/layers
import tensorflow as tf
import numpy as np
 
tf.set_random_seed(777)  # reproducibility
 
data_train, data_test = tf.keras.datasets.mnist.load_data()
 
# Parse images and labels
(images_train, labels_train) = data_train
labels_train = tf.keras.utils.to_categorical(labels_train,10)
train_dataset = tf.data.Dataset.from_tensor_slices((images_train, labels_train))
 
(images_test, labels_test) = data_test
labels_test = tf.keras.utils.to_categorical(labels_test,10)
 
# hyper parameters
learning_rate = 0.001
training_epochs = 20
batch_size = 100
 
train_dataset = train_dataset.repeat().batch(batch_size).prefetch(1)
train_iterator = train_dataset.make_one_shot_iterator()
next_batch_train = train_iterator.get_next()
 
class Model:
 
    def __init__(self, sess, name):
        self.sess = sess
        self.name = name
        self._build_net()
 
    def _build_net(self):
        with tf.variable_scope(self.name):
            # dropout (keep_prob) rate  0.7~0.5 on training, but should be 1
            # for testing
            self.training = tf.placeholder(tf.bool)
 
            # input place holders
            self.X = tf.placeholder(tf.float32, [None, 28, 28])
            X_img = tf.reshape(self.X, [-1, 28, 28, 1])
            
            # img 28x28x1 (black/white), Input Layer
            self.Y = tf.placeholder(tf.float32, [None, 10])
 
            # Convolutional Layer #1
            conv1 = tf.layers.conv2d(inputs=X_img, filters=32, kernel_size=[3, 3],
                                     padding="SAME", activation=tf.nn.relu)
            # Pooling Layer #1
            pool1 = tf.layers.max_pooling2d(inputs=conv1, pool_size=[2, 2],
                                            padding="SAME", strides=2)
            dropout1 = tf.layers.dropout(inputs=pool1,
                                         rate=0.3, training=self.training)
 
            # Convolutional Layer #2 and Pooling Layer #2
            conv2 = tf.layers.conv2d(inputs=dropout1, filters=64, kernel_size=[3, 3],
                                     padding="SAME", activation=tf.nn.relu)
            pool2 = tf.layers.max_pooling2d(inputs=conv2, pool_size=[2, 2],
                                            padding="SAME", strides=2)
            dropout2 = tf.layers.dropout(inputs=pool2,
                                         rate=0.3, training=self.training)
 
            # Convolutional Layer #3 and Pooling Layer #3
            conv3 = tf.layers.conv2d(inputs=dropout2, filters=128, kernel_size=[3, 3],
                                     padding="SAME", activation=tf.nn.relu)
            pool3 = tf.layers.max_pooling2d(inputs=conv3, pool_size=[2, 2],
                                            padding="SAME", strides=2)
            dropout3 = tf.layers.dropout(inputs=pool3,
                                         rate=0.3, training=self.training)
 
            # Dense Layer with Relu
            flat = tf.reshape(dropout3, [-1, 128 * 4 * 4])
            dense4 = tf.layers.dense(inputs=flat,
                                     units=625, activation=tf.nn.relu)
            dropout4 = tf.layers.dropout(inputs=dense4,
                                         rate=0.5, training=self.training)
 
            # Logits (no activation) Layer: L5 Final FC 625 inputs -> 10 outputs
            self.logits = tf.layers.dense(inputs=dropout4, units=10)
 
        # define cost/loss & optimizer
        self.cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits_v2(
            logits=self.logits, labels=self.Y))
        self.optimizer = tf.train.AdamOptimizer(
            learning_rate=learning_rate).minimize(self.cost)
 
        correct_prediction = tf.equal(
            tf.argmax(self.logits, 1), tf.argmax(self.Y, 1))
        self.accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
 
    def predict(self, x_test, training=False):
        return self.sess.run(self.logits,
                             feed_dict={self.X: x_test, self.training: training})
 
    def get_accuracy(self, x_test, y_test, training=False):
        return self.sess.run(self.accuracy,
                             feed_dict={self.X: x_test,
                                        self.Y: y_test, self.training: training})
 
    def train(self, x_data, y_data, training=True):
        return self.sess.run([self.cost, self.optimizer], feed_dict={
            self.X: x_data, self.Y: y_data, self.training: training})
 
# initialize
sess = tf.Session()
 
models = []
num_models = 2
for m in range(num_models):
    models.append(Model(sess, "model" + str(m)))
 
sess.run(tf.global_variables_initializer())
 
print('Learning Started!')
 
# train my model
for epoch in range(training_epochs):
    avg_cost_list = np.zeros(len(models))
    total_batch = int(len(labels_train) / batch_size)
    for i in range(total_batch):
        batch_xs, batch_ys = sess.run(next_batch_train)
 
        # train each model
        for m_idx, m in enumerate(models):
            c, _ = m.train(batch_xs, batch_ys)
            avg_cost_list[m_idx] += c / total_batch
 
    print('Epoch:', '%04d' % (epoch + 1), 'cost =', avg_cost_list)
 
print('Learning Finished!')
 
# Test model and check accuracy
test_size = len(labels_test)
predictions = np.zeros([test_size, 10])
for m_idx, m in enumerate(models):
    print(m_idx, 'Accuracy:', m.get_accuracy(
        images_test, labels_test))
    p = m.predict(images_test)
    predictions += p
 
ensemble_correct_prediction = tf.equal(
    tf.argmax(predictions, 1), tf.argmax(labels_test, 1))
ensemble_accuracy = tf.reduce_mean(
    tf.cast(ensemble_correct_prediction, tf.float32))
print('Ensemble accuracy:', sess.run(ensemble_accuracy))