(2)-5-2 RNN 구현

(1) RNN 구현

 

구현해야 할 신경망은 가로 방향으로 성장한 신경망입니다.

이 가로 방향으로 성장하는 신경망을 '하나의 계층'으로 구현하면 다음과 같습니다.

순환 구조를 펼친 후의 계층들을 하나의 계층으로 간주, 길이가 T인 시계열 데이터

위의 그림과 같이 상하 방향의 입출력을 하나로 묶으면 하나의 계층으로 볼 수 있습니다.

 

Time RNN 계층 내에서 한 단계의 작업을 수행하는 계층을 'RNN 계층'이라고 하고,

T개 단계분의 작업을 한꺼번에 처리하는 계층을 'Time RNN 계층'이라고 합니다.

(시계열 데이터를 한꺼번에 처리하는 계층 앞에는 'Time'이라는 표현을 붙이겠음)

 

 

① RNN 구현

 

미니 배치를 적용한 RNN의 식은 다음과 같습니다.

N:미니배치 크기, D:입력 벡터의 차원수, H: 은닉 상태 벡터의 차원수

이를 바탕으로 작성한 RNN 클래스의 코드는 다음과 같습니다.

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

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

        self.cache = (x, h_prev, h_next)
        return h_next

    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.dot(h_prev.T, dt)
        dh_prev = np.dot(dt, Wh.T)
        dWx = np.dot(x.T, dt)
        dx = np.dot(dt, Wx.T)

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

        return dx, dh_prev

 

우선 순전파를 보겠습니다.

params 변수에 가중치 2개와 편향 1개를 저장합니다.

grads 변수에 각 매개변수에 대응하는 형태로 기울기를 초기화한 후 저장합니다.

cache 변수에 역전파 계산 시 사용하는 중간 데이터를 담을 공간을 마련합니다.

 

순전파 진행은 위의 식대로 행렬 곱 후 tanh 처리를 해줍니다.

순전파의 입력 데이터는 h_prev, 출력 데이터는 h_next가 됩니다.

그림은 다음과 같이 그려집니다.

편향 b의 덧셈에서는 Repeat노드를 이용함(여기서 그냥 표기 안함) Repeat노드는 딥러닝 1-4글에서 볼 수 있습니다.

역전파는 순전파와 반대로 오차역전파법을 구현합니다.

 

 

 

② Time RNN 구현  

처음에 Time RNN의 구조를 그림으로 보여드렸습니다.

그리고 다음과 같이 RNN 계층의 은닉 상태 h를 인스턴스 변수를 '인계'받는 용도로 이용하기 위해 유지합니다.

Rnn 계층의 은닉 상태(h)를 Time RNN 계층에서 관리함으로써

Time RNN 사용자는 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

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

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

        if not self.stateful or self.h is None:
            self.h = np.zeros((N, H), dtype='f')

        for t in range(T):
            layer = RNN(*self.params)
            self.h = layer.forward(xs[:, t, :], self.h)
            hs[:, t, :] = self.h
            self.layers.append(layer)

        return hs

    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]
            dx, dh = layer.backward(dhs[:, t, :] + dh)
            dxs[:, t, :] = dx

            for i, grad in enumerate(layer.grads):
                grads[i] += grad

        for i, grad in enumerate(grads):
            self.grads[i][...] = grad
        self.dh = dh

        return dxs

    def set_state(self, h):
        self.h = h

    def reset_state(self):
        self.h = None

이를 나눠서 보겠습니다.

 

우선 생성자 초기화 부분을 보겠습니다.

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

    def set_state(self, h):
        self.h = h

    def reset_state(self):
        self.h = None
    # 확장성을 고려하여 Time RNN 계층의 은닉 상태를 설정하는 메서드(set, reset)를 구현함

초기화 메서드는 가중치와 편향, stateful이라는 boolean 값을 인수로 받습니다.

layers 인스턴스 변수는 RNN 계층을 리스트로 저장하는 용도입니다.

인스턴스 변수 h는 forward() 메서드를 불렀을 때 마지막 RNN 계층의 은닉 상태를 저장합니다.

dh는 backward()를 불렀을 때 하나 앞 블록의 은닉 상태의 기울기를 저장합니다.

 

stateful 인수는 은닉 상태를 유지할지 지정할 수 있는 변수입니다. (True면 유지)

'유지한다'라는 의미는 해당 시간대의 은닉 상태를 메모리에서 기억하고 있겠다는 의미입니다.

긴 시계열 데이터를 처리할 때 RNN의 은닉 상태를 유지해야 하며, 

이는 Time RNN 계층의 순전파를 끊지 않고 전파한다는 의미가 내포되어 있습니다.

 

순전파의 코드는 다음과 같습니다.

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

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

        if not self.stateful or self.h is None:
            self.h = np.zeros((N, H), dtype='f')

        for t in range(T):
            layer = RNN(*self.params) # *은 리스트의 원소들을 추출하여 메서드의 인수로 전달
            self.h = layer.forward(xs[:, t, :], self.h)
            hs[:, t, :] = self.h
            self.layers.append(layer)

        return hs

매개변수 인자 xs를 받으며

xs의 형상은 (N, T, D)입니다.

N : 미니 배치 크기

T : T개 분량 시계열 데이터를 하나로 모은 것

D : 입력 벡터의 차원 수

 

RNN 계층의 은닉 상태 h는 처음 호출 시 원소가 모두 0인 영행렬로 초기화됩니다.

그리고 인스턴스 변수 stateful이 False일 때도 항상 영행렬로 초기화됩니다.

 

hs는 출력 값을 담을 변수입니다.

 

이어서 총 T회 반복문으로 RNN 계층 T개 생성하며,

계층들을 layers변수에 layer변수를 이용하여 추가하고

RNN 계층이 각 시각 t의 은닉 상태 h를 계산하여 이를 hs에 해당 인덱스(시각)의 값으로 설정합니다.

이 반복문에서 RNN 계층에 Wx와 Wh와 b가 담겨 있는 self.params을 넘길 때 

T에 대해 매번 같은 놈을 넣는 것을 확인할 수 있습니다.

그 이유는 바로 RNN 본질적인 의미에서 알 수 있습니다.

지금 보고 있는 RNN 계층들은 사실 하나의 계층이기 때문입니다.

RNN 계층을 시간적으로 가로로 표현해주기 위해서 여러 계층처럼 보이는 것이지 

본질적으로는 하나의 계층입니다.

그러므로 매 번 같은 가중치 조건들을 넣어 주는 것입니다.

(일단은 한 Time RNN 계층 내의 RNN 계층들은 가중치가 모두 같음.

 하지만 서로 다른 Time RNN 계층의 가중치 조건은 사실은 서로 다름. 이 부분은 나중에 설명 )

(사실 이 부분은 앞서 간략하게 설명하긴 함)

 

Time RNN 계층의 forward() 메서드가 불리면, 

인스턴스 변수 h에는 마지막 RNN 계층은 은닉 상태가 저장됩니다.

그래서 다음번 forward() 메서드 호출 시 stateful이 True이면 먼저 저장된 h 값이 그대로 이용되고

False면 h가 다시 영행렬로 초기화됩니다.

 

 

 

 

역전파를 보겠습니다.

역전파에서는 Truncated BPTT를 수행하기 때문에

이 블록의 이전 시각 역전파는 필요하지 않습니다.

단, 이전 시각의 은닉 상태 기울기는 인스턴스 변수 dh에 저장해 놓겠습니다.

 

 

그림 중 2번째 RNN이 t번째 RNN 계층 

역전파에서 t번째 RNN 계층에서는 

위로부터의 기울기 dht와 

한 시각 뒤(미래) 계층으로부터의 기울기 dhnext가 전해집니다.

RNN 계층 순전파에서의 출력이 2개로 분기되었으므로

역전파에서는 각 기울기가 합산되어 전해집니다.

따라서 역전파 시 RNN 계층에는 합산된 기울기(dht + dhnext)가 입력됩니다.

 

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] # [np.zeros_like(Wx), np.zeros_like(Wh), np.zeros_like(b)]
        for t in reversed(range(T)): # 순전파와 반대로
            layer = self.layers[t]
            dx, dh = layer.backward(dhs[:, t, :] + dh)  #dhnext + dh 합산된 기울기
            dxs[:, t, :] = dx

            for i, grad in enumerate(layer.grads):
                grads[i] += grad  # RNN에서 backward하고 나온 grad를 grads에 넣기

        for i, grad in enumerate(grads):
            self.grads[i][...] = grad  # self.grads는 TimeRNN객체의 grads
        self.dh = dh

        return dxs

역전파에서도 하류로 흘려보낼 기울기를 담을 그릇인 dxs를 만들고,

순전파와는 반대 순서로 RNN 계층의 backward() 메서드를 호출하여,

각 시각의 기울기 dx를 구해서 dxs의 해당 인덱스(시각)에 저장합니다.

 

앞서 순전파부분 설명에서 '지금 설명하고 있는 RNN계층은 본질적으로 하나의 계층을 시간적으로 표현한 것이다' 라고 했습니다.

그러므로 순전파에서는 가중치들의 조건은 같았지만,

역전파에서는 시간마다 grad(기울기)들이 모두 다를 것입니다.

그 이유는 입력받은 미분(dht) 값들이 모두 다르기 때문입니다.

(순전파 입장으로 말하자면 Xs값들이 각 시각마다 다르기 때문)

 

 

가중치 매개변수에 대해서도 각 RNN 계층의 가중치 기울기를 합산하여 최종 결과를 

TimeRNN클래스의 멤버 변수 self.grads에 덮어씁니다.

위의 backward()코드 이해하는데 참조 코드

 

 

 

(2) 시계열 데이터 처리 계층 구현

하고자 하는 최종 목표는 RNN을 이용한 '언어 모델'을 구현하는 것입니다.

지금까지는 시계열 데이터를 한꺼번에 처리하는 RNN 계층을 구현했는데,

다른 계층도 더 구현해보겠습니다.

RNNLM(RNN Language Model)은 RNN을 사용한 언어 모델입니다.

 

 

① RNNLM

RNNLM의 전체 그림은 다음과 같습니다.

첫 번째(가장 아래) 층은 Embedding 계층으로 ID를 단어의 분산 표현(단어 벡터)으로 변환합니다.

그리고 그 분산 표현이 RNN 계층으로 입력됩니다.

RNN 계층은 은닉 상태를 다음 층(위쪽)으로는 Affine 계층과 softmax계층으로 출력하며,

같은 출력을 다음 시각의 RNN 계층 쪽으로 출력합니다.

 

자주 사용한 'I say goodbye you say hello' 예문을 RNNLM에 적용하면 다음과 같습니다.

softmax를 통해 나온 확률은 입력 다음에 나올 단어에 대한 확률들이 높은 것을 확인할 수 있습니다.

'say'부분에서는 'goodbye', 'hello' 둘 다 높게 나왔는데 둘 다 'you say' 다음으로 자연스럽게 나올 수 있습니다.

여기서 주목할 점은 RNN 계층이 'you say'라는 맥락을 기억하고 있다는 것입니다.

즉 과거의 정보를 응집된 은닉 상태 벡터로 저장해 두고 있습니다.

이처럼 RNNLM은 입력된 단어를 기억하고, 그것을 바탕으로 다음에 출현할 단어를 예측합니다.

이 일을 가능하게 하는 RNN 계층이며

과거에서 현재로 데이터를 계속 흘러 보내 줌으로써 과거의 정보를 인코딩해 기억할 수 있는 것입니다.

 

 

② Time 계층 구현

앞서 RNN 계층들을 한꺼번에 묶은 Time RNN처럼

Time Embedding, Time Affine 계층을 구현해보겠습니다.

 

Time Affine과 Time Embedding은

Affine 계층과 Embedding 계층을 T개 준비해서, 각 시각의 데이터를 개별적으로 처리하면 됩니다.

 

Time Affine 계층은 단순히 T개를 이용하는 방식 대신 한꺼번에 처리하는 효율적인 방법으로 했습니다.

class TimeAffine:
    def __init__(self, W, b):
        self.params = [W, b]
        self.grads = [np.zeros_like(W), np.zeros_like(b)]
        self.x = None

    def forward(self, x):
        N, T, D = x.shape
        W, b = self.params

        rx = x.reshape(N*T, -1)
        out = np.dot(rx, W) + b
        self.x = x
        return out.reshape(N, T, -1)

    def backward(self, dout):
        x = self.x
        N, T, D = x.shape
        W, b = self.params

        dout = dout.reshape(N*T, -1)
        rx = x.reshape(N*T, -1)

        db = np.sum(dout, axis=0)
        dW = np.dot(rx.T, dout)
        dx = np.dot(dout, W.T)
        dx = dx.reshape(*x.shape)

        self.grads[0][...] = dW
        self.grads[1][...] = db

        return dx

  

 

softmax 계층을 구현할 때 손실 오차를 구하는 Cross Entropy도 같이 하겠습니다.

x값들은 확률로 변환되기 전 점수고 t값들은 정답 레이블입니다.

최종 손실은 위와 같이 각 손실들의 평균입니다.

위의 최종 손실이라는 것은 한 블럭 당 최종 손실이지(미니 배치에 해당하는 손실의 평균)

전체 데이터에 대한 손실은 아닙니다.

그러므로 N개짜리 미니 배치라면, 그 N개의 손실을 더해 다시 N으로 나눠 데이터 1개당 평균 손실을 최종 손실로 구해 최종 출력을 내보냅니다.

 

코드는 다음과 같습니다.

class TimeSoftmaxWithLoss:
    def __init__(self):
        self.params, self.grads = [], []
        self.cache = None
        self.ignore_label = -1

    def forward(self, xs, ts):
        N, T, V = xs.shape

        if ts.ndim == 3:  # 정답 레이블이 원핫 벡터인 경우
            ts = ts.argmax(axis=2)

        mask = (ts != self.ignore_label)

        # 배치용과 시계열용을 정리(reshape)
        xs = xs.reshape(N * T, V)
        ts = ts.reshape(N * T)
        mask = mask.reshape(N * T)

        ys = softmax(xs)
        ls = np.log(ys[np.arange(N * T), ts])
        ls *= mask  # ignore_label에 해당하는 데이터는 손실을 0으로 설정
        loss = -np.sum(ls)
        loss /= mask.sum()

        self.cache = (ts, ys, mask, (N, T, V))
        return loss

    def backward(self, dout=1):
        ts, ys, mask, (N, T, V) = self.cache

        dx = ys
        dx[np.arange(N * T), ts] -= 1
        dx *= dout
        dx /= mask.sum()
        dx *= mask[:, np.newaxis]  # ignore_labelㅇㅔ 해당하는 데이터는 기울기를 0으로 설정

        dx = dx.reshape((N, T, V))

        return dx

 

 

 

(위의 내용은 밑바닥부터 시작하는 딥러닝 2 (한빛미디어)를 바탕으로 글을 작성하였습니다)