Transformer 를 사용한 seq2seq 모델 실습

개요

케창딥 11장에 나오는 Transformer Encoder 코드 학습 내용을 정리 합니다.

https://github.com/gilbutITbook/080315/blob/main/chapter11_part04_sequence-to-sequence-learning.ipynb

080315/chapter11_part04_sequence-to-sequence-learning.ipynb at main · gilbutITbook/080315

 

출처 : https://jalammar.github.io/illustrated-transformer/

 

앞서 구현한 Transformer Encoder 와 Decoder 를 사용하여 기계번역을 구현

https://gugalove.tistory.com/40

Transformer Encoder 구현 및 학습

class TransformerDecoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention_1 = layers.MultiHeadAttention(
            num_heads=num_heads, key_dim=embed_dim)
        self.attention_2 = layers.MultiHeadAttention(
            num_heads=num_heads, key_dim=embed_dim)
        self.dense_proj = keras.Sequential(
            [layers.Dense(dense_dim, activation="relu"),
             layers.Dense(embed_dim),]
        )
        self.layernorm_1 = layers.LayerNormalization()
        self.layernorm_2 = layers.LayerNormalization()
        self.layernorm_3 = layers.LayerNormalization()
        self.supports_masking = True

    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "num_heads": self.num_heads,
            "dense_dim": self.dense_dim,
        })
        return config

    def call(self, inputs, encoder_outputs, mask=None):
        attention_output_1 = self.attention_1(
            query=inputs,
            value=inputs,
            key=inputs,
            use_causal_mask=True)
        attention_output_1 = self.layernorm_1(inputs + attention_output_1)
        attention_output_2 = self.attention_2(
            query=attention_output_1,
            value=encoder_outputs,
            key=encoder_outputs
        )
        attention_output_2 = self.layernorm_2(
            attention_output_1 + attention_output_2)
        proj_output = self.dense_proj(attention_output_2)
        return self.layernorm_3(attention_output_2 + proj_output)

 

https://gugalove.tistory.com/41

Transformer Decoder 구현 및 학습

class TransformerEncoder(layers.Layer):
    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
        super().__init__(**kwargs)
        self.embed_dim = embed_dim
        self.dense_dim = dense_dim
        self.num_heads = num_heads
        self.attention = layers.MultiHeadAttention(
            num_heads=num_heads, key_dim=embed_dim)
        self.dense_proj = keras.Sequential(
            [layers.Dense(dense_dim, activation="relu"),
             layers.Dense(embed_dim),]
        )
        self.layernorm_1 = layers.LayerNormalization()
        self.layernorm_2 = layers.LayerNormalization()

    def call(self, inputs, mask=None):
        if mask is not None:
            mask = mask[:, tf.newaxis, :]
        attention_output = self.attention(
            inputs, inputs, attention_mask=mask)
        proj_input = self.layernorm_1(inputs + attention_output)
        proj_output = self.dense_proj(proj_input)
        return self.layernorm_2(proj_input + proj_output)

    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "num_heads": self.num_heads,
            "dense_dim": self.dense_dim,
        })
        return config

 

영어와 스페인어 데이터 셋 받기

TensorFlow에서 제공하는 데이터셋 중 하나로, 스페인어-영어 번역 데이터를 포함한 파일

 

• 간결한 구조: 스페인어와 영어 간의 1:1 대응 데이터로 구성되어 있어 전처리가 간단합니다.
• 오픈 소스: 공개적으로 제공되어 무료로 사용할 수 있습니다.
• 작은 크기: 초보자나 모델 실험을 위한 적합한 크기.

 

!wget http://storage.googleapis.com/download.tensorflow.org/data/spa-eng.zip
!unzip -q spa-eng.zip

text_file = "spa-eng/spa.txt"
with open(text_file) as f:
    lines = f.read().split("\n")[:-1]
text_pairs = []
for line in lines:
    english, spanish = line.split("\t")
    spanish = "[start] " + spanish + " [end]"
    text_pairs.append((english, spanish))

import random
print(random.choice(text_pairs))

('Everyone wants to feel loved.', '[start] Todos quieren sentirse amados. [end]')

import random
random.shuffle(text_pairs)
num_val_samples = int(0.15 * len(text_pairs))
num_train_samples = len(text_pairs) - 2 * num_val_samples
train_pairs = text_pairs[:num_train_samples]
val_pairs = text_pairs[num_train_samples:num_train_samples + num_val_samples]
test_pairs = text_pairs[num_train_samples + num_val_samples:]

 

 

영어와 스페인어 텍스트 쌍을 벡터화하기

import tensorflow as tf
from tensorflow.keras import layers
import string
import re

strip_chars = string.punctuation + "¿"
strip_chars = strip_chars.replace("[", "")
strip_chars = strip_chars.replace("]", "")

def custom_standardization(input_string):
    lowercase = tf.strings.lower(input_string)
    return tf.strings.regex_replace(
        lowercase, f"[{re.escape(strip_chars)}]", "")

vocab_size = 15000
sequence_length = 20

source_vectorization = layers.TextVectorization(
    max_tokens=vocab_size,
    output_mode="int",
    output_sequence_length=sequence_length,
)
target_vectorization = layers.TextVectorization(
    max_tokens=vocab_size,
    output_mode="int",
    output_sequence_length=sequence_length + 1,
    standardize=custom_standardization,
)
train_english_texts = [pair[0] for pair in train_pairs]
train_spanish_texts = [pair[1] for pair in train_pairs]
source_vectorization.adapt(train_english_texts)
target_vectorization.adapt(train_spanish_texts)

 

번역 작업을 위한 데이터셋 준비하기

batch_size = 64

def format_dataset(eng, spa):
    eng = source_vectorization(eng)
    spa = target_vectorization(spa)
    return ({
        "english": eng,
        "spanish": spa[:, :-1],
    }, spa[:, 1:])

def make_dataset(pairs):
    eng_texts, spa_texts = zip(*pairs)
    eng_texts = list(eng_texts)
    spa_texts = list(spa_texts)
    dataset = tf.data.Dataset.from_tensor_slices((eng_texts, spa_texts))
    dataset = dataset.batch(batch_size)
    dataset = dataset.map(format_dataset, num_parallel_calls=4)
    return dataset.shuffle(2048).prefetch(16).cache()

train_ds = make_dataset(train_pairs)
val_ds = make_dataset(val_pairs)

for inputs, targets in train_ds.take(1):
    print(f"inputs['english'].shape: {inputs['english'].shape}")
    print(f"inputs['spanish'].shape: {inputs['spanish'].shape}")
    print(f"targets.shape: {targets.shape}")

inputs['english'].shape: (64, 20)
inputs['spanish'].shape: (64, 20)
targets.shape: (64, 20)

 

기존 PositionalEmbedding 에서 에러가 발생 하여 수정

class PositionalEmbedding(layers.Layer):
    def __init__(self, sequence_length, input_dim, output_dim, **kwargs):
        super().__init__(**kwargs)
        self.token_embeddings = layers.Embedding(
            input_dim=input_dim, output_dim=output_dim)
        self.position_embeddings = layers.Embedding(
            input_dim=sequence_length, output_dim=output_dim)
        self.sequence_length = sequence_length
        self.input_dim = input_dim
        self.output_dim = output_dim

    def call(self, inputs):
        length = tf.shape(inputs)[-1]
        positions = tf.range(start=0, limit=length, delta=1)
        embedded_tokens = self.token_embeddings(inputs)
        embedded_positions = self.position_embeddings(positions)
        mask = tf.cast(tf.math.not_equal(inputs, 0), dtype=tf.float32)
        return (embedded_tokens + embedded_positions) * mask[..., tf.newaxis]

    def get_config(self):
        config = super(PositionalEmbedding, self).get_config()
        config.update({
            "output_dim": self.output_dim,
            "sequence_length": self.sequence_length,
            "input_dim": self.input_dim,
        })
        return config

 

수정 내용 :

에러의 원인은 PositionalEmbedding 클래스의 compute_mask 메서드에서 발생하고 있습니다. 

여기서 tf.math.not_equal(inputs, 0)이 문제가 되는 이유는, inputs이 KerasTensor이기 때문입니다.

Keras에서 KerasTensor는 심볼릭 텐서로, 일반적인 TensorFlow 연산과 호환되지 않습니다. 즉, Keras Functional API에서 Input을 사용하여 생성된 텐서는 심볼릭이기 때문에 일반적인 TensorFlow 함수 (tf.math.not_equal)에 직접적으로 사용될 수 없습니다.

 

위 코드와 같이 compute_mask 제거 하고 call 메서드에서 직접 처리  하였습니다.

 

Transformer 모델 정의

Transformer 구조를 구현한 것으로, 입력 시퀀스를 임베딩하고, 인코더와 디코더를 사용하여 번역을 수행하는 모델을 정의합니다. 이 Transformer 모델은 영어(소스 언어) 입력을 받아 스페인어(타겟 언어)로 번역하는 역할을 합니다.

• embed_dim: 각 단어 임베딩의 차원을 256으로 설정합니다.
• dense_dim: 피드포워드 네트워크에서 사용될 중간 차원을 2048로 설정합니다.
• num_heads: 멀티 헤드 어텐션의 헤드 수를 8로 설정합니다.
• encoder_inputs: 인코더 입력으로 사용될 텐서를 정의합니다. 여기서 입력은 가변적인 길이를 가진 시퀀스(shape=(None,))이며, 데이터 타입은 정수형(int64)입니다. 입력의 이름은 "english"로 설정됩니다.
• PositionalEmbedding: 입력 시퀀스의 각 단어에 대해 임베딩을 적용하고, 각 위치에 대해 위치 임베딩을 결합합니다.
  • sequence_length, vocab_size, embed_dim: 시퀀스의 길이, 어휘의 크기, 임베딩 차원을 인자로 받아 임베딩을 생성합니다.
  • encoder_inputs 를 임베딩 레이어에 전달하여 최종 임베딩 텐서를 얻습니다.
• TransformerEncoder: 인코더 부분의 핵심인 Transformer 인코더 레이어를 적용합니다.
  • embed_dim, dense_dim, num_heads 값을 통해 임베딩 차원, 피드포워드 네트워크 차원, 멀티 헤드 수를 설정하여 인코더 레이어의 출력을 얻습니다.
• decoder_inputs: 디코더 입력으로 사용될 텐서를 정의합니다. 입력은 영어 인코더와 마찬가지로 가변적인 길이를 가지며(shape=(None,)), 정수형 데이터 타입을 가집니다. 입력의 이름은 "spanish"로 설정됩니다.
• 디코더 입력에 대해서도 동일하게 PositionalEmbedding 레이어를 적용합니다. 각 단어와 위치에 대한 임베딩을 생성합니다.
• TransformerDecoder: 디코더 부분의 핵심인 Transformer 디코더 레이어를 적용합니다.
  • embed_dim, dense_dim, num_heads 값을 통해 임베딩 차원, 피드포워드 네트워크 차원, 멀티 헤드 수를 설정합니다.
  • 디코더 입력 (x)와 인코더 출력 (encoder_outputs)을 함께 사용하여 디코더의 출력을 생성합니다.
• Dropout: 과적합을 방지하기 위해 0.5의 비율로 드롭아웃을 적용합니다.
• Dense: 타겟 언어의 각 단어를 예측하기 위해 Dense 레이어를 사용합니다. 출력 크기는 어휘의 크기 (vocab_size)이며, 각 단어의 확률을 계산하기 위해 소프트맥스 활성화 함수를 사용합니다.
  • 이를 통해 번역된 각 단어의 확률 분포를 얻습니다.
• keras.Model: 전체 인코더와 디코더 구조를 하나의 모델로 정의합니다.
  • [encoder_inputs, decoder_inputs] 를 입력으로 받아, decoder_outputs 를 출력으로 하는 모델입니다.

embed_dim = 256
dense_dim = 2048
num_heads = 8

encoder_inputs = keras.Input(shape=(None,), dtype="int64", name="english")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(encoder_inputs)
encoder_outputs = TransformerEncoder(embed_dim, dense_dim, num_heads)(x)

decoder_inputs = keras.Input(shape=(None,), dtype="int64", name="spanish")
x = PositionalEmbedding(sequence_length, vocab_size, embed_dim)(decoder_inputs)
x = TransformerDecoder(embed_dim, dense_dim, num_heads)(x, encoder_outputs)
x = layers.Dropout(0.5)(x)
decoder_outputs = layers.Dense(vocab_size, activation="softmax")(x)
transformer = keras.Model([encoder_inputs, decoder_inputs], decoder_outputs)

 

모델 훈련(구글 코랩 PRO A100 GPU 사용)

transformer.compile(
    optimizer="rmsprop",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"])
transformer.fit(train_ds, epochs=30, validation_data=val_ds)

 

Transformer 모델을 사용해 새로운 문장 번역하기

• spa_vocab: 스페인어 어휘 사전을 가져옵니다.
• spa_index_lookup: 인덱스를 스페인어 단어로 매핑하기 위한 딕셔너리를 만듭니다.
• max_decoded_sentence_length: 디코딩 시 생성할 최대 문장 길이를 설정합니다.
• decode_sequence: 주어진 입력 문장을 번역하는 함수입니다.
  • 입력 문장을 벡터화하여 Transformer 모델의 입력으로 사용합니다.
  • [start] 토큰으로 시작하여 최대 길이까지 번역을 반복합니다.
  • np.argmax(predictions[0, i, :]): 현재 타임스텝에서 가장 높은 확률을 가진 단어를 선택합니다.
  • 종료 토큰이 생성되면 디코딩을 중단하고 번역된 문장을 반환합니다.

import numpy as np
spa_vocab = target_vectorization.get_vocabulary()
spa_index_lookup = dict(zip(range(len(spa_vocab)), spa_vocab))
max_decoded_sentence_length = 20

def decode_sequence(input_sentence):
    # 입력 문장을 벡터화하여 모델에 입력할 수 있도록 준비합니다.
    tokenized_input_sentence = source_vectorization([input_sentence])
    decoded_sentence = "[start]"
    
    # 최대 디코딩 길이 동안 반복하여 번역된 문장을 생성합니다.
    for i in range(max_decoded_sentence_length):
        # 현재까지 디코딩된 문장을 벡터화합니다 (마지막 토큰 제외).
        tokenized_target_sentence = target_vectorization([decoded_sentence])[:, :-1]
        
        # 인코더와 디코더 입력을 사용하여 예측을 수행합니다.
        predictions = transformer([tokenized_input_sentence, tokenized_target_sentence])
        
        # 가장 높은 확률을 가진 토큰을 선택합니다.
        sampled_token_index = np.argmax(predictions[0, i, :])
        sampled_token = spa_index_lookup[sampled_token_index]
        
        # 선택된 토큰을 현재 디코딩된 문장에 추가합니다.
        decoded_sentence += " " + sampled_token
        
        # 만약 종료 토큰("[end]")이 생성되면 반복을 중단합니다.
        if sampled_token == "[end]":
            break
    
    return decoded_sentence

# 테스트 데이터를 사용하여 번역 결과를 출력합니다.
test_eng_texts = [pair[0] for pair in test_pairs]
for _ in range(20):
    input_sentence = random.choice(test_eng_texts)
    print("-")
    print(input_sentence)
    print(decode_sequence(input_sentence))

-
We go to school because we want to learn.
[start] vamos a la escuela porque queremos aprender [end]
-
I'm e-mailing to follow up on our phone conversation this morning.
[start] estoy [UNK] a seguir el teléfono de nuestra conversación esta mañana [end]
-
I'm not hungry right now.
[start] no tengo hambre ahora [end]
-
The ship lowered its gangway after docking.
[start] el barco no [UNK] está [UNK] después de que maría [end]
-
Can you speak a little slower?
[start] puedes hablar un poco más despacio [end]
-
It's not easy writing a love letter in English.
[start] no es fácil escribir una carta de inglés [end]
-
If your prices are competitive, we will place a large order.
[start] si tus precios [UNK] son [UNK] una gran orden de gran [end]
-
I'm almost broke.
[start] soy casi se rompió [end]
-
She looks lonesome.
[start] parece ocupada [end]
-
Tom advised Mary not to go there by herself.
[start] tom le aconsejó a mary que no vaya ahí sola [end]
-
They are gathering nuts.
[start] ellos son hacia la reunión [end]
-
You don't remember, do you?
[start] no te acuerdas de hacer [end]
-
I do not have to study Japanese.
[start] no tengo que aprender japonés [end]
-
Tom is able to walk on his hands.
[start] tom puede caminar para [UNK] con las manos [end]
-
Tell those people to back off so that the helicopter can land.
[start] dile a esas personas que más que podemos [UNK] un helicóptero [UNK] de tierra [end]
-
If you don't want to go there, then we won't go there.
[start] si no quieres ir ahí a ir ahí [end]
-
I made her happy.
[start] le hice feliz [end]
-
I want to buy some shoes.
[start] quiero comprar algunos zapatos [end]
-
I don't like it when he does that.
[start] no me agrada cuando él hace eso [end]
-
He felt something crawl up his leg.
[start] Él se sentía cómo se sintió su pierna [end]