当前位置：首页 > news >正文

序列到序列学习

news 2025/7/6 5:11:49

seq2seq

就是把一个句子翻译到另外一个句子。
在这里插入图片描述

机器翻译

给定一个源语言的句子，自动翻译成目标语言
这两个句子可以有不同的长度

seq2seq 是一个 Encoder - Decoder 的架构
在这里插入图片描述
编码器是一个 RNN ，读取输入句子（可以是双向）
解码器使用另外一个 RNN 来输出

编码器 - 解码器细节

编码器是没有输出的 RNN
编码器最后时间步的隐状态用作解码器的初始隐状态
在这里插入图片描述
将最后一层的 RNN 的在最后一个时刻的隐藏状态，也就是输出，和句子的 Embedding 输入放在一起，作为输入

训练

训练时解码器使用目标句子作为输入
在这里插入图片描述
在训练的时候，是知道目标句子的，也就是知道真正的翻译。哪怕某一个时刻翻译错误，也是没有关系的，下一个时刻我还是可以给定正确的输入的。

推理

在这里插入图片描述
但是推理就不一样了，不知道目标句子的翻译，只能以当前时刻的输出作为下一时刻的翻译。

衡量生成序列的好坏的 BLEU

总结

Seq2seq 从一个句子生成另一个句子
编码器和解码器都是 RNN
将编码器最后时间隐状态来初始解码器隐状态来完成信息传递
常用BLEU来衡量生成序列的好坏

代码实现

首先就是导入必要的环境

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

实现循环神经网络编码器

# @save
class Seq2SeqEncoder(d2l.Encoder):"""用于序列到序列学习的循环神经网络编码器""""""vocab_size: 词汇表大小，决定了嵌入层的输入维度。embed_size: 嵌入层的输出维度，也就是每个词被映射为向量的维度。num_hiddens: 隐藏层单元的数量，这决定了GRU（门控循环单元）层的输出维度。num_layers: GRU层的数量，可以堆叠多层GRU以增加模型的深度。dropout: dropout比例，用于正则化，防止过拟合，默认值为0，即不使用dropout。"""def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,dropout=0, **kwargs):super(Seq2SeqEncoder, self).__init__(**kwargs)# 嵌入层"""创建了一个嵌入层self.embedding，它将输入的词汇索引转换成固定大小的密集向量。创建了一个GRU层self.rnn，该层接受来自嵌入层的输入，并基于这些输入生成一系列隐藏状态。GRU是一种特殊的循环神经网络（RNN），能够捕捉序列中的长期依赖关系。"""self.embedding = nn.Embedding(vocab_size, embed_size)self.rnn = nn.GRU(embed_size, num_hiddens, num_layers,dropout=dropout)def forward(self, X, *args):# 输出'X'的形状：(batch_size,num_steps,embed_size)X = self.embedding(X)# 在循环神经网络模型中，第一个轴对应于时间步X = X.permute(1, 0, 2)  # 使得时间步位于第一个维度，以便与GRU层的要求相匹配。# 如果未提及状态，则默认为0output, state = self.rnn(X)# output的形状:(num_steps,batch_size,num_hiddens)# state的形状:(num_layers,batch_size,num_hiddens)return output, state

上述编码器的实现

# 创建一个序列到序列学习的循环神经网络编码器实例
# vocab_size=10: 输入词汇表大小为10
# embed_size=8: 嵌入层的输出维度为8
# num_hiddens=16: GRU隐藏单元的数量为16
# num_layers=2: GRU的层数为2
encoder = Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16,num_layers=2)# 将编码器设置为评估模式（evaluation mode），
# 这会关闭dropout等仅在训练时使用的功能
encoder.eval()# 创建一个形状为(4, 7)的输入张量X，表示批量大小为4，时间步数为7
# 数据类型为torch.long，表示每个元素是一个长整型（通常用于表示词汇索引）
X = torch.zeros((4, 7), dtype=torch.long)# 将输入X传入编码器，获取输出和状态
# output: 编码器在每个时间步的输出，形状为(num_steps, batch_size, num_hiddens)
# state: 编码器的最终隐状态，形状为(num_layers, batch_size, num_hiddens)
output, state = encoder(X)# 打印output的形状，验证编码器的输出维度是否符合预期
output.shape
# torch.Size([7, 4, 16])state.shape
# torch.Size([2, 4, 16])

解码器

class Seq2SeqDecoder(d2l.Decoder):"""用于序列到序列学习的循环神经网络解码器"""def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,dropout=0, **kwargs):"""初始化解码器:param vocab_size: 词汇表大小，决定了嵌入层的输入维度:param embed_size: 嵌入层的输出维度，也就是每个词被映射为向量的维度:param num_hiddens: 隐藏层单元的数量，这决定了GRU（门控循环单元）层的输出维度:param num_layers: GRU层的数量，可以堆叠多层GRU以增加模型的深度:param dropout: dropout比例，用于正则化，防止过拟合，默认值为0，即不使用dropout"""super(Seq2SeqDecoder, self).__init__(**kwargs)# 嵌入层：将输入的词汇索引转换成固定大小的密集向量self.embedding = nn.Embedding(vocab_size, embed_size)# GRU层：接受来自嵌入层和上下文向量的输入，并生成一系列隐藏状态# 输入维度为 embed_size + num_hiddens，因为输入拼接了上下文向量self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers,dropout=dropout)# 输出层：将GRU的输出映射到词汇表大小的向量，用于预测下一个词元self.dense = nn.Linear(num_hiddens, vocab_size)def init_state(self, enc_outputs, *args):"""初始化解码器的隐状态:param enc_outputs: 编码器的输出，包含上下文向量和隐状态:return: 编码器的最终隐状态，用作解码器的初始隐状态"""return enc_outputs[1]  # 提取编码器的隐状态def forward(self, X, state):"""前向传播:param X: 解码器的输入，形状为 (batch_size, num_steps):param state: 解码器的隐状态，形状为 (num_layers, batch_size, num_hiddens):return: 解码器的输出和更新后的隐状态"""# 嵌入层：将输入的词汇索引转换为特征向量# 输出'X'的形状：(batch_size, num_steps, embed_size)X = self.embedding(X).permute(1, 0, 2)  # 调整维度以匹配GRU的输入要求# 获取上下文向量（编码器的最后一层隐状态）# 广播context，使其具有与X相同的时间步数 (num_steps)context = state[-1].repeat(X.shape[0], 1, 1)# 将输入特征向量和上下文向量拼接在一起# 拼接后的形状为 (num_steps, batch_size, embed_size + num_hiddens)# 这里也就是为啥前面的nn.GRU的输入维度是embed_size + num_hiddensX_and_context = torch.cat((X, context), 2)# GRU层：处理拼接后的输入，生成输出和更新后的隐状态# output的形状：(num_steps, batch_size, num_hiddens)# state的形状：(num_layers, batch_size, num_hiddens)output, state = self.rnn(X_and_context, state)# 输出层：将GRU的输出映射到词汇表大小的向量# 调整维度以匹配输出要求# output的最终形状：(batch_size, num_steps, vocab_size)output = self.dense(output).permute(1, 0, 2)# 返回解码器的输出和更新后的隐状态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# (torch.Size([4, 7, 10]), torch.Size([2, 4, 16]))

通过零值化屏蔽不相关的项

#@save
def sequence_mask(X, valid_len, value=0):"""在序列中屏蔽不相关的项。对于每个序列，根据其有效长度，将超出有效长度的部分用指定的值进行屏蔽。参数:X: 输入张量，形状为 (batch_size, num_steps)，表示批量中的序列数据。valid_len: 有效长度张量，形状为 (batch_size,)，表示每个序列的有效长度。value: 用于屏蔽的值，默认为 0。返回:屏蔽后的张量，形状与输入张量 X 相同。"""# 获取输入张量 X 的序列长度（即 num_steps）maxlen = X.size(1)# 创建一个布尔掩码张量，形状为 (batch_size, num_steps)# torch.arange(maxlen) 生成一个从 0 到 maxlen-1 的张量# valid_len[:, None] 将 valid_len 的形状从 (batch_size,) 扩展为 (batch_size, 1)# 比较操作生成一个布尔张量，表示每个序列中哪些位置是有效的mask = torch.arange((maxlen), dtype=torch.float32,device=X.device)[None, :] < valid_len[:, None]"""执行 valid_len[:, None] 后：valid_len 原本是一个形状为 (2,) 的一维张量：[1, 2]使用 [:, None] 或者 .unsqueeze(-1) 可以在指定位置增加一个维度，这里是在最后一个位置增加了一个维度，因此得到的结果是形状为 (2, 1) 的二维张量"""# 使用布尔掩码将无效位置设置为指定的屏蔽值# ~mask 表示取反操作，将无效位置标记为 True"""通过X[~mask] = value，所有标记为True的位置（即那些超过有效序列长度的位置）都会被赋值为value（在这个例子中是0）"""X[~mask] = value# 返回屏蔽后的张量return X# 示例输入张量 X，形状为 (2, 3)，表示两个序列，每个序列有 3 个时间步
X = torch.tensor([[1, 2, 3], [4, 5, 6]])# 有效长度张量 valid_len，形状为 (2,)，表示每个序列的有效长度
valid_len = torch.tensor([1, 2])# 调用 sequence_mask 函数，对输入张量进行屏蔽
sequence_mask(X, valid_len)

通过扩展 softmax 交叉熵损失函数来遮蔽不相关的预测

# @save
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):# 创建一个和 label 形状一样的全1张量，表示“初始情况下，所有位置都有效”。weights = torch.ones_like(label)# 将超出每个样本有效长度的位置设为 0。weights = sequence_mask(weights, valid_len)  # 有效的就保留下来，其余的就全部变成 0 。# PyTorch 中的 CrossEntropyLoss 默认会做平均或求和，但我们这里要自己控制怎么处理损失，# 所以设置为 'none'，这样它会返回每个时间步的损失值。self.reduction = 'none'"""注意输入的 pred 是 (batch_size, num_steps, vocab_size)，而 PyTorch 的 CrossEntropyLoss 要求输入是 (batch_size, vocab_size, num_steps)，所以要做一次 permute。这个操作后，unweighted_loss 的形状是 (batch_size, num_steps)，即每个时间步都有一个损失值。"""unweighted_loss = super().forward(pred.permute(0, 2, 1), label)"""先对损失乘以权重，这样 padding 位置的损失就变成 0。然后对每个句子的时间步取平均（dim=1），得到一个形状为 (batch_size,) 的损失张量。"""weighted_loss = (unweighted_loss * weights).mean(dim=1)  # 对每个句子的损失进行平均return weighted_loss

代码健全性检查

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

训练

#@save
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]"""这里构造了解码器的输入 dec_input：<bos> 表示 “begin of sentence”，告诉模型开始生成。Y[:, :-1] 是目标句子去掉最后一个词的部分。拼接后变成 [<bos>, y1, y2, ..., y_{n-1}]，这样解码器就可以在每个时间步根据前面的真实词来预测下一个词。这就是所谓的“强制教学”（teacher forcing）：用真实标签作为下一个输入，而不是自己预测的结果。"""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)

在这里插入图片描述
预测

#@save
def predict_seq2seq(net, src_sentence, src_vocab, tgt_vocab, num_steps,device, save_attention_weights=False):"""序列到序列模型的预测"""# 将模型设为评估模式，关闭dropout、batchnorm等训练专用操作net.eval()# 将输入句子转为小写并按空格分割成单词列表，然后查出每个词在源语言词汇表中的索引# 最后添加一个 <eos>（end of sentence）标记作为句子结束src_tokens = src_vocab[src_sentence.lower().split(' ')] + [src_vocab['<eos>']]# 创建 enc_valid_len 张量，表示编码器的有效长度（即不包括 padding 的实际词数）enc_valid_len = torch.tensor([len(src_tokens)], device=device)# 对源句子进行填充或截断，使其长度等于 num_steps，并用 <pad> 填充不足部分src_tokens = d2l.truncate_pad(src_tokens, num_steps, src_vocab['<pad>'])# 添加批量维度（batch dimension），因为模型输入需要 batch_size 维度# enc_X 形状：(1, num_steps)，其中 1 是 batch_size=1enc_X = torch.unsqueeze(torch.tensor(src_tokens, dtype=torch.long, device=device), dim=0)# 编码器前向传播，得到编码后的输出和隐藏状态enc_outputs = net.encoder(enc_X, enc_valid_len)# 使用编码器的输出初始化解码器的状态（例如 GRU 的初始隐藏状态）dec_state = net.decoder.init_state(enc_outputs, enc_valid_len)# 解码器的第一个输入是目标语言的起始标记 <bos># 同样添加批量维度，形状为 (1, 1)dec_X = torch.unsqueeze(torch.tensor([tgt_vocab['<bos>']], dtype=torch.long, device=device), dim=0)# 存放最终生成的目标语言句子的 token 索引output_seq = []# 如果启用保存注意力权重，则保存每一步的注意力权重attention_weight_seq = []# 循环最多 num_steps 次，逐步生成目标语言句子for _ in range(num_steps):# 解码器前向传播，输入当前的 dec_X 和当前状态，输出 Y 和新的状态Y, dec_state = net.decoder(dec_X, dec_state)# 从输出 Y 中选择概率最大的词元作为当前时间步的预测结果# dec_X 变为这个预测值，用于下一个时间步的输入（即 greedy search）dec_X = Y.argmax(dim=2)# 将预测结果转换为 Python 标量整数（词元索引）pred = dec_X.squeeze(dim=0).type(torch.int32).item()# 如果启用了保存注意力权重，则记录当前步的注意力权重if save_attention_weights:attention_weight_seq.append(net.decoder.attention_weights)# 如果预测的是 <eos>，说明句子已经生成完毕，提前退出循环if pred == tgt_vocab['<eos>']:break# 否则将当前预测的词元加入输出序列output_seq.append(pred)# 将预测的词元索引转换回对应的单词，并拼接成字符串返回return ' '.join(tgt_vocab.to_tokens(output_seq)), attention_weight_seq

BLEU 代码的实现

def bleu(pred_seq, label_seq, k):  #@save"""计算BLEU（Bilingual Evaluation Understudy）分数，用于评估预测序列与标签序列的相似度。参数:pred_seq: str，预测的序列（例如机器翻译的输出）。label_seq: str，标签序列（例如机器翻译的参考翻译）。k: int，最长的n元语法（n-grams）长度，用于计算BLEU分数。返回:float，BLEU分数。"""# 将预测序列和标签序列按空格分割成词元列表pred_tokens, label_tokens = pred_seq.split(' '), label_seq.split(' ')# 获取预测序列和标签序列的长度len_pred, len_label = len(pred_tokens), len(label_tokens)# 惩罚因子：如果预测序列比标签序列短，则会有惩罚# math.exp(min(0, 1 - len_label / len_pred)) 确保惩罚因子不大于1score = math.exp(min(0, 1 - len_label / len_pred))# 遍历从1到k的n元语法（n-grams）for n in range(1, k + 1):# 匹配的n元语法数量num_matches = 0# 用于存储标签序列中每个n元语法的出现次数label_subs = collections.defaultdict(int)# 遍历标签序列，提取所有长度为n的n元语法，并统计其出现次数for i in range(len_label - n + 1):label_subs[' '.join(label_tokens[i: i + n])] += 1# 遍历预测序列，提取所有长度为n的n元语法，并与标签序列中的n元语法进行匹配for i in range(len_pred - n + 1):ngram = ' '.join(pred_tokens[i: i + n])  # 当前的n元语法if label_subs[ngram] > 0:  # 如果标签序列中存在该n元语法num_matches += 1  # 匹配数加1label_subs[ngram] -= 1  # 减少标签序列中该n元语法的计数# 计算当前n元语法的精确度，并乘以权重（权重为0.5的幂次）# math.pow(0.5, n) 确保更长的n元语法权重更小score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n))# 返回最终的BLEU分数return score

将几个英文句子翻译成法语

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