损失函数与反向传播

"""CrossEntropyLoss 自带 softmax 的证明，以及手动再加一次 softmax 的后果。

手算路径：
    1) 对 logits 做 softmax，得到每一类的概率 p
    2) 取出真实类别那一格的概率 p[label]
    3) 这一条样本的损失 = -log(p[label])
    4) 一批样本取平均
把这四步写出来，和 nn.CrossEntropyLoss() 的结果对账。
"""
import math

import torch
import torch.nn as nn


def manual_cross_entropy(logits, labels):
    """纯 Python 版交叉熵，只用 math，不借助任何 torch 的损失函数。"""
    losses = []
    for row, label in zip(logits.tolist(), labels.tolist()):
        m = max(row)                                  # 减最大值，防指数溢出
        exps = [math.exp(v - m) for v in row]
        total = sum(exps)
        p_true = exps[label] / total
        losses.append(-math.log(p_true))
    return sum(losses) / len(losses)


def main():
    # 真实值必须是 int64 的类别下标，不是 one-hot，也不是浮点
    y_true = torch.tensor([1, 2], dtype=torch.int64)
    # 预测值是没过任何激活函数的原始分数 logits
    y_pred = torch.tensor([[0.2, 0.6, 0.2],
                           [0.1, 0.8, 0.1]], requires_grad=True)

    official = nn.CrossEntropyLoss()(y_pred, y_true)
    manual = manual_cross_entropy(y_pred.detach(), y_true)

    print('nn.CrossEntropyLoss = %.10f' % official.item())
    print('手算四步          = %.10f' % manual)
    print('两者之差 = %.12f —— 说明这个损失函数内部确实做了 softmax'
          % abs(official.item() - manual))

    # ---------------------------------------------------------- 错误示范
    # 如果模型最后一层已经加了 softmax，再交给 CrossEntropyLoss，
    # 等于把概率当 logits 又压了一次。
    probs = torch.softmax(y_pred.detach(), dim=-1)
    wrong = nn.CrossEntropyLoss()(probs, y_true)
    print('\n先手动 softmax 再进 CrossEntropyLoss = %.10f' % wrong.item())
    print('正确写法 = %.10f' % official.item())
    print('两次 softmax 之后概率被压平，损失数值失真，梯度也被削弱')
    print('压平前的概率分布：', [round(v, 4) for v in probs[0].tolist()])
    print('再压一次之后    ：',
          [round(v, 4) for v in torch.softmax(probs, dim=-1)[0].tolist()])

    # ---------------------------------------------------------- 取出预测类别
    # 推理时才需要 softmax，而且只是为了把分数变成好看的概率；
    # 只想知道预测哪一类的话，直接对 logits 取 argmax 就够了。
    print('\nargmax(logits) =', y_pred.argmax(dim=-1).tolist())
    print('argmax(softmax) =', probs.argmax(dim=-1).tolist(),
          '—— softmax 不改变大小顺序')

    # ---------------------------------------------------------- 二分类的对照
    # 二分类换成 sigmoid + BCELoss：预测值必须已经落在 (0,1)
    p = torch.tensor([0.6901, 0.5459, 0.2469], requires_grad=True)
    t = torch.tensor([0., 1., 0.])
    bce = nn.BCELoss()(p, t)
    hand = -sum(ti * math.log(pi) + (1 - ti) * math.log(1 - pi)
                for pi, ti in zip(p.tolist(), t.tolist())) / 3
    print('\nnn.BCELoss = %.10f   手算 = %.10f' % (bce.item(), hand))
    # 若模型最后一层没加 sigmoid，就该用 BCEWithLogitsLoss，它内部会补上


if __name__ == '__main__':
    main()

"""MSE / MAE / SmoothL1 三条曲线：比损失值更该看的是它们的梯度。

选损失函数其实是在选「误差变大时，惩罚涨得多快」：
    MSE      惩罚按误差平方涨，梯度 = 2·误差，离群点能把梯度顶上天
    MAE      惩罚按误差线性涨，梯度恒为 ±1，稳，但零点处不可导
    SmoothL1 误差小于 1 时像 MSE，大于 1 时像 MAE，两头的好处都要
"""
import torch
import torch.nn as nn


def loss_and_grad(criterion, err):
    """给定一个误差值，返回该损失函数的损失值与对预测值的梯度。"""
    pred = torch.tensor([err], requires_grad=True)
    target = torch.tensor([0.0])
    loss = criterion(pred, target)
    loss.backward()
    return loss.item(), pred.grad.item()


def table():
    mse, mae, smooth = nn.MSELoss(), nn.L1Loss(), nn.SmoothL1Loss()

    print('%8s | %10s %10s | %10s %10s | %10s %10s'
          % ('误差', 'MSE 值', 'MSE 梯度', 'MAE 值', 'MAE 梯度',
             'SL1 值', 'SL1 梯度'))
    for err in (0.1, 0.5, 1.0, 2.0, 5.0, 20.0, 100.0):
        lm, gm = loss_and_grad(mse, err)
        la, ga = loss_and_grad(mae, err)
        ls, gs = loss_and_grad(smooth, err)
        print('%8.1f | %10.2f %10.2f | %10.2f %10.2f | %10.2f %10.2f'
              % (err, lm, gm, la, ga, ls, gs))

    # 看最后一行：误差 100 时 MSE 的梯度是 200，MAE 和 SmoothL1 都只有 1。
    # 一条标错的样本就能让 MSE 主导整批梯度，这就是「离群点梯度爆炸」。


def outlier_effect():
    """同一批数据，只把一条改成离群点，看三种损失各自变化多少。"""
    pred = torch.tensor([1.0, 2.0, 3.0, 4.0])
    clean = torch.tensor([1.1, 2.1, 2.9, 3.9])
    dirty = clean.clone()
    dirty[3] = 100.0                       # 第四条标注错了

    for name, criterion in (('MSE', nn.MSELoss()),
                            ('MAE', nn.L1Loss()),
                            ('SmoothL1', nn.SmoothL1Loss())):
        a = criterion(pred, clean).item()
        b = criterion(pred, dirty).item()
        print('%-9s 干净数据 %8.3f -> 含一条离群点 %10.3f   放大 %8.1f 倍'
              % (name, a, b, b / a))


def zero_point():
    """MAE 在零点不光滑：左右两侧的导数是 -1 和 +1，跳变。"""
    for err in (-0.001, -1e-8, 0.0, 1e-8, 0.001):
        _, g = loss_and_grad(nn.L1Loss(), err)
        print('误差 %12.1e 时 MAE 的梯度 = %+.1f' % (err, g))
    print('梯度不随误差变小而变小，逼近最优解时会在极小值附近来回跨步')

    print()
    for err in (-0.001, -1e-8, 0.0, 1e-8, 0.001):
        _, g = loss_and_grad(nn.SmoothL1Loss(), err)
        print('误差 %12.1e 时 SmoothL1 的梯度 = %+.1e' % (err, g))
    print('SmoothL1 在零点附近梯度跟着误差一起变小，能稳稳停住')


if __name__ == '__main__':
    table()
    print('-' * 60)
    outlier_effect()
    print('-' * 60)
    zero_point()

"""手推一次反向传播，再让 PyTorch 算一遍，两个结果必须对得上。

网络：x(2,5) -> z = x @ w + b -> (2,3)，损失用 MSE，真实值全 0。
把链式法则一步步写出来，就知道 autograd 没有魔法。
"""
import torch

torch.manual_seed(42)

x = torch.ones(2, 5)                       # 2 个样本、5 个特征，全设成 1 方便手算
y = torch.zeros(2, 3)                      # 真实值全 0
w = torch.randn(5, 3, requires_grad=True)  # 5 个输入 -> 3 个输出
b = torch.randn(3, requires_grad=True)

# ---------- 前向 ----------
z = x @ w + b
loss = torch.nn.MSELoss()(z, y)
print('loss =', loss.item())

# ---------- 反向（交给 autograd）----------
loss.backward()
print('autograd 给的 w.grad:\n', w.grad)
print('autograd 给的 b.grad:\n', b.grad)

# ---------- 反向（自己推）----------
# MSE: loss = (1/N) * Σ (z - y)^2，这里 N = 2*3 = 6，且 y = 0
#   ∂loss/∂z = (2/N) * (z - y) = z / 3
#   z = x·w + b，所以 ∂z/∂w = x，∂z/∂b = 1
#   链式相乘并对样本维求和：
#      ∇w = xᵀ @ (z/3)
#      ∇b = (z/3) 按样本维求和
grad_z = (2.0 / y.numel()) * (z - y)
manual_w = x.t() @ grad_z
manual_b = grad_z.sum(dim=0)

print('手推 w 的最大误差 = %.10f' % (w.grad - manual_w).abs().max().item())
print('手推 b 的最大误差 = %.10f' % (b.grad - manual_b).abs().max().item())

"""梯度消失与梯度爆炸：把每一层收到的梯度大小逐层打出来。

链式法则是连乘。每一段的倍率小于 1，乘到前面就趋近 0（梯度消失）；
每一段大于 1，乘到前面就炸上天（梯度爆炸）。
这份代码把「倍率」这件事直接做成实验。
"""
import torch
import torch.nn as nn


def deep_net(depth, activation, weight_std=None):
    """搭一个 depth 层的窄网络，每层都是 16 -> 16。"""
    layers = []
    for _ in range(depth):
        linear = nn.Linear(16, 16)
        if weight_std is not None:
            nn.init.normal_(linear.weight, mean=0.0, std=weight_std)
            nn.init.zeros_(linear.bias)
        layers.append(linear)
        layers.append(activation())
    layers.append(nn.Linear(16, 1))
    return nn.Sequential(*layers)


def layer_grad_norms(model, x, y):
    """反向一次，返回每个线性层权重梯度的范数（从输入侧数到输出侧）。"""
    loss = ((model(x) - y) ** 2).mean()
    model.zero_grad()
    loss.backward()
    return [m.weight.grad.norm().item()
            for m in model if isinstance(m, nn.Linear)]


def compare_activation():
    torch.manual_seed(0)
    x = torch.randn(64, 16)
    y = torch.randn(64, 1)

    for name, act in (('Sigmoid', nn.Sigmoid), ('Tanh', nn.Tanh), ('ReLU', nn.ReLU)):
        torch.manual_seed(0)
        model = deep_net(10, act)
        norms = layer_grad_norms(model, x, y)
        print('%-8s 第 1 层 %.3e   第 5 层 %.3e   最后一层 %.3e   首尾比 %.2e'
              % (name, norms[0], norms[4], norms[-1],
                 norms[0] / max(norms[-1], 1e-30)))

    print('sigmoid 的导数上限是 0.25，十层连乘最好情形也只有 0.25^10 ≈ 9.5e-07，')
    print('所以靠近输入的那几层几乎收不到梯度，等于没在训练。')


def explode():
    """把权重初始化放大，制造梯度爆炸。"""
    torch.manual_seed(0)
    x = torch.randn(64, 16)
    y = torch.randn(64, 1)

    for std in (0.1, 0.5, 1.0, 2.0):
        torch.manual_seed(0)
        model = deep_net(10, nn.ReLU, weight_std=std)
        norms = layer_grad_norms(model, x, y)
        first = norms[0]
        flag = ' <- 已经溢出成 nan/inf' if not torch.isfinite(
            torch.tensor(first)) else ''
        print('权重 std=%-4s  第 1 层梯度范数 = %.3e%s' % (std, first, flag))

    print('std 每放大一档，连乘十层就放大十次方，很快超出 float32 能表示的范围。')


def clip_demo():
    """梯度裁剪：把整体梯度范数按比例压回阈值以内，方向不变、长度变短。"""
    torch.manual_seed(0)
    model = deep_net(6, nn.ReLU, weight_std=1.5)
    x, y = torch.randn(64, 16), torch.randn(64, 1)

    loss = ((model(x) - y) ** 2).mean()
    model.zero_grad()
    loss.backward()

    before = torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
    after = torch.sqrt(sum((p.grad ** 2).sum() for p in model.parameters()))
    print('裁剪前的总梯度范数 = %.4f' % before.item())
    print('裁剪后的总梯度范数 = %.4f（阈值 1.0）' % after.item())
    print('clip_grad_norm_ 必须放在 backward() 之后、step() 之前')


if __name__ == '__main__':
    compare_activation()
    print('-' * 60)
    explode()
    print('-' * 60)
    clip_demo()

"""五个损失函数的最小调用：先看清各自吃什么形状、什么类型。"""
import torch
import torch.nn as nn

# ---------- 多分类：CrossEntropyLoss ----------
# 真实值是类别下标，必须 int64；不用自己做 one-hot
y_true = torch.tensor([1, 2], dtype=torch.int64)
# 预测值是每一类的原始分数 logits，不要提前过 softmax
y_pred = torch.tensor([[0.2, 0.6, 0.2],
                       [0.1, 0.8, 0.1]], requires_grad=True)
print('CrossEntropy:', nn.CrossEntropyLoss()(y_pred, y_true).item())

# ---------- 二分类：BCELoss ----------
# 预测值必须已经过 sigmoid，落在 (0,1)；真实值是 0/1 的 float
y_pred = torch.tensor([0.6901, 0.5459, 0.2469], requires_grad=True)
y_true = torch.tensor([0, 1, 0], dtype=torch.float32)
print('BCE:', nn.BCELoss()(y_pred, y_true).item())

# ---------- 回归三兄弟 ----------
y_pred = torch.tensor([1.0, 1.0, 1.9], requires_grad=True)
y_true = torch.tensor([2.0, 2.0, 2.0])
print('L1  (MAE):', nn.L1Loss()(y_pred, y_true).item())        # (1+1+0.1)/3
print('MSE (L2) :', nn.MSELoss()(y_pred, y_true).item())       # (1+1+0.01)/3
print('SmoothL1 :', nn.SmoothL1Loss()(y_pred, y_true).item())  # 近处像 L2，远处像 L1

"""梯度累加实验：证明「不清零」到底会错成什么样。"""
import torch

# 同一个前向算三次反向，梯度会一路叠加
x = torch.tensor(3.0, requires_grad=True)
for i in range(1, 4):
    y = x ** 2                 # dy/dx = 2x = 6
    y.backward()
    print('第 %d 次反向后 x.grad = %.1f（正确值应当是 6.0）' % (i, x.grad.item()))
# 输出 6.0 / 12.0 / 18.0 —— 第二次起就已经错了

print('---- 每次反向前清零 ----')
x = torch.tensor(3.0, requires_grad=True)
for i in range(1, 4):
    y = x ** 2
    if x.grad is not None:
        x.grad.zero_()         # 训练循环里 optimizer.zero_grad() 干的就是这件事
    y.backward()
    print('第 %d 次反向后 x.grad = %.1f' % (i, x.grad.item()))
# 三次都是 6.0

"""两层小网络的反向传播：先纯手算，再让 PyTorch 算，两边逐个参数对账。

网络结构（激活函数全用 sigmoid，损失用 0.5·Σ(target − out)²）：
    输入 i1=0.05, i2=0.10
    隐藏层 h1, h2    权重 w1..w4，共享偏置 b1=0.35
    输出层 o1, o2    权重 w5..w8，共享偏置 b2=0.60
    目标 t1=0.01, t2=0.99

全程没有随机数，每个中间值都能用计算器复算。
"""
import math

import torch


def sigmoid(z):
    return 1.0 / (1.0 + math.exp(-z))


def manual():
    i1, i2 = 0.05, 0.10
    w1, w2, w3, w4, b1 = 0.15, 0.20, 0.25, 0.30, 0.35
    w5, w6, w7, w8, b2 = 0.40, 0.45, 0.50, 0.55, 0.60
    t1, t2 = 0.01, 0.99

    # ---------------- 前向 ----------------
    net_h1 = w1 * i1 + w2 * i2 + b1
    out_h1 = sigmoid(net_h1)
    net_h2 = w3 * i1 + w4 * i2 + b1
    out_h2 = sigmoid(net_h2)

    net_o1 = w5 * out_h1 + w6 * out_h2 + b2
    out_o1 = sigmoid(net_o1)
    net_o2 = w7 * out_h1 + w8 * out_h2 + b2
    out_o2 = sigmoid(net_o2)

    e1 = 0.5 * (t1 - out_o1) ** 2
    e2 = 0.5 * (t2 - out_o2) ** 2

    print('前向：net_h1=%.6f out_h1=%.9f' % (net_h1, out_h1))
    print('      net_h2=%.6f out_h2=%.9f' % (net_h2, out_h2))
    print('      net_o1=%.9f out_o1=%.9f' % (net_o1, out_o1))
    print('      net_o2=%.9f out_o2=%.9f' % (net_o2, out_o2))
    print('      E1=%.9f  E2=%.9f  E_total=%.9f' % (e1, e2, e1 + e2))

    # ---------------- 反向：先求 w5 ----------------
    # 链条三段：E 对 out_o1 → out_o1 对 net_o1 → net_o1 对 w5
    dE_dout_o1 = -(t1 - out_o1)                  # = out_o1 - t1
    dout_o1_dnet_o1 = out_o1 * (1 - out_o1)      # sigmoid 的导数形式
    dnet_o1_dw5 = out_h1                         # net_o1 = w5·out_h1 + ...
    dE_dw5 = dE_dout_o1 * dout_o1_dnet_o1 * dnet_o1_dw5

    print('\n反向 w5：%.9f × %.9f × %.9f = %.9f'
          % (dE_dout_o1, dout_o1_dnet_o1, dnet_o1_dw5, dE_dw5))
    print('学习率 0.5 时 w5 更新为 %.9f' % (w5 - 0.5 * dE_dw5))

    # ---------------- 反向：再求 w1 ----------------
    # w1 藏在第一层，它影响 out_h1，而 out_h1 同时喂给 o1 和 o2，
    # 所以「责任」要从两条路径加起来。
    dE1_dnet_o1 = dE_dout_o1 * dout_o1_dnet_o1
    dE2_dnet_o2 = -(t2 - out_o2) * out_o2 * (1 - out_o2)
    dE_dout_h1 = dE1_dnet_o1 * w5 + dE2_dnet_o2 * w7      # 两条路径求和
    dout_h1_dnet_h1 = out_h1 * (1 - out_h1)
    dE_dw1 = dE_dout_h1 * dout_h1_dnet_h1 * i1

    print('\n反向 w1：两条路径相加得 dE/dout_h1 = %.9f' % dE_dout_h1)
    print('        × %.9f × %.2f = %.9f'
          % (dout_h1_dnet_h1, i1, dE_dw1))
    print('学习率 0.5 时 w1 更新为 %.9f' % (w1 - 0.5 * dE_dw1))

    return dE_dw5, dE_dw1


def by_autograd():
    """同一个网络交给 PyTorch，看它算出来是不是同一组数。"""
    i = torch.tensor([0.05, 0.10])
    w_h = torch.tensor([[0.15, 0.20],       # h1 的权重 w1, w2
                        [0.25, 0.30]], requires_grad=True)   # h2 的 w3, w4
    b_h = torch.tensor([0.35, 0.35], requires_grad=True)
    w_o = torch.tensor([[0.40, 0.45],       # o1 的权重 w5, w6
                        [0.50, 0.55]], requires_grad=True)   # o2 的 w7, w8
    b_o = torch.tensor([0.60, 0.60], requires_grad=True)
    t = torch.tensor([0.01, 0.99])

    out_h = torch.sigmoid(w_h @ i + b_h)
    out_o = torch.sigmoid(w_o @ out_h + b_o)
    loss = (0.5 * (t - out_o) ** 2).sum()    # 注意是求和不是求平均

    loss.backward()
    print('\nautograd：loss = %.9f' % loss.item())
    print('autograd：dE/dw5 = %.9f' % w_o.grad[0, 0].item())
    print('autograd：dE/dw1 = %.9f' % w_h.grad[0, 0].item())
    return w_o.grad[0, 0].item(), w_h.grad[0, 0].item()


if __name__ == '__main__':
    m5, m1 = manual()
    a5, a1 = by_autograd()
    print('\nw5 两边之差 = %.12f' % abs(m5 - a5))
    print('w1 两边之差 = %.12f' % abs(m1 - a1))
    print('对得上，说明 autograd 做的就是这套链式法则，没有别的魔法')

"""把损失、反向、更新串成一个完整的 epoch，并打印损失是否真的在降。"""
import torch
from torch import nn, optim
from torch.utils.data import TensorDataset, DataLoader


def make_data(n=512, in_features=8, n_class=3, seed=0):
    """造一批线性可分的假数据：标签由一组固定权重决定，保证学得动。"""
    torch.manual_seed(seed)
    x = torch.randn(n, in_features)
    true_w = torch.randn(in_features, n_class)
    y = (x @ true_w).argmax(dim=1)          # 类别下标，int64
    return TensorDataset(x, y), in_features, n_class


class Net(nn.Module):
    def __init__(self, in_features, n_class):
        super().__init__()
        self.fc = nn.Linear(in_features, 32)
        self.out = nn.Linear(32, n_class)

    def forward(self, x):
        # 输出原始 logits，softmax 由 CrossEntropyLoss 内部完成
        return self.out(torch.relu(self.fc(x)))


def main():
    dataset, in_features, n_class = make_data()
    loader = DataLoader(dataset, batch_size=32, shuffle=True)

    model = Net(in_features, n_class)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=1e-3)

    for epoch in range(10):
        total_loss, total_num, correct = 0.0, 0, 0
        for bx, by in loader:
            logits = model(bx)                 # 前向
            loss = criterion(logits, by)       # 算损失

            optimizer.zero_grad()              # 清零：否则梯度累加
            loss.backward()                    # 反向：只算梯度，参数没动
            optimizer.step()                   # 更新：这一步才真的改参数

            # loss 是这一批的平均值，乘回样本数才能跨批次累加
            total_loss += loss.item() * len(by)
            total_num += len(by)
            correct += (logits.argmax(dim=1) == by).sum().item()

        print('epoch %2d  loss %.5f  acc %.4f'
              % (epoch + 1, total_loss / total_num, correct / total_num))


if __name__ == '__main__':
    main()

"""梯度累加模拟大 batch：小批跑 4 次只反向不更新，第 4 次之后才更新一次。

要证明的事：
    「batch_size=8 累加 4 步」得到的梯度，和「batch_size=32 直接算」完全相同。
证明方式：两条路径都从同一份初始权重、同一批数据出发，最后比梯度。
"""
import torch
from torch import nn


def build_model(seed=0):
    """每次都从同一组初始权重出发，两条路径才有可比性。"""
    torch.manual_seed(seed)
    return nn.Sequential(nn.Linear(6, 12), nn.ReLU(), nn.Linear(12, 1))


def big_batch_grad(x, y):
    """常规做法：一次性把 32 条样本喂进去。"""
    model = build_model()
    criterion = nn.MSELoss()

    loss = criterion(model(x), y)
    model.zero_grad()
    loss.backward()
    return [p.grad.clone() for p in model.parameters()], loss.item()


def accumulated_grad(x, y, micro=8, accum_steps=4):
    """省显存做法：每次只喂 8 条，连做 4 次反向，梯度自然叠加。"""
    model = build_model()
    criterion = nn.MSELoss()

    model.zero_grad()                       # 整轮开始前清一次，之后中途不清
    total = 0.0
    for step in range(accum_steps):
        bx = x[step * micro:(step + 1) * micro]
        by = y[step * micro:(step + 1) * micro]

        loss = criterion(model(bx), by)
        # 关键的一除：每个小批的 loss 是它自己那 8 条的均值，
        # 直接叠 4 次相当于把梯度放大 4 倍，等于偷偷把学习率调大了 4 倍。
        loss = loss / accum_steps
        loss.backward()                     # 只反向，不 step、不 zero_grad
        total += loss.item()

    return [p.grad.clone() for p in model.parameters()], total


def main():
    torch.manual_seed(42)
    x = torch.randn(32, 6)
    y = torch.randn(32, 1)

    big, big_loss = big_batch_grad(x, y)
    acc, acc_loss = accumulated_grad(x, y)

    print('一次性大 batch 的 loss = %.8f' % big_loss)
    print('累加四步得到的 loss     = %.8f' % acc_loss)

    worst = max((a - b).abs().max().item() for a, b in zip(big, acc))
    print('两条路径的梯度最大差异 = %.10f' % worst)
    print('差异为 0（或仅剩浮点误差），说明累加确实等效于大 batch')

    # ------------------------------------------------------------ 忘了除会怎样
    model = build_model()
    criterion = nn.MSELoss()
    model.zero_grad()
    for step in range(4):
        bx = x[step * 8:(step + 1) * 8]
        by = y[step * 8:(step + 1) * 8]
        criterion(model(bx), by).backward()     # 故意不除以 4
    no_div = [p.grad.clone() for p in model.parameters()]
    ratio = no_div[0].norm().item() / max(big[0].norm().item(), 1e-12)
    print('\n忘记除以累加步数时，梯度范数是正确值的 %.2f 倍' % ratio)
    print('等效学习率被放大同样的倍数，训练很容易发散')


def train_loop_sketch(model, loader, criterion, optimizer, accum_steps=4):
    """真实训练里的写法，重点在 step / zero_grad 的位置和收尾。"""
    model.train()
    optimizer.zero_grad()
    pending = False                          # 有没有还没生效的累计梯度

    for i, (bx, by) in enumerate(loader, start=1):
        loss = criterion(model(bx), by) / accum_steps
        loss.backward()
        pending = True

        if i % accum_steps == 0:
            optimizer.step()                 # 每 accum_steps 批才更新一次
            optimizer.zero_grad()
            pending = False

    # 收尾：样本数不能被整除时，最后剩下的几批也要补一次更新，
    # 否则这部分梯度算了却从未生效。
    if pending:
        optimizer.step()
        optimizer.zero_grad()


if __name__ == '__main__':
    main()

"""数值梯度校验：不信 autograd？用「挪一点点看损失变多少」验它一遍。

梯度的定义就是变化率：
    ∂loss/∂w ≈ [ loss(w + ε) − loss(w − ε) ] / (2ε)
这叫中心差分。它慢得没法用来训练，但用来核对反向传播写得对不对，
是自己实现层或自定义损失时最可靠的一招。
"""
import torch
import torch.nn as nn


def numeric_grad(model, params, x, y, criterion, eps=1e-3):
    """对每个参数逐个元素做一次中心差分，返回同形状的数值梯度。"""
    grads = []
    for p in params:
        g = torch.zeros_like(p)
        flat_p, flat_g = p.data.view(-1), g.view(-1)
        for i in range(flat_p.numel()):
            original = flat_p[i].item()

            flat_p[i] = original + eps
            loss_plus = criterion(model(x), y).item()

            flat_p[i] = original - eps
            loss_minus = criterion(model(x), y).item()

            flat_p[i] = original                    # 用完立刻还原，不能留痕
            flat_g[i] = (loss_plus - loss_minus) / (2 * eps)
        grads.append(g)
    return grads


def main():
    torch.manual_seed(0)

    # 小网络 + 小数据：数值梯度是 O(参数量) 次前向，规模一大就跑不动
    model = nn.Sequential(
        nn.Linear(3, 4),
        nn.Tanh(),          # 用 tanh 而不是 relu：relu 在 0 点不可导，差分会失真
        nn.Linear(4, 1),
    ).double()              # 双精度能把差分的舍入误差压下去

    x = torch.randn(8, 3, dtype=torch.float64)
    y = torch.randn(8, 1, dtype=torch.float64)
    criterion = nn.MSELoss()

    params = list(model.parameters())

    # ---------------------------------------------------------- autograd
    model.zero_grad()
    criterion(model(x), y).backward()
    auto = [p.grad.clone() for p in params]

    # ---------------------------------------------------------- 中心差分
    with torch.no_grad():
        numeric = numeric_grad(model, params, x, y, criterion)

    print('%-14s %-14s %-14s %s' % ('参数', 'autograd 范数', '数值梯度范数', '相对误差'))
    for (name, _), a, n in zip(model.named_parameters(), auto, numeric):
        rel = (a - n).norm() / (a.norm() + n.norm() + 1e-12)
        print('%-14s %-14.8f %-14.8f %.3e'
              % (name, a.norm().item(), n.norm().item(), rel.item()))

    print('\n相对误差在 1e-6 量级以内就说明反向传播是对的。')
    print('误差偏大时先查三件事：')
    print('  1) eps 太小被浮点舍入吃掉，或太大导致差分不准，先试 1e-3 ~ 1e-5')
    print('  2) 网络里有 relu 这类折点函数，正好踩在不可导点上')
    print('  3) 模型里有 dropout / BatchNorm 这类每次前向结果都变的层，')
    print('     校验前必须先 model.eval()')

    # ---------------------------------------------------------- 代价有多大
    n_param = sum(p.numel() for p in params)
    print('\n本例参数量 %d，数值校验做了 %d 次前向；' % (n_param, 2 * n_param))
    print('参数上百万时这个做法完全不可行——这正是反向传播的价值：')
    print('一次反向就拿到全部参数的梯度，代价约等于一次前向。')


if __name__ == '__main__':
    main()

"""损失函数选型骨架：按任务类型挑一条分支，其余删掉。"""
import torch
import torch.nn as nn

TASK = 'multiclass'     # TODO: 'multiclass' / 'binary' / 'regression'


def build_criterion(task=TASK):
    if task == 'multiclass':
        # 网络最后一层不加 softmax；标签是 int64 的类别下标
        return nn.CrossEntropyLoss()
    if task == 'binary':
        # 网络最后一层要加 sigmoid；标签是 0./1. 的 float
        return nn.BCELoss()
    if task == 'regression':
        # TODO: 数据干净用 MSELoss，离群点多用 L1Loss，折中用 SmoothL1Loss
        return nn.SmoothL1Loss()
    raise ValueError('未知任务类型：%s' % task)


def check_shapes(logits, target, task=TASK):
    """上手前先自检一遍形状和类型，能挡掉八成的 RuntimeError。"""
    if task == 'multiclass':
        assert logits.dim() == 2, 'logits 应为 (batch, 类别数)'
        assert target.dtype == torch.int64, '多分类标签必须是 int64'
        assert target.dim() == 1, '多分类标签应为 (batch,) 的类别下标'
    elif task == 'binary':
        assert target.dtype == torch.float32, '二分类标签必须是 float32'
        assert logits.shape == target.shape, '二分类预测与标签形状要一致'
    else:
        assert logits.shape == target.shape, '回归预测与标签形状要一致'


if __name__ == '__main__':
    criterion = build_criterion()
    logits = torch.randn(4, 3, requires_grad=True)   # TODO: 换成模型输出
    target = torch.tensor([0, 2, 1, 1])              # TODO: 换成真实标签
    check_shapes(logits, target)
    print(criterion(logits, target).item())

"""损失函数最常见的六种报错，原地复现一遍，看清报错原文和修法。

每一段都用 try/except 抓住异常，跑到底不中断。
自己触发一次，比读十条注意事项记得牢。
"""
import torch
import torch.nn as nn


def case(title, fn):
    print('=' * 58)
    print(title)
    try:
        fn()
        print('  -> 没有报错')
    except Exception as exc:                     # noqa: BLE001 演示用，要看原文
        print('  -> %s: %s' % (type(exc).__name__, str(exc).split('\n')[0]))


def wrong_label_dtype():
    """多分类标签写成了 float。"""
    logits = torch.randn(4, 3)
    labels = torch.tensor([0., 2., 1., 1.])      # 应该是 int64
    nn.CrossEntropyLoss()(logits, labels)


def right_label_dtype():
    logits = torch.randn(4, 3)
    labels = torch.tensor([0, 2, 1, 1], dtype=torch.int64)
    print('  修法：标签用 int64 的类别下标，损失 = %.6f'
          % nn.CrossEntropyLoss()(logits, labels).item())


def one_hot_label():
    """把标签做成了 one-hot 又当类别下标传。"""
    logits = torch.randn(2, 3)
    labels = torch.tensor([[0, 1, 0], [0, 0, 1]], dtype=torch.int64)
    nn.CrossEntropyLoss()(logits, labels)


def label_out_of_range():
    """类别下标越界：3 类的任务里出现了下标 3。"""
    logits = torch.randn(2, 3)
    labels = torch.tensor([1, 3], dtype=torch.int64)     # 合法范围是 0..2
    nn.CrossEntropyLoss()(logits, labels).backward()


def bce_without_sigmoid():
    """BCELoss 要求预测值落在 (0,1)，直接喂原始分数会报错。"""
    logits = torch.tensor([2.5, -1.2, 0.3])
    target = torch.tensor([1., 0., 1.])
    nn.BCELoss()(logits, target)


def bce_fixed():
    logits = torch.tensor([2.5, -1.2, 0.3])
    target = torch.tensor([1., 0., 1.])
    a = nn.BCELoss()(torch.sigmoid(logits), target).item()
    b = nn.BCEWithLogitsLoss()(logits, target).item()
    print('  修法一 sigmoid + BCELoss   = %.8f' % a)
    print('  修法二 BCEWithLogitsLoss   = %.8f（数值更稳，推荐）' % b)


def silent_broadcast():
    """最危险的一种：不报错，但算出来的损失是错的。"""
    pred = torch.randn(8, 1)          # 模型输出
    target = torch.randn(8)           # 标签少了一维
    loss = nn.MSELoss()(pred, target)
    print('  形状 %s 与 %s 被广播成 %s'
          % (tuple(pred.shape), tuple(target.shape),
             tuple((pred - target).shape)))
    print('  损失值 %.6f 看起来完全正常，实际是 64 个差值的均值' % loss.item())
    fixed = nn.MSELoss()(pred, target.reshape(-1, 1))
    print('  修正后 = %.6f，两个数完全不同' % fixed.item())


def backward_twice():
    """同一张计算图反向两次。"""
    w = torch.tensor(2.0, requires_grad=True)
    loss = (w * 3) ** 2
    loss.backward()
    loss.backward()                   # 图已被释放


if __name__ == '__main__':
    case('一、多分类标签是 float', wrong_label_dtype)
    right_label_dtype()
    case('二、多分类标签做成了 one-hot', one_hot_label)
    case('三、类别下标越界', label_out_of_range)
    case('四、BCELoss 收到未过 sigmoid 的原始分数', bce_without_sigmoid)
    bce_fixed()
    case('五、预测与标签形状不一致（不报错，最危险）', silent_broadcast)
    case('六、同一张计算图 backward 两次', backward_twice)
    print('=' * 58)
    print('第五条是唯一不抛异常的一条：算损失前先 assert 两边形状一致。')