深度学习理论基础（七）Transformer编码器和解码器

学习目录：

深度学习理论基础（一）Python及Torch基础篇
深度学习理论基础（二）深度神经网络DNN
深度学习理论基础（三）封装数据集及手写数字识别
深度学习理论基础（四）Parser命令行参数模块
深度学习理论基础（五）卷积神经网络CNN
深度学习理论基础（六）Transformer多头自注意力机制
深度学习理论基础（七）Transformer编码器和解码器

本文目录

学习目录：
前述： Transformer总体结构框图
一、编码器encoder
- 1. 编码器作用
- 2. 编码器部分
- - （1）单个编码器层代码
  - （2）编码器总体代码
二、解码器decoder
- 1. 解码器作用
- 2. 解码器部分
- - （1）单个解码器层代码
  - （2）解码器总体代码
三、Transformer
- 1. 框图
- 2. 代码

前述： Transformer总体结构框图

在这里插入图片描述
（1）嵌入层：编码器的输入是一系列的词嵌入（Word Embeddings），它将文本中的每个词映射到一个高维空间中的向量表示。
嵌入层的作用是将离散符号转换为连续的向量表示，降低数据维度同时保留语义信息，提供词语的上下文信息以及初始化模型参数，从而为深度学习模型提供了有效的输入表示。

class Embedder(nn.Module):
    def __init__(self, vocab_size, d_model):
        super().__init__()
        self.d_model = d_model
        self.embed = nn.Embedding(vocab_size, d_model)
    def forward(self, x):
        return self.embed(x)

（2）位置编码器：为了使 Transformer 能够处理序列信息，位置编码被引入到词嵌入中，以区分不同位置的词。通常使用正弦和余弦函数来生成位置编码。
一句话中同一个词，如果词语出现位置不同，意思可能发生翻天覆地的变化，就比如：我爱你和你爱我。这两句话的意思完全不一样。可见获取词语出现在句子中的位置信息是一件很重要的事情。但是Transformer 的是完全基于self-Attention地，而self-attention是不能获取词语位置信息的，就算打乱一句话中词语的位置，每个词还是能与其他词之间计算attention值，就相当于是一个功能强大的词袋模型，对结果没有任何影响。所以在我们输入的时候需要给每一个词向量添加位置编码。
位置编码是通过在输入序列的每个位置上添加一个与位置相关的固定向量来实现的。这个向量会在前向传播过程中与输入的词嵌入向量相加，以产生具有位置信息的新的向量表示。

●公式：
在这里插入图片描述
① PE：表示位置编码矩阵，可先使用 torch.zeros(行，列) 创建一个全0的矩阵，然后再通过操作将其中的某些元素赋值，形成含有位置信息的矩阵。
② pos：表示输入序列所在行的位置。范围0~输入序列矩阵行长度。
③ d_model :输入、输出向量的维度大小。
④ i：取值范围为 [ 0, d_model] ，但每次移动步长为2。 因为当 i 为偶数时，使用sin函数，当 i 为奇数时，使用cos函数。

●代码：

class PositionalEncoder(nn.Module):
    def __init__(self, d_model, max_seq_len = 200, dropout = 0.1):
        super().__init__()
        self.d_model = d_model   #输入维度
        self.dropout = nn.Dropout(dropout)

        pe = torch.zeros(max_seq_len, d_model)  # 创建一个形状为 (max_seq_len, d_model) 的位置编码矩阵，元素权威0
        for pos in range(max_seq_len):
            for i in range(0, d_model, 2):
                pe[pos, i]     = math.sin(pos / (10000 ** ((2 * i)/d_model)))
                pe[pos, i + 1] = math.cos(pos / (10000 ** ((2 * i)/d_model)))
                
        pe = pe.unsqueeze(0)   # 在第0维度上添加一个维度，变成 (1, max_seq_len, d_model)
        
        # PE位置编码矩阵其中的元素修改好后，可以注册位置编码矩阵为模型的 buffer，即封装起来。
        self.register_buffer('pe', pe) 
     
    def forward(self, x):
       # 将输入乘以一个与模型维度相关的缩放因子
        x = x * math.sqrt(self.d_model)
        # 将位置编码添加到输入中
        seq_len = x.size(1)
       	pe = self.pe[:, :seq_len].clone().detach()  # 从缓冲区中获取位置编码
       	
        if x.is_cuda:
            pe = pe.cuda()  # 将位置编码移到GPU上，如果输入张量也在GPU上
            
        x = x + pe         # 输入向量，加上位置编码器。
        return self.dropout(x)   # 加上丢弃层，防止过拟合。

（3）多头注意力层：这是编码器的核心组件之一。它允许模型在输入序列中学习词与词之间的依赖关系，通过计算每个词对其他词的注意力权重来实现。多头机制允许模型在不同的表示空间中并行地学习多种关注方向。多头注意力层详情查看地址！

def attention(q, k, v, d_k, mask=None, dropout=None): 
    scores = torch.matmul(q, k.transpose(-2, -1)) /  math.sqrt(d_k)
    
    if mask is not None:
        mask = mask.unsqueeze(1)
        scores = scores.masked_fill(mask == 0, -1e9)   
         
    scores = F.softmax(scores, dim=-1)
   
    if dropout is not None:
        scores = dropout(scores)
        
    output = torch.matmul(scores, v)
    return output

class MultiHeadAttention(nn.Module):
    def __init__(self, heads, d_model, dropout = 0.1):
        super().__init__()
        
        self.d_model = d_model  #输入向量的维度
        self.d_k = d_model // heads
        self.h = heads
        
        self.q_linear = nn.Linear(d_model, d_model)
        self.k_linear = nn.Linear(d_model, d_model)
        self.v_linear = nn.Linear(d_model, d_model)
               
        self.dropout = nn.Dropout(dropout)
        self.out = nn.Linear(d_model, d_model)
    
    def forward(self, q, k, v, mask=None):
        bs = q.size(0)
        # perform linear operation and split into N heads
        q = self.q_linear(q).view(bs, -1, self.h, self.d_k)
        k = self.k_linear(k).view(bs, -1, self.h, self.d_k) 
        v = self.v_linear(v).view(bs, -1, self.h, self.d_k)
        
        # transpose to get dimensions bs * N * sl * d_model
        k = k.transpose(1,2)
        q = q.transpose(1,2)
        v = v.transpose(1,2)
        
        scores = attention(q, k, v, self.d_k, mask, self.dropout)
        concat = scores.transpose(1,2).contiguous().view(bs, -1, self.d_model)
        output = self.out(concat)
    
        return output

 if __name__ == '__main__':  
		heads = 8
		d_model = 64
		
		# 创建输入张量 x2 和 mask
		# 假设 x2 和 mask 的形状为 [batch_size, seq_len, embedding_dim]
		batch_size = 16
		seq_len = 10
		embedding_dim = d_model
		
		x2 = torch.randn(batch_size, seq_len, embedding_dim)
		mask = torch.zeros(batch_size, seq_len)  # 假设没有特殊的 mask
		
		# 实例化 MultiHeadAttention 模型
		attn = MultiHeadAttention(heads, d_model, dropout=0.1)
		
		# 使用 MultiHeadAttention 进行前向传播
		output = attn(x2, x2, x2, mask)
		print(output.shape)  # 输出张量的形状

（4）残差连接与规范化层：在多头自注意力层之后，通常会添加残差连接和规范化层来加速训练和提高模型稳定性。残差连接主要是保存原图信息，防止过拟合

"""规范化层"""
class Norm(nn.Module):
    def __init__(self, d_model, eps = 1e-6):
        super().__init__()
    
        self.size = d_model
        
        # create two learnable parameters to calibrate normalisation
        self.alpha = nn.Parameter(torch.ones(self.size))
        self.bias = nn.Parameter(torch.zeros(self.size))
        
        self.eps = eps
    
    def forward(self, x):
        norm = self.alpha * (x - x.mean(dim=-1, keepdim=True)) \
        / (x.std(dim=-1, keepdim=True) + self.eps) + self.bias
        return norm

（5）前馈全连接层：在多头自注意力层之后，还有一个前馈全连接层，它对每个位置的词向量进行非线性变换和映射。

class FeedForward(nn.Module):
    def __init__(self, d_model, d_ff=2048, dropout = 0.1):
        super().__init__() 
    
        # We set d_ff as a default to 2048
        self.linear_1 = nn.Linear(d_model, d_ff)
        self.dropout = nn.Dropout(dropout)
        self.linear_2 = nn.Linear(d_ff, d_model)
    
    def forward(self, x):
        x = self.dropout(F.relu(self.linear_1(x)))
        x = self.linear_2(x)
        return x

一、编码器encoder

1. 编码器作用

编码器的作用是将输入序列转换为语义表示，学习输入序列中词与词之间的依赖关系，并提取输入序列的特征表示，为解码器生成目标序列提供有用的信息。
在这里插入图片描述

2. 编码器部分

（1）单个编码器层代码

在这里插入图片描述

class EncoderLayer(nn.Module):
    def __init__(self, d_model, heads, dropout=0.1):
        super().__init__()
        self.norm_1 = Norm(d_model)  #定义规范化层
        self.norm_2 = Norm(d_model)
        self.attn = MultiHeadAttention(heads, d_model, dropout=dropout) #定义多头注意力层
        self.ff = FeedForward(d_model, dropout=dropout)   #前馈全连接层
        self.dropout_1 = nn.Dropout(dropout)   #丢弃层
        self.dropout_2 = nn.Dropout(dropout)
        
    def forward(self, x, mask):
        x2 = self.norm_1(x)
        x = x + self.dropout_1(self.attn(x2,x2,x2,mask))    #残差连接,注意力中qkv全部来自自身。
        x2 = self.norm_2(x)
        x = x + self.dropout_2(self.ff(x2))
        return x

（2）编码器总体代码

这里需要将上述的编码器层进行克隆复制n份。
在这里插入图片描述

 """编码器总体""   
 #克隆多层编码器层  
 def get_clones(module, N):
    return nn.ModuleList([copy.deepcopy(module) for i in range(N)])
 
class Encoder(nn.Module):
    def __init__(self, vocab_size, d_model, N, heads, dropout):
        super().__init__()
        self.N = N      #n个编码器层
        self.embed = Embedder(vocab_size, d_model)    #嵌入层
        self.pe = PositionalEncoder(d_model, dropout=dropout)   #位置编码
        self.layers = get_clones(EncoderLayer(d_model, heads, dropout), N) #复制n个编码器层
        self.norm = Norm(d_model)  #规范化层
        
    def forward(self, src, mask):
        x = self.embed(src)
        x = self.pe(x)
        for i in range(self.N):
            x = self.layers[i](x, mask)
        return self.norm(x)

二、解码器decoder

1. 解码器作用

在这里插入图片描述

2. 解码器部分

（1）单个解码器层代码

在这里插入图片描述

class DecoderLayer(nn.Module):
    def __init__(self, d_model, heads, dropout=0.1):
        super().__init__()
        self.norm_1 = Norm(d_model)   #定义规范化层
        self.norm_2 = Norm(d_model)
        self.norm_3 = Norm(d_model)
        
        self.dropout_1 = nn.Dropout(dropout)  #定义丢弃层
        self.dropout_2 = nn.Dropout(dropout)
        self.dropout_3 = nn.Dropout(dropout)
        
        self.attn_1 = MultiHeadAttention(heads, d_model, dropout=dropout)  #多头注意力层
        self.attn_2 = MultiHeadAttention(heads, d_model, dropout=dropout)
        self.ff = FeedForward(d_model, dropout=dropout)  #前馈全连接层

    def forward(self, x, e_outputs, src_mask, trg_mask):
        x2 = self.norm_1(x)
        x = x + self.dropout_1(self.attn_1(x2, x2, x2, trg_mask))
        x2 = self.norm_2(x)
        x = x + self.dropout_2(self.attn_2(x2, e_outputs, e_outputs, \
        src_mask))
        x2 = self.norm_3(x)
        x = x + self.dropout_3(self.ff(x2))
        return x

（2）解码器总体代码

在这里插入图片描述

def get_clones(module, N):
    return nn.ModuleList([copy.deepcopy(module) for i in range(N)])

class Decoder(nn.Module):
    def __init__(self, vocab_size, d_model, N, heads, dropout):
        super().__init__()
        self.N = N
        self.embed = Embedder(vocab_size, d_model)   #嵌入层
        self.pe = PositionalEncoder(d_model, dropout=dropout) #位置编码
        self.layers = get_clones(DecoderLayer(d_model, heads, dropout), N) #复制n层解码器层
        self.norm = Norm(d_model)  #规范化层
    def forward(self, trg, e_outputs, src_mask, trg_mask):
        x = self.embed(trg)  
        x = self.pe(x)
        for i in range(self.N):
            x = self.layers[i](x, e_outputs, src_mask, trg_mask)
        return self.norm(x)

三、Transformer

1. 框图

在这里插入图片描述

2. 代码

class Transformer(nn.Module):
    def __init__(self, src_vocab, trg_vocab, d_model, N, heads, dropout):
        super().__init__()
        self.encoder = Encoder(src_vocab, d_model, N, heads, dropout)  #编码器总体
        self.decoder = Decoder(trg_vocab, d_model, N, heads, dropout)  #解码器总体
        self.out = nn.Linear(d_model, trg_vocab)  #全连接层输出
    def forward(self, src, trg, src_mask, trg_mask):
        e_outputs = self.encoder(src, src_mask) 
        d_output = self.decoder(trg, e_outputs, src_mask, trg_mask)
        output = self.out(d_output)
        return output
        
modle=Transformer()