1、编码器层的作用

编码器层(Encoder Layer)是 Transformer 模型的核心构建单元,其根本作用是:将输入的初始词表示(经 Embedding + 位置编码后)逐步转化为富含上下文语义的动态向量表示(Contextualized Representation)

下面从多个维度清晰阐述它的作用:


  1. 核心功能:上下文化(Contextualization)
  • 输入:每个 token 的初始向量(来自 Token Embedding + Positional Encoding),在进入编码器前彼此独立、无上下文交互。
  • 输出:每个 token 的新向量,融合了整个输入序列中所有其他 token 的相关信息

✅ 例如:

  • “bank” 在 “river bank” 中 → 向量靠近 water, shore
  • “bank” 在 “investment bank” 中 → 向量靠近 finance, money

这一上下文感知能力主要由自注意力机制实现


  1. 内部两大子层的分工协作

一个标准编码器层包含两个关键子层,各司其职:

🔹 子层 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 = 个人内化(基于会议结论,结合自身知识,形成最终观点)

  1. 架构保障:Pre-LN + 残差连接
  • Layer Normalization(Pre-LN):在每个子层之前对输入进行归一化(现代主流做法),显著提升深层模型的训练稳定性。
    • 对比:原始 Transformer 使用 Post-LN(子层之后归一化),需学习率 warmup;Pre-LN 则更鲁棒。
  • 残差连接(Residual Connection)output = x + Sublayer(LayerNorm(x))
    • 防止信息丢失
    • 缓解梯度消失
    • 允许网络在必要时“绕过”子层变换(类似 Highway Network)

这两者共同保障了即使堆叠数十层,模型仍能有效训练。


  1. 堆叠效应:层次化语义提取

单个编码器层只能捕获浅层模式。堆叠 N 层后形成深度、分层的语义表征

  • 底层(靠近输入):学习局部特征(如词性、短语边界、基本语法)
  • 中层:捕获句法结构(如主谓一致、依存关系、成分结构)
  • 高层(靠近输出):理解抽象语义(如情感倾向、逻辑关系、指代消解、任务意图)

📈 实验证明:BERT-large(24 层)在绝大多数 NLP 任务上显著优于 BERT-base(12 层),体现了深度带来的表征能力提升。


  1. 典型应用场景
  • 纯编码器架构:如 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

执行时发生了什么?

  1. SublayerConnection.forward(x, lambda...) 被调用;

  2. 它内部执行 x_norm = self.norm(x) → 得到归一化张量;

  3. 然后调用 sublayer_fn(x_norm),即:

    (lambda x_norm: ...)(x_norm)
    
  4. 因此,在 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

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_attnx 是否已归一化?❌ 没有,它是原始输入
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)

🔮 八、现代变体与演进

虽然基础结构稳固,但研究者仍在探索改进:

  1. FFN 变体

    • GLU (Gated Linear Unit):LLaMA 使用 SwiGLU,性能优于 ReLU。

      x = self.w2(F.silu(self.w1(x)) * self.v1(x))
      
  2. 注意力变体

    • Linear Attention:降低 O(T²) 复杂度。
    • Local Attention:限制感受野,提升长文本效率。
  3. Normalization 变体

    • RMSNorm:LLaMA 使用,比 LayerNorm 更快,效果相当。

      x_norm = x / torch.sqrt(x.pow(2).mean(-1, keepdim=True) + eps)
      

✅ 九、总结:编码器层的本质

编码器层是一个精心设计的“信息精炼厂”

  1. 第一车间(MHSA)

    • 任务:收集全局情报。
    • 方法:让每个工人(token)向所有其他工人提问并汇总答案。
    • 产出:一份包含团队共识的初步报告。
  2. 第二车间(FFN)

    • 任务:深度加工。
    • 方法:每个工人独立分析初步报告,结合自己的专业知识,撰写最终报告。
    • 产出:一份高度专业化、上下文精准的最终文档。
  3. 质量控制(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

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐