RNN을 이용한 디코더-인코더 - Seq2Seq

시퀀스 투 시퀀스(sequence to sequence)

* seq2seq model은 입력된 시퀀스로부터 다른 도메인의 시퀀스를 출력하는 데 주로 쓰인다. 예를 들어 한글을 입력하면(domain : 한글)

영어를 출력하는(domiani : 영어) 모델이나, 사용자가 입력을 집어넣으면(domain : 질문) 적절한 대답을 하는(domain : 대답) 챗봇 등이 될 수 있다. 이번 포스팅에서는 seq2seq model에 대해 살펴보되 번역기의 예를 들어 살펴보겠다(보통 가장 많이 사용하는 분야가 번역이기도 하다).

 

대략적인 구성

 

* seq2seq model의 인코더 및 디코더는 각각의 RNN모델로 구성되어 있다. 그렇다고 vanilla RNN을 그냥 사용하는 것은 아니고 성능 문제로 인해서 보통 LSTM이나 GRU를 사용하게 된다. RNN의 인코더의 각 timestep에는 출력을 output으로 내보내지 않는다. 다만, hidden state를 다음 timestep에 건내주기만 할 뿐이다. 이후 마지막 timestep의 hidden state는 context vector라는 이름으로 불리며 context vector는 디코더에 전달돼 첫 번째 timestep의 hidden state로 사용된다. 디코더는 기본적으로 RNNLM의 작동방식을 따르기 때문에 앞서 포스팅한 RNNLM 포스팅을 참고하면 앞으로 설명할 내용에 대해 이해하기 쉬울 것이다.

 

https://soki.tistory.com/44?category=1065364 

RNN 언어 모델(RNNLM)

 

훈련 과정과 테스트 과정의 차이

* 훈련 과정에서는 RNNLM과 같이 교사강요(Teacher forcing)를 사용한다. 즉, 인코더가 전달한 context vector와 디코더의 input을 입력 받았을 때 디코더는 자신이 현재 time step에서 출력하는 예측이 아닌 현재 time step의 정답 label을 다음 time step의 입력으로 사용하게 된다. 이는 간단히 말하면 예측이 한 번 어긋났을 때 그 뒤의 예측이 산으로 가지 않게 하기 위함이라고 설명할 수 있다. 예를 들면 정답 문장이 I ate pizza at lunch 라고 할 때 I 다음에 sleep이라고 예측해버리면 점심에 피자를 먹음(pizza at lunch)이라는 예측과는 전혀 다른 예측으로 이어질 것이기 때문에 이러한 상황을 방지하고자 정답 label을 사용하여 훈련을 실시하는 것이다. 때문에 

* 반면 테스트시에는 정 반대의 과정을 거친다. 정답 label을 사용하지 않고 순수하게 디코더의 예측을 다음 time step의 입력으로 사용한다.

* 디코더의 구조는 LSTM layer의 output vector를 Dense layer를 통과시켜 softmax를 통해 0~1사이의 확률 값으로 이뤄진 vector로 변환하고 가장 확률이 높은 index의 단어를 output으로 출력하는 형태를 띠고 있다. 이는 RNNLM에서 배운 내용과 동일하다.

 

2. 문자 레벨 기계 번역기(Character - Level Neural Machine Translation) 구현하기

 

1) 데이터셋 준비

import os
import shutil
import zipfile

import pandas as pd
import tensorflow as tf
import urllib3
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.utils import to_categorical

http=urllib3.PoolManager()
url='http://www.manythings.org/anki/fra-eng.zip'
filename='fra-eng.zip'
path=os.getcwd()
zipfilename=os.path.join(path,filename)
with http.request('GET', url, preload_content=False) as r, open(zipfilename, 'wb') as out_file:
  shutil.copyfileobj(r, out_file)

with zipfile.ZipFile(zipfilename, 'r') as zip_ref:
  zip_ref.extractall(path)
  
lines=pd.read_csv('fra.txt', names=['src', 'tar', 'lic'], sep='\t')
del lines['lic']
print('전체 샘플의 개수: ', len(lines))

# 전체 샘플의 개수:  191954

* 인코더-디코더 번역기의 입력 시퀀스와 출력 시퀀스의 길이는 같지 않다. 예를 들어 입력 시퀀스가 '나는 학생이다'로 3단어 일 때 출력 시퀀스가 'I am a student' 4단어 일 수도 있는 것이다.

* 우리가 사용할 데이터셋은 프랑스-영어 corpus인 fra-eng.zip 파일이다.

 

lines=lines.loc[:, 'src' : 'tar']
lines=lines[0:60000]
lines.sample(10)

* 총 17만 7천개의 병렬 문장 샘플 중에서 우리는 60,000개의 샘플만 가지고 사용할 것이다.

 

lines.tar=lines.tar.apply(lambda x : '\t ' + x + ' \n')
lines.sample(10)

* 출력의 정답 label인 프랑스어로 구성된 tar는 문장 시작인 <sos> token과 <eos> token을 포함해야 한다. 이를 위해 <sos> token으로 '\t'를 <eos> token으로 '\n'을 사용하였다.

src_set=set()
tar_set=set()

for line in lines.src:
  for char in line:
    src_set.add(char)

for line in lines.tar:
  for char in line:
    tar_set.add(char)

src_vocab_size=len(src_set)+1
tar_vocab_size=len(tar_set)+1

print('source vocab size: {}'.format(src_vocab_size))
print('target vocab size: {}'.format(tar_vocab_size))

# source vocab size: 79
# target vocab size: 105

* set을 구성해 line들의 character들을 하나씩 추가해서 vocabulary size를 생성하였다.

 

src_set=sorted(list(src_set))
tar_set=sorted(list(tar_set))
src_to_index=dict([(word, i+1) for i, word in enumerate(src_set)])
tar_to_index=dict([(word, i+1) for i, word in enumerate(tar_set)])
print(src_to_index)
print(tar_to_index)

# {' ': 1, '!': 2, '"': 3, '$': 4, '%': 5, '&': 6, "'": 7, ',': 8, '-': 9, '.': 10, '/': 11, '0': 12, '1': 13, '2': 14, '3': 15, '4': 16, '5': 17, '6': 18, '7': 19, '8': 20, '9': 21, ':': 22, '?': 23, 'A': 24, 'B': 25, 'C': 26, 'D': 27, 'E': 28, 'F': 29, 'G': 30, 'H': 31, 'I': 32, 'J': 33, 'K': 34, 'L': 35, 'M': 36, 'N': 37, 'O': 38, 'P': 39, 'Q': 40, 'R': 41, 'S': 42, 'T': 43, 'U': 44, 'V': 45, 'W': 46, 'X': 47, 'Y': 48, 'Z': 49, 'a': 50, 'b': 51, 'c': 52, 'd': 53, 'e': 54, 'f': 55, 'g': 56, 'h': 57, 'i': 58, 'j': 59, 'k': 60, 'l': 61, 'm': 62, 'n': 63, 'o': 64, 'p': 65, 'q': 66, 'r': 67, 's': 68, 't': 69, 'u': 70, 'v': 71, 'w': 72, 'x': 73, 'y': 74, 'z': 75, 'é': 76, '’': 77, '€': 78}
# {'\t': 1, '\n': 2, ' ': 3, '!': 4, '"': 5, '$': 6, '%': 7, '&': 8, "'": 9, '(': 10, ')': 11, ',': 12, '-': 13, '.': 14, '0': 15, '1': 16, '2': 17, '3': 18, '4': 19, '5': 20, '6': 21, '7': 22, '8': 23, '9': 24, ':': 25, '?': 26, 'A': 27, 'B': 28, 'C': 29, 'D': 30, 'E': 31, 'F': 32, 'G': 33, 'H': 34, 'I': 35, 'J': 36, 'K': 37, 'L': 38, 'M': 39, 'N': 40, 'O': 41, 'P': 42, 'Q': 43, 'R': 44, 'S': 45, 'T': 46, 'U': 47, 'V': 48, 'W': 49, 'X': 50, 'Y': 51, 'Z': 52, 'a': 53, 'b': 54, 'c': 55, 'd': 56, 'e': 57, 'f': 58, 'g': 59, 'h': 60, 'i': 61, 'j': 62, 'k': 63, 'l': 64, 'm': 65, 'n': 66, 'o': 67, 'p': 68, 'q': 69, 'r': 70, 's': 71, 't': 72, 'u': 73, 'v': 74, 'w': 75, 'x': 76, 'y': 77, 'z': 78, '\xa0': 79, '«': 80, '»': 81, 'À': 82, 'Ç': 83, 'É': 84, 'Ê': 85, 'Ô': 86, 'à': 87, 'â': 88, 'ç': 89, 'è': 90, 'é': 91, 'ê': 92, 'ë': 93, 'î': 94, 'ï': 95, 'ô': 96, 'ù': 97, 'û': 98, 'œ': 99, '\u2009': 100, '\u200b': 101, '‘': 102, '’': 103, '\u202f': 104}

* source set과 target set을 dictionary 형식으로 구성해 인덱스를 부여하였다.

 

encoder_input=[]
decoder_input=[]

for line in lines.src:
  tmp=[]
  for char in line:
    if char in src_to_index:
      tmp.append(src_to_index[char])
  encoder_input.append(tmp)

for line in lines.tar:
  tmp=[]
  for char in line:
    if char in tar_to_index:
      tmp.append(tar_to_index[char])
  decoder_input.append(tmp)
  
print('encoder input: {}'.format(encoder_input[:5]))
print('decoder input: {}'.format(decoder_input[:5]))

# encoder input: [[30, 64, 10], [30, 64, 10], [30, 64, 10], [31, 58, 10], [31, 58, 10]]
# decoder input: [[1, 3, 48, 53, 3, 4, 3, 2], [1, 3, 39, 53, 70, 55, 60, 57, 14, 3, 2], [1, 3, 28, 67, 73, 59, 57, 3, 4, 3, 2], [1, 3, 45, 53, 64, 73, 72, 3, 4, 3, 2], [1, 3, 45, 53, 64, 73, 72, 14, 3, 2]]

* encoder input과 decoder input에 문장들을 index로 encoding하여 저장하였다.

 

decoder_target=[]

for input in decoder_input:
  input.remove(1)
  decoder_target.append(input)

print('decoder target: {}'.format(decoder_target[:5]))

# decoder target: [[3, 48, 53, 3, 4, 3, 2], [3, 39, 53, 70, 55, 60, 57, 14, 3, 2], [3, 28, 67, 73, 59, 57, 3, 4, 3, 2], [3, 45, 53, 64, 73, 72, 3, 4, 3, 2], [3, 45, 53, 64, 73, 72, 14, 3, 2]]

* 디코더의 target, 즉 정답 label은 <sos>토큰이 존재하지 않아야 한다. 이는 decoder의 test과정을 생각해보면 쉽게 유추할 수 있는데, test과정에서 첫 번째 timestep에 입력되는 값이 <sos> 토큰이고, 첫 번째 timestep에 출력되는 target은 <sos>토큰 다음에 이어질 단어이기 때문이다. 그림으로 이해하기 쉽도록 앞서 살펴봤던 디코더 구조 자료를 확인해보자.

 

max_src_len=max([len(l) for l in encoder_input])
max_tar_len=max([len(l) for l in decoder_input])

# padding
encoder_input=pad_sequences(encoder_input, maxlen=max_src_len, padding='post')
decoder_input=pad_sequences(decoder_input, maxlen=max_src_len, padding='post')
decoder_target=pad_sequences(decoder_target, maxlen=max_src_len, padding='post')

# one-hot encoding
encoder_input=to_categorical(encoder_input, num_classes=src_vocab_size)
decoder_input=to_categorical(decoder_input, num_classes=tar_vocab_size)
decoder_target=to_categorical(decoder_target, num_classes = tar_vocab_size)

print(np.shape(encoder_input))
print(np.shape(decoder_input))

# (60000, 23, 79)
# (60000, 76, 105)

* 가장 큰 문장의 길이를 구한 이후 padding 작업을 해주고, one-hot encoding까지 완료하였다.

* 이제 encoder, decoder input들은 vocabulary size 크기의 onehot vector로 이뤄진 문장(길이 : 최대 문장 길이만큼 padding) 60000개로 변한되었다.

 

2) 모델 생성

from tensorflow.keras.layers import Input, LSTM, Embedding, Dense
from tensorflow.keras.models import Model
import numpy as np

encoder_inputs=Input(shape=(None, src_vocab_size))
encoder_lstm=LSTM(units=256, return_state=True)
encoder_outputs, state_h, state_c = encoder_lstm(encoder_inputs)
encoder_state=[state_h, state_c]

decoder_inputs=Input(shape=(None, tar_vocab_size))
decoder_lstm=LSTM(units=256, return_sequences=True, return_state=True)
decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=encoder_state)
decoder_softmax=Dense(tar_vocab_size, activation='softmax')
decoder_outputs=decoder_softmax(decoder_outputs)

model=Model([encoder_inputs, decoder_inputs], decoder_outputs)
model.compile(optimizer='rmsprop', loss='categorical_crossentropy')
model.fit(x=[encoder_input, decoder_input], y=decoder_target, batch_size=64, epochs=40, validation_split=0.2)

# Epoch 1/40
# 750/750 [==============================] - 47s 47ms/step - loss: 0.7360 - val_loss: 0.6574
# Epoch 2/40
# 750/750 [==============================] - 31s 42ms/step - loss: 0.4544 - val_loss: 0.5357
# Epoch 3/40
# 750/750 [==============================] - 33s 43ms/step - loss: 0.3839 - val_loss: 0.4721
# Epoch 4/40
# 750/750 [==============================] - 32s 42ms/step - loss: 0.3399 - val_loss: 0.4331
# Epoch 5/40
# 750/750 [==============================] - 32s 42ms/step - loss: 0.3108 - val_loss: 0.4079
# ... 중략
# Epoch 38/40
# 750/750 [==============================] - 32s 43ms/step - loss: 0.1386 - val_loss: 0.3977
# Epoch 39/40
# 750/750 [==============================] - 32s 43ms/step - loss: 0.1370 - val_loss: 0.3985
# Epoch 40/40
# 750/750 [==============================] - 32s 43ms/step - loss: 0.1353 - val_loss: 0.4013
# <keras.callbacks.History at 0x7fa26abfb110>

* lstm cell을 사용하는 encoder, decoder를 구성하여 40epoch 훈련을 시행하였다.

 

3) 모델 테스트

encoder_model=Model(inputs=encoder_inputs, outputs=encoder_state) #앞서 구성한 encoder model 그대로 사용

decoder_state_input_h=Input(shape=(256,)) #shape은 units 크기의 vector
decoder_state_input_c=Input(shape=(256,))
decoder_state_inputs=[decoder_state_input_h, decoder_state_input_c]
decoder_outputs, state_h, state_c = decoder_lstm(decoder_inputs, initial_state=decoder_state_inputs) #훈련된 lstm layer 사용
decoder_states=[state_h, state_c]
decoder_outputs=decoder_softmax(decoder_outputs)
decoder_model=Model(inputs=[decoder_inputs] + decoder_state_inputs, outputs=[decoder_outputs] + decoder_states)

# index : source, index : target 형태의 dictionary를 구성
index_to_src=dict((i, char) for char, i in src_to_index.items())
index_to_tar=dict((i, char) for char, i in tar_to_index.items())

* 모델을 테스트하기 위해서는 훈련을 통해 데이터에 fitting된 encoder lstm layer, decoder lstm layer를 그대로 사용해야 한다.

* 단, 훈련에 사용한 구조를 그대로 사용할 수는 없다. decoder의 output에는 hidden state, cell state가 포함되어 있어야 하기 때문이다. 그 이유는 다음 코드를 보며 설명하겠다.

def decode_sequence(input_seq):
  states_value = encoder_model.predict(input_seq)
  
  target_seq=np.zeros((1,1, tar_vocab_size))
  target_seq[0,0,tar_to_index['\t']] = 1.

  stop_condition=False
  decoded_sentence=""

  while not stop_condition:
    output_tokens, h, c = decoder_model.predict([target_seq] + states_value)

    sampled_token_index=np.argmax(output_tokens[0, -1, :])
    sampled_char = index_to_tar[sampled_token_index]

    decoded_sentence += sampled_char

    if (sampled_char == '\n' or len(decoded_sentence) > max_tar_len): 
        stop_condition=True
    
    target_seq=np.zero((1, 1, tar_vocab_size))
    target_seq[0,0,sampled_token_index] = 1.

    states_value = [h, c]

  return decoded_sentence

* 우리는 반복문을 통해 훈련된 decoder에 반복적으로 자신의 예측 값을 입력하는 작업을 할 것이다. 이 test 과정에서 encoder는 맨 처음 input sequence, 즉 번역할 문장을 집어넣어 hidden state를 생성하는 용도 외에는 사용하지 않는다. input sequence를 통해 생성한 hidden state는 decoder로 전달되고, decoder에는 token이 입력되어 output과 hidden state, cell state가 return 된다. 다음 반복에서는 decoder에 이전 반복의 output, hidden state, cell state가 입력으로 주어진다. 이 과정을 '\n'을 만나거나, decode된 문장의 길이가 max_tar_len보다 커질 때 까지 반복할 것이다. 그렇기 때문에 위에서 설명했듯이 decoder의 output에 hidden state, cell state가 포함된 구조를 새로 만든 것이다.

 

for seq_index in [3, 50, 100, 300, 1001]:
  input_seq = encoder_input[seq_index : seq_index+1]
  decoded_sentence=decode_sequence(input_seq)
  print(35 * "-")
  print('입력 문장: ', lines.src[seq_index])
  print('정답 문장: ', lines.tar[seq_index][2:len(lines.tar[seq_index])-1]) #정답 문장에는 \t, \n이 들어있으면 안된다.
  print('번역 문장: ', decoded_sentence[1:len(decoded_sentence)-1]) # \t는 <sos> token으로 출력되지 않기 떄문에 \n만 제외하고 print한다.
  
# -----------------------------------
# 입력 문장:  Hi.
# 정답 문장:  Salut ! 
# 번역 문장:  Salut. 
# -----------------------------------
# 입력 문장:  I see.
# 정답 문장:  Aha. 
# 번역 문장:  Je les adorais. 
# -----------------------------------
# 입력 문장:  Hug me.
# 정답 문장:  Serrez-moi dans vos bras ! 
# 번역 문장:  Serrez-moi dans vos bras ! 
# -----------------------------------
# 입력 문장:  Help me.
# 정답 문장:  Aidez-moi. 
# 번역 문장:  Aide-moi. 
# -----------------------------------
# 입력 문장:  I beg you.
# 정답 문장:  Je vous en prie. 
# 번역 문장:  Je vous ai eu.

* random한 index의 문장을 encoder_input에서 선택해 5번 번역 test를 해 본 결과 완벽하진 않지만 어느 정도 예측이 들어맞는 것을 확인할 수 있었다.