[밑시딥2] RNN을 사용한 문장 생성

언어 모델을 사용한 문장 생성

RNN을 사용한 문장 생성의 순서

언어 모델은 지금까지 주어진 단어들에서 다음에 출현하는 단어의 확률 분포를 출력한다.

이를 기초로 새로운 단어를 생성하는 방법은 아래의 두 방법을 생각할 수 있다.

• 확률이 가장 높은 단어를 선택하는 방법: 결정적인 방법, 결과가 항상 일정
• 확률적으로 선택하는 방법: 단어 확률에 맞게 선택하며, 확률에 따라 선택 가능성이 달라지지만 항상 같은 결과가 나오지 않는다.

여기서 결정적이란 알고리즘의 결과가 항상 일정함을 의미한다.

 

언어모델을 통해 확률 분포를 생성하면, 이를 기반으로 단어를 선택하고, 이를 반복한다.

이 작업을 원하는 만큼 수행하거나 <eos>와 같은 종결 기호가 나타날 때까지 반복하면 된다.

 

여기서 주목할 점은 이를 통해 생성한 모델은 훈련 데이터에는 존재하지 않는, 새로운 문장이라는 점이다.

문장 생성 구현

class RnnlmGen(Rnnlm):
    def generate(self, start_id, skip_ids=None, sample_size=100):
        word_ids = [start_id]

        x = start_id
        while len(word_ids) < sample_size:
            x = np.array(x).reshape(1, 1)
            score = self.predict(x)
            p = softmax(score.flatten())

            sampled = np.random.choice(len(p), size=1, p=p)
            if (skip_ids is None) or (sampled not in skip_ids):
                x = sampled
                word_ids.append(int(x))

        return word_ids

    def get_state(self):
        return self.lstm_layer.h, self.lstm_layer.c

    def set_state(self, state):
        self.lstm_layer.set_state(*state)

start_id는 시작 단어의 id, sample_size는 샘플링하는 단어 수이다.

skip_ids는 생성하지 않는 단어의 id 리스트이다.

더 좋은 문장으로

class BetterRnnlmGen(BetterRnnlm):
    def generate(self, start_id, skip_ids=None, sample_size=100):
        word_ids = [start_id]

        x = start_id
        while len(word_ids) < sample_size:
            x = np.array(x).reshape(1, 1)
            score = self.predict(x).flatten()
            p = softmax(score).flatten()

            sampled = np.random.choice(len(p), size=1, p=p)
            if (skip_ids is None) or (sampled not in skip_ids):
                x = sampled
                word_ids.append(int(x))

        return word_ids

    def get_state(self):
        states = []
        for layer in self.lstm_layers:
            states.append((layer.h, layer.c))
        return states

    def set_state(self, states):
        for layer, state in zip(self.lstm_layers, states):
            layer.set_state(*state)

BetterRnnlm 이용

seq2seq

seq2seq의 원리

seq2seq를 Encoder-Decoer 모델이라고 한다.

문장이 주어졌을 때, Encoder가 인코딩하며, 이 정보에는 번역에 필요한 정보가 조밀하게 응축되어 있다.

Decoder는 이 조밀하게 응축된 정보를 활용하여 새로운 도착어 문장을 생성한다.

 

Encoder의 역할은 시계열 데이터를 h라는 은닉 상태 백터로 변환하는 역할이다.

이 마짐가 은닉 상태 h에 입력 문장을 번역하는 데 필요한 정보가 인코딩된다.

여기서 주목할만한 점은 이 h가 고정 길이 벡터라는 사실이다.

즉, Encoder는 임의의 길이의 문장을 고정 길이 벡터로 변환하는 작업을 수행한다고 볼 수 있다.

 

Decoder는 앞 절의 신경망과 완전히 같은 구성이지만, 지금까지와 다르게 h를 입력받는다.

사실 완전히 입력받는 것이 없다고 보기보다는 영행렬을 지금까지 입력받아왔다고 볼 수 도 있다.

 

seq2seq는 이렇게 LSTM 2개로 구성되며, lstm의 은닉 상태가 encoder와 decoder를 서로 이어준다.

seq2seq의 전체 구성

시계열 데이터 변환용 장난감 문제

이 책에서는 토이프로젝트로 덧샘 문제를 해결한다.

문자로 작성된 간단한 덧샘 수식을 입력하여 이 문자들의 패턴을 통해 올바른 덧샘 결과가 도출되도록 하는 것이 목표이다.

가변 길이 시계열 데이터

사실 덧샘 문제는 전산학 기초에서 배웠던 전가산기 반가산기 같은 기계를 통해 해결할 수 있지만, 이번 과정에서의 의의는 문제마다 문자 수가 다르다는 것이다.

하지만 이를 미니배치 처리하려면 추가적인 노력이 필요하다.

 

이런한 문제를 해결하기 위한 기법으로 가장 단순한 방법은 패딩이다.

원래의 데이터에 의미 ㅇ벗는 데이터를 채워 모든 데이터의 길이를 균일하게 맞추는 기법이다.

 

문제는 원래 존재하지 않던 패딩용 문자를 함께 처리한다는 것인데, 정확성이 중요하다면, 이 패딩을 위한 전용 처리가 필요하다.

일반적으로 마스크라는 테크닉을 사용하여 해결한다.

이는 LSTM이 이전 시각의 입력을 그대로 출력하게 하며, 마치 처음부터 패딩이 없던 것처럼 인코딩할 수 있다.

seq2seq 구현

Encoder 클래스

Encoder의 입출력

Encoder 클래스는 임베딩 레이어와 LSTM 레이어로 구성되며, 입베딩 레이어에서는 문자를 문자 벡터로 변환하고, LSTM 레이어 그 벡터가 전달된다.

class Encoder:
    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')
        lstm_Wx = (rn(D, 4 * H) / np.sqrt(D)).astype('f')
        lstm_Wh = (rn(H, 4 * H) / np.sqrt(H)).astype('f')
        lstm_b = np.zeros(4 * H).astype('f')

        self.embed = TimeEmbedding(embed_W)
        self.lstm = TimeLSTM(lstm_Wx, lstm_Wh, lstm_b, stateful=False)

        self.params = self.embed.params + self.lstm.params
        self.grads = self.embed.grads + self.lstm.grads
        self.hs = None

    def forward(self, xs):
        xs = self.embed.forward(xs)
        hs = self.lstm.forward(xs)
        self.hs = hs
        return hs[:, -1, :]

    def backward(self, dh):
        dhs = np.zeros_like(self.hs)
        dhs[:, -1, :] = dh

        dout = self.lstm.backward(dhs)
        dout = self.embed.backward(dout)
        return dout

vocab_size는 어휘 수이며, wordvec_size는 문자 벡터의 차원 수, 그리고 hidden_size는 lstm 레이어의 은닉 상태 벡터의 차원수를 의미한다.

순전파 과정에서는 TimeLSTM 레이어의 마지막 은닉 상태만, 추출해 출력으로 전달한다.

Decoder 클래스

앞서 단어 생성에 대해서는 확률적으로 생성하는 것을 언급했지만, 여기서는 결정적인 답을 생성하도록 할 것이다.

Decoder 클래스 구조

Time Embedding, Time LSTM, Time Affine의 총 3가지 레이어로 구성된다.

여기서 Encoder와 다르게 Affine 레이어와 argmax가 추가로 사용되는데, 이는 점수가 가장 높은 결과를 추록하도록 하기 위해 추가된 노드이다.

class Decoder:
    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')
        lstm_Wx = (rn(D, 4 * H) / np.sqrt(D)).astype('f')
        lstm_Wh = (rn(H, 4 * H) / np.sqrt(H)).astype('f')
        lstm_b = np.zeros(4 * H).astype('f')
        affine_W = (rn(H, V) / np.sqrt(H)).astype('f')
        affine_b = np.zeros(V).astype('f')

        self.embed = TimeEmbedding(embed_W)
        self.lstm = TimeLSTM(lstm_Wx, lstm_Wh, lstm_b, stateful=True)
        self.affine = TimeAffine(affine_W, affine_b)

        self.params, self.grads = [], []
        for layer in (self.embed, self.lstm, self.affine):
            self.params += layer.params
            self.grads += layer.grads

    def forward(self, xs, h):
        self.lstm.set_state(h)

        out = self.embed.forward(xs)
        out = self.lstm.forward(out)
        score = self.affine.forward(out)
        return score

    def backward(self, dscore):
        dout = self.affine.backward(dscore)
        dout = self.lstm.backward(dout)
        dout = self.embed.backward(dout)
        dh = self.lstm.dh
        return dh

    def generate(self, h, start_id, sample_size):
        sampled = []
        sample_id = start_id
        self.lstm.set_state(h)

        for _ in range(sample_size):
            x = np.array(sample_id).reshape((1, 1))
            out = self.embed.forward(x)
            out = self.lstm.forward(out)
            score = self.affine.forward(out)

            sample_id = np.argmax(score.flatten())
            sampled.append(int(sample_id))

        return sampled

Decoder는 생성하는 역할도 수행하기 때문에 generate라는 함수가 추가되었다.

seq2seq 클래스

class Seq2seq(BaseModel):
    def __init__(self, vocab_size, wordvec_size, hidden_size):
        V, D, H = vocab_size, wordvec_size, hidden_size
        self.encoder = Encoder(V, D, H)
        self.decoder = Decoder(V, D, H)
        self.softmax = TimeSoftmaxWithLoss()

        self.params = self.encoder.params + self.decoder.params
        self.grads = self.encoder.grads + self.decoder.grads

    def forward(self, xs, ts):
        decoder_xs, decoder_ts = ts[:, :-1], ts[:, 1:]

        h = self.encoder.forward(xs)
        score = self.decoder.forward(decoder_xs, h)
        loss = self.softmax.forward(score, decoder_ts)
        return loss

    def backward(self, dout=1):
        dout = self.softmax.backward(dout)
        dh = self.decoder.backward(dout)
        dout = self.encoder.backward(dh)
        return dout

    def generate(self, xs, start_id, sample_size):
        h = self.encoder.forward(xs)
        sampled = self.decoder.generate(h, start_id, sample_size)
        return sampled

seq2seq 평가

# 데이터셋 읽기
(x_train, t_train), (x_test, t_test) = sequence.load_data('addition.txt')
char_to_id, id_to_char = sequence.get_vocab()

# 입력 반전 여부 설정 =============================================
is_reverse = False  # True
if is_reverse:
    x_train, x_test = x_train[:, ::-1], x_test[:, ::-1]
# ================================================================

# 하이퍼파라미터 설정
vocab_size = len(char_to_id)
wordvec_size = 16
hidden_size = 128
batch_size = 128
max_epoch = 25
max_grad = 5.0

# 일반 혹은 엿보기(Peeky) 설정 =====================================
model = Seq2seq(vocab_size, wordvec_size, hidden_size)
# model = PeekySeq2seq(vocab_size, wordvec_size, hidden_size)
# ================================================================
optimizer = Adam()
trainer = Trainer(model, optimizer)

acc_list = []
for epoch in range(max_epoch):
    trainer.fit(x_train, t_train, max_epoch=1,
                batch_size=batch_size, max_grad=max_grad)

    correct_num = 0
    for i in range(len(x_test)):
        question, correct = x_test[[i]], t_test[[i]]
        verbose = i < 10
        correct_num += eval_seq2seq(model, question, correct,
                                    id_to_char, verbose, is_reverse)

    acc = float(correct_num) / len(x_test)
    acc_list.append(acc)
    print('검증 정확도 %.3f%%' % (acc * 100))

여기서 평가 척도로 정확도가 사용되었다.

이를 통해 문제를 정확하게 풀었는지 평가한다.

seq2seq 개선

입력 데이터 반전

입력 데이터 반전

입력 데이터를 반전시키는 트릭은 seq2seq 논문에서 제안되었다.

이렇게 진행하면, 학습 진행이 빨라져서 최종적으로 정확도가 좋아진다고 한다.

반전으로 인해 단어와 가까우므로 기울기가 더 잘 전해져서 학습 효율이 더 좋아진다.

다만, 평군적인 거리는 그대로다.

 

나는 이에 대해 문자열 생성 트리를 빔서치와 같은 방법으로 구성할 때, 터미널노드 (잎파리)로부터 루트 노드 (뿌리)로 내려가는 과정으로 학습하면서, 반대의 과정을 생성할 때, 더 정확하게 추론할 수 있다고 생각한다.

엿보기 (Peeky)

이전에 구현한 seq2seq는 LSTM 레이어속 가장 처음의 LSTM cell 만이 처음에 들어온 이 은닉 상태 h를 사용했습니다.

여기서는 이를 다른 Cell들도 사용할 수 있도록 하여 정확도를 높인다.

모든 시각의 Affine 레이어와 LSTM Cell에 이를 전달하여 서로 공유한다.

이는 마치 집단 지성에 비유할 수 있다.

class PeekyDecoder:
    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')
        lstm_Wx = (rn(H + D, 4 * H) / np.sqrt(H + D)).astype('f')
        lstm_Wh = (rn(H, 4 * H) / np.sqrt(H)).astype('f')
        lstm_b = np.zeros(4 * H).astype('f')
        affine_W = (rn(H + H, V) / np.sqrt(H + H)).astype('f')
        affine_b = np.zeros(V).astype('f')

        self.embed = TimeEmbedding(embed_W)
        self.lstm = TimeLSTM(lstm_Wx, lstm_Wh, lstm_b, stateful=True)
        self.affine = TimeAffine(affine_W, affine_b)

        self.params, self.grads = [], []
        for layer in (self.embed, self.lstm, self.affine):
            self.params += layer.params
            self.grads += layer.grads
        self.cache = None

    def forward(self, xs, h):
        N, T = xs.shape
        N, H = h.shape

        self.lstm.set_state(h)

        out = self.embed.forward(xs)
        hs = np.repeat(h, T, axis=0).reshape(N, T, H)
        out = np.concatenate((hs, out), axis=2)

        out = self.lstm.forward(out)
        out = np.concatenate((hs, out), axis=2)

        score = self.affine.forward(out)
        self.cache = H
        return score

    def backward(self, dscore):
        H = self.cache

        dout = self.affine.backward(dscore)
        dout, dhs0 = dout[:, :, H:], dout[:, :, :H]
        dout = self.lstm.backward(dout)
        dembed, dhs1 = dout[:, :, H:], dout[:, :, :H]
        self.embed.backward(dembed)

        dhs = dhs0 + dhs1
        dh = self.lstm.dh + np.sum(dhs, axis=1)
        return dh

    def generate(self, h, start_id, sample_size):
        sampled = []
        char_id = start_id
        self.lstm.set_state(h)

        H = h.shape[1]
        peeky_h = h.reshape(1, 1, H)
        for _ in range(sample_size):
            x = np.array([char_id]).reshape((1, 1))
            out = self.embed.forward(x)

            out = np.concatenate((peeky_h, out), axis=2)
            out = self.lstm.forward(out)
            out = np.concatenate((peeky_h, out), axis=2)
            score = self.affine.forward(out)

            char_id = np.argmax(score.flatten())
            sampled.append(char_id)

        return sampled


class PeekySeq2seq(Seq2seq):
    def __init__(self, vocab_size, wordvec_size, hidden_size):
        V, D, H = vocab_size, wordvec_size, hidden_size
        self.encoder = Encoder(V, D, H)
        self.decoder = PeekyDecoder(V, D, H)
        self.softmax = TimeSoftmaxWithLoss()

        self.params = self.encoder.params + self.decoder.params
        self.grads = self.encoder.grads + self.decoder.grads

레퍼런스

밑바닥부터 시작하는 딥러닝 2 (사이토 고키 지음, 개앞맵시 옮김)

개앞맵시님의 밑시딥2 저장소: WegraLee/deep-learning-from-scratch-2: 『밑바닥부터 시작하는 딥러닝 ❷』(한빛미디어, 2019) (github.com)