回头看来,Attention Is All You Need这篇文章是所有大模型的根基,从MLP、CNN、RNN发展而来的船新版本特征提取器。这次跟着油管UP Umar Jamil对照论文通过手写transformer来学习这个划时代的算法。Coding a Transformer from scratch on PyTorch, with full explanation, training and inference。本文以翻译任务为例,展示如何使用pytorch手写一个将英语翻译为意大利语的Transformer。

整个Transformer模型可以大致分为两个部分,Encoder和Decoder。新建文件model.py用于编写Transformer模型。
1 Encoder
Encoder由个encoder模块首尾相接组合而成,每个encoder模块又包含了两个由Residual连接(也译为残差连接)的网络。
1.1 Input Embedding

第一步是embedding(也译为词嵌入),就是给将单词表里的每个词一个编号,然后将这个编号映射为一个词向量,在本文中词向量大小为512维度。需要注意的是在论文3.4节中提到embedding需要乘以。
import torchimport torch.nn as nnimport math
class InputEmbeddings(nn.Module):
# `d_model`: 模型维度,本文中为512 # `vocab_size`: 词表大小 def __init__(self, d_model: int, vocab_size: int) -> None: super().__init__() self.d_model = d_model self.vocab_size = vocab_size self.embedding = nn.Embedding(vocab_size, d_model) # 将一个数字映射为一个向量,这里的Embedding层包含vocab_size个大小为d_model的向量
def forward(self, x): # (batch, seq_len) --> (batch, seq_len, d_model) # 论文3.4节Embeddings and Softmax中提到embedding要乘以d_model的平方根 return self.embedding(x) * math.sqrt(self.d_model)1.2 Positional Encoding

第二步需要将每个单词在句子中的位置信息告诉模型,位置编码与embedding的维度相同(本文中为512),然后将这个位置编码与前面的embedding相加。位置编码矩阵大小为(seq_len, d_model)。

在位置向量的偶数元素使用公式,在奇数位置向量的奇数元素使用公式。为了计算的数值稳定性,我们对分母取对数和指数化简得到(这一项是分母所以代码中乘以这一项的负指数)。
class PositionalEncoding(nn.Module):
# `d_model`: 模型维度,本文中为512 # `seq_len`: 句子的最大长度,为每个位置分配一个向量 # `dropout`: 添加dropout层抑制过拟合 # 位置信息只计算一次,在之后的训练中用在每个句子上 def __init__(self, d_model: int, seq_len: int, dropout: float) -> None: super().__init__() self.d_model = d_model self.seq_len = seq_len self.dropout = nn.Dropout(dropout)
# 创建维度为(seq_len, d_model)的矩阵 pe = torch.zeros(seq_len, d_model) # 创建维度为(seq_len)的向量,元素取值[0, seq_len-1] position = torch.arange(0, seq_len, dtype=torch.float).unsqueeze(1) # (seq_len, 1) # 创建维度为(d_model)的向量 div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) # (d_model / 2) # 偶数项应用sin pe[:, 0::2] = torch.sin(position * div_term) # sin(position * (10000 ** (2i / d_model)) # 奇数项应用cos pe[:, 1::2] = torch.cos(position * div_term) # cos(position * (10000 ** (2i / d_model))
# 为位置编码添加batch维度,以方便一批一批的使用句子 pe = pe.unsqueeze(0) # (1, seq_len, d_model)
# 当你有一个不是习得的模型参数,而是表示模型状态的tensor,如果想要存放在模型中,则需要注册一个buffer,这个buffer就会跟模型文件一起保存在模型中 self.register_buffer('pe', pe)
def forward(self, x): # requires_grad_(False)告诉模型训练过程中不需要学习这一部分位置编码,因为他们是固定的 x = x + (self.pe[:, :x.shape[1], :]).requires_grad_(False) # (batch, seq_len, d_model) return self.dropout(x)1.3 Normaliation

Normaliation(也译为归一化),是取一个batch中的每个item,计算他们在本batch中相对于其他item的期望和方差,再使用和计算这一项的新值。此外,引入两个会被学习的参数,乘数gamma(或叫做alpha)和加数beta(或叫做bias),以增加数值的差异,因为如果将所有值都限定在[0, 1]之间,数值的起伏可能会过小,因此有必要让模型自己学习调整这两个能够引入更大数值起伏的参数。另外,之所以需要这个参数,主要是为了数值稳定性,比如过小时将会导致整个过大。
class LayerNormalization(nn.Module):
# `eps`: 批归一化的参数epsilon def __init__(self, eps: float = 10**-6) -> None: super().__init__() self.eps = eps self.alpha = nn.Parameter(torch.ones(1)) # 乘数 self.bias = nn.Parameter(torch.zeros(1)) # 加数
# `x`: 大小为(batch, seq_len, hidden_size) def forward(self, x): # `keepdim=True`让计算保持维度属性 # `dim=-1`沿最后一个维度计算均值和方差 mean = x.mean(dim=-1, keepdim=True) # (batch, seq_len, 1) std = x.std(dim=-1, keepdim=True) # (batch, seq_len, 1)
# eps保持数值稳定性,防止除数过小 return self.alpha * (x - mean) / (std + self.eps) + self.bias1.4 Feed Forward
一个简单的全连接层,论文提了用到的参数和激活函数relu。文中提到可以将这一步计算看做两个核大小为1的卷积层,输入输出维度为d_model=512,内部隐藏层维度为d_ff=2048。
class FeedForwardBlock(nn.Module):
# 一个简单的全连接层,论文3.3节提到d_model、d_ff的大小 def __init__(self, d_model: int, d_ff: int, dropout: float) -> None: super().__init__()
# Linear层默认就带有bias,因此我们无需新增参数 self.linear_1 = nn.Linear(d_model, d_ff) # w1 and b1 self.dropout = nn.Dropout(dropout) self.linear_2 = nn.Linear(d_ff, d_model) # w2 and b2
def forward(self, x): # (batch, seq_len, d_model) -[linear_1]-relu-dropout-> (batch, seq_len, d_ff) -[linear_2]-> (batch, seq_len, d_model) return self.linear_2(self.dropout(torch.relu(self.linear_1(x))))1.5 Multi-Head Attention

Encoder的重头戏,就是这个多头注意力层,需要注意的是,encoder和decoder的多头注意力层有些细微的区别。
先抛开batch维度,只关注算法,看看如何在一个句子上应用多头注意力模型。但需要记住,应用算法的时候,第一个维度是batch,程序会成批的输入句子。这里:
- 为句子最大长度
- 为embedding向量的大小(本文中为512);
- 为注意力头数
- 为
从流程上来说:
- 先将输入复制三份,即Q、K、V矩阵(分别代表query、key、value,维度均为);
- 分别乘以、、矩阵(维度均为),得到、、(维度均为);
- 沿embedding方向将、、拆分为个小矩阵、、(维度均为),即每个小矩阵都能拿到完整的句子,但都只拿到了embedding向量的一小部分;
- 应用注意力模型公式,计算出每个头,其中的每个头维度均为;
- 将合并成矩阵,维度为,即变回,再乘以矩阵(维度为),最终得到矩阵(维度为),这一步的整体运算为;
class MultiHeadAttentionBlock(nn.Module):
# `d_model`: 模型维度,本文中为512 # `h`: 注意力头数 # `dropout`: 添加dropout层抑制过拟合 def __init__(self, d_model: int, h: int, dropout: float) -> None: super().__init__() self.d_model = d_model self.h = h # 多头需要均分d_model,应确保d_model可以被h整除 assert d_model % h == 0, "d_model不可被h整除"
self.d_k = d_model // h self.w_q = nn.Linear(d_model, d_model) # W^Q self.w_k = nn.Linear(d_model, d_model) # W^K self.w_v = nn.Linear(d_model, d_model) # W^V
self.w_o = nn.Linear(d_model, d_model) # W^O self.dropout = nn.Dropout(dropout)关于mask参数

在计算时,我们会先使用和计算(即在乘以之前)。得到的是一个的矩阵,其内容为各词与其他词的关系表。如果我们不希望某词与其他某词产生联系,就需要将矩阵中两词对应的位置写成负无穷,即使其注意力分数变得极小,此时在进行运算时,矩阵该位置的值就几乎为零,也就意味着屏蔽了这两个词的注意力——这就是mask参数的作用。
@staticmethod def attention(query, key, value, mask, dropout: nn.Dropout): d_k = query.shape[-1]
# 计算论文中$\mathrm{softmax}\(\frac{QK^T}{\sqrt{d_k}}\)V$的部分 # (batch, h, seq_len, d_k) -[transpose]-> (batch, h, d_k, seq_len)] -[matmul]-> (batch, h, seq_len, seq_len) attention_scores = (query @ key.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: # 在mask==0的位置写一个非常小的数来代表负无穷 attention_scores.masked_fill_(mask == 0, -1e9) attention_scores = attention_scores.softmax(dim = -1) # (batch, h, seq_len, seq_len)再计算softmax if dropout is not None: attention_scores = dropout(attention_scores) # 乘以`value`后变为(batch, h, seq_len, seq_len) --> (batch, h, seq_len, d_k) # 这里额外返回`attention_scores`是为了可视化注意力分数 return (attention_scores @ value), attention_scores
def forward(self, q, k, v, mask): query = self.w_q(q) # (batch, seq_len, d_model) --> (batch, seq_len, d_model) key = self.w_k(k) # (batch, seq_len, d_model) --> (batch, seq_len, d_model) value = self.w_v(v) # (batch, seq_len, d_model) --> (batch, seq_len, d_model)
# (batch, seq_len, d_model) -[view]-> (batch, seq_len, h, d_k) -[transpose]-> (batch, h, seq_len, d_k) query = query.view(query.shape[0], query.shape[1], self.h, self.d_k).transpose(1, 2) key = key.view(key.shape[0], key.shape[1], self.h, self.d_k).transpose(1, 2) value = value.view(value.shape[0], value.shape[1], self.h, self.d_k).transpose(1, 2)
x, self.attention_scores = MultiHeadAttentionBlock.attention(query, key, value, mask, self.dropout)
# (batch, h, seq_len, d_k) -[transpose]-> (batch, seq_len, h, d_k) -[contiguous+view]-> (batch, seq_len, d_model) # `contiguous`用于重新调整内存布局,因为诸如`transpose`、`view`等函数只会改变tensor的元数据以调整其表现形式, # 这些操作可能导致tensor显示的布局与其内存中的布局不一致, # 详情参见https://stackoverflow.com/questions/48915810/what-does-contiguous-do-in-pytorch x = x.transpose(1, 2).contiguous().view(x.shape[0], -1, self.h * self.d_k)
# 乘以W^O,(batch, seq_len, d_model) --> (batch, seq_len, d_model) return self.w_o(x)1.6 Residual Connection
在Transformer模型中可以看到不少Residual连接(也译为残差连接),比如左侧一开始的Input Embedding+Positional Encoding既输出到Multi-Head Attention又输出到Add & Norm,或是接下来Add & Norm输出既到Feed Forward又到下一个Add & Norm。可以看出,模型中的残差连接都是用来连接Add & Norm及其前一层网络的:
class ResidualConnection(nn.Module):
def __init__(self, dropout: float) -> None: super().__init__() self.dropout = nn.Dropout(dropout) self.norm = LayerNormalization()
def forward(self, x, sublayer): # 有些实现是先计算`norm`再计算`sublayer`,比如论文中的实现。 # 这两种计算计算的先后顺序有些细微差别,本文按照先`norm`再`sublayer`的模式执行。 return x + self.dropout(sublayer(self.norm(x)))1.7 Encoder Block
现在可以实现左侧整个encoder模块了,可以看出这个模块在模型中又N个,前一个encoder的输出送入后一个encoder做输入,最后一个encoder的输出送入decoder。Encoder中包含了一层Multi-Head Attention、一层Feed Forward、两层Add & Norm:
class EncoderBlock(nn.Module):
def __init__(self, self_attention_block: MultiHeadAttentionBlock, feed_forward_block: FeedForwardBlock, dropout: float) -> None: super().__init__() self.self_attention_block = self_attention_block self.feed_forward_block = feed_forward_block # 返回一个列表,包括两层`ResidualConnection` self.residual_connection = nn.ModuleList([ResidualConnection(dropout) for _ in range(2)])
# `src_mask`应用在encoder的输入上,用于屏蔽padding词对正常词的影响, def forward(self, x, src_mask): # 从这里可以看到,`self_attention_block`之所以叫做"self attention", # 是因为其中的query、key、value都是同一个输入句子`x`自身, # 即这一层输入的句子中的每个词,只会跟本句的词产生关系。 # 在后文中可以看到decoder会有些许区别,将用到"cross attention", # 即decoder中的query将会与encoder中的key和value产生关系。 x = self.residual_connection[0](x, lambda x: self.self_attention_block(x, x, x, src_mask)) x = self.residual_connection[1](x, self.feed_forward_block) return x1.8 Encoder Module
然后将多个encoder模块首尾相接的组装在一起,最后添加一层Normalization形成模型的encoder:
class Encoder(nn.Module):
def __init__(self, layers: nn.ModuleList) -> None: super().__init__() self.layers = layers self.norm = LayerNormalization()
def forward(self, x, mask): for layer in self.layers: x = layer(x, mask) return self.norm(x)2 Decoder

再来观察一下Decoder。对于输入,Decoder的Output Embedding与Encoder的Input Embedding一样,都是用到同一张词表,而且Positional Encoding也是一样的。不一样的是上面的Decoder Block,可以看到首先是Masked Multi-Head Attention和残差连接到Add & Norm,然后又是Multi-Head Attention残差连接到Add & Norm,最后是Feed Forward残差连接到Add & Norm。可以发现我们前文中定义的MultiHeadAttentionBlock已经考虑到mask参数,因此只需要使用已经定义好的模块,把他们首尾相接,即可快速组装出这个decoder。
2.1 Decoder Block
编写decoder模块时需要注意,最下层的Multi-Head Attention仍然是”self attention”,而中间的Multi-Head Attention则变为”cross attention”,我们将会用到两种输入并计算他们之间的关系。另外,之所以代码中有src_mask和tgt_mask,是因为本文展示的是一个翻译任务,源语言为英语,目标语言为意大利语,所以从算法角度来说,也可以叫做enc_mask和dec_mask,只需要记住encoder和decoder各自都有一个mask。
class DecoderBlock(nn.Module):
def __init__(self, self_attention_block: MultiHeadAttentionBlock, cross_attention_block: MultiHeadAttentionBlock, feed_forward_block: FeedForwardBlock, dropout: float) -> None: super().__init__() self.self_attention_block = self_attention_block self.cross_attention_block = cross_attention_block self.feed_forward_block = feed_forward_block self.residual_connection = nn.ModuleList([ResidualConnection(dropout) for _ in range(3)])
# `x`: decoder的输入; # `encoder_output`: encoder的输出; # `src_mask`: encoder的mask # `tgt_mask`: decoder的mask def forward(self, x, encoder_output, src_mask, tgt_mask): x = self.residual_connection[0](x, lambda x: self.self_attention_block(x, x, x, tgt_mask)) x = self.residual_connection[1](x, lambda x: self.cross_attention_block(x, encoder_output, encoder_output, src_mask)) x = self.residual_connection[2](x, self.feed_forward_block) return x2.2 Decoder Module
同encoder类似,decoder也是多个decoder模块首尾相连,最后添加一层Normalization的结构:
class Decoder(nn.Module):
def __init__(self, layers: nn.ModuleList) -> None: super().__init__() self.layers = layers self.norm = LayerNormalization()
def forward(self, x, encoder_output, src_mask, tgt_mask): for layer in self.layers: x = layer(x, encoder_output, src_mask, tgt_mask) return self.norm(x)3 Linear & Softmax
现在,Decoder的输出维度为(仍然暂时忽略batch维度),我们想要将这些Embedding映射回词表,于是需要添加一层Linear将这些Embedding映射到词表中各词所在的位置上(因此也可以叫做Projection Layer投影层)。
class ProjectionLayer(nn.Module):
def __init__(self, d_model: int, vocab_size: int) -> None: super().__init__() self.proj = nn.Linear(d_model, vocab_size)
def forward(self, x): # (batch, seq_len, d_model) --> (batch, seq_len, vocab_size) # 这里为了数值稳定性,采用log softmax return torch.log_softmax(self.proj(x), dim=-1)4 Transformer
现在我们已经完成了Transformer模型中所有模块的编写,是时候将他们组装成一个完整的Transformer了。本文的例子为翻译任务,因此Transformer初始化时需要源语言的Embedding和目标语言的Embedding。我们在用三个函数分别实现encoder、decoder、project,而不是使用一个forward函数实现,因为在推理过程中,我们可以重用encoder的输出,而且将这部分输出另存一份有利于可视化理解注意力模型的计算过程。
class Transformer(nn.Module):
def __init__(self, encoder: Encoder, decoder: Decoder, src_embed: InputEmbeddings, tgt_embed: InputEmbeddings, src_pos: PositionalEncoding, tgt_pos: PositionalEncoding, projcetion_layer: ProjectionLayer) -> None: super().__init__() self.encoder = encoder self.decoder = decoder self.src_embed = src_embed self.tgt_embed = tgt_embed self.src_pos = src_pos self.tgt_pos = tgt_pos self.projcetion_layer = projcetion_layer
def encode(self, src, src_mask): src = self.src_embed(src) src = self.src_pos(src) return self.encoder(src, src_mask)
def decode(self, encoder_output, src_mask, tgt, tgt_mask): tgt = self.tgt_embed(tgt) tgt = self.tgt_pos(tgt) return self.decoder(tgt, encoder_output, src_mask, tgt_mask)
def project(self, x): return self.projcetion_layer(x)至此我们已经完成了所有模块的编写,还需要一个函数将这一大堆embedding、encoder、decoder模块赋超参数并完成初始化。
def build_transformer(src_vocab_size: int, tgt_vocab_size: int, src_seq_len: int, tgt_seq_len: int, d_model: int = 512, N: int = 6, h: int = 8, dropout: float = 0.1, d_ff = 2048) -> Transformer: """构造Transformer,创建超参数,初始化参数。
参数: `src_vocab_size`: 源语言的词表大小; `tgt_vocab_size`: 目标语言的词表大小; `src_seq_len`: 源语言语言句子最大长度; `tgt_seq_len`: 目标语言句子最大长度; (本文的范例为英语到意大利语的翻译,源和目标语言句子长度设置为相同大小。 但对于其他语言,可能存在源和目标语言相差非常大的情况,届时将这两个值设置为不同大小) `d_model`: 模型大小,与论文一致预设为`512`; `N`: encoder/decoder的数量,与论文一致预设为`6`; `h`: 注意力头数,与论文一致预设为`8`; `dropout`: 与论文一致预设为`0.1`; `d_ff`: 前馈网络大小,与论文一致预设为`2048`。 返回: 一个完成了参数初始化的`Transformer`。 """
# 初始化Embedding层 src_embed = InputEmbeddings(d_model, src_vocab_size) tgt_embed = InputEmbeddings(d_model, tgt_vocab_size)
# 初始化Positional Encoding层,实际上位置信息的编码只需要一组参数计算一次即可,本文使用相同的参数计算两层是为了更好的解释模型 src_pos = PositionalEncoding(d_model, src_seq_len, dropout) tgt_pos = PositionalEncoding(d_model, tgt_seq_len, dropout)
# 初始化N个Encoder encoder_blocks = [] for _ in range(N): encoder_self_attention_block = MultiHeadAttentionBlock(d_model, h, dropout) feed_forward_block = FeedForwardBlock(d_model, d_ff, dropout) encoder_block = EncoderBlock(encoder_self_attention_block, feed_forward_block, dropout) encoder_blocks.append(encoder_block)
# 初始化N个Decoder decoder_blocks = [] for _ in range(N): decoder_self_attention_block = MultiHeadAttentionBlock(d_model, h, dropout) decoder_cross_attention_block = MultiHeadAttentionBlock(d_model, h, dropout) feed_forward_block = FeedForwardBlock(d_model, d_ff, dropout) decoder_block = DecoderBlock(decoder_self_attention_block, decoder_cross_attention_block, feed_forward_block, dropout) decoder_blocks.append(decoder_block)
# 构造Encoder和Decoder encoder = Encoder(nn.ModuleList(encoder_blocks)) decoder = Decoder(nn.ModuleList(decoder_blocks))
# 初始化投影层 projection_layer = ProjectionLayer(d_model, tgt_vocab_size)
# 构造Transformer transformer = Transformer(encoder, decoder, src_embed, tgt_embed, src_pos, tgt_pos, projection_layer)
# 初始化超参数 for p in transformer.parameters(): if p.dim() > 1: nn.init.xavier_uniform_(p)
return transformer最终,我们完成了一个Transformer的编写,这个过程详细展示了Transformer的内部结构,包括各模块的总体架构、每层的输入输出、各层计算过程及维度、模块间的关系等信息。接下来将要展示如何使用Transformer,包括搜集训练集数据,以及编写训练循环、推理过程、可视化等部分。