深度学习之 seq2seq 进行 英文到法文的翻译

深度学习之 seq2seq 进行 英文到法文的翻译

import os
import torch
import random

source_path = "data/small_vocab_en"
target_path = "data/small_vocab_fr"
MAX_LENGTH = 100
SOS_token = 0
EOS_token = 1

def load_data(path):
    input_file = os.path.join(path)
    with open(input_file, 'r', encoding='utf-8') as f:
        data = f.read()
    return data

source_text = load_data(source_path)
target_text = load_data(target_path)


class Dictionary(object):
    def __init__(self):
        self.word2idx = {'<SOS>': 0, '<EOS>': 1}
        self.idx2word = {0: '<SOS>', 1: '<EOS>'}
        self.count = 2

    def add_word(self, word):
        if word not in self.word2idx:
            
            self.idx2word[self.count - 1] = word
            
            self.word2idx[word] = len(self.idx2word) - 1
            
            self.count += 1
        return self.word2idx[word]

    def __len__(self):
        return len(self.idx2word)

class Lang(object):
    def __init__(self, name):
        self.name = name
        self.dictionary = Dictionary()
    
    def addSentence(self, sentence):
        return [self.addWord(w) for w in sentence.split()]
    
    def addWord(self, word):
        return self.dictionary.add_word(word)
    
    def __len__(self):
        return len(self.dictionary)

def readLangs(source_name, source_lang_text, target_name, target_lang_text):
    source_lang = Lang(source_name)
    source_data = [source_lang.addSentence(s) for s in source_lang_text.lower().split('
')]
    
    target_lang = Lang(target_name)
    target_sentences = [ s + ' <EOS>' for s in target_lang_text.lower().split('
')]
    target_data = [target_lang.addSentence(s) for s in target_sentences]
    
    pairs = list(zip(source_data, target_data))
    return source_lang, target_lang, pairs

source_lang, target_lang, pairs_data = readLangs('en', source_text, 'fe', target_text)

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

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):
        result = Variable(torch.zeros(1, 1, self.hidden_size))
        return result

class DecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=MAX_LENGTH):
        super(DecoderRNN, 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):
        result = Variable(torch.zeros(1, 1, self.hidden_size))
        return result

epochs = 10
print_every = 2
hidden_size = 256
teacher_forcing_ratio = 0.5

encoder_model = EncoderRNN(len(source_lang), hidden_size)
att_decoder_model = DecoderRNN(hidden_size, len(target_lang), dropout_p=0.1)

def variablesFromIds(ids):
    return Variable(torch.LongTensor(ids).view(-1, 1))

def variablesFromPair(pair):
    input_var = variablesFromIds(pair[0])
    output_var = variablesFromIds(pair[1])
    return (input_var, output_var)

def train(input, target, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion, max_length=MAX_LENGTH):
    encoder_hidden = encoder.initHidden()
    
    encoder_optimizer.zero_grad()
    decoder_optimizer.zero_grad()
    
    input_length = input.size()[0]
    target_length = target.size()[0]
    
    encoder_outputs = Variable(torch.zeros(max_length, encoder.hidden_size))
    
    loss = 0
    
    for i in range(input_length):
        encoder_output, encoder_hidden = encoder(input[i], encoder_hidden)
        encoder_outputs[i] = encoder_output[0][0]
    
    decoder_input = Variable(torch.LongTensor([[SOS_token]]))
    
    decoder_hidden = encoder_hidden
    
    use_teacher_forcing = True if random.random() < teacher_forcing_ratio else False
    
    if use_teacher_forcing:
        for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            
            loss += criterion(decoder_output, target[di])
            decoder_input = target[di]
    else:
        for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            
            topv, topi = decoder_output.data.topk(1)
            ni = topi[0][0]
        
            decoder_input = Variable(torch.LongTensor([[ni]]))
            
            loss += criterion(decoder_output, target[di])
            
            if ni == EOS_token:
                break;
    loss.backward()
    
    encoder_optimizer.step()
    decoder_optimizer.step()
    
    return loss.data[0] / target_length

def trainIters(encoder, decoder, n_iters, print_every=10, learning_rate=0.01):
    encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
    decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)
    
    training_pairs = [variablesFromPair(random.choice(pairs_data)) for i in range(n_iters)]
    criterion = nn.NLLLoss()
    
    total_loss = 0
    
    for iter in range(1, n_iters + 1):
        training_pair = training_pairs[iter - 1]
        input_variable = training_pair[0]
        target_variable = training_pair[1]
        
        loss = train(input_variable, target_variable, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion)
        total_loss  += loss
        
        if iter % print_every == 0:
            
            print('(%d %d%%) loss %d total-loss %d percent %.4f' % (iter, iter / n_iters * 100, loss ,total_loss, total_loss / print_every))

trainIters(encoder_model, att_decoder_model, 5000)

def evaluate(encoder, decoder, sentence, max_length = MAX_LENGTH):
    input_variable = variablesFromIds(sentence)
    
    input_length = input_variable.size()[0]
    encoder_hidden = encoder.initHidden()
    
    encoder_outputs = Variable(torch.zeros(max_length, encoder.hidden_size))
    
    for ei in range(input_length):
        encoder_output, encoder_hidden = encoder(input_variable[ei], encoder_hidden)
        encoder_outputs[ei] = encoder_outputs[ei] + encoder_output[0][0]
    
    decoder_input = Variable(torch.LongTensor([[SOS_token]]))  # SOS
    
    decoder_hidden = encoder_hidden

    decoded_words = []
    decoder_attentions = torch.zeros(max_length, max_length)
    
    for di in range(max_length):
        decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
        decoder_attentions[di] = decoder_attention.data
        topv, topi = decoder_output.data.topk(1)
        ni = topi[0][0]
        
        if ni == EOS_token:
            decoded_words.append('<EOS>')
            break
        else:
            decoded_words.append(target_lang.dictionary.idx2word[ni])
        
        decoder_input = Variable(torch.LongTensor([[ni]]))
    return decoded_words, decoder_attentions[:di + 1]  

evaluateRandomly(encoder_model, att_decoder_model)

结论

训练少，正确率较低，后面再实现一个对话机器人