[NLP] RNN(순환 신경망)

저번 장에서는 word2vec에 대해서 살펴보았다. 

 

word2vec은 타깃 단어에 대해서 주변 문맥 단어(정보)를 입력하고 이를 단서로 활용하여 타깃을 예측하는 모델이었다. 

 

word2vec의 단점은, 주변 문맥 단어를 특정 크기(ex) 2)의 윈도우로 한정 짓기 때문에 

 

큰 맥락 관점에서 문장을 이해하지 못할 수 있다. 

 

 

RNN(순환 신경망)

반면에 RNN(순환 신경망)은 각 시점(time)에 따라서 은닉 상태(hidden state)를 다음 시점(time)으로 전파함으로써

 

전체 문맥 정보를 기억한다는 장점을 가진 모델이다. 

 

이제부터 순환신경망을 구현해보자. 

 

import numpy as np

class RNN:
    def __init__(self, Wx, Wh, b):
        self.params = [Wx, Wh, b]
        self.grads = [np.zeros_like(Wx), np.zeros_like(Wh), np.zeros_like(b)]
        self.cache = None

순환 신경망은 입력 받은 X (임베딩된 상태의 단어)를 은닉 상태(H)로 바꿔주고 이를 해당 RNN의 출력으로 동시에

 

다음 시점의 RNN에 전달해준다. 

 

따라서 RNN에 사용할 가중치는 입력받은 X를 은닉 상태 벡터 H로 바꿔주는 가중치 Wx, 그리고 이전 시점의 RNN에서 

 

전달받은 H_[t-1] 을 나의 RNN에 받아들이기 위해 사용하는 가중치 Wh가 필요하다. 편향 b도 사용한다.

 

    def forward(self, x, h_prev):
        Wx, Wh, b = self.params
        t = np.matmul(h_prev, Wh) + np.matmul(x, Wx) + b
        h_next = np.tanh(t)

        self.cache = (x, h_prev, h_next)
        # use when back-propagate
        return h_next

RNN 계층의 순전파는 위와 같다. 

 

이전 시점의 은닉상태벡터(h_prev)에 가중치(Wh)를 취하고, 입력 정보(x)에도 가중치(Wx)를 취하고,

 

편향을 더한 결과를 하이퍼볼릭 탄젠트(tanh)에 통과시킨다. 이 결과가 시점 t의 은닉상태벡터이다. 

 

이 은닉상태벡터는 다음 시점의 단어에 대한 예측 정보이며, 다음 시점 RNN에 전달된다. 

 

    def backward(self, dh_next):
        Wx, Wh, b = self.params
        x, h_prev, h_next = self.cache

        dt = dh_next * (1 - h_next ** 2)
        db = np.sum(dt, axis=0)
        dWh = np.matmul(h_prev.T, dt)
        dh_prev = np.matmul(dt, Wh.T)
        dWx = np.matmul(x.T, dt)
        dx = np.matmul(dt, Wx.T)

        self.grads[0][...] = dWx
        self.grads[1][...] = dWh
        self.grads[2][...] = db

        return dx, dh_prev

역전파 시에는 다음 시점에서 흘러들어온 역전파 정보(dh_next)를 통해 해당 시점의 가중치 정보들을 계산하여 저장한다.

 

이로써 한 시점에 대한 RNN 계층의 구현을 마쳤다. 이 RNN은 'She'를 입력하였을 때 'is'를 출력하는 단순한 모델이다. 

 

이 RNN을 훈련시키기 위해서는 다수의 시점이 연결된 RNN들의 모음이 필요하다. 

 

이를 TimeRNN이란 이름으로 구현해보자. 

 

import numpy as np
from RNN import RNN

class TimeRNN:
    def __init__(self, Wx, Wh, b, stateful =False):
        self.params = [Wx, Wh, b]
        self.grads = [np.zeros_like(Wx), np.zeros_like(Wh), np.zeros_like(b)]
        self.layers = None

        self.h, self.dh = None, None
        self.stateful = stateful

stateful을 true로 설정하면 '상태가 있는' RNN 계층이 된다. 상태가 있다는 것은 순전파 시에

 

이전 시점의 RNN으로부터 은닉 상태를 이어받는 것을 끊지 않고 계속 수행한다는 의미이다. 

    def set_state(self, h):
        # 이전 층에서 h를 이어 받을 경우
        self.h = h

    def reset_state(self):
        self.h = None

stateful 정보를 조작할 수 있다. 

 

    def forward(self, xs): #xs를 입력받아 hs를 반환
        Wx, Wh, b = self.params
        N, T, D = xs.shape
        #배치사이즈, time사이즈, 임베딩사이즈
        D, H = Wx.shape
        #임베딩사이즈, 히든상태사이즈

        self.layers = []
        hs = np.empty((N, T, D), dtype='f')

        if not self.stateful or self.h is None:
            #stateful이 false거나 h가 None일 경우(truncated)
            self.h = np.zeros((N, H), dtype='f')

        for t in range(T):
            layer = RNN(*self.params)
            # 파라미터를 주입한 RNN 생성
            self.h = layer.forward(xs[:, t, :], self.h)
            # t 시점의 x를 입력받아서(배치처리) 새로운 t 시점의 h를 생성
            hs[:, t, :] = self.h
            # 그렇게 생성한 h를 hs의 t시점 요소에 저장
            self.layers.append(layer)
            # t 시점을 처리하는 RNN을 layer에 쌓는다
        return hs

forward 시에는 입력 정보(x)의 모음(xs)들을 입력받아  RNN 계층의 각 시점마다 순회하면서

 

새로운 은닉 상태(h)들을 생성해낸다. 최종적으로 이 은닉상태들의 모음(hs)을 반환한다. 

 

위 구현에서는 RNN이 시점마다 여러 개가 존재한다 하더라도 TimeRNN에서 사용하는 가중치를 공유하는 것으로 보인다. 

 

    def backward(self, dhs):
        Wx, Wh, b = self.params
        N, T, H = dhs.shape
        D, H = Wx.shape

        dxs = np.empty((N, T, D), dtype='f')
        dh = 0
        grads = [0,0,0]
        for t in reversed(range(T)):
            layer = self.layers[t]
            # 마지막 t 시점의 RNN부터
            dx, dh = layer.backward(dhs[:, t, :] + dh)
            #순전파 시 h_t를 2개로 분기하므로 역전파에서는 흘러온 dh를 합쳐서 새로운 dx, dh 계산
            dxs[:, t, :] = dx
            for i, grad in enumerate(layer.grads):
                grads[i] += grad
                #각각 Wx, Wh, b의 grad를 누적(T 동안의)

        for i, grad in enumerate(grads):
            self.grads[i][...] = grad
            #역전파를 거치면서 누적된 grad(Wx, Wh, b)를 저장
        self.dh = dh
        #가장 최초 시점의 dh를 저장?

        return dxs

역전파 시에는 가장 마지막 시점의 RNN부터 역으로 순회하면서 이전 시점의 RNN 계층으로 역전파 정보를 전달한다.

 

각 RNN 계층마다 흘러들어온 역전파 정보(dh)와 위에서부터 들어온 역전파 정보(dhs[:, t, :])를 합산하여 

 

새로운 그레디언트를 계산한다. 각 시점을 순회하면서 가중치의 그레디언트가 누적된다. 

 

순환 신경망 예측 과정

RNN 계층을 통해서 은닉 상태를 계산해낸 것만으로는 예측을 수행할 수 없다. 

 

이 은닉 상태는 최종 예측을 수행하기 위한 중간 과정에 존재하는 단서 정보이므로 최종 예측을 위해서는 

 

softmax를 통해 출현할 단어의 확률 계산을 해야한다. 

 

[입력 단어 - 임베딩 - RNN - 어파인 변환 - 소프트맥스] 과정을 한번에 수행하는 계층을 구현한다. 

 

import sys
sys.path.append('...')
from common.time_layers import *
from timeRNN import TimeRNN

class SimpleRnnlm:
    def __init__(self, vocab_size, wordvec_size, hidden_size):
        V, D, H = vocab_size, wordvec_size, hidden_size
        rn = np.random.randn

        embed_W = (rn(V, D)/100).astype('f')
        # 임베딩 수행 가중치
        rnn_Wx = (rn(D, H)/np.sqrt(D)).astype('f')
        rnn_Wh = (rn(H, H)/np.sqrt(H)).astype('f')
        rnn_b = np.zeros(H).astype('f')
        # RNN을 수행하는 가중치
        affine_W = (rn(H,V)/np.sqrt(H)).astype('f')
        affine_b = np.zeros(V).astype('f')
        # 어파인 변환을 위한 가중치

        self.layers = [
            TimeEmbedding(embed_W),
            TimeRNN(rnn_Wx, rnn_Wh, rnn_b, stateful=True),
            TimeAffine(affine_W, affine_b)
        ]
        self.loss_layer = TimeSoftmaxWithLoss()
        self.rnn_layer = self.layers[1]
        #TimeRNN 계층

        self.params, self.grads = [], []
        for layer in self.layers:
            self.params += layer.params
            self.grads += layer.grads

임베딩 계층, RNN 계층, 어파인 변환 계층을 쌓고, 마지막에 소프트맥스와 크로스 엔트로피 손실 계산을 수행하는 

 

계층을 최종적으로 적층한 모델을 설계하였다. 

 

    def forward(self, xs, ts):
        for layer in self.layers:
            xs = layer.forward(xs)
        loss = self.loss_layer.forward(xs, ts)
        return loss

각 계층을 순서대로 순전파하였다.

 

시점이 전체 1~10이라고 하였을 때, 먼저 1~10 시점의 임베딩을 수행한다.

 

10개의 임베딩 정보가 TimeRNN 계층에 전달되어 다시 10개의 은닉상태정보를 출력한다. 

 

다시 10개의 은닉상태 정보가 어파인 계층으로, 소프트맥스 계층으로 전달된다. 

    def backward(self, dout = 1):
        dout = self.loss_layer.backward(dout)
        for layer in reversed(self.layers):
            dout = layer.backward(dout)
        return dout

역전파 시에도 마찬가지로 위 계층부터 순차대로 전체 시점에 대한 역전파를 수행한다. 

 

전체 모델 설계가 완료되었으니 모델의 성능을 평가해보자.

import sys
sys.path.append('..')
import matplotlib.pyplot as plt
import numpy as np
from common.optimizer import SGD
from dataset import ptb
from simpleRnnlm import SimpleRnnlm

batch_size = 10
wordvec_size = 100
hidden_size = 100
time_size = 5
lr = 0.1
max_epoch = 100

corpus, word_to_id, id_to_word = ptb.load_data('train')
corpus_size = 100
corpus = corpus[:corpus_size]
vocab_size = int(max(corpus)+1)

xs = corpus[:-1]
ts = corpus[1:]
data_size = len(xs)

max_iters = data_size // (batch_size*time_size)
time_idx = 0
total_loss = 0
loss_count = 0
ppl_list = []

model = SimpleRnnlm(vocab_size, wordvec_size, hidden_size)
optimizer = SGD(lr)

jump = (corpus_size - 1)//batch_size
offsets = [i*jump for i in range(batch_size)]

for epoch in range(max_epoch):
    for iter in range(max_iters):
        batch_x = np.empty((batch_size, time_size), dtype='i')
        batch_t = np.empty((batch_size, time_size), dtype='i')
        for t in range(time_size):
            for i, offset in enumerate(offsets):
                batch_x[i, t] = xs[(offset + time_idx)%data_size]
                batch_t[i, t] = ts[(offset + time_idx)%data_size]

        loss = model.forward(batch_x, batch_t)
        model.backward()
        optimizer.update(model.params, model.grads)
        total_loss += loss
        loss_count += 1

    ppl = np.exp(total_loss / loss_count)
    print('| epoch %d | perplexity %.2f'%(epoch+1, ppl))
    ppl_list.append(float(ppl))
    total_loss, loss_count = 0, 0

plt.plot(ppl_list)
plt.show()

평가 지표는 퍼플렉시티(Perplexity)로, 타깃 단어에 대한 모델이 계산한 확률의 역수이다.

 

타깃 단어가 출현할 확률이 높다고 평가했을 때 정확한 모델이므로 

 

이 확률이 높을수록, 역수인 퍼플렉시티가 낮을수록 좋은 모델이라고 할 수 있다. 

 

 

각 에포크 단계가 진전될수록 퍼플렉시티가 낮아지는 것을 확인할 수 있다.