AUGUST的深度学习笔记（四，现代循环神经网络与注意力机制）

开源代码
2025-09-10 09:57:02

AUGUST的深度学习笔记（四）

主要参考动手学深度学习及其电子jupyter notebook文档写成，动手学深度学习是我目前所见过的可实操性最好的深度学习教科书，我本篇笔记之前，我其实已经写过多篇深度学习笔记，可以在我的个人动态中找到。本篇笔记主要关注在原书籍的第九章，第十章，第十四章，第十五章。分别是现代循环神经网络，注意力机制，自然语言处理。侧重于为后面大语言模型的学习奠定基础。

现代循环神经网络

在前面的循环神经网络中，提到循环神经网络对于处理序列模型很占优势，并且在此基础上实现了基于循环神经网络的语言模型。但是循环神经网络仍然具有其缺点，例如，循环神经网络在实践中一个常见问题是数值不稳定性。尽管我们已经应用了梯度裁剪等技巧来缓解这个问题，但是仍需要通过设计更复杂的序列模型来进一步处理它。具体来说，我们将引入两个广泛使用的网络，即门控循环单元（gated recurrent units，GRU）和长短期记忆网络（long short-term memory，LSTM）。然后，我们将基于一个单向隐藏层来扩展循环神经网络架构。

事实上，语言建模只揭示了序列学习能力的冰山一角。在各种序列学习问题中，如自动语音识别、文本到语音转换和机器翻译，输入和输出都是任意长度的序列。为了阐述如何拟合这种类型的数据，我们将以机器翻译为例介绍基于循环神经网络的“编码器－解码器”架构和束搜索，并用它们来生成序列。

GRU循环神经网络

门控循环单元(GRU)门控循环单元与普通的循环神经网络之间的关键区别在于：前者支持隐状态的门控。这意味着模型有专门的机制来确定应该何时更新隐状态，以及应该何时重置隐状态。这些机制是可学习的，并且能够解决了上面列出的问题。例如，如果第一个词元非常重要，模型将学会在第一次观测之后不更新隐状态。同样，模型也可以学会跳过不相关的临时观测。最后，模型还将学会在需要的时候重置隐状态。借助GRU的循环神经网络实现代码如下：

import torch from torch import nn import d2l batch_size, num_steps = 32, 35 train_iter, vocab = d2l.load_data_time_machine(batch_size, num_steps) def get_params(vocab_size, num_hiddens, device): num_inputs = num_outputs = vocab_size def normal(shape): return torch.randn(size=shape, device=device)*0.01 def three(): return (normal((num_inputs, num_hiddens)), normal((num_hiddens, num_hiddens)), torch.zeros(num_hiddens, device=device)) W_xz, W_hz, b_z = three() # 更新门参数 W_xr, W_hr, b_r = three() # 重置门参数 W_xh, W_hh, b_h = three() # 候选隐状态参数 # 输出层参数 W_hq = normal((num_hiddens, num_outputs)) b_q = torch.zeros(num_outputs, device=device) # 附加梯度 params = [W_xz, W_hz, b_z, W_xr, W_hr, b_r, W_xh, W_hh, b_h, W_hq, b_q] for param in params: param.requires_grad_(True) return params def init_gru_state(batch_size, num_hiddens, device):#定义初始的状态 return (torch.zeros((batch_size, num_hiddens), device=device), ) def gru(inputs, state, params): W_xz, W_hz, b_z, W_xr, W_hr, b_r, W_xh, W_hh, b_h, W_hq, b_q = params H, = state outputs = [] for X in inputs: Z = torch.sigmoid((X @ W_xz) + (H @ W_hz) + b_z) R = torch.sigmoid((X @ W_xr) + (H @ W_hr) + b_r) H_tilda = torch.tanh((X @ W_xh) + ((R * H) @ W_hh) + b_h) H = Z * H + (1 - Z) * H_tilda Y = H @ W_hq + b_q outputs.append(Y) return torch.cat(outputs, dim=0), (H,) vocab_size, num_hiddens, device = len(vocab), 256, d2l.try_gpu()#调用GPU num_epochs, lr = 500, 1#训练轮数和学习率 model = d2l.RNNModelScratch(len(vocab), num_hiddens, device, get_params, init_gru_state, gru)#定义模型 d2l.train_ch8(model, train_iter, vocab, lr, num_epochs, device)#设定是用GOU加速

训练过程如下所示：

LSTM（长短期记忆网络）

长期以来，隐变量模型存在着长期信息保存和短期输入缺失的问题。解决这一问题的最早方法之一是长短期存储器（long short-term memory，LSTM）。它有许多与门控循环单元（ :numref:sec_gru）一样的属性。有趣的是，长短期记忆网络的设计比门控循环单元稍微复杂一些，却比门控循环单元早诞生了近20年。

可以说，长短期记忆网络的设计灵感来自于计算机的逻辑门。长短期记忆网络引入了记忆元（memory cell），或简称为单元（cell）。有些文献认为记忆元是隐状态的一种特殊类型，它们与隐状态具有相同的形状，其设计目的是用于记录附加的信息。为了控制记忆元，我们需要许多门。其中一个门用来从单元中输出条目，我们将其称为输出门（output gate）。另外一个门用来决定何时将数据读入单元，我们将其称为输入门（input gate）。我们还需要一种机制来重置单元的内容，由遗忘门（forget gate）来管理，这种设计的动机与门控循环单元相同，能够通过专用机制决定什么时候记忆或忽略隐状态中的输入。

构建lstm循环神经网络的代码如下：

import torch from torch import nn import d2l batch_size, num_steps = 32, 35 train_iter, vocab = d2l.load_data_time_machine(batch_size, num_steps)#加载数据 #batch_size:批量大小，num_steps:时间步数 #train_iter:数据迭代器，vocab:词汇表 def get_lstm_params(vocab_size, num_hiddens, device):#获取参数 num_inputs = num_outputs = vocab_size def normal(shape): return torch.randn(size=shape, device=device)*0.01 def three(): return (normal((num_inputs, num_hiddens)), normal((num_hiddens, num_hiddens)), torch.zeros(num_hiddens, device=device)) W_xi, W_hi, b_i = three() # 输入门参数 W_xf, W_hf, b_f = three() # 遗忘门参数 W_xo, W_ho, b_o = three() # 输出门参数 W_xc, W_hc, b_c = three() # 候选记忆元参数 # 输出层参数 W_hq = normal((num_hiddens, num_outputs)) b_q = torch.zeros(num_outputs, device=device) # 附加梯度 params = [W_xi, W_hi, b_i, W_xf, W_hf, b_f, W_xo, W_ho, b_o, W_xc, W_hc, b_c, W_hq, b_q] for param in params: param.requires_grad_(True) return params def init_lstm_state(batch_size, num_hiddens, device):#初始化lstm模型的状态 return (torch.zeros((batch_size, num_hiddens), device=device), torch.zeros((batch_size, num_hiddens), device=device)) #用于初始化隐藏状态和记忆元的函数 def lstm(inputs, state, params):#定义lstm模型，定义了前向传播 [W_xi, W_hi, b_i, W_xf, W_hf, b_f, W_xo, W_ho, b_o, W_xc, W_hc, b_c, W_hq, b_q] = params H, C = state outputs = [] for X in inputs: I = torch.sigmoid(torch.matmul(X, W_xi) + torch.matmul(H, W_hi) + b_i) F = torch.sigmoid(torch.matmul(X, W_xf) + torch.matmul(H, W_hf) + b_f) O = torch.sigmoid(torch.matmul(X, W_xo) + torch.matmul(H, W_ho) + b_o) C_tilde = torch.tanh(torch.matmul(X, W_xc) + torch.matmul(H, W_hc) + b_c) C = F * C + I * C_tilde H = O * torch.tanh(C) Y = torch.matmul(H, W_hq) + b_q outputs.append(Y) return torch.cat(outputs, dim=0), (H, C) vocab_size, num_hiddens, device = len(vocab), 256, d2l.try_gpu()#定义参数 num_epochs, lr = 500, 1 model = d2l.RNNModelScratch(len(vocab), num_hiddens, device, get_lstm_params, init_lstm_state, lstm) d2l.train_ch8(model, train_iter, vocab, lr, num_epochs, device)

训练的效果如下图所示：

深度循环神经网络

深度循环神经网络（Deep Recurrent Neural Network，简称Deep RNN）是指具有多个隐藏层的循环神经网络。在标准的循环神经网络中，通常只有一个隐藏层，而在深度循环神经网络中，隐藏层被堆叠起来，形成更深的网络结构。

以下是深度循环神经网络的一些关键特点：

多个隐藏层：与深度前馈神经网络类似，深度循环神经网络通过增加隐藏层的数量来提高模型的表达能力。每一层都可以有一个循环连接，使得信息可以在时间维度上传播，并且在不同层之间传递。

更复杂的模式识别：由于有了更多的隐藏层，深度循环神经网络能够学习更复杂的序列模式和数据表示。

梯度消失和梯度爆炸问题：在深度网络中，尤其是在长序列学习中，梯度消失和梯度爆炸问题变得更加严重。这通常需要特殊的技术来解决，如使用长短期记忆网络（LSTM）或门控循环单元（GRU）作为循环单元，或者采用梯度裁剪等策略。

训练难度：深度循环神经网络的训练通常比浅层网络更困难，因为需要更复杂的优化算法和更多的计算资源。

应用领域：深度循环神经网络适用于需要处理长序列数据和时间序列数据的任务，例如语音识别、机器翻译、文本生成、视频分析等。

示例代码如下：

import torch from torch import nn import d2l batch_size, num_steps = 32, 35 train_iter, vocab = d2l.load_data_time_machine(batch_size, num_steps) vocab_size, num_hiddens, num_layers = len(vocab), 256, 2#词汇表大小，隐藏层单元的数量，lstm层的数量 num_inputs = vocab_size#输入层的大小，与词汇表的大小相同 device = d2l.try_gpu() lstm_layer = nn.LSTM(num_inputs, num_hiddens, num_layers)#多层隐藏层，调用nn.LSTM model = d2l.RNNModel(lstm_layer, len(vocab)) model = model.to(device) num_epochs, lr = 500, 2 d2l.train_ch8(model, train_iter, vocab, lr*1.0, num_epochs, device)

值得注意的是，由于Lstm层的数量增加，训练时间大大增加。

双向循环神经网络

双向循环神经网络（Bidirectional Recurrent Neural Network，简称BiRNN）是一种特殊的循环神经网络（RNN），它在处理序列数据时能够同时考虑序列的前向和后向信息。这种网络结构通过两个独立的隐藏层（或称为隐藏状态流）来捕获输入序列的前向和后向上下文信息，然后将这两个隐藏层的信息合并起来以进行最终的预测。

以下是双向循环神经网络的主要特点：

两个隐藏层：BiRNN包含两个RNN层，一个处理正向输入序列（从序列的开始到结束），另一个处理反向输入序列（从序列的结束到开始）。

合并层：在每一个时间步，正向和反向隐藏层的输出被合并（通常是通过拼接或求和），以产生该时间步的最终输出。

上下文信息：由于BiRNN同时考虑了序列的前向和后向信息，因此它在处理诸如自然语言文本这样的序列数据时能够更准确地捕捉上下文信息。

应用广泛：BiRNN在自然语言处理（NLP）任务中特别有用，如文本分类、序列标注、语音识别等，因为它能够更好地理解单词在句子中的上下文。

以下是双向循环神经网络的工作流程：

正向传播：输入序列从时间步0开始，依次通过正向RNN层，每个时间步的输出被传递到下一个时间步。

反向传播：同时，输入序列从最后一个时间步开始，依次通过反向RNN层，每个时间步的输出被传递到前一个时间步。

合并输出：在每一个时间步，将正向和反向RNN层的隐藏状态合并，得到该时间步的最终输出。

输出层：合并后的隐藏状态可以用于生成最终的预测，例如通过一个全连接层进行分类或回归任务。

import torch from torch import nn import d2l # 加载数据 batch_size, num_steps, device = 32, 35, d2l.try_gpu() train_iter, vocab = d2l.load_data_time_machine(batch_size, num_steps) # 通过设置“bidirective=True”来定义双向LSTM模型 vocab_size, num_hiddens, num_layers = len(vocab), 256, 2 num_inputs = vocab_size lstm_layer = nn.LSTM(num_inputs, num_hiddens, num_layers, bidirectional=True) model = d2l.RNNModel(lstm_layer, len(vocab)) model = model.to(device) # 训练模型 num_epochs, lr = 500, 1 d2l.train_ch8(model, train_iter, vocab, lr, num_epochs, device)

示例代码如上：

机器翻译与数据集

语言模型是自然语言处理的关键，而机器翻译是语言模型最成功的基准测试。因为机器翻译正是将输入序列转换成输出序列的序列转换模型（sequence transduction）的核心问题。序列转换模型在各类现代人工智能应用中发挥着至关重要的作用，因此我们将其做为本章剩余部分和 :numref:chap_attention的重点。为此，本节将介绍机器翻译问题及其后文需要使用的数据集。

本节关注的是神经网络机器翻译方法，强调的是端到端的学习。加载数据集和处理的代码如下所示：

import os import torch import d2l #@save d2l.DATA_HUB['fra-eng'] = (d2l.DATA_URL + 'fra-eng.zip', '94646ad1522d915e7b0f9296181140edcf86a4f5') #@save def read_data_nmt(): """载入“英语－法语”数据集""" data_dir = d2l.download_extract('fra-eng') with open(os.path.join(data_dir, 'fra.txt'), 'r', encoding='utf-8') as f: return f.read() raw_text = read_data_nmt() print(raw_text[:75]) #@save 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) text = preprocess_nmt(raw_text) print(text[:80]) #@save def tokenize_nmt(text, num_examples=None): """词元化“英语－法语”数据数据集""" source, target = [], [] for i, line in enumerate(text.split('\n')): 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 source, target = tokenize_nmt(text) source[:6], target[:6] #@save def show_list_len_pair_hist(legend, xlabel, ylabel, xlist, ylist): """绘制列表长度对的直方图""" d2l.set_figsize() _, _, patches = d2l.plt.hist( [[len(l) for l in xlist], [len(l) for l in ylist]]) d2l.plt.xlabel(xlabel) d2l.plt.ylabel(ylabel) for patch in patches[1].patches: patch.set_hatch('/') d2l.plt.legend(legend) show_list_len_pair_hist(['source', 'target'], '# tokens per sequence', 'count', source, target) src_vocab = d2l.Vocab(source, min_freq=2, reserved_tokens=['<pad>', '<bos>', '<eos>']) len(src_vocab) #@save def truncate_pad(line, num_steps, padding_token): """截断或填充文本序列""" if len(line) > num_steps: return line[:num_steps] # 截断 return line + [padding_token] * (num_steps - len(line)) # 填充 truncate_pad(src_vocab[source[0]], 10, src_vocab['<pad>']) #@save def build_array_nmt(lines, vocab, num_steps): """将机器翻译的文本序列转换成小批量""" lines = [vocab[l] for l in lines] lines = [l + [vocab['<eos>']] for l in lines] array = torch.tensor([truncate_pad( l, num_steps, vocab['<pad>']) for l in lines]) valid_len = (array != vocab['<pad>']).type(torch.int32).sum(1) return array, valid_len #@save 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 train_iter, src_vocab, tgt_vocab = load_data_nmt(batch_size=2, num_steps=8) for X, X_valid_len, Y, Y_valid_len in train_iter: print('X:', X.type(torch.int32)) print('X的有效长度:', X_valid_len) print('Y:', Y.type(torch.int32)) print('Y的有效长度:', Y_valid_len) break

绘制出来的列表长度对的直方图如下所示：

这里补充说明一下Token是什么概念，token在大模型中经常涉及，比如购买多少万token多少钱，其实就是指的是让AI生成了多少东西，拿这个生成的给人来算钱。

在深度学习中，尤其是在自然语言处理（NLP）领域，"token"是一个常用的术语，它有不同的含义和用法，以下是一些常见的解释：

最小的语言单位：在许多情况下，token指的是文本中的最小单元，可以是单词、字符或者子词（subword）。例如，在处理英文文本时，单词通常被视为token。

单词token：在"the quick brown fox"这句话中，“the”、“quick”、“brown”、"fox"都是token。

字符token：如果以字符为单位，那么"t"、“h”、"e"等都是token。

子词token：在处理稀有词汇或形态丰富的语言时，可以使用子词token，如"un-", “friend-”, “ly”。

序列中的元素：在序列模型中，token指的是序列中的一个元素，可以是文本中的一个词、一个字符或者一个更复杂的单元。在处理序列数据时，模型通常是以token为单位进行操作的。

唯一标识符：在构建词嵌入（word embeddings）或处理大规模文本数据时，每个不同的token（如单词）会被分配一个唯一的整数标识符（ID）。这个过程称为tokenization，它是将文本转换为机器可以理解的数字表示的步骤。

输入表示：在深度学习模型中，token也可以指输入层中的一个元素，比如在Transformer模型中，输入序列中的每个词或子词都被转换为一个向量表示，这些向量就是模型的输入token。

特殊标记：在NLP中，还常常使用特殊token来表示特定的意义，如：

[CLS]：分类任务的开始标记。

[SEP]：序列分隔符，用于分隔句子或不同的片段。

[PAD]：填充标记，用于将序列填充到相同的长度。

总的来说，token是深度学习处理文本数据时的基本单元，它可以是单词、字符、子词或其他语言单位，其目的是将原始文本转换为机器学习模型可以理解和处理的形式。

编码器和解码器

在深度学习中，编码器（Encoder）和解码器（Decoder）是序列到序列（sequence to sequence）模型中的两个核心组件，它们分别用于处理输入序列和生成输出序列。编码器和解码器在机器翻译中经常会用到。以下是对编码器和解码器的基本解释：

编码器（Encoder）

编码器是一个神经网络模块，它的作用是将输入序列（如一段文本或语音信号）转换为一个固定大小的内部表示（通常是一个向量），这个内部表示捕获了输入序列的语义信息。编码器可以是一个循环神经网络（RNN）、长短期记忆网络（LSTM）、门控循环单元（GRU）或者更复杂的结构，如Transformer中的自注意力机制。

主要功能：

读取输入序列，并逐步构建一个上下文丰富的内部表示。

输出一个固定大小的向量（通常是最后一个时间步的隐藏状态），这个向量可以作为输入序列的摘要或编码。

解码器（Decoder）

解码器是另一个神经网络模块，它根据编码器提供的内部表示来生成输出序列。解码器通常会使用编码器的输出作为初始状态，并逐步生成输出序列的每个元素。

主要功能：

使用编码器的输出作为初始状态或上下文信息。

逐步生成输出序列，每一步都可能依赖于前一步的输出和编码器的输出。

在序列生成任务中，如机器翻译或文本摘要，解码器通常会预测下一个token，直到生成一个结束标记。

示例代码如下：

from torch import nn #@save class Encoder(nn.Module): """编码器-解码器架构的基本编码器接口""" def __init__(self, **kwargs): super(Encoder, self).__init__(**kwargs) def forward(self, X, *args): raise NotImplementedError #@save class Decoder(nn.Module): """编码器-解码器架构的基本解码器接口""" def __init__(self, **kwargs): super(Decoder, self).__init__(**kwargs) def init_state(self, enc_outputs, *args): raise NotImplementedError def forward(self, X, state): raise NotImplementedError #@save class EncoderDecoder(nn.Module): """编码器-解码器架构的基类""" def __init__(self, encoder, decoder, **kwargs): super(EncoderDecoder, self).__init__(**kwargs) self.encoder = encoder self.decoder = decoder def forward(self, enc_X, dec_X, *args): enc_outputs = self.encoder(enc_X, *args) dec_state = self.decoder.init_state(enc_outputs, *args) return self.decoder(dec_X, dec_state) 序列到序列学习

正如我们在 :numref:sec_machine_translation中看到的，机器翻译中的输入序列和输出序列都是长度可变的。为了解决这类问题，我们在 :numref:sec_encoder-decoder中设计了一个通用的”编码器－解码器“架构。本节，我们将使用两个循环神经网络的编码器和解码器，并将其应用于序列到序列（sequence to sequence，seq2seq）类的学习任务。遵循编码器－解码器架构的设计原则，循环神经网络编码器使用长度可变的序列作为输入，将其转换为固定形状的隐状态。换言之，输入序列的信息被编码到循环神经网络编码器的隐状态中。为了连续生成输出序列的词元，独立的循环神经网络解码器是基于输入序列的编码信息和输出序列已经看见的或者生成的词元来预测下一个词元。

序列到序列学习的示例代码如下：

import collections import math import torch from torch import nn import d2l #@save class Seq2SeqEncoder(d2l.Encoder): """用于序列到序列学习的循环神经网络编码器""" def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **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) def forward(self, X, *args): # 输出'X'的形状：(batch_size,num_steps,embed_size) X = self.embedding(X) # 在循环神经网络模型中，第一个轴对应于时间步 X = X.permute(1, 0, 2) # 如果未提及状态，则默认为0 output, state = self.rnn(X) # output的形状:(num_steps,batch_size,num_hiddens) # state的形状:(num_layers,batch_size,num_hiddens) return output, state 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.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) self.dense = nn.Linear(num_hiddens, vocab_size) def init_state(self, enc_outputs, *args): return enc_outputs[1] def forward(self, X, state): # 输出'X'的形状：(batch_size,num_steps,embed_size) X = self.embedding(X).permute(1, 0, 2) # 广播context，使其具有与X相同的num_steps context = state[-1].repeat(X.shape[0], 1, 1) X_and_context = torch.cat((X, context), 2) output, state = self.rnn(X_and_context, state) output = self.dense(output).permute(1, 0, 2) # 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 #@save def sequence_mask(X, valid_len, value=0): """在序列中屏蔽不相关的项""" maxlen = X.size(1) mask = torch.arange((maxlen), dtype=torch.float32, device=X.device)[None, :] < valid_len[:, None] X[~mask] = value return X X = 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) #@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): weights = torch.ones_like(label) weights = sequence_mask(weights, valid_len) self.reduction='none' unweighted_loss = super(MaskedSoftmaxCELoss, self).forward( pred.permute(0, 2, 1), label) 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])) #@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] 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): """序列到序列模型的预测""" # 在预测时将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>']: break output_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])] += 1 for i in range(len_pred - n + 1): if label_subs[' '.join(pred_tokens[i: i + n])] > 0: num_matches += 1 label_subs[' '.join(pred_tokens[i: i + n])] -= 1 score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n)) 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}')

训练过程如下所示：

此时的预测结果是很差的，需要指出，这是因为加载的数据集太小而且采用的技术不够先进。

注意力机制

注意力机制（Attention Mechanism）是深度学习中的一个重要概念，特别是在自然语言处理（NLP）和计算机视觉等领域。注意力机制的灵感来源于人类视觉系统：当我们观察一个场景时，并不会均匀地处理场景中的所有信息，而是会选择性地关注某些部分，对这些部分投入更多的注意力。在深度学习中，注意力机制模仿了这一过程，允许模型在处理数据时更加聚焦于重要的部分。

以下是注意力机制的基本原理和组成部分：

基本原理

权重分配：注意力机制会为输入序列中的每个元素分配一个权重，这个权重表示该元素在当前任务中的重要性。权重通常是通过一个可学习的函数计算得出的。

加权求和：根据分配的权重，对输入序列中的元素进行加权求和，得到一个加权表示，这个表示更加关注于重要的信息。

上下文向量：加权求和的结果称为上下文向量（context vector），它包含了输入序列中最重要的信息，用于后续的处理。

注意力汇聚：Nadaraya-Watson 核回归

注意力汇聚中又分为有参数注意力汇聚和无参数注意力汇聚，二者的区别如下所示：

有参数注意力汇聚和无参数注意力汇聚的主要区别在于它们如何计算注意力权重。以下是两种方法的详细对比：

有参数注意力汇聚（Parameterized Attention Pooling）

权重学习：在有参数的注意力汇聚中，注意力权重是通过学习得到的，通常是通过一个可学习的神经网络来计算。这意味着权重是模型参数的一部分，会在训练过程中通过梯度下降等优化算法进行更新。

复杂度：由于权重是通过学习得到的，这种方法可以捕捉到更加复杂的注意力模式，适应更加多样化的数据分布。

灵活性：有参数的注意力机制更加灵活，可以应用于更广泛的场景，包括但不限于序列模型、图像识别、自然语言处理等。

训练需求：这种方法需要大量的训练数据来学习有效的注意力权重。

示例：在Transformer模型中使用的自注意力机制就是一个有参数注意力汇聚的例子。它通过查询（Q）、键（K）和值（V）的线性变换来计算注意力权重。

无参数注意力汇聚（Non-Parameterized Attention Pooling）

固定权重：在无参数的注意力汇聚中，注意力权重不是通过学习得到的，而是基于固定的规则或函数计算。例如，在Nadaraya-Watson核回归中，权重是通过核函数直接计算得到的。

简单性：这种方法通常比较简单，不需要学习权重，因此计算量较小。

限制性：由于权重不是学习的，这种方法可能无法捕捉到数据中的复杂关系，适用性可能有限。

无需训练：无参数注意力汇聚不需要训练过程来学习权重，因此可以在没有训练数据的情况下使用。

示例：Nadaraya-Watson核回归就是一个无参数注意力汇聚的例子，其中权重是通过核函数和输入数据点之间的距离来计算的。

部分相关代码如下所示：

import torch from torch import nn import d2l n_train = 50 # 训练样本数 x_train, _ = torch.sort(torch.rand(n_train) * 5) # 排序后的训练样本 def f(x): return 2 * torch.sin(x) + x**0.8 y_train = f(x_train) + torch.normal(0.0, 0.5, (n_train,)) # 训练样本的输出 x_test = torch.arange(0, 5, 0.1) # 测试样本 y_truth = f(x_test) # 测试样本的真实输出 n_test = len(x_test) # 测试样本数 n_test def plot_kernel_reg(y_hat):#定义热力图的绘制函数 d2l.plot(x_test, [y_truth, y_hat], 'x', 'y', legend=['Truth', 'Pred'], xlim=[0, 5], ylim=[-1, 5]) d2l.plt.plot(x_train, y_train, 'o', alpha=0.5) y_hat = torch.repeat_interleave(y_train.mean(), n_test) plot_kernel_reg(y_hat) # X_repeat的形状:(n_test,n_train), # 每一行都包含着相同的测试输入（例如：同样的查询） X_repeat = x_test.repeat_interleave(n_train).reshape((-1, n_train)) # x_train包含着键。attention_weights的形状：(n_test,n_train), # 每一行都包含着要在给定的每个查询的值（y_train）之间分配的注意力权重 attention_weights = nn.functional.softmax(-(X_repeat - x_train)**2 / 2, dim=1) # y_hat的每个元素都是值的加权平均值，其中的权重是注意力权重 y_hat = torch.matmul(attention_weights, y_train) plot_kernel_reg(y_hat) d2l.show_heatmaps(attention_weights.unsqueeze(0).unsqueeze(0), xlabel='Sorted training inputs', ylabel='Sorted testing inputs') X = torch.ones((2, 1, 4)) Y = torch.ones((2, 4, 6)) torch.bmm(X, Y).shape weights = torch.ones((2, 10)) * 0.1 values = torch.arange(20.0).reshape((2, 10)) torch.bmm(weights.unsqueeze(1), values.unsqueeze(-1)) class NWKernelRegression(nn.Module): def __init__(self, **kwargs): super().__init__(**kwargs) self.w = nn.Parameter(torch.rand((1,), requires_grad=True)) def forward(self, queries, keys, values): # queries和attention_weights的形状为(查询个数，“键－值”对个数) queries = queries.repeat_interleave(keys.shape[1]).reshape((-1, keys.shape[1])) self.attention_weights = nn.functional.softmax( -((queries - keys) * self.w)**2 / 2, dim=1) # values的形状为(查询个数，“键－值”对个数) return torch.bmm(self.attention_weights.unsqueeze(1), values.unsqueeze(-1)).reshape(-1) # X_tile的形状:(n_train，n_train)，每一行都包含着相同的训练输入 X_tile = x_train.repeat((n_train, 1)) # Y_tile的形状:(n_train，n_train)，每一行都包含着相同的训练输出 Y_tile = y_train.repeat((n_train, 1)) # keys的形状:('n_train'，'n_train'-1) keys = X_tile[(1 - torch.eye(n_train)).type(torch.bool)].reshape((n_train, -1)) # values的形状:('n_train'，'n_train'-1) values = Y_tile[(1 - torch.eye(n_train)).type(torch.bool)].reshape((n_train, -1)) net = NWKernelRegression() loss = nn.MSELoss(reduction='none') trainer = torch.optim.SGD(net.parameters(), lr=0.5) animator = d2l.Animator(xlabel='epoch', ylabel='loss', xlim=[1, 5]) for epoch in range(5): trainer.zero_grad() l = loss(net(x_train, keys, values), y_train) l.sum().backward() trainer.step() print(f'epoch {epoch + 1}, loss {float(l.sum()):.6f}') animator.add(epoch + 1, float(l.sum())) # keys的形状:(n_test，n_train)，每一行包含着相同的训练输入（例如，相同的键） keys = x_train.repeat((n_test, 1)) # value的形状:(n_test，n_train) values = y_train.repeat((n_test, 1)) y_hat = net(x_test, keys, values).unsqueeze(1).detach() plot_kernel_reg(y_hat) d2l.show_heatmaps(net.attention_weights.unsqueeze(0).unsqueeze(0), xlabel='Sorted training inputs', ylabel='Sorted testing inputs') 注意力评分函数

注意力评分函数（Attention Scoring Function）是注意力机制中的一个关键组成部分，它用于计算查询（Query）和键（Key）之间的相关性或匹配程度，从而得到注意力权重。这些权重随后用于对值（Value）进行加权求和，以生成最终的注意力输出。以下是注意力评分函数的详细解释：

基本概念

查询（Query）：通常是我们想要关注的点或问题，它决定了我们将如何从键值对中提取信息。

键（Key）：与查询相关的信息，用于与查询进行比较以计算注意力权重。

值（Value）：实际的数据内容，将根据计算出的注意力权重进行加权求和。

将注意力汇聚的输出计算可以作为值的加权平均，选择不同的注意力评分函数会带来不同的注意力汇聚操作。当查询和键是不同长度的矢量时，可以使用可加性注意力评分函数。当它们的长度相同时，使用缩放的“点－积”注意力评分函数的计算效率更高。示例代码如下，分别采用了两种注意力机制，即缩放点积注意力和加性注意力。

import math import torch from torch import nn import d2l #@save def masked_softmax(X, valid_lens): """通过在最后一个轴上掩蔽元素来执行softmax操作""" # X:3D张量，valid_lens:1D或2D张量 if valid_lens is None: return nn.functional.softmax(X, dim=-1) else: shape = X.shape if valid_lens.dim() == 1: valid_lens = torch.repeat_interleave(valid_lens, shape[1]) else: valid_lens = valid_lens.reshape(-1) # 最后一轴上被掩蔽的元素使用一个非常大的负值替换，从而其softmax输出为0 X = d2l.sequence_mask(X.reshape(-1, shape[-1]), valid_lens, value=-1e6) return nn.functional.softmax(X.reshape(shape), dim=-1) masked_softmax(torch.rand(2, 2, 4), torch.tensor([2, 3])) masked_softmax(torch.rand(2, 2, 4), torch.tensor([[1, 3], [2, 4]])) #@save class AdditiveAttention(nn.Module): """加性注意力""" def __init__(self, key_size, query_size, num_hiddens, dropout, **kwargs): super(AdditiveAttention, self).__init__(**kwargs) self.W_k = nn.Linear(key_size, num_hiddens, bias=False) self.W_q = nn.Linear(query_size, num_hiddens, bias=False) self.w_v = nn.Linear(num_hiddens, 1, bias=False) self.dropout = nn.Dropout(dropout) def forward(self, queries, keys, values, valid_lens): queries, keys = self.W_q(queries), self.W_k(keys) # 在维度扩展后， # queries的形状：(batch_size，查询的个数，1，num_hidden) # key的形状：(batch_size，1，“键－值”对的个数，num_hiddens) # 使用广播方式进行求和 features = queries.unsqueeze(2) + keys.unsqueeze(1) features = torch.tanh(features) # self.w_v仅有一个输出，因此从形状中移除最后那个维度。 # scores的形状：(batch_size，查询的个数，“键-值”对的个数) scores = self.w_v(features).squeeze(-1) self.attention_weights = masked_softmax(scores, valid_lens) # values的形状：(batch_size，“键－值”对的个数，值的维度) return torch.bmm(self.dropout(self.attention_weights), values) queries, keys = torch.normal(0, 1, (2, 1, 20)), torch.ones((2, 10, 2)) # values的小批量，两个值矩阵是相同的 values = torch.arange(40, dtype=torch.float32).reshape(1, 10, 4).repeat( 2, 1, 1) valid_lens = torch.tensor([2, 6]) attention = AdditiveAttention(key_size=2, query_size=20, num_hiddens=8, dropout=0.1) attention.eval() attention(queries, keys, values, valid_lens) d2l.show_heatmaps(attention.attention_weights.reshape((1, 1, 2, 10)), xlabel='Keys', ylabel='Queries') #@save class DotProductAttention(nn.Module): """缩放点积注意力""" def __init__(self, dropout, **kwargs): super(DotProductAttention, self).__init__(**kwargs) self.dropout = nn.Dropout(dropout) # queries的形状：(batch_size，查询的个数，d) # keys的形状：(batch_size，“键－值”对的个数，d) # values的形状：(batch_size，“键－值”对的个数，值的维度) # valid_lens的形状:(batch_size，)或者(batch_size，查询的个数) def forward(self, queries, keys, values, valid_lens=None): d = queries.shape[-1] # 设置transpose_b=True为了交换keys的最后两个维度 scores = torch.bmm(queries, keys.transpose(1,2)) / math.sqrt(d) self.attention_weights = masked_softmax(scores, valid_lens) return torch.bmm(self.dropout(self.attention_weights), values) queries = torch.normal(0, 1, (2, 1, 2)) attention = DotProductAttention(dropout=0.5) attention.eval() attention(queries, keys, values, valid_lens) d2l.show_heatmaps(attention.attention_weights.reshape((1, 1, 2, 10)), xlabel='Keys', ylabel='Queries') Bahdanau 注意力

Bahdanau 注意力是一种在神经网络中用于增强序列到序列（seq2seq）模型能力的注意力机制，尤其是在机器翻译任务中。它由 Bahdanau 等人在 2015 年的论文《Neural Machine Translation by Jointly Learning to Align and Translate》中提出。Bahdanau 注意力的核心思想是在解码过程中，不是将整个编码器的输出视为一个固定的上下文向量，而是根据当前解码器的状态动态地选择编码器输出的相关部分。

以下是 Bahdanau 注意力的主要特点：

动态权重：在解码器的每个时间步，Bahdanau 注意力会计算一个权重系数，这些权重系数决定了编码器输出中每个部分对当前解码步骤的重要性。

加性模型：Bahdanau 注意力使用一个加性模型来计算注意力得分。具体来说，它将解码器当前的隐藏状态和编码器的每个隐藏状态通过一个全连接层（线性层）进行处理，然后将这两个处理后的向量相加，并通过一个 tanh 激活函数，最后通过另一个全连接层得到一个得分。

得分和权重：得到的得分通过 softmax 函数转换为概率分布，这些概率分布即为权重，表示编码器输出中每个部分的相对重要性。

上下文向量：将权重与编码器的隐藏状态相乘并求和，得到一个加权的上下文向量。这个上下文向量随后被用于解码器的当前时间步，帮助生成输出。

联合学习：Bahdanau 注意力机制使得编码器和解码器可以联合学习对齐和翻译，即在翻译的同时学习输入和输出序列之间的对应关系。

Bahdanau 注意力的引入显著提高了 seq2seq 模型在处理长序列时的性能，因为它允许模型更加灵活地关注输入序列的不同部分，而不是仅仅依赖于一个固定的上下文表示。这种方法在机器翻译、语音识别和其他序列处理任务中得到了广泛应用。