[NLP/자연어처리/Python] text generation 실습 (transformer 언어 번역)

 

(PyTorch) Text Generation

 

참고 : https://tutorials.pytorch.kr/beginner/translation_transformer.html?highlight=de_core_news_sm 

nn.Transformer 와 torchtext로 언어 번역하기

해당 tutorial을 진행하면서 공부한 내용을 정리해보았다.

tutorial의 목표는 독일어(German)를 영어(English)로 번역하는 모델을 학습시키는 것이다.

 

Transformer 이론에 대한 내용은 아래 게시글을 참고하면 된다.

2023.07.04 - [🤖 ai/natural language processing] - [NLP/자연어처리] 트랜스포머(Transformer)

 

필요한 library를 설치한다

!python -m spacy download de_core_news_sm
!python -m spacy download en_core_web_sm
!pip install torchdata
!pip install portalocker>=2.0.0

 

Dataset을 가져온다

# Multi30K는 Flickr30K 데이터셋의 확장으로, 
# 31,014개의 영어 설명에 대한 독일어 번역과 155,070개의 독일어 설명이 포함되어 있음
from torchtext.datasets import multi30k, Multi30k

# 원본 데이터의 링크가 동작하지 않으므로 데이터셋의 URL을 수정해야 합니다.
# 더 자세한 내용은 https://github.com/pytorch/text/issues/1756#issuecomment-1163664163 을 참고해주세요.
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 = {}

 

Dataset의 Tokenizer를 진행한 내용으로 Vocabulary를 생성한다.

from torchtext.data.utils import get_tokenizer
from torchtext.vocab import build_vocab_from_iterator
from typing import Iterable, List

# Source(de), Target(en)에 대한 Tokenizer 생성
# get_tokenizer (tokenizer function or library, language)
# get_tokenizer('basic_english') : 기본 영어 tokenizer func
# tokenizer library : spacy (문장구조와 분법을 분석), moses (통계적 방법), subword (BPE 알고리즘),
# language = 'de_core_news_sm' : sapcy의 독일어 모델
# language = 'en_core_web_sm' : sapcy의 영어 모델
# language = 'ko_core_news_sm' : spacy의 한글 모델
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) 함수
def yield_tokens(data_iter: Iterable, language: str) -> List[str]: # List[str] type의 값을 반환
    language_index = {SRC_LANGUAGE: 0, TGT_LANGUAGE: 1}
    for data_sample in data_iter:
        yield token_transform[language](data_sample[language_index[language]])

# 특수 기호(symbol)와 인덱스를 정의합니다
UNK_IDX, PAD_IDX, BOS_IDX, EOS_IDX = 0, 1, 2, 3
# 토큰들이 어휘집(vocab)에 인덱스 순서대로 잘 삽입되어 있는지 확인합니다
special_symbols = ['<unk>', '<pad>', '<bos>', '<eos>']

for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
    # 학습용 데이터 반복자(iterator)
    train_iter = Multi30k(split='train', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    #language_pair : Multi30k 데이터셋의 특징이라 할 수 있음. 원본언어, 대상언어 나타냄
    
    # torchtext의 Vocab(어휘집) 객체 생성
    vocab_transform[ln] = build_vocab_from_iterator(yield_tokens(train_iter, ln),
                                                    min_freq=1,
                                                    specials=special_symbols,
                                                    special_first=True)
    #  최소 빈도수가 1이고, 특수 기호가 포함된 어휘집을 구축한다. 특수 기호는 어휘집의 처음에 위치하게 된다

# ``UNK_IDX`` 를 기본 인덱스로 설정합니다. 이 인덱스는 토큰을 찾지 못하는 경우에 반환됩니다.
# 만약 기본 인덱스를 설정하지 않으면 어휘집(Vocabulary)에서 토큰을 찾지 못하는 경우
# ``RuntimeError`` 가 발생합니다.
for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
    vocab_transform[ln].set_default_index(UNK_IDX)

 

Token Embedding 구현하기

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

# 입력 인덱스의 텐서를 해당하는 토큰 임베딩의 텐서로 변환하기 위한 헬퍼 모듈(Module)
class TokenEmbedding(nn.Module):
    def __init__(self, vocab_size: int, emb_size):
        super(TokenEmbedding, self).__init__()

        self.embedding = nn.Embedding(vocab_size, emb_size) 
        # 단어 임베딩을 계산하는 layer, vocab_size x emb_size 행렬로 구성 (단어 개수 x 단어 벡터 크기)
        # Linear와 유사하게 작동. 학습 중에는 embedding matrix의 가중치가 업데이트 됨.
        self.emb_size = emb_size

    def forward(self, tokens: Tensor):
        return self.embedding(tokens.long()) * math.sqrt(self.emb_size)
        #  math.sqrt(self.emb_size) : 임베딩 벡터 스케일링

 

Positional Encoding 구현하기

Positional Encoding

$$PE_{(pos,\ 2i)}=sin(pos/10000^{2i/d_{model}})$$
$$PE_{(pos,\ 2i+1)}=cos(pos/10000^{2i/d_{model}})$$

den = torch.exp(- torch.arange(0, emb_size, 2)* math.log(10000) / emb_size) 을 살펴보면

1. torch.arange(0, emb_size, 2) : 0 ~ emb_size까지 2씩 증가하는 수열 생성
2. - torch.arange(0, emb_size, 2)* math.log(10000) :

    $ 0, -2*\log(10000), -4*\log(10000), ..., $
    $ = -2i*\log(10000) $
3. - torch.arange(0, emb_size, 2)* math.log(10000) / emb_size  :

    $ 0, -2*\log(10000)/\text{emb_size}, -4*\log(10000)/\text{emb_size}, ..., $
    $ = -2i*\log(10000)/\text{emb_size} $
    $ = -2i/\text{emb_size}*\log(10000) $
4. torch.exp(- torch.arange(0, emb_size, 2)* math.log(10000) / emb_size) :

    $e^{-2i/\text{emb_size}*\log(10000)}$
    $ = 10000^{-2i/\text{emb_size}} $
    $ = \frac{1}{10000^{2i/\text{emb_size}}}$
(e의 x * log(y) 제곱은 $y^x$)

 

코드가 굉장히 복잡해보이지만,

결국 pos_embedding[:, 0::2] = torch.sin(pos * den)가 나타내는 것은 $PE_{(pos,\ 2i)}=sin(pos/10000^{2i/d_{model}})$ 이다.

 

# 단어 순서 개념(notion)을 토큰 임베딩에 도입하기 위한 위치 인코딩(positional encoding)을 위한 헬퍼 모듈(Module)
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) 
        # 0부터 maxlen-1까지의 정수로 구성된 1차원 배열을 생성한 후, 
        # >> tensor([   0,    1,    2,  ..., 4997, 4998, 4999])

        # reshape 함수를 사용하여 이 배열을 maxlen x 1 크기의 2차원 배열로 변환
        # >>
        # tensor([[   0],
        # [   1],
        # [   2],
        # ...,
        # [4997],
        # [4998],
        # [4999]])

      
        pos_embedding = torch.zeros((maxlen, emb_size)) # maxlen x emb_size matrix
        pos_embedding[:, 0::2] = torch.sin(pos * den) # 짝수 index에는 sin. PE(pos, 2i) = sin()
        pos_embedding[:, 1::2] = torch.cos(pos * den) # 홀수 index에는 cos. PE(pos, 2i+1) = cos()
        pos_embedding = pos_embedding.unsqueeze(-2)
        # 배열의 마지막 차원 바로 앞에 새로운 차원을 추가함.
        # 이렇게 하면 pos_embedding 배열의 차원이 하나 늘어나게 된다.
        self.register_buffer('pos_embedding', pos_embedding) 
        # buffer : 모델의 상태를 저장하는 tensro, 학습중에 업데이트 되지 않음.

        self.dropout = nn.Dropout(dropout)      

    def forward(self, token_embedding: Tensor):
      #  입력 토큰의 임베딩과 위치 정보를 결합한 표현 (token embedding + positional embedding)
        return self.dropout(token_embedding + self.pos_embedding[:token_embedding.size(0), :])

 

모델 정의하기

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

        #torch.nn.Transformer(d_model=512, nhead=8, num_encoder_layers=6, num_decoder_layers=6, 
        # dim_feedforward=2048, dropout=0.1, activation=<function relu>, custom_encoder=None, custom_decoder=None, 
        # layer_norm_eps=1e-05, batch_first=False, norm_first=False, device=None, dtype=None)
        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.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)
        
        # 디코더의 출력을 타겟 언어의 단어장 크기로 변환하는 데 사용
        self.generator = nn.Linear(emb_size, tgt_vocab_size)

    def forward(self,
                # src, trg 문장
                src: Tensor,
                trg: Tensor,
                # mask tensor
                src_mask: Tensor,
                tgt_mask: Tensor,
                src_padding_mask: Tensor,
                tgt_padding_mask: Tensor,
                memory_key_padding_mask: Tensor):
        #  소스 문장과 타겟 문장에 대한 임베딩을 계산한 후, : self.src_tok_emb(src)
        # 위치 정보를 추가한 후 트랜스포머 모델의 입력으로 사용  : self.positional_encoding()
        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)
        
         # 트랜스포머 모델의 출력은 선형 레이어인 self.generator에 전달되어 최종 출력을 계산하고, 이 값을 반환
        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):
      # memory : 인코더의 출력
        return self.transformer.decoder(self.positional_encoding(
                          self.tgt_tok_emb(tgt)), memory,
                          tgt_mask)

 

마스크 정의하기

#  디코더에서는 현재 위치보다 미래의 정보를 참조하지 못하도록 해야한다 -> look-ahead mask 사용 필요
#  look-ahead mask : 현재 위치보다 미래의 정보를 1로 표시하여, 어텐션 계산에서 무시되도록 한다.
def generate_square_subsequent_mask(sz):
    mask = (torch.triu(torch.ones((sz, sz), device=DEVICE)) == 1).transpose(0, 1)
    # torch.ones((sz, sz), device=DEVICE) : sz x sz 크기의 룩어헤드 마스크를 생성, DEVICE(gpu, cpu..)에서 연산 수행
    # 이때 torch.ones((sz, sz), device=DEVICE)의 결과는 1로 채워진 sz x sz 가 되고
    # == 1 을 적용하면서, true로 채워진 sz x sz 가 된다.

    # torch.triu : 상삼각행렬 생성 (주대각선 아래의 모든 원소가 0인 행렬)
    # tensor([[True, True, True],
    #         [False, True, True],
    #         [False, False, True]])

    # .transpose(0, 1) : 전치
    # tensor([[True, False, False],
    #         [True, True, False],
    #         [True, True, True]])

    mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))
    # masked_fill(mask == 0, float('-inf')) : 0인 위치의 값을 -inf로 
    # masked_fill(mask == 1, float(0.0)) : 1인 위치의 값을 0으로
    return mask


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

    # target mask 생성
    tgt_mask = generate_square_subsequent_mask(tgt_seq_len) 
    # 디코더에서 자기 자신과 그 이전 단어들만을 참고할 수 있도록

    # source mask 생성
    src_mask = torch.zeros((src_seq_len, src_seq_len),device=DEVICE).type(torch.bool)
    # source sequence 길이 x source sequence 길이의 모든 값이 0인 matrix 생성


    # 문장의 빈 부분을 채우기 위해 패딩 토큰이 사용되는데, 이 패딩 토큰은 어텐션 계산에서 무시되어야 한다.
    # 이를 위해 패딩 마스크(padding mask)를 사용한다.
    # Padding이 있는 위치를 True로 바꾸고(실제 데이터는 False), 이를 전치한다.
    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

 

Tensor 묶음 및 Batch 처리 함수 생성하기

SRC_LANGUAGE, TGT_LANGUAGE을 Sqe2Sqe 신경망에서 처리할 수 있도록 tensor 묶음으로 변환해야 한다.

from torch.nn.utils.rnn import pad_sequence

# 순차적인 작업들을 하나로 묶는 헬퍼 함수
# *transforms : token_transform[ln] -> vocab_transform[ln] -> tensor_transform 순차적으로 적용하게 된다.
def sequential_transforms(*transforms):
    def func(txt_input): # 여기서 txt_input은 (text_transform[SRC_LANGUAGE](src_sample.rstrip("\n"))
        for transform in transforms:         
            txt_input = transform(txt_input)
        return txt_input
    return func

# BOS/EOS를 추가하고 입력 순서(sequence) 인덱스에 대한 텐서를 생성하는 함수
def tensor_transform(token_ids: List[int]):
    return torch.cat((torch.tensor([BOS_IDX]),
                      torch.tensor(token_ids),
                      torch.tensor([EOS_IDX])))

# 출발어(src)와 도착어(tgt) 원시 문자열들을 텐서 인덱스로 변환하는 변형(transform)
text_transform = {}
for ln in [SRC_LANGUAGE, TGT_LANGUAGE]:
    text_transform[ln] = sequential_transforms(token_transform[ln], # 토큰화(Tokenization)
                                               vocab_transform[ln], # 수치화(Numericalization)
                                               tensor_transform) # BOS/EOS를 추가하고 텐서를 생성


# 데이터를 텐서로 조합(collate)하는 함수
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"))) # .rstrip("\n") : 오른쪽 끝에서 개행 문자(“\n”)를 제거
        tgt_batch.append(text_transform[TGT_LANGUAGE](tgt_sample.rstrip("\n")))

    # 모든 시퀀스의 길이가 동일하도록 패딩을 적용. 이때 패딩 토큰으로 PAD_IDX가 사용
    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

 

모델 객체 생성 및 Loss Func, Optimizer 정의하기

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: #  매개변수의 차원이 1보다 큰 경우(즉, 가중치 행렬인 경우)에만 초기화를 수행
        nn.init.xavier_uniform_(p) # Xavier 초기화 사용

transformer = transformer.to(DEVICE)

# loss function으로 cross entropy loss를 사용. padding token은 무시
loss_fn = torch.nn.CrossEntropyLoss(ignore_index=PAD_IDX)

# optimizer로 Adam optimizer 사용
optimizer = torch.optim.Adam(transformer.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)
# lr=0.0001 : 학습률(learning rate). 기본값은 0.001
# betas=(0.9, 0.98) : 경사(gradient)와 제곱(square)의 이동 평균(running averages)을 계산하는 데 사용되는 계수들. 
#                     기본값은 (0.9, 0.999)
# eps=1e-9 : 수치 안정성을 향상시키기 위해 분모에 추가되는 작은 상수. 기본값은 1e-8

 

모델 학습 함수 및 평가 함수 정의하기

from torch.utils.data import DataLoader

def train_epoch(model, optimizer):
    model.train() # 모델 학습 모드로 설정

    losses = 0

    # Multi30K dataset의 train 분할을 이용해 train data iterator 생성
    train_iter = Multi30k(split='train', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    # data loader 생성
    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, :]
        # 첫 번째 토큰을 제외한 부분을 출력으로 사용
        tgt_out = tgt[1:, :]

        #“teacher forcing” 기법
        # ex) tgt = ["<sos>", "A", "B", "C", "<eos>"]
        # tgt_input = ["<sos>", "A", "B", "C"]
        # tgt_out = ["A", "B", "C", "<eos>"]
        # 디코더의 첫 번째 시간 단계에서 예측한 출력(tgt_input의 "<sos>" 다음은 뭘까?)은 "A"와 비교되고,
        # 두 번째 시간 단계에서 예측한 출력(tgt_input의 "A" 다음은 뭘까?)은 "B"와 비교할 수 있게 된다.

        # 마스크 생성
        src_mask, tgt_mask, src_padding_mask, tgt_padding_mask = create_mask(src, tgt_input)

        # Gradient 0으로 초기화
        optimizer.zero_grad()

        # Prediction
        logits = model(src, tgt_input, src_mask, tgt_mask,src_padding_mask, tgt_padding_mask, src_padding_mask)
              
        # Loss 계산
        loss = loss_fn(logits.reshape(-1, logits.shape[-1]), tgt_out.reshape(-1))

        # Backward pass (gradient 계산)
        loss.backward()

        # Parameter update
        optimizer.step()

        # 손실 누적
        losses += loss.item()
    return losses / len(list(train_dataloader)) # 평균 loss를 반환


def evaluate(model):
    model.eval()
    losses = 0

     # Multi30K dataset의 valid 분할을 이용해 valid data iterator 생성
    val_iter = Multi30k(split='valid', language_pair=(SRC_LANGUAGE, TGT_LANGUAGE))
    # data loader 생성
    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, :]
        tgt_out = tgt[1:, :]
        
        src_mask, tgt_mask, src_padding_mask, tgt_padding_mask = create_mask(src, tgt_input)
        
        # Prediction
        logits = model(src, tgt_input, src_mask, tgt_mask,src_padding_mask, tgt_padding_mask, src_padding_mask)
        # Loss 계산
        loss = loss_fn(logits.reshape(-1, logits.shape[-1]), tgt_out.reshape(-1))
        # 손실 누적
        losses += loss.item()
    return losses / len(list(val_dataloader)) # 평균 loss를 반환

 

모델 학습하기

NUM_EPOCHS = 18

for epoch in range(1, NUM_EPOCHS+1):
    train_loss = train_epoch(transformer, optimizer)
    val_loss = evaluate(transformer)
    print((f"Epoch: {epoch}, Train loss: {train_loss:.3f}, Val loss: {val_loss:.3f}"))

 

문자열을 번역하는 함수 구현하기

학습이 완료된 모델을 이용해,

source 문자열을 encode에 넣고 decode 출력을 통해 번역 결과를 받아 볼 수 있는 함수를 구현한다.

# 탐욕(greedy) 알고리즘을 사용하여 출력 순서(sequence)를 생성하는 함수
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) # encoder를 사용하여 소스 시퀀스의 메모리(memory)를 계산

    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) # decoder를 사용하여 출력을 계산. 현재까지 생성된 목표 시퀀스(ys), memory, target mask가 전달
        out = out.transpose(0, 1) # 출력 차원 전치

        # generator를 사용해 각 단어의 확률 계산. 
        # out[:, -1]는 마지막 시간 단계에서 예측한 단어의 확률 분포를 나타냄  
        prob = model.generator(out[:, -1])        

        _, next_word = torch.max(prob, dim=1) # 가장 높은 확률을 가진 단어의 인덱스를 선택
        next_word = next_word.item() #  선택된 단어의 인덱스를 스칼라 값으로 변환

        # 목표 시퀀스(ys)에 새로운 단어의 인덱스(next_word)를 추가
        # 새로운 단어의 인덱스는 torch.ones(1, 1).type_as(src.data).fill_(next_word) 코드를 사용하여 텐서로 변환
        ys = torch.cat([ys,
                        torch.ones(1, 1).type_as(src.data).fill_(next_word)], dim=0)
       
        # 선택된 단어가 종료 심볼(EOS)인 경우 반복을 종료
        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) # source 문장 전처리 후에 텐서로 변환
    # view(-1, 1) : -1은 해당 차원의 크기가 자동으로 계산되어야 함을 의미. (n, 1)로 변경된다.

    num_tokens = src.shape[0] # source sequence의 token 수 계산

    src_mask = (torch.zeros(num_tokens, num_tokens)).type(torch.bool) # source mask 생성

    tgt_tokens = greedy_decode(
        model,  src, src_mask, max_len=num_tokens + 5, start_symbol=BOS_IDX).flatten()

    # list(tgt_tokens.cpu().numpy())) : target sequence를 CPU로 이동시킨 후 NumPy 배열로 변환한 후 list로 변환
    # lookup_tokens 메서드를 사용하여 각 토큰의 인덱스를 해당하는 단어로 변환
    return " ".join(vocab_transform[TGT_LANGUAGE].lookup_tokens(list(tgt_tokens.cpu().numpy()))).replace("<bos>", "").replace("<eos>", "")

 

모델 평가하기

print(translate(transformer, "Eine Gruppe von Menschen steht vor einem Iglu ."))

결과 : A group of people stand in front an igloo .

toturial에 주어진 예시를 가지고 진행해보았을 때, 구글 번역기 결과와 동일하게 (of가 빠지긴 했지만) 나오는 것을 확인할 수 있다.

 

다른 예시로도 모델을 평가해보기 위해, 구글에 "독일어 명언"을 검색해 나오는 문장을 모델에 입력해보았다.

 

print(translate(transformer, "Es gibt Berge, über die man hinüber muss, Sonst geht der Weg nicht weiter."))

결과 : A female figure who is about to get to get into the ring from the trail.

 

결과가 이상하다..?

하이퍼파라미터 조정하여 다시 시도!

 

NHEAD = 8 -> 16

 

print(translate(transformer, "Eine Gruppe von Menschen steht vor einem Iglu ."))
print(translate(transformer, "Es gibt Berge, über die man hinüber muss, Sonst geht der Weg nicht weiter."))

결과 : A group of people stand in front of an igloo .

A mountains to make a moment to get to the moment of the trail .

 

독일어 명언의 결과는 여전히 이상하지만, 그래도 처음보다는 나아진 모습을 확인할 수 있다. 

(toturial에 주어진 예시에서는 빠졌던 of가 추가되었다)