[#E19] Recurrent Neural Network (RNN - PART 1/3)

RECURRENT NEURAL NETWORK (RNN) 은 도출된 이전의 시퀸스 weight 값이 다음 시퀸스 weight값에 영향을 미치면서 학습하는 모델입니다.

이것은 다음 그림과 같이 각 시퀸스들의 연결로 동작합니다.

 

위의 그림을 간단한 함수 식으로 나타내면 아래와 같습니다.

 

 

(h = hidden layer)

따라서 새로운 state는 이전의 state의 영향을 받는 구조가 됩니다. 이에 따른 weight는 총 3개가 존재하며, 각각 이전 state(Ht-1) 에 대한 weight, 새로운 state의 x값(Xt) 에 대한 weight, y 값에 대한 새로운 state(Ht) 의 weight 값입니다.

이 3개의 weight값을 이용해서 RNN 공식을 구현하면 다음과 같이 표현됩니다.

 

 

(이것은 크게 중요한 내용이 아닙니다.)

 

이제 hello 문자열을 출력하기 위한 학습에 RNN 모델을 적용해보겠습니다.

(각 레이어들은 담긴 데이터에 해당하는 weight 값을 곱해서 생성됩니다.)

사용된 x 데이터는 각각의 문자열을 ONE-HOT 인코딩한 것이며, 우리가 이 학습으로 바라는 이상적인 결과는 h 다음에는 e, e 다음에는 l, l 다음에는 l, 그리고 마지막 l 다음에는 o가 오는 것입니다.

즉 이상적인 y 예측값은 e, l, l, o 입니다.

output layers의 문자열 'h' 시퀸스를 보면, 가장 큰 값은 4.1으로 결과는 문자열 'o' (0, 0, 0, 1)를 지정합니다. 따라서 이는 부정확한 결과를 나타내므로 에러값 - 즉 Cost 값이 됩니다.

RNN은 이러한 Cost 값을 minimize하면서 정확한 예측값을 내게 됩니다.

이제 텐서플로우에서 이 모델을 바탕으로 각 시퀸스를 어떻게 구성하는지 알아보겠습니다.

 

먼저 문자열 h를 포함하는 하나의 시퀸스를 구현합니다.. 시퀸스의 구조는 다음과 같습니다.

hidden_size는 output이 갖는 결과의 수입니다. 이것은 사용자 임의대로 설정할 수 있습니다. 이번에는 hidden_size를 2로 지정한 후, 해당 시퀸스를 구현합니다.

(shape의 구조에 대해서는 잠시 후 설명합니다)

먼저 시퀸스를 포함하는 cell을 생성해야 하는데, 이것은 텐서플로우가 제공하는 BasicRNNCell을 이용해서 쉽게 구현할 수 있습니다.

cell = tf.contrib.rnn.BasicRNNCell(num_units=hidden_size)
...
outputs, _states = tf.nn.dynamic_rnn(cell, x_data, dtype=tf.float32)

 

아래의 코드는 학습이 아닌 단순히 제공한 x값에 대한 output (y) 값을 출력하는 예제입니다. 

SORUCE CODE

import tensorflow as tf
import numpy as np
import pprint

 
#pp setting
pp = pprint.PrettyPrinter(indent=4)
 
#new tensorflow session
sess = tf.InteractiveSession()
 
#data
h = [1, 0, 0, 0]
 
with tf.variable_scope('one_cell') as scope:
    # One cell RNN input_dim (4) -> output_dim (2)
    hidden_size = 2
 
    cell = tf.contrib.rnn.BasicRNNCell(num_units=hidden_size)
    print(cell.output_size, cell.state_size)
 
    x_data = np.array([[h]], dtype=np.float32) # x_data = [[[1,0,0,0]]]
 
    outputs, _states = tf.nn.dynamic_rnn(cell, x_data, dtype=tf.float32)

    sess.run(tf.global_variables_initializer())
    pp.pprint(outputs.eval())

OUTPUT

array([[[ 0.28406101,  0.53163123]]], dtype=float32)

 

이번에는 hello의 각 문자열을 포함한 시퀸스를 구현합니다. 이것은 총 5개의 시퀸스가 연결된 형태로, 아래의 그림을 참고하세요.

SOURCE CODE

import tensorflow as tf
import numpy as np
from tensorflow.contrib import rnn
import pprint
 
#pp setting
pp = pprint.PrettyPrinter(indent=4)
 
#new tensorflow session
sess = tf.InteractiveS
 
h = [1, 0, 0, 0]
e = [0, 1, 0, 0]
l = [0, 0, 1, 0]
o = [0, 0, 0, 1]
 
with tf.variable_scope('two_sequances') as scope:
    # One cell RNN input_dim (4) -> output_dim (2). sequence: 5
    hidden_size = 2
 
    cell = tf.contrib.rnn.BasicRNNCell(num_units=hidden_size)
    print(cell.output_size, cell.state_size)
 
    x_data = np.array([[h, e, l, l, o]], dtype=np.float32)
 
    outputs, _states = tf.nn.dynamic_rnn(cell, x_data, dtype=tf.float32)
 
    sess.run(tf.global_variables_initializer())
    pp.pprint(outputs.eval())

OUTPUT

array([[[ 0.50641137,  0.55783308],
        [-0.61545879, -0.04334207],
        [-0.77109283, -0.23411733],
        [-0.76478487, -0.07172935],
        [-0.72683465,  0.5266667 ]]], dtype=float32)

 

이번에는 batch를 사용한 시퀸스를 구현합니다. 구조는 아래의 그림을 참고하세요.

이 모델은 batch = 3이므로 3개의 문자열 데이터 셋을 사용합니다. [hello, eolll, lleel] 이것은 학습을 효율적으로 더 많이 진행할 수 있습니다.

이제 Shape는 각각 무엇을 의미하는지 알 수 있습니다. 위의 shape (3, 5, 4)에서 각각의 숫자는 batch size, squence 길이, output 출력 갯수를 의미합니다.

이제 RNN 모델을 활용하여 "hi hello"를 출력하는 RNN을 구성해보겠습니다.

 

구성할 RNN의 구조는 아래와 같습니다.

그림의 구조를 토대로, 먼저 데이터 및 hyper parameter을 지정합니다.

 

import tensorflow as tf
import numpy as np

 
idx2char = ['h', 'i', 'e', 'l', 'o']
# Teach hello: hihell -> ihello
x_data = [[0, 1, 0, 2, 3, 3]]   # hihell
x_one_hot = [[[1, 0, 0, 0, 0],   # h 0
              [0, 1, 0, 0, 0],   # i 1
              [1, 0, 0, 0, 0],   # h 0
              [0, 0, 1, 0, 0],   # e 2
              [0, 0, 0, 1, 0],   # l 3
              [0, 0, 0, 1, 0]]]  # l 3
 
y_data = [[1, 0, 2, 3, 3, 4]]    # ihello
 
#hyper parameter
num_classes = 5 # unique, h, i, e, l, o
input_dim = 5  # one-hot size // ex) h = [1, 0, 0, 0, 0]
hidden_size = 5  # output size
batch_size = 1   # only one sentence
sequence_length = 6  # |ihello| == 6
learning_rate = 0.1

 

이제 cell을 구성할 차례입니다. 이는 아래의 코드를 참고하세요.

X = tf.placeholder(
    tf.float32, [None, sequence_length, input_dim])  # X one-hot
Y = tf.placeholder(tf.int32, [None, sequence_length])  # Y label
 
cell = tf.contrib.rnn.BasicLSTMCell(num_units=hidden_size, state_is_tuple=True)
 
initial_state = cell.zero_state(batch_size, tf.float32)
 
outputs, _states = tf.nn.dynamic_rnn(
    cell, X, initial_state=initial_state, dtype=tf.float32)
 
weights = tf.ones([batch_size, sequence_length])
 
sequence_loss = tf.contrib.seq2seq.sequence_loss(
    logits=outputs, targets=Y, weights=weights)
 
loss = tf.reduce_mean(sequence_loss)
train = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)
 
prediction = tf.argmax(outputs, axis=2)

BasicLSTMCell 은 기존 BasicRNNCell보다 보완된 형태의 Cell입니다. 이는 Vanishing Gradient Problem을 예방합니다.

initial_state는 처음 시퀸스 값을 위한 변수입니다. 처음 시퀸스는 이전 시퀸스의 값이 없기 때문에 0으로 지정합니다.

weights는 일단 1이라고 생각합니다.

sequence_loss는 실제 Y값과 모델의 Y예측값 간의 차이입니다. (Cost)

중요한 것은 위의 코드처럼 output을 직접 logits에 대입하면 안된다는 점입니다. 이것은 SoftMax 함수 등으로 가공해야 하는데, 이 학습에서는 가공하지 않고 그대로 사용합니다.

이제 아래의 코드로 우리의 모델을 학습시킬 수 있습니다.

with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    for i in range(50):
        l, _ = sess.run([loss, train], feed_dict={X: x_one_hot, Y: y_data})
        result = sess.run(prediction, feed_dict={X: x_one_hot})
        print(i, "loss:", l, "prediction: ", result, "true Y: ", y_data)
 
        # print char using dic
        result_str = [idx2char[c] for c in np.squeeze(result)]
        print("\tPrediction str: ", ''.join(result_str))

실행시켜보면 loss는 계속 줄어들고, 제공한 Y 데이터 값 ihello를 성공적으로 출력합니다.

OUTPUT

49 loss: 0.71805996 prediction:  [[1 0 2 3 3 4]] true Y:  [[1, 0, 2, 3, 3, 4]]
    Prediction str:  ihello

 

그런데 학습 때마다 일일이 x 데이터의 ONE-HOT 인코딩된 값을 지정하거나, hyper parameter 값을 지정하는 것은 귀찮고 번거로운 일입니다.

따라서 이것을 자동화하면 훨씬 더 효율적으로 작업할 수 있습니다.

먼저 다음과 같이 x, y 데이터 및 hyper parameter를 설정합니다. 이번에는 미리 ONE-HOT 인코딩된 값을 주는 것이 아닌, 이를 코드로 직접 변환합니다.

또한 이외의 hyper parameter 값들도 다음과 같이 자동으로 계산하게끔 구현합니다.

import tensorflow as tf
import numpy as np
 
sample = " if you want you"
idx2char = list(set(sample))  # make this -> unique 'i', 'f', 'y' ...
char2idx = {c: i for i, c in enumerate(idx2char)}  # '1:i','2:f' ...
 
# hyper parameters
input_size = len(char2idx)  # RNN input size (one hot size)
hidden_size = len(char2idx)  # RNN output size
num_classes = len(char2idx)  # final output size (RNN or softmax, etc.)

batch_size = 1  # one sample data, one batch
sequence_length = len(sample) - 1  # number of lstm rollings (unit #)
learning_rate = 0.1
 
sample_idx = [char2idx[c] for c in sample]  # char to index
x_data = [sample_idx[:-1]]  # X data sample (0 ~ n-1) hello: hell
y_data = [sample_idx[1:]]   # Y label sample (1 ~ n) hello: ello
 

이제 자동화된 parameter들을 토대로 주어진 x_data를 ONE-HOT 인코딩 후 셀을 구성합니다. 셀 구성은 직접 ONE-HOT 인코딩을 진행하는 것을 제외하고는 위의 'Hi Hello' 트레이닝과 다르지 않습니다.

X = tf.placeholder(tf.int32, [None, sequence_length])  # X data
Y = tf.placeholder(tf.int32, [None, sequence_length])  # Y label
 
x_one_hot = tf.one_hot(X, num_classes)  # one hot: 1 -> 0 1 0 0 0 0 0 0 0 0
 
cell = tf.contrib.rnn.BasicLSTMCell(
    num_units=hidden_size, state_is_tuple=True)
 
initial_state = cell.zero_state(batch_size, tf.float32)
 
outputs, _states = tf.nn.dynamic_rnn(
    cell, x_one_hot, initial_state=initial_state, dtype=tf.float32)
 
weights = tf.ones([batch_size, sequence_length])
 
sequence_loss = tf.contrib.seq2seq.sequence_loss(
    logits=outputs, targets=Y, weights=weights)
 
loss = tf.reduce_mean(sequence_loss)
train = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)
 
prediction = tf.argmax(outputs, axis=2)

이제 구성된 RNN은 아래의 코드로 학습시킬 수 있습니다.

with tf.Session() as sess:
    sess.run(tf.global_variables_initializer())
    for i in range(50):
        l, _ = sess.run([loss, train], feed_dict={X: x_data, Y: y_data})
        result = sess.run(prediction, feed_dict={X: x_data})
 
        # print char using dic
        result_str = [idx2char[c] for c in np.squeeze(result)]
 
        print(i, "loss:", l, "Prediction:", ''.join(result_str))

OUTPUT

49 loss: 0.00029237865 Prediction: f you want you

마지막으로, 매우 긴 문장을 위한 LONG SEQEUNCE RNN 모델을 살펴보겠습니다. 이것은 위의 RNN과 크게 다르지 않으나, 각 x, y 데이터를 원하는 sequence size마다 잘라내어 순차적으로 학습합니다.

먼저 아까와 같이 hyper parameter들을 설정합니다.

import tensorflow as tf
import numpy as np

 
sentence = ("if you want to build a ship, don't drum up people together to "
            "collect wood and don't assign them tasks and work, but rather "
            "teach them to long for the endless immensity of the sea.")
 
idx2char = list(set(sentence))
char2idx = {w: i for i, w in enumerate(idx2char)}
 
data_dim = len(idx2char)
hidden_size = len(idx2char)
num_classes = len(idx2char)
sequence_length = 10  # Any arbitrary number
learning_rate = 0.1

이제 x, y 데이터를 분리해서 따로 dataX, dataY에 넣는 과정이 필요합니다. 이것은 아래의 코드로 구현할 수 있습니다.

dataX = []
dataY = []
 
for i in range(0, len(sentence) - sequence_length):
    x_str = sentence[i:i + sequence_length]
    y_str = sentence[i + 1: i + sequence_length + 1]
    print(i, x_str, '->', y_str)
 
    x = [char2idx[c] for c in x_str]  # x str to index
    y = [char2idx[c] for c in y_str]  # y str to index
 
    dataX.append(x)
    dataY.append(y)
 
batch_size = len(dataX)

이것은 순수한 파이썬 코드입니다. 이 과정을 통해 총 169개의 x 데이터가 생기는데, 이를 batch_size로 몇 개씩 학습할 지 임의로 숫자를 정할 수 있습니다. 여기에서는 모든 데이터를 한번에 학습합니다.

ONE-HOT 인코딩 및 셀 구성은 이전과 같습니다.

X = tf.placeholder(tf.int32, [None, sequence_length])
Y = tf.placeholder(tf.int32, [None, sequence_length])
 
# One-hot encoding
X_one_hot = tf.one_hot(X, num_classes)
print(X_one_hot)  # check out the shape
 
cell = tf.contrib.rnn.BasicLSTMCell(hidden_size, state_is_tuple=True)
 
initial_state = cell.zero_state(batch_size, tf.float32)
 
outputs, _states = tf.nn.dynamic_rnn(
    cell, X_one_hot, initial_state=initial_state, dtype=tf.float32)
 
weights = tf.ones([batch_size, sequence_length])
 
sequence_loss = tf.contrib.seq2seq.sequence_loss(
    logits=outputs, targets=Y, weights=weights)
 
loss = tf.reduce_mean(sequence_loss)
 
train = tf.train.AdamOptimizer(learning_rate=learning_rate).minimize(loss)
 
sess = tf.Session()
sess.run(tf.global_variables_initializer())
 

이제 다음 코드로 학습 및 테스트를 진행할 수 있습니다.

#learning
for i in range(500):
    _, l, results = sess.run(
        [train, loss, outputs], feed_dict={X: dataX, Y: dataY})
    for j, result in enumerate(results):
        index = np.argmax(result, axis=1)
        print(i, j, ''.join([idx2char[t] for t in index]), l)
 
# Let's print the last char of each result to check it works
results = sess.run(outputs, feed_dict={X: dataX})
for j, result in enumerate(results):
    index = np.argmax(result, axis=1)
    if j is 0:  # print all for the first result to make a sentence
        print(''.join([idx2char[t] for t in index]), end='')
    else:
        print(idx2char[index[-1]], end='') # get last index