一、Seq2Seq模型：编码-解码框架的开山之作
我们首先要了解的是seq2seq（Sequence-to-Sequence）模型。它最早由Google在2014年的一篇论文中提出，是第一个真正意义上的端到端的编码器-解码器（Encoder-Decoder）框架。
seq2seq模型主要由两部分组成：
- 编码器（Encoder）：通常是一个RNN（循环神经网络），用于将输入序列编码为一个固定长度的向量表示，即所谓的“语义向量”。你可以把编码器看作一位尽职的秘书,他会认真阅读你给他的所有材料(输入序列),然后为你写一份精炼的内容摘要(语义向量)。
- 解码器（Decoder）：同样是一个RNN，接收编码器输出的语义向量，从中解码出目标输出序列。解码器就像一位创意写手,他会根据秘书提供的摘要(语义向量),创作出一篇全新的文章(输出序列)。
然而，传统的seq2seq模型也存在一些局限性：
- 编码器需要将整个输入序列压缩为一个固定大小的向量,对于较长的输入序列,这样做会损失很多信息。就像秘书写摘要,输入内容太多的话,摘要就不可能面面俱到。
- 解码器只能利用编码器最后一个时间步的状态,无法充分利用编码器中间产生的语义信息。秘书的摘要再好,也只是对原始材料的高度概括,写手创作时无法参考材料的细节。
二、注意力机制:让写手"只选对的,不选多的"
为了克服上述问题，研究者提出了注意力机制（Attention Mechanism）。它的核心思想是:解码器在生成每个词时,都可以有选择地参考编码器不同时间步的输出,自动决定应该重点关注哪些部分。这个过程有点像写手在创作时翻阅秘书整理的资料,重点参考与当前内容最相关的部分。
在实现上，注意力机制会为编码器的每个时间步的输出分配一个权重，然后基于这些权重计算一个上下文向量（Context Vector），作为解码器的附加输入。权重的计算通常基于解码器当前的隐藏状态和编码器各时间步的输出。直观地说,权重越大,说明编码器该时间步的输出对解码器当前预测越重要,需要被重点"关注"。
引入注意力机制后,模型变得更加灵活和高效:
- 不再需要将整个输入序列压缩为固定长度的向量,缓解了长序列的信息损失问题。秘书不用再提供"一言以蔽之"的摘要,而是可以整理一份完整的材料供写手参考。
- 解码过程可以参考编码器各个时间步的输出,充分利用了输入序列的细节信息。写手可以有的放矢地参考材料,写出更精彩的内容。
首先导入库
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
# 使用PyTorch的Transformer模块来简化编码器和解码器的实现
from torch.nn import Transformer定义编码器类
编码器使用了PyTorch的TransformerEncoder模块，它由多个TransformerEncoderLayer堆叠而成。每个编码器层包含一个多头自注意力机制和一个前馈神经网络。此外，为了让模型能够区分输入序列中不同位置的词，我们还使用了位置编码（Positional Encoding）。位置编码与词嵌入相加,得到最终的输入表示。
class Encoder(nn.Module):
def __init__(self, input_dim, hid_dim, n_layers, n_heads, pf_dim, dropout):
super().__init__()
# 定义Transformer编码器层
self.encoder_layer = nn.TransformerEncoderLayer(
d_model=hid_dim,
nhead=n_heads,
dim_feedforward=pf_dim,
dropout=dropout
)
# 编码器由n_layers个编码器层堆叠而成
self.transformer_encoder = nn.TransformerEncoder(self.encoder_layer, num_layers=n_layers)
# 位置编码,用于为输入序列的每个位置添加位置信息
self.pos_embedding = nn.Embedding(1000, hid_dim)
# 输入嵌入层,将离散的输入token转换为连续的向量表示
self.embedding = nn.Embedding(input_dim, hid_dim)
# Dropout层,用于防止过拟合
self.dropout = nn.Dropout(dropout)
def forward(self, src):
# src的形状: (src_len, batch_size)
# 生成位置编码
src_pos = torch.arange(0, src.shape[0]).unsqueeze(0).repeat(src.shape[1], 1).permute(1, 0)
# 将输入序列转换为嵌入向量,并加上位置编码
src = self.embedding(src) + self.pos_embedding(src_pos)
# 通过Dropout层
src = self.dropout(src)
# 通过Transformer编码器获取最终的编码结果
# src的形状: (src_len, batch_size, hid_dim)
src = self.transformer_encoder(src)
return src定义注意力层
class Attention(nn.Module):
def __init__(self, hid_dim):
super().__init__()
self.attn = nn.Linear(hid_dim * 2, hid_dim)
self.v = nn.Linear(hid_dim, 1, bias=False)
def forward(self, hidden, encoder_outputs):
# hidden的形状: (batch_size, hid_dim)
# encoder_outputs的形状: (src_len, batch_size, hid_dim)
src_len = encoder_outputs.shape[0]
# 将hidden扩展为encoder_outputs的长度,以便拼接
hidden = hidden.unsqueeze(1).repeat(1, src_len, 1)
# 将hidden和encoder_outputs按最后一维拼接
energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim=2)))
# 计算注意力权重
attention = self.v(energy).squeeze(2)
return F.softmax(attention, dim=1)定义解码器类
解码器的核心也是一个TransformerDecoder模块，它由多个TransformerDecoderLayer组成。每个解码器层包含一个self-attention、一个cross-attention和一个前馈神经网络。其中，self-attention用于处理已生成的输出序列，cross-attention用于根据编码器的输出计算注意力权重。
在每个时间步，解码器首先将上一步的输出通过一个嵌入层和dropout层，然后用注意力层计算当前隐藏状态对编码器输出的注意力权重。接着,将注意力权重与编码器输出加权求和,得到一个上下文向量。这个上下文向量会与当前步的嵌入输入拼接，再通过一个线性层，作为transformer解码器的输入。最后，将transformer解码器的输出通过一个线性层和softmax层，得到当前步的词表分布。
class Decoder(nn.Module):
def __init__(self, output_dim, hid_dim, n_layers, n_heads, pf_dim, dropout, attention):
super().__init__()
self.output_dim = output_dim
self.hid_dim = hid_dim
self.n_layers = n_layers
self.n_heads = n_heads
self.pf_dim = pf_dim
self.dropout = dropout
# 注意力层
self.attention = attention
# 解码器层
self.decoder_layer = nn.TransformerDecoderLayer(
d_model=hid_dim,
nhead=n_heads,
dim_feedforward=pf_dim,
dropout=dropout
)
# 解码器由n_layers个解码器层堆叠而成
self.transformer_decoder = nn.TransformerDecoder(self.decoder_layer, num_layers=n_layers)
# 输出层,将transformer的输出转换为预测的词表分布
self.out = nn.Linear(hid_dim, output_dim)
# 输出词嵌入层
self.output_embedding = nn.Embedding(output_dim, hid_dim)
# Dropout层
self.dropout = nn.Dropout(dropout)
def forward(self, input, hidden, encoder_outputs):
# input的形状: (batch_size)
# hidden的形状: (batch_size, hid_dim)
# encoder_outputs的形状: (src_len, batch_size, hid_dim)
# 将输入转换为词嵌入
input = self.output_embedding(input.unsqueeze(0))
# 通过Dropout层
input = self.dropout(input)
# 通过注意力层,根据hidden和encoder_outputs计算注意力权重
a = self.attention(hidden, encoder_outputs)
# 将注意力权重与编码器输出相乘,得到注意力加权的上下文向量
# a的形状: (batch_size, src_len)
# encoder_outputs的形状: (src_len, batch_size, hid_dim)
# 将两者相乘得到 (batch_size, src_len, hid_dim)
# 然后在src_len维度求和,得到 (batch_size, hid_dim)
weighted = torch.bmm(a.unsqueeze(1), encoder_outputs.permute(1, 0, 2)).squeeze(1)
# 将加权的上下文向量与输入拼接,作为transformer解码器的输入
# input的形状: (1, batch_size, hid_dim)
# weighted的形状: (batch_size, hid_dim)
# 将两者拼接得到 (1, batch_size, 2*hid_dim)
input = torch.cat((input, weighted.unsqueeze(0)), dim=2)
# 将拼接后的输入通过线性层,将维度还原为hid_dim
input = F.relu(self.attn_combine(input))
# 通过transformer解码器
# input的形状: (1, batch_size, hid_dim)
# hidden的形状: (n_layers, batch_size, hid_dim)
# 注意我们这里只传入了encoder_outputs,因为解码器的self-attention会自动计算decoded_outputs
output, hidden = self.transformer_decoder(input, hidden, encoder_outputs)
# 将输出通过线性层转换为词表空间
output = self.out(output.squeeze(0))
# 应用softmax得到最终的词表分布
output = F.softmax(output, dim=1)
# 返回 output和新的隐藏状态
return output, hidden我们将编码器和解码器组合成完整的seq2seq模型：
在每个时间步,解码器根据上一步的输出、当前的隐藏状态和编码器的输出计算注意力权重,然后将注意力权重与编码器输出加权求和,得到一个上下文向量。这个上下文向量会作为transformer解码器的一部分输入，用于预测当前步的输出词。
class Seq2Seq(nn.Module):
def __init__(self, encoder, decoder, device):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.device = device
def forward(self, src, trg, teacher_forcing_ratio = 0.5):
# src的形状: (src_len, batch_size)
# trg的形状: (trg_len, batch_size)
# teacher_forcing_ratio是使用正确目标词的概率
batch_size = trg.shape[1]
trg_len = trg.shape[0]
trg_vocab_size = self.decoder.output_dim
# 存储解码器的输出序列
outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device)
# 编码输入序列
encoder_outputs = self.encoder(src)
# 初始化解码器的隐藏状态
hidden = encoder_outputs[-1,:,:]
# 解码器的第一个输入是 <sos> 标记
input = trg[0,:]
for t in range(1, trg_len):
# 根据当前的输入、隐藏状态和编码器输出,获取预测结果
output, hidden = self.decoder(input, hidden, encoder_outputs)
# 保存当前的输出
outputs[t] = output
# 根据teacher_forcing_ratio决定使用正确答案还是模型预测的词作为下一步的输入
teacher_force = random.random() < teacher_forcing_ratio
top1 = output.argmax(1)
input = trg[t] if teacher_force else top1
return outputs我们可以打印一下完整的model来看一下内部结构：
# 创建一些示例输入数据
src = torch.tensor([[1, 3, 4, 2]]) # 形状: (src_len, batch_size)
trg = torch.tensor([[1, 3, 4, 2]]) # 形状: (trg_len, batch_size)
# 定义模型参数
input_dim = len(input_vocab)
output_dim = len(output_vocab)
hid_dim = 256
n_layers = 3
n_heads = 8
pf_dim = 512
dropout = 0.1
# 实例化模型组件
attention = Attention(hid_dim)
encoder = Encoder(input_dim, hid_dim, n_layers, n_heads, pf_dim, dropout)
decoder = Decoder(output_dim, hid_dim, n_layers, n_heads, pf_dim, dropout, attention)
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# 实例化Seq2Seq模型
model = Seq2Seq(encoder, decoder, device).to(device)
print(model)输出：
Seq2Seq(
(encoder): Encoder(
(encoder_layer): TransformerEncoderLayer(
(self_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(linear1): Linear(in_features=256, out_features=512, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
(linear2): Linear(in_features=512, out_features=256, bias=True)
(norm1): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm2): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(dropout1): Dropout(p=0.1, inplace=False)
(dropout2): Dropout(p=0.1, inplace=False)
)
(transformer_encoder): TransformerEncoder(
(layers): ModuleList(
(0-2): 3 x TransformerEncoderLayer(
(self_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(linear1): Linear(in_features=256, out_features=512, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
(linear2): Linear(in_features=512, out_features=256, bias=True)
(norm1): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm2): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(dropout1): Dropout(p=0.1, inplace=False)
(dropout2): Dropout(p=0.1, inplace=False)
)
)
)
(pos_embedding): Embedding(1000, 256)
(embedding): Embedding(5, 256)
(dropout): Dropout(p=0.1, inplace=False)
)
(decoder): Decoder(
(attention): Attention(
(attn): Linear(in_features=512, out_features=256, bias=True)
(v): Linear(in_features=256, out_features=1, bias=False)
)
(decoder_layer): TransformerDecoderLayer(
(self_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(multihead_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(linear1): Linear(in_features=256, out_features=512, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
(linear2): Linear(in_features=512, out_features=256, bias=True)
(norm1): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm2): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm3): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(dropout1): Dropout(p=0.1, inplace=False)
(dropout2): Dropout(p=0.1, inplace=False)
(dropout3): Dropout(p=0.1, inplace=False)
)
(transformer_decoder): TransformerDecoder(
(layers): ModuleList(
(0-2): 3 x TransformerDecoderLayer(
(self_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(multihead_attn): MultiheadAttention(
(out_proj): NonDynamicallyQuantizableLinear(in_features=256, out_features=256, bias=True)
)
(linear1): Linear(in_features=256, out_features=512, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
(linear2): Linear(in_features=512, out_features=256, bias=True)
(norm1): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm2): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(norm3): LayerNorm((256,), eps=1e-05, elementwise_affine=True)
(dropout1): Dropout(p=0.1, inplace=False)
(dropout2): Dropout(p=0.1, inplace=False)
(dropout3): Dropout(p=0.1, inplace=False)
)
)
)
(out): Linear(in_features=256, out_features=5, bias=True)
(output_embedding): Embedding(5, 256)
(dropout): Dropout(p=0.1, inplace=False)
)
)三、指针生成网络:在"写原创"和"纯搬运"间找平衡
尽管有了注意力机制，传统的seq2seq模型在一些任务上的表现仍不尽如人意，尤其是在文本摘要领域。想象一下,如果让一个小说家去写新闻摘要,他很可能会添加一些原文中没有的内容。这在文学创作中是才华，在文本摘要里却是Bug。我们希望摘要模型既能准确抓取原文的关键信息,又能生成流畅自然的句子。
于是，斯坦福大学的研究者在2017年提出了指针生成网络（PGN）。它在传统的seq2seq+注意力框架的基础上，添加了一种新的混合生成方式：
- 生成模式:根据当前的解码器状态和注意力语境,从固定词表中生成新词。这一部分与普通的seq2seq没有区别，对应着摘要写手的“创作”能力。
- 拷贝模式:直接从输入序列中选择某个词,原封不动地"指针"到输出中。这一部分让模型有了"直接引用"的能力,对应着摘要写手的"摘抄"功能。
PGN通过一个可学习的"生成概率"来控制每一步应该选择生成模式还是拷贝模式。生成概率同时取决于解码器状态、注意力语境和当前输入,可以自动调节生成和拷贝的比例。打个比方,生成概率就像写手的思路开关:什么时候该发挥创造力,什么时候该直接引用原文,全凭她自己拿捏。
PGN在文本摘要、阅读理解等任务上取得了显著的效果提升。它成功地找到了"创作"与"摘编"的平衡:
- 在关键信息的提取和复述上更加准确,避免了关键细节的遗漏和僭改。写手知道什么地方该直接"照抄"。
- 生成的摘要更加自然流畅,可读性大大提高。写手也能适时发挥"文学才华",润色文字。
