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

import osimport torchimport randomsource_path = "data/small_vocab_en"target_path = "data/small_vocab_fr"MAX_LENGTH = 100SOS_token = 0EOS_token = 1def load_data(path):    input_file = os.path.join(path)    with open(input_file, 'r', encoding='utf-8') as f:        data = f.read()    return datasource_text = load_data(source_path)target_text = load_data(target_path)class Dictionary(object):    def __init__(self):        self.word2idx = {'
    
     ': 0, '
     
      ': 1}        self.idx2word = {0: '
      
       ', 1: '
       
        '}        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('\n')]        target_lang = Lang(target_name)    target_sentences = [ s + ' 
        
         ' for s in target_lang_text.lower().split('\n')]    target_data = [target_lang.addSentence(s) for s in target_sentences]        pairs = list(zip(source_data, target_data))    return source_lang, target_lang, pairssource_lang, target_lang, pairs_data = readLangs('en', source_text, 'fe', target_text)import torch.nn as nnfrom torch.autograd import Variablefrom torch import optimimport torch.nn.functional as Fclass 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 resultclass 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 resultepochs = 10print_every = 2hidden_size = 256teacher_forcing_ratio = 0.5encoder_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_lengthdef 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('
         
          ') 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)