从零开始手搓Transformer#Datawhale组队学习Task1#

从零开始手搓Transformer

缩放点积注意力DotProductAttention

多头注意力Multi-Head Attention

位置编码Position Encoder

前馈神经网络FFN

残差连接和层归一化（Add&Norm）

编码器Encoder

解码器Decoder

编码器-解码器Encoder-Decoder

训练函数：

Transformer%E6%9E%B6%E6%9E%84%E6%9D%A5%E5%AE%9E%E4%BE%8B%E5%8C%96%E7%BC%96%E7%A0%81%E5%99%A8%EF%BC%8D%E8%A7%A3%E7%A0%81%E5%99%A8%E6%A8%A1%E5%9E%8B-toc" style="margin-left:40px;"> 依照Transformer架构来实例化编码器－解码器模型

参考文献：

tiny-universe/content/TinyTransformer at main · datawhalechina/tiny-universe (github.com)

《attention is all you need》

《动手学深度学习PyTorch》

一些简介：Transformer 是一种在自然语言处理（NLP）领域具有里程碑意义的模型架构，首次在2017年的论文《Attention is All You Need》中提出。该架构摒弃了传统的递归神经网络（RNN）和卷积神经网络（CNN），完全依赖于自注意力机制（Self-Attention Mechanism），使得模型能够并行处理序列数据，大幅提升了训练效率。Transformer 主要由编码器（Encoder）和解码器（Decoder）两部分组成，每一部分都由多个结构相同的层堆叠而成。编码器负责将输入序列转换为中间表示，而解码器则基于这些表示生成输出序列。每个层内部包含多头自注意力（Multi-Head Self-Attention）模块和前馈神经网络（Feed-Forward Network），并通过残差连接（Residual Connection）和层归一化（Layer Normalization）技术增强了模型的表达能力和训练稳定性。Transformer 不仅在机器翻译任务上取得了卓越的性能，还广泛应用于文本摘要、问答系统等众多NLP任务中。

动手学深度学习

手搓Transformer

Transoformer的组成：Transformer由编码器-解码器架构组成，每个部分都使用了多头注意力机制（Multi-head)和前馈神经网络(FFN)，叠加位置编码模块，嵌入层等。

代码实现+知识讲解

缩放点积注意力DotProductAttention

Q：什么是注意力机制？

A：注意力机制（Attention Mechanism）是一种在序列到序列（Seq2Seq）模型中引入的重要机制，它的核心思想是让模型能够关注输入序列的不同部分，而不是平均对待所有输入。通过这种方式，模型可以更好地处理长距离依赖，并且在处理序列数据时更具灵活性。

Q：注意力机制都有哪些形式？

A：主要包括1）加性注意力（additive attention）；2）缩放点积注意力（scaled dot‐product attention）

Q：缩放点积注意力如何计算？

A：主要包括1）计算注意力得分；2）应用Softmax函数；3）加权求和Value向量。参考公式如下图所示：

代码实现：

class DotProductAttention(nn.Module):
    def __init__(self,dropout,**kwargs):
        super(DotProductAttention,self).__init__(**kwargs)
        self.dropout = nn.Dropout(dropout)
    # 前向计算
    def forward(self,queries,keys,values,valid_lens=None):
        # 获取查询向量的维度d
        d = queries.shape[-1]
        # 计算注意力得分，并缩放
        scores = torch.bmm(queries,keys.transpose(1,2))/torch.sqrt(d)
        # 应用掩码并计算注意力权重
        self.attention_weights = masked_softmax(scores,valid_lens)
         # 应用 dropout 并加权求和 values
        return torch.bmm(self.dropout(self.attention_weights),values)

masked_softmax函数（实现带掩码的缩放点积注意力计算）：

def sequence_mask(X, valid_lens, value=0):
    # 获取序列的最大长度
    max_len = X.size(1)
    # 创建一个范围张量，形状为 (1, max_len)
    row_vector = torch.arange(0, max_len, device=X.device).unsqueeze(0)
    # 将范围张量扩展为 (batch_size, max_len)
    # 将row_vector与valid_lens进行比较，生成一个布尔掩码矩阵mask
    mask = row_vector < valid_lens.unsqueeze(1)
    # 将掩码应用于输入张量 X
    Y = X.clone()
    # 将Y中对应mask为False的位置（即超出有效长度的位置）设置为 value
    Y[~mask] = value
    return Y

def masked_softmax(X,valid_lens):
    """"实现带掩码的softmax函数"""
    # 如果没有提供有效长度，则直接返回softmax结果
    if valid_lens is None:
        return nn.functional.softmax(X,dim=-1)
    else:
        shape = X.shape
        # 如果valid_lens是一维的，则将其重复以匹配X的第二维度
        if valid_lens.dim()==1:
            valid_lens = torch.repeat_interleave(valid_lens,shape[1])
        else:
            valid_lens = valid_lens.reshape[-1]
        # 将X展平为二维，并应用sequence_mask函数
        X = sequence_mask(X.reshape(-1,shape[-1]),valid_lens,value=-1e6)
        # 将展平后的结果恢复原始形状，并应用softmax
        return nn.functional.softmax(X,dim=-1)

多头注意力Multi-Head Attention

Q：什么是多头注意力？

A：多头注意力是指使用多个头head来独立计算注意力，然后再把这些独立计算的注意力合到一起，不同的头，可能会关注输入的不同部分，可以学习到不同的特征。

Q：多头注意力如何实现？

A：多头注意力的完整实现步骤如下：

输入投影：通过线性层W_q、W_k、W_v将输入的查询、键和值投影到不同的子空间。
分裂头部：使用transpose_qkv函数将投影后的张量分裂成多个头。
计算注意力权重：使用点积注意力机制计算每个头的注意力权重。
合并头部：使用transpose_output函数将多头的结果拼接在一起。
输出投影：通过线性层W_o将拼接后的输出投影回原始维度。

代码部分（关于代码部分的详细讲解参考注释）：

疑问or内部细节：

1.需要保证num_hideens/num_heads==0，即在多头注意力机制中，确实需要确保num_hiddens能够被num_heads整除，这样才能保证每个头的隐藏维度大小是整数。如果num_hiddens不能被num_heads整除，就会出现问题。（定位：transpose_qkv函数）

如何解决：

方法一：调整num_hiddens或num_heads

最简单的方法是在设计模型时确保num_hiddens能够被num_heads整除。这意味着你需要选择合适的num_hiddens和num_heads的组合。

方法二：动态调整

如果无法预先确定num_hiddens和num_heads的组合，可以在模型内部进行动态调整。例如，可以向上取整到最接近的能够整除的大小。

 # 确保num_hiddens可以被num_heads整除
 assert num_hiddens % num_heads == 0, "num_hiddens must be divisible by num_heads"

2.valid_lens 代码疑问，为什么在torch.repeat_interleave中使用dim=0，即沿着0维repeat？

 # 如果valid_lens不是None，则重复以适应多头数量
 # 沿着批次维度（即dim=0）重复valid_lens,重复num_heads次
 if valid_lens is not None:
    valid_lens = torch.repeat_interleave(valid_lens,repeats=self.num_heads,dim=0)

dim=0指的是沿着张量的第一维（即最外层的维度，批次维度）进行操作，在多头注意力机制中，我们需要将输入数据分成多个头，并且每个头都独立进行注意力计算。因此，如果原来的有效长度是针对整个批次的，那么在多头注意力中，每个头都需要对应一个有效长度。

为了让每个头都拥有自己的有效长度，我们需要将原来的valid_lens沿着批次维度进行重复，使得每个头都有相同的有效长度信息。

位置编码Position Encoder

Q:为什么要进行位置编码？

A:为了使用序列的顺序信息，通过在输入表示中添加位置编码（positional encoding）来注入绝对的或相对的位置信息。位置编码可以通过学习得到也可以直接固定得到。

Q：位置编码都有哪些实现形式呢？列举几种常见的并简述应用。

A：位置编码包括：1）固定式位置编码（Fixed Positional Encoding），Transformer原始论文提出并广泛应用；2）可学习位置编码（Learnable Positional Encoding），BERT使用可学习位置编码；3）相对位置编码（Relative Positional Encoding）；4）组合式位置编码（Combined Positional Encoding）等。

Q：Transformer原始论文中，位置编码是如何实现的？

A：使用的是固定位置编码，即正弦余弦函数。位置编码是根据位置和维度计算得出的，不随训练更新。

Q：图像处理领域，Transformer架构使用的位置编码？

A：图像领域主要使用的包括1）可学习位置编码（如ViT）；2）二维位置编码（2D Positional Encoding）

代码实现：（基于上述公式实现）

class PositionalEncoding(nn.Module):
    """"位置编码的是实现"""
    def __init__(self,num_hiddens,dropout,max_len=1000):
        super(PositionalEncoding,self).__init__()
        self.dropout = nn.Dropout(dropout)
        # 初始化编码矩阵，全0向量，用于存放位置编码
        self.P = torch.zeros(1,max_len,num_hiddens)
        # 计算位置和频率，X表示序列中每个位置的索引，freqs计算了每个维度的频率因子
        freqs = torch.arrange(0,num_hiddens,2,dtype=torch.float32)/num_hiddens
        X = torch.arange(max_len,dtype=torch.float32).reshape(-1,1)/torch.pow(10000,freqs)
        # 应用正弦函数和余弦函数，根据维度的奇偶性。
        self.P[:,:,0::2] = torch.sin(X)
        self.P[:,:,1::2] = torch.cos(X)
    
    # 前向计算
    def forward(self,X):
        # 将位置编码矩阵加到输入嵌入上
        X = X + self.P[:,:X.shape[1],:].to(X.device)
        return self.dropout(X)

疑问or内部细节：

关于P的维度的理解，即

“self.P[:,:,0::2] = torch.sin(X)”、“self.P[:,:,1::2] = torch.cos(X)”、“X = X + self.P[:,:X.shape[1],:].to(X.device)”这三条代码。

解释：

已知self.P 的形状为 (1, max_len, num_hiddens)，这里：

第一维度（第一个 :）：表示批次维度，这里只有一个批次，所以是全部选取。
第二维度（第二个 :）：表示序列长度维度，这里也是全部选取。
第三维度（0::2 和 1::2）：表示隐藏维度。
self.P[:, :, 0::2] 表示选取 self.P 中所有位置的所有偶数隐藏维度。
self.P[:, :, 1::2] 表示选取 self.P 中所有位置的所有奇数隐藏维度。
self.P[:, :X.shape[1], :]：我们选取 self.P 的前 X.shape[1] 个位置（即 X 的序列长度），并且选取所有的隐藏维度。

嵌入层embedding

层归一化LayerNorm

前馈神经网络FFN

Q：前馈神经网络的本质是什么？

A：多层感知机MLP

代码实现：

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

残差连接和层归一化（Add&Norm）

Q：什么是残差连接？

A：残差连接（Residual Connection）是一种在深度神经网络中使用的连接方式，它通过在层之间添加“跳跃连接”（skip connections）来改善深层网络的训练。这种连接方式最早是在 ResNet（残差网络）中提出的，后来被广泛应用于各种深度学习架构中，包括 Transformer。

Q：残差连接有什么作用？

A：1）解决梯度消失/梯度爆炸问题；2）加速收敛；3）提高模型性能

Q：什么是层归一化？

A：层归一化是一种重要的技术，它通过对输入数据进行归一化来减少内部协变量偏移，从而提高模型的训练效率和性能。

Q：层归一化和批量归一化的区别?简要解释。

A：BatchNorm: 一批次样本按每一个特征维度进行归一化；LayerNorm: 同一个样本的不同特征归一化。

代码实现：

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)

编码器Encoder

Q：Transformer编码器的组成？

A：Transformer编码器由多个编码器块Encoder Block堆叠而成，每个编码器块包括1）多头注意力；2）残差连接和层归一化；3）前馈神经网络等。

代码实现：

# 定义EncoderBlock类
class EncoderBlock(nn.Module):
    """"编码器块"""
    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 = 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))

# 定义TransformerEncoder类  
class TransformerEncoder(EncoderBlock):
    """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_hiddens
        self.embedding = nn.Embedding(vocab_size, num_hiddens)
        self.pos_encoding = 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_weights
                    return X

解码器Decoder

Q：什么是解码器，解码器的作用是什么？

A：Transformer解码器同样是由多个解码器块（DecoderBlock）堆叠而成，Transformer 的解码器模块负责将解码器的输入（通常是编码器的输出和先前生成的目标序列的一部分）转换为目标序列的预测。

Q：解码器块包含哪些组件？

A：解码器块相比编码器块，多了一层Masked多头注意力，然后接Add&&Norm，然后是一层编码器-解码器多头注意力，然后接一层Add&&Norm，再然后是FFN，接Add&&Norm。

代码实现：

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 = i
        self.attention1 = MultiHeadAttention(
            key_size, query_size, value_size, num_hiddens, num_heads, dropout)
        self.addnorm1 = AddNorm(norm_shape, dropout)
        self.attention2 = 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 = X
        else:
            key_values = torch.cat((state[2][self.i], X), axis=1)
        state[2][self.i] = key_values
        if 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(DecoderBlock):
    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_hiddens
        self.num_layers = num_layers
        self.embedding = nn.Embedding(vocab_size, num_hiddens)
        self.pos_encoding = 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_weights
        return self.dense(X), state

    @property
    def attention_weights(self):
        return self._attention_weights

编码器-解码器Encoder-Decoder

代码实现：

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)

训练函数：

代码实现：

#@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 = d2l.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)}')

Transformer%E6%9E%B6%E6%9E%84%E6%9D%A5%E5%AE%9E%E4%BE%8B%E5%8C%96%E7%BC%96%E7%A0%81%E5%99%A8%EF%BC%8D%E8%A7%A3%E7%A0%81%E5%99%A8%E6%A8%A1%E5%9E%8B"> 依照Transformer架构来实例化编码器－解码器模型

num_hiddens, num_layers, dropout, batch_size, num_steps = 32, 2, 0.1, 64, 10
lr, num_epochs, device = 0.005, 200, d2l.try_gpu()
ffn_num_input, ffn_num_hiddens, num_heads = 32, 64, 4
key_size, query_size, value_size = 32, 32, 32
norm_shape = [32]

train_iter, src_vocab, tgt_vocab = d2l.load_data_nmt(batch_size, num_steps)

encoder = TransformerEncoder(
    len(src_vocab), key_size, query_size, value_size, num_hiddens,
    norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,
    num_layers, dropout)
decoder = TransformerDecoder(
    len(tgt_vocab), key_size, query_size, value_size, num_hiddens,
    norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,
    num_layers, dropout)
net = d2l.EncoderDecoder(encoder, decoder)
d2l.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)

时间仓库，多有不周，还请见谅，有空的时候再来重新修改润色。

喜欢的小伙伴，点赞收藏关注吧。