9-7 序列到序列学习（seq2seq）

正如我们在 9-5节中看到的， 机器翻译中的输入序列和输出序列都是长度可变的。 为了解决这类问题，我们在 9-6节中 设计了一个通用的”编码器－解码器“架构。 本节，我们将使用两个循环神经网络的编码器和解码器， 并将其应用于序列到序列（sequence to sequence，seq2seq）类的学习任务 (Cho et al., 2014, Sutskever et al., 2014)。

遵循编码器－解码器架构的设计原则， 循环神经网络编码器使用长度可变的序列作为输入， 将其转换为固定形状的隐状态。 换言之，输入序列的信息被编码到循环神经网络编码器的隐状态中。 为了连续生成输出序列的词元， 独立的循环神经网络解码器是基于输入序列的编码信息 和输出序列已经看见的或者生成的词元来预测下一个词元。 图9.7.1演示了 如何在机器翻译中使用两个循环神经网络进行序列到序列学习。

在 图9.7.1中， 特定的“<eos>”表示序列结束词元。 一旦输出序列生成此词元，模型就会停止预测。

在循环神经网络解码器的初始化时间步，有两个特定的设计决定：

• 首先，特定的“<bos>”表示序列开始词元，它是解码器的输入序列的第一个词元。
• 其次，使用循环神经网络编码器最终的隐状态来初始化解码器的隐状态。 例如，在 (Sutskever et al., 2014)的设计中， 正是基于这种设计将输入序列的编码信息送入到解码器中来生成输出序列的。

在其他一些设计中 (Cho et al., 2014)， 如图9.7.1所示， 编码器最终的隐状态在每一个时间步都作为解码器的输入序列的一部分。 类似于 8-3节中语言模型的训练， 可以允许标签成为原始的输出序列， 从源序列词元“<bos>”“Ils”“regardent”“.” 到新序列词元 “Ils”“regardent”“.”“<eos>”来移动预测的位置。
下面，我们动手构建 图9.7.1的设计， 并将基于 9-5节中 介绍的“英－法”数据集来训练这个机器翻译模型。

import collections
import math
import torch
from torch import nn
from d2l import torch as d2l

编码器

class Seq2SeqEncoder(d2l.Encoder):"""用于序列到序列学习的循环神经网络编码器"""def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,dropout=0, **kwargs):# vocab_size: 词汇表的大小，即有多少不同的词汇。用于定义嵌入层的输入维度。# embed_size: 每个词的嵌入向量的维度。# num_hiddens: 循环神经网络（RNN）中隐藏层的单元数量（维度）。# num_layers: RNN 的层数。# dropout: Dropout 率，用于防止过拟合。# **kwargs: 其他可选的关键字参数，用于父类的初始化。super(Seq2SeqEncoder, self).__init__(**kwargs)# 嵌入层self.embedding = nn.Embedding(vocab_size, embed_size)# 创建了一个嵌入层，将词汇表中的词映射到一个指定维度的向量空间中。这个嵌入层会将输入的词索引转换为嵌入向量。self.rnn = nn.GRU(embed_size, num_hiddens, num_layers,dropout=dropout)# 创建了一个基于门控循环单元（GRU）的 RNN 层。GRU 是一种改进的循环神经网络，能够更好地捕捉长序列的依赖关系。它的输入是嵌入向量，输出是隐藏状态。dropout 参数用于防止模型过拟合。def forward(self, X, *args):# 输出'X'的形状：(batch_size,num_steps,embed_size)X = self.embedding(X)# 首先通过嵌入层将输入 X（通常是词索引的张量）转换为嵌入向量。输出的 X 形状为 (batch_size, num_steps, embed_size)，其中 batch_size 是批量大小，num_steps 是序列的长度，embed_size 是嵌入向量的维度。# 在循环神经网络模型中，第一个轴对应于时间步X = X.permute(1, 0, 2)# 对 X 进行维度转换，使用 permute 方法将张量的维度顺序调整为 (num_steps, batch_size, embed_size)，其中 num_steps 现在是第一个维度。这是因为 RNN 模型通常希望时间步在第一个维度上。# 如果未提及状态，则默认为0output, state = self.rnn(X)# 将处理后的 X 输入到 GRU 中，进行前向计算。rnn 返回两个值：# output 是 RNN 在每个时间步的输出，形状为 (num_steps, batch_size, num_hiddens)。# state 是 RNN 在最后一个时间步的隐藏状态，形状为 (num_layers, batch_size, num_hiddens)。# output的形状:(num_steps,batch_size,num_hiddens)# state的形状:(num_layers,batch_size,num_hiddens)return output, state

循环层返回变量的说明可以参考 8-6节。

下面，我们实例化上述编码器的实现： 我们使用一个两层门控循环单元编码器，其隐藏单元数为$16$。 给定一小批量的输入序列 X（批量大小为$4$，时间步为$7$）。 在完成所有时间步后， 最后一层的隐状态的输出是一个张量（output由编码器的循环层返回）， 其形状为（时间步数，批量大小，隐藏单元数）。

encoder = Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16,num_layers=2)
encoder.eval()
X = torch.zeros((4, 7), dtype=torch.long)
output, state = encoder(X)
output.shape

由于这里使用的是门控循环单元， 所以在最后一个时间步的多层隐状态的形状是 （隐藏层的数量，批量大小，隐藏单元的数量）。 如果使用长短期记忆网络，state中还将包含记忆单元信息。

state.shape

解码器

class Seq2SeqDecoder(d2l.Decoder):"""用于序列到序列学习的循环神经网络解码器"""def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,dropout=0, **kwargs):super(Seq2SeqDecoder, self).__init__(**kwargs)self.embedding = nn.Embedding(vocab_size, embed_size)self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers,dropout=dropout)# 创建了一个基于 GRU 的 RNN 层。与编码器不同的是，这里的 GRU 输入维度是嵌入向量维度 embed_size 与隐藏状态维度 num_hiddens 的总和，这是因为在解码过程中会将当前时间步的输入与上下文（编码器的输出）拼接在一起。self.dense = nn.Linear(num_hiddens, vocab_size)# 创建了一个线性层，用于将 RNN 的输出映射到词汇表的维度，从而得到每个词的概率分布。def init_state(self, enc_outputs, *args):return enc_outputs[1]# init_state 方法用于初始化解码器的隐藏状态。# 它接收 enc_outputs（编码器的输出）作为输入，并返回 enc_outputs 中的第二个元素，即编码器的隐藏状态 state，作为解码器的初始状态。def forward(self, X, state):# 输出'X'的形状：(batch_size,num_steps,embed_size)X = self.embedding(X).permute(1, 0, 2)# 将输入 X 通过嵌入层映射到嵌入向量，并使用 permute 将维度顺序调整为 (num_steps, batch_size, embed_size)，将时间步放在第一个维度。# 广播context，使其具有与X相同的num_stepscontext = state[-1].repeat(X.shape[0], 1, 1)# 从解码器的初始状态 state 中提取最后一层的隐藏状态（state[-1]），并通过 repeat 方法将其复制，使其在时间步维度上与 X 具有相同的长度。这一步是为了将编码器的上下文信息（通常是编码器的最后一层隐藏状态）传递给解码器的每一个时间步。X_and_context = torch.cat((X, context), 2)# 将输入 X 和上下文 context 在最后一个维度（即特征维度）上拼接在一起，形成新的输入 X_and_context，这个输入会被传递给 GRU。output, state = self.rnn(X_and_context, state)# 将拼接后的输入 X_and_context 和初始状态 state 传递给 GRU，进行前向计算。rnn 返回两个值：# output 是 GRU 在每个时间步的输出，形状为 (num_steps, batch_size, num_hiddens)。# state 是 GRU 在最后一个时间步的隐藏状态，形状为 (num_layers, batch_size, num_hiddens)。output = self.dense(output).permute(1, 0, 2)# 将 GRU 的输出通过线性层 self.dense 进行处理，将其映射到词汇表的维度，得到每个词的概率分布。然后再次使用 permute 将维度顺序调整回 (batch_size, num_steps, vocab_size)。# output的形状:(batch_size,num_steps,vocab_size)# state的形状:(num_layers,batch_size,num_hiddens)return output, state

下面，我们用与前面提到的编码器中相同的超参数来实例化解码器。 如我们所见，解码器的输出形状变为（批量大小，时间步数，词表大小）， 其中张量的最后一个维度存储预测的词元分布。

decoder = Seq2SeqDecoder(vocab_size=10, embed_size=8, num_hiddens=16,num_layers=2)
decoder.eval()
state = decoder.init_state(encoder(X))
output, state = decoder(X, state)
output.shape, state.shape

总之，上述循环神经网络“编码器－解码器”模型中的各层如 图9.7.2所示。

损失函数

在每个时间步，解码器预测了输出词元的概率分布。 类似于语言模型，可以使用softmax来获得分布， 并通过计算交叉熵损失函数来进行优化。 回想一下 9-5节中， 特定的填充词元被添加到序列的末尾， 因此不同长度的序列可以以相同形状的小批量加载。 但是，我们应该将填充词元的预测排除在损失函数的计算之外。

为此，我们可以使用下面的sequence_mask函数 通过零值化屏蔽不相关的项， 以便后面任何不相关预测的计算都是与零的乘积，结果都等于零。 例如，如果两个序列的有效长度（不包括填充词元）分别为$1$和$2$， 则第一个序列的第一项和第二个序列的前两项之后的剩余项将被清除为零。

# 在输入张量 X 中屏蔽掉超过 valid_len 指定长度的元素，用指定的 value 值替代
def sequence_mask(X, valid_len, value=0):"""在序列中屏蔽不相关的项"""maxlen = X.size(1)# maxlen 是 X 的第二个维度大小，即每个序列的长度（seq_len）。它表示每个序列的最大长度。mask = torch.arange((maxlen), dtype=torch.float32,device=X.device)[None, :] < valid_len[:, None]# 创建一个从 0 到 maxlen-1 的一维张量，数据类型为 float32，并与 X 保持相同的设备（即在同一个 CPU 或 GPU 上）。# [None, :] 为这个张量增加一个新的维度，使其形状变为 (1, maxlen)。# valid_len[:, None] 将 valid_len 张量扩展为 (batch_size, 1) 的形状。# mask = ... < valid_len[:, None] 是一个比较操作，生成一个布尔张量 mask，表示哪些位置的索引小于 valid_len。结果是一个形状为 (batch_size, maxlen) 的布尔张量，其中 True 表示对应位置的元素是有效的，False 表示需要屏蔽。X[~mask] = value# 通过布尔索引将 mask 为 False 的位置（即无效项）设为 value。~mask 代表取反操作，将 True 变为 False，将 False 变为 True。return XX = torch.tensor([[1, 2, 3], [4, 5, 6]])
sequence_mask(X, torch.tensor([1, 2]))

我们还可以使用此函数屏蔽最后几个轴上的所有项。如果愿意，也可以使用指定的非零值来替换这些项。

X = torch.ones(2, 3, 4)
sequence_mask(X, torch.tensor([1, 2]), value=-1)

现在，我们可以通过扩展softmax交叉熵损失函数来遮蔽不相关的预测。 最初，所有预测词元的掩码都设置为1。 一旦给定了有效长度，与填充词元对应的掩码将被设置为0。 最后，将所有词元的损失乘以掩码，以过滤掉损失中填充词元产生的不相关预测。

class MaskedSoftmaxCELoss(nn.CrossEntropyLoss):"""带遮蔽的softmax交叉熵损失函数"""# pred的形状：(batch_size,num_steps,vocab_size)# label的形状：(batch_size,num_steps)# valid_len的形状：(batch_size,)def forward(self, pred, label, valid_len):weights = torch.ones_like(label)# 创建一个与 label 形状相同的全 1 张量 weights，即 (batch_size, num_steps)，用于初始化权重。weights = sequence_mask(weights, valid_len)# 调用之前定义的 sequence_mask 函数，将 weights 中超过有效长度的部分屏蔽（置为 0），保留有效部分的权重为 1。这样，weights 将只在有效长度内为 1，其他位置为 0。self.reduction='none'# 将损失函数的 reduction 属性设置为 'none'，表示不对损失进行任何聚合操作，这样每个样本的损失都将保留在原始张量中。unweighted_loss = super(MaskedSoftmaxCELoss, self).forward(pred.permute(0, 2, 1), label)# 调用父类 nn.CrossEntropyLoss 的 forward 方法来计算未加权的交叉熵损失。# pred.permute(0, 2, 1) 将 pred 张量的维度从 (batch_size, num_steps, vocab_size) 变为 (batch_size, vocab_size, num_steps)，以符合 nn.CrossEntropyLoss 的输入要求。# label 作为目标标签输入，用于计算损失。# unweighted_loss 的形状为 (batch_size, num_steps)。weighted_loss = (unweighted_loss * weights).mean(dim=1)# 将 unweighted_loss 与 weights 相乘，以应用遮蔽操作。未加权损失的无效部分将被置为 0，有效部分保持不变。# .mean(dim=1) 计算每个序列的平均损失，即在 num_steps 维度上取平均值，得到每个样本的平均损失。结果 weighted_loss 的形状为 (batch_size,)。return weighted_loss# 返回最终的加权损失，形状为 (batch_size,)。

我们可以创建三个相同的序列来进行代码健全性检查， 然后分别指定这些序列的有效长度为$4$
、$2$和$0$。 结果就是，第一个序列的损失应为第二个序列的两倍，而第三个序列的损失应为零。

loss = MaskedSoftmaxCELoss()
loss(torch.ones(3, 4, 10), torch.ones((3, 4), dtype=torch.long),torch.tensor([4, 2, 0]))

训练

在下面的循环训练过程中，如 图9.7.1所示， 特定的序列开始词元（“<bos>”）和 原始的输出序列（不包括序列结束词元“<eos>”） 拼接在一起作为解码器的输入。 这被称为强制教学（teacher forcing）， 因为原始的输出序列（词元的标签）被送入解码器。 或者，将来自上一个时间步的预测得到的词元作为解码器的当前输入。

def train_seq2seq(net, data_iter, lr, num_epochs, tgt_vocab, device):"""训练序列到序列模型"""def xavier_init_weights(m):if type(m) == nn.Linear:nn.init.xavier_uniform_(m.weight)if type(m) == nn.GRU:for param in m._flat_weights_names:if "weight" in param:nn.init.xavier_uniform_(m._parameters[param])net.apply(xavier_init_weights)net.to(device)optimizer = torch.optim.Adam(net.parameters(), lr=lr)loss = MaskedSoftmaxCELoss()net.train()animator = d2l.Animator(xlabel='epoch', ylabel='loss',xlim=[10, num_epochs])for epoch in range(num_epochs):timer = d2l.Timer()metric = d2l.Accumulator(2)  # 训练损失总和，词元数量for batch in data_iter:optimizer.zero_grad()X, X_valid_len, Y, Y_valid_len = [x.to(device) for x in batch]bos = torch.tensor([tgt_vocab['<bos>']] * Y.shape[0],device=device).reshape(-1, 1)dec_input = torch.cat([bos, Y[:, :-1]], 1)  # 强制教学Y_hat, _ = net(X, dec_input, X_valid_len)l = loss(Y_hat, Y, Y_valid_len)l.sum().backward()      # 损失函数的标量进行“反向传播”d2l.grad_clipping(net, 1)num_tokens = Y_valid_len.sum()optimizer.step()with torch.no_grad():metric.add(l.sum(), num_tokens)if (epoch + 1) % 10 == 0:animator.add(epoch + 1, (metric[0] / metric[1],))print(f'loss {metric[0] / metric[1]:.3f}, {metric[1] / timer.stop():.1f} 'f'tokens/sec on {str(device)}')

现在，在机器翻译数据集上，我们可以 创建和训练一个循环神经网络“编码器－解码器”模型用于序列到序列的学习。

embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 300, d2l.try_gpu()train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers,dropout)
decoder = Seq2SeqDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers,dropout)
net = d2l.EncoderDecoder(encoder, decoder)
train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)

预测

为了采用一个接着一个词元的方式预测输出序列， 每个解码器当前时间步的输入都将来自于前一时间步的预测词元。 与训练类似，序列开始词元（“<bos>”） 在初始时间步被输入到解码器中。 该预测过程如 图9.7.3所示， 当输出序列的预测遇到序列结束词元（“<eos>”）时，预测就结束了。

我们将在 9-8节中介绍不同的序列生成策略。

def predict_seq2seq(net, src_sentence, src_vocab, tgt_vocab, num_steps,device, save_attention_weights=False):"""序列到序列模型的预测"""# 在预测时将net设置为评估模式net.eval()src_tokens = src_vocab[src_sentence.lower().split(' ')] + [src_vocab['<eos>']]enc_valid_len = torch.tensor([len(src_tokens)], device=device)src_tokens = d2l.truncate_pad(src_tokens, num_steps, src_vocab['<pad>'])# 添加批量轴enc_X = torch.unsqueeze(torch.tensor(src_tokens, dtype=torch.long, device=device), dim=0)enc_outputs = net.encoder(enc_X, enc_valid_len)dec_state = net.decoder.init_state(enc_outputs, enc_valid_len)# 添加批量轴dec_X = torch.unsqueeze(torch.tensor([tgt_vocab['<bos>']], dtype=torch.long, device=device), dim=0)output_seq, attention_weight_seq = [], []for _ in range(num_steps):Y, dec_state = net.decoder(dec_X, dec_state)# 我们使用具有预测最高可能性的词元，作为解码器在下一时间步的输入dec_X = Y.argmax(dim=2)pred = dec_X.squeeze(dim=0).type(torch.int32).item()# 保存注意力权重（稍后讨论）if save_attention_weights:attention_weight_seq.append(net.decoder.attention_weights)# 一旦序列结束词元被预测，输出序列的生成就完成了if pred == tgt_vocab['<eos>']:breakoutput_seq.append(pred)return ' '.join(tgt_vocab.to_tokens(output_seq)), attention_weight_seq

预测序列的评估

def bleu(pred_seq, label_seq, k):  #@save"""计算BLEU"""pred_tokens, label_tokens = pred_seq.split(' '), label_seq.split(' ')len_pred, len_label = len(pred_tokens), len(label_tokens)score = math.exp(min(0, 1 - len_label / len_pred))for n in range(1, k + 1):num_matches, label_subs = 0, collections.defaultdict(int)for i in range(len_label - n + 1):label_subs[' '.join(label_tokens[i: i + n])] += 1for i in range(len_pred - n + 1):if label_subs[' '.join(pred_tokens[i: i + n])] > 0:num_matches += 1label_subs[' '.join(pred_tokens[i: i + n])] -= 1score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n))return score

最后，利用训练好的循环神经网络“编码器－解码器”模型， 将几个英语句子翻译成法语，并计算BLEU的最终结果。

engs = ['go .', "i lost .", 'he\'s calm .', 'i\'m home .']
fras = ['va !', 'j\'ai perdu .', 'il est calme .', 'je suis chez moi .']
for eng, fra in zip(engs, fras):translation, attention_weight_seq = predict_seq2seq(net, eng, src_vocab, tgt_vocab, num_steps, device)print(f'{eng} => {translation}, bleu {bleu(translation, fra, k=2):.3f}')

小结

• 根据“编码器-解码器”架构的设计， 我们可以使用两个循环神经网络来设计一个序列到序列学习的模型。

• 在实现编码器和解码器时，我们可以使用多层循环神经网络。

• 我们可以使用遮蔽来过滤不相关的计算，例如在计算损失时。

• 在“编码器－解码器”训练中，强制教学方法将原始输出序列（而非预测结果）输入解码器。

• BLEU是一种常用的评估方法，它通过测量预测序列和标签序列之间的$n$元语法的匹配度来评估预测。