[NLP] seq2seq로 번역기 구현하기 (feat.딥러닝을 이용한 자연어 처리 입문)

-- 본 포스팅은 파이토치로 배우는 자연어 처리 (한빛미디어) 와 딥러닝을 이용한 자연어 처리 입문(위키독스) 저서를 참고해서 작성된 글입니다.

 

1. Sequence-to-Sequence (seq2seq)

• 번역기에서 대표적으로 사용되는 모델
• 인코더와 디코더로 구성된 인코더 - 디코더 모델의 일종
• 조건부 생성 모델 (conditioned generation model)의 일종이기도 함
  • 조건부 생성 모델이란? 입력표현 대신 일반적인 조건 문맥을 활용하여 디코더가 출력을 만드는 모델

1.1 seq2seq 모델의 구성

https://wikidocs.net/24996

1. 인코더는 입력 문장의 모든 단어를 순차적으로 입력받은 뒤에 마지막에 모든 단어의 정보를 압축해서 하나의 벡터로 만들어줌 (즉, context vector를 생성)
2. context 벡터로 모두 문장의 정보가 압축되면 디코더로 전송
3. 디코더는 context 벡터를 받아서 번역된 단어를 하나씩 순차적으로 출력

https://wikidocs.net/24996

• 인코더와 디코더를 자세히 보면 이렇게 RNN 아키텍쳐로 이루어져 있다.
  • 성능의 문제로 RNN보다는 주로 발전된 형태인 LSTM과 GRU셀을 사용한다.
• 기계는 단어보다 숫자를 더 잘 인식하기 때문에 각 셀에서 임베딩 과정을 거친다.
• 하나의 RNN(LSTM)셀에서는 t-1 에서의 은닉상태와, t에서의 입력벡터를 입력으로 받아서 t에서의 은닉상태 벡터를 생성해준다. (이전 포스팅참고)

▶ 인코더

• 입력 문장은 단어 토큰화를 통해 단어 단위로 쪼개지고, 모든 단어를 임베딩한다.
• 그리고 단어 토큰은 각각 RNN셀의 각 시점의 입력이 된다.
• RNN 각 셀의 마지막 시점의 은닉상태를 context 벡터로 만든 후, 디코더로 넘겨준다.

▶ 디코더

• 초기 입력으로 문장의 시작을 의미하는 <sos>가 입력됨
• 디코더는 <sos>가 입력되면 다음에 등장할 확률이 높은 단어를 예측
• 여기서는 첫번째 시점에 나올 단어를 Je로 예측
• 이렇게 디코더는 기본적으로 다음에 올 단어를 예측하고, 그 예측한 단어를 다음 시점의 RNN셀의 입력으로 넣는 행위를 반복
• 문장이 끝났다는 심볼인 <eos>가 예측될때까지 반복된다.

 

2. seq2seq로 기계 번역기 구현하기 (참고)

본 코드는 https://wikidocs.net/24996 해당 링크를 참고하여 작성했다. 우선은 글자 차원에서의 번역기를 구현할 것이다. (토큰의 단위가 단어가 아니라 글자(알파벳) 이라는 의미이다!)

2.1 데이터셋

▶ 데이터 출처

기계 번역을 훈련시키기 위해서 훈련 데이터로 병렬 코퍼스가 필요하다. http://www.manythings.org/anki 해당 링크에서 다운받은 프랑스-영어 병렬 코퍼스인 fran-eng.zip 이라는 파일을 사용할 것이다. 해당 파일의 압축을 풀고, fra.txt 라는 파일을 사용해서 실습해보자.

▶ 병렬 코퍼스

병렬 코퍼스는 '태깅'작업과는 살짝 다르다. 태깅 작업은 모든 데이터의 쌍의 길이가 같다는 특징이 있지만 병렬 데이터는 그렇지 않다. 예를들어 '나는 학생이다' 두개의 토큰으로 이루어진 문장은 'I am a student' 4개의 토큰으로 구성된 문장으로 번역된다. 이처럼 seq2seq는 입력 시퀀스와 출력 시퀀스의 길이가 다를 수 있다고 가정하고 실습을 진행한다.

▶ fra.txt의 구성

Watch me.           Regardez-moi !

이렇게 왼쪽의 영어 문장과 오른쪽의 프랑스어 문장 사이에 탭으로 구별되는 구조가 하나의 샘플이다.
이와 같은 데이터가 16만개의 병렬 문장 샘플을 포함하고 있다.

 

2.2 import

import pandas as pd
import urllib3
import zipfile
import shutil
import os
from tensorflow.keras.preprocessing.sequence import pad_sequences
from tensorflow.keras.utils import to_categorical

 

2.3 데이터 불러오기

from google.colab import drive
drive.mount('/content/gdrive/')
PATH = '/content/gdrive/MyDrive/Colab Notebooks/NLPstudy/'

• google colab을 사용했기에 google drive를 마운트해주고, 경로를 PATH에 저장해준다.

lines = pd.read_csv(PATH+'fra.txt', names = ['src',  'tar', 'lic'], sep='\t')
# src는 source의 줄임말로 입력 문장을 의미 / tar은 target으로 번역하고자하는 문장을 의미
del lines['lic']
len(lines) #데이터의 갯수 19만개정도

• 입력문장을 src, 번역하고자 하는 문장을 tar 으로 지정한다.
• 데이터는 총 19만개 정도 된다.

 

2.4 데이터 전처리

▶ 데이터 선택

lines = lines.loc[:, 'src':'tar']
lines = lines[0:60000] # 6만개만 저장
lines.sample(10) #랜덤으로 뽑은 10개의 샘플

• 실습은 6만개의 데이터만 사용할 것이다. 

▶ 데이터 형태 만져주기

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

• 위에 seq2seq모델 설명을 보면 번역하고 싶은 문장에는 시작 심볼인 <sos>와 문장이 끝나는 심볼인 <eos>가 존재한다.
• 해당 문장에는 시작과 끝의 심볼이 정해져 있지 않으니 따로 지정해서 넣어주자.
• <sos> : \t / <eos> : \n 으로 넣어주었고, tar 문장에 심볼이 정상적으로 입력됨을 확인하자.

▶ 글자 집합 구축

• 영어는 79글자, 프랑스어는 105글자로 구성되어있다. 글자를 일부만 출력해보면, 아래와 같다.

# 글자 집합 생성 (토큰단위가 아닌 '글자'단위로 집합을 구축)
src_vocab=set()
for line in lines.src: # 1줄씩 읽음
    for char in line: # 1개의 글자씩 읽음
        src_vocab.add(char)

tar_vocab=set()
for line in lines.tar:
    for char in line:
        tar_vocab.add(char)
        

src_vocab_size = len(src_vocab)+1
tar_vocab_size = len(tar_vocab)+1
print(src_vocab_size)
print(tar_vocab_size)

src_vocab = sorted(list(src_vocab))
tar_vocab = sorted(list(tar_vocab))
print(src_vocab[45:75]) #일부만 출력해보자
print(tar_vocab[45:75])

• 글장 인덱스를 부여해서 dictionary로 만들 것이다.

# 글자에 인덱스를 부여하여 dict으로 표현
src_to_index = dict([(word, i+1) for i, word in enumerate(src_vocab)])
tar_to_index = dict([(word, i+1) for i, word in enumerate(tar_vocab)])
print(src_to_index)
print(tar_to_index)

 

▶ 정수 인코딩 진행

# 인덱스가 부여된 글자 집합으로 부터 갖고 있는 훈련 데이터에 정수 인코딩을 수행
# 입력이 될 영어 문장 샘플에 대해 인코딩을 수행
encoder_input = []
for line in lines.src: #입력 데이터에서 1줄씩 문장을 읽음
    temp_X = []
    for w in line: #각 줄에서 1개씩 글자를 읽음
      temp_X.append(src_to_index[w]) # 글자를 해당되는 정수로 변환
    encoder_input.append(temp_X)
#예시 5개만 출력해보자
print(encoder_input[:5])

# 디코더의 입력이 될 프랑스어 데이터에 대해 정수 인코딩 수행
decoder_input = []
for line in lines.tar:
    temp_X = []
    for w in line:
      temp_X.append(tar_to_index[w])
    decoder_input.append(temp_X)
#예시 5개만 출력해보자
print(decoder_input[:5])

# 디코더의 예측값과 비교하기 위한 실제값이 필요함
# 실제값에는 시작 심볼 <sos>가 있을 필요가 없음 
# 시작심볼인 \t 를 제거해주자 
decoder_target = []
for line in lines.tar:
    t=0 # t가 0인 처음을 제외하고 temp_X에 append해주는과정을 거친다.
    temp_X = []
    for w in line:
      if t>0:
        temp_X.append(tar_to_index[w])
      t=t+1
    decoder_target.append(temp_X)
print(decoder_target[:5])

• decoder input에서는 모든 임베딩된 결과가 1로 시작했다. (시작심볼떄문)
• decoder target 에서는 1이 제외된 것을 보면, 정상적으로 시작 심볼이 제거됨을 알 수 있다.

 

▶ 패딩

# 영어와 프랑스어의 가장 긴 단어 탐색
max_src_len = max([len(line) for line in lines.src])
max_tar_len = max([len(line) for line in lines.tar])
print(max_src_len)
print(max_tar_len)

• 영어 24 / 프랑스어 76 로 패딩 진행
• 이번 병렬 데이터는 하나의 쌍이더라도 전부 길이가 다를 수 있으므로 (처음에 언급)
• 패딩을 할때도 두 개의 데이터 길이를 전부 동일하게 할 필요 없음

# 패딩 진행
encoder_input = pad_sequences(encoder_input, maxlen=max_src_len, padding='post')
decoder_input = pad_sequences(decoder_input, maxlen=max_tar_len, padding='post')
decoder_target = pad_sequences(decoder_target, maxlen=max_tar_len, padding='post')

 

▶ 원핫 인코딩

# 글자 단위 번역기 이므로 워드 임베딩은 별도로 사용하지 않을 것
# 예측과 오차 측정에 사용되는 실제값 뿐만 아니라 입력값도 원핫벡터를 사용함
encoder_input = to_categorical(encoder_input)
decoder_input = to_categorical(decoder_input)
decoder_target = to_categorical(decoder_target)

 

2.5 교사 강요 (Teacher Forcing)

• 이전 seq2seq 모델 설명을 보면, 현재 시점의 디코더 셀의 입력은 이전 디코더의 출력을 입력으로 받는다고 배웠다. 근데 왜 decoder_input이 필요한가?
• 이전 시점의 실제값을 현재 시점의 디코더 셀의 입력값으로 하는 방법을 사용할 것
• 이전 시점의 디코더 셀의 예측이 틀렸는데 이를 현재 시점의 디코더 셀의 입력값으로 사용하게되면 현재 시점의 디코더 셀의 예측까지 잘못될 가능성이 높고, 이는 연쇄적으로 작용하여 디코더 전체 예측을 어렵게 하기 떄문
• 이와 같이 RNN의 모든 시점에 대해서 이전 시점의 예측값 대신 실제값을 입력으로 주는 방법을 교사강요 라고 함

 

2.6 seq2seq 기계 번역 훈련시키기

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_outputs도 같이 리턴받기는 했지만 여기서는 필요없으므로 이 값은 버림.
encoder_states = [state_h, state_c]
# LSTM은 바닐라 RNN과는 달리 상태가 두 개. 바로 은닉 상태와 셀 상태.

• LSTM의 은닉상태 크기는 256으로 선택
• 인코더 내부상태를 디코더로 전달해야하기에 return_state = True로 설정
• LSTM에서 state_h, state_c를 리턴받는데 state_h는 은닉상태고 state_c는 셀상태에 해당
• 즉 은닉상태와 셀 상태를 전달해준다.
• 이 두개를 ecoder states에 저장하고, 이를 디코더에 전달하므로서 두가지 상태를 모두 디코더로 전달할 것.
• 앞서 배운 문맥벡터(context vector)가 encoder_state 에 해당하는 것!

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_states)
# 디코더의 첫 상태를 인코더의 은닉 상태, 셀 상태로 합니다.
decoder_softmax_layer = Dense(tar_vocab_size, activation='softmax')
decoder_outputs = decoder_softmax_layer(decoder_outputs)

model = Model([encoder_inputs, decoder_inputs], decoder_outputs)
model.compile(optimizer="rmsprop", loss="categorical_crossentropy")

• 디코더는 인코더의 마지막 상태를 초기 은닉상태로 사용 즉, initial_state의 인자값으로 encoder_state를 받는것이 이에 해당
• 디코더의 은닉상태도 256으로 주어짐
• 디코더도 은닉상태, 셀 상태를 리턴하기는 하지만 훈련 과정에서는 사용하지 않음
• 그 후 출력층에 프랑스어의 단어 집합의 크기만큼 뉴런을 배치한 후, 소프트맥스 함수를 사용하여 실제값과의 오차를 구함

model.fit(x=[encoder_input, decoder_input], y=decoder_target, batch_size=64, epochs=3, validation_split=0.2)

• 시간관계상 epochs를 3으로만 학습했다. (본 코드는 50으로 진행했다.)
• 입력으로는 encoder_input, 디코더의 실제값인 decoder_input을 넣는다.

 

2.7 seq2seq 기계 번역기 동작시키기

훈련과정과 동작과정은 다르다. 동작 과정에서는 encoder model과 decoder model을 따로 만들어서 입력한 문장에 대해서 기계번역을 하도록 모델을 조정한 후, 동작시켜볼 것이다. (훈련된 모델을 사용하지 않는 것인가? 흠.. )
전반적인 번역 동작 단계를 정리하면 아래와 같다.

1. 번역하고자 하는 입력 문장이 인코더에 들어가서 은닉 상태와 셀 상태를 얻는다.
2. 상태와 <sos>에 해당하는 \t 를 디코더로 보낸다.
3. 디코더가 <eos>에 해당하는 \n이 나올 때까지 다음 문자를 예측하는 행동을 반복한다.

▶ 인코더 모델 정의

# 앞에서 정의한 encoder_input = Input(shape=(None, src_vocab_size))
# outputs = encoder_states : encoder_lstm으로 부터 받은 은닉상태와 셀상태값 [state_h, state_c]
encoder_model = Model(inputs=encoder_inputs, outputs=encoder_states)

• 우선 인코더를 encoder_model 로 정의하자.

 

▶ 디코더 모델 정의

# 이전 시점의 상태들을 저장하는 텐서
decoder_state_input_h = Input(shape=(256,))
decoder_state_input_c = Input(shape=(256,))
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)
# 문장의 다음 단어를 예측하기 위해서 초기 상태(initial_state)를 이전 시점의 상태로 사용. 이는 뒤의 함수 decode_sequence()에 구현
decoder_states = [state_h, state_c]
# 훈련 과정에서와 달리 LSTM의 리턴하는 은닉 상태와 셀 상태인 state_h와 state_c를 버리지 않음.
decoder_outputs = decoder_softmax_layer(decoder_outputs)
decoder_model = Model(inputs=[decoder_inputs] + decoder_states_inputs, outputs=[decoder_outputs] + decoder_states)

• 이전 시점의 상태를 저장하는 텐서를 만들고, decoder_lstm 으로 부터 나 

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())

• 인덱스로 부터 단어를 얻을 수 있는 index_to_src / index_to_tar을 만들어준다.

def decode_sequence(input_seq):
    # 입력으로부터 인코더의 상태를 얻음
    states_value = encoder_model.predict(input_seq)

    # <SOS>에 해당하는 원-핫 벡터 생성
    target_seq = np.zeros((1, 1, tar_vocab_size))
    target_seq[0, 0, tar_to_index['\t']] = 1.

    stop_condition = False
    decoded_sentence = ""

    # stop_condition이 True가 될 때까지 루프 반복
    while not stop_condition:
        # 이점 시점의 상태 states_value를 현 시점의 초기 상태로 사용
        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

        # <eos>에 도달하거나 최대 길이를 넘으면 중단.
        if (sampled_char == '\n' or
           len(decoded_sentence) > max_tar_len):
            stop_condition = True

        # 현재 시점의 예측 결과를 다음 시점의 입력으로 사용하기 위해 저장
        target_seq = np.zeros((1, 1, tar_vocab_size))
        target_seq[0, 0, sampled_token_index] = 1.

        # 현재 시점의 상태를 다음 시점의 상태로 사용하기 위해 저장
        states_value = [h, c]

    return decoded_sentence

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][1:len(lines.tar[seq_index])-1]) # '\t'와 '\n'을 빼고 출력
    print('번역기가 번역한 문장:', decoded_sentence[:len(decoded_sentence)-1]) # '\n'을 빼고 출력

 

이렇게 글자 수준에서의 기계번역을 구현해 보았다. 단어 수준에서의 번역기 구현 방법이 궁금하다면 여기를 클릭하고 별도로 공부를 진행하자.