[Keras] Seq2Seq 문장 번역 예제

다른 분의 강좌를 보면서 따라했습니다.

사실 따라했다기 보다는 코드 복붙해놓고 이해하려고 노력했다는게 더 정확하지만 ...

코드:

# https://blog.keras.io/a-ten-minute-introduction-to-sequence-to-sequence-learning-in-keras.html
 
!pip install -U -q PyDrive
 
from keras import layers, models
from keras import datasets
from keras import backend
import matplotlib.pyplot as plt
import numpy as np
 
def get_data(file_id):
  """
  구글 드라이브에서 file_id에 해당하는 파일을 가져와 읽습니다.
  Colab에서 돌리기 위해 필요합니다.
  """
  
  # Install the PyDrive wrapper & import libraries.
  # This only needs to be done once per notebook.
  from pydrive.auth import GoogleAuth
  from pydrive.drive import GoogleDrive
  from google.colab import auth
  from oauth2client.client import GoogleCredentials
 
  # Authenticate and create the PyDrive client.
  # This only needs to be done once per notebook.
  auth.authenticate_user()
  gauth = GoogleAuth()
  gauth.credentials = GoogleCredentials.get_application_default()
  drive = GoogleDrive(gauth)
 
  # Download a file based on its file ID.
  #
  # A file ID looks like: laggVyWshwcyP6kEI-y_W3P8D26sz
  downloaded = drive.CreateFile({'id': file_id})
  return downloaded.GetContentString()
 
# 학습 정보
batch_size = 64
epochs = 64
latent_dim = 256
num_samples = 1024 # 학습 데이터 개수
 
# 문장 벡터화
input_texts = []
target_texts = []
input_characters = set()
target_characters = set()
 
data_set = {
    'kor': '12UL7MH38rxRFskhYuj1B9zU7wUXgv8Q7',
    'jpn': '1ULdUnurb_DEDTzJFaycdWHWkYWqPu4Bo'
}
 
lines = get_data(data_set['jpn']).split('\n')
for line in lines[: min(num_samples, len(lines) - 1)]:
  input_text, target_text = line.split('\t')
  # "\t"문자를 시작 문자로, "\n"문자를 종료 문자로 사용.
  target_text = '\t' + target_text + '\n'
  input_texts.append(input_text)
  target_texts.append(target_text)
 
  # 문자 집합 생성
  for char in input_text:
    if char not in input_characters:
      input_characters.add(char)
  for char in target_text:
    if char not in target_characters:
      target_characters.add(char)
 
num_samples = len(input_texts)
            
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])
 
print('Number of samples:', num_samples)
print('Number of unique input tokens:', num_encoder_tokens)
print('Number of unique output tokens:', num_decoder_tokens)
print('Max sequence length for inputs:', max_encoder_seq_length)
print('Max sequence length for outputs:', max_decoder_seq_length)
 
# 문자 -> 숫자 변환용 사전
input_token_index = dict(
    [(char, i) for i, char in enumerate(input_characters)])
target_token_index = dict(
    [(char, i) for i, char in enumerate(target_characters)])
 
# 학습에 사용할 데이터를 담을 3차원 배열
encoder_input_data = np.zeros(
    (num_samples, max_encoder_seq_length, num_encoder_tokens),
    dtype='float32')
decoder_input_data = np.zeros(
    (num_samples, max_decoder_seq_length, num_decoder_tokens),
    dtype='float32')
decoder_target_data = np.zeros(
    (num_samples, max_decoder_seq_length, num_decoder_tokens),
    dtype='float32')
 
# 문장을 문자 단위로 원 핫 인코딩하면서 학습용 데이터를 만듬
for i, (input_text, target_text) in enumerate(zip(input_texts, target_texts)):
  for t, char in enumerate(input_text):
    encoder_input_data[i, t, input_token_index[char]] = 1.
  for t, char in enumerate(target_text):
    decoder_input_data[i, t, target_token_index[char]] = 1.
    if t > 0:
      decoder_target_data[i, t - 1, target_token_index[char]] = 1.
      
# 인코더 생성
encoder_inputs = layers.Input(shape=(None, num_encoder_tokens))
encoder = layers.LSTM(latent_dim, return_state=True)
encoder_outputs, state_h, state_c = encoder(encoder_inputs)
# 인코더의 출력은 필요없고 상태만 중요
encoder_states = [state_h, state_c]
 
# 디코더 생성.
decoder_inputs = layers.Input(shape=(None, num_decoder_tokens))
decoder_lstm = layers.LSTM(latent_dim, return_sequences=True, return_state=True)
# 디코더의 내부 상태는 학습에 사용하지 않지만 추론(테스트) 단계에서는 필요함.
# 디코더의 초기 상태를 인코더의 최종 상태로 설정.
decoder_outputs, _, _ = decoder_lstm(decoder_inputs,
                                     initial_state=encoder_states)
decoder_dense = layers.Dense(num_decoder_tokens, activation='softmax')
decoder_outputs = decoder_dense(decoder_outputs)
 
# 모델 생성
model = models.Model([encoder_inputs, decoder_inputs], decoder_outputs)
 
# 학습
model.compile(optimizer='rmsprop', loss='categorical_crossentropy')
model.fit([encoder_input_data, decoder_input_data], decoder_target_data,
          batch_size=batch_size,
          epochs=epochs,
          validation_split=0.2,
          verbose=2)
 
# 추론(테스트)
 
# 추론 모델 생성
encoder_model = models.Model(encoder_inputs, encoder_states)
 
decoder_state_input_h = layers.Input(shape=(latent_dim,))
decoder_state_input_c = layers.Input(shape=(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 = models.Model(
    [decoder_inputs] + decoder_states_inputs,
    [decoder_outputs] + decoder_states)
 
# 숫자 -> 문자 변환용 사전
reverse_input_char_index = dict(
    (i, char) for char, i in input_token_index.items())
reverse_target_char_index = dict(
    (i, char) for char, i in target_token_index.items())
 
def decode_sequence(input_seq):
  # 입력 문장을 인코딩
  states_value = encoder_model.predict(input_seq)
 
  # 디코더의 입력으로 쓸 단일 문자
  target_seq = np.zeros((1, 1, num_decoder_tokens))
  # 첫 입력은 시작 문자인 '\t'로 설정
  target_seq[0, 0, target_token_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 = reverse_target_char_index[sampled_token_index]
    decoded_sentence += sampled_char
 
    # 종료 문자가 나왔거나 문장 길이가 한계를 넘으면 종료
    if (sampled_char == '\n' or len(decoded_sentence) > max_decoder_seq_length):
      stop_condition = True
 
    # 디코더의 다음 입력으로 쓸 데이터 갱신
    target_seq = np.zeros((1, 1, num_decoder_tokens))
    target_seq[0, 0, sampled_token_index] = 1.
    
    states_value = [h, c]
 
  return decoded_sentence
 
for seq_index in range(30):
  input_seq = encoder_input_data[seq_index: seq_index + 1]
  decoded_sentence = decode_sequence(input_seq)
  print('"{}" -> "{}"'.format(input_texts[seq_index], decoded_sentence.strip()))

결과:

Number of samples: 1024
Number of unique input tokens: 64
Number of unique output tokens: 485
Max sequence length for inputs: 12
Max sequence length for outputs: 20
Train on 819 samples, validate on 205 samples
Epoch 1/64
- 3s - loss: 1.8982 - val_loss: 1.7415
Epoch 2/64
- 1s - loss: 1.5878 - val_loss: 1.7715

(중략)

Epoch 63/64
- 1s - loss: 0.3502 - val_loss: 1.4305
Epoch 64/64
- 1s - loss: 0.3327 - val_loss: 1.4706
"Hi." -> "やっておい。"
"Hi." -> "やっておい。"
"Run." -> "走って！"
"Run." -> "走って！"
"Who?" -> "誰？"
"Wow!" -> "すごい！"
"Wow!" -> "すごい！"
"Wow!" -> "すごい！"
"Fire!" -> "火事！"
"Fire!" -> "火事！"
"Help!" -> "助けて！"
"Jump!" -> "飛び降りろ！"
"Jump!" -> "飛び降りろ！"
"Jump!" -> "飛び降りろ！"
"Jump!" -> "飛び降りろ！"
"Jump!" -> "飛び降りろ！"
"Jump." -> "飛び跳ねて！"
"Jump." -> "飛び跳ねて！"
"Stop!" -> "やめろ！"
"Stop!" -> "やめろ！"
"Wait!" -> "待って！"
"Go on." -> "進んで。"
"Go on." -> "進んで。"
"Go on." -> "進んで。"
"Go on." -> "進んで。"
"Hello!" -> "こんにちは！"
"Hello!" -> "こんにちは！"
"Hello!" -> "こんにちは！"
"Hurry!" -> "もちろん！"
"I see." -> "私はそうをいいます。"

Colab에서 돌린거라 구글 드라이브의 파일을 읽도록 수정했는데 다른 분들도 불러와질지 모르겠네요.

번역은 왜 일본어로 했냐면 한국어는 데이터가 부족해서 학습이 잘 안되더라고요...

계속 '톰', '자살' 타령;;

참고 : http://neurowhai.tistory.com/162

구현이 꽤 이해하기 힘드네요 ㅠㅠ

Keras에 익숙하지 않은 탓도 있지만 추론 모델을 만드는 과정은 이론적으론 알지만 코드는 복잡하게 생겨서 이해가;;

주석이 부족한건 제가 이해를 다 못해서 그렇습니다 ㅠ