「Transformer核心必读」编码器层深度解析:从原理机制到 PyTorch 源码实现
1、编码器层的作用
编码器层(Encoder Layer)是 Transformer 模型的核心构建单元,其根本作用是:将输入的初始词表示(经 Embedding + 位置编码后)逐步转化为富含上下文语义的动态向量表示(Contextualized Representation)。
下面从多个维度清晰阐述它的作用:
- 核心功能:上下文化(Contextualization)
- 输入:每个 token 的初始向量(来自 Token Embedding + Positional Encoding),在进入编码器前彼此独立、无上下文交互。
- 输出:每个 token 的新向量,融合了整个输入序列中所有其他 token 的相关信息。
✅ 例如:
- “bank” 在 “river bank” 中 → 向量靠近 water, shore
- “bank” 在 “investment bank” 中 → 向量靠近 finance, money
这一上下文感知能力主要由自注意力机制实现。
- 内部两大子层的分工协作
一个标准编码器层包含两个关键子层,各司其职:
🔹 子层 1:多头自注意力(Multi-Head Self-Attention, MHSA)
- 作用:建立 token 之间的全局依赖关系。
- 机制:每个 token 作为 Query,与其他所有 token(作为 Key/Value)计算注意力权重,并进行加权聚合。
- 效果:让每个 token “感知”整个序列的信息,实现跨位置的上下文融合。
- 关键特性:MHSA 是 token 间信息交互的唯一机制。若移除它,各 token 将无法相互影响(FFN 不跨 token 通信)。
🔹 子层 2:位置前馈网络(Position-wise Feed-Forward Network, FFN)
- 作用:对每个 token 的表示进行非线性变换与特征提炼。
- 机制:独立地对每个 token 应用相同的两层 MLP(通常中间维度扩大至 4 倍,如
d_ff = 4 × d_model)。 - 效果:
- 引入非线性(如 ReLU / GELU)
- 将注意力加权后的结果映射到更高维、更丰富的语义空间
- 相当于每个 token 的“个人专家”,基于上下文决定如何表达自己
💡 类比:
- MHSA = 开会讨论(听取所有人意见,形成共识)
- FFN = 个人内化(基于会议结论,结合自身知识,形成最终观点)
- 架构保障:Pre-LN + 残差连接
- Layer Normalization(Pre-LN):在每个子层之前对输入进行归一化(现代主流做法),显著提升深层模型的训练稳定性。
- 对比:原始 Transformer 使用 Post-LN(子层之后归一化),需学习率 warmup;Pre-LN 则更鲁棒。
- 残差连接(Residual Connection):
output = x + Sublayer(LayerNorm(x))- 防止信息丢失
- 缓解梯度消失
- 允许网络在必要时“绕过”子层变换(类似 Highway Network)
这两者共同保障了即使堆叠数十层,模型仍能有效训练。
- 堆叠效应:层次化语义提取
单个编码器层只能捕获浅层模式。堆叠 N 层后形成深度、分层的语义表征:
- 底层(靠近输入):学习局部特征(如词性、短语边界、基本语法)
- 中层:捕获句法结构(如主谓一致、依存关系、成分结构)
- 高层(靠近输出):理解抽象语义(如情感倾向、逻辑关系、指代消解、任务意图)
📈 实验证明:BERT-large(24 层)在绝大多数 NLP 任务上显著优于 BERT-base(12 层),体现了深度带来的表征能力提升。
- 典型应用场景
- 纯编码器架构:如 BERT、RoBERTa、DeBERTa —— 专用于理解型任务(文本分类、问答、命名实体识别、语义相似度等)
- 编码器-解码器架构:如 T5、BART —— 其中编码器负责全面理解输入序列,并将上下文表示传递给解码器以指导生成
✅ 总结一句话:
编码器层是一个“上下文感知的信息精炼器”:先通过多头自注意力让每个 token 感知全局语境,再通过位置前馈网络进行非线性深度加工,最终输出高度上下文化的向量表示,为各类自然语言理解任务奠定语义基础。
2、编码器层详情
🧱 一、宏观定位:编码器层在 Transformer 中的角色
1.1 整体架构回顾
Transformer 模型由 N 个相同的编码器层(Encoder Layer)堆叠而成。
Input Tokens → [Embedding + Pos Encoding] → Encoder Layer 1 → ... → Encoder Layer N → Output Contextualized Representations
- 输入:词表 ID 序列(如
[23, 456, 78, 0]) - 输出:每个 token 的上下文感知向量表示(Contextualized Embeddings),形状
(B, T, d_model)
1.2 核心目标
将静态的词嵌入(Static Word Embedding)转化为动态的上下文嵌入(Contextual Embedding)。
- 静态嵌入:单词 “bank” 无论出现在何处,其向量都相同。
- 上下文嵌入:
- 在 “river bank” 中,“bank” 的向量会靠近 “water”, “shore”
- 在 “investment bank” 中,“bank” 的向量会靠近 “finance”, “money”
1.3 编码器 vs 解码器
| 特性 | 编码器 (Encoder) | 解码器 (Decoder) |
|---|---|---|
| 注意力类型 | 仅自注意力 (Self-Attention) | 自注意力 + 交叉注意力 (Cross-Attention) |
| 信息流 | 全局双向(可看前后所有词) | 训练时可看全部(带 mask),推理时只能看前面(因果掩码) |
| 用途 | 理解输入(如 BERT) | 生成输出(如 GPT, T5) |
✅ 本篇聚焦:纯编码器(如 BERT 的核心)
🔍 二、编码器层的内部结构:两大子层详解
一个标准的编码器层包含 两个 Pre-LN 子层,按顺序执行:
Input x
──► [Sublayer 1: Multi-Head Self-Attention]
──► [Sublayer 2: Position-wise Feed-Forward Network]
──► Output
我们逐层拆解。
▶ 子层 1:多头自注意力(Multi-Head Self-Attention, MHSA)
2.1 功能:建立 Token 间的全局依赖关系
- 输入:一个序列的所有 token 表示 x = [ x 1 , x 2 , . . . , x t ] ∈ R T × d m o d e l x = [x₁, x₂, ..., xₜ] ∈ ℝ^{T×d_{model}} x=[x1,x2,...,xt]∈RT×dmodel
- 输出:每个 token 的新表示,融合了序列中所有其他 token 的信息。
2.2 为什么叫“自”注意力?
-
Query, Key, Value 都来自同一个输入
x:output = self_attn(x, x, x) -
这意味着每个 token 都在问自己:“序列中的哪些部分对我当前的理解最重要?”
2.3 多头机制:并行学习多种关系
-
单头局限:只能学习一种类型的依赖(如语法)。
-
多头优势:不同头可以关注不同方面:
- Head 1: 关注主谓一致
- Head 2: 关注指代消解(“he” 指谁?)
- Head 3: 关注语义相似性
-
数学形式:
MultiHead ( Q , K , V ) = Concat ( head 1 , . . . , head h ) W O where head i = Attention ( Q W i Q , K W i K , V W i V ) \text{MultiHead}(Q,K,V) = \text{Concat}(\text{head}_1, ..., \text{head}_h)W^O \\ \text{where } \text{head}_i = \text{Attention}(QW_i^Q, KW_i^K, VW_i^V) MultiHead(Q,K,V)=Concat(head1,...,headh)WOwhere headi=Attention(QWiQ,KWiK,VWiV)
2.4 Pre-LN 封装
# 在 EncoderLayer 中
x = self.sublayer1(x, lambda x_norm: self.self_attn(x_norm, x_norm, x_norm))
- 输入:原始表示
x - 处理:先 LayerNorm,再 MHSA,最后
x + Dropout(MHSA(LN(x))) - 输出:融合了全局上下文的新表示
💡 关键点:MHSA 是序列内通信的唯一机制。没有它,每个 token 就是孤岛。
▶ 子层 2:位置前馈网络(Position-wise Feed-Forward Network, FFN)
2.1 功能:对每个 token 的表示进行非线性变换和知识注入
- 输入:来自 MHSA 的上下文表示
z = [z₁, z₂, ..., zₜ] - 输出:进一步提炼后的表示。
2.2 为什么需要 FFN?
-
MHSA 的局限:它只是一个加权平均操作(线性组合)。
output i = ∑ j α i j v j \text{output}_i = \sum_j \alpha_{ij} v_j outputi=j∑αijvj- 无法引入非线性或复杂特征变换。
-
FFN 的作用:
- 通过 ReLU 引入非线性
- 将信息投影到更高维空间 (
d_ff ≈ 4*d_{model}) 再压缩回来 - 相当于一个 per-token 的“专家”,根据上下文决定如何表达自己。
2.3 “Position-wise” 的含义
-
独立处理:FFN 对每个 token 单独、并行地应用相同的 MLP。
# x.shape = (B, T, d_model) # linear1: (d_model -> d_ff) # linear2: (d_ff -> d_model) x = self.linear2(self.dropout(self.activation(self.linear1(x)))) -
不跨 token 通信:FFN 不知道其他 token 的存在,只关心“我这个 token 现在该怎么表示”。
2.4 Pre-LN 封装
# 在 EncoderLayer 中
x = self.sublayer2(x, self.ffn) # 因为 ffn 只需一个参数
- 输入:MHSA 的输出
- 处理:先 LayerNorm,再 FFN,最后
x + Dropout(FFN(LN(x))) - 输出:最终的、高度提炼的上下文表示
💡 哲学类比:
- MHSA = “开会讨论”(听取所有人意见)
- FFN = “个人思考”(基于会议结论,形成自己的最终观点)
⚙️ 三、完整编码器层的代码实现(生产级)
import torch
import torch.nn as nn
from typing import Optional
class EncoderLayer(nn.Module):
"""
单个 Transformer 编码器层(支持依赖注入 + Pre-LN 架构)。
结构:
x ──► [Pre-LN + Self-Attention + Residual] ──► [Pre-LN + FFN + Residual] ──► output
设计特点:
- 使用 Pre-Layer Normalization(训练更稳定)
- 自注意力模块和前馈网络通过构造函数注入(高内聚、低耦合)
- 支持灵活的 mask 输入格式
"""
def __init__(
self,
d_model: int,
self_attn: nn.Module,
ffn: nn.Module,
dropout: float = 0.1
):
"""
Args:
d_model (int): 模型维度(也即 token 表示的向量长度)
self_attn (nn.Module): 自注意力模块,需支持调用方式:
`self_attn(query, key, value, mask=mask)`
ffn (nn.Module): 前馈网络模块,需支持调用方式:
`ffn(x)`,输入输出形状均为 (B, T, d_model)
dropout (float): Dropout 概率,用于 SublayerConnection 内部的正则化
"""
super().__init__()
self.d_model = d_model
self.self_attn = self_attn
self.ffn = ffn
# 两个独立的子层连接(各自包含 LayerNorm + Dropout + 残差)
self.sublayer_attn = SublayerConnection(d_model, dropout)
self.sublayer_ffn = SublayerConnection(d_model, dropout)
def forward(
self,
x: torch.Tensor,
src_mask: Optional[torch.Tensor] = None
) -> torch.Tensor:
"""
前向传播。
Args:
x: 输入张量,shape (batch_size, seq_len, d_model)
src_mask: 可选的源序列 mask,支持:
- (B, T):每个位置是否为有效 token(True 有效)
- (B, 1, T) 或 (B, T, T):标准注意力 mask 格式
- None:无 mask
注意:mask 中 True/1 表示**参与计算**,False/0 表示**被屏蔽**。
Returns:
输出张量,shape (batch_size, seq_len, d_model)
"""
# 自动适配常见 mask 形状(如来自 pad_sequence 的 (B, T))
if src_mask is not None and src_mask.dim() == 2:
src_mask = src_mask.unsqueeze(1) # (B, T) → (B, 1, T)
# 子层 1: 自注意力(通过 lambda 适配单参数接口)
x = self.sublayer_attn(
x,
lambda x_norm: self.self_attn(x_norm, x_norm, x_norm, mask=src_mask)
)
# 子层 2: 前馈网络(直接传递,因 ffn(x) 符合接口)
# 如果 ffn 模块本身就是 ffn(x) 形式(即接收一个张量并返回一个张量),那么直接传 self.ffn 即可,无需包装成 lambda。
# 使用 lambda x_norm: self.ffn(x_norm) 是冗余的函数包装,增加调用开销,且无任何收益。
x = self.sublayer_ffn(x, self.ffn)
return x
🧩 核心机制:SublayerConnection 的内部逻辑
SublayerConnection 是实现 Pre-Layer Normalization (Pre-LN) 架构的核心组件。它的标准实现如下(这是理解一切的关键):
class SublayerConnection(nn.Module):
def __init__(self, d_model, dropout):
super().__init__()
self.norm = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x, sublayer_fn):
"""
Args:
x: 输入张量 (B, T, d_model) —— 此时尚未归一化!
sublayer_fn: 一个可调用对象,例如 lambda x_norm: self.self_attn(...)
Returns:
x + Dropout(sublayer_fn(LayerNorm(x))) # Pre-LN + 残差
"""
# 步骤 1: 先对输入 x 执行 LayerNorm → 得到归一化后的张量
x_norm = self.norm(x) # 👈 真正的归一化发生在这里!!!!!!!!!!!!!!!!!!!
# 步骤 2: 将归一化后的张量传给子层函数
sublayer_output = sublayer_fn(x_norm) # 👈 此时传入的是已归一化的数据!!!!!!!!!!!!!
# 步骤 3: 加上原始输入(残差连接)并应用 Dropout
return x + self.dropout(sublayer_output)
🔍 回到Lambda 表达式
现在看这行代码:
x = self.sublayer_attn(
x, # ← 这是原始输入,未归一化
lambda x_norm: self.self_attn(x_norm, x_norm, x_norm, mask=src_mask)
)
- 第一个参数
x:确实是未归一化的原始输入。 - 第二个参数:是一个函数,其形参名为
x_norm。
执行时发生了什么?
-
SublayerConnection.forward(x, lambda...)被调用; -
它内部执行
x_norm = self.norm(x)→ 得到归一化张量; -
然后调用
sublayer_fn(x_norm),即:(lambda x_norm: ...)(x_norm) -
因此,在 Lambda 函数内部,形参
x_norm绑定的就是那个已经过 LayerNorm 的张量。
✅ 结论:
x_norm这个名字是从 Lambda 函数的视角命名的——它接收到的实参确实是归一化后的数据。尽管外部传给SublayerConnection的是原始x,但SublayerConnection会在调用 Lambda 之前完成归一化。
📌 为什么这样设计?(Pre-LN 架构的本质)
这种模式是 Pre-Layer Normalization 的标准实现:
- “Pre” 指的是:LayerNorm 发生在子层(MHSA/FFN)计算之前
- 对比 Post-LN(原始 Transformer 使用):
- Post-LN:
x → Sublayer → LayerNorm → Residual - Pre-LN:
x → LayerNorm → Sublayer → Residual
- Post-LN:
Pre-LN 的优势在于训练更稳定,尤其在深层网络中无需学习率 warmup。
而 SublayerConnection + Lambda 的组合,正是为了将“归一化 + 子层调用”封装成一个可复用单元,同时保持子层实现的独立性。
💡 类比理解
想象 SublayerConnection 是一个“标准化处理站”:
def standardized_processor(raw_input, processing_fn):
clean_input = normalize(raw_input) # 先清洗
result = processing_fn(clean_input) # 专家只处理干净原料
return raw_input + add_noise(result) # 混合原始与处理结果(残差)
你在调用时写:
output = standardized_processor(dirty_x, lambda clean_x: expert(clean_x))
这里的 clean_x 显然指代“被清洗过的数据”,尽管你传入的是 dirty_x。函数形参的名字反映的是它在函数体内代表的含义,而不是调用时传入变量的原始状态。
✅ 总结
| 问题 | 回答 |
|---|---|
传入 sublayer_attn 的 x 是否已归一化? | ❌ 没有,它是原始输入 |
Lambda 的形参 x_norm 是否名副其实? | ✅ 是的!因为 SublayerConnection 会在调用 Lambda 前先执行 x_norm = LayerNorm(x),并将结果作为实参传入 |
| 这种设计是否符合 Pre-LN? | ✅ 完全符合:LayerNorm 发生在子层(MHSA/FFN)之前 |
| 命名是否合理? | ✅ 非常合理!x_norm 准确描述了 Lambda 函数内部接收到的数据语义 |
所以,代码没有错误,命名也非常恰当——只是需要理解 SublayerConnection 在背后完成了“先归一化,再调用子层”的工作。这正是 Pre-LN 架构的精髓所在,也是现代 Transformer 实现(如 BERT、T5)的主流做法。
📊 四、数据流与形状追踪
让我们一步步追踪数据形状:
# 初始输入
x = torch.randint(0, 500, (2, 4)) # (B=2, T=4)
# 1. Embedding + Positional Encoding
x = embed(x) # (2, 4, 512)
x = pos_enc(x) # (2, 4, 512)
# 2. 进入 EncoderLayer
# a. Sublayer 1: MHSA
# - norm(x) -> (2, 4, 512)
# - self_attn(...) -> (2, 4, 512)
# - x + dropout(...) -> (2, 4, 512)
# b. Sublayer 2: FFN
# - norm(x) -> (2, 4, 512)
# - ffn(...) -> (2, 4, 512)
# - x + dropout(...) -> (2, 4, 512)
# 最终输出
print(x.shape) # (2, 4, 512)
- 形状恒定:整个编码器层是等变的(equivariant),输入输出形状完全相同。
- 信息转换:虽然形状没变,但每个
(512,)向量的语义内容已被彻底重塑。
🧠 五、为什么是这两个子层?历史与动机
5.1 自注意力的诞生
- RNN/CNN 的痛点:
- RNN:长距离依赖困难,无法并行。
- CNN:感受野有限,需多层堆叠才能捕获全局信息。
- 自注意力的突破:
- 一步到位:任意两个 token 之间都有直接连接。
- 完全并行:所有 token 的计算可同时进行。
5.2 FFN 的必要性
- 纯注意力模型的缺陷:
- 注意力机制本质上是凸组合(权重和为1),输出空间受限于输入向量的凸包。
- 无法表示超出输入范围的复杂概念。
- FFN 的补足:
- 通过高维非线性映射,极大地扩展了模型的表达能力。
- 移除 FFN 会导致性能严重下降(实证结论)。
📚 经典引用:
“The attention mechanism learns to weight the input, but the feed-forward network learns what to do with the weighted sum.”
—— Ashish Vaswani et al., “Attention Is All You Need”
🔄 六、堆叠 N 层:深度的力量
单个编码器层只能捕获浅层依赖。堆叠多层后:
- 底层:学习局部模式(如词性、短语结构)
- 中层:学习句法关系(如主谓宾)
- 高层:学习语义和逻辑(如情感、意图)
这种层次化表征是深度学习的核心优势。
📈 实验事实:BERT-base (12层) vs BERT-large (24层),后者在几乎所有任务上都显著更强。
⚠️ 七、常见误区与陷阱
| 误区 | 正确理解 |
|---|---|
| “LayerNorm 可以放在最后(Post-LN)” | Pre-LN 更稳定,是现代标准 |
| “两个子层可以共享同一个 LayerNorm” | 必须独立,否则破坏模块化 |
| “FFN 可以省略” | FFN 提供非线性和表达能力,不可或缺 |
| “MHSA 的 Q/K/V 可以来自不同地方” | 在编码器中,必须是 Self-Attention(Q=K=V=x) |
🔮 八、现代变体与演进
虽然基础结构稳固,但研究者仍在探索改进:
-
FFN 变体:
-
GLU (Gated Linear Unit):LLaMA 使用
SwiGLU,性能优于 ReLU。x = self.w2(F.silu(self.w1(x)) * self.v1(x))
-
-
注意力变体:
- Linear Attention:降低 O(T²) 复杂度。
- Local Attention:限制感受野,提升长文本效率。
-
Normalization 变体:
-
RMSNorm:LLaMA 使用,比 LayerNorm 更快,效果相当。
x_norm = x / torch.sqrt(x.pow(2).mean(-1, keepdim=True) + eps)
-
✅ 九、总结:编码器层的本质
编码器层是一个精心设计的“信息精炼厂”。
-
第一车间(MHSA):
- 任务:收集全局情报。
- 方法:让每个工人(token)向所有其他工人提问并汇总答案。
- 产出:一份包含团队共识的初步报告。
-
第二车间(FFN):
- 任务:深度加工。
- 方法:每个工人独立分析初步报告,结合自己的专业知识,撰写最终报告。
- 产出:一份高度专业化、上下文精准的最终文档。
-
质量控制(Pre-LN + Residual):
- 原则:任何环节出错,都可以退回原始材料重做。
- 保障:信息永不丢失,梯度畅通无阻。
3、代码
import torch
import torch.nn as nn
import math
import torch.nn.functional as F
# 注意是 Embeddings, 不是 Embedding, 和 nn.Embedding 区分
# 仅仅是 词嵌入, 没有 位置编码
class Embeddings(nn.Module):
def __init__(self, vocabulary_size: int, d_model: int):
super().__init__()
self.vocabulary_size = vocabulary_size # 词表大小
self.d_model = d_model # 词向量维度
self.embed = nn.Embedding(num_embeddings=vocabulary_size, embedding_dim=d_model)
def forward(self, x):
# x.shape = (N, T)
# nn.Embedding 要求输入的 indices(即 x)必须是整数类型(如 torch.long 或 torch.int)
# math.sqrt 放缩的作用
x = self.embed(x) * math.sqrt(self.d_model)
return x
# 位置编码 【 Positional adj.位置的 】
class PositionalEncoding(nn.Module):
def __init__(self, d_model: int, max_len: int = 1024):
super().__init__()
self.d_model = d_model
self.max_len = max_len
# max_len # 最大 token 数量
# d_model # 词向量维度
# 创建大表格
pe = torch.zeros(size=(max_len, d_model))
# 创建位置索引
position = torch.arange(start=0, end=max_len, step=1, dtype=torch.float)
print('position.shape =', position.shape) # torch.Size([1024])
print(f'position: {position}')
# tensor([0.0000e+00, 1.0000e+00, 2.0000e+00, ..., 1.0210e+03, 1.0220e+03, 1.0230e+03])
# 目的:
# pos dim_0 dim_1 dim_2 dim_3 dim_4 ... dim_d_model
# 0 0*w0 0*w1 0*w2 0*w3 0*w4 ... 0*wd_model
# 1 1*w0 1*w1 1*w2 1*w3 1*w4 ... 1*wd_model
# 2 2*w0 2*w1 2*w2 2*w3 2*w4 ... 2*wd_model
# 3 3*w0 3*w1 3*w2 3*w3 3*w4 ... 3*wd_model
# 4 4*w0 4*w1 4*w2 4*w3 4*w4 ... 4*wd_model
# ...
# max_len m*w0 m*w1 m*w2 m*w3 m*w4 ... m*wd_model
# position 是 (max_len, ) 即 [0, 1, 2, ..., max_len]
# 每行都需要 d_model 哥 0, 要变成 (max_len, d_model) 个0
# 如果变成 (1, max_len), 那即使广播,也只能变成:
# [ [0, 1, 2, ..., max_len],
# [0, 1, 2, ..., max_len],
# ...
# [0, 1, 2, ..., max_len] ]
# 所以只能变成 (max_len, 1) -> 广播后:
# [ [0], [ [0, 0, 0, ..., 0],
# [1], [0, 0, 0, ..., 0],
# [2], [0, 0, 0, ..., 0],
# [3], [0, 0, 0, ..., 0],
# ..., ...,
# [max_len] ] [0, 0, 0, ..., 0] ]
position = position.unsqueeze(dim=1)
# 计算 w = 10000 ^ (- 2i/d) = exp( 2i * (-ln(10000) / d) )
div_term = torch.exp(
torch.arange(0, d_model, 2, dtype=torch.float) *
(-math.log(10000) / d_model)
)
print('div_term.shape =', div_term.shape) # torch.Size([34])
# div_term 需要变成 每行都是 [w0, w1, w2, w3, ..., w_d_model]
# 即:
# [ [w0, w1, w2, w3, ..., w_d_model],
# [w0, w1, w2, w3, ..., w_d_model],
# ...,
# [w0, w1, w2, w3, ..., w_d_model] ]
# 所以 只能变成 (1, d_model)
# 但是 position 已经变了, (max_len, 1) 与 (d_model, ) 运算
# 可以广播, (max_len, 1) -> (max_len, d_model)
# (1, d_model) -> (max_len, d_model)
# 所以 div_term 可以不进行升维,但也可以显示升维
# product 相当于算出来了 相位, 即 sin(x) 中的 x
product = position * div_term
print('product.shape =', product.shape) # torch.Size([1024, 34])
# 万事俱备,只差 sin、cos
# sin:
# 取所有行, 每行从0开始,取到末尾,步长为 2
pe[:, 0:: 2] = torch.sin(product)
# cos:
# 取所有行, 每行从1开始,取到末尾,步长为 2
pe[:, 1:: 2] = torch.cos(product)
print('pe.shape =', pe.shape) # torch.Size([1024, 68])
print(pe) # 只要 max_len 和 d_model 确定了,那 pe 就确定了额
# tensor([[ 0.0000e+00, 1.0000e+00, 0.0000e+00, ..., 1.0000e+00, 0.0000e+00, 1.0000e+00],
# [ 8.4147e-01, 5.4030e-01, 6.9087e-01, ..., 1.0000e+00, 1.3111e-04, 1.0000e+00],
# [ 9.0930e-01, -4.1615e-01, 9.9897e-01, ..., 1.0000e+00, 2.6223e-04, 1.0000e+00],
# ...,
# [ 9.9236e-01, -1.2340e-01, -1.1004e-02, ..., -6.7493e-01, 2.6875e-01, 9.6321e-01],
# [ 4.3234e-01, -9.0171e-01, 6.8216e-01, ..., -6.7505e-01, 2.6888e-01, 9.6317e-01],
# [-5.2517e-01, -8.5100e-01, 9.9852e-01, ..., -6.7518e-01, 2.6900e-01, 9.6314e-01]])
# 注册为 buffer
self.register_buffer(name='position_encoding', tensor=pe, persistent=True)
def forward(self, x):
T = x.size(dim=1) # 获取 T
print(f'T = {T}') # 3
if T > self.max_len:
print('序列长度超过最大序列长度, 位置编码失败!')
return None # 即使不写 None,也是默认返回 None
print(f'x.shape = {x.shape}') # torch.Size([2, 3, 68])
print(f'self.position_encoding.shape = {self.position_encoding.shape}') # torch.Size([1024, 68])
# 输入 x 可能只有 T=3 个 token(如 (2, 3, 68))。
# 所以不能直接加整个 pe,而应该只取前 T 行!
# x = x + self.position_encoding # 错误!
# 这里有讲究: position_encoding[: T] vs position_encoding[: T, : ]
# 在 《`pe[: T]` vs `pe[: T, : ]`》 里有详情
x = x + self.position_encoding[: T] # 这里有广播机制
return x
# 多头注意力
class MultiHeadAttn(nn.Module):
def __init__(
self,
d_model: int,
num_heads: int,
d_k: int = None, # query/key 的每个 head 维度
d_v: int = None, # value 的每个 head 维度
dropout: float = 0.1
):
super().__init__()
# 保证词嵌入维度 能整除 头数
# False 则报错,True 则正常运行
# 正确约束是:当 d_k 未指定时,d_model 必须能被 num_heads 整除(因为 d_k = d_model // num_heads)
# 一旦指定了 d_k,就没有任何整除要求!
# d_v 同理
if d_k is None or d_v is None:
assert d_model % num_heads == 0, f'词嵌入维度不能整除头数'
self.d_model = d_model
self.num_heads = num_heads
self.d_k = d_k if d_k is not None else d_model // num_heads
# 通常 d_v = d_k,但可以 d_v 不等于 d_k
self.d_v = d_v if d_v is not None else d_model // num_heads
print(f'd_k = {self.d_k}') # 64
print(f'd_v = {self.d_v}') # 64
# 把多个小矩阵拼接成大矩阵
self.W_q = nn.Linear(in_features=self.d_model, out_features=self.num_heads * self.d_k, bias=False)
self.W_k = nn.Linear(in_features=self.d_model, out_features=self.num_heads * self.d_k, bias=False)
self.W_v = nn.Linear(in_features=self.d_model, out_features=self.num_heads * self.d_v, bias=False)
# 在Softmax之后,在加权融合V之前
self.dropout = nn.Dropout(p=dropout)
# 加权融合V 得到的是 (token, h * d_v),所以这里的输入是 h * d_v
# 输出的线性投影, 拼接后的 (h * d_v) 需要映射回 d_model
self.W_o = nn.Linear(in_features=self.num_heads * self.d_v, out_features=self.d_model, bias=False)
def forward(
self,
query_input: torch.Tensor,
key_input: torch.Tensor,
value_input: torch.Tensor,
mask: torch.Tensor = None
):
B, T, D = query_input.shape # 需要保证 batch_first = True
# 为什么形参名字叫 query_input, key_input, value_input,在《关键:分清 X 到底是什么》有详情
# 线性变换, 矩阵是由多个heads小矩阵合并而成
Q = self.W_q(query_input) # (B, T, h * d_k)
K = self.W_k(key_input) # (B, T, h * d_k)
V = self.W_v(value_input) # (B, T, h * d_v)
# 拆分为多头 Q: (B, T, h * d_k) -> (B, T, h, d_k) -> (B, h, T, d_k)
# 为什么写 -1,详情在《拆分多头时:`Q.reshape(B, -1, self.num_heads, self.d_k)`》
Q_heads = Q.reshape(B, -1, self.num_heads, self.d_k).transpose(1, 2) # (B, h, T, d_k)
K_heads = K.reshape(B, -1, self.num_heads, self.d_k).transpose(1, 2) # (B, h, T, d_k)
V_heads = V.reshape(B, -1, self.num_heads, self.d_v).transpose(1, 2) # (B, h, T, d_v)
# 计算 缩放点击注意力(不是除以 h*d_k!) (B, h, T, T)
score = torch.matmul(Q_heads, K_heads.transpose(-1, -2)) / math.sqrt(self.d_k)
if mask is not None:
# masked_fill 返回新张量,不是 in_place 操作
score = score.masked_fill(mask == 0, -torch.inf)
# Softmax计算分布
attn_weight = F.softmax(score, dim=-1) # (B, h, T, T)
# 使用 dropout
attn_weight = self.dropout(attn_weight)
# 加权聚合 value 得到 上下文
context = torch.matmul(attn_weight, V_heads) # (B, h, T, d_v)
# 把所有 head 合并
# (B, h, T, d_v) --> (B, T, h, d_v) --> (B, T, h * d_v)
context = context.transpose(1, 2).reshape(B, -1, self.num_heads * self.d_v)
# 线性投影
output = self.W_o(context)
return output
# 前馈全连接层
class PositionWiseFeedForwardNetwork(nn.Module):
def __init__(
self,
d_model: int,
d_ff: int = None,
dropout: float = 0.1,
activation='relu'
):
super().__init__()
self.d_model = d_model
self.d_ff = d_ff if d_ff is not None else 4 * d_model # 默认 4 倍
self.linear1 = nn.Linear(in_features=d_model, out_features=self.d_ff)
if activation == 'relu':
self.activation = nn.ReLU()
else:
raise ValueError('还需要添加更多激活函数')
raise ValueError(f"Unsupported activation: '{activation}'. Currently only 'relu' is supported.")
self.dropout = nn.Dropout(p=dropout)
self.linear2 = nn.Linear(in_features=self.d_ff, out_features=self.d_model)
def forward(self, x: torch.Tensor) -> torch.Tensor:
# 返回值预计为 张量(但Python解释器可不管什么类型,即使不返回张量也不会报错)
"""
:param x: x.shape = (B, T, d_model)
:return: (B, T, d_model)
"""
x = self.linear1(x)
x = self.activation(x)
x = self.dropout(x)
x = self.linear2(x)
# 可一行搞定
# return self.linear2(self.dropout(self.activation(self.linear1(x))))
return x
# 规范化层
class LayerNorm(nn.Module):
def __init__(self, d_model: int, eps=1e-5):
super().__init__()
self.d_model = d_model
self.eps = eps
# 仿射变换 γ,用于缩放(初始为1,表示不缩放)
self.gamma = nn.Parameter(torch.ones(size=(self.d_model,)))
# 仿射变换 β,用于偏移(初始为0,表示不偏移)
self.beta = nn.Parameter(torch.zeros(size=(self.d_model,)))
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""
支持任意形状,只要最后一维是 d_model
例如: (B, d_model) 或 (B, T, d_model) 或 (B, T1, T2, d_model)
"""
# 计算均值
mean = x.mean(dim=-1, keepdim=True)
# 计算有偏方差
var = x.var(dim=-1, keepdim=True, unbiased=False)
# 标准化
x_norm = (x - mean) / (torch.sqrt(var + self.eps))
# 仿射变换
output = self.gamma * x_norm + self.beta
return output
# 子层连接结构
class SublayerConnection(nn.Module):
def __init__(self, d_model: int, dropout: float = 0.1):
super().__init__()
# 规范化层
self.norm = LayerNorm(d_model=d_model)
# Dropout
self.dropout = nn.Dropout(p=dropout)
def forward(self, x, sublayer_fn) -> torch.Tensor:
# 绝对不能写成 x = self.norm(x), 这样会覆盖原始的 x, 残差连接还要用到原始的 x
x_norm = self.norm(x) # 规范化层
sublayer_output = sublayer_fn(x_norm) # MHSA、FNN 等
output = x + self.dropout(sublayer_output) # 残差连接
# 一行搞定
# return x + self.dropout(sublayer_fn(self.norm(x)) )
return output
class EncoderLayer(nn.Module):
def __init__(self, d_model: int, self_attn: nn.Module, ffn: nn.Module, dropout: float = 0.1):
"""
:param self_attn: 多头自注意力机制(MHSA)
:param ffn: 前馈全连接(FFN)
"""
super().__init__()
self.d_model = d_model
self.self_attn = self_attn
self.ffn = ffn
self.dropout = dropout
# 第一个子层 —— 多头自注意力(MHSA Sub-layer)
self.sublayer_attn = SublayerConnection(d_model=self.d_model, dropout=self.dropout)
# 第二个子层 —— 前馈网络(FFN Sub-layer)
self.sublayer_ffn = SublayerConnection(d_model=self.d_model, dropout=self.dropout)
def forward(self, x, mask=None):
"""
:param mask: 这里先不管 mask,后续会对 mask 进行详细的改进
:param x: x.shape = (B, T, D)
"""
# 第一个子层
x1 = self.sublayer_attn(x, lambda x_norm: self.self_attn(x_norm, x_norm, x_norm, mask))
# 第二个子层
# 如果 self.ffn 本身就是一个接受单个张量输入并返回张量的模块(如 nn.Sequential 或自定义 FFN),那么直接传 self.ffn 即可。
# 使用 lambda x_norm: self.ffn(x_norm) 是冗余的函数包装,增加调用开销,且无任何收益。
x2 = self.sublayer_ffn(x1, self.ffn)
return x2
经典代码:【在《编码器层详情》中也有】
import torch
import torch.nn as nn
from typing import Optional
class EncoderLayer(nn.Module):
"""
单个 Transformer 编码器层(支持依赖注入 + Pre-LN 架构)。
结构:
x ──► [Pre-LN + Self-Attention + Residual] ──► [Pre-LN + FFN + Residual] ──► output
设计特点:
- 使用 Pre-Layer Normalization(训练更稳定)
- 自注意力模块和前馈网络通过构造函数注入(高内聚、低耦合)
- 支持灵活的 mask 输入格式
"""
def __init__(
self,
d_model: int,
self_attn: nn.Module,
ffn: nn.Module,
dropout: float = 0.1
):
"""
Args:
d_model (int): 模型维度(也即 token 表示的向量长度)
self_attn (nn.Module): 自注意力模块,需支持调用方式:
`self_attn(query, key, value, mask=mask)`
ffn (nn.Module): 前馈网络模块,需支持调用方式:
`ffn(x)`,输入输出形状均为 (B, T, d_model)
dropout (float): Dropout 概率,用于 SublayerConnection 内部的正则化
"""
super().__init__()
self.d_model = d_model
self.self_attn = self_attn
self.ffn = ffn
# 两个独立的子层连接(各自包含 LayerNorm + Dropout + 残差)
self.sublayer_attn = SublayerConnection(d_model, dropout)
self.sublayer_ffn = SublayerConnection(d_model, dropout)
def forward(
self,
x: torch.Tensor,
src_mask: Optional[torch.Tensor] = None
) -> torch.Tensor:
"""
前向传播。
Args:
x: 输入张量,shape (batch_size, seq_len, d_model)
src_mask: 可选的源序列 mask,支持:
- (B, T):每个位置是否为有效 token(True 有效)
- (B, 1, T) 或 (B, T, T):标准注意力 mask 格式
- None:无 mask
注意:mask 中 True/1 表示**参与计算**,False/0 表示**被屏蔽**。
Returns:
输出张量,shape (batch_size, seq_len, d_model)
"""
# 自动适配常见 mask 形状(如来自 pad_sequence 的 (B, T))
if src_mask is not None and src_mask.dim() == 2:
src_mask = src_mask.unsqueeze(1) # (B, T) → (B, 1, T)
# 子层 1: 自注意力(通过 lambda 适配单参数接口)
x = self.sublayer_attn(
x,
lambda x_norm: self.self_attn(x_norm, x_norm, x_norm, mask=src_mask)
)
# 子层 2: 前馈网络(直接传递,因 ffn(x) 符合接口)
# 如果 ffn 模块本身就是 ffn(x) 形式(即接收一个张量并返回一个张量),那么直接传 self.ffn 即可,无需包装成 lambda。
# 使用 lambda x_norm: self.ffn(x_norm) 是冗余的函数包装,增加调用开销,且无任何收益。
x = self.sublayer_ffn(x, self.ffn)
return x
更多推荐


所有评论(0)