1.词嵌入层输入层Transformer 词嵌入层 Token Embedding词向量 Positional Encoding位置编码Embedding层把离散单词编号token 索引→ 连续语义向量位置编码PE为向量加入序列位置信息最终输出x Embedding(Token) PositionalEncoding(pos)代码如下import math import torch import torch.nn as nn class Embedings(nn.Module): def __init__(self, d_module, vocab): super().__init__() self.d_module d_module self.lut nn.Embedding(vocab, d_module) def forward(self, x): return self.lut(x) * math.sqrt(self.d_module) # 定义位置编码器类, 我们同样把它看做一个层, 因此会继承nn.Module class PositionalEncoding(nn.Module): def __init__(self, d_model, dropout, max_len5000): super().__init__() self.dropout nn.Dropout(pdropout) # 初始化一个位置编码矩阵, 它是一个0阵矩阵的大小是max_len x d_model. 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) * -(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): x, 表示文本序列的词嵌入表示 x x self.pe[:, :x.size(1)] return self.dropout(x) class EmbeddingWithPosition(nn.Module): def __init__(self, d_module, vocab, dropout0.1): super().__init__() self.emb Embedings(d_module, vocab) self.pe PositionalEncoding(d_module, dropout) def forward(self, x): x self.emb(x) x self.pe(x) return x2.编码器层Transformer 编码器与传统 Seq2Seq 架构最大不同在于编码器摒弃了 RNN 串行循环结构采用多头自注意力机制进行全局并行特征建模。编码器自注意力属于自关联注意力满足QKV即查询向量、键向量、值向量均由同一输入序列映射得到用于建模序列内部词与词之间的依赖关系。同时编码器引入掩码 Mask 机制对填充无效位置进行屏蔽避免模型学习无意义信息再结合残差连接、LayerNorm 归一化与前馈全连接网络构成完整稳定的编码器结构。抛弃了传统的seq2seq的 RNN / LSTM 的循环结构不再逐词进行串行计算改用子注意力机制并行计算编码器内部是自注意力机制 self-attention 满足QKV使用掩码mask屏蔽无效位置 / 未来位置 防止信息泄露搭配残差连接 LayerNormal 前馈网络多头注意力机制中的核心张量维度转换1.输入的张量 [batch_size, seq_len] -- [6, 4] 一次送进去6句话每句话4个单词2.张量经过词嵌入层 [batch_size, seq_len, d_module] -- [6, 4, 512] 假设词嵌入曾的输出维度为5123.张量经过多头注意力机制层假设有8个头(1).张量先变成-- [batch_size, seq_len, head, d_k] -- [6, 4, 8, 64] 512 / 8 64(2).变换维度-- [batch_size, head, seq_len, d_k] -- [6, 8, 4, 64](3) 张量 Q * KT(转置) [6, 8, 4, 64] * [6, 8, 64, 4] -- [6, 8, 4, 4](4)使用掩码张量 mask 的形状为 [1, 1, seq_len, seq_len] 或者 [1, seq_len, seq_len] 或者[head, seq_len, seq_len] 这里使用了广播机制(5)将得到的张量 * V [6, 8, 4, 4] * [6, 8, 4, 64] -- [6, 8, 4, 64](6)最后先变化维度为 [6, 4, 8, 64] , 再变化维度为 [6, 4, 512] 这时与词嵌入层输出的维度一致 由N个编码器构成 一个norm层构成 每个编码器包含 1.多头注意力机制层 残差连接层 多头注意力机制 4个linear层 attention中加随机失活层 2.前馈全连接层 残差连接层 前馈全连接层 2个linear层 随机失活层 import copy import math import torch import torch.nn as nn from torch.nn import Dropout # 定义的模块克隆函数 def clones(module, N): return nn.ModuleList([copy.deepcopy(module) for _ in range(N)]) # attention函数 def attention(query, key, value, maskNone, dropoutNone): # 获得注意力张量 d_k query.size(-1) scores torch.matmul(query, key.transpose(-1, -2)) / math.sqrt(d_k) if mask is not None: scores scores.masked_fill(mask 0, -1e9) p_atten scores.softmax(dim-1) if dropout is not None: p_atten dropout(p_atten) return torch.matmul(p_atten, value), p_atten # 定义多头注意力机制 class MultiHeadAttention(nn.Module): def __init__(self, head, embedding_dim, dropout0.1): super().__init__() # 先判断词嵌入的维度是否可以整除head assert embedding_dim % head 0 # 得到词嵌入维度分头后的维度 self.d_k embedding_dim // head self.head head # 定义4个线性层Q,K,V需要先经过一个线性层输出还要经过一个线性层 self.linears clones(nn.Linear(embedding_dim, embedding_dim), 4) self.dropout nn.Dropout(pdropout) def forward(self, query, key ,value, maskNone): # 输入的掩码维度是三维但是要作用再分头后的结果上四维所以要扩展一维 if mask is not None: mask.unsqueeze(0) batch_size query.size(0) # 经过线性层并转换维度 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))] # query self.linears[0](query).view(batch_size, -1, self.head, self.d_k).transpose(1, 2) # key self.linears[1](key).view(batch_size, -1, self.head, self.d_k).transpose(1, 2) # value self.linears[2](value).view(batch_size, -1, self.head, self.d_k).transpose(1, 2) # 经过attention函数 atten_output, atten_weight attention(query, key, value, maskmask, dropoutself.dropout) # 将等到的注意力输出转化维度 atten_output atten_output.transpose(1, 2).contiguous().view(batch_size, -1, self.head * self.d_k) # 使用最后一个线性层返回结果 return self.linears[-1](atten_output) # 定义前馈连接层相当增加两个线性层 class FeedForward(nn.Module): def __init__(self, d_module, d_ff, dropout0.1): super().__init__() self.linear1 nn.Linear(d_module, d_ff) self.linear2 nn.Linear(d_ff, d_module) self.dropout Dropout(pdropout) def forward(self, x): # 经过第一个线性层再经过relu函数激活再经过随机失活层再经过线性层 return self.linear2(self.dropout(self.linear1(x).relu())) # 定义规范化层 class LayerNorm(nn.Module): def __init__(self, d_module, eps1e-6): super().__init__() # 包含一个残差计算和一个ADD层 # d_module 词嵌入的维度 eps防止分母出现0 self.k nn.Parameter(torch.ones(d_module)) self.b nn.Parameter(torch.zeros(d_module)) self.eps eps def forward(self, x): # 归一化 mean x.mean(-1, keepdimTrue) std x.std(-1, keepdimTrue) norm_x self.k * (x - mean) / (std self.eps) self.b return norm_x # 定义残差归一化层 class SublayerConnection(nn.Module): def __init__(self, d_module, dropout0.1): super().__init__() self.norm LayerNorm(d_module) self.dropout nn.Dropout(pdropout) def forward(self, x, sublayer): return x self.dropout(sublayer(self.norm(x))) # 定义编码器层 class EncoderLayer(nn.Module): def __init__(self, d_module, self_atten, feed_forward, dropout0.1): super().__init__() # d_module 词嵌入维度 self_atten 多头注意力机制对象 feed_forward 前馈全连接对象 self.self_atten self_atten self.feed_forward feed_forward self.d_module d_module self.sublayers clones(SublayerConnection(d_module, dropout), 2) def forward(self, x, mask): # FeedForward 只需要 x所以能直接传对象 # Attention 需要 x,x,x,mask所以必须用 lambda 包成单参数函数 x self.sublayers[0](x, lambda x: self.self_atten(x, x, x, mask)) return self.sublayers[1](x, lambda x: self.feed_forward(x)) # 最终的编码器 class Encoder(nn.Module): def __init__(self, encoderLayer, N): super().__init__() self.encoderLayers clones(encoderLayer, N) self.norm LayerNorm(encoderLayer.d_module) def forward(self, x, mask): # 经过N个编码器层的处理进行归一化返回得到结果 for layer in self.encoderLayers: x layer(x, mask) return self.norm(x) def test_encoder(): # 模拟输入 x torch.randn(2, 4, 512) mask torch.zeros(8, 4, 4) # 实例化多头注意力机制模型 atten_encoder_layer MultiHeadAttention(head8, embedding_dim512, dropout0.1) # 实例化前馈连接层 ff_layer FeedForward(512, 2048) # 实例化单个encoderLayer模型 single_encoder EncoderLayer(512, atten_encoder_layer, ff_layer, dropout0.1) # 实例化encoder encoder Encoder(single_encoder, 6) print(encoder) output encoder(x, mask) print(f编码器输出{output}{output.shape}) return output if __name__ __main__: test_encoder()3.解码器层解码器和编码器没什么大区别只不过又多了一层注意力机制import torch import torch.nn as nn import torch.nn.functional as F from torch.nn import Dropout from Transformer.encoder.encoder import * class DecoderLayer(nn.Module): def __init__(self, d_module, self_atten, src_atten, feed_forward, dropout0.1): super().__init__() # d_module 输入的词嵌入维度大小 # self_atten 多头注意力机制对象 QKV # src_atten 多头注意力对象 Q!KV # feed_forward 前馈连接层 self.d_module d_module self.self_atten self_atten self.src_atten src_atten self.feed_forward feed_forward self.dropout Dropout(pdropout) self.sublayers clones(SublayerConnection(d_module, dropout), 3) def forward(self, x, encoder_output, target_mask, source_mask): # x 目标输入的词嵌入 # encoder_output 编码器输出 # target_mask 目标句子的掩码 # source_mask 目标和源的掩码 # 经过第一个注意力机制 x self.sublayers[0](x, lambda x : self.self_atten(x, x, x, target_mask)) # 经过第二个注意力机制 x self.sublayers[1](x, lambda x : self.src_atten(x, encoder_output, encoder_output, source_mask)) # 经过前馈连接 return self.sublayers[2](x, lambda x: self.feed_forward(x)) class Decoder(nn.Module): def __init__(self, decoderlayer, N): super().__init__() self.layers clones(decoderlayer, N) self.norm LayerNorm(decoderlayer.d_module) def forward(self, x, encoder_output, target_mask, source_mask): for layer in self.layers: x layer(x, encoder_output, target_mask, source_mask) return self.norm(x) def test_decoder(): encoder_output test_encoder() # 模拟输入 y torch.randn(2, 6, 512) # 实例化多头注意力机制模型1 atten_decoder_layer1 MultiHeadAttention(head8, embedding_dim512, dropout0.1) atten_decoder_layer2 MultiHeadAttention(head8, embedding_dim512, dropout0.1) # 实例化前馈连接层 ff_layer FeedForward(512, 2048) signal_decoder_layer DecoderLayer(512, atten_decoder_layer1, atten_decoder_layer2, ff_layer, dropout0.1) target_mask torch.zeros(8, 6, 6) source_mask torch.zeros(8, 6, 4) decoder Decoder(signal_decoder_layer, 6) print(f解码器模型{decoder}) decoder_output decoder(y, encoder_output, target_mask, source_mask) print(f解码器输出{decoder_output},{decoder_output.shape}) if __name__ __main__: test_decoder()4.输出层通过一个线性层分类from Transformer.decoder.decoder import * class Generator(nn.Module): def __init__(self, d_module, vocab_size): super().__init__() self.output nn.Linear(d_module, vocab_size) def forward(self, x): return F.log_softmax(self.output(x), dim-1)5.整体模型组装from Transformer.encoder.encoder import * from Transformer.decoder.decoder import * from Transformer.output.generator import * from Transformer.embedding.embedding import * class EncoderDecoder(nn.Module): def __init__(self, encoder, decoder, src_embed, tgt_embed, generator): super().__init__() self.encoder encoder self.decoder decoder self.src_embed src_embed self.tgt_embed tgt_embed self.generator generator def forward(self, src, tgt, src_mask1, src_mask2, tgt_mask): # 获得输入词嵌入输出 src_embed_out self.src_embed(src) encoder_out self.encoder(src_embed_out, src_mask1) # 获得目标词嵌入输出 tgt_embed_out self.tgt_embed(tgt) decoder_out self.decoder(tgt_embed_out, encoder_out, tgt_mask, src_mask2) return self.generator(decoder_out) def test_EncoderDecoder(): # 实例化编码器 x torch.randn(2, 4, 512) mask torch.zeros(8, 4, 4) atten_encoder_layer MultiHeadAttention(head8, embedding_dim512, dropout0.1) ff_layer1 FeedForward(512, 2048) single_encoder EncoderLayer(512, atten_encoder_layer, ff_layer1, dropout0.1) encoder Encoder(single_encoder, 6) # 实例化解码器 atten_decoder_layer1 MultiHeadAttention(head8, embedding_dim512, dropout0.1) atten_decoder_layer2 MultiHeadAttention(head8, embedding_dim512, dropout0.1) ff_layer2 FeedForward(512, 2048) signal_decoder_layer DecoderLayer(512, atten_decoder_layer1, atten_decoder_layer2, ff_layer2, dropout0.1) decoder Decoder(signal_decoder_layer, 6) src_embed EmbeddingWithPosition(d_module512, vocab2000) tgt_embed EmbeddingWithPosition(d_module512, vocab2000) generator Generator(d_module512, vocab_size2000) encoderdecoder EncoderDecoder(encoder, decoder, src_embed, tgt_embed, generator) print(encoderdecoder) # 模拟输入 x torch.randint(0, 2000, (2, 4)).long() # 词汇表 0~1999 的整数 y torch.randint(0, 2000, (2, 6)).long() src_mask1 torch.zeros(8, 4, 4) src_mask2 torch.zeros(8, 6, 4) tgt_mask torch.zeros(8, 6, 6) output encoderdecoder(x, y, src_mask1, src_mask2, tgt_mask) print(output, output.shape) if __name__ __main__: test_EncoderDecoder()6.简单案例使用上面搭建的transformer做一个简单分类任务from Transformer.EncoderDecoder import EncoderDecoder from Transformer.encoder.encoder import * from Transformer.decoder.decoder import * from Transformer.output.generator import * from Transformer.embedding.embedding import * import torch # 生成真正的掩码矩阵 def subsequent_mask(size): 生成下三角掩码屏蔽未来位置 输出 shape: [1, size, size] 下三角为 1可见上三角为 0屏蔽 mask torch.ones(size, size).triu(diagonal1) return 1 - mask.unsqueeze(0) # 输出 [1, size, size] # 刺蛾二分类简易数据集 # 0padding1 体型小2 体型大3 绿色4 褐色5 有毒毛6 无毒毛7 花纹明显8 花纹模糊 # 刺蛾样本正类标签1 moth_spiny [ [3,5,1,7], # 绿色毒毛小体型刺蛾 [3,5,7,1], [5,3,1,7] ] # 非刺蛾负类标签0 moth_normal [ [4,6,2,8], # 褐色无毒毛非刺蛾 [6,4,8,2], [4,2,6,8] ] # 转tensor src_data torch.LongTensor(moth_spiny moth_normal) labels torch.LongTensor([1]*3 [0]*3) batch_size 6 seq_len 4 vocab_size 9 d_model 512 src_mask1 subsequent_mask(seq_len) src_mask2 subsequent_mask(seq_len) tgt_mask subsequent_mask(seq_len) # 1. 多头注意力 atten_encoder_layer MultiHeadAttention(head8, embedding_dim512, dropout0.1) ff_layer1 FeedForward(512, 2048) single_encoder EncoderLayer(512, atten_encoder_layer, ff_layer1, dropout0.1) encoder Encoder(single_encoder, N2) # 刺蛾简单任务不用6层2层足够 # 解码器 atten_decoder_layer1 MultiHeadAttention(head8, embedding_dim512, dropout0.1) atten_decoder_layer2 MultiHeadAttention(head8, embedding_dim512, dropout0.1) ff_layer2 FeedForward(512, 2048) signal_decoder_layer DecoderLayer(512, atten_decoder_layer1, atten_decoder_layer2, ff_layer2, dropout0.1) decoder Decoder(signal_decoder_layer, N2) # 词嵌入位置编码 src_embed EmbeddingWithPosition(d_module512, vocabvocab_size) tgt_embed EmbeddingWithPosition(d_module512, vocabvocab_size) # 二分类输出2个神经元刺蛾/非刺蛾 generator Generator(d_module512, vocab_size2) # 完整Transformer模型 model EncoderDecoder(encoder, decoder, src_embed, tgt_embed, generator) # 优化器损失 optimizer torch.optim.Adam(model.parameters(), lr1e-4) criterion F.nll_loss # 训练一轮演示刺蛾案例极简训练 model.train() for epoch in range(20): optimizer.zero_grad() # 前向输入刺蛾特征序列srctarget用自身序列自编码分类 out model(src_data, src_data, src_mask1, src_mask2, tgt_mask) # 取最后一个位置分类 pred out[:, -1, :] loss criterion(pred, labels) loss.backward() optimizer.step() if epoch % 5 0: print(fepoch:{epoch} loss:{loss.item():.4f}) # 测试预测 model.eval() with torch.no_grad(): out model(src_data, src_data, src_mask1, src_mask2, tgt_mask) pred out[:, -1, :].argmax(dim-1) print(真实标签:, labels.tolist()) print(预测结果:, pred.tolist()) print(刺蛾识别Transformer运行完成)