目录
- 动机
- 加入注意力
- Bahdanau注意力的架构
- 总结
- Bahdanau注意力代码
- 带有注意力机制的解码器基本接口
- 实现带有Bahdanau注意力的循环神经网络解码器
- 测试Bahdanau注意力解码器
- 该部分总代码
- 训练
- 从零实现总代码
- 简洁实现代码
- 将几个英语句子翻译成法语
- 该部分总代码
- 将注意力权重序列进行拼接
- 可视化注意力权重
- 该部分总代码
动机
动机：在机器翻译中，每个生成的词可能源自与句子中间的不同的词。
源句子：“hello world .”
其中hello比较重要，所以去翻译这个词的时候应该找到与它相似的（也就是bonjour）应该把关注点放在这个词上面，而不需要去看很多其它的东西。
例如：假设翻译的是“monde”，那么对应的是“world”这个词
所以说比较长的句子的时候，就是把所有东西压到一起，因为压到一起可能看不出来到底哪一块是前面、哪一块是后面。所以在翻译句子哪个部位的时候去看对应部位的原句子。
但是seq2seq做不了这个事情，因为seq2seq传过去的东西是最后那个词的隐藏状态，seq2seq只用了最后这个词输出的东西，根据这个东西去往下找，
所以它看不到前面那些词的输入。当然，前面那些词（hello word）已经在这个里面了，假设已经存了很多问题了，那么需要从这里还原出你的未知的信息。
所以这里的动机是说：我想要去翻译对应的词的时候，去让我的注意力关注在原句子中对应的部分。
加入注意力
Bahdanau注意力的架构
在做解码器的RNN的输入的时候，一块来自Embedding、另外一块来自前面编码器的RNN的最后一个时刻的最后一层的输出作为上下文一块传进去的。
现在说用最后一个时刻作为上下文，即：作为那个context传进去不好。应该根据当前预测任务去动态的选择，而不是只能是最后时刻的隐藏状态作为输入，可能是前面几个时刻对应的那些隐藏状态作为输入。
例如：
在翻译第一个的时候（hello），我不应该去把那个RNN最后那个时刻（即：那个句号）对应的state传到解码器的RNN，而是把第一个时刻的隐藏状态传到解码器的RNN。
①编码器对应每个词的输出作为key和value放进Attention里面。
②解码器RNN对上一个词的预测输出是query
③注意力的输出和下一个词的词嵌入合并进入RNN。
对比seq2seq的改进：
seq2seq永远是上一个RNN里面的最后那一个词的输出，现在是把所有的词拿出来对它做一个weight（做一个加权的平均），然后根据不同的词，比如一开始用前面的那些输出，到后面就用后面那些词的输出。
总结
①Seq2seq中通过隐状态在编码器和解码器中传递信息。
②注意力机制可以根据解码器RNN的输出来匹配到合适的编码器RNN的输出来更有效的传递信息。
Bahdanau注意力代码
import torch
from torch import nn
from d2l import torch as d2l
import os
带有注意力机制的解码器基本接口
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
实现带有Bahdanau注意力的循环神经网络解码器
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
测试Bahdanau注意力解码器
# 创建编码器和解码器的实例
encoder = d2l.Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
encoder.eval()
decoder = Seq2SeqAttentionDecoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
decoder.eval()
# 创建输入序列X
X = torch.zeros((4, 7), dtype=torch.long)
# 初始化解码器的状态
state = decoder.init_state(encoder(X), None)
# 执行解码器的前向计算
output, state = decoder(X, state)
# 输出结果的形状以及状态信息的长度和形状
print(output.shape)
print(len(state))
print(state[0].shape)
print(len(state[1]))
print(state[1][0].shape)
该部分总代码
import torch
from torch import nn
from d2l import torch as d2l
import os
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
# 这里的query形状是[4, 1, 16]
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
# 这里的context形状是[4, 1, 16]
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
# 创建编码器和解码器的实例
encoder = d2l.Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
encoder.eval()
decoder = Seq2SeqAttentionDecoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
decoder.eval()
# 创建输入序列X
X = torch.zeros((4, 7), dtype=torch.long)
# 初始化解码器的状态
state = decoder.init_state(encoder(X), None)
# 执行解码器的前向计算
output, state = decoder(X, state)
# 输出结果的形状以及状态信息的长度和形状
print(output.shape)
print(len(state))
print(state[0].shape)
print(len(state[1]))
print(state[1][0].shape)
训练
train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = d2l.Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建目标语言的带有注意力机制的解码器实例
decoder = Seq2SeqAttentionDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建编码-解码模型实例
net = d2l.EncoderDecoder(encoder, decoder)
# 训练序列到序列模型
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
d2l.plt.show()
从零实现总代码
import torch
from torch import nn
from d2l import torch as d2l
import os
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
# 这里的query形状是[4, 1, 16]
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
# 这里的context形状是[4, 1, 16]
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
def read_data_nmt():
"""载入 “英语-法语” 数据集 """
# 下载并解压数据集
data_dir = d2l.download_extract('fra-eng')
# 打开数据文件，使用utf-8编码读取文件内容
with open(os.path.join(data_dir, 'fra.txt'), 'r', encoding='utf-8') as f:
# 返回读取的文件内容
return f.read()
def preprocess_nmt(text):
"""预处理 “英语-法语” 数据集"""
def no_space(char, prev_char):
# 检查是否需要在字符之前添加空格
return char in set(',.!?') and prev_char != ''
# 替换特殊字符为空格并转换为小写
text = text.replace('\u202f', ' ').replace('\xa0', ' ').lower()
# 根据规则在字符之前添加空格
out = [
' ' + char if i > 0 and no_space(char, text[i - 1]) else char
for i, char in enumerate(text)]
# 将处理后的字符列表连接成字符串并返回
return ''.join(out)
def tokenize_nmt(text, num_examples=None):
"""词元化 “英语-法语” 数据数据集 """
# 初始化源语言和目标语言的列表
source, target = [], []
# 对每行文本进行遍历
for i, line in enumerate(text.split('\n')):
# 如果指定了num_examples，并且超过了指定数量，则退出循环
if num_examples and i > num_examples:
break
# 将每行文本按制表符分割成源语言和目标语言的部分
parts = line.split('\t')
# 如果分割后有两个部分，说明包含了源语言和目标语言的内容
if len(parts) == 2:
# 将源语言部分以空格为分隔符进行词元化，存储到源语言列表中
source.append(parts[0].split(' '))
# 将目标语言部分以空格为分隔符进行词元化，存储到目标语言列表中
target.append(parts[1].split(' '))
# 返回词元化后的源语言列表和目标语言列表
return source, target
def truncate_pad(line, num_steps, padding_token):
"""截断或填充文本序列"""
# 如果文本序列的长度超过了指定的步数
if len(line) > num_steps:
# 截断文本序列，保留前num_steps个元素
return line[:num_steps]
# 在文本序列末尾填充padding_token，使其长度达到num_steps
return line + [padding_token] * (num_steps - len(line))
def build_array_nmt(lines, vocab, num_steps):
"""将机器翻译的文本序列转换成小批量"""
# 将文本序列中的每个词根据词汇表转换为对应的索引
lines = [vocab[l] for l in lines]
# 在每个文本序列末尾添加<eos>表示句子结束的索引
lines = [l + [vocab['<eos>']] for l in lines]
# 将每个文本序列截断或填充为固定长度，并转换为tensor形式
array = torch.tensor([truncate_pad(l, num_steps, vocab['<pad>']) for l in lines])
# 计算有效长度，即非填充部分的长度
valid_len = (array != vocab['<pad>']).type(torch.int32).sum(1)
# 返回转换后的tensor形式的文本序列和有效长度
return array, valid_len
def load_data_nmt(batch_size, num_steps, num_examples=600):
"""返回翻译数据集的迭代器和词汇表"""
# 载入并预处理文本数据集
text = preprocess_nmt(read_data_nmt())
# 将文本数据集进行词元化
source, target = tokenize_nmt(text, num_examples)
# 构建源语言的词汇表对象
src_vocab = d2l.Vocab(source, min_freq=2,
reserved_tokens=['<pad>', '<bos>', '<eos>'])
# 构建目标语言的词汇表对象
tgt_vocab = d2l.Vocab(target, min_freq=2,
reserved_tokens=['<pad>', '<bos>', '<eos>'])
# 将源语言文本序列转换为小批量数据，并获取有效长度
src_array, src_valid_len = build_array_nmt(source, src_vocab, num_steps)
# 将目标语言文本序列转换为小批量数据，并获取有效长度
tgt_array, tgt_valid_len = build_array_nmt(target, tgt_vocab, num_steps)
# 构建数据集的数组
data_arrays = (src_array, src_valid_len, tgt_array, tgt_valid_len)
# 构建数据迭代器
data_iter = d2l.load_array(data_arrays, batch_size)
# 返回数据迭代器和词汇表对象
return data_iter, src_vocab, tgt_vocab
embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 250, d2l.try_gpu()
train_iter, src_vocab, tgt_vocab = load_data_nmt(batch_size, num_steps)
encoder = d2l.Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建目标语言的带有注意力机制的解码器实例
decoder = Seq2SeqAttentionDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建编码-解码模型实例
net = d2l.EncoderDecoder(encoder, decoder)
# 训练序列到序列模型
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
d2l.plt.show()
简洁实现代码
import torch
from torch import nn
from d2l import torch as d2l
import os
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
# 这里的query形状是[4, 1, 16]
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
# 这里的context形状是[4, 1, 16]
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 250, d2l.try_gpu()
train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = d2l.Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建目标语言的带有注意力机制的解码器实例
decoder = Seq2SeqAttentionDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建编码-解码模型实例
net = d2l.EncoderDecoder(encoder, decoder)
# 训练序列到序列模型
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
d2l.plt.show()
将几个英语句子翻译成法语
# 将几个英语句子翻译成汉语
# 英语句子列表
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):
# 使用训练好的模型net对英语句子进行翻译，并获取注意力权重序列
translation, dec_attention_weight_seq = d2l.predict_seq2seq(net, eng, src_vocab, tgt_vocab, num_steps, device, True)
# 打印英语句子、翻译结果和BLEU分数
print(f' {eng} => {translation}, ',
f'bleu {d2l.bleu(translation, fra, k=2):.3f}')
该部分总代码
import torch
from torch import nn
from d2l import torch as d2l
import os
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
# 这里的query形状是[4, 1, 16]
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
# 这里的context形状是[4, 1, 16]
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 250, d2l.try_gpu()
train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = d2l.Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建目标语言的带有注意力机制的解码器实例
decoder = Seq2SeqAttentionDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建编码-解码模型实例
net = d2l.EncoderDecoder(encoder, decoder)
# 训练序列到序列模型
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
# 将几个英语句子翻译成汉语
# 英语句子列表
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):
# 使用训练好的模型net对英语句子进行翻译，并获取注意力权重序列
translation, dec_attention_weight_seq = d2l.predict_seq2seq(net, eng, src_vocab, tgt_vocab, num_steps, device, True)
# 打印英语句子、翻译结果和BLEU分数
print(f' {eng} => {translation}, ',
f'bleu {d2l.bleu(translation, fra, k=2):.3f}')
将注意力权重序列进行拼接
# 将注意力权重序列进行拼接，并调整形状
attention_weights = torch.cat([step[0][0][0] for step in dec_attention_weight_seq], 0).reshape((1, 1, -1, num_steps))
可视化注意力权重
# 可视化注意力权重
# 显示注意力权重的热图，仅显示与输入英语句子对应的位置
d2l.show_heatmaps(attention_weights[:,:,:,:len(engs[-1].split()) + 1].cpu(),
xlabel = 'key positions', ylabel = 'Query posistions')
该部分总代码
import torch
from torch import nn
from d2l import torch as d2l
import os
# 带有注意力机制的解码器基本接口
class AttentionDecoder(d2l.Decoder):
"""带有注意力机制的解码器基本接口"""
def __init__(self, **kwargs):
super(AttentionDecoder, self).__init__(**kwargs)
@property
def attention_weight(self):
raise NotImplementedError
# 实现带有Bahdanau注意力的循环神经网络解码器
class Seq2SeqAttentionDecoder(AttentionDecoder):
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
# 调用父类AttentionDecoder的构造函数进行初始化
super(Seq2SeqAttentionDecoder, self).__init__(**kwargs)
# 创建一个加性注意力机制的实例，用于计算注意力权重
# 这里的key、value、query长度都是num_hiddens，所以可以用点乘的attention（Scaled Dot-Product Attention）
# 这里用加性的是因为这里可以学参数，效果会好一点
self.attention = d2l.AdditiveAttention(num_hiddens, num_hiddens, num_hiddens, dropout)
# 这三行和seq2seq一样的
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
# 这里相比之前多了enc_valid_lens，需要告诉哪些是填充的
def init_state(self, enc_outputs, enc_valid_lens, *args):
outputs, hidden_state = enc_outputs
# 对outputs进行维度变换，将batch维和时间步维交换，保持与解码器输入的一致性
# 返回初始化的解码器隐藏状态
return (outputs.permute(1, 0, 2), hidden_state, enc_valid_lens)
def forward(self, X, state):
enc_outputs, hidden_state, enc_valid_lens = state
# 对输入序列进行嵌入表示，并进行维度变换，将batch维和时间步维交换
X = self.embedding(X).permute(1, 0, 2)
outputs, self._attention_weights = [], []
# 每一步的context会变，所以对于decode的输入的每一个x
for x in X:
# hidden_state[-1]是取上一个时间的RNN的输出（最后一层的输出）
# 获取当前时间步的查询向量，将隐藏状态的最后一个时间步的特征进行维度扩展
# 加一个number of query这个维度进去，虽然这个query只有一个
# 这里的query形状是[4, 1, 16]
query = torch.unsqueeze(hidden_state[-1], dim=1)
# ⭐计算注意力上下文向量，通过注意力机制对编码器的输出进行加权求和
# query就是上一个时刻的最后一层的输出
# key就是enc_outputs（形状：[批量大小、时间步、num_hiddens]），value就是enc_outputs
# enc_valid_lens是一个长为bath_size的一个向量，其中第i个元素表示第i个样本的原始长度是几，因为不算填充
# 这里的context形状是[4, 1, 16]
context = self.attention(query, enc_outputs, enc_outputs, enc_valid_lens)
# 将注意力上下文向量与当前时间步的输入进行拼接，用于输入到循环神经网络
# 因为context里面多了一个长为1的维度（就是那个key的维度），所以x也加了一个维度，然后在最后一个维度concat起来
x = torch.cat((context, torch.unsqueeze(x, dim=1)), dim=-1)
# 执行循环神经网络的前向计算，得到输出和更新后的隐藏状态
out, hidden_state = self.rnn(x.permute(1, 0, 2), hidden_state)
# 将输出和注意力权重添加到对应的列表中
outputs.append(out)
self._attention_weights.append(self.attention.attention_weights)
# 将输出列表中的结果进行拼接，并通过线性层进行映射
outputs = self.dense(torch.cat(outputs, dim=0))
# 对输出进行维度变换，将batch维和时间步维交换，并返回更新后的状态信息
return outputs.permute(1, 0, 2), [enc_outputs, hidden_state, enc_valid_lens]
@property
def attention_weights(self):
# 返回注意力权重
return self._attention_weights
embed_size, num_hiddens, num_layers, dropout = 32, 32, 2, 0.1
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 250, d2l.try_gpu()
train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)
encoder = d2l.Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建目标语言的带有注意力机制的解码器实例
decoder = Seq2SeqAttentionDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
# 创建编码-解码模型实例
net = d2l.EncoderDecoder(encoder, decoder)
# 训练序列到序列模型
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
# 将几个英语句子翻译成汉语
# 英语句子列表
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):
# 使用训练好的模型net对英语句子进行翻译，并获取注意力权重序列
translation, dec_attention_weight_seq = d2l.predict_seq2seq(net, eng, src_vocab, tgt_vocab, num_steps, device, True)
# 打印英语句子、翻译结果和BLEU分数
print(f' {eng} => {translation}, ',
f'bleu {d2l.bleu(translation, fra, k=2):.3f}')
# 将注意力权重序列进行拼接，并调整形状
attention_weights = torch.cat([step[0][0][0] for step in dec_attention_weight_seq], 0).reshape((1, 1, -1, num_steps))
# 可视化注意力权重
# 显示注意力权重的热图，仅显示与输入英语句子对应的位置
d2l.show_heatmaps(attention_weights[:, :, :, :len(engs[-1].split()) + 1].cpu(),
xlabel='key positions', ylabel='Query posistions')
d2l.plt.show()
