目錄
1-大語言模型—理論基礎:詳解Transformer架構的實現(1)-CSDN博客https://blog.csdn.net/wh1236666/article/details/149443139?spm=1001.2014.3001.5502
2.3、殘差連接和層歸一化
2.3.1、什么是層歸一化?
2.3.2、層歸一化的核心特點(與其他歸一化對比)
2.3.3、特此說明
?2.4、編碼器和解碼器結構
2.4.1、 編碼器和解碼器到底是什么?
2.4.1.1、編碼器:負責 “看懂原文” 的翻譯官
2.4.1.2、解碼器:負責 “寫出譯文” 的秘書
2.4.1.3、總結:倆模塊的核心作用
2.4.2、編碼器(Encoder)
2.4.2.1、?整體結構
2.4.2.2、運算流程
2.4.2.3、核心機制:自注意力(Self-Attention)
2.4.3、解碼器(Decoder)
2.4.3.1、?整體結構
2.4.3.2、?運算流程
2.4.3.3、?核心機制:掩碼與交叉注意力
2.4.4、編碼器與解碼器的協作
2.4.5、完整代碼
2.5、Transformer整體邏輯
2.5.1、先看 “團隊架構”:編碼器與解碼器的核心組件
2.5.2、編碼器:用 “工具包” 把原文 “嚼碎成濃縮信息”
2.5.2.1、?多頭自注意力:像 “讀句子時同時抓多維度關系”
2.5.2.2、 前饋網絡:像 “基于關系提煉深層含義”
3. 殘差連接 + 層歸一化:像 “保持思路清晰,不混亂”
2.5.3、解碼器:用 “工具包” 把 “筆記” 變成 “通順譯文”
2.5.3.1、 掩碼多頭自注意力:像 “寫句子時只看自己已經寫的內容”
2.5.3.2、 編碼器 - 解碼器注意力:像 “寫譯文時回頭看原文筆記”
2.5.3.3、前饋網絡 + 殘差連接 + 層歸一化:和編碼器的作用一致
2.5.4、編碼器與解碼器的 “協作全流程”(以翻譯為例)
2.5.5、總結:為什么這套組合能 “超越傳統模型”?
2.6、完整代碼
2.6.1、Transforemers實現代碼
2.6.2、與LSTM對比實現代碼
2.7、實驗效果
2.7.1、Transforemers實驗效果
2.7.2、與LSTM對比實驗效果
前文:
1-大語言模型—理論基礎:詳解Transformer架構的實現(1)-CSDN博客
https://blog.csdn.net/wh1236666/article/details/149443139?spm=1001.2014.3001.5502
2.3、殘差連接和層歸一化
2.3.1、什么是層歸一化?
層歸一化的核心思想是:對單個樣本在某一層的所有特征(或隱藏單元)進行歸一化,讓這些特征的分布保持穩定(均值接近 0,方差接近 1),再通過可學習的參數進行縮放和平移,保留數據的原始特征信息。
具體計算步驟:
假設某一層的輸入為向量(d?為特征維度),層歸一化的計算過程如下:
- 計算均值:計算該向量所有元素的均值?
- 計算方差:計算該向量所有元素的方差?
- 歸一化:用均值和方差對原始數據進行標準化,消除量綱差異
?是一個極小值,避免分母為 0)
- 縮放和平移:通過可學習的參數?
(縮放因子)和?\(\beta\)(偏移因子)調整歸一化后的數據,保留原始特征的表達能力
2.3.2、層歸一化的核心特點(與其他歸一化對比)
為了更好理解層歸一化,我們可以與常用的批歸一化(Batch Normalization,BN)?對比:
特性 | 層歸一化(LN) | 批歸一化(BN) |
---|---|---|
歸一化維度 | 單個樣本的所有特征(特征維度) | 批次中所有樣本的同一特征(批次維度) |
依賴 “批次” 嗎? | 不依賴,單個樣本獨立計算 | 依賴,需基于整個批次的樣本計算 |
適用場景 | 序列模型(RNN、Transformer)、小批量數據 | 卷積神經網絡(CNN)、大批量數據 |
層歸一化是一種針對 “單個樣本特征” 的歸一化技術,其核心價值在于:不依賴批次、適配序列模型、穩定訓練并加速收斂。
2.3.3、特此說明
Transformer 模型中,層歸一化是核心組件之一,它被用于多頭注意力層和前饋網絡的輸入,確保了模型在處理長序列時的穩定性。
具體來說,在 Transformer 中,層歸一化的應用場景和作用可以更細致地拆解:
-
多頭注意力層的輸入與輸出:在多頭注意力機制計算前,會先對輸入的特征向量進行層歸一化,確保每個頭的注意力計算在穩定的數據分布上進行;而注意力層的輸出也會與輸入進行殘差連接后,再通過層歸一化處理,避免特征值因多次疊加而過大或分布失衡,保證后續前饋網絡能高效學習。
-
前饋網絡的輸入:經過注意力層和殘差連接、層歸一化后的數據,會作為前饋網絡的輸入。此時的層歸一化同樣起到 “校準” 作用,讓前饋網絡(由兩個線性層和激活函數組成)在處理特征時,無需適配波動劇烈的數據分布,從而更專注于學習特征間的非線性關系。
這種 “注意力層 + 層歸一化 + 殘差連接→前饋網絡 + 層歸一化 + 殘差連接” 的模塊化設計,是 Transformer 能處理超長序列(如長文本、長視頻幀)的重要保障。如果沒有層歸一化,隨著網絡深度增加(Transformer 通常有十幾到幾十層),特征分布會逐漸偏移甚至 “爆炸”,導致模型難以訓練或性能驟降。
2.3.4、完整代碼
"""
文件名: 2.1 transformer
作者: 墨塵
日期: 2025/7/18
項目名: LLM
備注:
"""import numpy as np
import math
import torch
from sympy.abc import q
from torch import nn
from d2l import torch as d2l
import matplotlib.pyplot as plt # 用于可視化注意力權重熱圖下·
import torch
import torch.nn as nn
import math
import torch.nn.functional as F# -------------------------- 2. 殘差連接 + 層規范化(AddNorm) --------------------------
# 作用:Transformer中每個子層(注意力/前饋網絡)的標配輸出處理,解決深層網絡訓練難題
# 核心邏輯:通過殘差連接保留原始信息,通過層規范化穩定特征分布,使模型可訓練數百層
class AddNorm(nn.Module):"""殘差連接后進行層規范化(Transformer子層輸出的標準處理)"""def __init__(self, normalized_shape, dropout, **kwargs):"""初始化參數參數詳解:normalized_shape: 層規范化的維度(通常為輸入特征的最后一維,如[seq_len, feature_dim])dropout: Dropout概率(隨機丟棄部分特征,防止過擬合)"""super(AddNorm, self).__init__(** kwargs)self.dropout = nn.Dropout(dropout) # Dropout層,僅作用于子層輸出(保護原始輸入)self.ln = nn.LayerNorm(normalized_shape) # 層規范化層(對每個樣本獨立歸一化,適合序列數據)def forward(self, X, Y):"""前向傳播:先殘差連接,再層規范化參數:X: 子層的原始輸入張量(形狀與Y必須一致,否則無法相加)Y: 子層(如注意力機制/前饋網絡)的輸出張量返回:經過處理的張量(形狀與X/Y一致,特征分布更穩定)"""# 步驟解析:# 1. 對Y應用Dropout:隨機丟棄部分特征,防止模型過度依賴子層輸出# 2. 殘差連接(X + dropout(Y)):保留原始輸入信息,緩解梯度消失(若Y無效,輸出≈X)# 3. 層規范化:對每個樣本計算均值和方差,將特征縮放到標準分布,加速訓練return self.ln(self.dropout(Y) + X)def main():# 設置參數batch_size = 2 # 批次大小seq_len = 5 # 序列長度feature_dim = 16 # 特征維度(與 normalized_shape 對應)dropout = 0.1 # Dropout概率# 初始化AddNorm層add_norm = AddNorm(normalized_shape=feature_dim, dropout=dropout)# 創建模擬輸入:X是子層原始輸入,Y是子層輸出X = torch.randn(batch_size, seq_len, feature_dim) # 原始輸入Y = torch.randn(batch_size, seq_len, feature_dim) # 子層(如注意力/前饋網絡)輸出# 應用AddNorm處理output = add_norm(X, Y)# 驗證形狀一致性print(f"原始輸入X形狀: {X.shape}")print(f"子層輸出Y形狀: {Y.shape}")print(f"AddNorm輸出形狀: {output.shape}") # 應與輸入形狀一致# 驗證殘差連接效果:輸出與輸入的差異應受Y影響# 計算X和output的相似度(應低于1.0,說明Y起作用)x_flat = X.flatten()output_flat = output.flatten()similarity = torch.cosine_similarity(x_flat.unsqueeze(0), output_flat.unsqueeze(0)).item()print(f"\nX與AddNorm輸出的余弦相似度: {similarity:.4f}") # 應顯著小于1.0# 驗證層規范化效果:輸出特征的均值應接近0,方差接近1mean = output.mean().item()var = output.var().item()print(f"AddNorm輸出的均值: {mean:.4f}") # 應接近0print(f"AddNorm輸出的方差: {var:.4f}") # 應接近1# 可視化處理前后的特征分布plt.figure(figsize=(10, 4))# 原始輸入X的特征分布plt.subplot(1, 2, 1)plt.hist(X.flatten().detach().numpy(), bins=20, alpha=0.7, label='原始輸入X')plt.axvline(X.mean().item(), color='r', linestyle='--', label=f'均值: {X.mean().item():.2f}')plt.title('原始輸入特征分布')plt.legend()# AddNorm輸出的特征分布plt.subplot(1, 2, 2)plt.hist(output.flatten().detach().numpy(), bins=20, alpha=0.7, label='AddNorm輸出')plt.axvline(output.mean().item(), color='r', linestyle='--', label=f'均值: {mean:.2f}')plt.title('AddNorm處理后的特征分布')plt.legend()plt.tight_layout()plt.show()if __name__ == "__main__":main()
通過直方圖直觀對比處理前后的特征分布,層規范化后的分布應更集中、波動更小。?
?2.4、編碼器和解碼器結構
2.4.1、 編碼器和解碼器到底是什么?
????????咱們可以把編碼器和解碼器想象成兩個人合作完成一項 “轉換任務”,比如把中文翻譯成英文,這樣就很好理解了:
2.4.1.1、編碼器:負責 “看懂原文” 的翻譯官
假設你要把一句中文 “我愛吃蘋果” 翻譯成英文,編碼器就像第一個翻譯官,他的工作是徹底理解這句話的意思。
- 他先看到每個字:“我”“愛”“吃”“蘋果”。
- 然后他會分析這些字的關系:“我” 是主語,“愛” 是謂語,“吃蘋果” 是賓語,整個句子說的是 “主語喜歡做‘吃蘋果’這件事”。
- 最后,他把這些信息整理成一份 “筆記”(專業上叫 “上下文向量”),里面不光有每個字的意思,還有它們之間的聯系(誰和誰相關,誰修飾誰)。
這份筆記會交給解碼器,相當于說:“我已經把原文吃透了,接下來看你的了!”
2.4.1.2、解碼器:負責 “寫出譯文” 的秘書
解碼器就像第二個角色,他的工作是根據編碼器的 “筆記”,一句一句寫出正確的英文。
- 他一開始不知道要寫什么,先從一個 “開始信號”(比如
<START>
)入手。 - 看到 “開始信號”,再對照編碼器的筆記(知道原文是 “我愛吃蘋果”),先寫出第一個詞 “I”。
- 寫完 “I” 之后,他會回頭看看自己剛寫的 “I”,再對照筆記,接著寫出 “like”(因為原文是 “愛”)。
- 然后再根據已經寫的 “I like” 和筆記,寫出 “eating”(對應 “吃”)。
- 最后寫出 “apples”(對應 “蘋果”),直到寫出 “結束信號”(比如
<END>
),整個翻譯就完成了。
這里有個關鍵點:解碼器寫的時候不能 “作弊”,比如寫 “I” 的時候,不能提前偷看后面要寫的 “like”,只能用自己已經寫過的內容,保證句子通順(這就是 “掩碼自注意力” 的作用)。
2.4.1.3、總結:倆模塊的核心作用
- 編碼器:把輸入的序列(句子、語音、圖像等)“嚼碎”,提取出所有關鍵信息和內部關系,變成一份 “濃縮的理解筆記”。
- 解碼器:拿著這份 “筆記”,從無到有地生成目標序列,并且保證生成的內容既符合原文意思,又符合目標語言的邏輯(比如語法、順序)。
就像兩個人合作:一個負責 “讀懂題意”,一個負責 “寫出答案”,缺一不可~
2.4.2、編碼器(Encoder)
編碼器負責處理輸入序列(如源語言句子),將其轉換為隱藏表示(特征向量),以便解碼器能夠理解并生成對應的輸出。
2.4.2.1、?整體結構
Transformer 的編碼器由 N 個相同的編碼層(Encoder Layer)?堆疊而成,每個編碼層包含兩個子層:
- 多頭自注意力層(Multi-Head Self-Attention):捕獲輸入序列內部的依賴關系(如句子中詞語之間的關聯)。
- 前饋神經網絡(Feed Forward Network):對注意力層的輸出進行非線性變換,增強模型表達能力。
每層之后還應用了殘差連接(Residual Connection)和層歸一化(Layer Normalization),以穩定訓練和防止梯度消失。
2.4.2.2、運算流程
以單個編碼層為例,其運算步驟如下:
輸入:X(上一層的輸出,初始為嵌入向量+位置編碼)1. 自注意力子層:- 對X進行線性變換,得到查詢(Q)、鍵(K)、值(V)三個矩陣- 計算注意力得分:Attention(Q, K, V) = softmax(Q·K?/√d?)·V- 多頭機制:將注意力計算分為多個“頭”并行處理,再拼接結果- 殘差連接:X? = X + MultiHead(Q, K, V)- 層歸一化:X? = LayerNorm(X?)2. 前饋網絡子層:- 線性變換+ReLU激活:FFN(X?) = max(0, X?·W? + b?)·W? + b?- 殘差連接:X? = X? + FFN(X?)- 層歸一化:X? = LayerNorm(X?)輸出:X?(作為下一層的輸入)
2.4.2.3、核心機制:自注意力(Self-Attention)
自注意力是編碼器的關鍵創新,允許模型關注輸入序列的不同部分來生成當前位置的表示。其核心公式為:
- Q, K, V?分別是查詢(Query)、鍵(Key)、值(Value)矩陣,通過輸入?X?線性變換得到。
是縮放因子,防止點積結果過大導致梯度消失。
- 多頭注意力將輸入分割為多個頭,并行計算注意力,捕獲不同子空間的信息。
2.4.3、解碼器(Decoder)
解碼器根據編碼器的輸出和已生成的部分輸出,逐步生成目標序列(如翻譯后的句子)。
2.4.3.1、?整體結構
Transformer 的解碼器同樣由N 個相同的解碼層(Decoder Layer)堆疊而成,但每個解碼層包含三個子層:
- 掩碼多頭自注意力層(Masked Multi-Head Self-Attention):與編碼器類似,但使用掩碼(Mask)防止看到未來位置的信息(確保生成時只依賴已生成的內容)。
- 編碼器 - 解碼器注意力層(Encoder-Decoder Attention):關注編碼器輸出的相關部分,建立輸入與輸出的關聯。
- 前饋神經網絡(Feed Forward Network):與編碼器相同,增強模型表達能力。
每層之后同樣應用殘差連接和層歸一化。
2.4.3.2、?運算流程
以單個解碼層為例,其運算步驟如下:
輸入:Y(上一層的輸出,初始為目標序列的嵌入向量+位置編碼)Encoder Output(編碼器的最終輸出)1. 掩碼自注意力子層:- 對Y進行線性變換,得到Q、K、V矩陣- 應用掩碼:在注意力得分計算中,將未來位置的得分設為負無窮(softmax后為0)- 計算注意力:Attention(Q, K, V) = softmax(Q·K?/√d?)·V- 殘差連接:Y? = Y + MaskedMultiHead(Y)- 層歸一化:Y? = LayerNorm(Y?)2. 編碼器-解碼器注意力子層:- 解碼器的Q來自Y?,K和V來自編碼器輸出- 計算注意力:Attention(Q, K, V) = softmax(Q·K?/√d?)·V- 殘差連接:Y? = Y? + MultiHead(Y?, Encoder Output, Encoder Output)- 層歸一化:Y? = LayerNorm(Y?)3. 前饋網絡子層:- 與編碼器相同:FFN(Y?) = max(0, Y?·W? + b?)·W? + b?- 殘差連接:Y? = Y? + FFN(Y?)- 層歸一化:Y? = LayerNorm(Y?)輸出:Y?(作為下一層的輸入)
2.4.3.3、?核心機制:掩碼與交叉注意力
- 掩碼(Mask):確保解碼器在生成第?t?個位置的輸出時,只關注?1?到?\(t-1\)?位置的輸入,避免信息泄露。
- 編碼器 - 解碼器注意力:解碼器通過查詢(Q)關注編碼器輸出的不同部分,建立源序列與目標序列的對齊關系(如機器翻譯中詞語的對應關系)。
2.4.4、編碼器與解碼器的協作
在完整的 Transformer 模型中,編碼器和解碼器的協作流程如下:
-
編碼階段:
- 輸入序列經過詞嵌入和位置編碼后,進入編碼器
- 編碼器逐層處理,生成最終的編碼表示(上下文向量)
-
解碼階段(自回歸生成):
- 解碼器從起始標記(如
<START>
)開始,每次生成一個詞 - 當前已生成的序列作為解碼器的輸入,結合編碼器輸出,預測下一個詞
- 重復此過程,直到生成結束標記(如
<END>
)或達到最大長度
- 解碼器從起始標記(如
2.4.5、完整代碼
????????????????????????后面一次給出包含實驗結果
2.5、Transformer整體邏輯
要理解 Transformer 中編碼器與解碼器的完整協作邏輯,我們可以用一個具體場景貫穿始終:把中文 “小明在公園給小紅送了一本他昨天買的書” 翻譯成英文。這個過程中,編碼器和解碼器就像兩個精密配合的 “翻譯團隊”,各自帶著一套 “工具包”(組件),分工協作完成從 “理解原文” 到 “生成譯文” 的全流程。
2.5.1、先看 “團隊架構”:編碼器與解碼器的核心組件
不管是編碼器還是解碼器,都遵循 “多層堆疊” 的設計(原論文中各堆了 6 層),每一層類似一個 “處理單元”。但因為兩者任務不同(編碼器 “理解輸入”,解碼器 “生成輸出”),“工具包” 略有差異:
模塊 | 編碼器每層包含 | 解碼器每層包含 | 核心目標 |
---|---|---|---|
注意力機制 | 多頭自注意力(Self-Attention) | 1. 掩碼多頭自注意力(Masked Self-Attention) 2. 編碼器 - 解碼器注意力(Encoder-Decoder Attention) | 捕捉 “關系”(輸入內部 / 生成序列內部 / 輸入與生成的關系) |
特征加工 | 前饋網絡(Feed-Forward Network) | 前饋網絡(Feed-Forward Network) | 深化單個位置的特征(從關系中提煉抽象含義) |
穩定機制 | 殘差連接(Add)+ 層歸一化(LayerNorm) | 殘差連接(Add)+ 層歸一化(LayerNorm) | 保證多層堆疊時訓練穩定,信息傳遞不 “跑偏” |
2.5.2、編碼器:用 “工具包” 把原文 “嚼碎成濃縮信息”
編碼器的任務是把輸入的中文句子 “嚼碎”,提煉出所有關鍵信息(誰、做了什么、關系如何),最終輸出一個 “濃縮的理解向量”(稱為 “編碼器輸出” 或 “上下文向量”)。它的 “工具包” 是這樣工作的:
2.5.2.1、?多頭自注意力:像 “讀句子時同時抓多維度關系”
面對 “小明在公園給小紅送了一本他昨天買的書”,編碼器需要同時理清:
- 主體與對象:“小明”→“小紅”(動作 “送” 的雙方);
- 動作與對象:“送”→“書”(送的是書);
- 指代關系:“他”→“小明”(避免混淆);
- 修飾關系:“他昨天買的”→“書”(書的來源)。
多頭自注意力就是干這個的:
- 每個 “頭” 是一個獨立的 “關系探測器”:有的頭專注抓 “誰對誰做了什么”,有的頭抓 “指代關系”,有的頭抓 “修飾關系”;
- 最后把所有頭的結果拼接起來,得到一個 “全方位的關系圖譜”—— 每個詞的表示都融入了和其他詞的關聯信息(比如 “書” 的表示里不僅有 “書” 本身,還有 “小明買的”“送給小紅” 這些信息)。
2.5.2.2、 前饋網絡:像 “基于關系提煉深層含義”
光有表面關系還不夠,需要進一步提煉抽象信息。比如:
- 從 “小明送小紅書”→ 隱含 “小明和小紅可能有關系”;
- 從 “昨天買的書”→ 隱含 “書是新的 / 特意準備的”。
前饋網絡就是做這個的:它是一個簡單的兩層神經網絡(線性變換 + ReLU 激活 + 線性變換),對每個詞的表示單獨 “深加工”—— 基于多頭注意力得到的關系,把具體的詞轉化為更抽象的 “語義特征”(類似人從具體事件中總結潛臺詞)。
3. 殘差連接 + 層歸一化:像 “保持思路清晰,不混亂”
編碼器是 6 層堆疊的(類似 “一層一層深入理解”),但多層處理容易出兩個問題:
- 信息 “越傳越歪”:比如第一層的輸出突然變大,第二層就很難處理(類似傳話游戲傳歪了);
- 深層 “學不動”:底層的參數因為梯度太小,學不到有效信息(類似推長鏈條,前端用力后端沒感覺)。
殘差連接(把每層的輸入直接加到輸出上)解決 “學不動” 問題 —— 讓信息和梯度能直接 “穿層而過”;
層歸一化(把輸出標準化,讓均值為 0、方差為 1)解決 “傳歪” 問題 —— 讓每層的輸入保持穩定范圍,方便下一層處理。
經過 6 層這樣的處理,編碼器最終輸出一個 “上下文向量”(本質是一串向量,每個位置對應輸入句的一個詞,但都融入了全局信息),相當于給解碼器遞了一份 “超詳細的原文理解筆記”。
2.5.3、解碼器:用 “工具包” 把 “筆記” 變成 “通順譯文”
解碼器的任務是拿著編碼器的 “筆記”,從無到有生成英文譯文(“Xiaoming gave Xiaohong a book he bought yesterday in the park”)。它的 “工具包” 更復雜 —— 因為它不僅要理解原文,還要保證生成的英文 “通順”(符合語法)、“對得上原文”(不跑偏)。
2.5.3.1、 掩碼多頭自注意力:像 “寫句子時只看自己已經寫的內容”
解碼器生成英文時,是 “逐詞推進” 的(先寫 “Xiaoming”,再寫 “gave”,再寫 “Xiaohong”……)。如果寫 “gave” 時偷看了后面的 “Xiaohong”,就可能寫出不符合語法的句子(比如先寫 “gave” 再補主語,這在英文里是錯的)。
掩碼多頭自注意力就是防止 “偷看” 的:
- 它和編碼器的 “多頭自注意力” 原理類似(抓詞之間的關系),但多了一個 “掩碼”(類似給未來的詞蓋了塊布)—— 計算當前詞和其他詞的關系時,只允許關注 “已經寫過的詞”(比如寫 “gave” 時,只能看 “Xiaoming”,不能看 “Xiaohong”“a book” 等還沒寫的詞)。
- 這樣生成的序列才能符合語言順序(比如英文必須 “主語→謂語→賓語”)。
2.5.3.2、 編碼器 - 解碼器注意力:像 “寫譯文時回頭看原文筆記”
生成英文時,必須保證每個詞都和原文對應(比如 “gave” 對應 “送”,“he” 對應 “他”)。
編碼器 - 解碼器注意力就是干這個的:
- 它讓解碼器 “盯著編碼器的筆記看”—— 計算解碼器當前生成的詞(比如 “gave”)與編碼器輸出的每個詞(比如 “小明”“送”“小紅”)的關聯程度;
- 比如生成 “he” 時,會重點關注編碼器中 “小明” 的位置(因為 “他” 指代 “小明”);生成 “book” 時,會重點關注 “書” 和 “買” 的位置。
2.5.3.3、前饋網絡 + 殘差連接 + 層歸一化:和編碼器的作用一致
- 前饋網絡:對解碼器生成的每個詞(比如 “gave”)做 “深加工”,提煉抽象含義(比如 “gave” 不僅是 “送”,還隱含 “過去式”“主動關系”);
- 殘差連接 + 層歸一化:保證解碼器的 6 層堆疊能穩定訓練,生成的序列 “層層優化”(從粗糙到精準)。
2.5.4、編碼器與解碼器的 “協作全流程”(以翻譯為例)
-
編碼器處理輸入:
中文句子→(嵌入層轉成初始向量)→ 經過 6 層編碼器(每層:多頭自注意力抓關系→前饋網絡深加工→殘差 + 歸一化穩定)→ 輸出 “上下文向量”(包含所有詞的關系和深層含義)。 -
解碼器生成輸出:
- 從 “開始信號”(
<START>
)出發,生成第一個詞(比如 “Xiaoming”); - 生成 “Xiaoming” 后,用掩碼自注意力關注 “Xiaoming”(確保只看已生成內容),用編碼器 - 解碼器注意力關注編碼器中 “小明” 的位置(保證對應),再經前饋網絡和歸一化優化;
- 重復上述步驟:生成 “gave” 時,關注已生成的 “Xiaoming” 和編碼器中 “送” 的位置;生成 “Xiaohong” 時,關注 “Xiaoming gave” 和編碼器中 “小紅” 的位置…… 直到生成 “結束信號”(
<END>
)。
- 從 “開始信號”(
-
最終結果:通過編碼器的 “透徹理解” 和解碼器的 “精準生成”,完成從中文到英文的轉換。
2.5.5、總結:為什么這套組合能 “超越傳統模型”?
Transformer 的編碼器 + 解碼器設計,本質是用 “注意力機制” 替代了 RNN 的 “序列依賴”(不用按順序處理,可并行計算),用 “多層堆疊 + 組件協作” 解決了 CNN 的 “局部視野局限”(能抓長距離關系)。
- 編碼器的組件讓它能 “吃透輸入”(全方位抓關系、挖含義、穩訓練);
- 解碼器的組件讓它能 “精準輸出”(不偷看未來、緊盯原文、保通順);
- 兩者協作,就像一個 “超級翻譯團隊”:一個把原文理解到骨子里,一個把理解轉化為完美譯文 —— 這也是 Transformer 能在翻譯、生成、問答等任務中表現頂尖的核心原因。
2.6、完整代碼
2.6.1、Transforemers實現代碼
# 導入必要的庫
import numpy as np # 用于數值計算和數組操作
import math # 用于數學運算(如平方根、對數)
import torch # PyTorch深度學習框架核心庫
from torch import nn # PyTorch神經網絡模塊
from d2l import torch as d2l # 深度學習工具庫(提供基礎組件和工具函數)
import matplotlib.pyplot as plt # 用于數據可視化(注意力權重熱圖等)
import torch.nn.functional as F # PyTorch函數式接口(如softmax、激活函數)class PositionalEncoding(nn.Module):"""位置編碼模塊:為序列注入位置信息(Transformer無循環結構,需顯式編碼位置)"""def __init__(self, d_model, max_seq_len=80):"""初始化位置編碼矩陣參數:d_model:模型特征維度(與詞嵌入維度一致)max_seq_len:最大序列長度(位置編碼的最大覆蓋范圍)"""super().__init__() # 繼承nn.Moduleself.d_model = d_model # 保存模型維度# 創建位置編碼矩陣:形狀為[max_seq_len, d_model],存儲每個位置的編碼pe = torch.zeros(max_seq_len, d_model)# 生成位置索引(0到max_seq_len-1),并增加維度為[max_seq_len, 1](便于廣播計算)position = torch.arange(0, max_seq_len, dtype=torch.float).unsqueeze(1)# 計算頻率除數(基于論文公式:div_term = 10000^(2i/d_model) 的倒數,用指數函數避免數值溢出)div_term = torch.exp(torch.arange(0, d_model, 2) # 生成偶數索引序列 [0,2,4,...,d_model-2](對應公式中的2i).float() # 轉為浮點數* (-math.log(10000.0) / d_model) # 等價于 1/10000^(2i/d_model))# 為偶數維度(0,2,4...)分配正弦編碼,奇數維度(1,3,5...)分配余弦編碼pe[:, 0::2] = torch.sin(position * div_term) # 0::2表示從0開始,步長為2的索引(偶數維度)pe[:, 1::2] = torch.cos(position * div_term) # 1::2表示從1開始,步長為2的索引(奇數維度)# 增加批次維度:形狀從[max_seq_len, d_model]變為[1, max_seq_len, d_model],適配批量輸入pe = pe.unsqueeze(0)# 將位置編碼注冊為緩沖區(非模型參數,不參與訓練,但會隨模型保存)self.register_buffer('pe', pe)def forward(self, x):"""將位置編碼添加到詞嵌入中參數:x:詞嵌入張量,形狀為[batch_size, seq_len, d_model]返回:注入位置信息的詞嵌入,形狀與x一致"""# 縮放詞嵌入:避免嵌入值與位置編碼值量級差異過大(穩定訓練)x = x * math.sqrt(self.d_model)# 獲取輸入序列的實際長度(每個樣本的詞元數量)seq_len = x.size(1) # x的形狀為[batch_size, seq_len, d_model],取第1維為序列長度# 將位置編碼中前seq_len個位置的編碼添加到詞嵌入中(截斷或補齊到實際序列長度)x = x + self.pe[:, :seq_len] # self.pe形狀為[1, max_seq_len, d_model],取前seq_len個位置return xclass MultiHeadAttention(nn.Module):"""多頭注意力機制模塊:將注意力拆分為多個并行子空間,捕獲多尺度特征"""def __init__(self, heads: int, d_model: int, dropout: float = 0.1):"""初始化多頭注意力參數:heads:注意力頭的數量(需滿足d_model能被heads整除)d_model:模型總維度(輸入/輸出特征維度)dropout:Dropout概率(防止過擬合)"""super().__init__()self.d_model = d_model # 模型總維度self.h = heads # 注意力頭數self.d_k = d_model // heads # 每個注意力頭的維度(d_model = heads * d_k)# 線性投影層:將輸入特征分別映射到Q(查詢)、K(鍵)、V(值)空間# 作用:區分Q、K、V的語義角色,為注意力計算做準備self.q_linear = nn.Linear(d_model, d_model) # Q的線性變換self.k_linear = nn.Linear(d_model, d_model) # K的線性變換self.v_linear = nn.Linear(d_model, d_model) # V的線性變換# 輸出投影層:將多頭注意力的結果合并后映射回d_model維度self.out = nn.Linear(d_model, d_model)# Dropout層:隨機丟棄部分注意力權重,防止過擬合self.dropout = nn.Dropout(dropout) # 訓練時以概率dropout丟棄元素,未丟棄元素縮放1/(1-dropout)# 縮放因子:用于縮放點積注意力的得分(避免得分過大導致softmax梯度消失)self.scale = math.sqrt(self.d_k) # 即1/sqrt(d_k)# 存儲注意力權重(用于后續可視化或分析)self.attention_weights = Nonedef create_mask(self, seq_len, valid_lens, device):"""創建注意力掩碼(用于屏蔽無效位置,如填充的PAD符號或未來信息)參數:seq_len:序列長度valid_lens:有效長度張量,形狀為[batch_size](每個樣本的有效長度)或[batch_size, seq_len](每個位置的有效性)device:設備(CPU/GPU)返回:掩碼張量,形狀為[batch_size, 1, seq_len, seq_len](適配多頭注意力的維度)"""if valid_lens is None: # 無掩碼時返回Nonereturn Nonebatch_size = valid_lens.size(0) # 批次大小if valid_lens.dim() == 1: # 情況1:每個樣本一個有效長度(如[5,4]表示第1個樣本有效長度5,第2個4)# 創建形狀為[batch_size, seq_len, seq_len]的掩碼:每行表示一個查詢位置的有效鍵位置mask = torch.arange(seq_len, device=device).expand(batch_size, seq_len, seq_len)# 將valid_lens擴展為[batch_size, 1, 1],便于廣播valid_lens = valid_lens.unsqueeze(1).unsqueeze(2)# 掩碼規則:位置索引 < 有效長度的位置為True(保留),否則為False(屏蔽)mask = mask < valid_lenselse: # 情況2:每個位置一個有效性標記(如[batch_size, seq_len]的0/1張量)# 擴展為[batch_size, seq_len, seq_len]:每個查詢位置共享相同的鍵位置有效性mask = valid_lens.unsqueeze(1).expand(batch_size, seq_len, seq_len)# 增加一個維度適配多頭注意力(形狀變為[batch_size, 1, seq_len, seq_len])return mask.unsqueeze(1)def attention(self, q: torch.Tensor, k: torch.Tensor, v: torch.Tensor,mask: torch.Tensor = None, dropout: nn.Dropout = None):"""計算單頭注意力(點積注意力)參數:q:查詢張量,形狀[batch_size, heads, seq_len_q, d_k]k:鍵張量,形狀[batch_size, heads, seq_len_k, d_k]v:值張量,形狀[batch_size, heads, seq_len_v, d_k](seq_len_k = seq_len_v)mask:掩碼張量,形狀[batch_size, 1, seq_len_q, seq_len_k]dropout:Dropout層返回:注意力輸出(加權聚合后的值),形狀[batch_size, heads, seq_len_q, d_k]注意力權重,形狀[batch_size, heads, seq_len_q, seq_len_k]"""# 計算注意力得分(Q與K的點積):形狀[batch_size, heads, seq_len_q, seq_len_k]# k.transpose(-2, -1):交換k的最后兩維,形狀變為[batch_size, heads, d_k, seq_len_k]# 點積后除以縮放因子self.scale(即1/sqrt(d_k)),防止得分過大導致softmax梯度消失scores = (torch.matmul(q, k.transpose(-2, -1)) / self.scale)# 應用掩碼:將無效位置的得分設為-1e9(softmax后接近0,即不關注)if mask is not None:scores = scores.masked_fill(mask == 0, -1e9) # mask==0的位置被屏蔽# 計算注意力權重:對得分做softmax(按最后一維歸一化,每行和為1)attn_weights = F.softmax(scores, dim=-1) # 形狀[batch_size, heads, seq_len_q, seq_len_k]# 應用Dropout(訓練時隨機丟棄部分權重)if dropout is not None:attn_weights = dropout(attn_weights)# 加權聚合值張量v:注意力權重 × v,得到最終注意力輸出output = torch.matmul(attn_weights, v) # 形狀[batch_size, heads, seq_len_q, d_k]return output, attn_weightsdef forward(self, q: torch.Tensor, k: torch.Tensor, v: torch.Tensor,mask: torch.Tensor = None):"""多頭注意力的前向傳播(核心邏輯)參數:q:查詢張量,形狀[batch_size, seq_len_q, d_model]k:鍵張量,形狀[batch_size, seq_len_k, d_model]v:值張量,形狀[batch_size, seq_len_v, d_model](seq_len_k = seq_len_v)mask:掩碼張量,可選,形狀[batch_size, 1, seq_len]或[batch_size, seq_len, seq_len]返回:多頭注意力輸出,形狀[batch_size, seq_len_q, d_model]注意力權重,形狀[batch_size, heads, seq_len_q, seq_len_k]"""batch_size = q.size(0) # 批次大小# 1. 線性投影并重塑為多頭結構# 將輸入特征通過線性層映射到d_model維度,再拆分為heads個注意力頭# 形狀變化:[batch_size, seq_len, d_model] → [batch_size, seq_len, heads, d_k]k = self.k_linear(k).view(batch_size, -1, self.h, self.d_k) # k的處理q = self.q_linear(q).view(batch_size, -1, self.h, self.d_k) # q的處理v = self.v_linear(v).view(batch_size, -1, self.h, self.d_k) # v的處理# 2. 調整維度順序:將heads維度提前,便于并行計算多頭注意力# 形狀變化:[batch_size, seq_len, heads, d_k] → [batch_size, heads, seq_len, d_k]k = k.transpose(1, 2) # 交換seq_len和heads維度q = q.transpose(1, 2)v = v.transpose(1, 2)# 3. 處理掩碼:將輸入掩碼轉換為適配多頭注意力的形狀[batch_size, 1, seq_len_q, seq_len_k]if mask is not None:# 若掩碼維度≤2(如[batch_size, seq_len]),調用create_mask生成標準掩碼if mask.dim() <= 2:mask = self.create_mask(q.size(2), mask, q.device) # q.size(2)是seq_len_q# 4. 計算多頭注意力:調用attention函數,得到每個頭的輸出和權重output, attn_weights = self.attention(q, k, v, mask, self.dropout)self.attention_weights = attn_weights # 保存注意力權重# 5. 重塑并合并多頭結果# 交換維度:[batch_size, heads, seq_len_q, d_k] → [batch_size, seq_len_q, heads, d_k]output = output.transpose(1, 2).contiguous() # contiguous()確保內存連續,便于后續view操作# 合并多頭:將heads和d_k維度合并為d_model(heads×d_k = d_model)# 形狀變化:[batch_size, seq_len_q, heads, d_k] → [batch_size, seq_len_q, d_model]output = output.view(batch_size, -1, self.d_model) # -1表示自動計算seq_len_q# 6. 最終線性投影:將合并后的結果映射回d_model維度(進一步調整特征)output = self.out(output)return output, attn_weightsclass PositionWiseFFN(nn.Module):"""基于位置的前饋網絡(Transformer子層):對序列中每個位置的特征獨立做非線性變換"""def __init__(self, ffn_num_input, ffn_num_hiddens, ffn_num_outputs, **kwargs):"""初始化前饋網絡參數:ffn_num_input:輸入特征維度(需與注意力機制輸出維度一致,即d_model)ffn_num_hiddens:隱藏層維度(通常大于輸入維度,形成"升維-降維"結構)ffn_num_outputs:輸出特征維度(需與輸入維度一致,才能參與殘差連接)"""super(PositionWiseFFN, self).__init__(** kwargs)# 第一層線性變換(升維):將輸入從ffn_num_input映射到ffn_num_hiddensself.dense1 = nn.Linear(ffn_num_input, ffn_num_hiddens)# 非線性激活函數:引入特征間的非線性交互(ReLU是常用選擇)self.relu = nn.ReLU()# 第二層線性變換(降維):將隱藏層映射回ffn_num_outputs(即d_model)self.dense2 = nn.Linear(ffn_num_hiddens, ffn_num_outputs)def forward(self, X):"""前向傳播:對每個位置的特征獨立應用相同的MLP參數:X:輸入張量,形狀為[batch_size, seq_len, feature_dim](feature_dim=ffn_num_input)返回:輸出張量,形狀與X一致([batch_size, seq_len, ffn_num_outputs])"""# 計算流程:輸入 → 升維(增強特征交互) → 非線性激活 → 降維(恢復原維度)return self.dense2(self.relu(self.dense1(X)))class AddNorm(nn.Module):"""殘差連接 + 層規范化(Transformer子層輸出的標準處理):解決深層網絡訓練難題"""def __init__(self, normalized_shape, dropout, **kwargs):"""初始化參數參數:normalized_shape:層規范化的維度(通常為輸入特征的最后一維,如d_model)dropout:Dropout概率(隨機丟棄部分特征,防止過擬合)"""super(AddNorm, self).__init__(** kwargs)# Dropout層:僅作用于子層輸出(保護原始輸入X)self.dropout = nn.Dropout(dropout)# 層規范化層:對每個樣本的特征做歸一化(均值0,方差1),穩定訓練# 與BatchNorm不同,LayerNorm在樣本內計算均值方差,更適合序列數據self.ln = nn.LayerNorm(normalized_shape)def forward(self, X, Y):"""前向傳播:殘差連接 + 層規范化參數:X:子層的原始輸入張量(形狀與Y必須一致,否則無法相加)Y:子層(如注意力/前饋網絡)的輸出張量返回:處理后的張量,形狀與X/Y一致(特征分布更穩定)"""# 步驟解析:# 1. 對Y應用Dropout:隨機丟棄部分特征,防止模型過度依賴子層輸出# 2. 殘差連接:X + dropout(Y) → 保留原始輸入信息,緩解梯度消失(若Y無效,輸出≈X)# 3. 層規范化:對每個樣本計算均值和方差,將特征縮放到標準分布,加速訓練return self.ln(self.dropout(Y) + X)class EncoderBlock(nn.Module):"""Transformer編碼器塊:編碼器的基本單元,堆疊N次形成完整編碼器"""def __init__(self, key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,dropout, use_bias=False, **kwargs):"""初始化編碼器塊參數:key_size/query_size/value_size:注意力機制中K/Q/V的特征維度(通常與num_hiddens一致)num_hiddens:隱藏層特征維度(即d_model,與詞嵌入維度一致)norm_shape:層規范化的維度(通常為[num_hiddens])ffn_num_input/ffn_num_hiddens:前饋網絡的輸入/隱藏層維度num_heads:注意力頭數dropout:Dropout概率(用于注意力和前饋網絡)use_bias:線性層是否使用偏置(控制模型復雜度)"""super(EncoderBlock, self).__init__(** kwargs)# 子層1:多頭自注意力機制(Q=K=V=輸入X,捕獲序列內的依賴關系)self.attention = MultiHeadAttention(heads=num_heads, d_model=num_hiddens, dropout=dropout)# 子層1的輸出處理:殘差連接 + 層規范化self.addnorm1 = AddNorm(norm_shape, dropout)# 子層2:基于位置的前饋網絡(對注意力輸出做非線性變換,增強特征表達)self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens, num_hiddens)# 子層2的輸出處理:殘差連接 + 層規范化self.addnorm2 = AddNorm(norm_shape, dropout)def forward(self, X, valid_lens):"""前向傳播:自注意力 → AddNorm → 前饋網絡 → AddNorm參數:X:輸入序列張量,形狀[batch_size, seq_len, num_hiddens]valid_lens:有效長度張量(屏蔽無效位置,如PAD)返回:經過編碼器塊處理的張量,形狀與X一致(已捕獲序列內依賴)"""# 步驟1:自注意力 + 殘差規范化# 自注意力中,Q=K=V=X,valid_lens控制僅關注有效位置Y = self.addnorm1(X, self.attention(X, X, X, valid_lens)[0]) # [0]取注意力輸出(忽略權重)# 步驟2:前饋網絡 + 殘差規范化# 對自注意力的輸出做非線性變換,增強特征表達return self.addnorm2(Y, self.ffn(Y))class TransformerEncoder(d2l.Encoder):"""Transformer編碼器:將輸入序列編碼為包含上下文信息的特征向量"""def __init__(self, vocab_size, key_size, query_size, value_size,num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, num_layers, dropout, use_bias=False, **kwargs):"""初始化編碼器參數:vocab_size:詞匯表大小(用于詞嵌入層)num_layers:編碼器塊的堆疊數量(層數越多,捕獲的上下文越復雜)其他參數:同EncoderBlock"""super(TransformerEncoder, self).__init__(** kwargs)self.num_hiddens = num_hiddens # 隱藏層維度(d_model)# 詞嵌入層:將詞ID(整數)轉換為向量,形狀[vocab_size, num_hiddens]self.embedding = nn.Embedding(vocab_size, num_hiddens)# 位置編碼層:注入序列順序信息(使用自定義的PositionalEncoding)self.pos_encoding = PositionalEncoding(d_model=num_hiddens, max_seq_len=100)# 堆疊num_layers個編碼器塊(用nn.Sequential管理)self.blks = nn.Sequential()for i in range(num_layers):self.blks.add_module(f"block{i}", # 為每個塊命名,便于調試EncoderBlock(key_size=key_size, query_size=query_size, value_size=value_size,num_hiddens=num_hiddens, norm_shape=norm_shape,ffn_num_input=ffn_num_input, ffn_num_hiddens=ffn_num_hiddens,num_heads=num_heads, dropout=dropout, use_bias=use_bias))def forward(self, X, valid_lens, *args):"""前向傳播:詞嵌入 → 位置編碼 → 多層編碼器塊參數:X:輸入詞ID序列,形狀[batch_size, seq_len]valid_lens:有效長度張量(屏蔽無效位置)返回:編碼后的特征向量,形狀[batch_size, seq_len, num_hiddens]"""# 1. 詞嵌入:將詞ID轉為向量,并縮放(平衡與位置編碼的量級)X = self.embedding(X) * math.sqrt(self.num_hiddens) # 縮放因子為sqrt(d_model)# 2. 注入位置編碼:將位置信息添加到詞嵌入中(Transformer無循環結構,需顯式位置信息)X = self.pos_encoding(X)# 3. 經過所有編碼器塊:逐層捕獲更復雜的上下文依賴self.attention_weights = [None] * len(self.blks) # 存儲各層的注意力權重(用于可視化)for i, blk in enumerate(self.blks):X = blk(X, valid_lens) # 每個塊處理后更新Xself.attention_weights[i] = blk.attention.attention_weights # 保存第i層的注意力權重return Xclass DecoderBlock(nn.Module):"""Transformer解碼器塊:解碼器的基本單元,堆疊N次形成完整解碼器"""def __init__(self, key_size, query_size, value_size, num_hiddens,norm_shape, ffn_num_input, ffn_num_hiddens, num_heads,dropout, i, **kwargs):"""初始化解碼器塊參數:i:當前塊的索引(用于管理歷史狀態)其他參數:同EncoderBlock(增加了解碼器特有的掩蔽自注意力)"""super(DecoderBlock, self).__init__(** kwargs)self.i = i # 塊索引(用于跟蹤歷史狀態)# 子層1:掩蔽多頭自注意力(Q=K=V=解碼器輸入,屏蔽未來位置信息)self.attention1 = MultiHeadAttention(heads=num_heads, d_model=num_hiddens, dropout=dropout)self.addnorm1 = AddNorm(norm_shape, dropout) # 殘差 + 層規范化# 子層2:編碼器-解碼器注意力(Q=解碼器輸出,K=V=編碼器輸出,結合源序列和目標序列信息)self.attention2 = MultiHeadAttention(heads=num_heads, d_model=num_hiddens, dropout=dropout)self.addnorm2 = AddNorm(norm_shape, dropout) # 殘差 + 層規范化# 子層3:前饋網絡(增強特征表達)self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens, num_hiddens)self.addnorm3 = AddNorm(norm_shape, dropout) # 殘差 + 層規范化def forward(self, X, state):"""前向傳播:掩蔽自注意力 → AddNorm → 編碼器-解碼器注意力 → AddNorm → 前饋網絡 → AddNorm參數:X:解碼器輸入序列,形狀[batch_size, seq_len, num_hiddens]state:狀態變量,包含:state[0]:編碼器輸出(enc_outputs)state[1]:編碼器有效長度(enc_valid_lens)state[2]:解碼器歷史狀態(每個塊的歷史輸入,用于推理時累積前文)返回:解碼器輸出 + 更新后的state(包含歷史狀態,用于下一時間步解碼)"""enc_outputs, enc_valid_lens = state[0], state[1] # 提取編碼器輸出和有效長度# 管理解碼器歷史狀態(推理時需累積已解碼的詞,訓練時直接用完整序列)if state[2][self.i] is None: # 訓練時:歷史狀態為空,鍵/值=當前輸入Xkey_values = Xelse: # 推理時:將歷史序列(已解碼的詞)與當前輸入拼接(確保關注前文)key_values = torch.cat((state[2][self.i], X), axis=1) # 沿序列長度維度拼接state[2][self.i] = key_values # 更新當前塊的歷史狀態# 訓練時:生成掩蔽(下三角矩陣),防止關注未來位置(如翻譯時第3個詞不能看第4個詞)if self.training:batch_size, num_steps, _ = X.shape # num_steps是當前輸入的序列長度# 生成形狀為[batch_size, num_steps]的有效長度(如[1,2,...,num_steps])dec_valid_lens = torch.arange(1, num_steps + 1, device=X.device).repeat(batch_size, 1)else: # 推理時:每次僅解碼一個詞,無需掩蔽(前文已包含在key_values中)dec_valid_lens = None# 子層1:掩蔽自注意力(確保解碼順序正確,不泄露未來信息)# Q=X,K=V=key_values(訓練時為完整序列+掩蔽,推理時為歷史+當前)output1, self.attention_weights1 = self.attention1(X, key_values, key_values, dec_valid_lens)Y = self.addnorm1(X, output1) # 殘差 + 層規范化# 子層2:編碼器-解碼器注意力(用編碼器輸出指導解碼,如英文→法文中結合英文信息)# Q=Y(解碼器自注意力輸出),K=V=enc_outputs(編碼器輸出)output2, self.attention_weights2 = self.attention2(Y, enc_outputs, enc_outputs, enc_valid_lens)Z = self.addnorm2(Y, output2) # 殘差 + 層規范化# 子層3:前饋網絡增強特征表達return self.addnorm3(Z, self.ffn(Z)), stateclass TransformerDecoder(d2l.AttentionDecoder):"""Transformer解碼器:將編碼器輸出轉換為目標序列(如翻譯任務中的目標語言)"""def __init__(self, vocab_size, key_size, query_size, value_size,num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens,num_heads, num_layers, dropout, **kwargs):"""初始化解碼器參數:vocab_size:目標語言詞匯表大小num_layers:解碼器塊的堆疊數量其他參數:同TransformerEncoder"""super(TransformerDecoder, self).__init__(** kwargs)self.num_hiddens = num_hiddens # 隱藏層維度(d_model)self.num_layers = num_layers # 解碼器塊數量# 目標語言詞嵌入層:將目標詞ID轉為向量self.embedding = nn.Embedding(vocab_size, num_hiddens)# 位置編碼層:注入目標序列的位置信息self.pos_encoding = PositionalEncoding(d_model=num_hiddens, max_seq_len=100)# 堆疊num_layers個解碼器塊self.blks = nn.Sequential()for i in range(num_layers):self.blks.add_module(f"block{i}",DecoderBlock(key_size=key_size, query_size=query_size, value_size=value_size,num_hiddens=num_hiddens, norm_shape=norm_shape,ffn_num_input=ffn_num_input, ffn_num_hiddens=ffn_num_hiddens,num_heads=num_heads, dropout=dropout, i=i))# 輸出層:將解碼器特征映射到目標詞匯表(vocab_size維度)self.dense = nn.Linear(num_hiddens, vocab_size)def init_state(self, enc_outputs, enc_valid_lens, *args):"""初始化解碼器狀態參數:enc_outputs:編碼器輸出enc_valid_lens:編碼器有效長度返回:初始狀態,包含:enc_outputs, enc_valid_lens, 空歷史狀態列表"""# state[2]為每個解碼器塊的歷史狀態(初始化為None)return [enc_outputs, enc_valid_lens, [None] * self.num_layers]def forward(self, X, state):"""前向傳播:詞嵌入 → 位置編碼 → 多層解碼器塊 → 輸出層參數:X:目標序列詞ID,形狀[batch_size, seq_len]state:解碼器初始狀態(來自init_state)返回:詞匯表概率分布(未歸一化,形狀[batch_size, seq_len, vocab_size]) + 更新后的state"""# 1. 詞嵌入 + 位置編碼(注入目標序列的位置信息)X = self.embedding(X) * math.sqrt(self.num_hiddens) # 縮放詞嵌入X = self.pos_encoding(X) # 添加位置編碼# 2. 經過所有解碼器塊# 存儲注意力權重:[0]為自注意力權重,[1]為編碼器-解碼器注意力權重self._attention_weights = [[None] * len(self.blks) for _ in range(2)]for i, blk in enumerate(self.blks):X, state = blk(X, state) # 每個塊處理后更新X和state# 保存當前塊的兩種注意力權重(用于可視化)self._attention_weights[0][i] = blk.attention_weights1 # 自注意力權重self._attention_weights[1][i] = blk.attention_weights2 # 編碼器-解碼器注意力權重# 3. 輸出層:映射到目標詞匯表(未用softmax,訓練時結合交叉熵損失)return self.dense(X), state@propertydef attention_weights(self):"""返回注意力權重(用于可視化)"""return self._attention_weightsclass CustomEncoderDecoder(nn.Module):"""自定義編碼器-解碼器模型:協調編碼器和解碼器工作,適配訓練函數"""def __init__(self, encoder, decoder):super(CustomEncoderDecoder, self).__init__()self.encoder = encoder # 編碼器實例self.decoder = decoder # 解碼器實例def forward(self, enc_X, dec_X, enc_valid_lens=None):"""前向傳播:編碼器編碼 → 解碼器解碼參數:enc_X:源序列(如英文句子詞ID),形狀[batch_size, src_seq_len]dec_X:目標序列(如法語句子詞ID,訓練時用"強制教學"),形狀[batch_size, tgt_seq_len]enc_valid_lens:源序列有效長度返回:解碼器輸出(詞匯表概率分布) + 解碼器狀態(與訓練函數兼容)"""enc_outputs = self.encoder(enc_X, enc_valid_lens) # 編碼器編碼源序列dec_state = self.decoder.init_state(enc_outputs, enc_valid_lens) # 初始化解碼器狀態# 解碼器解碼:輸入目標序列和解碼器狀態,返回輸出和狀態return self.decoder(dec_X, dec_state)def main():"""主函數:測試Transformer模型的前向傳播和關鍵特性"""# 超參數設置(極簡配置,便于測試)vocab_size = 100 # 詞匯表大小(模擬小詞匯表)d_model = 16 # 模型維度(d_model)num_heads = 2 # 注意力頭數num_layers = 2 # 編碼器/解碼器層數batch_size = 2 # 批次大小seq_len = 5 # 序列長度(每個樣本包含5個詞元)# 1. 創建編碼器和解碼器encoder = TransformerEncoder(vocab_size=vocab_size,key_size=d_model,query_size=d_model,value_size=d_model,num_hiddens=d_model,norm_shape=[d_model],ffn_num_input=d_model,ffn_num_hiddens=32, # 前饋網絡隱藏層維度num_heads=num_heads,num_layers=num_layers,dropout=0.1)decoder = TransformerDecoder(vocab_size=vocab_size,key_size=d_model,query_size=d_model,value_size=d_model,num_hiddens=d_model,norm_shape=[d_model],ffn_num_input=d_model,ffn_num_hiddens=32,num_heads=num_heads,num_layers=num_layers,dropout=0.1)# 2. 生成模擬輸入(隨機詞ID序列)src_seq = torch.randint(0, vocab_size, (batch_size, seq_len)) # 源序列:[2,5]tgt_seq = torch.randint(0, vocab_size, (batch_size, seq_len)) # 目標序列:[2,5]valid_lens = torch.tensor([seq_len, seq_len - 1]) # 有效長度:第1個樣本全有效,第2個樣本最后1個無效# 3. 編碼器前向傳播:測試編碼過程enc_output = encoder(src_seq, valid_lens)print(f"源序列形狀: {src_seq.shape}") # 期望:[2,5]print(f"編碼器輸出形狀: {enc_output.shape}") # 期望:[2,5,16](batch, seq_len, d_model)# 4. 解碼器前向傳播:測試解碼過程model = CustomEncoderDecoder(encoder, decoder) # 封裝編碼器-解碼器dec_output, _ = model(src_seq, tgt_seq, valid_lens)print(f"目標序列形狀: {tgt_seq.shape}") # 期望:[2,5]print(f"解碼器輸出形狀: {dec_output.shape}") # 期望:[2,5,100](batch, seq_len, vocab_size)# 5. 驗證編碼器注意力權重的有效性enc_attn_weights = encoder.attention_weights[0] # 取第0層的注意力權重print(f"編碼器第0層注意力權重形狀: {enc_attn_weights.shape}") # 期望:[2,2,5,5](batch, heads, seq_len, seq_len)# 檢查注意力權重歸一化(每行和應為1.0,因softmax歸一化)head0_row0_sum = enc_attn_weights[0, 0, 0].sum().item() # 第0樣本、第0頭、第0行的權重和print(f"編碼器第0層第0個頭的權重和: {head0_row0_sum:.4f}") # 期望接近1.0# 6. 可視化解碼器自注意力權重(第0層第0個頭)dec_self_attn = decoder.attention_weights[0][0] # 解碼器第0層自注意力權重print(f"解碼器第0層自注意力權重形狀: {dec_self_attn.shape}") # 期望:[2,2,5,5]attn_matrix = dec_self_attn[0, 0].detach().numpy() # 取第0樣本、第0頭的權重矩陣print(f"待可視化的注意力矩陣形狀: {attn_matrix.shape}") # 期望:(5,5)# 繪制注意力熱圖(顏色越深表示關注度越高)plt.figure(figsize=(6, 6))plt.imshow(attn_matrix, cmap='viridis') # 熱圖可視化plt.colorbar(label='注意力權重')plt.title('解碼器第0層自注意力權重')plt.xlabel('鍵位置(Key Position)')plt.ylabel('查詢位置(Query Position)')plt.show()# 7. 驗證編碼器-解碼器注意力的形狀(結合源和目標序列)enc_dec_attn = decoder.attention_weights[1][0] # 第0層編碼器-解碼器注意力print(f"編碼器-解碼器注意力形狀: {enc_dec_attn.shape}") # 期望:[2,2,5,5](batch, heads, tgt_seq_len, src_seq_len)# 程序入口:執行main函數
if __name__ == "__main__":main()
2.6.2、與LSTM對比實現代碼
"""
文件名: 對比實驗
作者: 墨塵
日期: 2025/7/18
項目名: dl_env
備注:
"""
import mathimport torch
import torch.nn as nn
import torch.optim as optim
import matplotlib.pyplot as plt
import numpy as np
from torch.utils.data import Dataset, DataLoaderfrom LLM import PositionalEncoding# -------------------------- 數據生成(帶長距離依賴的序列) --------------------------
class SequenceDataset(Dataset):"""生成帶長距離依賴的序列數據:預測序列的下一個元素,其中偶數位置依賴前2個位置的元素"""def __init__(self, seq_len=10, num_samples=1000):self.seq_len = seq_lenself.num_samples = num_samplesself.data = self._generate_data()def _generate_data(self):"""生成序列:規律為 x[i] = x[i-2] + 噪聲(增強長距離依賴)"""data = []for _ in range(self.num_samples):# 隨機初始化前2個元素seq = [np.random.randn() for _ in range(2)]# 生成后續元素(依賴前2個位置,制造長距離依賴)for i in range(2, self.seq_len + 1): # +1 是因為需要預測下一個元素seq.append(seq[i - 2] + 0.1 * np.random.randn()) # x[i] = x[i-2] + 噪聲data.append(seq)return np.array(data, dtype=np.float32)def __len__(self):return self.num_samplesdef __getitem__(self, idx):seq = self.data[idx]x = seq[:-1] # 輸入序列(前seq_len個元素)y = seq[1:] # 目標序列(后seq_len個元素,即下一個元素預測)return torch.tensor(x), torch.tensor(y)# -------------------------- 模型定義 --------------------------
class TransformerModel(nn.Module):"""簡化的Transformer模型(用于序列預測)"""def __init__(self, input_dim=1, d_model=32, num_heads=2, num_layers=2, dropout=0.1):super().__init__()self.d_model = d_model# 輸入維度映射(將1維序列映射到d_model維)self.input_proj = nn.Linear(input_dim, d_model)self.pos_encoding = PositionalEncoding(d_model, max_seq_len=100)# Transformer編碼器層encoder_layers = nn.TransformerEncoderLayer(d_model=d_model, nhead=num_heads, dim_feedforward=64, dropout=dropout, batch_first=True)self.transformer_encoder = nn.TransformerEncoder(encoder_layers, num_layers=num_layers)# 輸出層(映射回1維)self.output_proj = nn.Linear(d_model, 1)def forward(self, x):# x形狀:[batch_size, seq_len, 1]x = self.input_proj(x) # [batch_size, seq_len, d_model]x = self.pos_encoding(x) # 注入位置信息x = self.transformer_encoder(x) # [batch_size, seq_len, d_model]return self.output_proj(x) # [batch_size, seq_len, 1]class LSTMModel(nn.Module):"""對比用的LSTM模型(參數規模與Transformer相近)"""def __init__(self, input_dim=1, hidden_dim=32, num_layers=2, dropout=0.1):super().__init__()self.lstm = nn.LSTM(input_size=input_dim,hidden_size=hidden_dim,num_layers=num_layers,dropout=dropout,batch_first=True)self.output_proj = nn.Linear(hidden_dim, 1) # 輸出層def forward(self, x):# x形狀:[batch_size, seq_len, 1]lstm_out, _ = self.lstm(x) # [batch_size, seq_len, hidden_dim]return self.output_proj(lstm_out) # [batch_size, seq_len, 1]# -------------------------- 訓練與評估函數 --------------------------
def train_model(model, train_loader, epochs=50, lr=0.001):criterion = nn.MSELoss()optimizer = optim.Adam(model.parameters(), lr=lr)model.train()losses = []for epoch in range(epochs):total_loss = 0.0for x, y in train_loader:# 調整輸入形狀:[batch_size, seq_len] → [batch_size, seq_len, 1]x = x.unsqueeze(-1)y = y.unsqueeze(-1)optimizer.zero_grad()output = model(x)loss = criterion(output, y)loss.backward()optimizer.step()total_loss += loss.item() * x.size(0)avg_loss = total_loss / len(train_loader.dataset)losses.append(avg_loss)if (epoch + 1) % 10 == 0:print(f"Epoch {epoch + 1}/{epochs}, Loss: {avg_loss:.6f}")return lossesdef evaluate_long_sequence(model, seq_len=50):"""評估模型在長序列上的預測能力(測試長距離依賴捕獲)"""model.eval()# 生成一個長序列(長度為seq_len)seq = [np.random.randn() for _ in range(2)]for i in range(2, seq_len):seq.append(seq[i - 2] + 0.1 * np.random.randn()) # 遵循x[i] = x[i-2] + 噪聲# 用模型預測后續元素x = torch.tensor(seq[:20]).unsqueeze(0).unsqueeze(-1).float() # 取前20個元素作為輸入with torch.no_grad():pred = model(x).squeeze().numpy() # 預測接下來的20個元素# 計算與真實值的MSE(關注后10個元素,體現長距離依賴)true = seq[1:21] # 真實的后續元素long_range_mse = np.mean((pred[-10:] - true[-10:]) ** 2) # 僅計算最后10個元素的誤差return long_range_mse# -------------------------- 對比實驗主函數 --------------------------
def main():# 實驗參數seq_len = 20 # 序列長度(包含一定長距離依賴)batch_size = 32epochs = 50hidden_dim = 32 # 確保兩個模型的參數規模相近# 1. 生成數據dataset = SequenceDataset(seq_len=seq_len, num_samples=1000)train_loader = DataLoader(dataset, batch_size=batch_size, shuffle=True)# 2. 初始化模型transformer = TransformerModel(d_model=hidden_dim)lstm = LSTMModel(hidden_dim=hidden_dim)# 3. 訓練模型print("=== 訓練Transformer ===")transformer_losses = train_model(transformer, train_loader, epochs=epochs)print("\n=== 訓練LSTM ===")lstm_losses = train_model(lstm, train_loader, epochs=epochs)# 4. 評估長序列預測能力(測試長距離依賴)transformer_long_mse = evaluate_long_sequence(transformer, seq_len=50)lstm_long_mse = evaluate_long_sequence(lstm, seq_len=50)print(f"\n長序列預測MSE(越小越好):")print(f"Transformer: {transformer_long_mse:.6f}")print(f"LSTM: {lstm_long_mse:.6f}")# 5. 可視化損失曲線plt.figure(figsize=(10, 5))plt.plot(transformer_losses, label='Transformer')plt.plot(lstm_losses, label='LSTM')plt.xlabel('Epoch')plt.ylabel('MSE Loss')plt.title('訓練損失對比')plt.legend()plt.grid(True)plt.show()# 復用之前定義的PositionalEncoding類class PositionalEncoding(nn.Module):def __init__(self, d_model, max_seq_len=80):super().__init__()self.d_model = d_modelpe = torch.zeros(max_seq_len, d_model)position = torch.arange(0, max_seq_len, dtype=torch.float).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):x = x * math.sqrt(self.d_model)seq_len = x.size(1)x = x + self.pe[:, :seq_len]return xif __name__ == "__main__":main()