代码解读
第一部分:Tokenizer 类解读
该类主要功能是封装 Google 的 SentencePiece 库,为大型语言模型(如 LLaMa)提供一个标准化的文本编码(tokenization)和解码(decoding)接口。
核心功能总结:
- 初始化 (
__init__): 加载一个预先训练好的 SentencePiece 模型文件。这个模型定义了词汇表(vocabulary)和分词规则。同时,它会获取并存储一些特殊的 token ID,如句子开头(BOS)、句子结尾(EOS)和填充(PAD)的 ID。 - 编码 (
encode): 将一个普通的字符串(人类可读的文本)转换成一串数字 ID(模型可读的格式)。它还提供了选项,可以在序列的开头和结尾添加特殊的BOS和EOS标记,这对于训练语言模型至关重要。 - 解码 (
decode): 将模型生成的数字 ID 序列转换回人类可读的字符串。
这个 Tokenizer 是任何基于 Transformer 的自然语言处理模型都不可或缺的基础组件。
import os
from sentencepiece import SentencePieceProcessor
from logging import getLogger
from typing import List
# 获取日志记录器
logger = getLogger()
class Tokenizer:
"""
一个使用 SentencePiece 模型进行文本编码和解码的分词器类。
这个类封装了 SentencePieceProcessor,并提供了方便的方法来处理文本和 token ID 之间的转换。
"""
def __init__(self, model_path: str):
"""
初始化 Tokenizer 类。
Args:
model_path (str): 指向 SentencePiece 模型文件的路径。
"""
# 重新加载分词器模型
# 断言:确保模型文件存在于指定路径,否则抛出异常。
assert os.path.isfile(model_path), model_path
# 从指定路径加载预训练的 SentencePiece 模型。
self.sp_model = SentencePieceProcessor(model_file=model_path)
# 记录日志,表明模型已成功加载。
logger.info(f"Reloaded SentencePiece model from {model_path}")
# 获取词汇表大小和特殊 token 的 ID
# self.n_words: 词汇表中的单词(或 subword)总数。
self.n_words: int = self.sp_model.vocab_size()
# self.bos_id: "Beginning Of Sentence" (句子开头) 标记的 ID。
self.bos_id: int = self.sp_model.bos_id()
# self.eos_id: "End Of Sentence" (句子结尾) 标记的 ID。
self.eos_id: int = self.sp_model.eos_id()
# self.pad_id: 用于填充的 token 的 ID。
self.pad_id: int = self.sp_model.pad_id()
# 记录日志,输出词汇表大小和关键 token ID 的信息。
logger.info(
f"#words: {self.n_words} - BOS ID: {self.bos_id} - EOS ID: {self.eos_id}"
)
# 断言:确保词汇表大小与模型中的 piece 数量一致,这是一个完整性检查。
assert self.sp_model.vocab_size() == self.sp_model.get_piece_size()
def encode(self, s: str, bos: bool, eos: bool) -> List[int]:
"""
将输入的字符串编码为 token ID 列表。
Args:
s (str): 要编码的输入字符串。
bos (bool): 是否在编码结果的开头添加 BOS (句子开头) 标记。
eos (bool): 是否在编码结果的结尾添加 EOS (句子结尾) 标记。
Returns:
List[int]: 编码后的 token ID 列表。
"""
# 断言:确保输入是一个字符串类型。
assert type(s) is str
# 使用 SentencePiece 模型将字符串 s 编码成一个 token ID 列表。
t = self.sp_model.encode(s)
# 如果 bos 参数为 True,则在列表的开头插入 BOS ID。
if bos:
t = [self.bos_id] + t
# 如果 eos 参数为 True,则在列表的末尾追加 EOS ID。
if eos:
t = t + [self.eos_id]
# 返回最终的 token ID 列表。
return t
def decode(self, t: List[int]) -> str:
"""
将 token ID 列表解码回原始的字符串。
Args:
t (List[int]): 要解码的 token ID 列表。
Returns:
str: 解码后的字符串。
"""
# 使用 SentencePiece 模型将 token ID 列表解码成一个字符串。
return self.sp_model.decode(t)
第二部分:RMSNorm 类代码解读与注释
RMSNorm 是一种在 LLaMa 和其他现代 Transformer 模型中用于提升训练稳定性的归一化(Normalization)技术。它比更常见的 LayerNorm 计算上更高效。
下面是为您添加了详细中文注释的 RMSNorm 代码。
import torch
import torch.nn as nn
class RMSNorm(torch.nn.Module):
"""
均方根层归一化 (Root Mean Square Layer Normalization)。
为了提高训练稳定性,LLaMa 对每个 Transformer 子层的输入进行归一化,
而不是对输出进行归一化(这被称为 Pre-normalization)。
RMSNorm 是 LayerNorm 的一种变体,计算上更简单高效。
Args:
dim (int): 输入张量的最后一个维度的尺寸,即特征维度。
eps (float): 一个很小的数,添加到分母中以防止除以零。默认为 1e-6。
"""
def __init__(self, dim: int, eps: float = 1e-6):
"""
初始化 RMSNorm 层。
"""
# 调用父类 nn.Module 的构造函数。
super().__init__()
# 保存 epsilon 值,用于数值稳定性。
self.eps = eps
# 创建一个可学习的缩放参数 `weight`。
# 它的大小与特征维度 `dim` 相同,并初始化为全 1。
# 这个参数允许模型在归一化后学习如何重新缩放特征。
self.weight = nn.Parameter(torch.ones(dim))
def _norm(self, x: torch.Tensor) -> torch.Tensor:
"""
执行核心的 RMSNorm 计算。
公式为:x / sqrt(mean(x^2) + eps)
Args:
x (torch.Tensor): 输入张量。
Returns:
torch.Tensor: 归一化后的张量。
"""
# 1. x.pow(2): 对输入张量 x 的每个元素进行平方。
# 2. .mean(-1, keepdim=True): 沿着最后一个维度(特征维度)计算均值。
# `keepdim=True` 确保结果张量的维度与输入相同,以便进行广播。
# 3. + self.eps: 添加 epsilon 以防止分母为零。
# 4. torch.rsqrt(...): 计算平方根的倒数 (1 / sqrt(...)),这比直接除以 sqrt 在计算上更高效。
# 5. x * ...: 将原始输入 x 与计算出的缩放因子相乘,完成归一化。
return x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""
定义模块的前向传播逻辑。
Args:
x (torch.Tensor): 输入张量,通常形状为 (batch_size, sequence_length, feature_dim)。
Returns:
torch.Tensor: 经过 RMSNorm 处理后的输出张量。
"""
# 1. self._norm(x.float()): 将输入 x 转换为浮点数类型进行归一化计算,以保证数值精度。
# 得到归一化后的输出 `output`。
# 2. .type_as(x): 将计算结果的数据类型转换回原始输入 x 的数据类型(例如,如果是半精度浮点数)。
# 3. output * self.weight: 将归一化后的输出与可学习的缩放参数 `weight` 相乘。
# 这给了模型恢复或调整信号幅度的能力,增加了模型的表达能力。
output = self._norm(x.float()).type_as(x)
return output * self.weight
RMSNorm 解读总结:
-
为什么需要归一化?
- 在深度神经网络中,每一层的输出都会成为下一层的输入。随着网络层数的加深,输入的分布可能会发生剧烈变化(称为“内部协变量偏移”),这会减慢训练速度,使训练过程不稳定。
- 归一化层(如
LayerNorm或RMSNorm)的作用是将每层的输入或输出重新调整到一个标准的分布(例如,均值为0,方差为1),从而使训练更加稳定和快速。
-
RMSNorm和LayerNorm的区别是什么?- LayerNorm:
(x - mean(x)) / sqrt(variance(x) + eps) * gain + bias。它同时对输入进行中心化(减去均值)和缩放(除以标准差)。 - RMSNorm:
x / sqrt(mean(x^2) + eps) * gain。它只对输入进行缩放(通过均方根),省略了中心化步骤(减去均值)。 - 优势: LLaMa 的作者发现,省略中心化步骤对性能影响不大,但可以简化计算,从而在 GPU 上运行得更快。
- LayerNorm:
-
代码中的
self.weight是什么角色?self.weight是一个可学习的参数(也称为增益gain)。虽然归一化过程强制性地将数据的尺度调整了,但这可能会限制模型的表达能力。- 通过引入一个可学习的
weight,模型可以在训练过程中自己学习每个特征维度最合适的缩放比例。在极端情况下,如果模型发现归一化是有害的,它可以学习将weight设置为某个值,以部分或完全抵消归一化的影响。这增加了模型的灵活性。