Translation with a SeqToSeq Network and Attention - (1)

소스코드 원본 링크 : https://pytorch.org/tutorials/intermediate/seq2seq_translation_tutorial.html

코드의 전반적인 메커니즘을 이해하는데 큰 도움이 되는 자료:
https://medium.com/@jongdae.lim/%EA%B8%B0%EA%B3%84-%ED%95%99%EC%8A%B5-machine-learning-%EC%9D%80-%EC%A6%90%EA%B2%81%EB%8B%A4-part-5-83b7a44b797a
2019-01-12 2 17 24
Sequence to Sequence 네트워크는 두가지 RNN이 작동한다. 첫번째로 인코더 네트워크는 입력 문장(시퀀스)을 벡터로 변환하고, 두번째로 디코더 네트워크는 그 벡터를 unfold 해주고 새로운 문장(시퀀스)로 변환한다.

Import packages and modules

from __future__ import unicode_literals, print_function, division
from io import open
import unicodedata
import string
import re
import random

import torch
import torch.nn as nn
from torch import optim
import torch.nn.functional as F

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

Loading Data Files

init : 임의의 언어를 초기화시키고 생성자 파라미터에 어떤 언어가 입력되었는지 체크하고 변수에 언어 이름을 저장한다. 입력 문장이 들어왔을 때 문장의 구성요소인 단어들에게 인덱스를 부여하고 몇번 중복되었는지 체크한다. SOS는 시작부분 EOS는 끝부분을 지정해주기 위해 만들었다.
addSentence : 입력 문장이 들어왔을 때 공백으로 split 하고 각 단어에 대해 addWord 함수를 실행한다.
addWord : 단어가 들어왔을 때 조건문에 따라 어떤식으로 딕셔너리를 처리할지 결정한다.

SOS_token = 0
EOS_token = 1

class Lang:
    def __init__(self, name):
        self.name = name
        self.word2index = {}
        self.word2count = {}
        self.index2word = {0: "SOS", 1: "EOS"}
        self.n_words = 2  # Count SOS and EOS

    def addSentence(self, sentence):
        for word in sentence.split(' '):
            self.addWord(word)

    def addWord(self, word):
        if word not in self.word2index:
            self.word2index[word] = self.n_words
            self.word2count[word] = 1
            self.index2word[self.n_words] = word
            self.n_words += 1
        else:
            self.word2count[word] += 1

유니코드를 아스키코드로 바꾸고 정규표현식을 사용해서 normalize 하는 코드

# Turn a Unicode string to plain ASCII, thanks to
# https://stackoverflow.com/a/518232/2809427
def unicodeToAscii(s):
    return ''.join(
        c for c in unicodedata.normalize('NFD', s)
        if unicodedata.category(c) != 'Mn'
    )

# Lowercase, trim, and remove non-letter characters

def normalizeString(s):
    s = unicodeToAscii(s.lower().strip())
    s = re.sub(r"([.!?])", r" \1", s)
    s = re.sub(r"[^a-zA-Z.!?]+", r" ", s)
    return s

print(lines[0]) –> Go. Va !
print(lines[1]) –> Run! Cours !
print(pairs[0]) –> [‘go .’, ‘va !’]
print(pairs[1]) –> [‘run !’, ‘cours !’]

def readLangs(lang1, lang2, reverse=False):
    print("Reading lines...")

    # Read the file and split into lines
    lines = open('data/%s-%s.txt' % (lang1, lang2), encoding='utf-8').\
        read().strip().split('\n')

    # Split every line into pairs and normalize
    pairs = [[normalizeString(s) for s in l.split('\t')] for l in lines]

    # Reverse pairs, make Lang instances
    if reverse:
        pairs = [list(reversed(p)) for p in pairs]
        input_lang = Lang(lang2)
        output_lang = Lang(lang1)
    else:
        input_lang = Lang(lang1)
        output_lang = Lang(lang2)

    return input_lang, output_lang, pairs

english 문장이 eng_prefixes에 있는 각 문장들로 시작하고 영어 문장, 프랑스어 문장 각각도 공백으로 split 했을 때 길이가 10보다 작아야 한다.

MAX_LENGTH = 10

eng_prefixes = (
    "i am ", "i m ",
    "he is", "he s ",
    "she is", "she s ",
    "you are", "you re ",
    "we are", "we re ",
    "they are", "they re "
)

def filterPair(p):
    return len(p[0].split(' ')) < MAX_LENGTH and \
        len(p[1].split(' ')) < MAX_LENGTH and \
        p[1].startswith(eng_prefixes)

def filterPairs(pairs):
    return [pair for pair in pairs if filterPair(pair)]

def prepareData(lang1, lang2, reverse=False):
    input_lang, output_lang, pairs = readLangs(lang1, lang2, reverse)
    print("Read %s sentence pairs" % len(pairs))
    pairs = filterPairs(pairs)
    print("Trimmed to %s sentence pairs" % len(pairs))
    print("Counting words...")
    for pair in pairs:
        input_lang.addSentence(pair[0])
        output_lang.addSentence(pair[1])
    print("Counted words:")
    print(input_lang.name, input_lang.n_words)
    print(output_lang.name, output_lang.n_words)
    return input_lang, output_lang, pairs

input_lang, output_lang, pairs = prepareData('eng', 'fra', True)
print(random.choice(pairs))

The Encoder

encode
인코더는 입력 단어를 출력 벡터와 hidden state를 반환한다. hidden state는 다음 입력 단어에 사용한다.

class EncoderRNN(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(EncoderRNN, self).__init__()
        self.hidden_size = hidden_size

        self.embedding = nn.Embedding(input_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)

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

    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)

The Decoder

decode
디코더는 인코더의 출력 벡터를 받고 번역을 위한 단어를 생성한다. 인코더의 마지막 출력을 context 벡터라고 부른다. 이 벡터는 디코더의 초기 hidden state로 사용된다. 매 단계마다 디코더는 입력 토큰과 hidden state를 받는다. 초기 입력 토큰은 SOS 토큰이고 초기 hidden state는 context 벡터이다.

class DecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size):
        super(DecoderRNN, self).__init__()
        self.hidden_size = hidden_size

        self.embedding = nn.Embedding(output_size, hidden_size)
        self.gru = nn.GRU(hidden_size, hidden_size)
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.LogSoftmax(dim=1)

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

    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)

Attention Decoder

attention

지금까지 context 벡터가 전체 문장을 인코딩하는 부담을 지는 것처럼 보였다. 실제로 단순한 인코딩-디코딩 시스템은 여러 문제가 발생할 수 있다고 한다. 위에 그림에 보이는 어텐션 네트워크는 디코더의 입력과 hidden을 받고 어텐션 weight들을 계산하고 인코더의 출력과 곱하여 결과를 반환한다. 디코더의 결과는 문장이다. 어텐션 weights은 디코더가 더 자연스러운 번역을 하도록 도와주는 상수값이라고 생각하면 되겠다.
attention-decoder-network

class AttnDecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=MAX_LENGTH):
        super(AttnDecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.output_size = output_size
        self.dropout_p = dropout_p
        self.max_length = max_length

        self.embedding = nn.Embedding(self.output_size, self.hidden_size)
        self.attn = nn.Linear(self.hidden_size * 2, self.max_length)
        self.attn_combine = nn.Linear(self.hidden_size * 2, self.hidden_size)
        self.dropout = nn.Dropout(self.dropout_p)
        self.gru = nn.GRU(self.hidden_size, self.hidden_size)
        self.out = nn.Linear(self.hidden_size, self.output_size)

    def forward(self, input, hidden, encoder_outputs):
        embedded = self.embedding(input).view(1, 1, -1)
        embedded = self.dropout(embedded)

        attn_weights = F.softmax(
            self.attn(torch.cat((embedded[0], hidden[0]), 1)), dim=1)
        attn_applied = torch.bmm(attn_weights.unsqueeze(0),
                                 encoder_outputs.unsqueeze(0))

        output = torch.cat((embedded[0], attn_applied[0]), 1)
        output = self.attn_combine(output).unsqueeze(0)

        output = F.relu(output)
        output, hidden = self.gru(output, hidden)

        output = F.log_softmax(self.out(output[0]), dim=1)
        return output, hidden, attn_weights

    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)