[NLP] 코드로 보는 RNN

김성훈 교수님의 모두의 딥러닝 강의의 RNN 구현 코드는 최신 Tensorflow 버전에서는 안 돌아간다. 그래서 구글링해가면서 다시 구현했다. 이왕 구현하는 김에 객체화해서 구현해보았다.

 

1. 모델 구축

def build(self):
        self.x_data = tf.placeholder(shape=self.input_size, dtype=tf.float32)
        rnn_cell = tf.nn.rnn_cell.BasicRNNCell(self.rnn_size)
        initial_state = tf.random_normal(shape=(self.batch_size, rnn_cell.state_size), mean=1.0)
        # cell, inputs, dtype
        outputs, _ = tf.nn.dynamic_rnn(rnn_cell, self.x_data, dtype=tf.float32, 
                                       initial_state=initial_state)

먼저, tf.nn.rnn_cell.BasicRNNCell 함수를 이용해서 RNN cell을 생성해준다. 인수는 사용할 hidden_state 개수라고 생각하면 될 것 같다.

RNN size를 1로 하면, 위와 같은 RNN Cell 하나가 생성된다.

혹은, RNN size를 조정하여 위 그림처럼 다양한 모델을 생성할 수 있다.

RNN에서 현 시점의 hidden state는 이전 hidden state와 현 시점 입력 값에 의해 결정된다. 하지만 첫 시점의 hidden state는 이전 hidden state가 없기 때문에 별도의 초기 state가 필요하다. 따라서 initial_state를 tf.random_normal 함수로 초기화하였다. 사실 tf.zero로 초기 상태를 모두 0으로 정의하거나 초기 상태를 만들지 않아도 상관없다. 초기 학습이 조금 빠르지 않을까 하는 생각에 했지만, 이런 작은 모델에서는 약간의 차이도 볼 수 없었다.

tf.nn.dynamic_rnn 함수는 입력과 RNN Cell을 받아 실행하여 결과 값을 출력하는 함수다. tf.nn.static_rnn 이라는 함수도 있는데 이 둘의 차이는 What's the difference between tensorflow dynamic_rnn and rnn? 에 잘 나와있다. 대충 요약하면, dynamic_rnn이 더 빠르게 그래프를 생성하고 다양한 크기의 batch를 받아들일 수 있다.

 

2. 모델 훈련

​x
def fit(self, train_data, target_data):
        # [batch_size, sequence_length, num_decoder_symbols]
        logits = tf.reshape(self.outputs, [self.batch_size, self.rnn_size, self.num_decoder])
        # [batch_size, sequence_length]
        targets = tf.reshape(target_data, (self.batch_size, self.rnn_size))
        # [batch_size, sequence_length]
        weights = tf.ones(shape=[self.batch_size, self.rnn_size])
        
        self.loss = tf.contrib.seq2seq.sequence_loss(logits, targets, weights)
        cost = tf.reduce_sum(self.loss) / self.batch_size
        self.train_op = tf.train.RMSPropOptimizer(0.01, 0.9).minimize(cost)
​
        # Launch the graph in a session
        with tf.Session() as sess:
            tf.global_variables_initializer().run()
            for i in range(self.iter):
                _, loss = sess.run((self.train_op, self.loss),
                                   feed_dict={self.x_data: train_data})
                self.result = sess.run(tf.argmax(logits, -1),
                                       feed_dict={self.x_data: train_data})
                accuracy = np.mean(self.result==target_data)
                if i % 20 == 0:
                    print("step: {}, loss: {:.3f}, acc: {:.2f}".format(i, loss, accuracy))

tf.contrib.seq2seq.sequence_loss 함수를 통해 loss를 계산할 수 있다. 범용적으로 만들어진 함수이기 때문에 반드시 입력 형태를 맞춰주어야 한다. 입력은 모델에서 계산된 logits, 매칭시킬 targets, 그리고 weights 총 세 가지다. 각각의 입력 형태는 다음과 같다.

• logits : (batch_size, sequence_length, num_decoder_symbols)
• targets : (batch_size, sequence_length)
• weights : (batch_size, sequence_length)

sequence_length는 rnn_size라고 생각하면 될듯하다.

optimizer는 RMSProp과 Adam 두 개로 실험해봤는데 RMSProp이 더 안정된 성능을 보였다.

 

3. 입력 형태

학습할 단어를 one-hot 벡터로 임베딩하여 사용하기 위해 사전(Vocabulary)가 필요하다. 일단 강의에 나온 것처럼 "hello"를 학습시키기 위해 4개의 글자로 이루어진 리스트를 정의했다. 그런 다음 word: id로 이루어진 사전을 생성한다.

xxxxxxxxxx
char_rdic = ['h', 'e', 'l', 'o']
# create vocab
char_dic = {word : id for id, word in enumerate(char_rdic)}
data = ["hello"]

현재는 하나의 단어만 학습하지만 여러 개의 단어를 한 번에 학습할 수 있도록 입력 형태를 만들어준다. 여기서는 마지막 글자를 제외한 글자들이 학습 데이터이고 첫 글자를 제외한 나머지 글자들이 target 데이터가 된다.

xxxxxxxxxx
samples = []
x_data = []
for word in data:
    samples.append([char_dic[c] for c in word[1:]])
    x_data.append([c for c in word[:-1]])   
​
x_data = one_hot_embedding(np.asarray(x_data), char_dic)

one hot 임베딩을 거치기 전 x_data와 samples의 형태는 다음과 같다.

xxxxxxxxxx
x_data = [['h', 'e', 'l', 'l']]
samples = [[1, 2, 2, 3]]

x_data는 one hot 임베딩을 거치면 다음과 같이 학습을 위한 최종 형태가 완성된다.

xxxxxxxxxx
x_data = [[[1,0,0,0],
           [0,1,0,0],
           [0,0,1,0],
           [0,0,1,0]]]

이제 학습을 시키면 다음과 같이 안정적으로 학습이 되는 것을 확인할 수 있다.

xxxxxxxxxx
step: 0, loss: 1.606, acc: 0.25
step: 20, loss: 1.459, acc: 0.25
step: 40, loss: 1.108, acc: 0.50
step: 60, loss: 0.958, acc: 0.75
step: 80, loss: 0.781, acc: 1.00
step: 100, loss: 0.756, acc: 1.00

 

전체 코드

xxxxxxxxxx
import numpy as np
import tensorflow as tf
from embedding import one_hot_embedding
​
class Char_RNN(object):
    def __init__(self, input_size, rnn_size, num_decoder, vocab_dict, iter=100):
        self.vocab_dict = vocab_dict
        self.input_size = input_size
        self.rnn_size = rnn_size
        self.time_step_size = rnn_size
        self.batch_size = input_size[0]
        self.num_decoder = num_decoder
        self.iter = iter
        self.build()
​
    def build(self):
        self.x_data = tf.placeholder(shape=self.input_size, dtype=tf.float32)
        rnn_cell = tf.nn.rnn_cell.BasicRNNCell(self.rnn_size)
        initial_state = tf.random_normal(shape=(self.batch_size, rnn_cell.state_size), mean=1.0)
        # cell, inputs, dtype
        outputs, _ = tf.nn.dynamic_rnn(rnn_cell, self.x_data, dtype=tf.float32, 
                                       initial_state=initial_state)
        # [batch_size, sequence_length, num_decoder_symbols]
        self.logits = tf.reshape(outputs, [self.batch_size, self.rnn_size, self.num_decoder])
​
    def fit(self, train_data, target_data):
        # [batch_size, sequence_length]
        targets = tf.reshape(target_data, (self.batch_size, self.rnn_size))
        # [batch_size, sequence_length]
        weights = tf.ones(shape=[self.batch_size, self.rnn_size])
        
        self.loss = tf.contrib.seq2seq.sequence_loss(self.logits, targets, weights)
        cost = tf.reduce_sum(self.loss) / self.batch_size
        self.train_op = tf.train.RMSPropOptimizer(0.01, 0.9).minimize(cost)
​
        # Launch the graph in a session
        with tf.Session() as sess:
            tf.global_variables_initializer().run()
            for i in range(self.iter):
                _, loss = sess.run((self.train_op, self.loss), feed_dict={self.x_data:
                                                                          train_data})
                self.result = sess.run(tf.argmax(self.logits, -1), feed_dict={self.x_data:
                                                                              train_data})
                accuracy = np.mean(self.result==target_data)
                if i % 50 == 0:
                    print("step: {}, loss: {:.3f}, acc: {:.2f}".format(i, loss, accuracy))
    
    def decode(self):
        decoded = []
        reverse_vocab = {id : w for w, id in self.vocab_dict.items()}
        for batch in self.result:
            decoded.append(''.join([reverse_vocab[id] for id in batch]))
        return decoded
​
if __name__=="__main__":
    char_rdic = ['h', 'e', 'l', 'o', 'u','r','s'] # id -> char
    # create vocab
    char_dic = {word : id for id, word in enumerate(char_rdic)} # char -> id
    
    data = ["hello"]
    samples = []
    x_data = []
    for word in data:
        samples.append([char_dic[c] for c in word[1:]])
        x_data.append([c for c in word[:-1]])   
            
    x_data = one_hot_embedding(np.asarray(x_data), char_dic)
​
    char_rnn = Char_RNN(x_data.shape, x_data.shape[1], x_data.shape[1], char_dic, iter=100)
    char_rnn.fit(x_data, samples)
    print(char_rnn.decode())

 

4. 모델 변경

위와 비슷하게 단어의 앞 세 글자가가 주어지면 뒤에 올 마지막 글자를 맞추는 모델로 변경할 수도 있다. 입력의 크기를 조절하는 단계에서 잘 되지 않아 sequence_loss 함수를 사용하지 않고 구현했다.

코드