Lab11-5 RNN seq2seq

부스트코스를 보면서 따라했다!!

www.boostcourse.org/ai214/lecture/43778/?isDesc=false

파이토치로 시작하는 딥러닝 기초

 

 

Sequence를 입력받고 Sequence로 출력하는 것을 다룬다.

 

Seq2Seq

[RNN과 Seq2Seq의 차이]

 

ex. seq2seq가 잘 적용된 것 : chatbot 

- RNN : Today's perfect weather makes me much sad.라는 문장을 보고 한 단어씩 추출하여 답을 생성한다면, 이미 위로랑은 거리가 먼 문장을 생성해낼 것이다. 끝에서 살짝 변하는 말에 대해 대응하지 못한다.(한국어는 끝까지 들어봐야 아는데,,,)

- seq2seq : encoder과 decoder의 구조를 가진다. 뒤의 Apply seq2seq에서의 설명을 유심히 보자.

 

Apply Seq2Seq

- Encoder-Decoder

seq2seq는 encoder와 decoder로 구성되어 있다.

encoder : 입력된 sequence를 벡터의 형태로 압축하여 decoder로 전달한다.

decoder : encoder에서 나온 것을 첫 번째 시작하는 부분에 start flag와 함께 넣어주고, 결과값이 나오면 output은 답변으로도 역할을 수행하지만 hidden state로도 넘어간다.

 

이는 깃허브에서 200줄을 볼 수 있지만 그중 10줄 정도 중요한 친구들만 강의에서 찝어주셨다.

(github.com/deeplearningzerotoall/PyTorch/blob/master/lab-11_5_seq2seq.ipynb)

Source Code test는 영어로 되어 있지만, 이를 target text에서 한국어 문장으로 번역하는 모델이다.

 

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

# declare max length for sentence
SOURCE_MAX_LENGTH = 10
TARGET_MAX_LENGTH = 12

# preprocess the corpus
# source, target// train, test data을 Preprocess에서 나눈다.
load_pairs, load_source_vocab, load_target_vocab = preprocess(raw, SOURCE_MAX_LENGTH, TARGET_MAX_LENGTH)
print(random.choice(load_pairs))

# declare the encoder and the decoder
# enccoder hidden size와 decoder hidden size는 같게 설정해준다.
enc_hidden_size = 16
dec_hidden_size = enc_hidden_size
enc = Encoder(load_source_vocab.n_vocab, enc_hidden_size).to(device)
dec = Decoder(dec_hidden_size, load_target_vocab.n_vocab).to(device)

# train seq2seq model
# Encoder에서 나온 것을 Decoder에 넣는 코드를 포함하고 있을 것이다.
train(load_pairs, load_source_vocab, load_target_vocab, enc, dec, 5000, print_every=1000)


# check the model with given data
evaluate(load_pairs, load_source_vocab, load_target_vocab, enc, dec, TARGET_MAX_LENGTH)

 

code를 차근차근 살펴보자.

 

- Data Preprocessing

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

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


# 데이터를 직접 선언한 경우이다. 영어 tab 한국어 구조
raw = ["I feel hungry.	나는 배가 고프다.",
       "Pytorch is very easy.	파이토치는 매우 쉽다.",
       "Pytorch is a framework for deep learning.	파이토치는 딥러닝을 위한 프레임워크이다.",
       "Pytorch is very clear to use.	파이토치는 사용하기 매우 직관적이다."]
       
# fix token for "start of sentence" and "end of sentence"
# decoder의 마지막 출력물을 받아오고 첫번째 step의 index를 받아온다.
SOS_token = 0 
EOS_token = 1 # 문장의 종료를 알려주는 것

# read and preprocess the corpus data
# Preprocessing은 Source text와 target text를 나누어 가진다.
def preprocess(corpus, source_max_length, target_max_length):
    print("reading corpus...")
    pairs = []
    for line in corpus:
        # tab으로 구분되어 있는 것을 나누어 가진다.
        pairs.append([s for s in line.strip().lower().split("\t")])
    print("Read {} sentence pairs".format(len(pairs)))

    pairs = [pair for pair in pairs if filter_pair(pair, source_max_length, target_max_length)]
    print("Trimmed to {} sentence pairs".format(len(pairs)))

    # Vocab()을 생성하여 단어 사전 생성
    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

 

 

- Neural Net Setting

# declare simple encoder
class Encoder(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(Encoder, self).__init__()
        self.hidden_size = hidden_size
        # embedding은 거대한 행렬같다. 
        # Source에서 사용되고 있는 단어의 개수에서 hidden size 만큼의 vector로 줄인다.
        # 가로로 긴 vector에 세로로 긴 one-hot encoding이 곱해져서 줄어든 vector가 나온 후, 이를 gru에서 처리한다.
        self.embedding = nn.Embedding(input_size, hidden_size)
        # 들어오는 dimension과 나오는 dimension만 정해주면 된다.
        self.gru = nn.GRU(hidden_size, hidden_size)

    # encoder구조를 알 수 있다.
    def forward(self, x, hidden):
        # embedding을 통과해서 나오고 , 이것이 gru에 input으로 들어가서 나온다.
        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)
        # decoder part의 gru
        self.gru = nn.GRU(hidden_size, hidden_size)
        # encoder와 유사하지만, encoder만큼의 차원으로 복원하는 과정을 수행한다.
        # hidden state로 나온 것을 target에 맞추어 그만큼 복원해주어야 한다.
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.LogSoftmax(dim=1)

    def forward(self, x, hidden):
        x = self.embedding(x).view(1, 1, -1)
        x, hidden = self.gru(x, hidden)
        x = self.softmax(self.out(x[0]))
        return x, hidden

 

 

- Training

# convert sentence to the index tensor
# sentence를 입력 받아서 onehot encoding을 하고 이를 tensor로 바꿔주는 역할을 한다.
def tensorize(vocab, sentence):
    indexes = [vocab.vocab2index[word] for word in sentence.split(" ")]
    indexes.append(vocab.vocab2index["<EOS>"])
    return torch.Tensor(indexes).long().to(device).view(-1, 1)
    
# training seq2seq
def train(pairs, source_vocab, target_vocab, encoder, decoder, n_iter, print_every=1000, learning_rate=0.01):
    loss_total = 0

    encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
    decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)

    training_batch = [random.choice(pairs) for _ in range(n_iter)]
    training_source = [tensorize(source_vocab, pair[0]) for pair in training_batch]
    training_target = [tensorize(target_vocab, pair[1]) for pair in training_batch]

    # category vector를 비교할 때 많이 사용되는 loss이다.
    # Negative Log Likelihood Loss
    # 마지막에 nn.LogSoftmax 를 해주어야한다.
    criterion = nn.NLLLoss()

    
    for i in range(1, n_iter + 1):
        source_tensor = training_source[i - 1]
        target_tensor = training_target[i - 1]

        # hidden state
        encoder_hidden = torch.zeros([1, 1, encoder.hidden_size]).to(device)

        encoder_optimizer.zero_grad()
        decoder_optimizer.zero_grad()

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

        loss = 0

        
        for enc_input in range(source_length):
            # encoder의 hiddeen state를 꺼내온다.
            _, encoder_hidden = encoder(source_tensor[enc_input], encoder_hidden)

        # encoder 의 마지막 hidden state를 decoder의 첫 hidden state에 넣어준다.
        # Start_Of_Token도 입력값으로 들어간다.
        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 = decoder(decoder_input, decoder_hidden)
            loss += criterion(decoder_output, target_tensor[di])]
            # GRU에서 나온 값을 넣어줄 수도 있지만 실제 값을 넣어줘서 훈련하는 방법이 있다.
            # 후자는 빠르지만, 네트워크가 불안정할 수 있다.
            decoder_input = target_tensor[di]  # teacher forcing

        loss.backward()

        encoder_optimizer.step()
        decoder_optimizer.step()

        loss_iter = loss.item() / target_length
        loss_total += loss_iter

        if i % print_every == 0:
            loss_avg = loss_total / print_every
            loss_total = 0
            print("[{} - {}%] loss = {:05.4f}".format(i, i / n_iter * 100, loss_avg))

 

 

 

- Evaluation

# insert given sentence to check the training
def evaluate(pairs, source_vocab, target_vocab, encoder, decoder, target_max_length):
    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, encoder.hidden_size]).to(device)

        for ei in range(source_length):
            _, encoder_hidden = encoder(source_tensor[ei], encoder_hidden)

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

        for di in range(target_max_length):
            decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
            _, 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("")