深度学习常用组件(下):手写 Adam 优化器 + 中文文本分类完整实战

开篇:前两篇搭好了"骨架",这篇注入"灵魂"

回顾一下我们的旅程:

  • 上篇:Embedding / Padding / Pooling / Dropout / BatchNorm——数据的进出通道;
  • 中篇:RNN / LSTM——按顺序阅读并记忆的核心引擎。

但到目前为止,网络里的参数都是随机初始化的,什么都"不会"。让参数从随机变成有用的过程叫训练,而训练的执行者是优化器。这篇分两部分:

  1. 手写 Adam:当今最流行的优化器,我们把它的更新公式逐行用 NumPy 实现,并和 PyTorch 对答案;
  2. 完整实战:把三篇学过的所有组件组装成一个"中文句子关键词分类器",从造数据到训练到推理,一个文件跑通。

一、从 SGD 到 Adam:优化器在干什么

1.1 SGD 的局限

最原始的梯度下降(SGD)只有一行:

θ ← θ - η · ∇L(θ)

η 是全局固定的学习率。问题很明显:

  • 学习率难调:大了来回震荡,小了收敛如龟速;
  • 一刀切:所有参数用同一个学习率。但有的参数梯度又大又稳(该小步走),有的梯度稀疏又小(该大步走),一碗水端平并不合理。

1.2 Adam 的思路:给每个参数配"自适应学习率"

一句话版本:Adam 每次更新参数前,先回答两个问题——"最近梯度一直朝哪走?"(决定方向)和"最近梯度抖动大不大?"(决定步子大小)。

Adam(Adaptive Moment Estimation)为每个参数单独维护两个统计量

  • 一阶矩 m:梯度的指数移动平均——"最近梯度大概朝哪走"(动量,防止摇摆);
  • 二阶矩 v:梯度平方的指数移动平均——"最近梯度波动有多大"(波动大就自动缩小步长)。

1.3 逐行拆解 Adam 公式(含数值实例)

完整更新流程(每步):

g  = 当前梯度
m  = β1·m + (1-β1)·g          ← 更新一阶矩(默认 β1=0.9)
v  = β2·v + (1-β2)·g²         ← 更新二阶矩(默认 β2=0.999)
m̂  = m / (1 - β1ᵗ)             ← 偏差修正
v̂  = v / (1 - β2ᵗ)             ← 偏差修正
w  = w - lr · m̂ / (√v̂ + ε)    ← 参数更新

符号多不要怕,我们设定一个具体场景,跟着数字走一遍:参数 w = 1.0,当前梯度 g = 0.5,第 1 步(t=1),lr = 0.001

① m(一阶矩):梯度方向的"滑动平均" → 决定朝哪走

m = β1·m + (1-β1)·g  =  0.9×0 + 0.1×0.5  =  0.05

这就是指数移动平均——程序员可以把它理解为一个不用存历史数据的"近期平均值":新平均值 = 0.9×旧平均值 + 0.1×新数据。β1=0.9 意味着"90% 相信历史,10% 采纳新数据"。

作用类似动量/惯性:如果梯度连续 10 步都指向左边,哪怕这一步突然指向右,m 依然 mostly 朝左,不会被单步的噪声带偏。就像开车下坡时的惯性会推着你继续走。

② v(二阶矩):梯度抖动的"滑动平均" → 决定步子大小

v = β2·v + (1-β2)·g²  =  0.999×0 + 0.001×0.5²  =  0.00025

结构和 m 完全一样,只是平均的对象从 g 换成了 。平方有两个作用:消掉正负号(+5 和 -5 的"抖动幅度"一样大)、放大大梯度。所以 v 衡量的就是"最近梯度的波动量级"——可以联想上篇讲的标准差:v 大 ≈ 梯度很"散"、很不稳定。

③ 偏差修正:把被 0 初始化"压低"的估计拉回来

m̂ = m / (1 - β1ᵗ)  =  0.05 / (1 - 0.9)    =  0.5
v̂ = v / (1 - β2ᵗ)  =  0.00025 / (1 - 0.999) =  0.25

为什么需要这步?因为 m、v 初始值是 0。看上面的数字:第一步真实梯度明明是 0.5,但算出来 m = 0.05——被初始的 0 拖累了,只有真实值的十分之一。除以 (1-0.9¹) = 0.1 正好放大 10 倍,还原回 0.5。

而且修正因子 1-βᵗ 会随 t 增大趋近于 1(t=100 时 1-0.9¹⁰⁰ ≈ 1),所以这个修正只在训练初期起作用,之后自动退出——非常优雅的设计。

④ 最终更新:方向 ÷ 波动

w = w - lr · m̂ / (√v̂ + ε)  =  1.0 - 0.001 × 0.5/(√0.25 + ε)  =  0.999

这是整个 Adam 最精华的一行,读法是:

  • 分子 m̂:提供方向(和一点惯性);
  • 分母 √v̂:梯度波动越大,分母越大,实际步长自动越小。这就是"自适应学习率"的含义——不需要你手动调,每个参数根据自己的历史波动自动决定走多大步。波动大的参数走小步(怕震荡),波动小而稳定的参数走大步(放心冲);
  • ε:防除零的小常数,和 BatchNorm 里的 eps 一个作用。

一个有趣的推论:第 1 步时 m̂ ≈ g、v̂ ≈ g²,所以 m̂/√v̂ ≈ g/|g| = ±1——第一步所有参数的更新量都约等于 lr,只有符号不同。这保证了 Adam 在训练初期不会发疯。

串起来记:m 管方向(动量),v 管步长(自适应),偏差修正管初期,ε 管数值稳定。optimizer.step() 就是对每个参数做这一组数组运算。

1.4 手写一步 Adam,和 PyTorch 逐位对答案

老规矩:从 PyTorch 抠出梯度和初始权重,手动按公式算一步,对比结果。

import torch
import torch.nn as nn
import numpy as np

torch.manual_seed(42)
np.random.seed(42)

# ── 超参数(PyTorch Adam 默认值)──
LR, BETA1, BETA2, EPS = 1e-3, 0.9, 0.999, 1e-8

# ── 一个极简网络:单层线性,便于核验 ──
net = nn.Linear(4, 2, bias=False)
x      = torch.randn(8, 4)       # 8 个样本,4 维输入
target = torch.randn(8, 2)

optimizer = torch.optim.Adam(net.parameters(), lr=LR,
                             betas=(BETA1, BETA2), eps=EPS)

# ══ Part 1: PyTorch 走一步 ══
W_init = net.weight.data.clone()          # 保存初始权重

loss = nn.MSELoss()(net(x), target)
optimizer.zero_grad()
loss.backward()

grad_torch = net.weight.grad.clone()      # 反向传播拿到梯度
optimizer.step()                          # PyTorch Adam 更新一步
W_after_torch = net.weight.data.clone()

# ══ Part 2: 手动按公式走一步 ══
W_manual = W_init.numpy().copy()
grad     = grad_torch.numpy().copy()

t  = 0
mt = np.zeros_like(W_manual)              # 一阶矩初始化为 0
vt = np.zeros_like(W_manual)              # 二阶矩初始化为 0

t  = t + 1                                # step 1
mt = BETA1 * mt + (1 - BETA1) * grad      # 一阶矩更新
vt = BETA2 * vt + (1 - BETA2) * grad**2   # 二阶矩更新

mth = mt / (1 - BETA1**t)                 # 偏差修正
vth = vt / (1 - BETA2**t)

W_manual = W_manual - LR * mth / (np.sqrt(vth) + EPS)   # 权重更新

# ══ Part 3: 对答案 ══
diff = np.abs(W_after_torch.numpy() - W_manual)
print(f"最大误差: {diff.max():.2e}")
print("结论:", "完全一致" if diff.max() < 1e-6 else "有差异,检查公式")
# 最大误差约 1e-9 量级 → 手写版与 PyTorch 完全一致

跑通这个实验后,Adam 对你而言就再没有黑盒了:所谓优化器,无非是在 optimizer.step() 里对每个参数做了一组确定性的数组运算


二、终极实战:中文关键词分类器(把三篇的知识串起来)

2.1 任务定义

句子中含有关键词(好/棒/赞/喜欢/满意)→ 正样本(1),否则 → 负样本(0)。

  • 正例:"这家餐厅真的很棒,下次还来"
  • 负例:"今天天气阴沉,出门忘带雨伞"

模型架构(每一层都是前三篇讲过的组件):

Embedding → RNN → MaxPooling → BatchNorm → Dropout → Linear → Sigmoid

优化器用 Adam,损失用 MSELoss。数据不用下载,纯 Python 现场生成。

2.2 数据生成:模板造句

import random
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

# ─── 超参数 ───
SEED        = 42
N_SAMPLES   = 4000
MAXLEN      = 32
EMBED_DIM   = 64
HIDDEN_DIM  = 64
LR          = 1e-3
BATCH_SIZE  = 64
EPOCHS      = 20
TRAIN_RATIO = 0.8

random.seed(SEED)
torch.manual_seed(SEED)

POS_KEYS = ['好', '棒', '赞', '喜欢', '满意']

TEMPLATES_POS = [
    '这家{}真的很{},下次还来',
    '这款{}设计让我{}',
    '{}的服务态度让我感到{}',
    '{}体验非常{}',
    '这次购物感觉{}极了',
]

TEMPLATES_NEG = [
    '今天天气阴沉,出门忘带雨伞',
    '这部电影情节比较平淡',
    '下午开了三个小时的会议',
    '路上堵车耽误了不少时间',
    '这道题做了很久还没解出来',
    '最近工作任务比较繁重',
    '超市里人很多,排队结账',
    '这个季节换季容易感冒',
    '今天作业布置得有点多',
    '公交车又晚点了十分钟',
]

OBJ_WORDS = ['店铺', '餐厅', '产品', '服务', '环境', '系统', '设计', '课程']
ADJ_WORDS = ['方便', '简洁', '独特', '舒适', '高效']

def make_positive():
    kw, tmpl, obj = (random.choice(POS_KEYS),
                     random.choice(TEMPLATES_POS),
                     random.choice(OBJ_WORDS))
    try:
        sent = tmpl.format(obj, kw)
    except Exception:
        sent = obj + kw + random.choice(ADJ_WORDS)
    # 30% 概率再随机插入一个关键词,增加多样性
    if random.random() < 0.3:
        extra = random.choice(POS_KEYS)
        pos   = random.randint(0, len(sent))
        sent  = sent[:pos] + extra + sent[pos:]
    return sent

def make_negative():
    base = random.choice(TEMPLATES_NEG)
    if random.random() < 0.4:                 # 40% 概率拼接两句,变长
        base += random.choice(TEMPLATES_NEG)
    return base

def build_dataset(n=N_SAMPLES):
    data = []
    for _ in range(n // 2):
        data.append((make_positive(), 1))
        data.append((make_negative(), 0))
    random.shuffle(data)
    return data

2.3 词表与编码(上篇的知识)

def build_vocab(data):
    vocab = {'<PAD>': 0, '<UNK>': 1}       # 0 留给 pad,1 留给未登录字
    for sent, _ in data:
        for ch in sent:
            if ch not in vocab:
                vocab[ch] = len(vocab)
    return vocab

def encode(sent, vocab, maxlen=MAXLEN):
    ids  = [vocab.get(ch, 1) for ch in sent]
    ids  = ids[:maxlen]                     # 超长截断
    ids += [0] * (maxlen - len(ids))        # 不足补 pad(0)
    return ids

class TextDataset(Dataset):
    def __init__(self, data, vocab):
        self.X = [encode(s, vocab) for s, _ in data]
        self.y = [lb for _, lb in data]

    def __len__(self):
        return len(self.y)

    def __getitem__(self, i):
        return (torch.tensor(self.X[i], dtype=torch.long),
                torch.tensor(self.y[i], dtype=torch.float))

2.4 模型定义(每层都能对上号)

class KeywordRNN(nn.Module):
    """
    Embedding → RNN → MaxPool → BatchNorm → Dropout → Linear → Sigmoid
    """
    def __init__(self, vocab_size, embed_dim=EMBED_DIM,
                 hidden_dim=HIDDEN_DIM, dropout=0.3):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim,
                                      padding_idx=0)      # ← 上篇:pad 行恒为 0
        self.rnn       = nn.RNN(embed_dim, hidden_dim,
                                batch_first=True)         # ← 中篇:循环记忆
        self.bn        = nn.BatchNorm1d(hidden_dim)       # ← 上篇:稳定训练
        self.dropout   = nn.Dropout(dropout)              # ← 上篇:防过拟合
        self.fc        = nn.Linear(hidden_dim, 1)

    def forward(self, x):
        # x: (batch, seq_len)
        e, _ = self.rnn(self.embedding(x))   # (B, L, hidden)
        pooled = e.max(dim=1)[0]             # ← 上篇:MaxPool 抓关键词峰值
        pooled = self.dropout(self.bn(pooled))
        out = torch.sigmoid(self.fc(pooled).squeeze(1))  # (B,)
        return out

数据流逐步看形状变化:

(batch, 32)  --Embedding-->  (batch, 32, 64)
             --RNN------->   (batch, 32, 64)   每步一个隐藏向量
             --MaxPool--->   (batch, 64)       序列维消失,一句话一个向量
             --BN+Dropout->  (batch, 64)
             --Linear---->   (batch, 1) → squeeze → (batch,)
             --Sigmoid--->   0~1 概率,>0.5 判正样本

为什么用 MaxPooling 而不是取最后一步?因为这个任务的答案(关键词)可能出现在句子任何位置,MaxPool 会在整个序列上挑出"最像关键词"的那个信号;而 RNN 最后一步的记忆对长句子开头的关键词可能已经淡忘。

2.5 训练与评估(Adam 上场)

def evaluate(model, loader):
    model.eval()                       # ← 关键:关掉 Dropout,BN 用累计统计量
    correct = total = 0
    with torch.no_grad():
        for X, y in loader:
            pred     = (model(X) > 0.5).long()
            correct += (pred == y.long()).sum().item()
            total   += len(y)
    return correct / total

def train():
    print("生成数据集...")
    data  = build_dataset(N_SAMPLES)
    vocab = build_vocab(data)
    print(f"  样本数:{len(data)},词表大小:{len(vocab)}")

    split       = int(len(data) * TRAIN_RATIO)
    train_data  = data[:split]
    val_data    = data[split:]

    train_loader = DataLoader(TextDataset(train_data, vocab),
                              batch_size=BATCH_SIZE, shuffle=True)
    val_loader   = DataLoader(TextDataset(val_data, vocab),
                              batch_size=BATCH_SIZE)

    model     = KeywordRNN(vocab_size=len(vocab))
    criterion = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=LR)   # ← 本篇主角

    total_params = sum(p.numel() for p in model.parameters())
    print(f"  模型参数量:{total_params:,}\n")

    for epoch in range(1, EPOCHS + 1):
        model.train()                    # ← 关键:开启 Dropout,BN 用 batch 统计量
        total_loss = 0.0
        for X, y in train_loader:
            pred = model(X)
            loss = criterion(pred, y)
            optimizer.zero_grad()        # 清空旧梯度
            loss.backward()              # 反向传播
            optimizer.step()             # Adam 更新参数
            total_loss += loss.item()

        avg_loss = total_loss / len(train_loader)
        val_acc  = evaluate(model, val_loader)
        print(f"Epoch {epoch:2d}/{EPOCHS}  loss={avg_loss:.4f}  val_acc={val_acc:.4f}")

    # ── 推理示例 ──
    print("\n--- 推理示例 ---")
    model.eval()
    test_sents = [
        '这款产品真的很棒,非常满意',
        '今天天气有点阴沉,出门带了雨伞',
        '服务太赞了,下次还来',
        '等了很久公交车终于来了',
    ]
    with torch.no_grad():
        for sent in test_sents:
            ids   = torch.tensor([encode(sent, vocab)], dtype=torch.long)
            prob  = model(ids).item()
            label = '正样本' if prob > 0.5 else '负样本'
            print(f"  [{label}({prob:.2f})]  {sent}")

if __name__ == '__main__':
    train()

运行后你会看到 loss 一路下降,验证准确率很快冲到 0.99 以上,最后四句推理全部判断正确(正例概率接近 1.0,负例接近 0.0)。

2.6 训练循环里的三个"铁律"再强调一遍

  1. zero_grad() → backward() → step() 顺序不能乱:PyTorch 梯度默认累加,不清零就会把上个 batch 的梯度也算进去;
  2. model.train() / model.eval() 必须切换:Dropout 和 BatchNorm 在两个模式下行为完全不同;
  3. 推理包在 torch.no_grad():省内存、省算力,还避免误改计算图。

三、全系列总结:一张组件地图

三篇走完,从文字到训练好的分类器,每个组件的位置和职责如下:

原始文本
   │
   ├─ encode(截断 + 补 pad)          ← 上篇:Padding
   ▼
token id 序列  [batch, seq_len]
   │
   ├─ nn.Embedding(padding_idx=0)      ← 上篇:查表 + pad 冻结
   ▼
词向量序列  [batch, seq_len, embed_dim]
   │
   ├─ nn.RNN / nn.LSTM                 ← 中篇:顺序阅读 + 记忆
   ▼
每步隐藏状态  [batch, seq_len, hidden]
   │
   ├─ MaxPooling                       ← 上篇:抓关键词峰值,压缩序列
   ▼
句子向量  [batch, hidden]
   │
   ├─ BatchNorm + Dropout              ← 上篇:稳定训练 + 防过拟合
   ├─ Linear + Sigmoid
   ▼
概率 → 标签
   │
   ├─ MSELoss 计算损失
   ├─ backward() 反向传播
   └─ Adam.step() 更新参数             ← 下篇:自适应学习率优化

回顾每篇的核心收获

  • 上篇:Embedding 是可训练的查表;Padding 对齐 batch;Pooling 压缩序列;Dropout/BatchNorm 是训练辅助,且训练/推理行为不同(记得切 train()/eval())。
  • 中篇:RNN 就是"带状态的 for 循环";LSTM 用三个门保护长期记忆;双向就是正反向两个独立实例拼输出。手写一遍再对答案,是祛魅的最好方式。
  • 下篇:Adam = 动量(一阶矩)+ 自适应步长(二阶矩)+ 偏差修正;所有组件可以几十行代码组装成一个真实可用的文本分类器。

可以继续探索的方向

  1. 把 RNN 换成 LSTM 或双向 LSTM,观察小数据集上的效果差异(中篇代码可直接复用);
  2. 把 MaxPooling 换成 AvgPooling,对比哪个更抗干扰;
  3. 手动实现两步、三步 Adam,验证 t 增大后偏差修正因子确实趋近于 1;
  4. 给模型加上 LayerNorm 试试(课件里提到过:NLP 领域它比 BatchNorm 更常用,不依赖 batch size)。

动手改、动手跑、动手对答案——组件不多,但全部吃透之后,你看任何 NLP 模型的源码都会觉得"似曾相识"。

©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

友情链接更多精彩内容