Transformer 및 Torchtext를 사용한 Seq2Seq 모델
출처
https://pytorch.org/tutorials/beginner/transformer_tutorial.html#sequence-to-sequence-modeling-with-nn-transformer-and-torchtext
Transformer 모델은 더 많은 병렬화가 가능하지만 많은 Seq2Seq 문제에서 품질이 우수하다는 것이 입증되었습니다.
nn.Transformer 모듈은 입력과 출력 간의 전역 종속성을 그리기 위해 Attention 메커니즘(nn.MultiheadAtention으로 구현 된 다른 모듈)에 전적으로 의존합니다.
모델 정의
nn.TransformerEncoder 언어 모델링 업무에 대한 모델을 학습 합니다. 언어 모델링 작업은 주어진 단어(또는 단어 시퀀스)가 단어 시퀀스를 따를 가능성을 할당하는 것입니다. 일련의 토큰이 먼저 임베드 레이어로 전달되고 단어 순서를 설명하기 위해 위치 인코딩 레이어가 이어집니다. nn.TransformerEncoderLayer는 nn.TransformerEncoder 여러 계층으로 구성됩니다. 입력 시퀀스와 함께 자체 레이어 nn.TransformerEncoder는 시퀀스의 이전 위치에만 참석할 수 있으므로ㅗ 사각형 주의 마스크가 필요합니다. 언어 모델링 작업의 경우 향후 위치에 있는 모든 토큰을 마스크해야합니다. 실제 단어를 가지려면 nn.TransformerEncoder 모델은 최종 선형 레이어로 전송되고 그 뒤에 log-Softmax 함수가 나타납니다.
import math
import torch
import torch.nn as nn
import torch.nn.functional as F
class TransformerModel(nn.Module):
def __init__(self, ntoken, ninp, nhead, nhid, nlayers, dropout=0.5):
super(TransformerModel, self).__init__()
from torch.nn import TransformerEncoder, TransformerEncoderLayer
self.model_type = 'Transformer'
self.src_mask = None
self.pos_encoder = PositionalEncoding(ninp, dropout)
encoder_layers = TransformerEncoderLayer(ninp, nhead, nhid, dropout)
self.transformer_encoder = TransformerEncoder(encoder_layers, nlayers)
self.encoder = nn.Embedding(ntoken, ninp)
self.ninp = ninp
self.decoder = nn.Linear(ninp, ntoken)
self.init_weights()
def _generate_square_subsequent_mask(self, sz):
mask = (torch.triu(torch.ones(sz, sz)) == 1).transpose(0, 1)
mask = mask.float().masked_fill(mask == 0, float('-inf')).masked_fill(mask == 1, float(0.0))
return mask
def init_weights(self):
initrange = 0.1
self.encoder.weight.data.uniform_(-initrange, initrange)
self.decoder.bias.data.zero_()
self.decoder.weight.data.uniform_(-initrange, initrange)
def forward(self, src):
if self.src_mask is None or self.src_mask.size(0) != len(src):
device = src.device
mask = self._generate_square_subsequent_mask(len(src)).to(device)
self.src_mask = mask
src = self.encoder(src) * math.sqrt(self.ninp)
src = self.pos_encoder(src)
output = self.transformer_encoder(src, self.src_mask)
output = self.decoder(output)
return output
PositionalEncoding 모듈은 시퀀스에서 토큰의 상대 또는 절대 위치에 대한 정보를 주입합니다. 위치 인코딩은 포함과 동일한 차원을 가지므로 둘을 합칠 수 있습니다. Sine, Cosine 주파수 기능을 사용 합니다.
class PositionalEncoding(nn.Module):
def __init__(self, d_model, dropout=0.1, max_len=5000):
super(PositionalEncoding, self).__init__()
self.dropout = nn.Dropout(p=dropout)
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
pe = pe.unsqueeze(0).transpose(0, 1)
self.register_buffer('pe', pe)
def forward(self, x):
x = x + self.pe[:x.size(0), :]
return self.dropout(x)코드로 알아보는 PositionalEncoding.
출처 : https://towardsdatascience.com/how-to-code-the-transformer-in-pytorch-24db27c8f9ec
모델이 문장을 이해하기 위해서는 각 단어에 대해 두가지를 알아야합니다.
- 단어는 무엇을 의미합니까?
- 문장에서 그 위치는 무엇입니까?
각 단어에 대한 임베딩 벡터는 의미를 배우므로 이제 단어의 위치를 네트워크에 알려주는 것을 입력해야합니다.
일정한 위치 별 값을 생성하기 위해 다음 함수를 사용하여 이문제에 답변했습니다.
이 상수는 2차원 행렬입니다. Pos 는 문장의 순서를 나타내며 i는 임베딩 벡터 차원의 위치를 나타냅니다.
그런 다음 pos / i 행렬의 각 값은 위의 방정식을 사용하여 계산됩니다.
위치 인코딩 매트릭스는 그 값이 상기 방정식에 의해 정의되는 상수이다. 임베딩 매트릭스에 추가될 때, 각각의 워드 임베딩은 그 위치에 특정한 방식으로 변경된다.
class PositionalEncoder(nn.Module):
def __init__(self, d_model, max_seq_len = 80):
super().__init__()
self.d_model = d_model
# create constant 'pe' matrix with values dependant on
# pos and i
pe = torch.zeros(max_seq_len, d_model)
for pos in range(max_seq_len):
for i in range(0, d_model, 2):
pe[pos, i] = \
math.sin(pos / (10000 ** ((2 * i)/d_model)))
pe[pos, i + 1] = \
math.cos(pos / (10000 ** ((2 * (i + 1))/d_model)))
pe = pe.unsqueeze(0)
self.register_buffer('pe', pe)
def forward(self, x):
# make embeddings relatively larger
x = x * math.sqrt(self.d_model)
#add constant to embedding
seq_len = x.size(1)
x = x + Variable(self.pe[:,:seq_len], \
requires_grad=False).cuda()
return x위와 같이 모듈을 사용하면 임베딩 벡터에 위치 인코딩을 추가하여 모델에 대한 구조 정보를 제공 할 수 있습니다.
추가하기 전에 임베딩 값을 증가시키는 이유는 위치 인코딩을 상대적으로 작게하기 위해서입니다.
즉, 임베딩 벡터의 원래 의미를 함께 추가해도 손실되지 않습니다.
로드 및 배치 데이터
Wikitext-2 데이터 세트를 사용 torchtext 합니다. vocab 객체는 train 데이터 셋을 기반으로하며 토큰을 텐서로 수치화하는 데 사용됩니다. 순차 데이터에서 시작 batchify() 이 함수는 데이터 세트를 열로 배열하여 데이터 크기 단위로 나눈 후 남은 토큰을 제거합니다. batch_size이 함수는 알파벳을 시퀀스(총 길이 26)와 배치 크기가 4인 경우 알파벳을 길이가 4인 4 시퀀스로 나눕니다.
이 열은 모델에 의해 독립적으로 취급되므로 의존도 G 및 F 학습이 불가능하지만 보다 효율적인 일괄 처리가 가능합니다.
import torchtext
from torchtext.data.utils import get_tokenizer
TEXT = torchtext.data.Field(tokenize=get_tokenizer("basic_english"),
init_token='<sos>',
eos_token='<eos>',
lower=True)
train_txt, val_txt, test_txt = torchtext.datasets.WikiText2.splits(TEXT)
TEXT.build_vocab(train_txt)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
def batchify(data, bsz):
data = TEXT.numericalize([data.examples[0].text])
# numericalize = 번호순으로 정렬하다
# Divide the dataset into bsz parts.
nbatch = data.size(0) // bsz
# Trim off any extra elements that wouldn't cleanly fit (remainders).
data = data.narrow(0, 0, nbatch * bsz)
# Evenly divide the data across the bsz batches.
data = data.view(bsz, -1).t().contiguous()
return data.to(device)
batch_size = 20
eval_batch_size = 10
train_data = batchify(train_txt, batch_size)
val_data = batchify(val_txt, eval_batch_size)
test_data = batchify(test_txt, eval_batch_size)
Error : No module torchtext
Trochtext에 파일이 없다면 아래의 명령어를 입력해주세요.
그런 다음 다시 위에 코드를 실행시켜 줍니다.
!pip install torchtext예제를 통한 이해 <Field>
출처 : https://tutorials.pytorch.kr/beginner/torchtext_translation_tutorial.html
torchtext에는 언어 변환 모델을 만들 때 쉽게 사용할 수 있는 데이터 셋을 만들기 적합한 다양한 도구가 있습니다. 그 중에서도 중요한 클래스 중 하나인 Field는 각 문장이 어떻게 전처리되어야 하는지 지정합니다.
Tensor로 변환하기 위한 지시 사항과 함께 데이터 유형을 정의합니다.
필드 클래스는 텐서로 표시 할 수있는 공통 텍스트 처리 데이터 유형을 모델링합니다. 필드의 요소에 대한 가능한 값 세트와 해당 숫자 표시를 정의하는 Vocab 오브젝트를 보유합니다. Field 객체는 또한 토큰화 방법 및 생성해야하는 Tensor의 종류와 같이 데이터 유형을 수치화하는 방법과 관련된 다른 매개 변수를 보유합니다.
데이터 세트의 두 열간에 공유되는 경우(QA 데이터 세트의 질문과 답변) 공유 어휘를 갖습니다.
torchtext의 Field에 있는 유용한 기능은 바로 build_vocab method로 각 언어와 연관된 어휘들을 만들어 낼 수 있습니다.
TEXT.vocab.stoi 는 어휘에 해당하는 토큰을 키로, 관련된 색인을 값으로 가지는 사전(dict)이 됩니다.
TEXT.vocab.itos 역시 사전(dict)이지만, 키와 값이 서로 반대입니다.
- stoi - collections.defaultdict 인스턴트에서 트큰 문자열을 숫자 식별자로 매핑합니다.
- itos - 숫자 식별자로 색인화 된 토큰 문자열 목록
입력 및 대상 시퀀스를 생성하는 기능
get_batch() 함수는 변압기 모델의 입력 및 대상 시퀀스를 생성합니다.
bptt : 소스 데이터를 length 청크로 세분화합니다.
Target : 언어 모델링 작업의 경우 모델에는 다음과 같은 단어가 필요합니다. 예를 들어 bptt 값이 2이면 i = 0에 대해 다음 두가지 변수가 나타납니다.
청크는 s 트랜스포머 모델의 치수와 일치하는 치수 0을 따릅니다. 배치 차원 N은 차원 1을 따릅니다.
bptt = 35
def get_batch(source, i):
seq_len = min(bptt, len(source) - 1 - i)
data = source[i:i+seq_len]
target = source[i+1:i+1+seq_len].view(-1)
return data, target
인스턴스 시작
모델은 아래 하이퍼 파라미터로 설정됩니다. vocab 크기는 voczb 객체의 길이와 같습니다.
ntokens = len(TEXT.vocab.stoi) # the size of vocabulary
emsize = 200 # embedding dimension
nhid = 200 # the dimension of the feedforward network model in nn.TransformerEncoder
nlayers = 2 # the number of nn.TransformerEncoderLayer in nn.TransformerEncoder
nhead = 2 # the number of heads in the multiheadattention models
dropout = 0.2 # the dropout value
model = TransformerModel(ntokens, emsize, nhead, nhid, nlayers, dropout).to(device)
모델 실행
CrossEntropyLoss는 손실을 추적하기 위해 적용되며 SGD는 확률 적 구배 하강 방법을 최적화 프로그램으로 구현합니다. 초기 학습 속도는 5.0으로 설정되어 있습니다. StepLR은 에포크를 통해 학습 속도를 조정하는 데 적용됩니다. 훈련 중에 nn.utils.clip_grad_norm_ 함수를 사용하여 폭발을 방지하기 위해 모든 그라디언트를 함께 스케일링합니다.
criterion = nn.CrossEntropyLoss()
lr = 5.0 # learning rate
optimizer = torch.optim.SGD(model.parameters(), lr=lr)
scheduler = torch.optim.lr_scheduler.StepLR(optimizer, 1.0, gamma=0.95)
import time
def train():
model.train() # Turn on the train mode
total_loss = 0.
start_time = time.time()
ntokens = len(TEXT.vocab.stoi)
for batch, i in enumerate(range(0, train_data.size(0) - 1, bptt)):
data, targets = get_batch(train_data, i)
optimizer.zero_grad()
output = model(data)
loss = criterion(output.view(-1, ntokens), targets)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5)
optimizer.step()
total_loss += loss.item()
log_interval = 200
if batch % log_interval == 0 and batch > 0:
cur_loss = total_loss / log_interval
elapsed = time.time() - start_time
print('| epoch {:3d} | {:5d}/{:5d} batches | '
'lr {:02.2f} | ms/batch {:5.2f} | '
'loss {:5.2f} | ppl {:8.2f}'.format(
epoch, batch, len(train_data) // bptt, scheduler.get_lr()[0],
elapsed * 1000 / log_interval,
cur_loss, math.exp(cur_loss)))
total_loss = 0
start_time = time.time()
def evaluate(eval_model, data_source):
eval_model.eval() # Turn on the evaluation mode
total_loss = 0.
ntokens = len(TEXT.vocab.stoi)
with torch.no_grad():
for i in range(0, data_source.size(0) - 1, bptt):
data, targets = get_batch(data_source, i)
output = eval_model(data)
output_flat = output.view(-1, ntokens)
total_loss += len(data) * criterion(output_flat, targets).item()
return total_loss / (len(data_source) - 1)
epoch을 반복합니다. 지금까지 본 Validation loss 가장 좋은 경우 모델을 저장합니다. 각 epoch 이후 학습 속도를 조정합니다.
best_val_loss = float("inf")
epochs = 3 # The number of epochs
best_model = None
for epoch in range(1, epochs + 1):
epoch_start_time = time.time()
train()
val_loss = evaluate(model, val_data)
print('-' * 89)
print('| end of epoch {:3d} | time: {:5.2f}s | valid loss {:5.2f} | '
'valid ppl {:8.2f}'.format(epoch, (time.time() - epoch_start_time),
val_loss, math.exp(val_loss)))
print('-' * 89)
if val_loss < best_val_loss:
best_val_loss = val_loss
best_model = model
scheduler.step()
Test data set 이용 모델 평가
테스트 데이터 세트로 결과를 확인하려면 최상의 모델을 적용합니다.
test_loss = evaluate(best_model, test_data)
print('=' * 89)
print('| End of training | test loss {:5.2f} | test ppl {:8.2f}'.format(
test_loss, math.exp(test_loss)))
print('=' * 89)
+ 내용 설명 및 코드 설명