【深度学习实战】【详细解读】基于Seq2Seq模型实现简单的机器翻译

官网地址：NLP FROM SCRATCH: TRANSLATION WITH A SEQUENCE TO SEQUENCE NETWORK AND ATTENTION 数据下载：数据     基础知识：【深度学习实战】从零开始深度学习（四）：RNN与自然语言处理  

1. 导入需要的包

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

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

import matplotlib.pyplot as plt
import matplotlib.ticker as ticker
import numpy as np

  同时设置是否使用GPU进行训练。  

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

2. 读入数据

  首先建立一个字典类，“SOS”和“EOS”是一句话开始和结束的标志，先给它们附上0和1这两个标签；并定义两个函数分析输入的句子。  

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

  接下来，通过定义两个函数将Unicode字符都转化成ASCII字符，并且让所有的字母都变成小写，并去掉所有标点  

def unicodeToAscii(s):
    return ''.join(
        c for c in unicodedata.normalize('NFD', s)
        if unicodedata.category(c) != 'Mn')

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

  接下来，对数据集进行处理。我们先看一下数据的样子：  
  由于数据是一行一行对应的英文与法文，所以要对输入的数据做一下处理。  

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

    # 读取文件，并分割成行
    lines = open('data/%s-%s.txt' % (lang1, lang2), encoding='utf-8').read().strip().split('\n')

    # 将每一行的数据分割成单词对
    pairs = [[normalizeString(s) for s in l.split('\t')] for l in lines]

    # 设置是否反转，默认是英文翻译成法文，如果设置reverse，则是法文翻译成英文
    # 设置两个语言的词典
    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

  为了简化计算，我们需要对数据进行筛选。我们选择某些固定开头且长度小于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))
'''
out:
Reading lines...
Read 135842 sentence pairs
Trimmed to 10599 sentence pairs
Counting words...
Counted words:
fra 4345
eng 2803
['nous nous dirigeons a l ouest .', 'we re going west .']
'''

  可以看到，总的数据集是有135842个句子，经过筛选，保留10599个句子作为训练样本。  

3. 数据预处理——建立数据集

  为了训练，我们需要把这些词/句子对转化成输入张量(输入句子中单词的索引)和一个目标张量(目标句子中单词的索引)。在创建这些向量时，我们将把“SOS”和“EOS”分别加载句子的头尾，作为标识符。  

# indexesFromSentence函数的作用是将返回句子中每个词的在字典中的索引
def indexesFromSentence(lang, sentence):
    return [lang.word2index[word] for word in sentence.split(' ')]

# tensorFromSentence函数的作用是将句子妆花成张量
def tensorFromSentence(lang, sentence):
    indexes = indexesFromSentence(lang, sentence)
    indexes.append(EOS_token)
    return torch.tensor(indexes, dtype=torch.long, device=device).view(-1, 1)

# 将句子对转化成输入张量和目标张量
def tensorsFromPair(pair):
    input_tensor = tensorFromSentence(input_lang, pair[0])
    target_tensor = tensorFromSentence(output_lang, pair[1])
    return (input_tensor, target_tensor)

4. 模型搭建

  在准备好数据之后，第二步是搭建模型。  

4.1 编码器

  首先，我们搭建一个 编码器（encoder）。  

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向量扁平化，并设置seq=1，batch=1
        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)

  这个编码器其实很简单，就是一个简单的RNN网络。网络有两层，一层embedding层，用于得到单词的词向量；一层单层GRU层，用于将单词编码。inithidden函数用于初始化h0。  
  编码器的input_size是词典的大小。hidden_size可以自己定义，是word embedded的维度，常取的数值有50,100,200,256,300等。  

hidden_size = 256
encoder1 = EncoderRNN(input_lang.n_words, hidden_size).to(device)

4.2 解码器

  接下来搭建解码器（decoder） 。  

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)

  解码器的结构也很简单。  
  首先先将输入的单词进入embedding层进行词向量计算。如果是句首则输入句首标识符“SOS”；如果不是句首，则将上一个解码出来的单词当成输入；词向量经过非线性激活函数，和h0一起进入循环神经网络层。   需要注意的是，解码器的h0应该等于编码器的hn。   通过GRU层的计算，可以得到一个输出和解码后的hn。解码器的输出维度应该是翻译目标语言的词典大小，通过softmax，输出概率最大的那个单词作为翻译后的单词。   上面这个就是简单解码器的结构，但是它的效率实在是太低了，因此我们引入注意力机制，改善我们的解码器结构。  

4.3 注意力机制——改善解码器

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))

        combine = torch.cat((embedded[0], attn_applied[0]), 1)
        
        output = self.attn_combine(combine).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)

  引入注意力机制的解码器相对来说更复杂一些。（好吧，复杂的多。。。。）   我们先来看一下网络的基本结构（__init__函数部分），网络一共有五层（其中Dropout可以看做是对网络结构的一个优化，不算是一层）：  

1. embedding层——获得输入词语的词向量；
2. attn层——一个线性层；
3. attn_combine层——另一个线性层；
4. gru层——循环神经网络层；
5. out层——第三个线性层；

  这些层组成了网络的基本框架，它们怎么工作、发挥怎样的作用要看forward函数部分。推荐大家可以先看下这篇文章：注意力机制的基本思想和实现原理（很详细），对理解后面的代码很有帮助。   1.首先，将输入的单词经过embedding层和dropout处理，获得其词向量。和简单解码器一样，如果是句首则输入句首标识符“SOS”；如果不是句首，则将上一个解码出来的单词当成输入。  
    2.接着，计算注意力分配概率。通过torch.cat()函数将h0和输入的词向量首尾相接，作为下一个线性层的输入。所以我们可以从网络结构的设置看到，attn这个线性层的输入维度是2*hidden_size，输出是max_length（也就是句子中最多有用多少个词），这个线性层的含义我认为是提供一个计算注意力参数的权重W（这个W可以跟着网络一起训练），根据计算得到的值进行softmax操作，得到最终的注意力分配概率分布值。  
  这里借用文章注意力机制的基本思想和实现原理（很详细）中的一张图，可以更方便地理解 attn_weights = F.softmax(self.attn(torch.cat((embedded[0], hidden[0]), 1)), dim=1) 这行代码的作用：  
    3.接着，利用torch.bmm()函数，把注意力分配到输入的内容向量上，得到经过注意力机制后的内容输入向量attn_applied。  
  这里torch.bmm()用于对存储在两个批batch1和batch2内的矩阵进行批矩阵乘操作，两个tensor的维度必须为3，所以要将attn_weights和encoder_outputs增加一个维度。     4.这时候，我们就拥有了两个结果，一个是目前读入的单词的词向量embedded，一个是经过注意力机制调整的句子的内容向量attn_applied。我们将这两个向量同样利用torch.cat()函数收尾相接，得到一个新的combine向量。这个combine向量既包括当前单词的信息，又包括句子内容的信息，还包含注意力。  
    5.接下来的操作就和简单解码器是一样的了。最终输出三个参数：output，hidden和注意力权值attn_weights。  
 

5. 模型训练

  我们首先看一下训练框架：  

def trainIters(encoder, decoder, n_iters, print_every=1000, plot_every=100, learning_rate=0.01):
    start = time.time()
    plot_losses = []
    print_loss_total = 0  # Reset every print_every
    plot_loss_total = 0  # Reset every plot_every

    encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
    decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)
    criterion = nn.NLLLoss()
    
    training_pairs = [tensorsFromPair(random.choice(pairs))
                      for i in range(n_iters)]
    
    for iter in range(1, n_iters + 1):
        training_pair = training_pairs[iter - 1]
        input_tensor = training_pair[0]
        target_tensor = training_pair[1]

        loss = train(input_tensor, target_tensor, encoder,decoder, encoder_optimizer, decoder_optimizer, criterion)
        
        print_loss_total += loss
        plot_loss_total += loss

        if iter % print_every == 0:
            print_loss_avg = print_loss_total / print_every
            print_loss_total = 0
            print('%s (%d %d%%) %.4f' % (timeSince(start, iter / n_iters),iter, iter / n_iters * 100, print_loss_avg))

        if iter % plot_every == 0:
            plot_loss_avg = plot_loss_total / plot_every
            plot_losses.append(plot_loss_avg)
            plot_loss_total = 0

    showPlot(plot_losses)

  我们首先设置encoder和decoder的优化器和损失函数： encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate) decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate) criterion = nn.NLLLoss()   接着，我们创建训练数据集tranning pairs。训练数据集从总数据集中随机选取创建，训练数据集的样本个数为n_iters。由于是随机选取的，所以会有重复的样本出现。 training_pairs = [tensorsFromPair(random.choice(pairs)) for i in range(n_iters)]   下面就是开始通过for循环开始训练。每次选取一个样本，利用该样本去训练模型，并更新模型参数。每训练print_every个样本，输出相关参数，并记录loss值，以便于最后的图像绘制。时间参数和绘图的代码如下：  

def asMinutes(s):
    m = math.floor(s / 60)
    s -= m * 60
    return '%dm %ds' % (m, s)


def timeSince(since, percent):
    now = time.time()
    s = now - since
    es = s / (percent)
    rs = es - s
    return '%s (- %s)' % (asMinutes(s), asMinutes(rs))

def showPlot(points):
    plt.figure()
    fig, ax = plt.subplots()
    # this locator puts ticks at regular intervals
    loc = ticker.MultipleLocator(base=0.2)
    ax.yaxis.set_major_locator(loc)
    plt.plot(points)
    plt.show()

    那么，怎么得到loss值呢？ 通过一个train()函数来定义—— loss = train(input_tensor, target_tensor, encoder,decoder, encoder_optimizer, decoder_optimizer, criterion) 这也是模型训练的核心部分之一  

teacher_forcing_ratio = 0.5

def train(input_tensor, target_tensor, 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_tensor.size(0)
    target_length = target_tensor.size(0)

    encoder_outputs = torch.zeros(max_length, encoder.hidden_size, device=device)

    loss = 0

    for ei in range(input_length):
        encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
        encoder_outputs[ei] = encoder_output[0, 0]

    decoder_input = torch.tensor([[SOS_token]], device=device)
    
    decoder_hidden = encoder_hidden

    use_teacher_forcing = True if random.random() < teacher_forcing_ratio else False

    if use_teacher_forcing:
        # Teacher forcing: Feed the target as the next input
        for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            loss += criterion(decoder_output, target_tensor[di])
            decoder_input = target_tensor[di]  # Teacher forcing
    else:
        # Without teacher forcing: use its own predictions as the next input
        for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            topv, topi = decoder_output.topk(1)
            decoder_input = topi.squeeze().detach()  # detach from history as input

            loss += criterion(decoder_output, target_tensor[di])
            if decoder_input.item() == EOS_token:
                break

    loss.backward()

    encoder_optimizer.step()
    decoder_optimizer.step()

    return loss.item() / target_length

    在train()函数中，我们首先将输入句子的每个词输入encoder中进行编码，进而得到这个句子的内容编码encoder_outputs。   接着，我们将decoder的输入decoder_input设置为句子的开始标识符“SOS”；并将decoder的h0设置为句子的内容编码。 decoder_input = torch.tensor([[SOS_token]], device=device) decoder_hidden = encoder_hidden   解码器的训练通过后面的for循环来实现。   在模型训练过程中，我们加入了一个teaching force的阈值，表示使用teaching force的概率。当随机生成的数字大于这个阈值时，使用teaching force；否则不使用。teacher force就是在翻译的过程中“抄答案”，将正确的单词作为后面decoder的输入。   我们可以从的decoder的代码中再深入体会一下teacher force的作用。   首先我们先看不使用teacher_force的代码：  

# Without teacher forcing: use its own predictions as the next input
        for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            topv, topi = decoder_output.topk(1)
            decoder_input = topi.squeeze().detach()  # detach from history as input

  这个代码的流程可以用一张图来表示：  
  可以看到，我们将这一次翻译的结果当做后面翻译的输入。通过得到的decode_output里面最大值的下标，到词典中找到对应的词，认为这个词是目前翻译的结果，并当做后面继续翻译的输入。那么这个翻译好的词有可能是正确的，也有可能是错误的；如果翻译错了，很有可能会影响后面的计算。   而采用teacher force的代码 是这样子的：  

for di in range(target_length):
            decoder_output, decoder_hidden, decoder_attention = decoder(decoder_input, decoder_hidden, encoder_outputs)
            loss += criterion(decoder_output, target_tensor[di])
            decoder_input = target_tensor[di]  # Teacher forcing

  我们可以看到，如果使用了teacher force，不管翻译的结果是否正确，我们都使用正确的答案参与后面的decoder的计算中。简单来说，就是 “师傅带进门，修行靠个人” 。 decoder_input = target_tensor[di] # Teacher forcing   最后就是设置模型参数，开始训练：  

hidden_size = 256
encoder1 = EncoderRNN(input_lang.n_words, hidden_size).to(device)
attn_decoder1 = AttnDecoderRNN(hidden_size, output_lang.n_words, dropout_p=0.1).to(device)

trainIters(encoder1, attn_decoder1, 75000, print_every=5000)

  这个模型在我的电脑上面要跑将近70分钟（T^T），跑出来的loss函数的图如下所示：  
 

6. 模型评估

  模型评估和模型训练部分的代码基本一致，就是把输出的向量根据字典换成了词语。  

def evaluate(encoder, decoder, sentence, max_length=MAX_LENGTH):
    with torch.no_grad():
        input_tensor = tensorFromSentence(input_lang, sentence)
        input_length = input_tensor.size()[0]
        encoder_hidden = encoder.initHidden()

        encoder_outputs = torch.zeros(max_length, encoder.hidden_size, device=device)

        for ei in range(input_length):
            encoder_output, encoder_hidden = encoder(input_tensor[ei],
                                                     encoder_hidden)
            encoder_outputs[ei] += encoder_output[0, 0]

        decoder_input = torch.tensor([[SOS_token]], device=device)  # 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)
            if topi.item() == EOS_token:
                decoded_words.append('<EOS>')
                break
            else:
                decoded_words.append(output_lang.index2word[topi.item()])

            decoder_input = topi.squeeze().detach()

        return decoded_words, decoder_attentions[:di + 1]

  我们可以调用一下下面的代码来看一下具体的翻译效果：  

def evaluateRandomly(encoder, decoder, n=10):
    for i in range(n):
        pair = random.choice(pairs)
        print('>', pair[0])
        print('=', pair[1])
        output_words, attentions = evaluate(encoder, decoder, pair[0])
        output_sentence = ' '.join(output_words)
        print('<', output_sentence)
        print('')

evaluateRandomly(encoder1, attn_decoder1)

    输出结果如下：  

> je ne me sens pas offense .
= i m not offended .
< i m not offended . <EOS>

> je me fais du souci pour lui .
= i am worried about him .
< i am worried about him . <EOS>

> je suis un petit enfant .
= i m a little child .
< i m a little child . <EOS>

> je suis occupee .
= i m busy .
< i m busy . <EOS>

> elle va juste faire des courses .
= she is just going shopping .
< she is just going shopping . <EOS>

> je suis ravi de vous aider .
= i am glad to help you .
< i am glad to help you . <EOS>

> je suis avec toi .
= i m by your side .
< i m in you . <EOS>

> tout va bien .
= i m fine .
< i m fine . <EOS>

> je m ennuie juste un peu .
= i m just a little bored .
< i m just a little a little . <EOS>

> vous vous etes trompe d avion .
= you are on the wrong plane .
< you are on the wrong . . <EOS>

  看起来还是不错的  

7. 注意力权重可视化

output_words, attentions = evaluate(
    encoder1, attn_decoder1, "je suis trop froid .")
plt.matshow(attentions.numpy())

  可以得到注意力权重图如下：  
  还可以使用加强版的可视化函数：  

def showAttention(input_sentence, output_words, attentions):
    # Set up figure with colorbar
    fig = plt.figure()
    ax = fig.add_subplot(111)
    cax = ax.matshow(attentions.numpy(), cmap='bone')
    fig.colorbar(cax)

    # Set up axes
    ax.set_xticklabels([''] + input_sentence.split(' ') +
                       ['<EOS>'], rotation=90)
    ax.set_yticklabels([''] + output_words)

    # Show label at every tick
    ax.xaxis.set_major_locator(ticker.MultipleLocator(1))
    ax.yaxis.set_major_locator(ticker.MultipleLocator(1))

    plt.show()


def evaluateAndShowAttention(input_sentence):
    output_words, attentions = evaluate(
        encoder1, attn_decoder1, input_sentence)
    print('input =', input_sentence)
    print('output =', ' '.join(output_words))
    showAttention(input_sentence, output_words, attentions)


evaluateAndShowAttention("elle a cinq ans de moins que moi .")
evaluateAndShowAttention("elle est trop petit .")
evaluateAndShowAttention("je ne crains pas de mourir .")
evaluateAndShowAttention("c est un jeune directeur plein de talent .")

  最后可以得到下面这些图：