Transformer详解

Posted Twq

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Transformer详解相关的知识,希望对你有一定的参考价值。

1.理论知识讲解

transfromer这个模型在机器翻译方面就是做如下事情由一种语言到另一种语言


下图中六个encoder在结构上是完全相同的但是每个encoder的内部的参数不完全相同,也就是在训练的时候6个encoder都在训练,并不是一个在训练,然后其它五个去拷贝这个encoder,六个decoder的结构也是完全相同的;每个encoder和decoder是不一样的

1.1 Encoder

1.1.1输入部分

该部分分为两个小部分

1⃣️Embedding

下图中有12个字,按字切分,然后每个字定义一个512维的字向量(可以使用word2vec或者随机初始化)

2⃣️位置嵌入

使用transformer为什么需要位置嵌入,因为在encoder中的第二个部分中多头注意力机制中是将输入的词并行输入的,并不是像rnn那样是一个词一个词的输入,但是这样并行输入有缺少了rnn那种词与词之间的先后关系,因此需要引入词位置的编码。
下图rnn是一个词一个词的输入,输入“我”,处理“我”,在输入“爱”,在接收“我”的信息之后再处理“爱”,这样保证词与词之间的先后顺序不会乱,但是一个词一个词的输入导致效率低

位置编码公式:
公式中的pos就是单词或者是字的位置,2i和2i+1分别对应,在偶数位置使用sin,在基数位置使用cos。对于爱这个字所在的位置pos,如果爱这个字对应的词向量为512维,则对于这512位的向量偶数位用cos来计算,基数位置用cos来计算
公式中的d_model它表示模型中每个输入和输出 token 的向量表示的维度大小。也就是说,模型中的每个单词都被表示为一个长度为 d_model 的向量,对于一下这个例子d_model的值为512,在我的代码中d_model的值为128。


由公式计算出位置编码之后,与原来的词向量进行相加得到最终整个transformer的输入。

1.1.2注意力机制

1⃣️基本的注意力机制
下图中,对于“婴儿在干嘛”这句话应该关注图片中的哪个区域,通过公式计算得到这句话应该更关注图片中的哪个位置


注意力机制的公式,其中Q,K,V,三个都是矩阵

2⃣️在transformer中怎么操作
x1与wq矩阵相乘的到q1矩阵,依次类推


计算attention值
Score的值等于,你当前关注的词的q,分别乘以这个句子中所有的词的k,就会得到句子中每个词对当前的词的打分结果

1.1.3残差和LayNorm


1⃣️残差


可以有效缓解梯度消失

1.1.3前馈神经网络

1.2decoder

在多头注意力机制中多了一个mask机制,表示将当前单词和之后的单词做mask


为什么使用mask的原因:

在 Transformer 模型中,为了生成下一个目标单词,解码器需要访问已经生成的单词和源序列的信息。在训练时,我们可以将目标序列的所有单词都输入到解码器中,并在每个时间步生成一个单词。但是,在生成过程中,我们需要逐步地生成目标序列中的单词,而在每个时间步中,我们只能访问当前时刻之前生成的单词。这意味着,在生成第 i 个单词时,我们不能使用第 i+1 个单词的信息。因此为了解决这一问题才引入了mask机制

k v矩阵是由encoder得到,而q矩阵是decoder中得到,encoder中得到的每个k v矩阵将会和decoder中的q矩阵进行交互


下图讲述了mask的过程,一行一行的看,当输入的s的时候就只能看到s看不到 后面的“卷”,“起”,“来”,因为此时这三个字的mask为1,当输入“卷”的时候就只能看到“S”和“卷”,看不到后面的两个字,这也是代码中上三角矩阵的作用

2.代码部分讲解

import torch
import numpy as np
from config import ngpu, device


# 计算角度:pos * 1/(10000^(2i/d))
def get_angles(pos, i, d_model):
    # 2*(i//2)保证了2i,这部分计算的是1/10000^(2i/d)
    angle_rates = 1 / np.power(10000, 2 * (i // 2) / np.float32(d_model))  # => [1, 512]
    return pos * angle_rates  # [50,1]*[1,512]=>[50, 512]

# np.arange()函数返回一个有终点和起点的固定步长的排列,如[1,2,3,4,5],起点是1,终点是5,步长为1
# 注意:起点终点是左开右闭区间,即start=1,end=6,才会产生[1,2,3,4,5]
# 只有一个参数时,参数值为终点,起点取默认值0,步长取默认值1。
def positional_encoding(position, d_model):  #d_model是位置编码的长度,相当于position encoding的embedding_dim?
    angle_rads = get_angles(np.arange(position)[:, np.newaxis],  # [50, 1]
                            np.arange(d_model)[np.newaxis, :],  # [1, d_model=512]
                            d_model)
    angle_rads[:, 0::2] = np.sin(angle_rads[:, 0::2])  #从0开始步长为2,2i
    angle_rads[:, 1::2] = np.cos(angle_rads[:, 1::2])  #从1开始步长为2,2i+1

    pos_encoding = angle_rads[np.newaxis, ...]  #[50, 512] => [1,50,512]
    return torch.tensor(pos_encoding, dtype=torch.float32)

pos_encoding = positional_encoding(50, 512)

pad = 1  #重要!
def create_padding_mask(seq):  # seq [b, seq_len]
    # seq = torch.eq(seq, torch.tensor(0)).float() # pad=0的情况
    seq = torch.eq(seq, torch.tensor(pad)).float()  # pad!=0
    return seq[:, np.newaxis, np.newaxis, :]  # =>[b, 1, 1, seq_len]

def create_look_ahead_mask(size):  # seq_len
    mask = torch.triu(torch.ones((size, size)), diagonal=1)
    # mask = mask.device() #
    return mask  # [seq_len, seq_len]

def scaled_dot_product_attention(q, k, v, mask=None):
    """
    #计算注意力权重。
    q, k, v 必须具有匹配的前置维度。 且dq=dk
    k, v 必须有匹配的倒数第二个维度,例如:seq_len_k = seq_len_v。
    #虽然 mask 根据其类型(填充或前瞻)有不同的形状,
    #但是 mask 必须能进行广播转换以便求和。

    #参数:
        q: 请求的形状 == (..., seq_len_q, depth)
        k: 主键的形状 == (..., seq_len_k, depth)
        v: 数值的形状 == (..., seq_len_v, depth_v)  seq_len_k = seq_len_v
        mask: Float 张量,其形状能转换成
              (..., seq_len_q, seq_len_k)。默认为None。

    #返回值:
        #输出,注意力权重
    """
    # matmul(a,b)矩阵乘:a b的最后2个维度要能做乘法,即a的最后一个维度值==b的倒数第2个纬度值,
    # 除此之外,其他维度值必须相等或为1(为1时会广播)
    matmul_qk = torch.matmul(q, k.transpose(-1, -2))  # 矩阵乘 =>[..., seq_len_q, seq_len_k]

    # 缩放matmul_qk
    dk = torch.tensor(k.shape[-1], dtype=torch.float32)  # k的深度dk,或叫做depth_k
    scaled_attention_logits = matmul_qk / torch.sqrt(dk)  # [..., seq_len_q, seq_len_k]

    # 将 mask 加入到缩放的张量上(重要!)
    if mask is not None:  # mask: [b, 1, 1, seq_len]
        # mask=1的位置是pad,乘以-1e9(-1*10^9)成为负无穷,经过softmax后会趋于0
        scaled_attention_logits += (mask * -1e9)

    # softmax 在最后一个轴(seq_len_k)上归一化
    attention_weights = torch.nn.functional.softmax(scaled_attention_logits, dim=-1)  # [..., seq_len_q, seq_len_k]

    output = torch.matmul(attention_weights, v)  # =>[..., seq_len_q, depth_v]
    return output, attention_weights  # [..., seq_len_q, depth_v], [..., seq_len_q, seq_len_k]

class MultiHeadAttention(torch.nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        self.num_heads = num_heads
        self.d_model = d_model

        assert d_model % self.num_heads == 0  #因为输入要被(平均?)split到不同的head

        self.depth = d_model//self.num_heads  #512/8=64,所以在scaled dot-product atten中dq=dk=64,dv也是64
        self.wq = torch.nn.Linear(d_model, d_model)
        self.wk = torch.nn.Linear(d_model, d_model)
        self.wv = torch.nn.Linear(d_model, d_model)
        self.final_linear = torch.nn.Linear(d_model, d_model)

    def split_heads(self, x, batch_size):  # x [b, seq_len, d_model]
        x = x.view(batch_size, -1, self.num_heads,
                   self.depth)  # [b, seq_len, d_model=512]=>[b, seq_len, num_head=8, depth=64]
        return x.transpose(1, 2)  # [b, seq_len, num_head=8, depth=64]=>[b, num_head=8, seq_len, depth=64]

    def forward(self, q, k, v, mask):  # q=k=v=x [b, seq_len, embedding_dim] embedding_dim其实也=d_model
        batch_size = q.shape[0]

        q = self.wq(q)  # => [b, seq_len, d_model]
        k = self.wk(k)  # => [b, seq_len, d_model]
        v = self.wv(v)  # => [b, seq_len, d_model]

        q = self.split_heads(q, batch_size)  # => [b, num_head=8, seq_len, depth=64]
        k = self.split_heads(k, batch_size)  # => [b, num_head=8, seq_len, depth=64]
        v = self.split_heads(v, batch_size)  # => [b, num_head=8, seq_len, depth=64]

        scaled_attention, attention_weights = scaled_dot_product_attention(q, k, v, mask)
        # => [b, num_head=8, seq_len_q, depth=64], [b, num_head=8, seq_len_q, seq_len_k]

        scaled_attention = scaled_attention.transpose(1, 2)  # =>[b, seq_len_q, num_head=8, depth=64]
        # 转置操作让张量存储结构扭曲,直接使用view方法会失败,可以使用reshape方法
        concat_attention = scaled_attention.reshape(batch_size, -1, self.d_model)  # =>[b, seq_len_q, d_model=512]

        output = self.final_linear(concat_attention)  # =>[b, seq_len_q, d_model=512]
        return output, attention_weights  # [b, seq_len_q, d_model=512], [b, num_head=8, seq_len_q, seq_len_k]

# 点式前馈网络
def point_wise_feed_forward_network(d_model, dff):
    feed_forward_net = torch.nn.Sequential(
        torch.nn.Linear(d_model, dff),  # [b, seq_len, d_model]=>[b, seq_len, dff=2048]
        torch.nn.ReLU(),
        torch.nn.Linear(dff, d_model),  # [b, seq_len, dff=2048]=>[b, seq_len, d_model=512]
    )
    return feed_forward_net

class EncoderLayer(torch.nn.Module):
    def __init__(self, d_model, num_heads, dff, rate=0.1):
        super(EncoderLayer, self).__init__()

        self.mha = MultiHeadAttention(d_model, num_heads)  # 多头注意力(padding mask)(self-attention)
        self.ffn = point_wise_feed_forward_network(d_model, dff)#前馈神经网络层

        self.layernorm1 = torch.nn.LayerNorm(normalized_shape=d_model, eps=1e-6)
        self.layernorm2 = torch.nn.LayerNorm(normalized_shape=d_model, eps=1e-6)

        self.dropout1 = torch.nn.Dropout(rate)
        self.dropout2 = torch.nn.Dropout(rate)

    # x [b, inp_seq_len, embedding_dim] embedding_dim其实也=d_model
    # mask [b,1,1,inp_seq_len]
    def forward(self, x, mask):
        attn_output, _ = self.mha(x, x, x, mask)  # =>[b, seq_len, d_model] self-attention
        attn_output = self.dropout1(attn_output)
        out1 = self.layernorm1(x + attn_output)  # 残差&层归一化 =>[b, seq_len, d_model]  残差网络思路

        ffn_output = self.ffn(out1)  # =>[b, seq_len, d_model]
        ffn_output = self.dropout2(ffn_output)
        out2 = self.layernorm2(out1 + ffn_output)  # 残差&层归一化 =>[b, seq_len, d_model]

        return out2  # [b, seq_len, d_model]

class DecoderLayer(torch.nn.Module):
    def __init__(self, d_model, num_heads, dff, rate=0.1):
        super(DecoderLayer, self).__init__()

        self.mha1 = MultiHeadAttention(d_model,
                                       num_heads)  # masked的多头注意力(look ahead mask 和 padding mask)(self-attention)
        self.mha2 = MultiHeadAttention(d_model, num_heads)  # 多头注意力(padding mask)(encoder-decoder attention)

        self.ffn = point_wise_feed_forward_network(d_model, dff)

        self.layernorm1 = torch.nn.LayerNorm(normalized_shape=d_model, eps=1e-6)
        self.layernorm2 = torch.nn.LayerNorm(normalized_shape=d_model, eps=1e-6)
        self.layernorm3 = torch.nn.LayerNorm(normalized_shape=d_model, eps=1e-6)

        self.dropout1 = torch.nn.Dropout(rate)
        self.dropout2 = torch.nn.Dropout(rate)
        self.dropout3 = torch.nn.Dropout(rate)

    # x [b, targ_seq_len, embedding_dim] embedding_dim其实也=d_model=512
    # look_ahead_mask [b, 1, targ_seq_len, targ_seq_len] 这里传入的look_ahead_mask应该是已经结合了look_ahead_mask和padding mask的mask
    # enc_output [b, inp_seq_len, d_model]
    # padding_mask [b, 1, 1, inp_seq_len]
    def forward(self, x, enc_output, look_ahead_mask, padding_mask):
        attn1, attn_weights_block1 = self.mha1(x, x, x, look_ahead_mask)  # =>[b, targ_seq_len, d_model], [b, num_heads, targ_seq_len, targ_seq_len]  self-attention,训练阶段用于编码结果
        attn1 = self.dropout1(attn1)
        out1 = self.layernorm1(x + attn1)  # 残差&层归一化 [b, targ_seq_len, d_model]

        # Q: receives the output from decoder\'s first attention block,即 masked multi-head attention sublayer
        # K V: V (value) and K (key) receive the encoder output as inputs
        attn2, attn_weights_block2 = self.mha2(out1, enc_output, enc_output,
                                               padding_mask)  # =>[b, targ_seq_len, d_model], [b, num_heads, targ_seq_len, inp_seq_len]
        attn2 = self.dropout2(attn2)
        out2 = self.layernorm2(out1 + attn2)  # 残差&层归一化 [b, targ_seq_len, d_model]

        ffn_output = self.ffn(out2)  # =>[b, targ_seq_len, d_model]
        ffn_output = self.dropout3(ffn_output)
        out3 = self.layernorm3(out2 + ffn_output)  # 残差&层归一化 =>[b, targ_seq_len, d_model]

        return out3, attn_weights_block1, attn_weights_block2
        #[b, targ_seq_len, d_model], [b, num_heads, targ_seq_len, targ_seq_len], [b, num_heads, targ_seq_len, inp_seq_len]

class Encoder(torch.nn.Module):
    def __init__(self,
                 num_layers,  # N个encoder layer
                 d_model,  # 当前网络的维度
                 num_heads,
                 dff,  # 点式前馈网络内层fn的维度
                 input_vocab_size,  # 输入词表大小(源语言(法语))
                 maximun_position_encoding,
                 rate=0.1):
        super(Encoder, self).__init__()

        self.num_layers = num_layers
        self.d_model = d_model

        self.embedding = torch.nn.Embedding(num_embeddings=input_vocab_size, embedding_dim=d_model)#可以对应看transformer的图,这个是编码器的词向量层
        self.pos_encoding = positional_encoding(maximun_position_encoding,
                                                d_model).to(device)  #在这个相当于关系图中编码的词嵌入层 =>[1, max_pos_encoding, d_model=512] # 相当于作者手动实现了pos_emb

        # self.enc_layers = [EncoderLayer(d_model, num_heads, dff, rate).cuda() for _ in range(num_layers)] # 不行
        #对num_layers个encoder进行堆叠
        self.enc_layers = torch.nn.ModuleList([EncoderLayer(d_model, num_heads, dff, rate) for _ in range(num_layers)])

        self.dropout = torch.nn.Dropout(rate)

    # x [b, inp_seq_len]
    # mask [b, 1, 1, inp_sel_len]
    def forward(self, x, mask):
        inp_seq_len = x.shape[-1]
        # adding embedding and position encoding
        x = self.embedding(x)  # [b, inp_seq_len]=>[b, inp_seq_len, d_model]
        # 缩放 embedding 原始论文的3.4节有提到: In the embedding layers, we multiply those weights by \\sqrtd_model.
        x *= torch.sqrt(torch.tensor(self.d_model, dtype=torch.float32))
        pos_encoding = self.pos_encoding[:, :inp_seq_len, :]
        #pos_encoding = pos_encoding.cuda()  #调用了显卡资源
        x += pos_encoding  #将得到的位置编码于原来数据的输入进行相加得到最终的输入 [b, inp_seq_len, d_model]
        x = self.dropout(x)
        for i in range(self.num_layers):#将每一层decoder的输出当作下一层decoder的输入
            x = self.enc_layers[i](x, mask)  # [b, inp_seq_len, d_model]=>[b, inp_seq_len, d_model]
        return x  # [b, inp_seq_len, d_model]

class Decoder(torch.nn.Module):
    def __init__(self,
                 num_layers,  # N个encoder layer
                 d_model,
                 num_heads,
                 dff,  # 点式前馈网络内层fn的维度
                 target_vocab_size,  # target词表大小(目标语言(英语))
                 maximun_position_encoding,
                 rate=0.1):
        super(Decoder, self).__init__()

        self.num_layers = num_layers
        self.d_model = d_model

        self.embedding = torch.nn.Embedding(num_embeddings=target_vocab_size, embedding_dim=d_model)
        self.pos_encoding = positional_encoding(maximun_position_encoding,
                                                d_model).to(device)  # =>[1, max_pos_encoding, d_model=512]

        # self.dec_layers = [DecoderLayer(d_model, num_heads, dff, rate).cuda() for _ in range(num_layers)] # 不行
        self.dec_layers = torch.nn.ModuleList([DecoderLayer(d_model, num_heads, dff, rate) for _ in range(num_layers)])  # decoder比encoder多了一组attention层

        self.dropout = torch.nn.Dropout(rate)

    # x [b, targ_seq_len]
    # look_ahead_mask [b, 1, targ_seq_len, targ_seq_len] 这里传入的look_ahead_mask应该是已经结合了look_ahead_mask和padding mask的mask
    # enc_output [b, inp_seq_len, d_model]
    # padding_mask [b, 1, 1, inp_seq_len]
    def forward(self, x, enc_output, look_ahead_mask, padding_mask):
        targ_seq_len = x.shape[-1]

        attention_weights = 

        # adding embedding and position encoding
        x = self.embedding(x)  # [b, targ_seq_len]=>[b, targ_seq_len, d_model]
        # 缩放 embedding 原始论文的3.4节有提到: In the embedding layers, we multiply those weights by \\sqrtd_model.
        x *= torch.sqrt(torch.tensor(self.d_model, dtype=torch.float32))
        # x += self.pos_encoding[:, :targ_seq_len, :]  # [b, targ_seq_len, d_model]
        pos_encoding = self.pos_encoding[:, :targ_seq_len, :]  # [b, targ_seq_len, d_model]
        #pos_encoding = pos_encoding.cuda() #调用显卡资源
        x += pos_encoding  # [b, inp_seq_len, d_model]

        x = self.dropout(x)

        for i in range(self.num_layers):
            x, attn_block1, attn_block2 = self.dec_layers[i](x, enc_output, look_ahead_mask, padding_mask)  # 因为是解码器,需要将编码器的输出结果enc_output传入
            # => [b, targ_seq_len, d_model], [b, num_heads, targ_seq_len, targ_seq_len], [b, num_heads, targ_seq_len, inp_seq_len]

            attention_weights[f\'decoder_layeri + 1_block1\'] = attn_block1
            attention_weights[f\'decoder_layeri + 1_block2\'] = attn_block2
        return x, attention_weights

class Transformer(torch.nn.Module):
    def __init__(self,
                 num_layers,  # N个encoder layer
                 d_model,
                 num_heads,
                 dff,  # 点式前馈网络内层fn的维度
                 input_vocab_size,  # input此表大小(源语言(法语))
                 target_vocab_size,  # target词表大小(目标语言(英语))
                 pe_input,  # input max_pos_encoding
                 pe_target,  # input max_pos_encoding
                 rate=0.1):
        super(Transformer, self).__init__()

        self.encoder = Encoder(num_layers,# Encoder中的EncoderLayer层数
                               d_model,#模型的维度大小,即每个词语的向量维度大小
                               num_heads,#多头注意力机制中head的数量。
                               dff,#前向网络中的隐层大小
                               input_vocab_size,
                               pe_input,
                               rate)
        self.decoder = Decoder(num_layers,
                               d_model,
                               num_heads,
                               dff,
                               target_vocab_size,
                               pe_target,
                               rate)
        self.final_layer = torch.nn.Linear(d_model, target_vocab_size)

    # inp [b, inp_seq_len]
    # targ [b, targ_seq_len]
    # enc_padding_mask [b, 1, 1, inp_seq_len]
    # look_ahead_mask [b, 1, targ_seq_len, targ_seq_len]
    # dec_padding_mask [b, 1, 1, inp_seq_len] # 注意这里的维度是inp_seq_len
    def forward(self, inp, targ, enc_padding_mask, look_ahead_mask, dec_padding_mask):
        enc_output = self.encoder(inp, enc_padding_mask)  # =>[b, inp_seq_len, d_model]
        dec_output, attention_weights = self.decoder(targ, enc_output, look_ahead_mask, dec_padding_mask)
        # => [b, targ_seq_len, d_model],
        # \'..block1\': [b, num_heads, targ_seq_len, targ_seq_len],
        #  \'..block2\': [b, num_heads, targ_seq_len, inp_seq_len], ...
        final_output = self.final_layer(dec_output)  # =>[b, targ_seq_len, target_vocab_size]
        return final_output, attention_weights

以上是关于Transformer详解的主要内容,如果未能解决你的问题,请参考以下文章

Transformer详解(附代码)

[转] Transformer详解

Vision Transformer详解

Swin-Transformer网络结构详解

Transformer模型入门详解及代码实现

广告行业中那些趣事系列4:详解从配角到C位出道的Transformer