尧图网站建设 
写这篇文章的目的是为了理清现在很多模型的基础Transformer模型是怎么回事。为此自己也是跟着视频敲了一遍代码但仍然感觉没有吃透所以再写一篇。下图是模型的架构。source sequence说模型必须弄明白输入和输出的数据结构形状是怎么样的有位学长说过现在的模型是概率分词器而在计算机里所有的词都是以一串数据表示换言之Transform模型是根据输入的数据去运算来判断用什么词输出的概率更大。说回输入这里提到词是以一串数据去输入的那么如下代码所示。src_vocab 10000 tgt_vocab 10000 model Transformer(src_vocab,tgt_vocab) src torch.randint(0,src_vocab,(32,10)) tgt torch.randint(0,tgt_vocab,(32,10))src_vocab 是输入的词汇总数tgt是输出的词汇总数。src和tgt分别代表了输入和输出词汇的一个案例简单说就是一个句子0维行表示一个token(词)1维列即一个词的特有的表示序列。PositionalEncoding输入句子后需要编码因为模型要知道单独的词在句子中的位置好知道和其他词是什么关系方便后面选择词的概率计算。代码如下。class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len 1000): super().__init__() #创建整个输入句子的编码矩阵 pe torch.zeros(max_len, d_model) #创建句子的每个词位置编码的数字 position torch.arange(0, max_len).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() *(-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x): seq_len x.size(1) return x self.pe[:, :seq_len ,:]这个代码按照原论文描述的公式去编码可能其他延伸模型有在这里改动过。用正/余弦编码的好处是易推导且不容易出问题也是为了长距离运算控制在范围内这里代码手算符合式子原来数值PE。Encoder整个Transform模型的核心多头注意力机制。如以下代码class SelfAttention(nn.Module): def __init__(self, drop_out): super().__init__() self.drop_out nn.Dropout(drop_out) self.softmax nn.Softmax(dim-1) def forward(self, Q, K, V, maskNone): d_k Q.size(-1) scores torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) attn self.softmax(scores) attn self.drop_out(attn) output torch.matmul(attn, V) return output, attn class MultiHeadAttention(nn.Module): def __init__(self, d_model, n_head, dropout0.1): super().__init__() assert d_model % n_head 0 self.n_head n_head self.d_k d_model // n_head self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.fc nn.Linear(d_model, d_model) self.attention SelfAttention(dropout) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, q, k, v, maskNone): batch_size q.size(0) Q self.W_q(q).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) K self.W_k(k).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) V self.W_v(v).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) output, attn self.attention(Q, K, V, mask) output output.transpose(1, 2).contiguous().view(batch_size, -1, self.n_head * self.d_k) output self.fc(output) output self.drop(output) return self.norm(outputq), attnSelfAttention这里先说自注意力机制没有单头这个概念。如下图所示很多教程都讲烂了先前的输入矩阵X每一行token x都会非线性成三部分Q、K、V。先说这三个怎么来的。q,k,v是同一个东西输入的句子但是这里肯定不止一句所以形状就成了b,len,word第零维就是输入句子的多少(batch)。然后经过线性变换成Q、K、V。然后是把Q、K、V传入一个自注意力中mask掩码在编码器里一般不用。引用菜鸟的比喻就是QQuery 你在搜索框里输入的关键词代表我要查什么KKey 每个网页的标题标签代表我能匹配哪些词VValue 每个网页的实际内容代表我能提供什么信息注意力权重 Q 与每个 K 的相似度最终结果 用权重对所有 V 加权求和。当然这里要始终认清词token是一组数据字符串Q*K是代表这组字符串对其他字符串相乘产生的比重V是模型中字符串的本身质量即代表句子的矩阵。softmax归一化是为了方便之后好找每个词与其他词对应的概率也为了比较各个样本之间的概率。sqrt(d_k)是为了避免坡度值缩小好优化参数这里可以参考用1216去归一化后的结果88%、12%原数只差4导致的概率差就够大的了。然后输出的out是各个词与其他词的概率attn矩阵乘以原来矩阵V然后还是一个二维张量。这里drop_out是为了舍弃部分参数避免发生过拟合情况最简单的Transform可以没有。MultiHeadAttention这里多头注意力机制可以类比CNN的多卷积核每个卷积注意不同的地方如下图所示。光看图可能很简单这里就直接说代码吧n_head是模型使用的头的数量、d_model是输入输出词的长度、这决定了d_k的数量是n_h//d_m即一个头注意的长度最后再全部拼接。W_q、k、v是线性变换重要参数这里要使用其去参与训练和学习最后预测也是其与输入词汇运行处理。所以搞成nn.Linear(d_model, d_model)为了分两步走将d_model维的向量切分成 n_head每份d_k维让每个头在不同的子空间中独立学习注意力模式。这样经过一次自注意力机制训练实际上已经完成了多头注意力机制的训练了然后将d_k维数组全部合并在送入全连接层一是为了特征融合相互交换多头的信息二是为了增加模型的表达能力因为自注意力本身是线性的不能产生新的概率。最后一步需要舍弃部分参数然后残差连接再归一化这是为了避免运算结果至少要和原来相差不大避免深层网络的梯度消失也让模型可以选择性忽略某层的变换。最后得到的结果经过编码后更加丰富的句子FeedForward经历了多头注意力机制后下一步需要进行前反馈神经网络运算对每个词进行独立的非线性变换。代码如下class FeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout0.1): super().__init__() self.fc1 nn.Linear(d_model, d_ff) self.fc2 nn.Linear(d_ff, d_model) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, x): out self.fc1(x) out torch.relu(out) out self.drop(out) # dropout应该在激活后 out self.fc2(out) return self.norm(out x)比较简单就是两次连接层去做非线性变换中间有一次激活函数然后结尾残差连接和归一化EncoderLayer然后我们可以根据多头注意力机制和前反馈神经网络构建编码层class EncoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, src,src_maskNone): out,_ self.self_attn(src,src,src,src_mask) out self.ffn(out) return out编码器根据编码层来构建整个编码器然后注意所有归一化操作已经在多头注意力机制和前反馈神经网络完成了。class Encoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model, max_len) self.layers nn.ModuleList([EncoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) def forward(self, src, src_maskNone): out self.embedding(src) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, src_mask) return out这里注意src就是输入的句子emdedding给每一个token编入一个IDDecoderMaskedMultiHeadAttention这里解释掩码的生成和运作原理def generate_mask(size): mask torch.triu(torch.ones(size, size), diagonal 1).bool() return mask 0使用函数生成一个下三角行列式然后与scores去做运算。这里之前犯了一个错误以为掩盖的是token本身实际是本token和其他之间的比重联系然后再进行归一化操作得到自注意力矩阵attn这一操作是为了避免出现后面的token对前者产生影响直观点说是模型学会偷懒直接复制未来信息概率相当于会定死然后训练效果达不到预期DecoderLayer解码层分成三步掩码自注意力、交叉注意力、一个前馈神经网络。class DecoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.cross_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, tgt,memory,tgt_maskNone,memory_maskNone): out,_ self.self_attn(tgt,tgt,tgt,tgt_mask) out,_ self.cross_attn(out,memory,memory,memory_mask) out self.ffn(out) return out自注意力Self-Attention让目标序列中的每个token关注其他位置捕捉内部依赖关系。tgt_mask是三角掩码防止看到未来位置。交叉注意力Cross-Attention让目标序列关注源序列编码器输出实现翻译/对齐功能。这里的memory就是编码器的输出。这个是区别编码器的地方额外引入交叉注意力实现翻译。比如翻译 I love you → 我爱你输入目标序列 [我, 爱, 你]输出每个词只能看到目标序列内部信息模型完全不知道我对应哪个源词翻译变成了随便生成一句话而不是翻译 。前馈网络ffn对每个位置独立进行非线性变换增强模型表达能力。解码器解码器和编码器一样由解码层堆叠后再归一化成且数量一致。不同的是还需要从编码器中得到memory。所以这里的K和V使用了编码器中的数据代表之前的可以查到的信息和原信息。Q是经过解码器后产生的代表当前要查询的信息。表示本token和其他token经过翻译后的值产生的概率生成的词。如以下代码所示class Decoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model,max_len) self.layers nn.ModuleList([DecoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) self.fc_out nn.Linear(d_model, vocab_size) def forward(self, tgt, memory, tgt_maskNone, memory_maskNone): out self.embedding(tgt) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, memory, tgt_mask, memory_mask) out self.fc_out(out) return outTransform经过了编码器和解码器的原理和代码的解释后我们可以将两者拼接成一个模型这步就非常简单了。class Transformer(nn.Module): def __init__(self, src_vocab, tgt_vocab, d_model 512, n_head 8, num_encoder_layers 6, num_decoder_layers 6, d_ff 2048, dropout0.1, max_len 5000,): super().__init__() self.encoder Encoder(src_vocab,d_model,n_head,num_encoder_layers,d_ff,dropout,max_len) self.decoder Decoder(tgt_vocab,d_model,n_head,num_decoder_layers,d_ff,dropout,max_len) def forward(self, src, tgt, src_maskNone, tgt_maskNone, memory_maskNone): memory self.encoder(src, src_masksrc_mask) out self.decoder(tgt, memory, tgt_masktgt_mask, memory_maskmemory_mask) return out这里还是推荐读一遍原论文 attention is all you need.最后完整代码如下所示import torch import math from torch import nn class SelfAttention(nn.Module): def __init__(self, drop_out): super().__init__() self.drop_out nn.Dropout(drop_out) self.softmax nn.Softmax(dim-1) def forward(self, Q, K, V, maskNone): d_k Q.size(-1) scores torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) attn self.softmax(scores) attn self.drop_out(attn) output torch.matmul(attn, V) return output, attn class MultiHeadAttention(nn.Module): def __init__(self, d_model, n_head, dropout0.1): super().__init__() assert d_model % n_head 0 self.n_head n_head self.d_k d_model // n_head self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.fc nn.Linear(d_model, d_model) self.attention SelfAttention(dropout) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, q, k, v, maskNone): batch_size q.size(0) Q self.W_q(q).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) K self.W_k(k).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) V self.W_v(v).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) output, attn self.attention(Q, K, V, mask) output output.transpose(1, 2).contiguous().view(batch_size, -1, self.n_head * self.d_k) output self.fc(output) output self.drop(output) return self.norm(outputq), attn class FeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout0.1): super().__init__() self.fc1 nn.Linear(d_model, d_ff) self.fc2 nn.Linear(d_ff, d_model) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, x): out self.fc1(x) out torch.relu(out) out self.drop(out) # dropout应该在激活后 out self.fc2(out) return self.norm(out x) # def forward(self, x): # out self.fc2(self.drop(torch.relu(self.fc1(x)))) # return self.norm(outx) class EncoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, src,src_maskNone): out,_ self.self_attn(src,src,src,src_mask) out self.ffn(out) return out class DecoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.cross_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, tgt,memory,tgt_maskNone,memory_maskNone): out,_ self.self_attn(tgt,tgt,tgt,tgt_mask) out,_ self.cross_attn(out,memory,memory,memory_mask) out self.ffn(out) return out class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len 1000): super().__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() *(-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x): seq_len x.size(1) return x self.pe[:, :seq_len ,:] class Encoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model, max_len) self.layers nn.ModuleList([EncoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) def forward(self, src, src_maskNone): out self.embedding(src) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, src_mask) return out class Decoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model,max_len) self.layers nn.ModuleList([DecoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) self.fc_out nn.Linear(d_model, vocab_size) def forward(self, tgt, memory, tgt_maskNone, memory_maskNone): out self.embedding(tgt) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, memory, tgt_mask, memory_mask) out self.fc_out(out) return out class Transformer(nn.Module): def __init__(self, src_vocab, tgt_vocab, d_model 512, n_head 8, num_encoder_layers 6, num_decoder_layers 6, d_ff 2048, dropout0.1, max_len 5000,): super().__init__() self.encoder Encoder(src_vocab,d_model,n_head,num_encoder_layers,d_ff,dropout,max_len) self.decoder Decoder(tgt_vocab,d_model,n_head,num_decoder_layers,d_ff,dropout,max_len) def forward(self, src, tgt, src_maskNone, tgt_maskNone, memory_maskNone): memory self.encoder(src, src_masksrc_mask) out self.decoder(tgt, memory, tgt_masktgt_mask, memory_maskmemory_mask) return out def generate_mask(size): mask torch.triu(torch.ones(size, size), diagonal 1).bool() return mask 0 if __name__ __main__: src_vocab 10000 tgt_vocab 10000 model Transformer(src_vocab,tgt_vocab) src torch.randint(0,src_vocab,(32,10)) tgt torch.randint(0,tgt_vocab,(32,10)) tgt_mask generate_mask(tgt.size(1)).to(tgt.device) output model(src,tgt,tgt_masktgt_mask) print(output.shape) Transformer模型个人总结
写这篇文章的目的是为了理清现在很多模型的基础Transformer模型是怎么回事。为此自己也是跟着视频敲了一遍代码但仍然感觉没有吃透所以再写一篇。下图是模型的架构。source sequence说模型必须弄明白输入和输出的数据结构形状是怎么样的有位学长说过现在的模型是概率分词器而在计算机里所有的词都是以一串数据表示换言之Transform模型是根据输入的数据去运算来判断用什么词输出的概率更大。说回输入这里提到词是以一串数据去输入的那么如下代码所示。src_vocab 10000 tgt_vocab 10000 model Transformer(src_vocab,tgt_vocab) src torch.randint(0,src_vocab,(32,10)) tgt torch.randint(0,tgt_vocab,(32,10))src_vocab 是输入的词汇总数tgt是输出的词汇总数。src和tgt分别代表了输入和输出词汇的一个案例简单说就是一个句子0维行表示一个token(词)1维列即一个词的特有的表示序列。PositionalEncoding输入句子后需要编码因为模型要知道单独的词在句子中的位置好知道和其他词是什么关系方便后面选择词的概率计算。代码如下。class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len 1000): super().__init__() #创建整个输入句子的编码矩阵 pe torch.zeros(max_len, d_model) #创建句子的每个词位置编码的数字 position torch.arange(0, max_len).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() *(-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x): seq_len x.size(1) return x self.pe[:, :seq_len ,:]这个代码按照原论文描述的公式去编码可能其他延伸模型有在这里改动过。用正/余弦编码的好处是易推导且不容易出问题也是为了长距离运算控制在范围内这里代码手算符合式子原来数值PE。Encoder整个Transform模型的核心多头注意力机制。如以下代码class SelfAttention(nn.Module): def __init__(self, drop_out): super().__init__() self.drop_out nn.Dropout(drop_out) self.softmax nn.Softmax(dim-1) def forward(self, Q, K, V, maskNone): d_k Q.size(-1) scores torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) attn self.softmax(scores) attn self.drop_out(attn) output torch.matmul(attn, V) return output, attn class MultiHeadAttention(nn.Module): def __init__(self, d_model, n_head, dropout0.1): super().__init__() assert d_model % n_head 0 self.n_head n_head self.d_k d_model // n_head self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.fc nn.Linear(d_model, d_model) self.attention SelfAttention(dropout) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, q, k, v, maskNone): batch_size q.size(0) Q self.W_q(q).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) K self.W_k(k).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) V self.W_v(v).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) output, attn self.attention(Q, K, V, mask) output output.transpose(1, 2).contiguous().view(batch_size, -1, self.n_head * self.d_k) output self.fc(output) output self.drop(output) return self.norm(outputq), attnSelfAttention这里先说自注意力机制没有单头这个概念。如下图所示很多教程都讲烂了先前的输入矩阵X每一行token x都会非线性成三部分Q、K、V。先说这三个怎么来的。q,k,v是同一个东西输入的句子但是这里肯定不止一句所以形状就成了b,len,word第零维就是输入句子的多少(batch)。然后经过线性变换成Q、K、V。然后是把Q、K、V传入一个自注意力中mask掩码在编码器里一般不用。引用菜鸟的比喻就是QQuery 你在搜索框里输入的关键词代表我要查什么KKey 每个网页的标题标签代表我能匹配哪些词VValue 每个网页的实际内容代表我能提供什么信息注意力权重 Q 与每个 K 的相似度最终结果 用权重对所有 V 加权求和。当然这里要始终认清词token是一组数据字符串Q*K是代表这组字符串对其他字符串相乘产生的比重V是模型中字符串的本身质量即代表句子的矩阵。softmax归一化是为了方便之后好找每个词与其他词对应的概率也为了比较各个样本之间的概率。sqrt(d_k)是为了避免坡度值缩小好优化参数这里可以参考用1216去归一化后的结果88%、12%原数只差4导致的概率差就够大的了。然后输出的out是各个词与其他词的概率attn矩阵乘以原来矩阵V然后还是一个二维张量。这里drop_out是为了舍弃部分参数避免发生过拟合情况最简单的Transform可以没有。MultiHeadAttention这里多头注意力机制可以类比CNN的多卷积核每个卷积注意不同的地方如下图所示。光看图可能很简单这里就直接说代码吧n_head是模型使用的头的数量、d_model是输入输出词的长度、这决定了d_k的数量是n_h//d_m即一个头注意的长度最后再全部拼接。W_q、k、v是线性变换重要参数这里要使用其去参与训练和学习最后预测也是其与输入词汇运行处理。所以搞成nn.Linear(d_model, d_model)为了分两步走将d_model维的向量切分成 n_head每份d_k维让每个头在不同的子空间中独立学习注意力模式。这样经过一次自注意力机制训练实际上已经完成了多头注意力机制的训练了然后将d_k维数组全部合并在送入全连接层一是为了特征融合相互交换多头的信息二是为了增加模型的表达能力因为自注意力本身是线性的不能产生新的概率。最后一步需要舍弃部分参数然后残差连接再归一化这是为了避免运算结果至少要和原来相差不大避免深层网络的梯度消失也让模型可以选择性忽略某层的变换。最后得到的结果经过编码后更加丰富的句子FeedForward经历了多头注意力机制后下一步需要进行前反馈神经网络运算对每个词进行独立的非线性变换。代码如下class FeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout0.1): super().__init__() self.fc1 nn.Linear(d_model, d_ff) self.fc2 nn.Linear(d_ff, d_model) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, x): out self.fc1(x) out torch.relu(out) out self.drop(out) # dropout应该在激活后 out self.fc2(out) return self.norm(out x)比较简单就是两次连接层去做非线性变换中间有一次激活函数然后结尾残差连接和归一化EncoderLayer然后我们可以根据多头注意力机制和前反馈神经网络构建编码层class EncoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, src,src_maskNone): out,_ self.self_attn(src,src,src,src_mask) out self.ffn(out) return out编码器根据编码层来构建整个编码器然后注意所有归一化操作已经在多头注意力机制和前反馈神经网络完成了。class Encoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model, max_len) self.layers nn.ModuleList([EncoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) def forward(self, src, src_maskNone): out self.embedding(src) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, src_mask) return out这里注意src就是输入的句子emdedding给每一个token编入一个IDDecoderMaskedMultiHeadAttention这里解释掩码的生成和运作原理def generate_mask(size): mask torch.triu(torch.ones(size, size), diagonal 1).bool() return mask 0使用函数生成一个下三角行列式然后与scores去做运算。这里之前犯了一个错误以为掩盖的是token本身实际是本token和其他之间的比重联系然后再进行归一化操作得到自注意力矩阵attn这一操作是为了避免出现后面的token对前者产生影响直观点说是模型学会偷懒直接复制未来信息概率相当于会定死然后训练效果达不到预期DecoderLayer解码层分成三步掩码自注意力、交叉注意力、一个前馈神经网络。class DecoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.cross_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, tgt,memory,tgt_maskNone,memory_maskNone): out,_ self.self_attn(tgt,tgt,tgt,tgt_mask) out,_ self.cross_attn(out,memory,memory,memory_mask) out self.ffn(out) return out自注意力Self-Attention让目标序列中的每个token关注其他位置捕捉内部依赖关系。tgt_mask是三角掩码防止看到未来位置。交叉注意力Cross-Attention让目标序列关注源序列编码器输出实现翻译/对齐功能。这里的memory就是编码器的输出。这个是区别编码器的地方额外引入交叉注意力实现翻译。比如翻译 I love you → 我爱你输入目标序列 [我, 爱, 你]输出每个词只能看到目标序列内部信息模型完全不知道我对应哪个源词翻译变成了随便生成一句话而不是翻译 。前馈网络ffn对每个位置独立进行非线性变换增强模型表达能力。解码器解码器和编码器一样由解码层堆叠后再归一化成且数量一致。不同的是还需要从编码器中得到memory。所以这里的K和V使用了编码器中的数据代表之前的可以查到的信息和原信息。Q是经过解码器后产生的代表当前要查询的信息。表示本token和其他token经过翻译后的值产生的概率生成的词。如以下代码所示class Decoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model,max_len) self.layers nn.ModuleList([DecoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) self.fc_out nn.Linear(d_model, vocab_size) def forward(self, tgt, memory, tgt_maskNone, memory_maskNone): out self.embedding(tgt) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, memory, tgt_mask, memory_mask) out self.fc_out(out) return outTransform经过了编码器和解码器的原理和代码的解释后我们可以将两者拼接成一个模型这步就非常简单了。class Transformer(nn.Module): def __init__(self, src_vocab, tgt_vocab, d_model 512, n_head 8, num_encoder_layers 6, num_decoder_layers 6, d_ff 2048, dropout0.1, max_len 5000,): super().__init__() self.encoder Encoder(src_vocab,d_model,n_head,num_encoder_layers,d_ff,dropout,max_len) self.decoder Decoder(tgt_vocab,d_model,n_head,num_decoder_layers,d_ff,dropout,max_len) def forward(self, src, tgt, src_maskNone, tgt_maskNone, memory_maskNone): memory self.encoder(src, src_masksrc_mask) out self.decoder(tgt, memory, tgt_masktgt_mask, memory_maskmemory_mask) return out这里还是推荐读一遍原论文 attention is all you need.最后完整代码如下所示import torch import math from torch import nn class SelfAttention(nn.Module): def __init__(self, drop_out): super().__init__() self.drop_out nn.Dropout(drop_out) self.softmax nn.Softmax(dim-1) def forward(self, Q, K, V, maskNone): d_k Q.size(-1) scores torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) attn self.softmax(scores) attn self.drop_out(attn) output torch.matmul(attn, V) return output, attn class MultiHeadAttention(nn.Module): def __init__(self, d_model, n_head, dropout0.1): super().__init__() assert d_model % n_head 0 self.n_head n_head self.d_k d_model // n_head self.W_q nn.Linear(d_model, d_model) self.W_k nn.Linear(d_model, d_model) self.W_v nn.Linear(d_model, d_model) self.fc nn.Linear(d_model, d_model) self.attention SelfAttention(dropout) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, q, k, v, maskNone): batch_size q.size(0) Q self.W_q(q).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) K self.W_k(k).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) V self.W_v(v).view(batch_size, -1, self.n_head, self.d_k).transpose(1, 2) output, attn self.attention(Q, K, V, mask) output output.transpose(1, 2).contiguous().view(batch_size, -1, self.n_head * self.d_k) output self.fc(output) output self.drop(output) return self.norm(outputq), attn class FeedForward(nn.Module): def __init__(self, d_model, d_ff, dropout0.1): super().__init__() self.fc1 nn.Linear(d_model, d_ff) self.fc2 nn.Linear(d_ff, d_model) self.drop nn.Dropout(dropout) self.norm nn.LayerNorm(d_model) def forward(self, x): out self.fc1(x) out torch.relu(out) out self.drop(out) # dropout应该在激活后 out self.fc2(out) return self.norm(out x) # def forward(self, x): # out self.fc2(self.drop(torch.relu(self.fc1(x)))) # return self.norm(outx) class EncoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, src,src_maskNone): out,_ self.self_attn(src,src,src,src_mask) out self.ffn(out) return out class DecoderLayer(nn.Module): def __init__(self, d_model, n_head, d_ff, dropout0.1): super().__init__() self.self_attn MultiHeadAttention(d_model, n_head, dropout) self.cross_attn MultiHeadAttention(d_model, n_head, dropout) self.ffn FeedForward(d_model, d_ff, dropout) def forward(self, tgt,memory,tgt_maskNone,memory_maskNone): out,_ self.self_attn(tgt,tgt,tgt,tgt_mask) out,_ self.cross_attn(out,memory,memory,memory_mask) out self.ffn(out) return out class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len 1000): super().__init__() pe torch.zeros(max_len, d_model) position torch.arange(0, max_len).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() *(-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) pe pe.unsqueeze(0) self.register_buffer(pe, pe) def forward(self, x): seq_len x.size(1) return x self.pe[:, :seq_len ,:] class Encoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model, max_len) self.layers nn.ModuleList([EncoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) def forward(self, src, src_maskNone): out self.embedding(src) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, src_mask) return out class Decoder(nn.Module): def __init__(self, vocab_size, d_model, n_head, num_layer, d_ff, dropout0.1, max_len5000): super().__init__() self.embedding nn.Embedding(vocab_size, d_model) self.position PositionalEncoding(d_model,max_len) self.layers nn.ModuleList([DecoderLayer(d_model, n_head, d_ff, dropout) for _ in range(num_layer) ]) self.fc_out nn.Linear(d_model, vocab_size) def forward(self, tgt, memory, tgt_maskNone, memory_maskNone): out self.embedding(tgt) * math.sqrt(self.embedding.embedding_dim) out self.position(out) for layer in self.layers: out layer(out, memory, tgt_mask, memory_mask) out self.fc_out(out) return out class Transformer(nn.Module): def __init__(self, src_vocab, tgt_vocab, d_model 512, n_head 8, num_encoder_layers 6, num_decoder_layers 6, d_ff 2048, dropout0.1, max_len 5000,): super().__init__() self.encoder Encoder(src_vocab,d_model,n_head,num_encoder_layers,d_ff,dropout,max_len) self.decoder Decoder(tgt_vocab,d_model,n_head,num_decoder_layers,d_ff,dropout,max_len) def forward(self, src, tgt, src_maskNone, tgt_maskNone, memory_maskNone): memory self.encoder(src, src_masksrc_mask) out self.decoder(tgt, memory, tgt_masktgt_mask, memory_maskmemory_mask) return out def generate_mask(size): mask torch.triu(torch.ones(size, size), diagonal 1).bool() return mask 0 if __name__ __main__: src_vocab 10000 tgt_vocab 10000 model Transformer(src_vocab,tgt_vocab) src torch.randint(0,src_vocab,(32,10)) tgt torch.randint(0,tgt_vocab,(32,10)) tgt_mask generate_mask(tgt.size(1)).to(tgt.device) output model(src,tgt,tgt_masktgt_mask) print(output.shape)