6-29 RNN과 CNN을 결합한 Hybrid 알고리듬에 의한 MNIST 머신러닝

지난해 겨울 텐서플로우 OpenCV 출간 이후 7개월 만에 2권에 해당하는 “Scikit PyTorch 머신러닝”을 출간하게 되었다. 각권이 450 페이지이므로 합 900 페이지에 달하는 내용이라 무슨 머신러닝을 공부하는데 분량이 왜 이렇게 많은가? 하고 의문을 가질 수도 있겠으나 그 내용이 튜토리얼성에 가까워 사실 그렇게 큰 부담은 없는 책이다. 물론 책 내부에 파이선 코드를 끼워 넣는 먹통 짓은 지금 세상에서는 할 필요가 없을 것이다. 해당 책의 머리말에 써둔 하이퍼링크 목차를 다운받으면 블로그를 직접 열어 볼 수 있으며 거기서 예제 코드를 다운 받을 수 있다.

 

1권에 해당하는 텐서플로우 OpenCV 머신러닝에서는 2017년 9월경부터 2018년 12월 사이에 머신러닝을 이해하고자 하는 필자의 열공(?) 내용을 담아 보았다면 2권에서는 1권에서 제기되었던 여러 내용들에 대해서 해답을 찾아가는 내용들을 꽤 많이 포함하였다. 2권이라고 해서 내용적으로 완전히 정리된 것은 아니기 때문에 다시 3권의 출발점이 될 수도 있을 것이다.

 

다소 아쉬운 점은 흥미 위주로 시작했던 1, 2권의 Softmax 관련 내용을 완전히 정리하지는 못했는데 이 그 이유는 R&D 영역으로 넘어갔기 때문이다. 2020년에는그 내용까지도 포함하여 출간할 계획이다.

 

본 서의 출간 목적은 작가들과 출판사가 염원하는 베스트셀러 화가 목표가 아니다. 사회적으로 인공지능(머신러닝)에 대한 이해 필요성이 점증하는 시기이며, 인공지능 분야의 발전 속도가 상당히 빠르기 때문에 그에 맞춰서 비전공자라 할지라도 머신러닝에 입문해 볼 수 있도록 경험과 생각을 공유해 보고자 하는 것이다.

+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

RNN 에 의한 MNIST 문자 데이터 처리를 위한 구조이다. MNIST 문자 데이터 베이스를 읽어 들이는 명령은 다음과 같다. 즉 하나의 문자를 784개의 1차원 픽셀 배열이 아니라 28X28 매트릭스 배열 형태로 읽어 들이게 된다.

 

x = tf.placeholder("float",[None, time_steps, n_input])

 

이 데이터를 RNN 처리하기 위해서는 위 그림과 같이 28개 열로 나누어 각가의 28개 픽셀 데이터로 구성된 개개의 데이터를 28개 Cell 들이 연결되어 구성된 RNN에 입력하도록 하자.

 

    28X28 매트릭스 형태로 읽어 들인 데이터 x를 아래와 같이 unstack 명령을 사용하여 28개 열(row)로 분할하도록 한다. time_steps 는 28개의 연결된 Cell 상태 수를 의미한다.

처리 결과를 확인하기 위해서 변수 input을 출력(print)처리하도록 하자. TensorFlow에서 그래프 영역의 print 문은 아래의 결과와 같이 사용 중인 변수의 텐서 규격을 보여준다.

이렇게 처리된 데이터 input은 rnn_static 루틴에서 입력 파라메터로 사용된다. 이 점에서 왜 tensorflow 의 명령인 tf.nn.dynamic_rnn을 사용치 않는가라는 의문을 제기할 수 있는데 그 이유는 dynamic_rnn 루틴에서 사용하는 입력 데이터의 요구되는 shape의 rank 값이 3 이어야 하기 때문이다.

 

데이터 x 로부터 입력 데이터의 규격을 살펴보면 (time_steps, n_INPUT) 임을 알 수 있다. 하지만 dynamic_rnn을 흔히 사용하는 문자 레벨 RNN 에서는 (batch_size, time_steps, n_iput) 형태의 shape 데이터를 사용한다는 점이며 파라메터 batch_size는 MNIST 데이터를 임의 추출 방식으로 읽어 들이는 개수와는 다른 의미로 사용될 수도 있음에 유의하자.

 

BasicLSTMCell 에서는 필수 파라메터인 hidden layer 의 출력 수 값인 num_units 값을 사용자가 임의로 설정하나 MNIST 문제를 풀기 위해서는 784로 설정하기로 한다. Unstack 된 상태의 input 데이터를 사용하여 rnn.static_rnn 루틴을 부르면 내부적으로 hidden layer 의 웨이트 매트릭스를 수반하는 RNN 레이어 준비가 완료 되며 outputs 와 _states를 출력하게 된다. outputs 는 RNN 연산에 결과 리스트 데이터 전체를 출력해 주는 반면에 _states 는 0.0 의 값을 가지는 리스트 데이터들을 배제한 간추린 상태의 출력 값을 보여준다. 784개의 데이터를 가지는 outputs를 대상으로 선형회귀 형태의 prediction(hypothesis) 함수를 설정하고 Cross Entropy 형을 사용하는 loss 함수는 prediction 과 라벨 값 y를 입력해 줘야 한다.

 

이런 형태로 RNN 코드가 구성되는 과정에 CNN 코드를 인터페이스 하는 방법을 알아보자. Unstack 된 데이터를 RNN 데이터 생성을 위해 사용한 후 다음과 같이 CNN에서 사용할 수 있도록 다시 reshape 하도록 한다.

 

한편 잘 알려진 CNN 코드 구조를 살펴보도록 하자. 이 코드는 유튜브에서 머신러닝(인공지능)을 강의하는 Sung Kim 동영상에서 공유하여 사용하는 GIthub 오픈소스 중 lab-11-1-mnist_cnn.py 에서 취하였다.

즉 RNN 과 CNN 코드를 인터페이스 하려면 X 데이터를 아예 28X28 로 읽어 unstack 하고 RNN 형태로 일차 처리 후 선형회귀 형태로 만들되 CNN 입력이 28X28 기준이므로 RNN에서 데이터 준비 시 LSTMCell 의 hidden layer 수인 num_units = 784 으로 설정하여 아래에서처럼 웨이트 매트릭스 W0 와 b0를 준비하면 prediction이 1X784 가 되어 기존의 CNN 코드와 인터페이스가 가능해진다.

1X784인 CNN 의 입력 데이터 준비를 위해 784X784 웨이트 매트릭스 W0 와 784개의 바이아스 b0 가 추가되는 셈이다.

 

다음의 계산 결과를 살펴보자. Sung Kim 이 제공하는 코드 하단부에 첨부된 계산 사례에 의하면 98.85% 의 accuracy를 보여주고 있다. 반면에 RNN 이 추가된 경우 epoch 설정에 따라 변동이 있기는 하지만 99.1% 의 정밀도를 보여 준다. 비록 0.25% accuracy 향상에 불과하나 CNN 코드 알고리듬의 완벽성에 비춰 보아 99.0% 대에서 0.1% 의 향상도 쉽지 않다는 점에 유의하자.

그밖에 CNN 코드에 accuracy를 증가시키려고 적용하는 Dropout 기법에 의해 보다 높은 99.5% 수준의 accuracy 가 얻어질 수 있으나 RNN 과의 시너지 효과가 있을지는 의문이다.

 

#mnist_rnn_cnn_01.py

import tensorflow as tf
from tensorflow.contrib import rnn
import random

from tensorflow.examples.tutorials.mnist import input_data

tf.set_random_seed(777)  # reproducibility

mnist = input_data.read_data_sets("MNIST_data/", one_hot=True)

tf.reset_default_graph()
time_steps = 28 #unrolled through 28 time steps, sequence_length
num_units = 784 #hidden LSTM units

n_input = 28 #rows of 28 pixels
learning_rate = 0.001 #learning rate for adam
n_classes = 10
batch_size = 100 #size of batch
training_epochs = 20

W0=tf.Variable(tf.random_normal([num_units,784]))
b0=tf.Variable(tf.random_normal([784]))

X = tf.placeholder("float",[None, time_steps, n_input])
Y = tf.placeholder("float",[None, n_classes])
#from [batch_size,n_steps,n_input] to "time_steps" number of [batch_size,n_input] tensors
input = tf.unstack(X, time_steps, 1)

#defining the network
lstm_layer = tf.contrib.rnn.BasicLSTMCell(num_units, forget_bias=1)
outputs,_ = rnn.static_rnn(lstm_layer, input, dtype="float32")

#converting last output of dimension [batch_size,num_units] to [batch_size,n_classes] by out_weight multiplication
prediction = (tf.matmul(outputs[-1], W0) + b0)

X_img = tf.reshape(prediction, [-1, 28, 28, 1])   # img 28x28x1 (black/white)
print(X_img)
Y = tf.placeholder(tf.float32, [None, 10])

# L1 ImgIn shape=(?, 28, 28, 1)
W1 = tf.Variable(tf.random_normal([3, 3, 1, 32], stddev=0.01))
#    Conv     -> (?, 28, 28, 32)
#    Pool     -> (?, 14, 14, 32)
L1 = tf.nn.conv2d(X_img, W1, strides=[1, 1, 1, 1], padding='SAME')
L1 = tf.nn.relu(L1)
L1 = tf.nn.max_pool(L1, ksize=[1, 2, 2, 1],strides=[1, 2, 2, 1], padding='SAME')
'''
Tensor("Conv2D:0", shape=(?, 28, 28, 32), dtype=float32)
Tensor("Relu:0", shape=(?, 28, 28, 32), dtype=float32)
Tensor("MaxPool:0", shape=(?, 14, 14, 32), dtype=float32)
'''

# L2 ImgIn shape=(?, 14, 14, 32)
W2 = tf.Variable(tf.random_normal([3, 3, 32, 64], stddev=0.01))
#    Conv      ->(?, 14, 14, 64)
#    Pool      ->(?, 7, 7, 64)
L2 = tf.nn.conv2d(L1, W2, strides=[1, 1, 1, 1], padding='SAME')
L2 = tf.nn.relu(L2)
L2 = tf.nn.max_pool(L2, ksize=[1, 2, 2, 1],
                    strides=[1, 2, 2, 1], padding='SAME')
L2_flat = tf.reshape(L2, [-1, 7 * 7 * 64])
'''
Tensor("Conv2D_1:0", shape=(?, 14, 14, 64), dtype=float32)
Tensor("Relu_1:0", shape=(?, 14, 14, 64), dtype=float32)
Tensor("MaxPool_1:0", shape=(?, 7, 7, 64), dtype=float32)
Tensor("Reshape_1:0", shape=(?, 3136), dtype=float32)
'''

# Final FC 7x7x64 inputs -> 10 outputs
W3 = tf.get_variable("W3", shape=[7 * 7 * 64, 10],
                     initializer=tf.contrib.layers.xavier_initializer())
b = tf.Variable(tf.random_normal([10]))
logits = tf.matmul(L2_flat, W3) + b

# define cost/loss & optimizer
cost = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits(
    logits=logits, labels=Y))
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(cost)

# initialize
sess = tf.Session()
sess.run(tf.global_variables_initializer())

# train my model
print('Learning started. It takes sometime.')
for epoch in range(training_epochs):
    avg_cost = 0
    total_batch = int(mnist.train.num_examples / batch_size)

    for i in range(total_batch):
        batch_xs, batch_ys = mnist.train.next_batch(batch_size=batch_size)
        batch_xs = batch_xs.reshape((batch_size, time_steps, n_input))
        feed_dict = {X: batch_xs, Y: batch_ys}
        c, _ = sess.run([cost, optimizer], feed_dict=feed_dict)
        avg_cost += c / total_batch

    print('Epoch:', '%04d' % (epoch + 1), 'cost =', '{:.9f}'.format(avg_cost))

print('Learning Finished!')

# Test model and check accuracy
correct_prediction = tf.equal(tf.argmax(logits, 1), tf.argmax(Y, 1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
test_data = mnist.test.images.reshape((-1, time_steps, n_input))
test_label = mnist.test.labels
print('Accuracy:', sess.run(accuracy, feed_dict={
      X: test_data, Y: test_label}))