Seq2Seq 문장번역

파이토치 Seq2Seq 예제

import random
import torch
import torch.nn as nn
import torch.optim as optim

torch.manual_seed(0)
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

raw = ['I called Tom for help.	나는 톰에게 도움을 요청했다.',
'I do not like science.	나는 과학이 싫어.',
'I hate myself as well.	나도 내 자신을 싫어해.',
'I knew Tom would lose.	톰이 질 거라는 것을 난 알고 있었어.',
'I know Tom personally.	난 톰을 개인적으로 알고 있어.',
'I like Korean cuisine.	전 한국 요리가 좋아요.',
'I like Korean cuisine.	전 한국 요리를 좋아해요.',
'I like helping others.	나는 남을 돕는 것을 좋아한다.',
'I really like puppies.	저는 강아지가 정말 좋아요.',
'I run faster than Tom.	나는 톰보다 빠르게 달릴 수 있어.',
'I think Tom is lonely.	톰이 외로워하는 것 같아.',
'I think they like you.	그들이 널 좋아하는 것 같아.',
'I want to go to sleep.	나 자러 가고 싶어.',
'I want to go to sleep.	나 자고 싶어.',
'I want to visit Korea.	나는 한국에 들르고 싶다.']

사용한 데이터는 http://www.manythings.org/anki/ 에서 kor-eng.zip 파일을 다운로드 받아 일부 데이터만 사용했습니다. 해당 사이트에 들어가면 한국어 외에도 다양한 형태의 파일을 다운 받을 수 있습니다.

SOS_token = 0 # 문장의 시작 Start of Sentence
EOS_token = 1 #  문장의 끝 End of Sentence

class Vocab:
    def __init__(self):
        self.vocab2index = {"<SOS>":SOS_token, "<EOS>":EOS_token}
        self.index2vocab = {SOS_token:"<SOS>", EOS_token:"<SOS>"}
        self.vocab_count = {}
        self.n_vocab = len(self.vocab2index)
    
    def add_vocab(self, sentence):
        for word in sentence.split(' '):
            if word not in self.vocab2index:
                self.vocab2index[word] = self.n_vocab
                self.vocab_count[word] = 1
                self.index2vocab[self.n_vocab] = word
                self.n_vocab += 1
            else:
                self.vocab_count[word] += 1

# declare simple encoder
class Encoder(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(Encoder, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(input_size, hidden_size) # Embedding(17, 16)
        self.gru = nn.GRU(hidden_size, hidden_size)

    def forward(self, x, hidden):
        x = self.embedding(x).view(1, 1, -1)
        x, hidden = self.gru(x, hidden)
        return x, hidden

    
# declare simple decoder
class Decoder(nn.Module):
    def __init__(self, hidden_size, output_size):
        super(Decoder, self).__init__()
        self.hidden_size = hidden_size
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size, num_layers=1, batch_first=True)
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.Softmax(dim=1)

    def forward(self, x, hidden):
        x = self.embedding(x).view(1, 1, -1)
        x, hidden = self.gru(x, hidden) # lstm을 사용할 경우 해당 위치 수정
        x = self.softmax(self.out(x[0]))
        return x, hidden

# read and preprocess the corpus data
def preprocess(corpus):
    print("reading corpus...")
    pairs = []
    for line in corpus:
        pairs.append([s for s in line.strip().lower().split("\t")])
    print("Read {} sentence pairs".format(len(pairs)))

    pairs = [pair for pair in pairs]
    print("Trimmed to {} sentence pairs".format(len(pairs)))

    source_vocab = Vocab()
    target_vocab = Vocab()

    print("Counting words...")
    for pair in pairs:
        source_vocab.add_vocab(pair[0])
        target_vocab.add_vocab(pair[1])
    print("source vocab size =", source_vocab.n_vocab)
    print("target vocab size =", target_vocab.n_vocab)

    return pairs, source_vocab, target_vocab

# 데이터셋, 입력단어정보, 출력단어정보
pairs, source_vocab, target_vocab = preprocess(raw)

훈련용 입출력 데이터셋을 위와 같이 만든후 이제 인코더, 디코더 모델을 만들어야 합니다. 먼저 만들기 전에 인코더-디코더의 입출력 정보에 대하여 직접 그림으로 그려보시기를 추천합니다. 가장 좋은 것은 노트에 펜으로 그려보시는 것이 좋겠지만 그렇지 않다면 머리속으로 어떤 입력이 들어오고 어떤 출력이 나가는지에 대한 정보를 설계하는 과정이 필요합니다.

이런 과정이 없으면 나중에 인코더와 디코더를 설계할 때에 혼동하기 쉽기 때문에 반드시 모델의 입출력 흐름을 구상해보시기 바랍니다.

본 예제의 인코더-디코더 정보는 다음과 같습니다.
인코더 : input_vector(41) -> Embedding(41,30) -> LSTM(30,30)
디코더 : Embedding(52,30) -> LSTM(30, 52) – hidden_vector(52)

enc_hidden_size = 30
dec_hidden_size = enc_hidden_size
enc = Encoder(source_vocab.n_vocab, enc_hidden_size).to(device)
dec = Decoder(dec_hidden_size, target_vocab.n_vocab).to(device)

def tensorize(vocab, sentence):
    idx = [vocab.vocab2index[word] for word in sentence.lower().split(' ')]
    idx.append(vocab.vocab2index['<EOS>'])
    return torch.Tensor(idx).long().to(device).view(-1,1)

tensorize(source_vocab, 'I called Tom for help.')
output : tensor([[2], [3], [4], [5], [6], [1]])

training_source = [tensorize(source_vocab, pair[0]) for pair in pairs]
training_target = [tensorize(target_vocab, pair[1]) for pair in pairs]

Train

loss_total = 0
number_epoch = 5001

encoder_optimizer = optim.SGD(enc.parameters(), lr=0.01)
decoder_optimizer = optim.SGD(dec.parameters(), lr=0.01)

criterion = nn.NLLLoss()

for epoch in range(number_epoch):
    epoch_loss = 0
    
    for i in range(len(training_source)):
        encoder_optimizer.zero_grad()
        decoder_optimizer.zero_grad()
        
        source_tensor = training_source[i]
        target_tensor = training_target[i]

        encoder_hidden = torch.zeros([1, 1, enc.hidden_size]).to(device)

        source_length = source_tensor.size(0)
        target_length = target_tensor.size(0)
        
        loss = 0

        for enc_input in range(source_length):
            _, encoder_hidden = enc(source_tensor[enc_input], encoder_hidden)

        decoder_input = torch.Tensor([[SOS_token]]).long().to(device)
        decoder_hidden = encoder_hidden # connect encoder output to decoder input

        for di in range(target_length):
            decoder_output, decoder_hidden = dec(decoder_input, decoder_hidden)
            #print(decoder_output, target_tensor[di], criterion(decoder_output, target_tensor[di]))
            loss += criterion(decoder_output, target_tensor[di])
            decoder_input = target_tensor[di]  # teacher forcing
        
        loss.backward()

        encoder_optimizer.step()
        decoder_optimizer.step()
        
        #print(loss.item(),target_length)
        epoch_loss += loss.item()/target_length
        #loss_total += loss_epoch
    if epoch % 100 == 0:
        print('--- epoch {}, total loss {} '.format(epoch,float(epoch_loss/15)))

Evaluate

for pair in pairs:
    print(">", pair[0])
    print("=", pair[1])
    source_tensor = tensorize(source_vocab, pair[0])
    source_length = source_tensor.size()[0]
    encoder_hidden = torch.zeros([1, 1, enc.hidden_size]).to(device)

    for ei in range(source_length):
        _, encoder_hidden = enc(source_tensor[ei], encoder_hidden)
        #print(encoder_hidden.size()) # 1,1,16

    decoder_input = torch.Tensor([[SOS_token]], device=device).long()
    decoder_hidden = encoder_hidden
    decoded_words = []

    for di in range(20):
        decoder_output, decoder_hidden = dec(decoder_input, decoder_hidden)
        #print('decoder_iput',decoder_input, 'decoder_output',decoder_output)
        _, top_index = decoder_output.data.topk(1)
        if top_index.item() == EOS_token:
            decoded_words.append("<EOS>")
            break
        else:
            decoded_words.append(target_vocab.index2vocab[top_index.item()])

        decoder_input = top_index.squeeze().detach()

    predict_words = decoded_words
    predict_sentence = " ".join(predict_words)
    print("<", predict_sentence)
    print("")

> i called tom for help.
= 나는 톰에게 도움을 요청했다.
< 나는 톰에게 도움을 요청했다. <EOS>

> i do not like science.
= 나는 과학이 싫어.
< 나는 과학이 싫어. <EOS>

> i hate myself as well.
= 나도 내 자신을 싫어해.
< 나도 내 자신을 싫어해. <EOS>

> i knew tom would lose.
= 톰이 질 거라는 것을 난 알고 있었어.
< 톰이 질 거라는 것을 난 알고 있었어. <EOS>

> i know tom personally.
= 난 톰을 개인적으로 알고 있어.
< 난 톰을 개인적으로 알고 있어. <EOS>

> i like korean cuisine.
= 전 한국 요리가 좋아요.
< 전 한국 요리를 좋아해요. <EOS>

> i like korean cuisine.
= 전 한국 요리를 좋아해요.
< 전 한국 요리를 좋아해요. <EOS>

> i like helping others.
= 나는 남을 돕는 것을 좋아한다.
< 나는 남을 돕는 것을 좋아한다. <EOS>

> i really like puppies.
= 저는 강아지가 정말 좋아요.
< 저는 강아지가 정말 좋아요. <EOS>

> i run faster than tom.
= 나는 톰보다 빠르게 달릴 수 있어.
< 나는 톰보다 빠르게 달릴 수 있어. <EOS>

> i think tom is lonely.
= 톰이 외로워하는 것 같아.
< 톰이 외로워하는 것 같아. <EOS>

> i think they like you.
= 그들이 널 좋아하는 것 같아.
< 그들이 널 좋아하는 것 같아. <EOS>

> i want to go to sleep.
= 나 자러 가고 싶어.
< 나 자고 싶어. <EOS>

> i want to go to sleep.
= 나 자고 싶어.
< 나 자고 싶어. <EOS>

> i want to visit korea.
= 나는 한국에 들르고 싶다.
< 나는 한국에 들르고 싶다. <EOS>