[keras, NLP] Seq2Seq로 번역 모델 구현하기

seq2seq 모델로 번역 모델을 만들어보자.

 

1. 데이터 전처리

seq2seq 모델을 번역 모델로 학습하기 위해서는 세 가지 종류의 데이터가 필요하다.

1. 인코더의 입력으로써 넣을 원 언어의 텍스트 데이터
2. 디코더의 입력으로써 넣을 타겟 언어의 텍스트 데이터
3. Teacher forcing을 위한 타겟 언어의 텍스트 데이터

학습에는 영어 텍스와 이에 대해 프랑스어로 변역된 텍스트의 모임인 코퍼스 데이터를 사용할 것이다.

Data download
http://www.manythings.org/anki/fra-eng.zip

이 데이터는 각 줄에 하나의 (영어, 프랑스어) 텍스트 쌍이 있고 이는 Tab으로 구분된다. 이를 다음과 같이 source 텍스트와 target 텍스트로 나눈다.

​x
with open('fra-eng/fra.txt', 'r', encoding='utf-8') as f:
        lines = f.read().split('\n')
​
input_texts = []
target_texts = []
target_text = ""
for line in lines[:3000]:
    input_text, target_text = line.split('\t')
    input_texts.append(input_text)
    # Each '\t' and '\n' is used as 'start' and 'end' sequence character
    target_text = '\t' + target_text + '\n'
    target_texts.append(target_text)

target 텍스트는 정답의 시작과 끝을 알리는 용도로 Tab 문자와 개행 문자를 사용했다.

이제, 학습 모델에 넣어줄 수 있도록 one-hot 벡터로 벡터화 해야한다.

각 데이터의 형태는 (텍스트 개수, 최대 문장 길이, 사용되는 토큰 수) 이 될 것이다. 여기서 사용되는 토큰 수 라는 건 텍스트를 이루는 문자 개수, 영어에서는 알파벳 수 정도로 생각하면 된다.

xxxxxxxxxx
latent_dim = 256
input_characters = set()
target_characters = set()
for input_text, target_text in zip(input_texts, target_texts):
    for ch in input_text:
        if ch not in input_characters:
            input_characters.add(ch)
    for ch in target_text:
        if ch not in target_characters:
            target_characters.add(ch)
​
input_characters = sorted(list(input_characters))
target_characters = sorted(list(target_characters))
num_encoder_tokens = len(input_characters)
num_decoder_tokens = len(target_characters)
max_encoder_seq_length = max([len(txt) for txt in input_texts])
max_decoder_seq_length = max([len(txt) for txt in target_texts])
​
input_token_index = { char: id for id, char in enumerate(input_characters)}
target_token_index = { char: id for id, char in enumerate(target_characters)}
​
encoder_input_data = np.zeros(shape=(len(input_texts), max_encoder_seq_length,                                  num_encoder_tokens), dtype='float32')
decoder_input_data = np.zeros(shape=(len(target_texts), max_decoder_seq_length,                                 num_decoder_tokens), dtype='float32')
decoder_target_data = np.zeros(shape=(len(target_texts), max_decoder_seq_length,                                num_decoder_tokens), dtype='float32')
​
for i, (input_text, target_text) in enumerate(zip(input_texts, target_texts)):
    for j, ch in enumerate(input_text):
        encoder_input_data[i, j, input_token_index[ch]] = 1.
    for j, ch in enumerate(target_text):
        decoder_input_data[i, j, target_token_index[ch]] = 1.
        if j > 0:
            decoder_target_data[i, j-1, target_token_index[ch]] = 1.

 

2. 모델 구축

학습 시 데이터의 주입 형태는 위 그림과 동일하다. 내부적으로는 인코더와 디코더로 이루어져 있고 각 모듈은 LSTM을 사용했다. 디코더는 인코더의 최종 hidden state와 cell state 값을 받아 초기 상태를 초기화하고 teacher forcing 과정을 진행하며 학습한다.

xxxxxxxxxx
from keras.models import Model, load_model
from keras.layers import Input, LSTM, Dense, BatchNormalization
​
# a part of encoder
encoder_inputs = Input(shape=(None, num_encoder_tokens), name='encoder_input')
encoder = LSTM(latent_dim, return_sequences=True, return_state=True, name='encoder')
encoder_outputs, state_h, state_c = encoder(encoder_inputs)
encoder_states = [state_h, state_c]
​
# a part of decoder
decoder_inputs = Input(shape=(None, num_decoder_tokens), name='decoder_input')
decoder_lstm = LSTM(latent_dim, return_sequences=True, return_state=True, name='decoder')
decoder_outputs, _, _ = decoder_lstm(decoder_inputs, initial_state=encoder_states)
batchNorm = BatchNormalization()
decoder_dense = Dense(num_decoder_tokens, activation='softmax')
decoder_outputs = decoder_dense(batchNorm(decoder_outputs))
​
# a model to train
model = Model([encoder_inputs, decoder_inputs], decoder_outputs)

학습이 조금 더 빠르게 되기 위해 Dense 레이어 전에 Batch normalization 레이어를 추가했다.

 

3. 학습하기

one-hot 벡터를 사용하여 다음에 어떤 글자가 나타날지 예측하는 것이기 때문에 손실 함수는 당연히 categorical cross entropy를 사용한다.

xxxxxxxxxx
batch_size=64
epochs=100
optimizer='rmsprop'
loss='categorical_crossentropy'
load_model_path=None, save_model_path='s2s.h5'
​
if not load_model_path == None:
    load_model(load_model_path)
model.compile(optimizer, loss)
model.fit([encoder_input_data, decoder_input_data], decoder_target_data, 
          batch_size = batch_size,
          epochs=epochs,
          validation_split=0.2)
model.save(save_model_path)

validation_split 인수는 총 데이터의 몇 퍼센트를 검증 데이터셋으로 사용할지를 결정한다. 만약 0.2로 값을 설정한다면, 총 데이터의 20%를 검증 데이터셋으로 사용한다. 이 검증 데이터셋은 학습 동안에 셔플되지 않는다.

 

4. 모델 사용

만들고자 하는 것은 번역 모델이므로, 이 모델에 영어 텍스트를 주입하여 프랑스어로 번역된 텍스트를 얻을 수 있어야 한다. 따라서 타겟 데이터를 하나씩 만들어서 다음 문자들을 하나씩 예측하는 형태로 번역된 텍스트를 얻는 방법을 사용하여 번역 모델을 완성시켜보겠다.

xxxxxxxxxx
# encoder model to decode
encoder_model = Model(encoder_inputs, encoder_states)
​
# decoder model to decode
decoder_state_input_h = Input(shape=(self.latent_dim,))
decoder_state_input_c = Input(shape=(self.latent_dim,))
decoder_states_inputs = [decoder_state_input_h, decoder_state_input_c]
decoder_outputs, state_h, state_c = decoder_lstm(
decoder_inputs, initial_state=decoder_states_inputs)
decoder_states = [state_h, state_c]
decoder_outputs = decoder_dense(decoder_outputs)
decoder_model = Model(
    [decoder_inputs] + decoder_states_inputs,
    [decoder_outputs] + decoder_states)

인코더 모델과 디코더 모델을 따로 구분하여 다시 정의하는 이유는 인코더로부터 얻은 상태를 decoder의 초기 상태로 한 다음, 처음 시작 문자\t로 시작하여 디코더가 예측한 문자로 타겟 입력을 계속 바꾸어 다음 문자를 예측할 수 있게 하기 위함이다. 이 때 종료 문자 \n이 예측되면 번역이 완료된 것이다.

xxxxxxxxxx
def decode_sequence(self, input_seq):
    # Encode the input as state vectors. [state_h, state_c]
    states_value = self.encoder_model.predict(input_seq)
​
    # Generate empty target sequence of length 1.
    target_seq = np.zeros((1, 1, self.num_decoder_tokens))
    # Populate the first character of target sequence with the start character.
    target_seq[0, 0, self.target_token_index['\t']] = 1.
​
    # 점점 디코드된 문자열을 추가해나감.
    stop_condition = False
    decoded_sentence = ''
    while not stop_condition:
        output_tokens, h, c = self.decoder_model.predict([target_seq] + states_value)
​
         # Sample a token
         sampled_token_index = np.argmax(output_tokens[0, -1, :])
         sampled_char = self.reverse_target_char_index[sampled_token_index]
         decoded_sentence += sampled_char
​
        # Exit condition: either hit max length
         # or find stop character.
         if (sampled_char == '\n' or
            len(decoded_sentence) > self.max_decoder_seq_length):
                stop_condition = True
​
        # Update the target sequence (of length 1).
         target_seq = np.zeros((1, 1, self.num_decoder_tokens))
         target_seq[0, 0, sampled_token_index] = 1.
​
        # Update states
         states_value = [h, c]
         
    return decoded_sentence

 

xxxxxxxxxx
def translate(self, input_seq):
    reverse_target_char_index = dict((i, char) for char, i in self.target_token_index.items())
    # 입력을 모델에 주입할 수 있도록 벡터화한다.
    input_vec = np.zeros(shape=(len(input_seq), self.max_encoder_seq_length,                                        num_encoder_tokens), dtype='float32')
    for i, txt in enumerate(input_seq):
        for j, ch in enumerate(txt):
            input_vec[i, j, self.input_token_index[ch]] = 1.
​
    decoded_sequences = []
    for seq_idx in range(input_vec.shape[0]):
        decoded_sequences.append(decode_sequence(input_vec[seq_idx: seq_idx+1]))
​
    return decoded_sequences

여기서 decoded_sequence 함수의 입력 형태는 (1, 최대 문장 길이, 사용된 토큰 개수)를 맞춰주어야 한다.

 

모듈화된 전체 코드는 여기에서 확인할 수 있다.