2024년 2월 6일 화요일

파이토치 트랜스포머 모델 사용자화를 통한 간단한 자연어 번역기 개발 방법 소개

이 글은 앞서 소개한 트랜스포머 모델을 사용자화하여, 간단한 자연어 번역기를 개발하는 방법을 보여준다. 자연어 번역기 개발을 위한 개념과 사용방법을 간략히 설명한다. 사용된 코드는 Github를 통해 제공된다. 참고로, 이 글에 있는 예제를 실행하기 위해서는 파이토치 라이브러리, GPU 드라이버가 컴퓨터에 설치되어 있어야 한다.
트랜스포머 기반 자연어 번역 개념도

머리말
앞의 글에서 보았듯이 트랜스포머는 자연어 번역을 위해 개발된 것이다. 이 글은 자연어 번역을 위해, 미리 잘 만들어진 파이토치 트랜스포머를 이용하기로 한다. 이 글은 파이토치 관련 레퍼런스와 예제를 참고하였다. 이 글에서 구현되는 코드는 다음 github를 확인한다.
트랜스포머에 대한 상세 동작 메커니즘은 다음 링크를 참고한다.
본문
TorchText 라이브러리는 자연어 데이터 세트를 만들기 위한 유틸리티가 있다. 이 예는 torchtext의 내장 데이터 세트를 사용하는 방법을 보여준다. 원시 텍스트 문장을 토큰화하고, 어휘를 구축하고, 토큰을 임베딩하여 텐서로 변환한다. 학습을 위해 Multi30k 데이터셋를 이용해 한 쌍의 원본-대상 문장을 준비한다. 

torchtext 데이터 세트를 다운로드하려면 https://github.com/pytorch/data 에 따라 torchdata를 설치한다. 코드는 다음과 같다.
multi30k.URL["train"] = "https://raw.githubusercontent.com/neychev/small_DL_repo/master/datasets/Multi30k/training.tar.gz"
multi30k.URL["valid"] = "https://raw.githubusercontent.com/neychev/small_DL_repo/master/datasets/Multi30k/validation.tar.gz"

SRC_LANGUAGE = 'de'
TGT_LANGUAGE = 'en'

# Place-holders
token_transform = {}
vocab_transform = {}

token_transform[SRC_LANGUAGE] = get_tokenizer('spacy', language='de_core_news_sm')
token_transform[TGT_LANGUAGE] = get_tokenizer('spacy', language='en_core_web_sm')

print(token_transform[SRC_LANGUAGE]("Eine Gruppe von Menschen steht vor einem Iglu ."))
print(token_transform[TGT_LANGUAGE]("A group of people are standing in front of an igloo ."))  

토큰화 결과

소스, 대상 언어 토크나이저를 위해, 관련된 라이브러리 종속을 설치한다.
pip install -U torchdata
pip install -U spacy
python -m spacy download en_core_web_sm
python -m spacy download de_core_news_sm

학습 데이터를 다음과 같이 준비한다.
token_transform[SRC_LANGUAGE] = get_tokenizer('spacy', language='de_core_news_sm')
token_transform[TGT_LANGUAGE] = get_tokenizer('spacy', language='en_core_web_sm')

# helper function to yield list of tokens
def yield_tokens(data_iter: Iterable, language: str) -> List[str]:
    language_index = {SRC_LANGUAGE: 0, TGT_LANGUAGE: 1}

    for data_sample in data_iter:
        yield token_transform[language](data_sample[language_index[language]])

# Define special symbols and indices
UNK_IDX, PAD_IDX, BOS_IDX, EOS_IDX = 0, 1, 2, 3
# Make sure the tokens are in order of their indices to properly insert them in vocab
special_symbols = ['<unk>', '<pad>', '<bos>', '<eos>']

for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
    # Training data Iterator
    train_iter = Multi30k(split='train', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    # Create torchtext's Vocab object
    vocab_transform[ln] = build_vocab_from_iterator(yield_tokens(train_iter, ln),
                                                    min_freq=1,
                                                    specials=special_symbols,
                                                    special_first=True)

# Set ``UNK_IDX`` as the default index. This index is returned when the token is not found.
# If not set, it throws ``RuntimeError`` when the queried token is not found in the Vocabulary.
for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
  vocab_transform[ln].set_default_index(UNK_IDX)

트랜스포머 기반 Seq2Seq 신경망 구현
이제, Transformer를 사용하는 Seq2Seq 네트워크를 생성한다. 네트워크 세 부분으로 구성되어 있다. 

첫 번째, 임베딩 레이어이다. 이 레이어는 입력 문장들을 텐서로 변환한다. 이 임베딩은 위치 인코딩을 사용하여 입력 토큰의 위치 정보를 모델에 제공한다. 
두 번째, 트랜스포머를 구현한다.
세 번째, 트랜스포머 모델의 출력은 선형 레이어를 통과해, 대상 언어의 각 토큰에 대한 생성 확률을 제공한다.

코드 구현은 다음과 같다. 
from torch import Tensor
import torch
import torch.nn as nn
from torch.nn import Transformer
import math
DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# 앞의 글에서 설명한 포지션 인코딩과 동일하다.
class PositionalEncoding(nn.Module):
    def __init__(self,
                 emb_size: int,
                 dropout: float,
                 maxlen: int = 5000):
        super(PositionalEncoding, self).__init__()
        den = torch.exp(- torch.arange(0, emb_size, 2)* math.log(10000) / emb_size)
        pos = torch.arange(0, maxlen).reshape(maxlen, 1)
        pos_embedding = torch.zeros((maxlen, emb_size))
        pos_embedding[:, 0::2] = torch.sin(pos * den)
        pos_embedding[:, 1::2] = torch.cos(pos * den)
        pos_embedding = pos_embedding.unsqueeze(-2)

        self.dropout = nn.Dropout(dropout)
        self.register_buffer('pos_embedding', pos_embedding)

    def forward(self, token_embedding: Tensor):
        return self.dropout(token_embedding + self.pos_embedding[:token_embedding.size(0), :])

# 토큰 임베딩 텐서 변환 모듈
class TokenEmbedding(nn.Module):
    def __init__(self, vocab_size: int, emb_size):
        super(TokenEmbedding, self).__init__()
        self.embedding = nn.Embedding(vocab_size, emb_size)
        self.emb_size = emb_size

    def forward(self, tokens: Tensor):
        return self.embedding(tokens.long()) * math.sqrt(self.emb_size)

# Seq2Seq 모듈
class Seq2SeqTransformer(nn.Module):
    def __init__(self,
                 num_encoder_layers: int,
                 num_decoder_layers: int,
                 emb_size: int,
                 nhead: int,
                 src_vocab_size: int,
                 tgt_vocab_size: int,
                 dim_feedforward: int = 512,
                 dropout: float = 0.1):
        super(Seq2SeqTransformer, self).__init__()
        self.transformer = Transformer(d_model=emb_size,
                                       nhead=nhead,
                                       num_encoder_layers=num_encoder_layers,
                                       num_decoder_layers=num_decoder_layers,
                                       dim_feedforward=dim_feedforward,
                                       dropout=dropout)  # 토치에 내장된 트랜스포머 사용
        self.generator = nn.Linear(emb_size, tgt_vocab_size)
        self.src_tok_emb = TokenEmbedding(src_vocab_size, emb_size)
        self.tgt_tok_emb = TokenEmbedding(tgt_vocab_size, emb_size)
        self.positional_encoding = PositionalEncoding(
            emb_size, dropout=dropout)

    def forward(self,
                src: Tensor,
                trg: Tensor,
                src_mask: Tensor,
                tgt_mask: Tensor,
                src_padding_mask: Tensor,
                tgt_padding_mask: Tensor,
                memory_key_padding_mask: Tensor):
        src_emb = self.positional_encoding(self.src_tok_emb(src))   # 소스 토큰 임베딩
        tgt_emb = self.positional_encoding(self.tgt_tok_emb(trg))   # 대상 토큰 임베딩
        outs = self.transformer(src_emb, tgt_emb, src_mask, tgt_mask, None,
                                src_padding_mask, tgt_padding_mask, memory_key_padding_mask)  # 트랜스포머 호출
        return self.generator(outs)  # 선형 레이어 입력 처리

    def encode(self, src: Tensor, src_mask: Tensor):
        return self.transformer.encoder(self.positional_encoding(
                            self.src_tok_emb(src)), src_mask)

    def decode(self, tgt: Tensor, memory: Tensor, tgt_mask: Tensor):
        return self.transformer.decoder(self.positional_encoding(
                          self.tgt_tok_emb(tgt)), memory,
                          tgt_mask)

훈련하는 동안 모델이 다음을 학습하지 못하도록, 후속 단어 마스크가 필요하다. 예측을 할 때 미래 단어를 숨길 마스크를 준비한다. 소스 및 대상 패딩 토큰 등도 준비한다. 
def generate_square_subsequent_mask(sz): 
    mask = (torch.triu(torch.ones((sz, sz), device=DEVICE)) == 1).transpose(0, 1)
    mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))
    return mask

def create_mask(src, tgt):
    src_seq_len = src.shape[0]
    tgt_seq_len = tgt.shape[0]

    tgt_mask = generate_square_subsequent_mask(tgt_seq_len)
    src_mask = torch.zeros((src_seq_len, src_seq_len),device=DEVICE).type(torch.bool)

    src_padding_mask = (src == PAD_IDX).transpose(0, 1)
    tgt_padding_mask = (tgt == PAD_IDX).transpose(0, 1)
    return src_mask, tgt_mask, src_padding_mask, tgt_padding_mask

이제 모델의 매개 변수를 정의하고 동일하게 인스턴스화한다. 여기서는 교차 엔트로피 손실 함수와 옵티마이저를 정의한다.
torch.manual_seed(0)

SRC_VOCAB_SIZE = len(vocab_transform[SRC_LANGUAGE])
TGT_VOCAB_SIZE = len(vocab_transform[TGT_LANGUAGE])
EMB_SIZE = 512
NHEAD = 8
FFN_HID_DIM = 512
BATCH_SIZE = 128
NUM_ENCODER_LAYERS = 3
NUM_DECODER_LAYERS = 3

transformer = Seq2SeqTransformer(NUM_ENCODER_LAYERS, NUM_DECODER_LAYERS, EMB_SIZE,
                                 NHEAD, SRC_VOCAB_SIZE, TGT_VOCAB_SIZE, FFN_HID_DIM)
for p in transformer.parameters():
    if p.dim() > 1:
        nn.init.xavier_uniform_(p)  # 각 레이어의 신경망 가중치를 적절히 초기화함

transformer = transformer.to(DEVICE)

loss_fn = torch.nn.CrossEntropyLoss(ignore_index=PAD_IDX)
optimizer = torch.optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)

데이터 학습
데이터 학습은 한 쌍의 소스-대상 문자열을 이용한다. 이러한 문자열 쌍을 신경망에서 처리할 수 있는 배치 데이터 텐서로 변환한다(collate 함수). 
from torch.nn.utils.rnn import pad_sequence

# 문장을 텐서로 변환하기 위한 헬퍼 함수
def sequential_transforms(*transforms):
    def func(txt_input):
        for transform in transforms:
            txt_input = transform(txt_input)
        return txt_input
    return func

# BOS, EOS 추가. 텐서 변환
def tensor_transform(token_ids: List[int]):
    return torch.cat((torch.tensor([BOS_IDX]),
                      torch.tensor(token_ids),
                      torch.tensor([EOS_IDX])))

# 소스, 타겟 언어 문장을 텐서로 변환
text_transform = {}
for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
    text_transform[ln] = sequential_transforms(token_transform[ln], # 문장 토큰화
                                               vocab_transform[ln], # 토큰 수치화
                                               tensor_transform) # BOS/EOS 추가. 텐서 생성

# 학습용 배치 데이터셋 생성. 소스-대상 문장 토큰 대상.
def collate_fn(batch):
    src_batch, tgt_batch = [], []
    for src_sample, tgt_sample in batch:
        src_batch.append(text_transform[SRC_LANGUAGE](src_sample.rstrip("\n")))
        tgt_batch.append(text_transform[TGT_LANGUAGE](tgt_sample.rstrip("\n")))

    src_batch = pad_sequence(src_batch, padding_value=PAD_IDX)
    tgt_batch = pad_sequence(tgt_batch, padding_value=PAD_IDX)
    return src_batch, tgt_batch

다음과 같이 학습 함수를 정의한다.
from torch.utils.data import DataLoader

def train_epoch(model, optimizer): # 학습 함수
    model.train()
    losses = 0
    train_iter = Multi30k(split='train', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    train_dataloader = DataLoader(train_iter, batch_size=BATCH_SIZE, collate_fn=collate_fn)

    for src, tgt in train_dataloader:
        src = src.to(DEVICE)
        tgt = tgt.to(DEVICE)
        tgt_input = tgt[:-1, :]

        src_mask, tgt_mask, src_padding_mask, tgt_padding_mask = create_mask(src, tgt_input) # 마스크 
        logits = model(src, tgt_input, src_mask, tgt_mask,src_padding_mask, tgt_padding_mask, src_padding_mask)

        optimizer.zero_grad() 
        tgt_out = tgt[1:, :]
        loss = loss_fn(logits.reshape(-1, logits.shape[-1]), tgt_out.reshape(-1))  # 손실 계산
        loss.backward()   # 역전파

        optimizer.step()
        losses += loss.item()

    return losses / len(list(train_dataloader))

def evaluate(model):  # 평가 함수
    model.eval()
    losses = 0

    val_iter = Multi30k(split='valid', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    val_dataloader = DataLoader(val_iter, batch_size=BATCH_SIZE, collate_fn=collate_fn)

    for src, tgt in val_dataloader:
        src = src.to(DEVICE)
        tgt = tgt.to(DEVICE)
        tgt_input = tgt[:-1, :]
        src_mask, tgt_mask, src_padding_mask, tgt_padding_mask = create_mask(src, tgt_input)

        logits = model(src, tgt_input, src_mask, tgt_mask,src_padding_mask, tgt_padding_mask, src_padding_mask)

        tgt_out = tgt[1:, :]
        loss = loss_fn(logits.reshape(-1, logits.shape[-1]), tgt_out.reshape(-1))
        losses += loss.item()

    return losses / len(list(val_dataloader))

이제 준비된 데이터로 학습한다.
from timeit import default_timer as timer
NUM_EPOCHS = 18

for epoch in range(1, NUM_EPOCHS+1):
    start_time = timer()
    train_loss = train_epoch(transformer, optimizer)  # 에포크 만큼 학습
    end_time = timer()
    val_loss = evaluate(transformer)
    print((f"Epoch: {epoch}, Train loss: {train_loss:.3f}, Val loss: {val_loss:.3f}, "f"Epoch time = {(end_time - start_time):.3f}s"))

# function to generate output sequence using greedy algorithm
def greedy_decode(model, src, src_mask, max_len, start_symbol):
    src = src.to(DEVICE)
    src_mask = src_mask.to(DEVICE)

    memory = model.encode(src, src_mask)
    ys = torch.ones(1, 1).fill_(start_symbol).type(torch.long).to(DEVICE)
    for i in range(max_len-1):
        memory = memory.to(DEVICE)
        tgt_mask = (generate_square_subsequent_mask(ys.size(0))
                    .type(torch.bool)).to(DEVICE)
        out = model.decode(ys, memory, tgt_mask)
        out = out.transpose(0, 1)
        prob = model.generator(out[:, -1])
        _, next_word = torch.max(prob, dim=1)
        next_word = next_word.item()

        ys = torch.cat([ys,
                        torch.ones(1, 1).type_as(src.data).fill_(next_word)], dim=0)
        if next_word == EOS_IDX:
            break
    return ys

# 실제 입력에 대한 대상으로 번역(변환)
def translate(model: torch.nn.Module, src_sentence: str):
    model.eval()
    src = text_transform[SRC_LANGUAGE](src_sentence).view(-1, 1)
    num_tokens = src.shape[0]
    src_mask = (torch.zeros(num_tokens, num_tokens)).type(torch.bool)
    tgt_tokens = greedy_decode(
        model,  src, src_mask, max_len=num_tokens + 5, start_symbol=BOS_IDX).flatten()
    return " ".join(vocab_transform[TGT_LANGUAGE].lookup_tokens(list(tgt_tokens.cpu().numpy()))).replace("<bos>", "").replace("<eos>", "")

학습 결과 다음과 같이 자연어 번역 기능이 제대로 동작하는 것을 확인할 수 있다. 참고로, 학습에는 NVIDIA GPU RAM 7G 정도가 사용되었으며, 학습 시간은 한 시간 내에 처리된다. 
print(translate(transformer, "Eine Gruppe von Menschen steht vor einem Iglu ."))
> A group of people stand in front of an igloo .

실행 결과

마무리
토치에서 제공되는 트랜스포머 코드를 확인해 보면, 앞서 설명된 트랜스포머 내부 코드 구현과 거의 유사한 것을 알 수 있다. 실행해보면, 계산 속도가 좀 더 빠르고, 좀 더 많은 양의 데이터를 학습해서 결과다 더 나은 것을 알 수 있다.

트랜스포머가 자연어 번역 이외에 멀티모달의 핵심 기술이 된 이유는 비정형 데이터를 연산 가능한 차원으로 수치화할 수 있는 임베딩 기술의 발전과 트랜스포머의 Key, Query, Value 입력을 통한 여러 학습 데이터 조합이 가능한 특징이 크게 작용했다. 
트랜스포머 기반 멀티모달 구현 전략 개념도

멀티모달 시작을 알린 OpenAI의 CLIP 모델(Learning Transferable Visual Models From Natural Language Supervision, 2021)

다음은 트랜스포머와 VAE를 이용한 CLIP 멀티모달 네트웍을 좀 더 깊게 파헤쳐 보도록 한다. 

레퍼런스
작성자: 시간:
라벨:

댓글 없음:

댓글 쓰기

피드 구독하기: 댓글 (Atom)