transformer大模型有监督学习和生成原理记录
·
代码中每个model类都有个前向传播和反向传播。
而最外层的前向和反向是Transformer类中的。相关代码地址：【免费】transformer-有监督学习，带详细注释资源-CSDN下载
-
transformer 模型训练（有监督学习）
-
步骤
-
加载transformer配置
参数:
vocab_size (int): 词表大小，决定输入输出维度
- 小数据集建议: 100-1000
- 大数据集建议: 10000+
d_model (int): 模型维度，必须能被 n_heads 整除
- 小型模型: 64-128
- 中型模型: 256-512
- 大型模型: 768-1024
n_heads (int): 多头注意力头数
- 必须满足 d_model % n_heads == 0
d_ff (int): 前馈网络中间层维度，通常为 4*d_model
n_layers (int): Encoder 和 Decoder 的堆叠层数
max_seq_len (int): 最大序列长度（输入+输出）
dropout_rate (float): Dropout 丢弃概率，0-1之间
learning_rate (float): Adam 优化器学习率
beta1 (float): Adam 一阶矩衰减率
beta2 (float): Adam 二阶矩衰减率
epsilon (float): Adam 数值稳定小常数
epochs (int): 训练总轮数
batch_size (int): 每批样本数
max_grad_norm (float): 梯度裁剪最大范数
-
加载清理后的 成对的训练数据
-
将每个字符映射为唯一整数 ID，并补充特殊token
这几个特殊token是表示开始、结束的。还有填充字符，保证每个训练数据转成数组后的长度是一样的
# 特殊 token 定义
PAD_TOKEN = '<PAD>' # 填充用，ID=0
SOS_TOKEN = '<SOS>' # 序列开始 (Start Of Sequence)，ID=1
EOS_TOKEN = '<EOS>' # 序列结束 (End Of Sequence)，ID=2
UNK_TOKEN = '<UNK>' # 未知字符 (Unknown)，ID=3
# 特殊 token 对应 ID（固定，不可更改）
PAD_ID = 0
SOS_ID = 1
EOS_ID = 2
UNK_ID = 3-
准备训练数据
可以分成四部分。按分割符分成源数据和目标数据。 然后再把总的训练数据提取处理十几对作为测试数据, 如下面4个变量名:
# 3. 准备训练数据
src_train, tgt_train, src_test, tgt_test = prepare_batch_data(
texts, tokenizer, max_len=config.max_seq_len
)
print(f"训练样本: {len(src_train)}, 测试样本: {len(src_test)}")-
开始训练
主体代码：
def train(self, src_data, tgt_data, epochs=None, batch_size=None, verbose=True):
"""
完整训练循环
参数:
src_data (numpy.ndarray): 所有源序列，形状 (N, src_len)
示例: (100, 20) - 100 个样本，每个最长 20
tgt_data (numpy.ndarray): 所有目标序列，形状 (N, tgt_len)
示例: (100, 20)
epochs (int|None): 训练轮数，None 使用 config 中的值
batch_size (int|None): 批次大小，None 使用 config 中的值
verbose (bool): 是否打印进度
返回:
dict: 训练历史 {'loss': [...], 'epoch': [...]}
数据样例:
>>> history = trainer.train(src_data, tgt_data, epochs=10, batch_size=4)
>>> print(history['loss'])
[4.5, 3.2, 2.8, ...]
"""
epochs = epochs or self.config.epochs
#batch_size注意下，是每个批次处理的个数，不是批次是
batch_size = batch_size or self.config.batch_size
n_samples = len(src_data)
n_batches = max(1, n_samples // batch_size)
if verbose:
print(f"开始训练: {n_samples} 样本, {n_batches} 批次/轮, {epochs} 轮")
print(f"配置: d_model={self.config.d_model}, n_heads={self.config.n_heads}, "
f"layers={self.config.n_layers}")
start_time = time.time()
"""
先记录一下：这个训练过程 比如说epochs=50 训练50次
n_samples样本数是112，batch_size每次处理4个
经过indices 打乱了后，每个批次序列的样本在原样本中的顺序也乱了
"""
for epoch in range(1, epochs + 1):
epoch_losses = []
# 随机打乱数据 目的是防止模型‘背顺序’，让模型记住的是内容，而不是位置或者巧合
indices = np.random.permutation(n_samples)
for i in range(n_batches):
# batch_index是生成一组随机数组下标，供后面使用
batch_indices = indices[i * batch_size:(i + 1) * batch_size]
# 经过indices 打乱了后，每个批次序列的样本在原样本中的顺序也乱了
# src和tgt是双层数组，外层表示句子，里层表示每个句子包含的词对应的数字
src_batch = src_data[batch_indices]
tgt_batch = tgt_data[batch_indices]
# 有了原始和对比数据，开始训练
loss = self.train_step(src_batch, tgt_batch)
epoch_losses.append(loss)
# 记录本轮平均损失
avg_loss = np.mean(epoch_losses)
self.history['loss'].append(avg_loss)
self.history['epoch'].append(epoch)
# 打印日志
if verbose and (epoch == 1 or epoch % self.log_interval == 0 or epoch == epochs):
elapsed = time.time() - start_time
print(f"轮次 {epoch}/{epochs} | 损失: {avg_loss:.4f} | "
f"用时: {elapsed:.1f}s")
if verbose:
total_time = time.time() - start_time
print(f"\n训练完成! 总用时: {total_time:.1f}s")
print(f"最终损失: {self.history['loss'][-1]:.4f}")
return self.history
-
每次训练过程
def forward(self, src, tgt, src_mask=None, tgt_mask=None, training=True):
"""
完整前向传播
参数:
src (numpy.ndarray): 源序列，形状 (batch_size, src_len)
示例: [[1, 5, 3, 0, 0]] # batch=1, src_len=5
tgt (numpy.ndarray): 目标序列，形状 (batch_size, tgt_len)
示例: [[1, 7, 2, 0]] # batch=1, tgt_len=4
src_mask (numpy.ndarray|None): 源序列 padding mask
None 时自动创建
tgt_mask (numpy.ndarray|None): 目标序列组合掩码（padding + look-ahead）
None 时自动创建
training (bool): 是否训练模式
返回:
numpy.ndarray: 输出 logits，形状 (batch_size, tgt_len, vocab_size)
示例: (1, 4, 100) - 每个位置对每个词的分数
"""
# shape是获取数组长度
batch_size = src.shape[0]
src_len = src.shape[1]
tgt_len = tgt.shape[1]
# 自动创建 mask（如果未提供） 源数据掩码
if src_mask is None:
src_mask = create_padding_mask(src) # (B, 1, 1, src_len)
if tgt_mask is None: #目标数据掩码
# 组合掩码: padding mask + look-ahead mask
# 注意这里是给目标数据做掩码处理，防止看未来。
# create_padding_mask是屏蔽padding追加的无用字符
#create_look_ahead_mask 防止看未来位置，生成一个上三角矩阵，
# 包含对角线，0可见 1不可见
tgt_padding_mask = create_padding_mask(tgt) # (B, 1, 1, tgt_len)
look_ahead = create_look_ahead_mask(tgt_len) # (tgt_len, tgt_len)
# 扩展 look_ahead mask 到 batch 维度
# np.newaxis就是 None，作用是 增加一个新的维度（axis）。
# 非常典型且重要的 NumPy 广播（Broadcasting）操作
# 给 look_ahead这个二维矩阵，在前面“插”两个大小为 1 的维度，
# 使其形状从 (T, T)变成 (1, 1, T, T)，
# 以便和 (Batch, Heads, T, T)的 Attention Score 进行广播计算。
# 后面前向训练后，权重等维度也会跟这个一样
look_ahead = look_ahead[np.newaxis, np.newaxis, :, :] # (1, 1, tgt_len, tgt_len)
tgt_mask = np.maximum(tgt_padding_mask, look_ahead) # (B, 1, tgt_len, tgt_len)
# 1. 源序列嵌入 + 位置编码
# *self.scale是为了防止过大，其值前面说了是d_model的平方根
src_emb = self.src_embed.forward(src) * self.scale
src_emb = self.pos_encoding.forward(src_emb)
# 2. 目标序列嵌入 + 位置编码
tgt_emb = self.tgt_embed.forward(tgt) * self.scale
tgt_emb = self.pos_encoding.forward(tgt_emb)
# 3. Encoder
memory = self.encoder.forward(src_emb, src_mask, training)
# 4. Decoder
dec_output = self.decoder.forward(tgt_emb, memory, tgt_mask, src_mask, training)
# 5. 输出投影 -> logits
logits = dec_output @ self.W_out + self.b_out # (B, tgt_len, vocab_size)
# 缓存用于反向传播
self._cache = {
'src': src, 'tgt': tgt,
'src_emb': src_emb, 'tgt_emb': tgt_emb,
'memory': memory, 'dec_output': dec_output,
'src_mask': src_mask, 'tgt_mask': tgt_mask
}
return logits-
先打乱数据，防止背顺序
此时，数据的维度还是二维的，列是每个的词向量维度，也是赋予词的含义维度数据，常用64，128，512。行是训练数据拆成词之后的词的个数。
-
批量处理
因为训练数据很多。这里批量处理，处理更快，也能加大联想。
-
前向传播
- 创建填充掩码 (Padding Mask)
用于屏蔽填充位置（<PAD>），使模型不关注这些位置。
原理:
在注意力计算中，将被屏蔽位置的注意力分数设为 -inf（或极大负数），
使其经过 softmax 后权重趋近于 0。
def create_padding_mask(seq, pad_id=0):
"""
创建填充掩码 (Padding Mask)
用于屏蔽填充位置（<PAD>），使模型不关注这些位置。
参数:
seq (numpy.ndarray): 输入序列，形状 (batch_size, seq_len)
示例: [[1, 4, 5, 0, 0], # batch=2, seq_len=5
[1, 7, 8, 9, 0]]
pad_id (int): 填充 token 的 ID，默认 0
返回:
numpy.ndarray: 掩码矩阵，形状 (batch_size, 1, 1, seq_len)
值为 0（不屏蔽）或 1（屏蔽）
示例: [[[[0, 0, 0, 1, 1]]], # 最后两个是 PAD
[[[0, 0, 0, 0, 1]]]]
原理:
在注意力计算中，将被屏蔽位置的注意力分数设为 -inf（或极大负数），
使其经过 softmax 后权重趋近于 0。
"""
"""
这一步是升维操作，目的是为后续计算时跟后面的注意力分数：Q,K,V保持一致。好计算
距离看把，比如说sql是（3,5）
seq = np.array([
[1, 2, 3, 0, 0],
[4, 5, 0, 0, 0],
[6, 7, 8, 9, 0]
])
(seq == pad_id)→ bool mask：
[
[False, False, False, True, True],
[False, False, True, True, True],
[False, False, False, False, True]
]
.astype(np.float32)→ 数值 mask
[
[0., 0., 0., 1., 1.],
[0., 0., 1., 1., 1.],
[0., 0., 0., 0., 1.]
]
升维后：mask = mask.reshape(3, 1, 1, 5)
[
# batch 0
[
[ # 第1个 new axis
[0., 0., 0., 1., 1.] # 第2个 new axis
]
],
# batch 1
[
[
[0., 0., 1., 1., 1.]
]
],
# batch 2
[
[
[0., 0., 0., 0., 1.]
]
]
]
Returns:
_type_: _description_
"""
# (batch_size, seq_len) -> (batch_size, 1, 1, seq_len)
# 这行代码要从左往右看，(seq == pad_id)是把数组转成bool数组，等于pad_id的元素维true
# 然后把bool值转成float型，最后升维，连升两次
mask = (seq == pad_id).astype(np.float32)[:, np.newaxis, np.newaxis, :]
return mask-
给目标数据做掩码处理，防止看未来
就是一个上三角的矩阵，包含对角线，表示 0可见 1不可见。
-
源序列嵌入 + 位置编码
即词嵌入层和位置编码层
前面我们有个把训练数据打乱的操作。这里相当于还原了。但是把词和位置关系分开了。
什么是词嵌入层？
官方解释：把一个离散的、无结构的符号，嵌入到一个连续的、有结构的向量空间中
让语义变成了距离，让关系变成了方向。
我们录入的词是很多孤立的符号，而嵌入就是把这些词投射到三维空间中，
语义相近的词靠的近，相似的词方向一致也就是这个词在不在你这个句子中，它都有其自己的意思。也就是就独立的词而言，其本身之间就是有距离关系的。
class Embedding:
"""
词嵌入层 (Embedding Layer)
理解下什么是嵌入层？
官方解释：把一个离散的、无结构的符号，嵌入到一个连续的、有结构的向量空间中
让语义变成了距离，让关系变成了方向。
我们录入的词是很多孤立的符号，而嵌入就是把这些词投射到三维空间中，
语义相近的词靠的近，相似的词方向一致
将整数 token ID 映射为 d_model 维的稠密向量。
使用查找表（权重矩阵）实现。
权重形状: (vocab_size, d_model)
示例:
>>> emb = Embedding(vocab_size=100, d_model=64)
>>> input_ids = np.array([[1, 5, 3]]) # batch=1, seq_len=3
>>> output = emb.forward(input_ids)
>>> output.shape
(1, 3, 64)
"""
def __init__(self, vocab_size, d_model, seed=None):
"""
初始化嵌入层
参数:
vocab_size (int): 词表大小 V
d_model (int): 嵌入维度 D
seed (int|None): 随机种子，用于可复现
返回: Embedding 实例
权重初始化:
使用 Xavier/Glorot 初始化: N(0, sqrt(2/(vocab_size + d_model)))
帮助梯度在前向/反向传播中保持稳定。
"""
self.vocab_size = vocab_size
self.d_model = d_model
if seed is not None:
np.random.seed(seed)
# 权重矩阵: (vocab_size, d_model)
# Xavier 初始化
limit = np.sqrt(6.0 / (vocab_size + d_model))
self.W = np.random.uniform(-limit, limit, (vocab_size, d_model)).astype(
np.float32
)
# 梯度缓存
# zeros_like是啥？ 先看zeros，np.zeros(3)->array([0., 0., 0.]) 也就是创建一个全是0的数组
# zeros_like也是一样的， np.zeros_like(self.W) 等价于 np.zeros(self.W.个数)
self.dW = np.zeros_like(self.W)
self._input_ids = None # 缓存输入用于反向传播
def forward(self, x):
"""
前向传播: 将 token IDs 映射为向量
参数:
x (numpy.ndarray): token ID 数组，形状 (batch_size, seq_len)
示例: [[1, 5, 3, 0]] # 0 是 PAD
返回:
numpy.ndarray: 嵌入向量，形状 (batch_size, seq_len, d_model)
示例: (1, 4, 64) 的浮点数组
self.w[x] 很难让人看懂啊，w,x都是数组，怎么从根据x查w呢？
W[x]的结果，就是把 x中的每个 token id，替换成对应的那一行动量。
举例说明：
x = [
[1, 0, 2],
[3, 1, 4]
]
Step 2：逐个 id 查表
token id 取哪一行
1 W[1]
0 W[0]
2 W[2]
3 W[3]
1 W[1]
4 W[4]
result = self.W[x]
# 等价于
result = [
[ W[1], W[0], W[2] ],
[ W[3], W[1], W[4] ]
]
"""
self._input_ids = x # 缓存用于反向传播
# 通过索引查找: W[x] 形状为 (batch_size, seq_len, d_model)
return self.W[x]
def backward(self, d_out):
"""
反向传播: 计算梯度并更新权重
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, seq_len, d_model)
返回:
numpy.ndarray: 输入梯度（无需返回，Embedding 是第一层）
梯度计算:
将 d_out 累加到对应 token ID 的权重行上
"""
# 清零梯度
self.dW.fill(0)
# 将梯度累加到对应位置
batch_size, seq_len = self._input_ids.shape
for i in range(batch_size):
for j in range(seq_len):
token_id = self._input_ids[i, j]
self.dW[token_id] += d_out[i, j]
return None # 不需要向更前层传播
def get_params(self):
"""获取参数和梯度（用于优化器）"""
return [{"W": self.W, "dW": self.dW}]
但是句子本身词位置关系我们也要考虑，所以就加了个位置编码层。
class PositionalEncoding:
"""
先解释下 位置编码层 与 embadding层的关系：
embadding告诉这个词是谁或者含义
位置编码层告诉这个词的位置
或者说：当模型处理第 pos个词时，就直接从这张表里拿出第 pos行的向量，和词向量加在一起，告诉模型：“我站在这里”。
正弦/余弦位置编码层 (Positional Encoding)
为序列中的每个位置添加位置信息，使模型能感知 token 的顺序。
使用固定（不可学习）的正弦和余弦函数。
PE(pos, 2i) = sin(pos / 10000^(2i/d_model))
PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model))
示例:
>>> pe = PositionalEncoding(d_model=64, max_seq_len=128)
>>> x = np.zeros((1, 10, 64))
>>> output = pe.forward(x)
>>> output.shape
(1, 10, 64)
"""
def __init__(self, d_model, max_seq_len=512):
"""
初始化位置编码
参数:
d_model (int): 模型维度 D
max_seq_len (int): 最大支持的序列长度
返回: PositionalEncoding 实例
预计算:
在初始化时预计算所有位置的位置编码矩阵，避免每次前向传播重复计算。
它在 Transformer 中是如何使用的？
假设你的输入句子是 "猫 吃 鱼"：
Embedding 层 把这三个字变成向量（假设也是 4 维）：
E(猫) = [0.5, 0.1, -0.3, 0.8]
E(吃) = [0.2, 0.4, 0.1, -0.1]
E(鱼) = [-0.1, 0.5, 0.6, 0.2]
加上位置编码（把上面表格里的行对应加上）：
Input(猫) = E(猫) + pe[0]
=[0.5,0.1,−0.3,0.8]+[0.0,1.0,0.0,1.0]
=[0.5,1.1,−0.3,1.8]
Input(吃) = E(吃) + pe[1]
=[0.2,0.4,0.1,−0.1]+[0.841,0.540,0.010,1.000]
=[1.041,0.940,0.110,0.900]
Input(鱼) = E(鱼) + pe[2]
=[−0.1,0.5,0.6,0.2]+[0.909,−0.416,0.020,0.999]
=[0.809,0.084,0.620,1.199]
"""
self.d_model = d_model
self.max_seq_len = max_seq_len
# 预计算位置编码矩阵: (max_seq_len, d_model)
self.pe = np.zeros((max_seq_len, d_model), dtype=np.float32)
# 位置索引: (max_seq_len, 1)
# np.arange(5) -> 结果：[ [0] [1] [2] [3] [4] [5] ] 就是生成步长位1的一维数组
# 同理 np.arange(start, stop, step)
# [:, np.newaxis]将其转换为列向量，形状为 (max_seq_len, 1)。
position = np.arange(max_seq_len)[:, np.newaxis]
# 维度索引的除数项: 也是一个矩阵，但是数量只有d_model的一半了
# 这一步很难看明白，先看(np.log(10000.0) / d_model) 没明白，啥作用
# exp里面是生成一个一维数组，以d_model为长度，步长是2生成的
# 然后每个数组元素都乘 -(np.log(10000.0) / d_model)
# 最终得到一个频率除数 这里的结果在0-1之间，说是除数，其实已转为0-1的百分比形式，
# 后面使用时，用乘就行了。
div_term = np.exp(np.arange(0, d_model, 2) * -(np.log(10000.0) / d_model))
# position是一个max_seq_len行 1列的数组，
# position * div_term 每行的值都乘以频率除数 会得到一个[position长度的列，position长度的列/2]
# self矩阵的 0::2表示偶数列 1::2表示奇数列
# 偶数维度用 sin
self.pe[:, 0::2] = np.sin(position * div_term)
# 奇数维度用 cos
self.pe[:, 1::2] = np.cos(position * div_term)
def forward(self, x):
"""
前向传播: 将位置编码加到输入上
参数:
x (numpy.ndarray): 输入嵌入，形状 (batch_size, seq_len, d_model)
示例: (2, 10, 64)
返回:
numpy.ndarray: 加位置编码后的输出，形状与输入相同
示例: (2, 10, 64)
"""
seq_len = x.shape[1]
# 如果序列长度超过预计算的 pe，动态扩展
"""
理解下这块内容，先说为啥需要位置，因为前面我们也看到了 embadding把录入的句子词语给打乱了。
也就是词之间现在是没有位置关系的。所以这里我们要给他加上位置向量。
总之，就是怎么表示一个位置呢？就是跟表示词一样，用数组来表示，各个元素一叠加就能标识一个位置。
至于这里的sin cos是把开始赋予的数组元素值改为0-1的小数， 而且还能提供泛化（即根据历史数据自动推理）
"""
if seq_len > self.pe.shape[0]:
# 动态计算额外的位置编码
d_model = self.d_model
pe_ext = np.zeros((seq_len, d_model), dtype=np.float32)
position = np.arange(self.pe.shape[0], seq_len)[:, np.newaxis]
div_term = np.exp(np.arange(0, d_model, 2) * -(np.log(10000.0) / d_model))
pe_ext[:, 0::2] = np.sin(position * div_term)
pe_ext[:, 1::2] = np.cos(position * div_term)
self.pe = np.concatenate([self.pe, pe_ext], axis=0)
# 取前 seq_len 个位置的编码，广播加到 batch 上
x = x + self.pe[:seq_len]
return x
def backward(self, d_out):
"""
反向传播: 位置编码是常量，梯度直接通过
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, seq_len, d_model)
返回:
numpy.ndarray: 与 d_out 相同的梯度
"""
return d_out
def get_params(self):
"""位置编码没有可学习参数"""
return []最后，词嵌入向量+词位置向量 = 词在句子中全部信息
-
编码器层EncoderLayer
堆叠 n_layers 个 EncoderLayer。
将源序列编码为连续的上下文表示（memory）。
这一层处理的对象是源数据，
让 token 互相交流 → 保留自己 → 再独立思考 → 再保留自己**
注意此时处理数据是3维的，即(batch_size, src_len, d_model)
class EncoderLayer:
"""
编码器层 (Encoder Layer)
结构: 自注意力 -> 残差 -> LayerNorm -> FFN -> 残差 -> LayerNorm
输入输出形状: (batch_size, src_len, d_model)
示例:
>>> enc_layer = EncoderLayer(d_model=64, n_heads=4, d_ff=256)
>>> x = np.random.randn(2, 10, 64)
>>> out = enc_layer.forward(x, src_mask=None)
>>> out.shape
(2, 10, 64)
"""
def __init__(self, d_model, n_heads, d_ff, dropout_rate=0.1):
"""
初始化编码器层
参数:
d_model (int): 模型维度
n_heads (int): 注意力头数
d_ff (int): 前馈网络中间维度
dropout_rate (float): dropout 概率
"""
# 多头自注意力
self.self_attn = MultiHeadAttention(d_model, n_heads, dropout_rate)
# FFN
self.ffn = FeedForward(d_model, d_ff, dropout_rate)
# LayerNorm（两个，分别用于 attention 后和 FFN 后）
self.norm1 = LayerNorm(d_model)
self.norm2 = LayerNorm(d_model)
# Dropout（用于残差连接）
self.dropout = Dropout(dropout_rate)
# 缓存
self._cache = {}
def forward(self, x, src_mask=None, training=True):
"""
前向传播
让 token 互相交流 → 保留自己 → 再独立思考 → 再保留自己**
参数:
x (numpy.ndarray): 输入，形状 (batch_size, src_len, d_model)
src_mask (numpy.ndarray|None): 源序列 padding mask
形状: (batch_size, 1, 1, src_len)
training (bool): 是否训练模式
返回:
numpy.ndarray: 输出，形状 (batch_size, src_len, d_model)
"""
# 1. 多头自注意力 + 残差 + LayerNorm
# Self-Attention: Q=K=V=x（自注意力，自己关注自己）
attn_out = self.self_attn.forward(x, x, x, mask=src_mask, training=training)
# 随机关掉一部分注意力结果，防止过拟合， 强迫模型不依赖某个token
attn_out = self.dropout.forward(attn_out, training=training)
# 残差即保留原始信息：原始的+改进的 防止跑偏
x = self.norm1.forward(x + attn_out) # 残差连接 + LayerNorm
# 2. FFN + 残差 + LayerNorm
# ffn:前馈网络：Self-Attention所有token间互换信息，FFN是对每个token单独做线性变换
# 前者 社交，后者自主独立思考
ffn_out = self.ffn.forward(x, training=training)
ffn_out = self.dropout.forward(ffn_out, training=training)
# 再次残差 跟第一次一样
x = self.norm2.forward(x + ffn_out) # 残差连接 + LayerNorm
return x
def backward(self, d_out):
"""
反向传播
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, src_len, d_model)
返回:
numpy.ndarray: 输入梯度，形状 (batch_size, src_len, d_model)
"""
# 反向: norm2 -> ffn -> norm1 -> self_attn
# 2. FFN 路径
d_norm2 = self.norm2.backward(d_out)
d_ffn_out = self.dropout.backward(d_norm2)
d_ffn_x = self.ffn.backward(d_ffn_out)
d_out = d_norm2 + d_ffn_x # 残差分支
# 1. Self-Attention 路径
d_norm1 = self.norm1.backward(d_out)
d_attn_out = self.dropout.backward(d_norm1)
d_query, d_key, d_value = self.self_attn.backward(d_attn_out)
d_x = d_norm1 + d_query # 三个梯度相同（Q=K=V）
return d_x
def get_params(self):
"""获取所有可训练参数"""
params = []
params.extend(self.self_attn.get_params())
params.extend(self.ffn.get_params())
params.extend(self.norm1.get_params())
params.extend(self.norm2.get_params())
return params
-
多头注意力 最重要的一步，可以说前面都在准备，这里真正的开始计算
此时才是真正的开始计算了。引入了其Q,K,V
虽然 query、key、value初始值相同（都是 x），但它们分别乘以了 三个不同的矩阵：
|
变换 |
作用 |
|---|---|
|
|
把输入变成“提问视角” |
|
|
把输入变成“索引视角” |
|
|
把输入变成“内容视角” |
所以经过变化之后已经不一样了。
然后在计算分数的时候 ，用到了这一行代码：
scores = (Q @ K.transpose(0, 1, 3, 2))
这里可以理解成句子内部词之间的点乘，也就是序列内部在互相观察
比如句子是："我爱中国"
-
Q（我） 去点乘 K（爱、中、国）
-
Q（爱） 去点乘 K（我、中、国）
点乘在算什么？
点乘（Dot Product）算的是 “相似度”。
|
情况 |
点乘结果 |
含义 |
|---|---|---|
|
“我” 看 “爱” |
高分 ✅ |
关系密切 |
|
“我” 看 “国” |
中分 |
有点关系 |
|
“我” 看 “的” |
低分 ❌ |
没什么关系 |
**“在 Self-Attention 中，Q、K、V 虽然源自同一输入，但通过不同的可学习线性变换被映射到不同的表示空间，赋予其查询、键和值的语义。
所谓‘自己点乘自己’，实质上是序列中每个位置（Query）与所有位置（Key）计算相似度，从而实现全局信息交互，而非单个 Token 的自相关。”**
最后：
scores 我想不想听你讲
mask 空椅子不许发言
softmax 时间分配比例 即归一化处理
context 我综合后的理解 机器自己捣鼓的结果 去中和 真实可信任的结果
output 我最终表达的观点 输出线性投影 我们在找规律，再加一个除QKV之外的权重参数是有必要的。
class MultiHeadAttention:
"""
多头自注意力层 (Multi-Head Self-Attention)
多头注意力就是同时计算词在各个维度上的占比，比如说 语法 / 语义 / 位置
一开始这个头数怎么定呢？也顺便说下各种参数的值的定位：其实大多是还是经验所得
很难说清楚其中缘由。
多头注意力的头数，是由 d_model决定的，目标是让每个头的维度保持在 64 左右，兼顾表达能力、数值稳定性和工程效率。
d_model = 512 → num_heads = 8
d_model = 768 → num_heads = 12
d_model = 1024 → num_heads = 16
d_model = 2048 → num_heads = 32
所以要去d_model 必须要能被头数整除
核心思想: 将注意力计算分成多个"头"，每个头学习不同的关注模式，
最后拼接所有头的结果。
计算流程:
1. 线性投影: Q = X·W_q, K = X·W_k, V = X·W_v
2. 分头: reshape 为 (batch, n_heads, seq_len, d_head)
3. 缩放点积注意力: Attention(Q,K,V) = softmax(Q·K^T/√d_k)·V
4. 拼接多头结果
5. 线性输出投影
示例:
>>> mha = MultiHeadAttention(d_model=64, n_heads=4)
>>> x = np.random.randn(2, 10, 64) # batch=2, seq=10
>>> out = mha.forward(x, x, x) # 自注意力
>>> out.shape
(2, 10, 64)
"""
def __init__(self, d_model, n_heads, dropout_rate=0.1, seed=None):
"""
初始化多头注意力层
参数:
d_model (int): 模型维度 D，必须能被 n_heads 整除
n_heads (int): 注意力头数 H
dropout_rate (float): dropout 概率
seed (int|None): 随机种子
返回: MultiHeadAttention 实例
权重初始化:
W_q, W_k, W_v: (d_model, d_model) - 查询/键/值投影矩阵
W_o: (d_model, d_model) - 输出投影矩阵
"""
self.d_model = d_model
self.n_heads = n_heads
self.d_head = d_model // n_heads # 每个头的维度
self.dropout_rate = dropout_rate
if seed is not None:
np.random.seed(seed)
# Xavier 初始化缩放因子
# 除以 √d_model是为了把 Attention 分数的方差拉回 1，防止点积结果过大导致 softmax 梯度消失。
# 因为后面在做 Self-Attention 的线性投影层 中（如 W_q, W_k, W_v）的 权重更新的时候，权重值会随着计算变大，可能会爆炸
# 至于这个6 听说是方差求导后的值，切记住吧
limit = np.sqrt(6.0 / (d_model + d_model))
# Q, K, V 投影权重
# 下面应该好理解了 就是计算自注意力时需要的几个权重
"""
为什么 Attention 需要三个投影？
因为 Attention 的机制决定了它们角色不同：
Query (Q)“我在找什么”
Key (K)“我有什么”
Value (V) “我实际给什么信息”
之所以叫“投影”，是因为 q k v
通过矩阵乘法，把输入向量从原始的高维空间，线性变换到新的、专门为注意力设计的表示空间中。
再者 通过矩阵变换确实降维了。可以把512维降到64维。但是注意力是多头的
怎么实现降维的呢，是通过q,k,v一层层来的哦
比如说有5个词，每个词的d_model是512，现在我要转成64维的。
那就要[5,512]*[512,64] -> [5,64]
这样就得到了，
然后是多头注意力。就是把512个维度投射到8个64维，能够并行计算，防止计算爆炸，而且也能多视角学习
也就是大家说的 Reshape重组
比如：
# 配置
d_model = 512
num_heads = 8
head_dim = d_model // num_heads # 64
# 1. 定义大权重矩阵 (512 -> 8 * 64)
# 这里把 8 个头的权重打包在一起
W_q_multi = np.random.randn(d_model, num_heads * head_dim) * 0.01
# 2. 输入
X = np.random.randn(1, 5, 512)
# 3. 一次性投影到 8 个头
# 输出形状: (1, 5, 8 * 64) = (1, 5, 512)
multi_head_raw = X @ W_q_multi
"""
#至于下面几个投影为啥维度还是大矩阵维度，因为这是为了方便计算，
# 现阶段还没有并行操作，所以还是大矩阵快，
# 后面reshape向前传播阶段和会降下来的,并行计算就快了
# 大矩阵就是[512,512]。而多头注意力矩阵 np.random.randn(1, 8, 64)，
# 后者需要调用8次64维矩阵算法，在串行计算下会比大矩阵慢
# **初始化时：造一辆 8 厢列车（512 米长） Forward 时：把车厢拆开，让 8 节车厢并行跑**
# 张量是什么？其实就是数组，只不过在transformer中在经过大矩阵到reshape的变化后，
# 二者的值还是不变的，但是计算速度加快了。
self.W_q = np.random.uniform(-limit, limit, (d_model, d_model)).astype(
np.float32
)
self.W_k = np.random.uniform(-limit, limit, (d_model, d_model)).astype(
np.float32
)
self.W_v = np.random.uniform(-limit, limit, (d_model, d_model)).astype(
np.float32
)
self.W_o = np.random.uniform(-limit, limit, (d_model, d_model)).astype(
np.float32
)
# 偏置项
self.b_q = np.zeros(d_model, dtype=np.float32)
self.b_k = np.zeros(d_model, dtype=np.float32)
self.b_v = np.zeros(d_model, dtype=np.float32)
self.b_o = np.zeros(d_model, dtype=np.float32)
# 梯度缓存
self.dW_q = np.zeros_like(self.W_q)
self.dW_k = np.zeros_like(self.W_k)
self.dW_v = np.zeros_like(self.W_v)
self.dW_o = np.zeros_like(self.W_o)
self.db_q = np.zeros_like(self.b_q)
self.db_k = np.zeros_like(self.b_k)
self.db_v = np.zeros_like(self.b_v)
self.db_o = np.zeros_like(self.b_o)
# 前向传播缓存（用于反向传播）
self._cache = {}
def forward(self, query, key, value, mask=None, training=True):
"""
前向传播: 计算多头注意力
参数:
query (numpy.ndarray): 查询输入，形状 (batch_size, q_len, d_model)
示例: (2, 10, 64)
key (numpy.ndarray): 键输入，形状 (batch_size, k_len, d_model)
示例: (2, 10, 64)
value (numpy.ndarray): 值输入，形状 (batch_size, v_len, d_model)
示例: (2, 10, 64)
mask (numpy.ndarray|None): 掩码矩阵
- Padding mask: (batch_size, 1, 1, seq_len)
- Look-ahead mask: (seq_len, seq_len) 或组合掩码
示例: (2, 1, 1, 10) 或 (10, 10)
training (bool): 是否为训练模式（控制 dropout）
返回:
numpy.ndarray: 注意力输出，形状 (batch_size, q_len, d_model)
示例: (2, 10, 64)
注意力分数形状:
scores: (batch_size, n_heads, q_len, k_len)
attn_weights: (batch_size, n_heads, q_len, k_len)
context: (batch_size, n_heads, q_len, d_head)
"""
# 总结这里在干嘛：
# 让每个 token 拿着自己的“问题（Q）”，去找所有 token 的“名片（K）”匹配，然后用匹配程度去拿别人的“干货（V）”。
# 进来的qkvs是三维的，中间经过转换变4维，最后变回来。
batch_size = query.shape[0]
q_len = query.shape[1]
k_len = key.shape[1]
# 保存输入用于反向传播
self._cache = {
"query": query,
"key": key,
"value": value,
"mask": mask,
"batch_size": batch_size,
}
"""
角色 心态
Q “我想找什么”
K “我适合被谁找到”
V “我实际能提供什么”
代码 会议室
Q 我想问什么
K 你能答什么
V 你实际说的内容
scores 我想不想听你讲
mask 空椅子不许发言
softmax 时间分配比例
context 我综合后的理解
output 我最终表达的观点
"""
# 1. 线性投影: Q, K, V
# (batch, seq, d_model) @ (d_model, d_model) -> (batch, seq, d_model)
Q = query @ self.W_q + self.b_q # (B, q_len, D)
K = key @ self.W_k + self.b_k # (B, k_len, D)
V = value @ self.W_v + self.b_v # (B, v_len, D)
# 讲下transpose函数，就是把原数组按照transpose参数下班重新变化
# 比如transpose（1，0） 就是把原数组第一维和地位变化下位置
# 2. 分头: (batch, seq, d_model) -> (batch, n_heads, seq, d_head)
# 比如 头1关注语法，头2关注指向，头3关注语义 头四关注位置
Q = Q.reshape(batch_size, q_len, self.n_heads, self.d_head).transpose(
0, 2, 1, 3
)
K = K.reshape(batch_size, k_len, self.n_heads, self.d_head).transpose(
0, 2, 1, 3
)
V = V.reshape(batch_size, k_len, self.n_heads, self.d_head).transpose(
0, 2, 1, 3
)
# 现在 Q, K, V: (batch_size, n_heads, seq_len, d_head)
# 3. 缩放点积注意力 打分 q\k做相似度匹配 越高越像
# scores: (batch, n_heads, q_len, k_len)
scores = (Q @ K.transpose(0, 1, 3, 2)) / np.sqrt(self.d_head)
# 应用掩码（如果有） 隐藏掉空值，空值不能影响结果
if mask is not None:
# mask 需要广播到 (batch, n_heads, q_len, k_len)
# mask 为 1 的位置设为 -1e9（softmax 后接近 0）
scores = scores + (mask * -1e9)
# Softmax 得到注意力权重 归一化，所有值都放到0-1之间
attn_weights = softmax(scores, axis=-1) # (B, H, q_len, k_len)
# Dropout（训练时） 随机丢掉某些值，防止过度依赖，提升泛化
# 不用担心被丢掉的值，我们是多次训练，不影响。
if training and self.dropout_rate > 0:
dropout_mask = (
np.random.rand(*attn_weights.shape) > self.dropout_rate
).astype(np.float32)
attn_weights = attn_weights * dropout_mask / (1 - self.dropout_rate)
self._cache["dropout_mask"] = dropout_mask
# 4. 加权求和: (B, H, q_len, k_len) @ (B, H, k_len, d_head) -> (B, H, q_len, d_head)
# 机器自己捣鼓的结果 去中和 真实可信任的结果
context = attn_weights @ V
# 5. 拼接多头: (B, H, q_len, d_head) -> (B, q_len, D)
# 计算结束 还原了
context = context.transpose(0, 2, 1, 3).reshape(batch_size, q_len, self.d_model)
# 6. 输出线性投影
# 我们在找规律，再加一个权重参数是有必要的。
output = context @ self.W_o + self.b_o # (B, q_len, D)
# 缓存中间结果用于反向传播
self._cache.update(
{
"Q": Q,
"K": K,
"V": V,
"scores": scores,
"attn_weights": attn_weights,
"context": context,
}
)
return output
def backward(self, d_out):
"""
反向传播: 计算注意力层的梯度
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, q_len, d_model)
返回:
numpy.ndarray: 输入梯度，形状 (batch_size, q_len, d_model)
梯度计算流程:
1. W_o 的梯度
2. context 的梯度
3. V 的梯度（通过注意力权重）
4. 注意力权重的梯度
5. scores 的梯度
6. Q, K 的梯度
7. W_q, W_k, W_v 的梯度
8. 输入的梯度
"""
batch_size = self._cache["batch_size"]
q_len = d_out.shape[1]
k_len = self._cache["key"].shape[1]
Q, K, V = self._cache["Q"], self._cache["K"], self._cache["V"]
attn_weights = self._cache["attn_weights"]
context = self._cache["context"]
query, key, value = (
self._cache["query"],
self._cache["key"],
self._cache["value"],
)
mask = self._cache["mask"]
# 1. W_o, b_o 的梯度
# output = context @ W_o + b_o
# dW_o = context^T @ d_out
self.dW_o = context.reshape(-1, self.d_model).T @ d_out.reshape(
-1, self.d_model
)
self.db_o = np.sum(d_out, axis=(0, 1))
d_context = d_out @ self.W_o.T # (B, q_len, D)
# 2. 恢复多头形状
d_context = d_context.reshape(batch_size, q_len, self.n_heads, self.d_head)
d_context = d_context.transpose(0, 2, 1, 3) # (B, H, q_len, d_head)
# 3. dV = attn_weights^T @ d_context
dV = attn_weights.transpose(0, 1, 3, 2) @ d_context # (B, H, k_len, d_head)
# 4. d_attn_weights = d_context @ V^T
d_attn_weights = d_context @ V.transpose(0, 1, 3, 2) # (B, H, q_len, k_len)
# Dropout 梯度
if "dropout_mask" in self._cache:
d_attn_weights = (
d_attn_weights * self._cache["dropout_mask"] / (1 - self.dropout_rate)
)
# 5. d_scores = d_attn_weights * softmax_derivative
# softmax 导数: d_softmax = softmax * (d - sum(d * softmax))
d_scores = d_attn_weights * attn_weights
d_scores_sum = np.sum(d_scores, axis=-1, keepdims=True)
d_scores = d_scores - attn_weights * d_scores_sum
# 缩放因子
d_scores = d_scores / np.sqrt(self.d_head)
# 6. dQ, dK
dQ = d_scores @ K # (B, H, q_len, d_head)
dK = d_scores.transpose(0, 1, 3, 2) @ Q # (B, H, k_len, d_head)
# 7. 恢复形状并计算 W_q, W_k, W_v 的梯度
dQ = dQ.transpose(0, 2, 1, 3).reshape(batch_size, q_len, self.d_model)
dK = dK.transpose(0, 2, 1, 3).reshape(batch_size, k_len, self.d_model)
dV = dV.transpose(0, 2, 1, 3).reshape(batch_size, key.shape[1], self.d_model)
# W_q 的梯度
self.dW_q = query.reshape(-1, self.d_model).T @ dQ.reshape(-1, self.d_model)
self.db_q = np.sum(dQ, axis=(0, 1))
# W_k 的梯度
self.dW_k = key.reshape(-1, self.d_model).T @ dK.reshape(-1, self.d_model)
self.db_k = np.sum(dK, axis=(0, 1))
# W_v 的梯度
self.dW_v = value.reshape(-1, self.d_model).T @ dV.reshape(-1, self.d_model)
self.db_v = np.sum(dV, axis=(0, 1))
# 8. 输入梯度
d_query = dQ @ self.W_q.T
d_key = dK @ self.W_k.T
d_value = dV @ self.W_v.T
return d_query, d_key, d_value
def get_params(self):
"""获取所有可训练参数和梯度"""
return [
{"W": self.W_q, "dW": self.dW_q, "b": self.b_q, "db": self.db_q},
{"W": self.W_k, "dW": self.dW_k, "b": self.b_k, "db": self.db_k},
{"W": self.W_v, "dW": self.dW_v, "b": self.b_v, "db": self.db_v},
{"W": self.W_o, "dW": self.dW_o, "b": self.b_o, "db": self.db_o},
]
FFN 前馈神经网络层(Feed-Forward Network)
其实就是lstm中的激活函数，但是又有不同
FFN 是一个位置独立的前馈网络，由两个线性变换和一个非线性激活函数组成。
它的作用是增加模型的非线性表达能力，并对每个位置的特征进行进一步加工。
与 LSTM 中的激活函数不同，FFN 是一个独立的子层，不负责时序信息传递，
仅对每个位置的表示进行变换。class FeedForward:
"""
前馈神经网络层 (Feed-Forward Network)
**FFN 是每个 Token 的“私人加工厂”。
：引入非线性，让模型能学复杂模式。
**“FFN 是一个位置独立的前馈网络，由两个线性变换和一个非线性激活函数组成。它的作用是增加模型的非线性表达能力，并对每个位置的特征进行进一步加工。
与 LSTM 中的激活函数不同，FFN 是一个独立的子层，不负责时序信息传递，仅对每个位置的表示进行变换。”*
Attention 负责“社交”（看别人），FFN 负责“独处”（自己想）。**
每个位置独立应用的两层全连接网络:
FFN(x) = GELU(x @ W1 + b1) @ W2 + b2
中间维度 d_ff 通常为 4 * d_model。
示例:
>>> ffn = FeedForward(d_model=64, d_ff=256)
>>> x = np.random.randn(2, 10, 64)
>>> out = ffn.forward(x)
>>> out.shape
(2, 10, 64)
"""
def __init__(self, d_model, d_ff, dropout_rate=0.1, seed=None):
"""
初始化前馈层
参数:
d_model (int): 输入/输出维度
d_ff (int): 中间隐藏层维度（通常为 4*d_model）
dropout_rate (float): dropout 概率
seed (int|None): 随机种子
返回: FeedForward 实例
权重:
W1: (d_model, d_ff) - 第一层投影
b1: (d_ff,) - 第一层偏置
W2: (d_ff, d_model) - 第二层投影
b2: (d_model,) - 第二层偏置
"""
self.d_model = d_model
# 隐藏层维度
self.d_ff = d_ff
self.dropout_rate = dropout_rate
if seed is not None:
np.random.seed(seed)
# Xavier 初始化
# 至于为啥用6 且记住吧 因为方法的求导后是6
limit1 = np.sqrt(6.0 / (d_model + d_ff))
limit2 = np.sqrt(6.0 / (d_ff + d_model))
self.W1 = np.random.uniform(-limit1, limit1, (d_model, d_ff)).astype(np.float32)
self.b1 = np.zeros(d_ff, dtype=np.float32)
self.W2 = np.random.uniform(-limit2, limit2, (d_ff, d_model)).astype(np.float32)
self.b2 = np.zeros(d_model, dtype=np.float32)
# 梯度
self.dW1 = np.zeros_like(self.W1)
self.db1 = np.zeros_like(self.b1)
self.dW2 = np.zeros_like(self.W2)
self.db2 = np.zeros_like(self.b2)
# 缓存
self._cache = {}
def forward(self, x, training=True):
"""
前向传播
参数:
x (numpy.ndarray): 输入，形状 (batch_size, seq_len, d_model)
示例: (2, 10, 64)
training (bool): 是否训练模式
返回:
numpy.ndarray: 输出，形状 (batch_size, seq_len, d_model)
示例: (2, 10, 64)
中间形状:
hidden: (batch_size, seq_len, d_ff)
"""
# 第一层: (B, S, D) @ (D, d_ff) -> (B, S, d_ff)
hidden = x @ self.W1 + self.b1
hidden_activated = gelu(hidden)
# Dropout
if training and self.dropout_rate > 0:
dropout_mask = (
np.random.rand(*hidden_activated.shape) > self.dropout_rate
).astype(np.float32)
hidden_activated = hidden_activated * dropout_mask / (1 - self.dropout_rate)
self._cache["dropout_mask"] = dropout_mask
# 第二层: (B, S, d_ff) @ (d_ff, D) -> (B, S, D)
output = hidden_activated @ self.W2 + self.b2
self._cache = {"x": x, "hidden": hidden, "hidden_activated": hidden_activated}
return output
def backward(self, d_out):
"""
反向传播
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, seq_len, d_model)
返回:
numpy.ndarray: 输入梯度，形状 (batch_size, seq_len, d_model)
"""
x = self._cache["x"]
hidden = self._cache["hidden"]
hidden_activated = self._cache["hidden_activated"]
# W2, b2 梯度
self.dW2 = hidden_activated.reshape(-1, self.d_ff).T @ d_out.reshape(
-1, self.d_model
)
self.db2 = np.sum(d_out, axis=(0, 1))
d_hidden_activated = d_out @ self.W2.T # (B, S, d_ff)
# Dropout 梯度
if "dropout_mask" in self._cache:
d_hidden_activated = (
d_hidden_activated
* self._cache["dropout_mask"]
/ (1 - self.dropout_rate)
)
# GELU 导数
d_hidden = d_hidden_activated * gelu_derivative(hidden)
# W1, b1 梯度
self.dW1 = x.reshape(-1, self.d_model).T @ d_hidden.reshape(-1, self.d_ff)
self.db1 = np.sum(d_hidden, axis=(0, 1))
d_x = d_hidden @ self.W1.T
return d_x
def get_params(self):
"""获取可训练参数"""
return [
{"W": self.W1, "dW": self.dW1, "b": self.b1, "db": self.db1},
{"W": self.W2, "dW": self.dW2, "b": self.b2, "db": self.db2},
]
-
归一化
就是对一组数据缩放，等比例的放到0-1之间
-
dropout层，丢弃层
训练时随机丢弃一部分神经元，防止过拟合。
推理时不丢弃，但输出乘以 (1 - dropout_rate) 以保持期望一致。
class Dropout:
"""
Dropout 层
训练时随机丢弃一部分神经元，防止过拟合。
推理时不丢弃，但输出乘以 (1 - dropout_rate) 以保持期望一致。
示例:
>>> drop = Dropout(rate=0.1)
>>> x = np.ones((2, 10, 64))
>>> out = drop.forward(x, training=True)
>>> out.shape
(2, 10, 64)
"""
def __init__(self, rate=0.1):
"""
初始化 Dropout
参数:
rate (float): 丢弃概率，0-1 之间
"""
self.rate = rate
self._mask = None
def forward(self, x, training=True):
"""
前向传播
参数:
x (numpy.ndarray): 输入，任意形状
training (bool): 是否为训练模式
True: 随机丢弃部分元素
False: 直接返回输入（不丢弃）
返回:
numpy.ndarray: dropout 后的输出，形状与输入相同
"""
if not training or self.rate == 0:
return x
# 生成随机掩码: 1 表示保留，0 表示丢弃
self._mask = (np.random.rand(*x.shape) > self.rate).astype(np.float32)
# 缩放: 保留的元素除以 (1 - rate)，保证期望值不变
return x * self._mask / (1 - self.rate)
def backward(self, d_out):
"""
反向传播
参数:
d_out (numpy.ndarray): 输出梯度
返回:
numpy.ndarray: 输入梯度
"""
if self._mask is None:
return d_out
return d_out * self._mask / (1 - self.rate)
def get_params(self):
"""Dropout 没有可训练参数"""
return []
-
残差连接
残差即保留原始信息：原始的+改进的 防止跑偏
把前面丢弃后的，再加上原来的，就是残差连接
-
总结前向传播的几个过程
|
模块 |
人脑活动 |
|---|---|
|
Attention |
开会讨论（听别人说） |
|
FFN |
回家自己琢磨（深度思考） |
|
残差连接 |
保留原始记忆 |
|
LayerNorm |
整理思绪 |
-
解码器层DecoderLayer
结构: 掩码自注意力 -> 残差 -> LayerNorm
-> 交叉注意力 -> 残差 -> LayerNorm
-> FFN -> 残差 -> LayerNorm
注意跟编码层的区别，尤其是传入的参数对象，也是处理的对象。编码层是处理源数据的。
编码层是处理目标数据和编程层处理后的memory(将源序列编码为连续的上下文表示。
然后处理方式上好像和编码差不多
class Decoder:
"""
解码器 (Decoder)
堆叠 n_layers 个 DecoderLayer。
根据目标序列和 Encoder 输出（memory）生成目标序列的表示。
示例:
>>> decoder = Decoder(d_model=64, n_heads=4, d_ff=256, n_layers=2)
>>> x = np.random.randn(2, 8, 64)
>>> memory = np.random.randn(2, 10, 64)
>>> out = decoder.forward(x, memory, look_ahead_mask=None)
>>> out.shape
(2, 8, 64)
"""
def __init__(self, d_model, n_heads, d_ff, n_layers, dropout_rate=0.1):
"""
初始化解码器
参数:
d_model (int): 模型维度
n_heads (int): 注意力头数
d_ff (int): 前馈网络中间维度
n_layers (int): 解码器层数
dropout_rate (float): dropout 概率
"""
self.layers = [
DecoderLayer(d_model, n_heads, d_ff, dropout_rate)
for _ in range(n_layers)
]
def forward(self, x, memory, look_ahead_mask=None, src_mask=None, training=True):
"""
前向传播
参数:
x (numpy.ndarray): 目标序列嵌入，形状 (batch_size, tgt_len, d_model)
memory (numpy.ndarray): Encoder 输出，形状 (batch_size, src_len, d_model)
look_ahead_mask (numpy.ndarray|None): 前瞻掩码
src_mask (numpy.ndarray|None): 源序列 padding mask
training (bool): 是否训练模式
返回:
numpy.ndarray: 解码后的表示，形状 (batch_size, tgt_len, d_model)
"""
for layer in self.layers:
x = layer.forward(x, memory, look_ahead_mask, src_mask, training)
return x
def backward(self, d_out):
"""
反向传播
参数:
d_out (numpy.ndarray): 输出梯度，形状 (batch_size, tgt_len, d_model)
返回:
tuple: (d_x, d_memory)
d_x: 目标序列梯度
d_memory: 编码器输出梯度
"""
d_memory_total = None
for layer in reversed(self.layers):
d_out, d_memory = layer.backward(d_out)
if d_memory_total is None:
d_memory_total = d_memory
else:
d_memory_total += d_memory
return d_out, d_memory_total
def get_params(self):
"""获取所有可训练参数"""
params = []
for layer in self.layers:
params.extend(layer.get_params())
return params前向传播结束
会返回个线性函数，表示最终源数据和目标数据间的关系
-
计算损失和梯度 交叉熵
这一步很重要了，前向传播 我们最终得出了一个线性函数，那么我们带入源数据值到这个线性函数中是不是就是我们的预测值了呢。然后拿这个预测值跟目标值对比，是不是就能计算出损失了呢。
用交叉熵拟合预测值和目标值 就能得到损失值，然后求导得出损失梯度。
class CrossEntropyLoss:
"""
交叉熵损失函数
用于分类任务，衡量预测概率分布与真实标签的差异。
公式: L = -Σ(y_true * log(y_pred)) / N
同时处理 padding 位置（mask 掉 <PAD> 的损失）。
示例:
>>> loss_fn = CrossEntropyLoss()
>>> logits = np.random.randn(2, 5, 10) # (B, S, V)
>>> targets = np.array([[1, 3, 5, 0, 0], [2, 4, 6, 7, 0]])
>>> loss, d_logits = loss_fn.forward(logits, targets)
"""
def __init__(self, pad_id=0, label_smoothing=0.0):
"""
初始化交叉熵损失
参数:
pad_id (int): 填充 token ID，这些位置不参与损失计算
label_smoothing (float): 标签平滑系数，0 表示不平滑
"""
self.pad_id = pad_id
self.label_smoothing = label_smoothing
def forward(self, logits, targets):
"""
前向传播: 计算损失和梯度
参数:
logits (numpy.ndarray): 模型输出，形状 (batch_size, seq_len, vocab_size)
示例: (2, 5, 100)
targets (numpy.ndarray): 真实标签（token IDs），形状 (batch_size, seq_len)
示例: [[1, 3, 5, 0, 0], [2, 4, 6, 7, 0]]
返回:
tuple: (loss, d_logits)
loss (float): 标量损失值
d_logits (numpy.ndarray): 对 logits 的梯度，形状与 logits 相同
示例: (2, 5, 100)
"""
batch_size, seq_len, vocab_size = logits.shape
# 创建 mask（排除 padding 位置）
mask = (targets != self.pad_id).astype(np.float32) # (B, S)
# 展平
logits_flat = logits.reshape(-1, vocab_size) # (B*S, V)
targets_flat = targets.reshape(-1) # (B*S,)
mask_flat = mask.reshape(-1) # (B*S,)
# Softmax（数值稳定）
logits_max = np.max(logits_flat, axis=-1, keepdims=True)
logits_shifted = logits_flat - logits_max
exp_logits = np.exp(logits_shifted)
probs = exp_logits / np.sum(exp_logits, axis=-1, keepdims=True) # (B*S, V)
# 有效位置数量
valid_count = np.sum(mask_flat)
if valid_count == 0:
return 0.0, np.zeros_like(logits)
# 交叉熵损失: -log(p[target])
# 解释下 为啥叫交叉熵？ 熵是表示 混乱程度
# 就是拿预测的数据跟真实的数据拟合在一起， 拟合结果越高，误差就越低喽。
log_probs = np.log(np.clip(probs, 1e-10, 1.0))
# 只对有效位置计算损失
target_log_probs = log_probs[np.arange(len(targets_flat)), targets_flat]
loss = -np.sum(target_log_probs * mask_flat) / valid_count
# 梯度: dL/d_logit = p - one_hot(target)
d_logits_flat = probs.copy()
d_logits_flat[np.arange(len(targets_flat)), targets_flat] -= 1
# 应用 mask 和缩放
d_logits_flat *= mask_flat[:, np.newaxis]
d_logits_flat /= valid_count
# 恢复形状
d_logits = d_logits_flat.reshape(batch_size, seq_len, vocab_size)
return loss, d_logits-
反向传播
backward 的任务只有一个：
把“最后的错误（d_logits）”，一步步拆回去，告诉每一层：
“你当初哪里算错了，应该怎么改。
def backward(self, d_logits):
"""
反向传播
**backward 的任务只有一个：
把“最后的错误（d_logits）”，一步步拆回去，告诉每一层：
“你当初哪里算错了，应该怎么改。”**
参数:
d_logits (numpy.ndarray): logits 梯度，形状 (batch_size, tgt_len, vocab_size)
返回: 无（梯度存储在各层的参数中）
"""
dec_output = self._cache['dec_output']
memory = self._cache['memory']
src_emb = self._cache['src_emb']
tgt_emb = self._cache['tgt_emb']
tgt_mask = self._cache['tgt_mask']
src_mask = self._cache['src_mask']
# 5. 输出投影梯度 就是误差是多少，计算出后面拿来改
self.dW_out = dec_output.reshape(-1, self.config.d_model).T @ d_logits.reshape(-1, self.config.vocab_size)
self.db_out = np.sum(d_logits, axis=(0, 1))
d_dec_output = d_logits @ self.W_out.T # (B, tgt_len, d_model)
# 4. Decoder 反向
d_tgt_emb, d_memory = self.decoder.backward(d_dec_output)
# 3. Encoder 反向
d_src_emb = self.encoder.backward(d_memory)
# 2. 目标序列嵌入反向
d_tgt_scaled = self.pos_encoding.backward(d_tgt_emb)
d_tgt = self.tgt_embed.backward(d_tgt_scaled * self.scale)
# 1. 源序列嵌入反向
d_src_scaled = self.pos_encoding.backward(d_src_emb)
d_src = self.src_embed.backward(d_src_scaled * self.scale)
return d_src, d_tgt-
梯度裁剪
防梯度爆炸 这就不用多说了，每个权重的梯度结果是可能大于1的，累加后可能会爆炸
def _clip_gradients(self):
"""
梯度裁剪（防止梯度爆炸）
将所有参数的梯度限制在 max_grad_norm 范围内。
公式: grad = grad * (max_norm / max(total_norm, max_norm))
"""
# 计算所有梯度的总范数
total_norm = 0.0
params = self.model.get_params()
for param in params:
for key in ['dW', 'db', 'dgamma', 'dbeta']:
if key in param:
total_norm += np.sum(param[key] ** 2)
total_norm = np.sqrt(total_norm)
# 裁剪
clip_coef = self.config.max_grad_norm / (total_norm + 1e-6)
if clip_coef < 1.0:
for param in params:
for key in ['dW', 'db', 'dgamma', 'dbeta']:
if key in param:
param[key] *= clip_coef-
优化器更新 （动态步长，动量）
自我调节步长 根据“错误信号（梯度）”，决定参数“往哪走、走多大一步”的教练。
之前学习lst时，步长的调价时SGD模式，即步长是固定的，这就太死板了
- 解决的3个问题：
1，步长太大 -》 冲过山谷，跳过了我们要找的最低点
2，步长太小-> 一百年下不来
3，坑坑洼洼，卡在局部最低点
怎么解决呢？
引入了动量的概念。
# 就是超某个方向变化时，会加速变化，进而快速收敛
# 如果一直是来回抖动 就需要减少步长了
# 如果一直某个方向，就需要增大步长了
class AdamOptimizer:
"""
Adam 优化器 (Adaptive Moment Estimation)
结合动量（Momentum）和 RMSProp 的优点:
m_t = β1 * m_{t-1} + (1-β1) * g_t (一阶矩估计)
v_t = β2 * v_{t-1} + (1-β2) * g_t² (二阶矩估计)
m̂_t = m_t / (1-β1^t) (偏差修正)
v̂_t = v_t / (1-β2^t) (偏差修正)
θ_t = θ_{t-1} - lr * m̂_t / (√v̂_t + ε)
示例:
>>> optimizer = AdamOptimizer(lr=0.001, beta1=0.9, beta2=0.98, eps=1e-9)
>>> params = [{'W': np.ones((3,3)), 'dW': np.zeros((3,3))}]
>>> optimizer.step(params, timestep=1)
"""
def __init__(self, lr=0.001, beta1=0.9, beta2=0.98, eps=1e-9):
"""
初始化 Adam 优化器
参数:
lr (float): 学习率，默认 0.001
beta1 (float): 一阶矩衰减率，默认 0.9
beta2 (float): 二阶矩衰减率，默认 0.98
eps (float): 数值稳定常数，默认 1e-9
返回: AdamOptimizer 实例
"""
self.lr = lr
self.beta1 = beta1
self.beta2 = beta2
self.eps = eps
self.t = 0 # 时间步（更新次数）
self.m_states = {} # 一阶矩缓存
self.v_states = {} # 二阶矩缓存
def step(self, params, timestep=None):
"""
解决的3个问题：
1，步长太大 -》 冲过山谷，跳过了我们要找的最低点
2，步长太小-> 一百年下不来
3，坑坑洼洼，卡在局部最低点
代码 含义
m_states 动量（惯性）
v_states 二阶动量（抖不抖）
beta1 惯性保留程度
beta2 抖动记忆程度
eps 防止除以 0
lr 基础步长
执行一步参数更新
参数:
params (list): 参数列表，每个元素是包含 'W'/'dW' 的字典
示例: [{'W': np.array([...]), 'dW': np.array([...])}, ...]
timestep (int|None): 当前时间步，None 则使用内部计数器
返回: 无（直接修改参数）
数据样例:
>>> opt = AdamOptimizer(lr=0.1)
>>> W = np.array([1.0, 2.0])
>>> dW = np.array([0.1, -0.2])
>>> params = [{'W': W, 'dW': dW}]
>>> opt.step(params, timestep=1)
>>> print(W) # 更新后的值
[0.9 2.1]
"""
if timestep is None:
self.t += 1
t = self.t
else:
t = timestep
for idx, param in enumerate(params):
for key in ['W', 'b', 'gamma', 'beta']:
if key not in param or 'd' + key not in param:
continue
theta = param[key] # 参数值
grad = param['d' + key] # 梯度
# 创建唯一键
state_key = f"{idx}_{key}"
# 初始化状态
if state_key not in self.m_states:
self.m_states[state_key] = np.zeros_like(theta)
self.v_states[state_key] = np.zeros_like(theta)
# 更新矩估计
# 这块看不懂，应该就是所谓的动量了
# 就是超某个方向变化时，会加速变化，进而快速收敛
self.m_states[state_key] = self.beta1 * self.m_states[state_key] + (1 - self.beta1) * grad
# 如果一直是来回抖动 就需要减少步长了
# 如果一直某个方向，就需要增大步长了
self.v_states[state_key] = self.beta2 * self.v_states[state_key] + (1 - self.beta2) * (grad ** 2)
# 偏差修正
m_hat = self.m_states[state_key] / (1 - self.beta1 ** t)
v_hat = self.v_states[state_key] / (1 - self.beta2 ** t)
# 更新参数
# 最终更新公式 方向由动量决定，步幅由历史抖动决定。
theta -= self.lr * m_hat / (np.sqrt(v_hat) + self.eps)
def zero_grad(self, params):
"""
清零所有梯度
参数:
params (list): 参数列表
返回: 无
"""
for param in params:
for key in param:
if key.startswith('d') and isinstance(param[key], np.ndarray):
param[key].fill(0)-
一轮训练结束
-
保存模型
需要保存的内容很多，配置、词表的对应关系，模型权重等。像7B模型，7B就是有70亿个参数之巨。
def save_model(model, tokenizer, config, filepath, include_weights=True):
"""
保存模型到 JSON 文件
保存内容:
- 配置参数（config）
- 词表（tokenizer）
- 模型权重（各层的 W, b 参数）
参数:
model (Transformer): Transformer 模型
tokenizer (CharTokenizer): 分词器
config (TransformerConfig): 配置
filepath (str): 保存路径，例如 "./model.json"
include_weights (bool): 是否包含模型权重（仅保存配置和词表则设为 False）
返回:
dict: 保存的数据字典
数据样例:
>>> data = save_model(model, tokenizer, config, "model.json")
>>> print(data.keys())
dict_keys(['config', 'vocab', 'weights', 'info'])
"""
data = {
'info': {
'description': 'Transformer model saved in human-readable JSON format',
'format_version': '1.0'
},
'config': config.to_dict(),
'vocab': tokenizer.vocab_to_dict()
}
if include_weights:
weights = {}
params = model.get_params()
for idx, param in enumerate(params):
for key in ['W', 'b', 'gamma', 'beta']:
if key in param:
weights[f"param_{idx}_{key}"] = param[key]
data['weights'] = _numpy_to_list(weights)
# 保存到文件
os.makedirs(os.path.dirname(filepath) if os.path.dirname(filepath) else '.', exist_ok=True)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 打印摘要
file_size = os.path.getsize(filepath)
print(f"模型已保存: {filepath}")
print(f"文件大小: {file_size / 1024:.1f} KB")
print(f"词表大小: {config.vocab_size}")
print(f"参数数量: {sum(v.size for v in weights.values()) if include_weights else 'N/A'}")
return data-
开始对话
-
加载模型
def load_model(filepath, model_class, tokenizer_class, config_class):
"""
从 JSON 文件加载模型
参数:
filepath (str): 模型文件路径
model_class: Transformer 类
tokenizer_class: CharTokenizer 类
config_class: TransformerConfig 类
返回:
tuple: (model, tokenizer, config)
model: 加载权重后的 Transformer 模型
tokenizer: 加载词表后的分词器
config: 配置对象
数据样例:
>>> model, tokenizer, config = load_model("model.json", Transformer, CharTokenizer, TransformerConfig)
>>> print(config.vocab_size)
100
"""
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
# 恢复配置
config = config_class.from_dict(data['config'])
# 恢复分词器
tokenizer = tokenizer_class()
tokenizer.vocab_from_dict(data['vocab'])
# 创建新模型
model = model_class(config)
# 加载权重
if 'weights' in data:
weights = _list_to_numpy(data['weights'])
params = model.get_params()
for idx, param in enumerate(params):
for key in ['W', 'b', 'gamma', 'beta']:
weight_key = f"param_{idx}_{key}"
if weight_key in weights and key in param:
# 确保形状匹配
loaded = weights[weight_key]
if isinstance(loaded, np.ndarray) and loaded.shape == param[key].shape:
param[key] = loaded
else:
print(f"警告: 参数 {weight_key} 形状不匹配，跳过")
print(f"模型已加载: {filepath}")
print(f"词表大小: {tokenizer.vocab_size}")
return model, tokenizer, config生成对话的逻辑
Transformer 生成句子采用的是自回归（Auto-Regressive）机制。首先，Encoder 对输入序列进行编码，
形成上下文语义表示（Memory）；随后，Decoder 以 <SOS>起始符为起点，
在每一步利用 因果掩码（Causal Mask） 确保当前位置只能关注到之前的 token，
结合 Encoder 的 Memory 通过 Cross-Attention 进行解码，
输出当前步的隐层向量。该向量会与词嵌入矩阵（Weight Tying）进行点积计算，
这里的词矩阵是该向量与词典中所有的词放在一个大矩阵中同时计算，即使词典中有几十万的词。也能瞬间得出相似度得分（logits)。
logits是线性变换的意思
得到包含所有词表信息的 Logits，因为要考虑到所有词，才能进行选举出最合适的词，经 Softmax 转化为概率分布。
最后，模型根据 Temperature 和 Top-k 等策略进行采样，
选出下一个 Token 并拼接到输入序列尾部，循环往复，
直到生成 <EOS>终止符或达到最大长度。def chat(self, user_input, max_len=50, temperature=0.8):
"""
进行一轮对话
参数:
user_input (str): 用户输入文本
max_len (int): 最大回复长度
temperature (float): 生成温度
返回:
str: 助手回复
数据样例:
>>> bot = ChatBot(model, tokenizer, config)
>>> bot.chat("你好")
'你好！很高兴见到您。'
>>> bot.chat("今天天气怎么样？")
'我是一个AI助手，无法获取实时天气信息。'
"""
# 构建提示
prompt = self._build_prompt(user_input)
# 生成回复
# “Transformer 生成句子采用的是自回归（Auto-Regressive）机制。首先，Encoder 对输入序列进行编码，
# 形成上下文语义表示（Memory）；随后，Decoder 以 <SOS>起始符为起点，
# 在每一步利用 因果掩码（Causal Mask） 确保当前位置只能关注到之前的 token，
# 结合 Encoder 的 Memory 通过 Cross-Attention 进行解码，
# 输出当前步的隐层向量。该向量会与词嵌入矩阵（Weight Tying）进行点积计算，
# 这里的词矩阵是该向量与词典中所有的词放在一个大矩阵中同时计算，即使词典中有几十万的词。也能瞬间得出相似度得分（logits)。
# logits是线性变换的意思
# 得到包含所有词表信息的 Logits，因为要考虑到所有词，才能进行选举出最合适的词，经 Softmax 转化为概率分布。
# 最后，模型根据 Temperature 和 Top-k 等策略进行采样，
# 选出下一个 Token 并拼接到输入序列尾部，循环往复，
# 直到生成 <EOS>终止符或达到最大长度。”
response = self.generate(prompt, max_len=max_len, temperature=temperature)
# 更新历史
self.history.append((user_input, response))
# 限制历史长度
if len(self.history) > self.max_turns:
self.history = self.history[-self.max_turns:]
return response生成逻辑
def generate(self, prompt, max_len=50, temperature=0.8, top_k=None):
"""
根据提示生成回复
参数:
prompt (str): 输入提示文本
max_len (int): 最大生成长度（token 数）
temperature (float): 温度参数
- 0.1-0.5: 保守/确定性输出
- 0.6-1.0: 平衡（推荐）
- >1.0: 更有创造性/随机
top_k (int|None): 只从概率最高的 k 个 token 中采样，None 表示不限制
返回:
str: 生成的文本
数据样例:
>>> bot = ChatBot(model, tokenizer, config)
>>> response = bot.generate("用户: 你好\\n助手: ", max_len=20, temperature=0.8)
>>> print(response)
"你好！有什么我可以帮您的吗？"
"""
# 编码输入
src_ids = self.tokenizer.encode(prompt, add_sos=True, add_eos=False)
src = np.array([src_ids], dtype=np.int32)
# 生成 这一步最重要
tgt_ids = self.model.generate(
src,
max_len=max_len + len(src_ids),
start_token=self.tokenizer.SOS_ID,
end_token=self.tokenizer.EOS_ID,
temperature=temperature
)[0]
# 解码（只取生成的新部分，不包括输入）
generated_text = self.tokenizer.decode(tgt_ids, remove_special=True)
# 移除输入部分（prompt）
if generated_text.startswith(prompt):
response = generated_text[len(prompt):].strip()
else:
# 如果没有完全匹配，尝试找到 "助手: " 之后的内容
marker = "助手: "
if marker in generated_text:
# 找到最后一次出现
idx = generated_text.rfind(marker)
response = generated_text[idx + len(marker):].strip()
else:
response = generated_text.strip()
# 清理：截断到下一个 "用户:" 之前
user_marker = "用户:"
if user_marker in response:
response = response[:response.index(user_marker)].strip()
return responsedef generate(self, src, max_len=50, start_token=1, end_token=2, temperature=1.0):
"""
自回归生成（推理模式）
逐 token 生成目标序列，每次将已生成的序列作为 Decoder 输入。
参数:
src (numpy.ndarray): 源序列，形状 (batch_size, src_len)
max_len (int): 最大生成长度
start_token (int): <SOS> token ID
end_token (int): <EOS> token ID
temperature (float): 温度参数，控制随机性
- 1.0: 标准采样
- 0.1-0.5: 更确定性（贪婪）
- >1.0: 更随机
返回:
list[list[int]]: 生成的 token ID 序列
示例: [[1, 5, 8, 3, 2]]
"""
batch_size = src.shape[0]
# 初始化目标序列（只有 <SOS>）
tgt = np.full((batch_size, 1), start_token, dtype=np.int32)
# 预计算 Encoder 输出（避免重复计算）
src_mask = create_padding_mask(src)
src_emb = self.src_embed.forward(src) * self.scale
src_emb = self.pos_encoding.forward(src_emb)
memory = self.encoder.forward(src_emb, src_mask, training=False)
for _ in range(max_len - 1):
tgt_len = tgt.shape[1]
# 创建掩码
tgt_padding_mask = create_padding_mask(tgt)
look_ahead = create_look_ahead_mask(tgt_len)
look_ahead = look_ahead[np.newaxis, np.newaxis, :, :]
tgt_mask = np.maximum(tgt_padding_mask, look_ahead)
# 目标嵌入
tgt_emb = self.tgt_embed.forward(tgt) * self.scale
tgt_emb = self.pos_encoding.forward(tgt_emb)
# Decoder
dec_output = self.decoder.forward(
tgt_emb, memory, tgt_mask, src_mask, training=False
)
# 输出投影（只取最后一个位置）
logits = dec_output[:, -1:, :] @ self.W_out + self.b_out # (B, 1, V)
logits = logits[:, 0, :] # (B, V)
# 应用温度
logits = logits / temperature
# Softmax 获取概率
probs = softmax(logits, axis=-1)
# 采样（贪婪: 取最大概率；随机: 按概率采样）
if temperature < 0.1:
next_token = np.argmax(probs, axis=-1, keepdims=True)
else:
next_token = np.array(
[np.random.choice(self.config.vocab_size, p=p) for p in probs]
)[:, np.newaxis]
# 拼接到目标序列
tgt = np.concatenate([tgt, next_token.astype(np.int32)], axis=1)
# 检查是否所有序列都生成了 <EOS>
if np.all(next_token == end_token):
break
return tgt.tolist()
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
6
0- 0
已为社区贡献3条内容
所有评论(0)