Transformer模型完全基于注意力机制，没有任何卷积层或循环神经网络层。
Transformer原先应用于文本数据上序列到序列的学习，现在已经被应用语言、视觉、语音和强化学习领域。
1. 模型架构
Transformer是由编码器和解码器组成的，其编码器和解码器是由自注意力的模块叠加而成的。
源（输入）序列和目标（输出）序列的嵌入 (embedding) 表示将加上位置编码(positional encoding)，再分别输入到编码器和解码器中。
编码器：Transformer的编码器是由多个相同的层叠加而成的，每个层都有两个子层(sublayer)：
- 多头注意力(multi-head self-attention)汇聚：计算自注意力时，query, key, value都来自前一个编码器层的输出。
- 基于位置的前馈网络(positionwise feed-forward network)
每个子层后需要进行add & norm：
- 每个子层都采用了
残差连接(residual connection)。Transformer中，序列中任何位置的输入\(x \in R^d\)，经过子层的输出维度仍为\(sublayer(x) \in R^d\)，以便进行残差连接 \(x + sublayer(x) \in R^d\)。 - 在残差连接之后，应用
层规范化(layer normalization)。
解码器：Transformer的解码器也是由多个相同的层叠加而成的，并且层之间使用了残差连接和层规范化。除了编码器中提到的两个子层，解码器中还包含了编码器-解码器注意力(encoder-decoder attention)。
- 编码器-解码器注意力：query来自上一个解码器层的输出；key, value来自整个编码器的输出
- 解码器自注意力：query, key, value都来自上一个解码器层的输出。
解码器中的每个位置只能考虑该位置之前的位置，这种掩码(masked)注意力保留了自回归属性，确保预测仅依赖于已生成的输出词元。
2. 基于位置的前馈网络
基于位置的前馈网络，本质上一个多层感知机(MLP)。
会改变张量的最里层维度的尺寸：
- 输入：(batch_size, num_steps, ffn_num_input)
- 输出：(batch_size, num_steps, ffn_num_output)
class PositionWiseFFN(nn.Module):"""基于位置的前馈网络"""def __init__(self, ffn_num_input, ffn_num_hiddens, ffn_num_outputs,**kwargs):super(PositionWiseFFN, self).__init__(**kwargs)self.dense1 = nn.Linear(ffn_num_input, ffn_num_hiddens)self.relu = nn.ReLU()self.dense2 = nn.Linear(ffn_num_hiddens, ffn_num_outputs)def forward(self, X):return self.dense2(self.relu(self.dense1(X)))
3. 残差连接和层规范化
- 计算机视觉领域(CV)常用批量归一化(batch normalization)
- 自然语言处理领域(NLP)常用层归一化(layer normalization)，基于特征维度进行规范化。
ln = nn.LayerNorm(2)
bn = nn.BatchNorm1d(2)
X = torch.tensor([[1, 2], [2, 3]], dtype=torch.float32)
# 在训练模式下计算X的均值和方差
print('layer norm:', ln(X), '\nbatch norm:', bn(X))layer norm: tensor([[-1.0000, 1.0000],[-1.0000, 1.0000]], grad_fn=<NativeLayerNormBackward0>)
batch norm: tensor([[-1.0000, -1.0000],[ 1.0000, 1.0000]], grad_fn=<NativeBatchNormBackward0>)
class AddNorm(nn.Module):"""残差连接后进行层规范化"""def __init__(self, normalized_shape, dropout, **kwargs):super(AddNorm, self).__init__(**kwargs)self.dropout = nn.Dropout(dropout)self.ln = nn.LayerNorm(normalized_shape)def forward(self, X, Y):return self.ln(self.dropout(Y) + X)
4. 编码器
首先实现编码器中的一个基本块
class EncoderBlock(nn.Module):"""Transformer编码器块"""def __init__(self, key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,dropout, use_bias=False, **kwargs):super(EncoderBlock, self).__init__(**kwargs)self.attention = d2l.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout,use_bias)self.addnorm1 = AddNorm(norm_shape, dropout)self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens, num_hiddens)self.addnorm2 = AddNorm(norm_shape, dropout)def forward(self, X, valid_lens):Y = self.addnorm1(X, self.attention(X, X, X, valid_lens))return self.addnorm2(Y, self.ffn(Y))
Transfomer的编码器堆叠了num_layers个编码器块。
因为这里使用的时值范围在[-1, 1]之间的固定位置编码，因此通过学习得到的输入的嵌入表示的值需要先乘以维度的平方根进行重新缩放，然后再与位置编码相加。
class TransformerEncoder(d2l.Encoder):"""Transformer编码器"""def __init__(self, vocab_size, key_size, query_size, value_size,num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, num_layers, dropout, use_bias=False, **kwargs):super(TransformerEncoder, self).__init__(**kwargs)self.num_hiddens = num_hiddensself.embedding = nn.Embedding(vocab_size, num_hiddens)self.pos_encoding = d2l.PositionalEncoding(num_hiddens, dropout)self.blks = nn.Sequential()for i in range(num_layers):self.blks.add_module("block"+str(i),EncoderBlock(key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, dropout, use_bias))def forward(self, X, valid_lens, *args):# 因为位置编码值在-1和1之间，# 因此嵌入值乘以嵌入维度的平方根进行缩放，# 然后再与位置编码相加。X = self.pos_encoding(self.embedding(X) * math.sqrt(self.num_hiddens))self.attention_weights = [None] * len(self.blks)for i, blk in enumerate(self.blks):X = blk(X, valid_lens)self.attention_weights[i] = blk.attention.attention.attention_weightsreturn X
5. 解码器
关于序列到序列模型（sequence-to-sequence model）：
- 在训练阶段，其输出序列的所有位置（时间步）的词元都是已知的
- 在预测阶段，其输出序列的词元是逐个生成的。
因此，在任何解码器时间步中，只有生成的词元才能用于解码器的自注意力计算中。
为了在解码器中保留自回归的属性，其掩蔽自注意力设定了参数dec_valid_lens，以便任何查询都只会与解码器中所有已经生成词元的位置（即直到该查询位置为止）进行注意力计算。
class DecoderBlock(nn.Module):"""解码器中第i个块"""def __init__(self, key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,dropout, i, **kwargs):super(DecoderBlock, self).__init__(**kwargs)self.i = iself.attention1 = d2l.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout)self.addnorm1 = AddNorm(norm_shape, dropout)self.attention2 = d2l.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout)self.addnorm2 = AddNorm(norm_shape, dropout)self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens,num_hiddens)self.addnorm3 = AddNorm(norm_shape, dropout)def forward(self, X, state):enc_outputs, enc_valid_lens = state[0], state[1]# 训练阶段，输出序列的所有词元都在同一时间处理，# 因此state[2][self.i]初始化为None。# 预测阶段，输出序列是通过词元一个接着一个解码的，# 因此state[2][self.i]包含着直到当前时间步第i个块解码的输出表示if state[2][self.i] is None:key_values = Xelse:key_values = torch.cat((state[2][self.i], X), axis=1)state[2][self.i] = key_valuesif self.training:batch_size, num_steps, _ = X.shape# dec_valid_lens的开头:(batch_size,num_steps),# 其中每一行是[1,2,...,num_steps]dec_valid_lens = torch.arange(1, num_steps + 1, device=X.device).repeat(batch_size, 1)else:dec_valid_lens = None# 自注意力X2 = self.attention1(X, key_values, key_values, dec_valid_lens)Y = self.addnorm1(X, X2)# 编码器－解码器注意力。# enc_outputs的开头:(batch_size,num_steps,num_hiddens)Y2 = self.attention2(Y, enc_outputs, enc_outputs, enc_valid_lens)Z = self.addnorm2(Y, Y2)return self.addnorm3(Z, self.ffn(Z)), state
class TransformerDecoder(d2l.AttentionDecoder):def __init__(self, vocab_size, key_size, query_size, value_size,num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, num_layers, dropout, **kwargs):super(TransformerDecoder, self).__init__(**kwargs)self.num_hiddens = num_hiddensself.num_layers = num_layersself.embedding = nn.Embedding(vocab_size, num_hiddens)self.pos_encoding = d2l.PositionalEncoding(num_hiddens, dropout)self.blks = nn.Sequential()for i in range(num_layers):self.blks.add_module("block"+str(i),DecoderBlock(key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, dropout, i))self.dense = nn.Linear(num_hiddens, vocab_size)def init_state(self, enc_outputs, enc_valid_lens, *args):return [enc_outputs, enc_valid_lens, [None] * self.num_layers]def forward(self, X, state):X = self.pos_encoding(self.embedding(X) * math.sqrt(self.num_hiddens))self._attention_weights = [[None] * len(self.blks) for _ in range (2)]for i, blk in enumerate(self.blks):X, state = blk(X, state)# 解码器自注意力权重self._attention_weights[0][i] = blk.attention1.attention.attention_weights# “编码器－解码器”自注意力权重self._attention_weights[1][i] = blk.attention2.attention.attention_weightsreturn self.dense(X), state@propertydef attention_weights(self):return self._attention_weights
