📐 优化理论¶

import numpy as np
import matplotlib.pyplot as plt

def gradient_descent_2d(f, grad_f, x0, lr=0.1, n_iters=50):
    """2D梯度下降可视化"""
    path = [x0.copy()]
    x = x0.copy()
    for _ in range(n_iters):
        g = grad_f(x)
        x = x - lr * g
        path.append(x.copy())
    return np.array(path)  # np.array创建NumPy数组

# Rosenbrock函数（经典难优化函数）
def rosenbrock(x):
    return (1 - x[0])**2 + 100*(x[1] - x[0]**2)**2

def rosenbrock_grad(x):
    dx = -2*(1 - x[0]) - 400*x[0]*(x[1] - x[0]**2)
    dy = 200*(x[1] - x[0]**2)
    return np.array([dx, dy])

x0 = np.array([-1.0, 1.0])
path = gradient_descent_2d(rosenbrock, rosenbrock_grad, x0, lr=0.001, n_iters=200)
print(f"起点: {x0}, 终点: {path[-1]}, 最优: [1, 1]")

import numpy as np

def gradient_descent_convergence_demo():
    """演示梯度下降收敛性"""
    # 二次函数 f(x) = 0.5 * x^T A x - b^T x
    # 最优解 x* = A^{-1} b
    A = np.array([[2.0, 0.5], [0.5, 1.0]])  # 正定矩阵
    b = np.array([1.0, 1.0])
    x_star = np.linalg.solve(A, b)  # np.linalg线性代数运算
    f_star = 0.5 * x_star @ A @ x_star - b @ x_star

    L = np.linalg.eigvalsh(A).max()  # Lipschitz常数
    mu = np.linalg.eigvalsh(A).min()  # 强凸参数
    eta = 1.0 / L  # 最优学习率

    x = np.array([5.0, 5.0])
    errors = []

    for t in range(100):
        grad = A @ x - b
        x = x - eta * grad
        f_val = 0.5 * x @ A @ x - b @ x
        errors.append(f_val - f_star)

    print(f"L = {L:.2f}, μ = {mu:.2f}, 条件数 = {L/mu:.2f}")
    print(f"理论收敛率: O((1 - μ/L)^t) = O({1 - mu/L:.4f}^t)")
    print(f"实际收敛: f(x_10) - f* = {errors[9]:.6f}, f(x_50) - f* = {errors[49]:.10f}")

gradient_descent_convergence_demo()

class AdamOptimizer:
    """手写Adam优化器"""

    def __init__(self, lr=0.001, beta1=0.9, beta2=0.999, eps=1e-8):  # __init__构造方法，创建对象时自动调用
        self.lr = lr
        self.beta1 = beta1
        self.beta2 = beta2
        self.eps = eps
        self.m = None  # 一阶矩
        self.v = None  # 二阶矩
        self.t = 0     # 时间步

    def step(self, params, grads):
        if self.m is None:
            self.m = np.zeros_like(params)
            self.v = np.zeros_like(params)

        self.t += 1
        self.m = self.beta1 * self.m + (1 - self.beta1) * grads
        self.v = self.beta2 * self.v + (1 - self.beta2) * grads**2

        # 偏差修正
        m_hat = self.m / (1 - self.beta1**self.t)
        v_hat = self.v / (1 - self.beta2**self.t)

        params -= self.lr * m_hat / (np.sqrt(v_hat) + self.eps)
        return params

# 对比SGD vs Momentum vs Adam
def compare_optimizers():
    """在Rosenbrock函数上对比优化器"""
    x_sgd = np.array([-1.0, 1.0])
    x_mom = np.array([-1.0, 1.0])
    x_adam = np.array([-1.0, 1.0])

    adam = AdamOptimizer(lr=0.01)
    v_mom = np.zeros(2)

    paths = {"SGD": [x_sgd.copy()], "Momentum": [x_mom.copy()], "Adam": [x_adam.copy()]}

    for _ in range(500):
        g = rosenbrock_grad(x_sgd)
        x_sgd = x_sgd - 0.0005 * g
        paths["SGD"].append(x_sgd.copy())

        g = rosenbrock_grad(x_mom)
        v_mom = 0.9 * v_mom + 0.0005 * g
        x_mom = x_mom - v_mom
        paths["Momentum"].append(x_mom.copy())

        g = rosenbrock_grad(x_adam)
        x_adam = adam.step(x_adam, g)
        paths["Adam"].append(x_adam.copy())

    for name, path in paths.items():
        p = np.array(path)
        print(f"{name:10s}: 终点=({p[-1,0]:.4f}, {p[-1,1]:.4f}), "
              f"距最优={np.linalg.norm(p[-1] - [1, 1]):.6f}")

compare_optimizers()

import numpy as np
import matplotlib.pyplot as plt

# 模拟不同学习率策略
steps = 100
lrs = {}

# Step Decay
lr = 0.1
lrs["StepDecay"] = []
for i in range(steps):
    if i in [30, 60, 80]: lr *= 0.1
    lrs["StepDecay"].append(lr)

# Cosine Annealing
lrs["Cosine"] = [0.1 * 0.5 * (1 + np.cos(np.pi * i / steps)) for i in range(steps)]  # 列表推导式，简洁创建列表

# Warmup + Cosine (Transformer标配)
warmup = 10
lrs["Warmup+Cosine"] = []
for i in range(steps):
    if i < warmup:
        lrs["Warmup+Cosine"].append(0.1 * (i + 1) / warmup)
    else:
        progress = (i - warmup) / (steps - warmup)
        lrs["Warmup+Cosine"].append(0.1 * 0.5 * (1 + np.cos(np.pi * progress)))

# 1/sqrt(t) decay (原始Transformer)
lrs["InvSqrt"] = [0.1 * min(1/(i+1)**0.5, (i+1)*warmup**(-1.5)) for i in range(steps)]

for name, lr_list in lrs.items():
    plt.plot(lr_list, label=name)
plt.xlabel("Step"); plt.ylabel("LR"); plt.legend(); plt.title("学习率调度策略对比")
plt.show()

# 可视化：高维空间中鞍点远多于局部最小值
# Hessian特征值：n维空间，每个方向独立为正/负（概率各50%）
# 所有方向都为正（局部最小）的概率 = 0.5^n → 指数衰减！

import matplotlib.pyplot as plt

dims = np.arange(1, 101)
p_local_min = 0.5 ** dims

plt.semilogy(dims, p_local_min)
plt.xlabel("参数维度")
plt.ylabel("随机临界点是局部最小值的概率")
plt.title("高维空间中鞍点远多于局部最小值")
plt.grid(True)
plt.show()

# 这就是为什么SGD实际效果好——大多数"卡住"是鞍点，SGD噪声能逃出

from sklearn.svm import SVC
from sklearn.datasets import make_classification

X, y = make_classification(n_samples=100, n_features=2,
                           n_redundant=0, random_state=42)

svc = SVC(kernel='linear', C=1.0)
svc.fit(X, y)

print(f"支持向量数量: {svc.n_support_}")
print(f"总样本数: {len(X)}")
print(f"支持向量比例: {sum(svc.n_support_)/len(X):.2%}")
# 互补松弛条件: 只有支持向量的α>0，其余都=0

import torch

def gradient_clipping_demo():
    """梯度裁剪演示 — 大模型训练常用"""
    model = torch.nn.Linear(10, 10)

    # 模拟大梯度
    x = torch.randn(1, 10) * 100
    loss = model(x).sum()
    loss.backward()  # 反向传播计算梯度

    # 裁剪前
    total_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=float('inf'))
    print(f"裁剪前梯度范数: {total_norm:.2f}")

    # 重新计算
    model.zero_grad()
    loss = model(x).sum()
    loss.backward()

    # 裁剪（max_norm=1.0是LLM常用值）
    # 注意: clip_grad_norm_ 返回裁剪前的梯度范数
    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
    post_clip_norm = sum(p.grad.norm()**2 for p in model.parameters() if p.grad is not None).sqrt()
    print(f"裁剪后梯度范数: {post_clip_norm:.2f}")

gradient_clipping_demo()

import torch

class AdamW:
    """手写AdamW优化器"""
    def __init__(self, params, lr=1e-3, betas=(0.9, 0.999), eps=1e-8, weight_decay=0.01):
        self.params = list(params)
        self.lr = lr
        self.beta1, self.beta2 = betas
        self.eps = eps
        self.wd = weight_decay
        self.t = 0
        self.m = [torch.zeros_like(p) for p in self.params]
        self.v = [torch.zeros_like(p) for p in self.params]

    def step(self):
        self.t += 1
        for i, p in enumerate(self.params):  # enumerate同时获取索引和元素
            if p.grad is None:
                continue
            g = p.grad.data

            # 一阶矩（均值）
            self.m[i] = self.beta1 * self.m[i] + (1 - self.beta1) * g
            # 二阶矩（方差）
            self.v[i] = self.beta2 * self.v[i] + (1 - self.beta2) * g ** 2

            # 偏差校正
            m_hat = self.m[i] / (1 - self.beta1 ** self.t)
            v_hat = self.v[i] / (1 - self.beta2 ** self.t)

            # 参数更新 = Adam步 + 解耦权重衰减
            p.data -= self.lr * (m_hat / (v_hat.sqrt() + self.eps) + self.wd * p.data)

    def zero_grad(self):
        for p in self.params:
            if p.grad is not None:
                p.grad.zero_()

# 验证：对比PyTorch官方实现
model = torch.nn.Linear(10, 1)
opt_official = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=0.01)
print("AdamW手写实现完成")

import math

class CosineAnnealingWarmup:
    """Warmup + Cosine退火调度器"""
    def __init__(self, optimizer, warmup_steps, total_steps, min_lr=1e-6):
        self.optimizer = optimizer
        self.warmup_steps = warmup_steps
        self.total_steps = total_steps
        self.min_lr = min_lr
        self.base_lr = optimizer.param_groups[0]['lr']
        self.step_count = 0

    def step(self):
        self.step_count += 1
        if self.step_count <= self.warmup_steps:
            # 线性Warmup
            lr = self.base_lr * self.step_count / self.warmup_steps
        else:
            # Cosine退火
            progress = (self.step_count - self.warmup_steps) / (self.total_steps - self.warmup_steps)
            lr = self.min_lr + 0.5 * (self.base_lr - self.min_lr) * (1 + math.cos(math.pi * progress))

        for pg in self.optimizer.param_groups:
            pg['lr'] = lr
        return lr

# 可视化调度曲线
model = torch.nn.Linear(10, 1)
opt = torch.optim.AdamW(model.parameters(), lr=3e-4)
scheduler = CosineAnnealingWarmup(opt, warmup_steps=1000, total_steps=50000)

lrs = [scheduler.step() for _ in range(50000)]
print(f"Warmup结束lr: {lrs[999]:.6f}")
print(f"中间lr: {lrs[25000]:.6f}")
print(f"最终lr: {lrs[-1]:.6f}")

def train_with_gradient_accumulation(model, dataloader, optimizer, accumulation_steps=4):
    """梯度累积：用小显存实现大batch效果"""
    model.train()  # train()开启训练模式
    optimizer.zero_grad()

    for step, (x, y) in enumerate(dataloader):
        loss = model(x, y) / accumulation_steps  # 除以累积步数
        loss.backward()  # 梯度累积

        if (step + 1) % accumulation_steps == 0:
            # 梯度裁剪（防梯度爆炸）
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
            optimizer.step()  # 根据梯度更新模型参数
            optimizer.zero_grad()

# 等价于batch_size = micro_batch * accumulation_steps * n_gpus
print("梯度累积: 4次micro_batch=8 ≈ 1次batch=32")

# PyTorch AMP (Automatic Mixed Precision)
scaler = torch.amp.GradScaler('cuda')

for x, y in dataloader:
    optimizer.zero_grad()

    # 前向：自动选fp16/fp32
    with torch.amp.autocast(device_type='cuda'):
        output = model(x)
        loss = criterion(output, y)

    # 反向：loss scale防止fp16下溢
    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()