與基于RNN的方法相比，Transformer 不需要循環，主要是由Attention 機制組成，因而可以充分利用python的高效線性代數函數庫，大量節省訓練時間。

可是，卻經常聽到有人抱怨，Transformer學過就忘，總是不得要領。

怎么辦？那就自己搭一個Transformer吧！

上圖是谷歌提出的transformer 架構，其本質上是一個Encoder-Decoder的結構。把英文句子輸入模型，模型會輸出法文句子。

要搭建Transformer，我們必須要了解5個過程:

詞向量層

位置編碼

創建Masks

多頭注意層（The Multi-Head Attention layer）

Feed Forward層

詞向量

詞向量是神經網絡機器翻譯（NMT）的標準訓練方法，能夠表達豐富的詞義信息。

在pytorch里很容易實現詞向量：

class Embedder(nn.Module): def __init__(self, vocab_size, d_model): super().__init__() self.embed = nn.Embedding(vocab_size, d_model) def forward(self, x): return self.embed(x)

當每個單詞進入后，代碼就會查詢和檢索詞向量。模型會把這些向量當作參數進行學習，并隨著梯度下降的每次迭代而調整。

給單詞賦予上下文語境：位置編程

模型理解一個句子有兩個要素：一是單詞的含義，二是單詞在句中所處的位置。

每個單詞的嵌入向量會學習單詞的含義，所以我們需要輸入一些信息，讓神經網絡知道單詞在句中所處的位置。

Vasmari用下面的函數創建位置特異性常量來解決這類問題：

這個常量是一個2D矩陣。Pos代表了句子的順序，i代表了嵌入向量所處的維度位置。在pos/i矩陣中的每一個值都可以通過上面的算式計算出來。

位置編碼矩陣是一個常量，它的值可以用上面的算式計算出來。把常量嵌入矩陣，然后每個嵌入的單詞會根據它所處的位置發生特定轉變。

位置編輯器的代碼如下所示：

class PositionalEncoder(nn.Module): def __init__(self, d_model, max_seq_len = 80): super().__init__() self.d_model = d_model # create constant 'pe' matrix with values dependant on # pos and i pe = torch.zeros(max_seq_len, d_model) for pos in range(max_seq_len): for i in range(0, d_model, 2): pe[pos, i] = \ math.sin(pos / (10000 ** ((2 * i)/d_model))) pe[pos, i + 1] = \ math.cos(pos / (10000 ** ((2 * (i + 1))/d_model))) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) def forward(self, x): # make embeddings relatively larger x = x * math.sqrt(self.d_model) #add constant to embedding seq_len = x.size(1) x = x + Variable(self.pe[:,:seq_len], \ requires_grad=False).cuda() return x

以上模塊允許我們向嵌入向量添加位置編碼（positional encoding），為模型架構提供信息。

在給詞向量添加位置編碼之前，我們要擴大詞向量的數值，目的是讓位置編碼相對較小。這意味著向詞向量添加位置編碼時，詞向量的原始含義不會丟失。

創建Masks

Masks在transformer模型中起重要作用，主要包括兩個方面：

在編碼器和解碼器中：當輸入為padding，注意力會是0。

在解碼器中：預測下一個單詞，避免解碼器偷偷看到后面的翻譯內容。

輸入端生成一個mask很簡單：

batch = next(iter(train_iter))input_seq = batch.English.transpose(0,1)input_pad = EN_TEXT.vocab.stoi['']# creates mask with 0s wherever there is padding in the inputinput_msk = (input_seq != input_pad).unsqueeze(1)

同樣的，Target_seq也可以生成一個mask，但是會額外增加一個步驟：

# create mask as beforetarget_seq = batch.French.transpose(0,1)target_pad = FR_TEXT.vocab.stoi['']target_msk = (target_seq != target_pad).unsqueeze(1)size = target_seq.size(1) # get seq_len for matrixnopeak_mask = np.triu(np.ones(1, size, size),k=1).astype('uint8')nopeak_mask = Variable(torch.from_numpy(nopeak_mask) == 0)target_msk = target_msk & nopeak_mask

目標語句（法語翻譯內容）作為初始值輸進解碼器中。解碼器通過編碼器的全部輸出，以及目前已翻譯的單詞來預測下一個單詞。

因此，我們需要防止解碼器偷看到還沒預測的單詞。為了達成這個目的，我們用到了nopeak_mask函數：

當在注意力函數中應用mask，每一次預測都只會用到這個詞之前的句子。

多頭注意力

一旦我們有了詞向量(帶有位置編碼)和masks，我們就可以開始構建模型層了。

下圖是多頭注意力的結構：

多頭注意力層，每一個輸入都會分成多頭（multiple heads），從而讓網絡同時“注意”每一個詞向量的不同部分。

V,K和Q分別代表“key”、“value”和“query”，這些是注意力函數的相關術語，但我不覺得解釋這些術語會對理解這個模型有任何幫助。

在編碼器中，V、K和G將作為詞向量（加上位置編碼）的相同拷貝。它們具有維度Batch_size * seq_len * d_model.

在多頭注意力中，我們把嵌入向量分進N個頭中，它們就有了維度（batch_size * N * seq_len * (d_model / N).

我們定義最終維度 (d_model / N )為d_k。

讓我們來看看解碼器模塊的代碼：

class MultiHeadAttention(nn.Module): def __init__(self, heads, d_model, dropout = 0.1): super().__init__() self.d_model = d_model self.d_k = d_model // heads self.h = heads self.q_linear = nn.Linear(d_model, d_model) self.v_linear = nn.Linear(d_model, d_model) self.k_linear = nn.Linear(d_model, d_model) self.dropout = nn.Dropout(dropout) self.out = nn.Linear(d_model, d_model)

def forward(self, q, k, v, mask=None): bs = q.size(0) # perform linear operation and split into h heads k = self.k_linear(k).view(bs, -1, self.h, self.d_k) q = self.q_linear(q).view(bs, -1, self.h, self.d_k) v = self.v_linear(v).view(bs, -1, self.h, self.d_k) # transpose to get dimensions bs * h * sl * d_model k = k.transpose(1,2) q = q.transpose(1,2) v = v.transpose(1,2)# calculate attention using function we will define next scores = attention(q, k, v, self.d_k, mask, self.dropout) # concatenate heads and put through final linear layer concat = scores.transpose(1,2).contiguous()\ .view(bs, -1, self.d_model) output = self.out(concat) return output

計算注意力

計算注意力的公式

圖解公式

這是另一個我們需要了解的公式，上面這幅圖很好地解釋了這個公式。

圖中的每個箭頭代表了公式的一部分。

首先，我們要用Q乘以K的轉置函數（transpose），然后通過除以d_k的平方根來實現scaled函數。

方程中沒有顯示的一個步驟是masking。在執行Softmax之前，我們使用mask，減少輸入填充（padding）的值。

另一個未顯示的步驟是dropout，我們將在Softmax之后使用它。

最后一步是在目前為止的結果和V之間做點積（dot product）。

下面是注意力函數的代碼：

def attention(q, k, v, d_k, mask=None, dropout=None): scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(d_k)if mask is not None: mask = mask.unsqueeze(1) scores = scores.masked_fill(mask == 0, -1e9)scores = F.softmax(scores, dim=-1) if dropout is not None: scores = dropout(scores) output = torch.matmul(scores, v) return output

前饋網絡

好了，如果你現在已經理解以上部分，我們就進入最后一步！

這一層由兩個線性運算組成，兩層中夾有relu和dropout 運算。

class FeedForward(nn.Module): def __init__(self, d_model, d_ff=2048, dropout = 0.1): super().__init__() # We set d_ff as a default to 2048 self.linear_1 = nn.Linear(d_model, d_ff) self.dropout = nn.Dropout(dropout) self.linear_2 = nn.Linear(d_ff, d_model) def forward(self, x): x = self.dropout(F.relu(self.linear_1(x))) x = self.linear_2(x) return x

最后一件事：歸一化

在深度神經網絡中，歸一化是非常重要的。它可以防止層中值變化太多，這意味著模型訓練速度更快，具有更好的泛化。

我們在編碼器/解碼器的每一層之間歸一化我們的結果，所以在構建我們的模型之前，讓我們先定義這個函數:

class Norm(nn.Module): def __init__(self, d_model, eps = 1e-6): super().__init__() self.size = d_model # create two learnable parameters to calibrate normalisation self.alpha = nn.Parameter(torch.ones(self.size)) self.bias = nn.Parameter(torch.zeros(self.size)) self.eps = eps def forward(self, x): norm = self.alpha * (x - x.mean(dim=-1, keepdim=True)) \ / (x.std(dim=-1, keepdim=True) + self.eps) + self.bias return norm

把所有內容結合起來！

如果你已經清楚了上述相關細節，那么你就能理解Transformer模型啦。剩下的就是把一切都組裝起來。

讓我們再來看看整體架構，然后開始構建:

最后一個變量:如果你仔細看圖，你可以看到編碼器和解碼器旁邊有一個“Nx”。實際上，上圖中的編碼器和解碼器分別表示編碼器的一層和解碼器的一層。N是層數的變量。比如，如果N=6，數據經過6個編碼器層(如上所示的結構)，然后將這些輸出傳給解碼器，解碼器也由6個重復的解碼器層組成。

現在，我們將使用上面模型中所示的結構構建編碼器層和解碼器層模塊。在我們構建編碼器和解碼器時，我們可以決定層的數量。

# build an encoder layer with one multi-head attention layer and one # feed-forward layerclass EncoderLayer(nn.Module): def __init__(self, d_model, heads, dropout = 0.1): super().__init__() self.norm_1 = Norm(d_model) self.norm_2 = Norm(d_model) self.attn = MultiHeadAttention(heads, d_model) self.ff = FeedForward(d_model) self.dropout_1 = nn.Dropout(dropout) self.dropout_2 = nn.Dropout(dropout) def forward(self, x, mask): x2 = self.norm_1(x) x = x + self.dropout_1(self.attn(x2,x2,x2,mask)) x2 = self.norm_2(x) x = x + self.dropout_2(self.ff(x2)) return x # build a decoder layer with two multi-head attention layers and# one feed-forward layerclass DecoderLayer(nn.Module): def __init__(self, d_model, heads, dropout=0.1): super().__init__() self.norm_1 = Norm(d_model) self.norm_2 = Norm(d_model) self.norm_3 = Norm(d_model) self.dropout_1 = nn.Dropout(dropout) self.dropout_2 = nn.Dropout(dropout) self.dropout_3 = nn.Dropout(dropout) self.attn_1 = MultiHeadAttention(heads, d_model) self.attn_2 = MultiHeadAttention(heads, d_model) self.ff = FeedForward(d_model).cuda()def forward(self, x, e_outputs, src_mask, trg_mask): x2 = self.norm_1(x) x = x + self.dropout_1(self.attn_1(x2, x2, x2, trg_mask)) x2 = self.norm_2(x) x = x + self.dropout_2(self.attn_2(x2, e_outputs, e_outputs, src_mask)) x2 = self.norm_3(x) x = x + self.dropout_3(self.ff(x2)) return x# We can then build a convenient cloning function that can generate multiple layers:def get_clones(module, N): return nn.ModuleList([copy.deepcopy(module) for i in range(N)])

我們現在可以構建編碼器和解碼器了：

class Encoder(nn.Module): def __init__(self, vocab_size, d_model, N, heads): super().__init__() self.N = N self.embed = Embedder(vocab_size, d_model) self.pe = PositionalEncoder(d_model) self.layers = get_clones(EncoderLayer(d_model, heads), N) self.norm = Norm(d_model) def forward(self, src, mask): x = self.embed(src) x = self.pe(x) for i in range(N): x = self.layers[i](x, mask) return self.norm(x) class Decoder(nn.Module): def __init__(self, vocab_size, d_model, N, heads): super().__init__() self.N = N self.embed = Embedder(vocab_size, d_model) self.pe = PositionalEncoder(d_model) self.layers = get_clones(DecoderLayer(d_model, heads), N) self.norm = Norm(d_model) def forward(self, trg, e_outputs, src_mask, trg_mask): x = self.embed(trg) x = self.pe(x) for i in range(self.N): x = self.layers[i](x, e_outputs, src_mask, trg_mask) return self.norm(x)

Transformer模型構建完畢！

class Transformer(nn.Module): def __init__(self, src_vocab, trg_vocab, d_model, N, heads): super().__init__() self.encoder = Encoder(src_vocab, d_model, N, heads) self.decoder = Decoder(trg_vocab, d_model, N, heads) self.out = nn.Linear(d_model, trg_vocab) def forward(self, src, trg, src_mask, trg_mask): e_outputs = self.encoder(src, src_mask) d_output = self.decoder(trg, e_outputs, src_mask, trg_mask) output = self.out(d_output) return output# we don't perform softmax on the output as this will be handled# automatically by our loss function

訓練模型

構建完transformer，接下來要做的是用EuroParl數據集進行訓練。編碼部分非常簡單，但是要等兩天，模型才會開始converge!

讓我們先來定義一些參數：

d_model = 512heads = 8N = 6src_vocab = len(EN_TEXT.vocab)trg_vocab = len(FR_TEXT.vocab)model = Transformer(src_vocab, trg_vocab, d_model, N, heads)for p in model.parameters(): if p.dim() > 1: nn.init.xavier_uniform_(p)# this code is very important! It initialises the parameters with a# range of values that stops the signal fading or getting too big.# See this blog for a mathematical explanation.optim = torch.optim.Adam(model.parameters(), lr=0.0001, betas=(0.9, 0.98), eps=1e-9)

現在，我們可以開始訓練了：

def train_model(epochs, print_every=100): model.train() start = time.time() temp = start total_loss = 0 for epoch in range(epochs): for i, batch in enumerate(train_iter): src = batch.English.transpose(0,1) trg = batch.French.transpose(0,1) # the French sentence we input has all words except # the last, as it is using each word to predict the next trg_input = trg[:, :-1] # the words we are trying to predict targets = trg[:, 1:].contiguous().view(-1) # create function to make masks using mask code above src_mask, trg_mask = create_masks(src, trg_input) preds = model(src, trg_input, src_mask, trg_mask) optim.zero_grad() loss = F.cross_entropy(preds.view(-1, preds.size(-1)), results, ignore_index=target_pad) loss.backward() optim.step() total_loss += loss.data[0] if (i + 1) % print_every == 0: loss_avg = total_loss / print_every print("time = %dm, epoch %d, iter = %d, loss = %.3f, %ds per %d iters" % ((time.time() - start) // 60, epoch + 1, i + 1, loss_avg, time.time() - temp, print_every)) total_loss = 0 temp = time.time()

示例訓練輸出：經過幾天的訓練后，模型的損失函數收斂到了大約1.3。

測試模型

我們可以使用下面的函數來翻譯句子。我們可以直接輸入句子，或者輸入自定義字符串。

翻譯器通過運行一個循環來工作。我們對英語句子進行編碼。把 token輸進解碼器，編碼器輸出。然后，解碼器對第一個單詞進行預測，使用 token將其加進解碼器的輸入。接著，重新運行循環，獲取下一個單詞預測，將其加入解碼器的輸入，直到 token完成翻譯。

def translate(model, src, max_len = 80, custom_string=False): model.eval()if custom_sentence == True: src = tokenize_en(src) sentence=\ Variable(torch.LongTensor([[EN_TEXT.vocab.stoi[tok] for tok in sentence]])).cuda()src_mask = (src != input_pad).unsqueeze(-2) e_outputs = model.encoder(src, src_mask) outputs = torch.zeros(max_len).type_as(src.data) outputs[0] = torch.LongTensor([FR_TEXT.vocab.stoi['']])for i in range(1, max_len): trg_mask = np.triu(np.ones((1, i, i), k=1).astype('uint8') trg_mask= Variable(torch.from_numpy(trg_mask) == 0).cuda() out = model.out(model.decoder(outputs[:i].unsqueeze(0), e_outputs, src_mask, trg_mask)) out = F.softmax(out, dim=-1) val, ix = out[:, -1].data.topk(1) outputs[i] = ix[0][0] if ix[0][0] == FR_TEXT.vocab.stoi['']: breakreturn ' '.join( [FR_TEXT.vocab.itos[ix] for ix in outputs[:i]] )

Transformer模型的構建過程大致就是這樣。