概述
语言模型的核心目标是对自然语言的概率分布进行建模，这一任务在自然语言处理研究中占据重要地位，是其基础性工作之一。随着基于 Transformer 架构的语言模型不断发展，以及预训练-微调范式在各类自然语言处理任务中取得突破性成果，自 2020 年 OpenAI 发布 GPT-3 以来，大语言模型的研究逐步深入。目前市面上所有的大语言模型都是在Transformer架构的基础上进行改进，要相对大模型的理论有所了解，就必须要深入了解Transformer的基本原理。
Transformer 结构是由 Google 在 2017 年提出并首先应用于机器翻译的神经网络模型架构，原论文就是大名鼎鼎的《Attention is All You Need》。机器翻译的目标是从源语言（Source Language）转换到目标语言（Target Language）。Transformer结构完全通过注意力机制完成对源语言序列和目标语言序列全局依赖的建模。Transformer是一种典型的Seq2Seq架构，分为编码器和解码器两部分，编码器负责对源语言语义进行分析和理解，解码器则负责将其语义映射到另一种语言上，本质是一个生成任务。
Transfomer的结构由下图所示，左侧是编码器，右侧是解码器，他们都由多个Transfomer块组成，每一个Transfomer块都接受一个向量序列 $ {x_i}^t_{i=1} ，并 输 出 等 长 {y_i}^t_{i=1} $。
image-20250614190335599
嵌入表示层
语言本身是一种离散的抽象信息，如果想要对文本进行计算就必须将其先进行数值化。其整体的流程是先去建立词表，将训练样本按照词表进行切分，切分后按照词表的索引进行token化，这样就将词转化成了数值，例如“我”这个词在词表中的索引值是10，那么它的token值就是10。但是若将词直接映射为整数，模型会错误假设类别间存在数值关系，除此之外也无法计算词之间的距离，并且不能进行批量计算，所以要将其进行向量化。最开始的时候使用one-hot编码，每一个词根据词表的长度来生成指定维数的向量，比如词表长度为5，“我”在此表中的索引为3，那么就构造一个[0,0,0,1,0]的向量，每个样本仅在对应类别维度为1，其余99%以上为0，这种方式构造的样本矩阵过于稀疏，当词表过长时，会出现维度爆炸的问题，计算效率非常低，而且仍然无法捕捉词语之间的关联关系，后面就提出了像word2vec和glove这种可训练的embedding模型，用于词嵌入操作，他们的本质还是神经网络，比如word2vec的Skip-Gram 模型架构，就是通过中心词预测上下文词的方式来进行训练，输入中心词的one-hot向量，通过隐藏层来生成中心词向量，最后输出预测窗口内所有上下文词的概率。Transformer并没有使用这些独立的Embedding方法，而是使用了torch自带的nn.Embedding方法来进行词嵌入，它的底层是一种lookup Table，是一个可学习的矩阵，形状为 [词汇表大小, 嵌入维度]，输入是离散的Token ID（如单词索引），而输出对应的稠密向量（嵌入向量），例如下面的代码：
1 2 embedding = nn.Embedding(num_embeddings=10000 , embedding_dim=300 ) output = embedding(torch.tensor([42 ]))
这样做的好处，第一点就是作为模型一部分，随任务端到端优化，并且可动态适配下游任务。
虽然对词进行了嵌入操作，将其转换成了向量的形式，但是仍然有一些问题，RNN或者LSTM是基于循环的方式建模文本输入的，每次将单个词送入到模型中计算，这种循环结构天然带有顺序关系，每一个词都可以根据自己的循环顺序关系来确定位置关系，但是transformer没有这种循环结构，它是直接将整个样本文本输入到模型中的，所以，在送入编码器端建模其上下文语义之前，一个非常重要的操作是在词嵌入中加入位置编码（PositionalEncoding）。具体来说，序列中每一个单词所在的位置都对应一个向量。这一向量会与单词表示对应的词向量相加并送入后续模块中做进一步处理。在训练过程中，模型会自动地学习到如何利用这部分位置信息。那如何做这种位置编码呢，很自然的就能想到直接用这个词在句子中的位置进行编码，比如说“我爱大语言模型”，假如说按照词表拆分成了[“我”， “爱”， “大”， “语言”， “模型”]，token化之后变成[1201, 15, 20, 100,1500]，之后进行word embedding，如果word embedding的长度也是5，那就是一个5x5的矩阵，位置可以看成[0,1,2,3,4]，之后再将这些绝对位置编码成one-hot，和其embedding矩阵维度一致，最后将其相加，但是这种绝对位置编码的方式存在几个比较大的问题。
第一个问题，数值过大且泛化能力差 ，如果说序列长度可变的情况下，当句子长度过长时，pos的值也会变得非常大，这种大数值会破坏模型的稳定性，且模型难以泛化到训练时未见过长度的序列，如果说训练时最大序列长度为512，推理时遇到600长度的序列，模型就无法处理超出范围的位置索引，而且直接使用大数值的绝对位置编码会导致位置编码的数值范围无界，影响梯度计算和模型收敛。第二个问题，归一化后的相对位置不一致 ，比如将pos归一化到[0,1]区间（如pos/max_length），会导致相同归一化值在不同长度序列中表示不同绝对位置 ，举个例子，长度为10的序列中，0.1对应第一个词，而到了100的序列中，则表示的是第10个词。第三个问题，无法捕捉相对位置信息 ，自然语言中相对位置 （如词序方向、距离）对语义至关重要，而pos仅提供绝对位置，无法直接表达相对关系，模型无法理解正常的语义顺序，从而导致语义混淆。第四个问题就是缺乏多尺度位置感知 ，使用这种绝对位置编码时，所有维度使用相同的线性值，无法区分局部细节 （相邻词顺序）和全局结构 （句首/句尾特征）。总的来说这种绝对位置编码方式，并不满足理想位置编码的核心要求，只满足了唯一性，而缺失了相对位置一致性、有界性、外推能力和方向感知能力。
Transformer引入了正余弦函数来进行相对位置编码，基本解决上述几个关键问题，它使用不同频率的正余弦函数来让模型感知感知词序关系，其公式如下： $$
PE(pos, 2i) = sin(\frac{pos}{10000^{2i/d}}) \\
PE(pos, 2i + 1) = cos(\frac{pos}{10000^{2i/d}})
$$ pos表示词的绝对位置，i 是维度索引，d 是模型的总维度数。首先说一下d，它表示词嵌入向量的总维度数，也就是每个词向量的长度，这个值必须是偶数，在transformer的原论文中，这个值是512，像BERT则是768。，2i和2i+1是用于区分位置编码向量中不同维度的索引设计 ，其核心目的是通过交替的正弦（sin）和余弦（cos）函数 生成位置编码，使模型能同时捕捉绝对位置 和相对位置 信息。i的值是从0到d/2-1的遍历结果，用于定位向量中的具体维度位置 ，也就是说如果d是512，那么就需要遍历256次，每次上面两个公式要同时计算，如果是偶数维度就计算$PE(pos, 2i) ，如 果 是 奇 数 维 度 就 计 算 PE(pos, 2i + 1) $，这里举个简单的例子，假设模型维度 `d=4`（简化计算），则 `i` 的取值范围为 `0` 到 `1`（因 `d/2=2`）。以位置 `pos=1` 为例，当i为0时，2i=0为偶数，计算$ sin()，结 果 大 概 是 0.8415，2i + 1 = 1为 奇 数 ，计 算 cos()，结 果 大 概 是 0.5403，当 i 为 1时 ，2i = 2为 偶 数 ，计 算 sin()，结 果 大 概 是 0.01，2i + 1 = 3为 奇 数 ，计 算 cos()$，结果大概是0.999，最终位置编码向量：PE(1) = [0.8415, 0.5403, 0.0100, 0.9999]，以此类推PE(2) =[0.9093，-0.4161， 0.0200， 0.9998] ，通过上面的计算可以看出来，i 增大时，分母项 100002i /d 指数增长，导致波长从短到长变化，低维度时，短波长位置变化敏感可以捕捉局部位置差异，比如相邻词顺序，高维度时，长波长位置变化缓慢可以捕捉全局位置特征。
这种引入正余弦函数的方法比较符合理想位置编码的核心要求，首先，通过计算之后，每个位置生成了唯一的编码向量，符合唯一性，第二，依据三角函数的基本性质，可以得知第pos+k个位置编码是第pos个位置编码的线性组合，这就意味着位置编码中蕴含着单词之间的距离信息。公式表达如下，其中M(k)是一个变换矩阵： P E (p o s + k ) = P E (p o s ) + M (k )
计算得到Word Embedding和Position Embedding后，因为这两个Embedding的维度一致，所以可以直接将其相加，得到最后的Embedding结果。为什么这种结合有效？第一，位置编码作为低频信号，词嵌入作为高频信号，相加后形成多频复合信号，第二，相加操作等价于将Token ID、位置ID分别One-hot后拼接，再通过单层全连接层，第三，位置编码为词向量提供“坐标轴”，使自注意力能计算词间相对位置。
以上就是Transformer的Embedding的整个过程。位置编码的代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 class PositionalEncoder (nn.Module): def __init__ (self, d_model, max_seq_len = 80 ): super ().__init__() self .d_model = d_model 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 ** (i/d_model))) pe[pos, i + 1 ] = math.cos(pos / (10000 ** (i/d_model))) 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 + Variable(self .pe[:,:seq_len], requires_grad=False ).cuda() return x
注意力层
人类大脑在处理视觉信息时，会优先关注场景中的显著区域，而去忽略冗余背景，深度学习中的注意力机制就是模拟这种人类聚焦能力的计算技术，通过动态分配权重突出输入数据中的关键信息，从而提升模型处理效率与精度。在深度学习中，有非常多的注意力机制，比如说cv中的通道注意力机制、空间注意力等，这些都用来关注到最重要的区域上，而在自然语言处理中，注意力机制的核心目标是动态分配计算资源，使模型能够聚焦输入序列中与当前任务最相关的部分，解决传统序列模型（如RNN/LSTM）的长程依赖和固定表示瓶颈问题。
一些常用的机器翻译模型都是Seq2Seq结构，编码器将输入文本的特征提取后会送入到解码器中，但是只会将最后一个隐藏层的特征传递给解码器，他们之间的唯一联系就是下图中固定长度的语义向量c。
img
这种方式会让模型只关注原句子末尾的语义特征而遗忘其他的位置的语义，使得整体的效果下降，序列越长这种现象就越严重，为了缓解这个问题，一些研究人员引入了注意力机制，拿最经典的soft attention来说，他的核心思想就是让解码器不再完全依赖编码器的最后一层，而是想办法让解码器关注到编码器前面的一些隐藏层输出，那如何做呢，在翻译任务中，虽然句子之间的语法和单词以及序列的长度不同，但是他们之间总会有语义对应的地方，比如说“我想喝杯咖啡”可以翻译成“I want a cup of coffee”，其中我和i对应，想和want对应，咖啡和coffee对应，这些词的语义是有相似性的，注意力机制就是通过计算查询向量Q（编码器的隐状态）与各键K（解码器的隐状态）的相似度得到权重，再用Softmax归一化后对值V（解码器的隐状态）加权求和，实现输入信息的动态聚焦。也就是将下面的公式： $$
\begin{equation}
\begin{aligned}
&\mathbf{y}_{1} = \mathbf{f}(\mathbf{C}) \\
&\mathbf{y}_{2} = \mathbf{f}\left(\mathbf{C}, \mathbf{y}_{1}\right) \\
&\mathbf{y}_{3} = \mathbf{f}\left(\mathbf{C}, \mathbf{y}_{1}, \mathbf{y}_{2}\right)
\end{aligned}
\end{equation}
$$ $$
\begin{equation}
\begin{aligned}
&\mathbf{y}_{1} = \mathbf{f1}\left(\mathbf{C}_{1}\right) \\
&\mathbf{y}_{2} = \mathbf{f1}\left(\mathbf{C}_{2}, \mathbf{y}_{1}\right) \\
&\mathbf{y}_{3} = \mathbf{f1}\left(\mathbf{C}_{3}, \mathbf{y}_{1}, \mathbf{y}_{2}\right) % 修正逗号位置错误
\end{aligned}
\end{equation}
$$
原先解码器在计算输出时，完全以来编码器的隐状态C C 1 , C 2 , C 3
e t , i s t − 1T h i s t − 1h i i $$
α_{t,i} = \frac{exp(e_{i,j})}{\sum_{j=1}^{T}exp(e_{i,j})}
$$ Softmax 函数 沿输入序列长度维度（T）归一化。第三步，计算上下文向量（Context Vector），对编码器隐藏状态加权求和，生成动态上下文信息： $$
c_{t} = \sum^{T}_{i=1}α_{t,i}·h_{i}
$$ α t , i h i i c t
但是这种传统的注意力机制也存在着一系列。首先，传统的Attention机制依赖于编码器序列和解码器序列两个独立序列，只能计算跨序列关系，无法建模统一序列内部元素之间的依赖（如橘子中词与词之间的语法关系）。第二点，在RNN/LSTM+Attention架构中，编码器需逐步传递序列信息 ，导致远端位置信息衰减（梯度消失）。第三点，计算效率低，难以并行化，传统注意力的计算需严格依赖编码器顺序输出 ，解码过程是串行的。第四点，位置信息表达不足，注意力权重仅依赖内容相似性 （如词向量接近则权重高），但忽略位置信息 。
2017年google团队提出了Transfomer，其中的自注意力机制对上述的问题进行了改进和优化。它的计算过程是这样的，首先，将由单词语义嵌入及其位置编码叠加得到的输入表示为{x i R d i L ，其中d q i k i v i W Q R d × d q W K R d × d k W V R d × d v x i q i R d q k i R d k v i R d v x i R d i L $$
Q = XW^Q \\
K = XW^K \\
V = XW^V \\
$$
image-20250614171755220
为了得到编码单词x i i q i k 1 , q i k 2 , · · ·,q i k t $\sqrt{q}$ 以稳定优化。放缩后的 得分经过Softmax归一化为概率，与其他位置的值向量相乘来聚合希望关注的上下文信息，并最 小化不相关信息的干扰。上述计算过程可以被形式化地表述如下： $$
Z = Attention(Q, K, V) = Softmax(\frac{QK^T}{\sqrt{d}}V)
$$ Q ∈ R L × d q K ∈ R L × d k V ∈ R L × d v q , k , v L Z ∈ R L × d v 多头注意力机制，以关注上下文的不同侧面。具体来说，上下文中每一个单词的表示x i $$
在此基础上，将所有的$Z_i$进行拼接，然后经过线性变换$W_O∈R^{(Nd_v)×d}$来综合不同子空间中的上下文表示并形成注意力层，最终的输出$\{ x_i \in \mathbb{R}^d \}^{L}_{i=1}$。多头自注意力（Multi-HeadSelf-Attention）可表示为：
$$ 任意其他位置 ，解决了传统注意力机制只关注局部而无法关注全局的问题，彻底解决长距离依赖问题。第三，所有位置的注意力权重通过矩阵乘法一次性计算 ，可以并行计算，效率远超RNN的单步传递。第四，引入了位置编码，解决了位置信息表达不足的问题。其代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 import torchimport torch.nn as nnimport torch.nn.functional as Fimport mathclass MultiHeadAttention (nn.Module): def __init__ (self, heads, d_model, dropout=0.1 ): super ().__init__() assert d_model % heads == 0 , "d_model must be divisible by heads" self .d_model = d_model self .d_k = d_model // heads self .h = heads self .q_linear = nn.Linear(d_model, d_model) self .k_linear = nn.Linear(d_model, d_model) self .v_linear = nn.Linear(d_model, d_model) self .dropout = nn.Dropout(dropout) self .out = nn.Linear(d_model, d_model) def attention (self, q, k, v, mask=None ): """计算缩放点积注意力""" scores = torch.matmul(q, k.transpose(-2 , -1 )) / math.sqrt(self .d_k) if mask is not None : scores = scores.masked_fill(mask == 0 , -1e9 ) attn_weights = F.softmax(scores, dim=-1 ) attn_weights = self .dropout(attn_weights) if self .dropout else attn_weights output = torch.matmul(attn_weights, v) return output, attn_weights def forward (self, q, k, v, mask=None ): batch_size = q.size(0 ) q = self .q_linear(q) k = self .k_linear(k) v = self .v_linear(v) q = q.view(batch_size, -1 , self .h, self .d_k).transpose(1 , 2 ) k = k.view(batch_size, -1 , self .h, self .d_k).transpose(1 , 2 ) v = v.view(batch_size, -1 , self .h, self .d_k).transpose(1 , 2 ) scores, attn_weights = self .attention(q, k, v, mask) scores = scores.transpose(1 , 2 ).contiguous() concat = scores.view(batch_size, -1 , self .d_model) output = self .out(concat) return output, attn_weights
残差连接与LayerNorm
Transformer的并行结构加上多块堆叠，导致其网络结构一般都非常庞大，它的每一层中都包含复杂的非线性映射，这就导致模型的训练比较困难。为了解决这个问题，Transfomer引入了残差连接和层归一化技术，以进一步提升训练的稳定性。残差网络被大量用在深度网络网络中，它首先解决了梯度消失的问题，支持像Transfomer这种非常深的网络的训练，第二点是保留原始输入信息，降低学习难度，因为深层网络中一些原始特征经多次非线性变换易被扭曲或丢失，通过残差结构可以强制模型仅学习输入与输出间的差异，而非完整映射，第三就是避免在优化过程中因网络过深而产生潜在的梯度消失问题，并且可以加速模型训练收敛，提升稳定性，其公式如下： x l + 1f (x l x l x l l f ( · )
再说LayerNorm，归一化是一种常用的数据预处理技术，旨在将不同量纲或取值范围的数据转换到相同的尺度上，通常是 [0, 1] 或 [-1, 1]。这一过程可以消除特征间的数量级差异，从而提高数据分析和机器学习模型的性能。常用的归一化方法有BatchNorm、LayerNorm和GroupNorm等，其中BatchNorm在CV任务中被大量运用，它是在Batch维度上对数据进行归一化，也就是说把多个图像样本中相同特征通道看做一个分布进行归一化，比如说一个Batch中有四张特征图，这四张特征图经过几层卷积之后，变成了一个通道数为32的特征图，那么这个归一化过程会将四个样本第一个特征图进行归一化，同理，再将四个样本第二个特征图进行归一化，以此类推，如下图所示。
BN1示意图
但是这种方式就很难用到自然语言处理任务，因为在CV任务中，一个batch中的特征图通过相同的卷积层之后，他们的通道数始终是一致的，但是NLP任务中每个样本句子的长度不一致，一些短的句子一般情况下会进行数据填充，如果在Batch维度上进行归一化，也就是把一个batch中不同句子但是相同位置的词进行归一化，这些填充的词就会对归一化进行干扰。比如说有两个句子，“我只能永远读着对白“，”读着我对你的伤害”，假如进行分词后是这样的：[“我”，“只能”，“永远”，“读着”， “对白”]，[“读着”, “我”, “对”, “你”, “的”, “伤害”]，两个句子的长度不一样，[“我”，“只能”，“永远”，“读着”， “对白”]会填充成[“我”，“只能”，“永远”，“读着”， “对白”, ""]，之后会token化，然后再通过embedding转成向量，如果使用BatchNorm，那么会将前面句的”我“对应的向量和后面句子的”读着”对应的向量进行归一化，将前面句的“只能”对应的向量和后面句子的“我”对应的向量进行归一化，以此类推，但是最后一个会将前句填充的“”对应的向量和后句的“伤害”对应的向量进行归一化，这显然是不合理的，会干扰整个归一化过程，所以Transfomer中没有BatchNorm，而是用的LayerNorm，LayerNorm的原理就是对单一样本的所有特征 计算均值和方差，也就是将句子内部所有词对应的向量进行归一化操作，如下图所示：
请添加图片描述
这样就合理多了，其公式如下： $$
LN(x) = α·\frac{x-μ}{σ}+b
$$ µ σ α b
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 import torchimport torch.nn as nnclass Norm (nn.Module): def __init__ (self, d_model: int , eps: float = 1e-6 ) -> None : """ 层归一化模块 参数: d_model: 输入特征维度 eps: 数值稳定系数，防止除零错误 """ super ().__init__() self .size = d_model self .eps = eps self .alpha = nn.Parameter(torch.ones(self .size)) self .bias = nn.Parameter(torch.zeros(self .size)) def forward (self, x: torch.Tensor ) -> torch.Tensor: """ 前向传播计算 参数: x: 输入张量 (batch_size, ..., d_model) 返回: 归一化后的张量 """ mean = x.mean(dim=-1 , keepdim=True ) std = x.std(dim=-1 , keepdim=True , unbiased=False ) normalized = (x - mean) / (std + self .eps) return self .alpha * normalized + self .bia
前馈层
为了增强Transfomer的特征提取能力和表达能力，Transfomer还在自注意力层之后增加了一个前馈层，也就是FFN（Feed-Forward Network）层，它的本质就是一个带有 ReLU 激活函数的两层全连接网络，对自注意力层的输入进行更复杂的非线性变换，它注意力机制互补，分工协作，对注意力机制筛选的重要信息进行进一步感知，实验证明，这一非线性变换会对模型最终的性能产生重要的影响，其公式如下： F F N (x ) = R e L U (x W 1 + b 1 )W 2 + b 2 W 1 b 1 W 2 b 2 增大前馈子层隐状态的维度有利于提高最终翻译结果的质量，因此，前馈子层隐状态的维度一般比自注意力子层要大。其代码如下。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 import torchimport torch.nn as nnimport torch.nn.functional as Fclass FeedForward (nn.Module): """位置感知的前馈神经网络模块（Position-wise Feed-Forward Network） Transformer 架构中的核心组件，通过非线性变换增强模型表达能力。 结构：线性升维 → ReLU激活 → Dropout → 线性降维 参数： d_model: 输入/输出的特征维度（通常512/768） d_ff: 前馈层内部隐藏层维度（默认2048，推荐d_model的4倍） dropout: Dropout比率（默认0.1） """ def __init__ (self, d_model: int , d_ff: int = 2048 , dropout: float = 0.1 ) -> None : super ().__init__() self .linear_1 = nn.Linear(d_model, d_ff) self .linear_2 = nn.Linear(d_ff, d_model) self .dropout = nn.Dropout(dropout) def forward (self, x: torch.Tensor ) -> torch.Tensor: """前向传播过程 参数： x: 输入张量，形状为 [batch_size, seq_len, d_model] 返回： 输出张量，形状与输入相同 """ x = F.relu(self .linear_1(x)) x = self .dropout(x) return self .linear_2(x)
编码器解码器结构
Transfomer的编码器和解码器功能是不同的，编码器端主要用于编码源语言序列的信息，而这个序列是完全已知的，因而编码器仅需要考虑如何融合上下文语义信息就可以。解码器端则负责生成目标语言序列，这一生成过程是自回归的，即对于每一个单词的生成过程，仅有当前单词之前的目标语言序列是可以被观测的，因此这一额外增加的掩码是用来掩盖后续的文本信息的，以防模型在训练阶段直接看到后续的文本序列，进而无法得到有效的训练。
此外，解码器端额外增加了一个多头交叉注意力（Multi-Head Cross-Attention）模块，使用交叉注意力（Cross-Attention）方法，同时接收来自编码器端的输出和当前 Transformer 块的前一个掩码注意力层的输出 。查询是通过解码器前一层的输出进行投影的，而键和值是使用编码器的输出进行投影的。它的作用是在翻译的过程中，为了生成合理的目标语言序列，观测待翻译的源语言序列是什么。基于上述编码器和解码器结构，待翻译的源语言文本经过编码器端的每个 Transformer块对其上下文语义进行层层抽象，最终输出每一个源语言单词上下文相关的表示。解码器端以自回归的方式生成目标语言文本，即在每个时间步 t，根据编码器端输出的源语言文本表示，以及前t − 1
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 import torchimport torch.nn as nnimport torch.nn.functional as Fclass EncoderLayer (nn.Module): """Transformer 编码器层，包含多头自注意力和前馈网络子层 设计要点： - 采用 Pre-LN 结构（层归一化在子层前）提升训练稳定性 - 分离注意力掩码逻辑，支持填充掩码(Padding Mask)和因果掩码(Causal Mask) - 残差连接缓解梯度消失，Dropout 增强泛化能力 参数： d_model: 特征维度 (默认512) heads: 注意力头数 dropout: 随机失活比率 (默认0.1) """ def __init__ (self, d_model: int , heads: int , dropout: float = 0.1 ): super ().__init__() self .norm_attn = nn.LayerNorm(d_model) self .attn = MultiHeadAttention(heads, d_model, dropout) self .dropout_attn = nn.Dropout(dropout) self .norm_ffn = nn.LayerNorm(d_model) self .ffn = FeedForward(d_model, dropout=dropout) self .dropout_ffn = nn.Dropout(dropout) def forward (self, x: torch.Tensor, mask: torch.Tensor = None ) -> torch.Tensor: """前向传播流程 参数： x: 输入张量 [batch, seq_len, d_model] mask: 注意力掩码 [batch, seq_len] 或 [batch, seq_len, seq_len] 返回： 编码后的张量 [batch, seq_len, d_model] """ residual = x x = self .norm_attn(x) attn_output = self .attn(x, x, x, mask) x = residual + self .dropout_attn(attn_output) residual = x x = self .norm_ffn(x) ffn_output = self .ffn(x) x = residual + self .dropout_ffn(ffn_output) return x class Encoder (nn.Module): """Transformer 编码器堆栈，包含嵌入层、位置编码和N层编码器 关键设计： - 支持动态序列长度，通过位置编码注入位置信息 - 堆叠N个相同结构的编码器层实现深度特征提取 - 最终输出层归一化稳定特征分布 参数： vocab_size: 词表大小 d_model: 特征维度 N: 编码器层堆叠数量 heads: 注意力头数 dropout: 全局Dropout比率 """ def __init__ (self, vocab_size: int , d_model: int , N: int , heads: int , dropout: float ): super ().__init__() self .embed = nn.Embedding(vocab_size, d_model) self .pos_encoder = PositionalEncoder(d_model, dropout) self .layers = nn.ModuleList([ EncoderLayer(d_model, heads, dropout) for _ in range (N) ]) self .norm = nn.LayerNorm(d_model) def forward (self, src: torch.Tensor, mask: torch.Tensor = None ) -> torch.Tensor: """编码器前向传播 参数： src: 输入词索引 [batch, seq_len] mask: 序列掩码 [batch, seq_len] 返回： 上下文感知的特征表示 [batch, seq_len, d_model] """ x = self .embed(src) x = self .pos_encoder(x) for layer in self .layers: x = layer(x, mask) return self .norm(x)
解码器代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 import torchimport torch.nn as nnimport torch.nn.functional as Fclass DecoderLayer (nn.Module): """Transformer解码器层，包含自注意力、交叉注意力和前馈网络 关键组件： - Masked Multi-Head Self-Attention：防止未来信息泄露 - Encoder-Decoder Cross-Attention：融合编码器上下文 - Position-wise Feed-Forward Network：非线性特征变换 - 残差连接+层归一化：稳定训练过程 参数： d_model: 特征维度 (默认512) heads: 注意力头数 dropout: Dropout比率 (默认0.1) """ def __init__ (self, d_model: int , heads: int , dropout: float = 0.1 ): super ().__init__() self .norm_1 = nn.LayerNorm(d_model) self .attn_1 = MultiHeadAttention(heads, d_model, dropout=dropout) self .dropout_1 = nn.Dropout(dropout) self .norm_2 = nn.LayerNorm(d_model) self .attn_2 = MultiHeadAttention(heads, d_model, dropout=dropout) self .dropout_2 = nn.Dropout(dropout) self .norm_3 = nn.LayerNorm(d_model) self .ff = FeedForward(d_model, dropout=dropout) self .dropout_3 = nn.Dropout(dropout) def forward ( self, x: torch.Tensor, e_outputs: torch.Tensor, src_mask: torch.Tensor, trg_mask: torch.Tensor ) -> torch.Tensor: """ 前向传播流程 参数： x: 解码器输入 [batch, trg_len, d_model] e_outputs: 编码器输出 [batch, src_len, d_model] src_mask: 源序列掩码 [batch, 1, src_len] trg_mask: 目标序列掩码 [batch, trg_len, trg_len] 返回： 解码特征 [batch, trg_len, d_model] """ residual = x attn_output_1 = self .attn_1(x, x, x, trg_mask) x = residual + self .dropout_1(attn_output_1) x = self .norm_1(x) residual = x attn_output_2 = self .attn_2(x, e_outputs, e_outputs, src_mask) x = residual + self .dropout_2(attn_output_2) x = self .norm_2(x) residual = x ff_output = self .ff(x) x = residual + self .dropout_3(ff_output) x = self .norm_3(x) return x class Decoder (nn.Module): """Transformer解码器堆栈，包含嵌入层、位置编码和N层解码器 关键功能： - 词嵌入+位置编码：将离散token转化为连续向量 - 堆叠N个DecoderLayer：逐步生成目标序列 - 输出层归一化：稳定特征分布 参数： vocab_size: 目标词表大小 d_model: 特征维度 N: 解码器层数 heads: 注意力头数 dropout: 全局Dropout比率 """ def __init__ ( self, vocab_size: int , d_model: int , N: int , heads: int , dropout: float ): super ().__init__() self .N = N self .embed = nn.Embedding(vocab_size, d_model) self .pe = PositionalEncoder(d_model, dropout=dropout) self .layers = nn.ModuleList([ DecoderLayer(d_model, heads, dropout) for _ in range (N) ]) self .norm = nn.LayerNorm(d_model) def forward ( self, trg: torch.Tensor, e_outputs: torch.Tensor, src_mask: torch.Tensor, trg_mask: torch.Tensor ) -> torch.Tensor: """ 解码器前向传播 参数： trg: 目标序列 [batch, trg_len] e_outputs: 编码器输出 [batch, src_len, d_model] src_mask: 源序列掩码 [batch, 1, src_len] trg_mask: 目标序列掩码 [batch, trg_len, trg_len] 返回： 解码结果 [batch, trg_len, d_model] """ x = self .embed(trg) x = self .pe(x) for layer in self .layers: x = layer(x, e_outputs, src_mask, trg_mask) return self .norm(x)
Transformer的整体结构代码如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 import torchimport torch.nn as nnclass Transformer (nn.Module): """Transformer 序列到序列模型 结构组成： - 编码器堆栈：提取源序列特征 - 解码器堆栈：生成目标序列 - 线性输出层：映射到目标词表空间 参数： src_vocab: 源语言词表大小 trg_vocab: 目标语言词表大小 d_model: 特征维度 (默认512) N: 编码器/解码器堆叠层数 (默认6) heads: 注意力头数 (默认8) dropout: Dropout比率 (默认0.1) """ def __init__ ( self, src_vocab: int , trg_vocab: int , d_model: int = 512 , N: int = 6 , heads: int = 8 , dropout: float = 0.1 ) -> None : super ().__init__() self .encoder = Encoder(src_vocab, d_model, N, heads, dropout) self .decoder = Decoder(trg_vocab, d_model, N, heads, dropout) self .out = nn.Linear(d_model, trg_vocab) def forward ( self, src: torch.Tensor, trg: torch.Tensor, src_mask: torch.Tensor, trg_mask: torch.Tensor ) -> torch.Tensor: """ 前向传播流程 参数： src: 源序列 [batch, src_len] trg: 目标序列 [batch, trg_len] src_mask: 源序列掩码 [batch, 1, src_len] trg_mask: 目标序列掩码 [batch, trg_len, trg_len] 返回： 预测概率分布 [batch, trg_len, trg_vocab] """ e_outputs = self .encoder(src, src_mask) d_output = self .decoder(trg, e_outputs, src_mask, trg_mask) return self .out(d_output)
模型的训练代码：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 import torchimport torch.nn as nnimport torch.nn.functional as Fimport timefrom torchtext.data import Field, BucketIteratorfrom model import Transformer device = torch.device('cuda' if torch.cuda.is_available() else 'cpu' ) EN_TEXT = Field(tokenize='spacy' , tokenizer_language='en' , lower=True , init_token='<sos>' , eos_token='<eos>' , batch_first=True ) FR_TEXT = Field(tokenize='spacy' , tokenizer_language='fr' , lower=True , init_token='<sos>' , eos_token='<eos>' , batch_first=True ) d_model = 512 heads = 8 N = 6 src_vocab_size = len (EN_TEXT.vocab) trg_vocab_size = len (FR_TEXT.vocab) source_pad_idx = EN_TEXT.vocab.stoi['<pad>' ] target_pad_idx = FR_TEXT.vocab.stoi['<pad>' ] model = Transformer(src_vocab_size, trg_vocab_size, d_model, N, heads).to(device) for p in model.parameters(): if p.dim() > 1 : nn.init.xavier_uniform_(p) optim = torch.optim.Adam( model.parameters(), lr=0.0001 , betas=(0.9 , 0.98 ), eps=1e-9 ) def create_masks (src: torch.Tensor, trg: torch.Tensor ) -> tuple : """ 创建源序列掩码和目标序列掩码 参数: src: 源序列张量 [batch_size, src_len] trg: 目标序列张量 [batch_size, trg_len] 返回: src_mask: 源序列掩码 [batch_size, 1, 1, src_len] trg_mask: 目标序列掩码 [batch_size, 1, trg_len, trg_len] """ src_mask = (src != source_pad_idx).unsqueeze(1 ).unsqueeze(2 ) trg_len = trg.shape[1 ] trg_pad_mask = (trg != target_pad_idx).unsqueeze(1 ).unsqueeze(2 ) nopeak_mask = torch.tril(torch.ones(1 , trg_len, trg_len), diagonal=0 ).bool ().to(device) trg_mask = trg_pad_mask & nopeak_mask return src_mask, trg_mask def train_model (epochs: int , print_every: int = 100 ): """ Transformer模型训练循环 参数: epochs: 训练轮数 print_every: 每隔多少批次打印一次状态 """ model.train() start = time.time() temp = start total_loss = 0 train_iter = BucketIterator.splits( (train_data,), batch_size=64 , device=device, sort_within_batch=True , sort_key=lambda x: len (x.src) )[0 ] for epoch in range (epochs): for i, batch in enumerate (train_iter): src = batch.English.to(device) trg = batch.French.to(device) trg_input = trg[:, :-1 ] targets = trg[:, 1 :] src_mask, trg_mask = create_masks(src, trg_input) preds = model(src, trg_input, src_mask, trg_mask) loss = F.cross_entropy( preds.view(-1 , preds.size(-1 )), targets.contiguous().view(-1 ), ignore_index=target_pad_idx ) optim.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0 ) optim.step() total_loss += loss.item() if i % print_every == 0 : avg_loss = total_loss / print_every elapsed = time.time() - temp print (f"Epoch: {epoch+1 :02d} | Batch: {i:04d} | " f"Loss: {avg_loss:.4 f} | Time/Batch: {elapsed:.2 f} s" ) total_loss = 0 temp = time.time() torch.save({ 'epoch' : epoch, 'model_state_dict' : model.state_dict(), 'optimizer_state_dict' : optim.state_dict(), 'loss' : loss.item(), }, f'transformer_epoch_{epoch+1 } .pt' ) print (f"Total Training Time: {(time.time()-start)/60 :.2 f} minutes" ) if __name__ == "__main__" : train_model(epochs=10 , print_every=100 )
