Transformer总体架构

一、Transformer架构解析

1.1 认识Transformer

Transformer模型的作用：
Transformer 模型是一种用于自然语言处理任务的深度学习模型，特别适用于序列到序列的任务，如机器翻译、文本生成、问答系统等。它的主要特点是没有使用传统的循环神经网络（RNN）或卷积神经网络（CNN），而是完全依赖于自注意力机制（Self-Attention Mechanism）来捕捉输入序列中的依赖关系。

自注意力机制：
自注意力机制是 Transformer 模型的核心组件之一，它允许模型在处理每个位置的词汇时，考虑整个输入序列中的所有词汇，从而捕捉长距离依赖关系。自注意力机制通过计算查询（Query）、键（Key）和值（Value）之间的相似度，生成加权和的表示。

查询（Query, Q）
作用：查询就像是你在图书馆里想要找一本书时，心里想的那个书名或关键词。
计算：查询是从输入序列中提取出来的，用来表示每个词汇对其他词汇的关注程度。
类比：假设你手头有一本书的目录，你想找到某个章节的内容，你脑子里想的那个章节标题就是查询。
键（Key, K）
作用：键就像是图书馆里的书名或关键词，是你用来匹配查询的信息。
计算：键也是从输入序列中提取出来的，用来表示每个词汇对其他词汇的被关注程度。
类比：继续上面的例子，图书馆里每本书的标题或关键词就是键。
值（Value, V）
作用：值就像是图书馆里书的内容，是你最终想要获取的信息。
计算：值也是从输入序列中提取出来的，用来表示每个词汇的实际内容。
类比：图书馆里每本书的具体内容就是值。

自注意力机制的简单解释：

自注意力机制的工作原理可以类比为在图书馆里找书的过程：

1. 准备查询：你脑子里有一个关键词（查询 Q），比如“人工智能”。
2. 查找键：你去图书馆的目录（键 K），看看哪些书的标题或关键词包含了“人工智能”。
3. 计算相似度：你计算每个书的标题或关键词与你的关键词的相似度（计算 Q 和 K 的点积）。
4. 选择书：你根据相似度选择几本书（应用 Softmax 函数，得到注意力权重）。
5. 阅读内容：你阅读这几本书的内容（加权求和，得到最终的输出）。

1.2 Transformer总体架构

Transformer总体架构可分为四个部分：

• 输入部分
• 输出部分
• 编码部分
• 解码部分

输入部分包括：

• 源文本嵌入层及其位置编码器
• 目标文本嵌入层及其位置编码器

输出部分包含：

• 线性层
• softmax层

编码器部分：

• 由N个编码器堆叠而成
• 每个编码器层由两个子层连接结构组成
• 第一个子层连接结构包括一个多头自注意力子层和一个残差连接
• 第二个子层连接结构包括一个前馈全连接子层和规范化以及一个残差连接
  

解码器部分：

• 由N个解码器堆叠而成
• 每个解码器层由三个子层连接结构组成
• 第一个子层连接结构包括一个多头自注意力子层和规范化层以及一个残差连接
• 第二个子层连接包括一个多头注意力子层和规范化层以及一个残差连接
• 第三个子层连接结构包括一个前馈全连接子层和规范化层以及一个残差连接
  

二、Transformer架构实现

使用Transformer模型架构处理从一种语言文本到另一种语言文本的翻译工作.

# 导包
import torch
import torch.nn as nn
import math
import numpy as np
from torch.autograd import Variable
import matplotlib.pyplot as plt
import torch.nn.functional as F
from torch.xpu import device

2.1 输入部分的实现

2.1.1 文本嵌入层

无论是源文本嵌入还是目标文本嵌入，都是为了将文本中词汇的数字表示转变为向量表示,希望在这样的高维空间捕捉词汇间的关系.

# 文本嵌入层代码
class Embeddings(nn.Module):
    def __init__(self,d_model,vocab):
        super(Embeddings,self).__init__()
        # 创建嵌入层，将词汇表中的每个单词映射到一个d_model维的向量
        self.lut = nn.Embedding(vocab,d_model)
        self.d_model = d_model
    def forward(self,x):
        # 确保嵌入向量的尺度与其他部分的尺度一致，从而有助于模型的稳定性和收敛速度
        return self.lut(x) * math.sqrt(self.d_model)

2.1.2 位置编码器

因为在Transformer的编码器结构中, 并没有针对词汇位置信息的处理，因此需要在Embedding层后加入位置编码器，将词汇位置不同可能会产生不同语义的信息加入到词嵌入张量中, 以弥补位置信息的缺失.

class PositionalEncoding(nn.Module):
    def __init__(self,d_model,dropout,max_len=5000):
        '''
        d_model:词嵌入维度
        dropout:神经元失活率
        max_len:每个句子的最大长度
        '''
        super(PositionalEncoding,self).__init__()
        # 实例化
        self.dropout = nn.Dropout(p=dropout)
        #初始化一个位置矩阵,max_len * d_model
        pe = torch.zeros(max_len,d_model)
        # 初始化一个绝对位置矩阵,词汇的绝对位置就是用它的索引去表示
        # max_len * 1
        position = torch.arange(0,max_len).unsqueeze(1)
        # 将绝对位置信息加入到位置编码矩阵中
        # 直接将max_len * 1 矩阵变为max_len * d_model,然后覆盖pe
        # 需要一个1 * d_model的变换矩阵div_term
        div_term = torch.exp(torch.arange(0,d_model,2) * -(math.log(10000.0)/d_model))
        # 矩阵偶数位置上填充
        pe[:,0::2] = torch.sin(position * div_term)
        # 矩阵奇数位置上填充
        pe[:,1::2] = torch.cos(position * div_term)

        # pe矩阵(二维)要想和embedding(三维)的输出相加就必须拓展一个维度
        pe = pe.unsqueeze(0)
        # 把pe位置编码矩阵注册成buffer
        # buffer是对模型有帮助的,但是却不是模型结构中超参数或者参数,不需要随着优化步骤进行更新的增益对象。
        # 注册之后我们可以在模型保存后重新加载时和模型一同被加载
        self.register_buffer('pe',pe)

    def forward(self,x):
        # x 的形状是 (batch_size, seq_len, d_model)
        # 默认max_len太长为5000,一般很难有句子包含5000个词汇,所以要进行与输入张量的适配
        x = x + torch.tensor(self.pe[:,:x.size(1)],requires_grad=False)
        # 进行失活操作
        return self.dropout(x)

# 绘制词汇向量中特征的分布曲线
plt.figure(figsize=(15,6))
pe = PositionalEncoding(20,0)
y = pe(torch.tensor(torch.zeros(1,100,20)))
plt.plot(torch.arange(100),y[0,:,4:8].data.numpy())
plt.legend(["dim %d" % p for p in [4,5,6,7]])

2.2 编码器部分的实现

2.2.1 生成掩码

在transformer中, 掩码张量的主要作用在应用attention时，有一些生成的attention张量中的值计算有可能已知了未来信息而得到的，未来信息被看到是因为训练时会把整个输出结果都一次性进性Embedding，但是理论上解码器的的输出却不是一次就能产生最终结果的，而是一次次通过上一次结果综合得出的，因此，未来的信息可能被提前利用.

def subsequent_mask(size):
    # 定义掩码张量的形状
    attn_shape = (1,size,size)
    # 提取上三角矩阵
    subsequent_mask = np.triu(np.ones(attn_shape),k=1).astype('uint8')
    return torch.from_numpy(1 - subsequent_mask)

plt.figure(figsize=(5,5))
plt.imshow(subsequent_mask(20)[0])

2.2.2 注意力机制

# attention 函数实现了注意力机制的核心部分，通过计算查询向量和键向量之间的相似度分数，
# 应用掩码和 softmax 得到注意力权重，最后将这些权重应用于值向量，得到加权和输出。
# 提高了模型在处理长序列和复杂模式时的性能和表达能力
def attention(query,key,value,mask=None,dropout=None):

    # 计算注意力分数
    scores = torch.matmul(query,key.transpose(1,2)) 

    # d_k嵌入维度的大小，我们需要对注意力分数进行缩放，缩放因子为 sqrt(d_k)，以防止分数过大导致 softmax 函数的梯度消失或爆炸。
    d_k = query.size(-1)
    scores = scores / math.sqrt(d_k)

    # 在某些情况下（如序列填充或自注意力机制中的未标记掩码），我们希望忽略某些位置的注意力分数。
    # 通过将这些位置的分数设为极小值，softmax 后这些位置的权重将几乎为 0
    if mask is not None:
        scores = scores.masked_fill(mask == 0,-1e9)
   
    # 计算注意力权重，将注意力分数转换为概率分布
    p_attn = F.softmax(scores,dim=-1)

    # 随机将一些注意力权重设为 0，以防止过拟合。
    if dropout is not None:
        p_attn = dropout(p_attn)

    # 根据注意力权重对值向量进行加权求和，得到每个查询向量对应的输出向量。
    return torch.matmul(p_attn,value),p_attn

2.2.3 多头注意力机制

多头注意力机制（Multi-Head Attention）是 Transformer 模型中的一个重要组成部分，它通过将输入序列分成多个头（Head），每个头独立计算自注意力，然后将结果拼接在一起，再通过一个线性层进行投影，以增加模型的表达能力。

import copy

def clones(module,N):
    # 在多头注意力机制中，需要多个线性层来处理查询、键和值向量
    return nn.ModuleList([copy.deepcopy(module) for _ in range(N)])

# 实现多头注意力机制的处理
class MultiHeadedAttention(nn.Module):
    def __init__(self,head,embedding_dim,dropout=0.1):
        super(MultiHeadedAttention,self).__init__()
        # 断言，确保嵌入维度可以被头数整除
        assert embedding_dim % head == 0
        # 每个头的维度
        self.d_k = embedding_dim // head
        self.head = head
        self.linears = clones(nn.Linear(embedding_dim,embedding_dim),4)
        # 存储注意力权重
        self.attn = None
        self.dropout = nn.Dropout(p=dropout)
    def forward(self,query,key,value,mask=None):
        '''
        query: 查询向量，形状为 [batch_size, num_queries, embedding_dim]
        key: 键向量，形状为 [batch_size, num_keys, embedding_dim]
        value: 值向量，形状为 [batch_size, num_keys, embedding_dim]
        mask: 可选的掩码，形状为 [batch_size, num_queries, num_keys]
        '''
        # 确保掩码的形状与批量数据匹配
        if mask is not None:
            mask = mask.unsqueeze(0)
        batch_size = query.size(0)
        # 使用前三个线性层分别对查询、键和值进行线性变换。
        # view将变换后的张量重塑为 [batch_size, num_queries/num_keys, head, d_k]
        query,key,value = [model(x).view(batch_size,-1,self.head,self.d_k).transpose(1,2) for model,x in zip(self.linears,(query,key,value))]
        x,self.attn = attention(query,key,value,mask=mask,dropout=self.dropout)
        # 将注意力加权和输出 x 重塑为 [batch_size, num_queries, embedding_dim]
        # contiguous()确保张量在内存中是连续的，便于后续操作
        x = x.transpose(1,2).contiguous().view(batch_size,-1,self.head * self.d_k)
        # 使用最后一个线性层 self.linears[-1] 对注意力加权和输出 x 进行最终变换，得到最终的输出张量。
        # 将每个头的输出合并成一个最终的输出
        return self.linears[-1](x)

2.2.4 前馈全连接层

概念：在Transformer中前馈全连接层就是具有两层线性层的全连接网络.
作用：考虑注意力机制可能对复杂过程的拟合程度不够, 通过增加两层网络来增强模型的能力

class PositionwiseFeedForward(nn.Module):
    def __init__(self,d_model,d_ff,dropout=0.1):
        '''
        d_model:输入和输出特征维度
        d_ff:中间隐藏层特征维度
        '''
        super(PositionwiseFeedForward,self).__init__()
        # 将输入特征从 d_model 映射到 d_ff
        self.w1 = nn.Linear(d_model,d_ff)
        self.w2 = nn.Linear(d_ff,d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self,x):
        return self.w2(self.dropout(F.relu(self.w1(x))))

2.2.5 规范化层

它是所有深层网络模型都需要的标准网络层，因为随着⽹络层数的增加，通过多层的计算后参数可能开始出现过大或过小的情况，这样可能会导致学习过程出现异常，模型可能收敛⾮常的慢. 因此都会在⼀定层数后接规范化层进⾏数值的规范化，使其特征数值在合理范围内.

class LayerNorm(nn.Module):
    def __init__(self,features,eps=1e-6):
        '''
        features:词嵌入维度
        '''
        super(LayerNorm,self).__init__()
        # 可学习缩放参数
        self.a2 = nn.Parameter(torch.ones(features))
        # 可学习偏移参数
        self.b2 = nn.Parameter(torch.zeros(features))
        self.eps = eps
    def forward(self,x):
        mean = x.mean(-1,keepdim=True)
        std = x.std(-1,keepdim=True)
        # 允许模型学习每个特征的缩放和偏移，从而更好地适应数据。
        return self.a2 * (x-mean) / (std+self.eps) + self.b2

2.2.6 子层连接结构

输入到每个子层以及规范化层的过程中，还使用了残差链接（跳跃连接），因此我们把这⼀部分结构整体叫做子层连接（代表子层及其链接结构）, 在每个编码器层中，都有两个子层，这两个子层加上周围的链接结构就形成了两个子层连接结构。

class SunlayerConnection(nn.Module):
    def __init__(self,size,dropout=0.1):
        '''
        size:词嵌入维度大小
        '''
        super(SunlayerConnection,self).__init__()
        # 实例化规范化对象
        self.norm = LayerNorm(size)
        self.dropout = nn.Dropout(p=dropout)
    def forward(self,x,sublayer):
        '''
        x:上一层或者子层的输入
        sublayer:该子层连接中的子层函数
        '''
        return x + self.dropout(sublayer(self.norm(x)))

2.2.7 编码器层

作为编码器的组成单元, 每个编码器层完成⼀次对输入的特征提取过程, 即编码过程.

class EncoderLayer(nn.Module):
    def __init__(self,size,self_attn,feed_forward,dropout):
        ''''
        self_attn: 多有自注意力子层实例化对象
        feed_forward: 前馈全连接层实例化对象
        '''
        super(EncoderLayer,self).__init__()
        self.self_attn = self_attn
        self.forward = feed_forward
        # 编码层有两个子层连接结构
        self.sublayer = clones(SunlayerConnection(size,dropout),2)
        self.size = size
    def forward(self,x,mask):
        '''
        x:上一层的输出
        mask:掩码张量，mask主要用于处理输入序列中的填充部分（padding）。
        这是因为不同的输入序列可能具有不同的长度，而在进行批量处理时，
        需要将这些序列填充至相同的长度。这样做虽然便于并行计算，
        但也引入了一个问题：模型可能会错误地将填充部分（通常是用特殊符号如 <pad> 表示）视为有效信息。
        为了避免这种情况，我们使用 mask 来指示哪些部分是实际的输入，哪些部分是填充的
        '''
        
        x = self.sublayer[0](x,lambda x: self.self_atten(x,x,x,mask))
        return self.sublayer[1](x,self.feed_forward)

2.2.8 编码器

编码器用于对输⼊进行指定的特征提取过程, 也称为编码, 由N个编码器层堆叠而成.

class Encoder(nn.Module):
    def __init__(self,layer,N):
        '''
        layer:EncoderLayer实例
        '''
        super(Encoder,self).__init__()
        self.layers = clones(layer,N)
        # layer.size获取EncoderLayer实例的输入特征维度大小
        self.norm = LayerNorm(layer.size)
    def forward(self,x,mask):
        for layer in self.layers:
            x = layer(x,mask)
        return self.norm(x)

2.3 解码器部分的实现

2.3.1 解码器层

作为解码器的组成单元, 每个解码器层根据给定的输⼊向目标方向进行特征提取操作，即解码过程

class DecoderLayer(nn.Module):
    def __init__(self,size,self_attn,src_attn,feed_forward,dropout):
        '''
        self_attn:多头自注意力对象 Q=K=V
        src_attn:多头注意力对象 Q!=K=V
        '''
        super(DecoderLayer,self).__init__()
        self.size = size
        self.self_attn = self_attn
        self.src_attn = src_attn
        self.feed_forward = feed_forward
        self.sublayer = clones(SunlayerConnection(size,dropout),3)
    def forward(self,x,memory,source_mask,target_mask):
        '''
        x:上一层的输入
        memory:编码器层输出的特征向量矩阵
        source_mask:源数据掩码张量
        target_mask:目标数据源码张量
        '''
        m = memory
        # target_mask防止模型在生成某个位置的输出时看到未来的位置。
        # 在生成目标序列时，模型不能看到未来的词。例如，在生成第 t 个词时，
        # 模型只能看到第 t 个词之前的所有词，不能看到第 t+1 及以后的词。
        # target_mask 用于实现这一点，通常是一个上三角矩阵，对角线及其以下的部分为1，对角线上方的部分为0或负无穷大
        x = self.sublayer[0](x,lambda x:self.self_attn(x,x,x,target_mask))
        # source_mask忽略某些位置的输入
        # 在实际应用中，不同长度的源序列会被填充到相同的长度。
        # source_mask 用于标记哪些位置是填充的（通常是0），哪些位置是有效的输入（通常是1）
        # 这样在计算自注意力时，可以忽略掉填充的位置，避免引入噪声。
        x = self.sublayer[1](x,lambda x:self.src_attn(x,m,m,source_mask))
        # 输出了由编码器输⼊和⽬标数据⼀同作⽤的特征提取结果.
        return self.sublayer[2](x,self.feed_forward)

2.3.2 解码器

class Decoder(nn.Module):
    def __init__(self,layer,N):
        super(Decoder,self).__init__()
        self.layers = clones(layer,N)
        self.norm = LayerNorm(layer.size)
    def forward(self,x,memory,source_mask,target_mask):
        for layer in self.layers:
            x = layer(x,memory,source_mask,target_mask)
        return self.norm(x)

2.4 输出部分实现

线性层：对上一步的线性变化得到指定维度的输出, 也就是转换维度的作用
softmax层：使最后一维的向量中的数字缩放到0-1的概率值域内, 并满足他们的和为1.

class Generator(nn.Module):
    def __init__(self,d_model,vocab_size):
        '''
        d_model:词嵌⼊维度,通常是从解码器的最后一个子层（如前馈全连接层）输出的特征向量的维度。
        vocab_size:词汇表的大小，表示模型需要生成的词的总数。
        '''
        super(Generator,self).__init__()
        self.project = nn.Linear(d_model,vocab_size)
    def forward(self,x):
        return F.log_softmax(self.project(x),dim=-1)

2.5 模型构建(编码器-解码器结构)

class EncoderDecoder(nn.Module):
    def __init__(self,encoder,decoder,source_embed,target_embed,generator):
        '''
        encoder:编码器部分，负责处理源序列并生成记忆（memory）。
        decoder:解码器部分，负责处理目标序列并生成最终的输出。
        source_embed:源序列的嵌入层，将源序列的词转换为固定维度的向量。
        target_embed:目标序列的嵌入层，将目标序列的词转换为固定维度的向量。
        generator:生成器部分，用于生成最终的输出概率分布。
        '''
        super(EncoderDecoder, self).__init__()
         # 将参数传⼊到类中
        self.encoder = encoder
        self.decoder = decoder
        self.source_embed = source_embed
        self.target_embed = target_embed
        self.generator = generator
    
    def forward(self,source,target,source_mask,target_mask):
        '''
        source:源序列的输入张量
        target:目标序列的输入张量
        source_mask:源序列的掩码
        target_mask:目标序列的掩码
        '''
        # 编码
        memory = self.encode(source, source_mask)
        # 解码
        decoder_output = self.decode(memory, source_mask, target, target_mask)
        # 生成最终的输出概率分布
        output = self.generator(decoder_output)
        return output

    # 编码函数,生成编码器的输出 memory
    def encode(self,source,source_mask):
        return self.encoder(self.source_embed(source), source_mask)
    
    # 解码函数,生成解码器的输出
    def decode(self,memory,source_mask,target,target_mask):
        return self.decoder(self.target_embed(target), memory, source_mask,target_mask)

2.6 模型构建

def make_model(source_vocab,target_vocab,N=6,d_model=512,d_ff=2048,head=8,dropout=0.1):
    '''
    :param source_vocab: 原词汇大小
    :param target_vocab: 目标词汇大小
    :param N: 编码器和解码器层数
    :param d_model: 模型的隐藏维度
    :param d_ff: 前馈网络中间维度
    :param head: 多头注意力机制头数
    :param dropout: 神经元失活概率
    :return: 返回Transformer模型
    '''
    c = copy.deepcopy
    attn = MultiHeadedAttention(head,d_model)
    ff = PositionwiseFeedForward(d_model,d_ff,dropout)
    position = PositionalEncoding(d_model,dropout)
    model = EncoderDecoder(
        Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout), N),
        Decoder(DecoderLayer(d_model, c(attn), c(attn),c(ff), dropout), N),
        nn.Sequential(Embeddings(d_model, source_vocab), c(position)),
        nn.Sequential(Embeddings(d_model, target_vocab), c(position)),
        Generator(d_model, target_vocab)
    )
    for p in model.parameters():
        if p.dim() > 1:
            # 进行 Xavier 均匀分布初始化
            nn.init.xavier_uniform_(p)
    return model

2.7 测试与运行

source_vocab = 11
target_vocab = 11
N=6
# 检查GPU是否可用,选择cuda1
device = torch.device("cuda:1" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")
# 实例化模型
model = make_model(source_vocab, target_vocab, N)
model.to(device)
model

原文地址：https://blog.csdn.net/qq_56246012/article/details/142750432

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！