Cao Note

外观

Python 深度学习（PyTorch） ​

基于 PyTorch 的深度学习入门，涵盖核心概念、模型构建与训练流程。

深度学习概述 ​

深度学习是机器学习的一个子领域，其核心是使用多层神经网络从数据中自动学习特征表示。与传统机器学习方法不同，深度学习无需手工设计特征，而是通过端到端的学习方式自动提取层次化的特征。

深度学习的典型应用 ​

  • 计算机视觉：图像分类、目标检测、语义分割、人脸识别
  • 自然语言处理：文本分类、机器翻译、情感分析、问答系统
  • 语音识别：语音转文字、说话人识别
  • 推荐系统：用户行为预测、个性化推荐
  • 强化学习：游戏 AI、机器人控制、自动驾驶

为什么选择 PyTorch ​

PyTorch 由 Meta（原 Facebook）于 2016 年开源，凭借其简洁的设计与出色的灵活性，迅速成为学术研究与工业部署的主流深度学习框架。

PyTorch vs TensorFlow ​

当前（2025）市场格局 ​

PyTorch 在学术界已是绝对主流，绝大多数顶会论文（CVPR、ICCV、NeurIPS、ICLR 等）基于 PyTorch 实现。在工业界，PyTorch 通过 TorchScript、TorchServe 和 ONNX 导出等工具链，已广泛部署于生产环境。TensorFlow 仍在使用 Keras 的团队和部分已有基础设施的大厂中存在，但新项目选型普遍倾向 PyTorch。

对比维度PyTorchTensorFlow
计算图动态图（Define-by-Run）2.x 默认 Eager Execution，但历史包袱重
调试体验原生 Python 调试器（pdb）直接可用需要 tfdbg，调试体验相对笨重
学习曲线平缓，接近 NumPy 风格较陡峭，API 分层复杂
生态torchvision、torchaudio、torchtext、HuggingFaceTF Hub、TF Datasets、Keras
生产部署TorchServe、ONNX、LibTorch（C++）TF Serving、TF Lite、TF.js
分布式训练原生 DDP（Distributed Data Parallel）tf.distribute.Strategy
社区活跃度极高，学术界首选依赖 Google 维护，社区热度下降

核心差异解读 ​

动态图 vs 静态图：PyTorch 的执行方式是与 Python 运行时同步的，每一行代码执行时图结构即时构建，这使得条件分支和循环可以直接使用 Python 原生语法。TensorFlow 1.x 时代需要先构建静态计算图再通过 Session 执行，调试极为不便。TensorFlow 2.x 虽默认启用了 Eager Execution，但其底层架构仍保留了静态图的复杂性。

调试体验：在 PyTorch 中，你可以在前向传播的任何位置插入 print() 或 breakpoint()，就像调试普通 Python 代码一样。TensorFlow 的调试则需要通过 tf.print()（张量打印需在图中执行）或 tf.debugging 工具集，体验上不如 PyTorch 直观。

何时选择 PyTorch ​

  • 学术研究、论文复现、快速原型开发
  • 需要灵活的动态图控制流（如循环神经网络、树形结构模型）
  • 团队以 Python 工程师为主，希望降低学习成本
  • 需要与 HuggingFace Transformers、Diffusers 等现代生态集成

何时仍可考虑 TensorFlow ​

  • 已有大规模 TensorFlow 生产基础设施，迁移成本过高
  • 需要移动端部署（TFLite）或 Web 端部署（TF.js）
  • 团队深度绑定 Google Cloud 生态（TPU、Vertex AI）

PyTorch 核心概念 ​

Tensor（张量） ​

Tensor 是 PyTorch 中最基本的数据结构，可以理解为支持 GPU 加速的多维数组，其接口设计高度接近 NumPy。

python
import torch
import numpy as np

# ── 创建张量 ──────────────────────────────────────────

# 从列表创建
x = torch.tensor([[1, 2], [3, 4]])
print(x.shape)   # torch.Size([2, 2])
print(x.dtype)   # torch.int64

# 特殊张量
zeros = torch.zeros(2, 3)            # 全零
ones = torch.ones(2, 3)              # 全一
eye = torch.eye(3)                   # 单位矩阵
rand = torch.randn(3, 4)             # 标准正态分布随机
rand_uniform = torch.rand(3, 4)      # [0,1) 均匀分布
arange = torch.arange(0, 10, 2)      # [0, 2, 4, 6, 8]
linspace = torch.linspace(0, 1, 5)   # [0.0, 0.25, 0.5, 0.75, 1.0]

# 指定数据类型和设备
x = torch.tensor([1, 2, 3], dtype=torch.float32)
x = torch.tensor([1, 2, 3], device="cuda" if torch.cuda.is_available() else "cpu")

# ── Tensor 与 NumPy 互转 ──────────────────────────────

arr = np.array([[1, 2], [3, 4]])
x = torch.from_numpy(arr)   # NumPy → Tensor（共享内存）
back_to_arr = x.numpy()     # Tensor → NumPy（共享内存）

# 注意：共享内存意味着修改一方会影响另一方
arr[0, 0] = 999
print(x)  # tensor([[999,   2], [  3,   4]])

# ── 设备管理 ──────────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
x = torch.randn(3, 4)
x = x.to(device)          # 移动到 GPU
x = x.cpu()               # 移回 CPU

# 创建时直接指定设备
x = torch.randn(3, 4, device=device)

# ── 常用操作 ──────────────────────────────────────────

a = torch.randn(2, 3)
b = torch.randn(2, 3)

# 算术运算
c = a + b          # 加法
c = torch.add(a, b)
c = a * b          # 逐元素乘法
c = torch.matmul(a, b.T)  # 矩阵乘法
c = a @ b.T        # 等价写法（Python 3.5+）

# 形状操作
print(a.shape)       # torch.Size([2, 3])
print(a.size())      # 同上
print(a.ndim)        # 维度数：2

c = a.view(3, 2)     # 重塑（必须保证总元素数不变）
c = a.reshape(3, 2)  # 类似 view，但会自动处理连续性
c = a.transpose(0, 1) # 交换维度
c = a.unsqueeze(0)   # 在 dim=0 处插入一个维度 [1, 2, 3]
c = a.squeeze()      # 移除所有长度为 1 的维度

# 聚合操作
mean = a.mean()
sum_ = a.sum()
max_val, max_idx = a.max(dim=1)  # 沿 dim=1 取最大值
import torch
import numpy as np

# ── 创建张量 ──────────────────────────────────────────

# 从列表创建
x = torch.tensor([[1, 2], [3, 4]])
print(x.shape)   # torch.Size([2, 2])
print(x.dtype)   # torch.int64

# 特殊张量
zeros = torch.zeros(2, 3)            # 全零
ones = torch.ones(2, 3)              # 全一
eye = torch.eye(3)                   # 单位矩阵
rand = torch.randn(3, 4)             # 标准正态分布随机
rand_uniform = torch.rand(3, 4)      # [0,1) 均匀分布
arange = torch.arange(0, 10, 2)      # [0, 2, 4, 6, 8]
linspace = torch.linspace(0, 1, 5)   # [0.0, 0.25, 0.5, 0.75, 1.0]

# 指定数据类型和设备
x = torch.tensor([1, 2, 3], dtype=torch.float32)
x = torch.tensor([1, 2, 3], device="cuda" if torch.cuda.is_available() else "cpu")

# ── Tensor 与 NumPy 互转 ──────────────────────────────

arr = np.array([[1, 2], [3, 4]])
x = torch.from_numpy(arr)   # NumPy → Tensor（共享内存）
back_to_arr = x.numpy()     # Tensor → NumPy（共享内存）

# 注意：共享内存意味着修改一方会影响另一方
arr[0, 0] = 999
print(x)  # tensor([[999,   2], [  3,   4]])

# ── 设备管理 ──────────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
x = torch.randn(3, 4)
x = x.to(device)          # 移动到 GPU
x = x.cpu()               # 移回 CPU

# 创建时直接指定设备
x = torch.randn(3, 4, device=device)

# ── 常用操作 ──────────────────────────────────────────

a = torch.randn(2, 3)
b = torch.randn(2, 3)

# 算术运算
c = a + b          # 加法
c = torch.add(a, b)
c = a * b          # 逐元素乘法
c = torch.matmul(a, b.T)  # 矩阵乘法
c = a @ b.T        # 等价写法（Python 3.5+）

# 形状操作
print(a.shape)       # torch.Size([2, 3])
print(a.size())      # 同上
print(a.ndim)        # 维度数：2

c = a.view(3, 2)     # 重塑（必须保证总元素数不变）
c = a.reshape(3, 2)  # 类似 view，但会自动处理连续性
c = a.transpose(0, 1) # 交换维度
c = a.unsqueeze(0)   # 在 dim=0 处插入一个维度 [1, 2, 3]
c = a.squeeze()      # 移除所有长度为 1 的维度

# 聚合操作
mean = a.mean()
sum_ = a.sum()
max_val, max_idx = a.max(dim=1)  # 沿 dim=1 取最大值

Autograd（自动微分） ​

Autograd 是 PyTorch 的自动微分引擎，它是训练神经网络的基础。通过记录张量上的所有操作，Autograd 能够自动计算梯度。

python
# ── 基本原理 ──────────────────────────────────────────

# 创建需要梯度的张量
x = torch.randn(3, requires_grad=True)
print(x.requires_grad)  # True

y = x * 2               # 记录计算图
z = y.mean()            # 继续向前传播

z.backward()            # 反向传播，计算梯度
print(x.grad)           # 查看 x 的梯度值

# 手动验证梯度：
# z = mean(y) = 1/3 * (y0 + y1 + y2)
# y = 2 * x
# dz/dx_i = dz/dy_i * dy_i/dx_i = (1/3) * 2 = 2/3
print(x.grad)           # tensor([0.6667, 0.6667, 0.6667])

# ── 梯度累积与清零 ────────────────────────────────────

# 默认情况下，梯度会累积
for _ in range(3):
    y = (x * 2).mean()
    y.backward()
print(x.grad)  # 梯度值累积了 3 次

# 需要手动清零
x.grad.zero_()

# ── 停止追踪梯度 ──────────────────────────────────────

# 推理阶段或冻结参数时，可用 torch.no_grad() 禁用梯度追踪
with torch.no_grad():
    y = x * 2         # 此处不会记录计算图
    print(y.requires_grad)  # False

# 等价方法：detach()
y = x.detach()        # 返回新的张量，不追踪梯度

# ── requires_grad 设置 ────────────────────────────────

# 修改已有张量的 requires_grad
x.requires_grad_(True)   # 原地修改
x.requires_grad_(False)  # 关闭梯度追踪

# 查看梯度是否有效
print(x.requires_grad)   # bool
# ── 基本原理 ──────────────────────────────────────────

# 创建需要梯度的张量
x = torch.randn(3, requires_grad=True)
print(x.requires_grad)  # True

y = x * 2               # 记录计算图
z = y.mean()            # 继续向前传播

z.backward()            # 反向传播，计算梯度
print(x.grad)           # 查看 x 的梯度值

# 手动验证梯度：
# z = mean(y) = 1/3 * (y0 + y1 + y2)
# y = 2 * x
# dz/dx_i = dz/dy_i * dy_i/dx_i = (1/3) * 2 = 2/3
print(x.grad)           # tensor([0.6667, 0.6667, 0.6667])

# ── 梯度累积与清零 ────────────────────────────────────

# 默认情况下，梯度会累积
for _ in range(3):
    y = (x * 2).mean()
    y.backward()
print(x.grad)  # 梯度值累积了 3 次

# 需要手动清零
x.grad.zero_()

# ── 停止追踪梯度 ──────────────────────────────────────

# 推理阶段或冻结参数时，可用 torch.no_grad() 禁用梯度追踪
with torch.no_grad():
    y = x * 2         # 此处不会记录计算图
    print(y.requires_grad)  # False

# 等价方法：detach()
y = x.detach()        # 返回新的张量，不追踪梯度

# ── requires_grad 设置 ────────────────────────────────

# 修改已有张量的 requires_grad
x.requires_grad_(True)   # 原地修改
x.requires_grad_(False)  # 关闭梯度追踪

# 查看梯度是否有效
print(x.requires_grad)   # bool

Autograd 工作原理简述：前向传播时，PyTorch 会动态构建一个计算图（DAG，有向无环图），图中的每个节点是一个张量，每条边代表操作。调用 backward() 时，Autograd 沿着计算图从输出到输入遍历，利用链式法则计算每个参数的梯度。计算完成后，该计算图被自动释放以供下一轮使用（默认为动态图，每次迭代重新构建）。

构建神经网络 ​

nn.Module ​

nn.Module 是所有神经网络模块的基类。自定义模型时，你需要继承它并实现 __init__（定义层）和 forward（定义前向传播逻辑）。

python
import torch.nn as nn
import torch.nn.functional as F

class SimpleNet(nn.Module):
    """一个简单的全连接网络：784 → 256 → 10"""
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(784, 256)    # 第一层：784 维输入 → 256 维隐藏层
        self.relu = nn.ReLU()             # ReLU 激活函数
        self.fc2 = nn.Linear(256, 10)     # 第二层：256 维 → 10 维输出（分类数）

    def forward(self, x):
        """
        前向传播
        x shape: [batch_size, 784]
        """
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x  # 原始 logits，未经过 softmax

# 实例化模型
model = SimpleNet()
print(model)

# 查看模型参数
for name, param in model.named_parameters():
    print(f"{name}: {param.shape}")

# 前向传播
dummy_input = torch.randn(32, 784)   # batch_size=32
output = model(dummy_input)
print(output.shape)  # torch.Size([32, 10])
import torch.nn as nn
import torch.nn.functional as F

class SimpleNet(nn.Module):
    """一个简单的全连接网络：784 → 256 → 10"""
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(784, 256)    # 第一层：784 维输入 → 256 维隐藏层
        self.relu = nn.ReLU()             # ReLU 激活函数
        self.fc2 = nn.Linear(256, 10)     # 第二层：256 维 → 10 维输出（分类数）

    def forward(self, x):
        """
        前向传播
        x shape: [batch_size, 784]
        """
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        return x  # 原始 logits，未经过 softmax

# 实例化模型
model = SimpleNet()
print(model)

# 查看模型参数
for name, param in model.named_parameters():
    print(f"{name}: {param.shape}")

# 前向传播
dummy_input = torch.randn(32, 784)   # batch_size=32
output = model(dummy_input)
print(output.shape)  # torch.Size([32, 10])

Sequential 容器 ​

对于层结构简单的线性堆叠，可使用 nn.Sequential 简化代码：

python
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 10),
)
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 10),
)

但使用继承 nn.Module 的方式更为灵活，便于添加复杂控制流（条件分支、循环、多输入等）。

常用网络层 ​

python
# ── 全连接层（Linear） ────────────────────────────────
nn.Linear(in_features=784, out_features=256, bias=True)

# ── 卷积层（Conv2d） ──────────────────────────────────
nn.Conv2d(
    in_channels=3,      # 输入通道数（RGB 图像为 3）
    out_channels=16,    # 输出通道数（卷积核数量）
    kernel_size=3,      # 卷积核大小 3×3
    stride=1,           # 步长
    padding=1,          # 填充（保持空间尺寸）
    padding_mode="zeros"
)

# ── 池化层 ────────────────────────────────────────────
nn.MaxPool2d(kernel_size=2, stride=2)  # 最大值池化，尺寸减半
nn.AvgPool2d(kernel_size=2, stride=2)  # 平均池化
nn.AdaptiveAvgPool2d((1, 1))           # 自适应池化到指定尺寸

# ── Dropout ───────────────────────────────────────────
nn.Dropout(p=0.5)       # 训练时随机丢弃 50% 的神经元（防止过拟合）
nn.Dropout2d(p=0.2)     # 针对卷积层的通道级 Dropout

# ── 批归一化 ──────────────────────────────────────────
nn.BatchNorm1d(num_features=256)    # 全连接层后使用
nn.BatchNorm2d(num_features=16)     # 卷积层后使用（通道维）

# ── 激活函数 ──────────────────────────────────────────
nn.ReLU()
nn.LeakyReLU(negative_slope=0.01)
nn.Sigmoid()       # 二分类输出层
nn.Tanh()
nn.Softmax(dim=1)  # 多分类概率输出
nn.LogSoftmax(dim=1)  # 常与 nn.NLLLoss 配合使用
# ── 全连接层（Linear） ────────────────────────────────
nn.Linear(in_features=784, out_features=256, bias=True)

# ── 卷积层（Conv2d） ──────────────────────────────────
nn.Conv2d(
    in_channels=3,      # 输入通道数（RGB 图像为 3）
    out_channels=16,    # 输出通道数（卷积核数量）
    kernel_size=3,      # 卷积核大小 3×3
    stride=1,           # 步长
    padding=1,          # 填充（保持空间尺寸）
    padding_mode="zeros"
)

# ── 池化层 ────────────────────────────────────────────
nn.MaxPool2d(kernel_size=2, stride=2)  # 最大值池化，尺寸减半
nn.AvgPool2d(kernel_size=2, stride=2)  # 平均池化
nn.AdaptiveAvgPool2d((1, 1))           # 自适应池化到指定尺寸

# ── Dropout ───────────────────────────────────────────
nn.Dropout(p=0.5)       # 训练时随机丢弃 50% 的神经元（防止过拟合）
nn.Dropout2d(p=0.2)     # 针对卷积层的通道级 Dropout

# ── 批归一化 ──────────────────────────────────────────
nn.BatchNorm1d(num_features=256)    # 全连接层后使用
nn.BatchNorm2d(num_features=16)     # 卷积层后使用（通道维）

# ── 激活函数 ──────────────────────────────────────────
nn.ReLU()
nn.LeakyReLU(negative_slope=0.01)
nn.Sigmoid()       # 二分类输出层
nn.Tanh()
nn.Softmax(dim=1)  # 多分类概率输出
nn.LogSoftmax(dim=1)  # 常与 nn.NLLLoss 配合使用

损失函数与优化器 ​

python
import torch.optim as optim

# ── 常用损失函数 ──────────────────────────────────────

# 分类任务
criterion = nn.CrossEntropyLoss()            # 多分类（自带 Softmax）
criterion = nn.BCEWithLogitsLoss()           # 多标签二分类
criterion = nn.BCELoss()                     # 二分类（需先接 Sigmoid）
criterion = nn.NLLLoss()                     # 负对数似然（需先接 LogSoftmax）

# 回归任务
criterion = nn.MSELoss()                     # 均方误差
criterion = nn.L1Loss()                      # 平均绝对误差
criterion = nn.SmoothL1Loss()                # Huber 损失，对异常值更鲁棒

# ── 常用优化器 ────────────────────────────────────────

model = SimpleNet()

optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)
optimizer = optim.Adam(model.parameters(), lr=0.001)              # 最常用
optimizer = optim.AdamW(model.parameters(), lr=0.001, weight_decay=0.01)  # Adam + 正确权重衰减
optimizer = optim.RMSprop(model.parameters(), lr=0.001, alpha=0.99)

# ── 学习率调度器 ───────────────────────────────────────

scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5)
import torch.optim as optim

# ── 常用损失函数 ──────────────────────────────────────

# 分类任务
criterion = nn.CrossEntropyLoss()            # 多分类（自带 Softmax）
criterion = nn.BCEWithLogitsLoss()           # 多标签二分类
criterion = nn.BCELoss()                     # 二分类（需先接 Sigmoid）
criterion = nn.NLLLoss()                     # 负对数似然（需先接 LogSoftmax）

# 回归任务
criterion = nn.MSELoss()                     # 均方误差
criterion = nn.L1Loss()                      # 平均绝对误差
criterion = nn.SmoothL1Loss()                # Huber 损失，对异常值更鲁棒

# ── 常用优化器 ────────────────────────────────────────

model = SimpleNet()

optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.9)
optimizer = optim.Adam(model.parameters(), lr=0.001)              # 最常用
optimizer = optim.AdamW(model.parameters(), lr=0.001, weight_decay=0.01)  # Adam + 正确权重衰减
optimizer = optim.RMSprop(model.parameters(), lr=0.001, alpha=0.99)

# ── 学习率调度器 ───────────────────────────────────────

scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=10, gamma=0.1)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)
scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=5)

训练流程 ​

标准训练循环 ​

python
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader

# 超参数
num_epochs = 10
learning_rate = 0.001
batch_size = 64

# 模型、损失函数、优化器
model = SimpleNet()
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 假设已有 dataloader
# dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

for epoch in range(num_epochs):
    model.train()  # 设置为训练模式（影响 Dropout / BatchNorm 等层）

    running_loss = 0.0
    for batch_idx, (inputs, labels) in enumerate(dataloader):
        # 将数据移动到设备（GPU）
        inputs, labels = inputs.to(device), labels.to(device)

        # ── 前向传播 ──
        outputs = model(inputs)
        loss = criterion(outputs, labels)

        # ── 反向传播 ──
        optimizer.zero_grad()   # 清空上一轮的梯度
        loss.backward()         # 计算梯度
        optimizer.step()        # 更新参数

        # 统计
        running_loss += loss.item()

        if batch_idx % 100 == 99:
            avg_loss = running_loss / 100
            print(f"Epoch [{epoch+1}/{num_epochs}], Batch [{batch_idx+1}], Loss: {avg_loss:.4f}")
            running_loss = 0.0
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader

# 超参数
num_epochs = 10
learning_rate = 0.001
batch_size = 64

# 模型、损失函数、优化器
model = SimpleNet()
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 假设已有 dataloader
# dataloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

for epoch in range(num_epochs):
    model.train()  # 设置为训练模式（影响 Dropout / BatchNorm 等层）

    running_loss = 0.0
    for batch_idx, (inputs, labels) in enumerate(dataloader):
        # 将数据移动到设备（GPU）
        inputs, labels = inputs.to(device), labels.to(device)

        # ── 前向传播 ──
        outputs = model(inputs)
        loss = criterion(outputs, labels)

        # ── 反向传播 ──
        optimizer.zero_grad()   # 清空上一轮的梯度
        loss.backward()         # 计算梯度
        optimizer.step()        # 更新参数

        # 统计
        running_loss += loss.item()

        if batch_idx % 100 == 99:
            avg_loss = running_loss / 100
            print(f"Epoch [{epoch+1}/{num_epochs}], Batch [{batch_idx+1}], Loss: {avg_loss:.4f}")
            running_loss = 0.0

验证 / 测试 ​

python
def evaluate(model, dataloader, device):
    """在验证集或测试集上评估模型"""
    model.eval()  # 设置为评估模式
    correct = 0
    total = 0
    total_loss = 0.0
    criterion = nn.CrossEntropyLoss()

    # 禁用梯度计算（节省内存和计算）
    with torch.no_grad():
        for inputs, labels in dataloader:
            inputs, labels = inputs.to(device), labels.to(device)

            outputs = model(inputs)
            loss = criterion(outputs, labels)
            total_loss += loss.item()

            # 获取预测类别
            _, predicted = torch.max(outputs, dim=1)
            #          或 torch.argmax(outputs, dim=1)

            total += labels.size(0)
            correct += (predicted == labels).sum().item()

    accuracy = 100.0 * correct / total
    avg_loss = total_loss / len(dataloader)

    print(f"Test Results: Loss: {avg_loss:.4f}, Accuracy: {accuracy:.2f}%")
    return accuracy, avg_loss

# 调用评估
# test_accuracy, test_loss = evaluate(model, test_loader, device)
def evaluate(model, dataloader, device):
    """在验证集或测试集上评估模型"""
    model.eval()  # 设置为评估模式
    correct = 0
    total = 0
    total_loss = 0.0
    criterion = nn.CrossEntropyLoss()

    # 禁用梯度计算（节省内存和计算）
    with torch.no_grad():
        for inputs, labels in dataloader:
            inputs, labels = inputs.to(device), labels.to(device)

            outputs = model(inputs)
            loss = criterion(outputs, labels)
            total_loss += loss.item()

            # 获取预测类别
            _, predicted = torch.max(outputs, dim=1)
            #          或 torch.argmax(outputs, dim=1)

            total += labels.size(0)
            correct += (predicted == labels).sum().item()

    accuracy = 100.0 * correct / total
    avg_loss = total_loss / len(dataloader)

    print(f"Test Results: Loss: {avg_loss:.4f}, Accuracy: {accuracy:.2f}%")
    return accuracy, avg_loss

# 调用评估
# test_accuracy, test_loss = evaluate(model, test_loader, device)

完整训练 + 验证循环 ​

python
def train_model(model, train_loader, val_loader, criterion, optimizer,
                num_epochs, device, save_path="best_model.pth"):
    """完整训练流程，包含验证和模型保存"""
    best_acc = 0.0

    for epoch in range(num_epochs):
        # ── 训练阶段 ──
        model.train()
        train_loss = 0.0
        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)

            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            train_loss += loss.item()

        # ── 验证阶段 ──
        model.eval()
        val_loss = 0.0
        correct = 0
        total = 0
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)

                val_loss += loss.item()
                _, predicted = torch.max(outputs, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()

        val_acc = 100.0 * correct / total

        print(f"Epoch [{epoch+1}/{num_epochs}] "
              f"Train Loss: {train_loss/len(train_loader):.4f} "
              f"Val Loss: {val_loss/len(val_loader):.4f} "
              f"Val Acc: {val_acc:.2f}%")

        # 保存最佳模型
        if val_acc > best_acc:
            best_acc = val_acc
            torch.save(model.state_dict(), save_path)
            print(f"  -> Model saved (best acc: {best_acc:.2f}%)")

    print(f"Training completed. Best accuracy: {best_acc:.2f}%")
    return best_acc
def train_model(model, train_loader, val_loader, criterion, optimizer,
                num_epochs, device, save_path="best_model.pth"):
    """完整训练流程，包含验证和模型保存"""
    best_acc = 0.0

    for epoch in range(num_epochs):
        # ── 训练阶段 ──
        model.train()
        train_loss = 0.0
        for inputs, labels in train_loader:
            inputs, labels = inputs.to(device), labels.to(device)

            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            train_loss += loss.item()

        # ── 验证阶段 ──
        model.eval()
        val_loss = 0.0
        correct = 0
        total = 0
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)

                val_loss += loss.item()
                _, predicted = torch.max(outputs, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()

        val_acc = 100.0 * correct / total

        print(f"Epoch [{epoch+1}/{num_epochs}] "
              f"Train Loss: {train_loss/len(train_loader):.4f} "
              f"Val Loss: {val_loss/len(val_loader):.4f} "
              f"Val Acc: {val_acc:.2f}%")

        # 保存最佳模型
        if val_acc > best_acc:
            best_acc = val_acc
            torch.save(model.state_dict(), save_path)
            print(f"  -> Model saved (best acc: {best_acc:.2f}%)")

    print(f"Training completed. Best accuracy: {best_acc:.2f}%")
    return best_acc

数据加载 ​

Dataset 与 DataLoader ​

python
from torch.utils.data import Dataset, DataLoader, random_split

# ── 自定义数据集 ──────────────────────────────────────

class MyDataset(Dataset):
    """
    自定义数据集类，必须实现 __len__ 和 __getitem__。
    """
    def __init__(self, data, labels, transform=None):
        """
        Args:
            data: 特征数据 (numpy array 或 torch.Tensor)
            labels: 标签 (numpy array 或 torch.Tensor)
            transform: 数据增强/预处理函数
        """
        self.data = data
        self.labels = labels
        self.transform = transform

    def __len__(self):
        """返回数据集的样本总数"""
        return len(self.data)

    def __getitem__(self, idx):
        """返回索引 idx 对应的样本和标签"""
        sample = self.data[idx]
        label = self.labels[idx]

        if self.transform:
            sample = self.transform(sample)

        return sample, label

# ── 使用示例 ──────────────────────────────────────────

import numpy as np

# 模拟数据
data = np.random.randn(1000, 784).astype(np.float32)
labels = np.random.randint(0, 10, size=1000)

# 创建数据集
dataset = MyDataset(data, labels)

# 划分训练集和验证集
train_size = int(0.8 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = random_split(
    dataset, [train_size, val_size],
    generator=torch.Generator().manual_seed(42)
)

print(f"Train size: {len(train_dataset)}, Val size: {len(val_dataset)}")

# ── DataLoader ────────────────────────────────────────

train_loader = DataLoader(
    dataset=train_dataset,
    batch_size=32,          # 每批样本数
    shuffle=True,           # 每个 epoch 打乱数据
    num_workers=2,          # 加载数据的子进程数（Windows 建议设为 0）
    pin_memory=True,        # 加快 GPU 数据传输
    drop_last=False         # 是否丢弃最后不足 batch_size 的样本
)

val_loader = DataLoader(
    dataset=val_dataset,
    batch_size=32,
    shuffle=False,          # 验证集不需要打乱
    num_workers=2,
    pin_memory=True
)

# 迭代数据
for batch_x, batch_y in train_loader:
    print(f"Batch x shape: {batch_x.shape}, y shape: {batch_y.shape}")
    break
from torch.utils.data import Dataset, DataLoader, random_split

# ── 自定义数据集 ──────────────────────────────────────

class MyDataset(Dataset):
    """
    自定义数据集类，必须实现 __len__ 和 __getitem__。
    """
    def __init__(self, data, labels, transform=None):
        """
        Args:
            data: 特征数据 (numpy array 或 torch.Tensor)
            labels: 标签 (numpy array 或 torch.Tensor)
            transform: 数据增强/预处理函数
        """
        self.data = data
        self.labels = labels
        self.transform = transform

    def __len__(self):
        """返回数据集的样本总数"""
        return len(self.data)

    def __getitem__(self, idx):
        """返回索引 idx 对应的样本和标签"""
        sample = self.data[idx]
        label = self.labels[idx]

        if self.transform:
            sample = self.transform(sample)

        return sample, label

# ── 使用示例 ──────────────────────────────────────────

import numpy as np

# 模拟数据
data = np.random.randn(1000, 784).astype(np.float32)
labels = np.random.randint(0, 10, size=1000)

# 创建数据集
dataset = MyDataset(data, labels)

# 划分训练集和验证集
train_size = int(0.8 * len(dataset))
val_size = len(dataset) - train_size
train_dataset, val_dataset = random_split(
    dataset, [train_size, val_size],
    generator=torch.Generator().manual_seed(42)
)

print(f"Train size: {len(train_dataset)}, Val size: {len(val_dataset)}")

# ── DataLoader ────────────────────────────────────────

train_loader = DataLoader(
    dataset=train_dataset,
    batch_size=32,          # 每批样本数
    shuffle=True,           # 每个 epoch 打乱数据
    num_workers=2,          # 加载数据的子进程数（Windows 建议设为 0）
    pin_memory=True,        # 加快 GPU 数据传输
    drop_last=False         # 是否丢弃最后不足 batch_size 的样本
)

val_loader = DataLoader(
    dataset=val_dataset,
    batch_size=32,
    shuffle=False,          # 验证集不需要打乱
    num_workers=2,
    pin_memory=True
)

# 迭代数据
for batch_x, batch_y in train_loader:
    print(f"Batch x shape: {batch_x.shape}, y shape: {batch_y.shape}")
    break

使用 torchvision 加载经典数据集 ​

python
import torchvision
import torchvision.transforms as transforms

# ── 数据预处理流水线 ──────────────────────────────────

transform = transforms.Compose([
    transforms.ToTensor(),                              # PIL/NumPy → Tensor
    transforms.Normalize(mean=[0.5], std=[0.5]),        # 标准化到 [-1, 1]
])

# 对图像数据更复杂的增强
train_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),             # 随机水平翻转
    transforms.RandomRotation(degrees=10),              # 随机旋转 ±10度
    transforms.ColorJitter(brightness=0.2, contrast=0.2),  # 颜色抖动
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],    # ImageNet 均值
                         std=[0.229, 0.224, 0.225]),    # ImageNet 标准差
])

# ── 加载 MNIST ────────────────────────────────────────

mnist_train = torchvision.datasets.MNIST(
    root="./data", train=True, download=True,
    transform=transforms.ToTensor()
)

mnist_test = torchvision.datasets.MNIST(
    root="./data", train=False, download=True,
    transform=transforms.ToTensor()
)

train_loader = DataLoader(mnist_train, batch_size=64, shuffle=True)
test_loader = DataLoader(mnist_test, batch_size=64, shuffle=False)

# ── 加载 CIFAR-10 ─────────────────────────────────────

cifar_train = torchvision.datasets.CIFAR10(
    root="./data", train=True, download=True,
    transform=train_transform
)

cifar_test = torchvision.datasets.CIFAR10(
    root="./data", train=False, download=True,
    transform=transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406],
                             std=[0.229, 0.224, 0.225]),
    ])
)

cifar_classes = ["airplane", "automobile", "bird", "cat", "deer",
                 "dog", "frog", "horse", "ship", "truck"]
import torchvision
import torchvision.transforms as transforms

# ── 数据预处理流水线 ──────────────────────────────────

transform = transforms.Compose([
    transforms.ToTensor(),                              # PIL/NumPy → Tensor
    transforms.Normalize(mean=[0.5], std=[0.5]),        # 标准化到 [-1, 1]
])

# 对图像数据更复杂的增强
train_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(p=0.5),             # 随机水平翻转
    transforms.RandomRotation(degrees=10),              # 随机旋转 ±10度
    transforms.ColorJitter(brightness=0.2, contrast=0.2),  # 颜色抖动
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],    # ImageNet 均值
                         std=[0.229, 0.224, 0.225]),    # ImageNet 标准差
])

# ── 加载 MNIST ────────────────────────────────────────

mnist_train = torchvision.datasets.MNIST(
    root="./data", train=True, download=True,
    transform=transforms.ToTensor()
)

mnist_test = torchvision.datasets.MNIST(
    root="./data", train=False, download=True,
    transform=transforms.ToTensor()
)

train_loader = DataLoader(mnist_train, batch_size=64, shuffle=True)
test_loader = DataLoader(mnist_test, batch_size=64, shuffle=False)

# ── 加载 CIFAR-10 ─────────────────────────────────────

cifar_train = torchvision.datasets.CIFAR10(
    root="./data", train=True, download=True,
    transform=train_transform
)

cifar_test = torchvision.datasets.CIFAR10(
    root="./data", train=False, download=True,
    transform=transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406],
                             std=[0.229, 0.224, 0.225]),
    ])
)

cifar_classes = ["airplane", "automobile", "bird", "cat", "deer",
                 "dog", "frog", "horse", "ship", "truck"]

CNN 示例（图像分类） ​

以下是一个完整的卷积神经网络示例，使用 CIFAR-10 数据集进行图像分类。

python
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader

# ── 1. 设置设备 ───────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")

# ── 2. 数据加载与预处理 ───────────────────────────────

batch_size = 64

transform_train = transforms.Compose([
    transforms.RandomHorizontalFlip(),
    transforms.RandomCrop(32, padding=4),
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

transform_test = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

train_dataset = torchvision.datasets.CIFAR10(
    root="./data", train=True, download=True, transform=transform_train
)
test_dataset = torchvision.datasets.CIFAR10(
    root="./data", train=False, download=True, transform=transform_test
)

train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=2)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=2)

# ── 3. 定义 CNN 模型 ──────────────────────────────────

class ConvNet(nn.Module):
    """
    一个简单的卷积神经网络用于 CIFAR-10 分类：
    卷积块 × 2 → 全连接分类器
    """
    def __init__(self, num_classes=10):
        super().__init__()
        # 第一个卷积块：3 → 32 通道，3×3 卷积
        self.conv_block1 = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),
            nn.Conv2d(32, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 32×32 → 16×16
            nn.Dropout2d(0.25),
        )

        # 第二个卷积块：32 → 64 通道
        self.conv_block2 = nn.Sequential(
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 16×16 → 8×8
            nn.Dropout2d(0.25),
        )

        # 第三个卷积块：64 → 128 通道
        self.conv_block3 = nn.Sequential(
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.Conv2d(128, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 8×8 → 4×4
            nn.Dropout2d(0.25),
        )

        # 全连接分类器
        self.classifier = nn.Sequential(
            nn.Linear(128 * 4 * 4, 256),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(256, num_classes),
        )

    def forward(self, x):
        x = self.conv_block1(x)
        x = self.conv_block2(x)
        x = self.conv_block3(x)
        x = x.view(x.size(0), -1)   # 展平： [batch, 128, 4, 4] → [batch, 128*4*4]
        x = self.classifier(x)
        return x

model = ConvNet(num_classes=10).to(device)
print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

# ── 4. 训练配置 ───────────────────────────────────────

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-4)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)

# ── 5. 训练循环 ───────────────────────────────────────

num_epochs = 50

for epoch in range(num_epochs):
    model.train()
    train_loss = 0.0
    correct = 0
    total = 0

    for inputs, labels in train_loader:
        inputs, labels = inputs.to(device), labels.to(device)

        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        train_loss += loss.item()
        _, predicted = outputs.max(1)
        total += labels.size(0)
        correct += predicted.eq(labels).sum().item()

    train_acc = 100.0 * correct / total

    # 验证
    model.eval()
    test_loss = 0.0
    correct = 0
    total = 0
    with torch.no_grad():
        for inputs, labels in test_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            outputs = model(inputs)
            loss = criterion(outputs, labels)

            test_loss += loss.item()
            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()

    test_acc = 100.0 * correct / total
    scheduler.step()

    if (epoch + 1) % 5 == 0:
        print(f"Epoch [{epoch+1}/{num_epochs}] "
              f"Train Loss: {train_loss/len(train_loader):.4f} ({train_acc:.2f}%) "
              f"Test Loss: {test_loss/len(test_loader):.4f} ({test_acc:.2f}%)")

# ── 6. 最终评估 ───────────────────────────────────────

print(f"Final test accuracy: {test_acc:.2f}%")
torch.save(model.state_dict(), "cifar10_cnn.pth")
print("Model saved to cifar10_cnn.pth")
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torchvision
import torchvision.transforms as transforms
from torch.utils.data import DataLoader

# ── 1. 设置设备 ───────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {device}")

# ── 2. 数据加载与预处理 ───────────────────────────────

batch_size = 64

transform_train = transforms.Compose([
    transforms.RandomHorizontalFlip(),
    transforms.RandomCrop(32, padding=4),
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

transform_test = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2470, 0.2435, 0.2616)),
])

train_dataset = torchvision.datasets.CIFAR10(
    root="./data", train=True, download=True, transform=transform_train
)
test_dataset = torchvision.datasets.CIFAR10(
    root="./data", train=False, download=True, transform=transform_test
)

train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=2)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=2)

# ── 3. 定义 CNN 模型 ──────────────────────────────────

class ConvNet(nn.Module):
    """
    一个简单的卷积神经网络用于 CIFAR-10 分类：
    卷积块 × 2 → 全连接分类器
    """
    def __init__(self, num_classes=10):
        super().__init__()
        # 第一个卷积块：3 → 32 通道，3×3 卷积
        self.conv_block1 = nn.Sequential(
            nn.Conv2d(3, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),
            nn.Conv2d(32, 32, kernel_size=3, padding=1),
            nn.BatchNorm2d(32),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 32×32 → 16×16
            nn.Dropout2d(0.25),
        )

        # 第二个卷积块：32 → 64 通道
        self.conv_block2 = nn.Sequential(
            nn.Conv2d(32, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, padding=1),
            nn.BatchNorm2d(64),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 16×16 → 8×8
            nn.Dropout2d(0.25),
        )

        # 第三个卷积块：64 → 128 通道
        self.conv_block3 = nn.Sequential(
            nn.Conv2d(64, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.Conv2d(128, 128, kernel_size=3, padding=1),
            nn.BatchNorm2d(128),
            nn.ReLU(),
            nn.MaxPool2d(kernel_size=2, stride=2),   # 8×8 → 4×4
            nn.Dropout2d(0.25),
        )

        # 全连接分类器
        self.classifier = nn.Sequential(
            nn.Linear(128 * 4 * 4, 256),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(256, num_classes),
        )

    def forward(self, x):
        x = self.conv_block1(x)
        x = self.conv_block2(x)
        x = self.conv_block3(x)
        x = x.view(x.size(0), -1)   # 展平： [batch, 128, 4, 4] → [batch, 128*4*4]
        x = self.classifier(x)
        return x

model = ConvNet(num_classes=10).to(device)
print(f"Model parameters: {sum(p.numel() for p in model.parameters()):,}")

# ── 4. 训练配置 ───────────────────────────────────────

criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-4)
scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=50)

# ── 5. 训练循环 ───────────────────────────────────────

num_epochs = 50

for epoch in range(num_epochs):
    model.train()
    train_loss = 0.0
    correct = 0
    total = 0

    for inputs, labels in train_loader:
        inputs, labels = inputs.to(device), labels.to(device)

        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

        train_loss += loss.item()
        _, predicted = outputs.max(1)
        total += labels.size(0)
        correct += predicted.eq(labels).sum().item()

    train_acc = 100.0 * correct / total

    # 验证
    model.eval()
    test_loss = 0.0
    correct = 0
    total = 0
    with torch.no_grad():
        for inputs, labels in test_loader:
            inputs, labels = inputs.to(device), labels.to(device)
            outputs = model(inputs)
            loss = criterion(outputs, labels)

            test_loss += loss.item()
            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()

    test_acc = 100.0 * correct / total
    scheduler.step()

    if (epoch + 1) % 5 == 0:
        print(f"Epoch [{epoch+1}/{num_epochs}] "
              f"Train Loss: {train_loss/len(train_loader):.4f} ({train_acc:.2f}%) "
              f"Test Loss: {test_loss/len(test_loader):.4f} ({test_acc:.2f}%)")

# ── 6. 最终评估 ───────────────────────────────────────

print(f"Final test accuracy: {test_acc:.2f}%")
torch.save(model.state_dict(), "cifar10_cnn.pth")
print("Model saved to cifar10_cnn.pth")

迁移学习 ​

迁移学习是利用在大规模数据集（如 ImageNet）上预训练的模型，针对特定任务进行微调的技术。它能在数据量有限时显著提升模型性能。

python
import torchvision.models as models

# ── 加载预训练模型 ────────────────────────────────────

# ResNet-18（ImageNet 预训练）
resnet = models.resnet18(pretrained=True)

# 其他可用模型
# models.resnet50(pretrained=True)
# models.vgg16(pretrained=True)
# models.densenet121(pretrained=True)
# models.efficientnet_b0(pretrained=True)
# models.mobilenet_v2(pretrained=True)

# ── 方法一：特征提取器（冻结所有层，仅训练分类头） ─────

# 冻结所有参数
for param in resnet.parameters():
    param.requires_grad = False

# 查看原始分类头
print(resnet.fc)  # Linear(in_features=512, out_features=1000, bias=True)

# 替换分类头（假设新任务有 5 个类别）
num_classes = 5
resnet.fc = nn.Linear(resnet.fc.in_features, num_classes)

# 只有分类头的参数需要梯度（约 0.4% 的参数）
trainable_params = sum(p.numel() for p in resnet.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in resnet.parameters())
print(f"Trainable: {trainable_params:,} / Total: {total_params:,} "
      f"({100 * trainable_params / total_params:.2f}%)")

# ── 方法二：全模型微调（解冻部分层） ──────────────────

# 先冻结全部
for param in resnet.parameters():
    param.requires_grad = False

# 解冻最后几个残差块
for param in resnet.layer4.parameters():
    param.requires_grad = True

# 分类头始终可训练
resnet.fc = nn.Linear(512, num_classes)

# ── 训练迁移学习模型 ──────────────────────────────────

model = resnet.to(device)
criterion = nn.CrossEntropyLoss()

# 不同层使用不同学习率（分类头学习率更大）
optimizer = optim.Adam([
    {"params": model.layer4.parameters(), "lr": 0.0001},
    {"params": model.fc.parameters(), "lr": 0.001},
])

# 也可以统一使用小学习率
# optimizer = optim.Adam(model.parameters(), lr=0.0001)

# 后续训练流程与标准循环一致
import torchvision.models as models

# ── 加载预训练模型 ────────────────────────────────────

# ResNet-18（ImageNet 预训练）
resnet = models.resnet18(pretrained=True)

# 其他可用模型
# models.resnet50(pretrained=True)
# models.vgg16(pretrained=True)
# models.densenet121(pretrained=True)
# models.efficientnet_b0(pretrained=True)
# models.mobilenet_v2(pretrained=True)

# ── 方法一：特征提取器（冻结所有层，仅训练分类头） ─────

# 冻结所有参数
for param in resnet.parameters():
    param.requires_grad = False

# 查看原始分类头
print(resnet.fc)  # Linear(in_features=512, out_features=1000, bias=True)

# 替换分类头（假设新任务有 5 个类别）
num_classes = 5
resnet.fc = nn.Linear(resnet.fc.in_features, num_classes)

# 只有分类头的参数需要梯度（约 0.4% 的参数）
trainable_params = sum(p.numel() for p in resnet.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in resnet.parameters())
print(f"Trainable: {trainable_params:,} / Total: {total_params:,} "
      f"({100 * trainable_params / total_params:.2f}%)")

# ── 方法二：全模型微调（解冻部分层） ──────────────────

# 先冻结全部
for param in resnet.parameters():
    param.requires_grad = False

# 解冻最后几个残差块
for param in resnet.layer4.parameters():
    param.requires_grad = True

# 分类头始终可训练
resnet.fc = nn.Linear(512, num_classes)

# ── 训练迁移学习模型 ──────────────────────────────────

model = resnet.to(device)
criterion = nn.CrossEntropyLoss()

# 不同层使用不同学习率（分类头学习率更大）
optimizer = optim.Adam([
    {"params": model.layer4.parameters(), "lr": 0.0001},
    {"params": model.fc.parameters(), "lr": 0.001},
])

# 也可以统一使用小学习率
# optimizer = optim.Adam(model.parameters(), lr=0.0001)

# 后续训练流程与标准循环一致

迁移学习适用场景 ​

场景建议策略数据量要求
新数据集与 ImageNet 高度相似冻结全部特征提取层，只训练分类头数百张即可
新数据集差异较大（如医学影像）解冻后几层 + 新分类头进行微调数千张起
数据量充足全模型微调，但使用较低学习率上万张
新任务与预训练任务完全不同可以仅使用预训练权重初始化，全局微调需要较多数据

模型保存与加载 ​

python
import torch

# ── 保存与加载模型参数（推荐方式） ────────────────────

# 保存 state_dict（仅参数，不保存网络结构）
torch.save(model.state_dict(), "model.pth")

# 加载
model = ConvNet()                     # 先实例化模型
model.load_state_dict(torch.load("model.pth", map_location=device))
model.eval()                          # 切换到评估模式

# ── 保存完整模型（不推荐，兼容性较差） ─────────────────

torch.save(model, "model_complete.pth")
model = torch.load("model_complete.pth", map_location=device)

# ── 保存检查点（恢复训练用） ──────────────────────────

checkpoint = {
    "epoch": epoch,
    "model_state_dict": model.state_dict(),
    "optimizer_state_dict": optimizer.state_dict(),
    "scheduler_state_dict": scheduler.state_dict(),
    "loss": loss,
    "best_acc": best_acc,
}
torch.save(checkpoint, "checkpoint.pth")

# 恢复检查点
checkpoint = torch.load("checkpoint.pth", map_location=device)
model.load_state_dict(checkpoint["model_state_dict"])
optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
scheduler.load_state_dict(checkpoint["scheduler_state_dict"])
start_epoch = checkpoint["epoch"] + 1
import torch

# ── 保存与加载模型参数（推荐方式） ────────────────────

# 保存 state_dict（仅参数，不保存网络结构）
torch.save(model.state_dict(), "model.pth")

# 加载
model = ConvNet()                     # 先实例化模型
model.load_state_dict(torch.load("model.pth", map_location=device))
model.eval()                          # 切换到评估模式

# ── 保存完整模型（不推荐，兼容性较差） ─────────────────

torch.save(model, "model_complete.pth")
model = torch.load("model_complete.pth", map_location=device)

# ── 保存检查点（恢复训练用） ──────────────────────────

checkpoint = {
    "epoch": epoch,
    "model_state_dict": model.state_dict(),
    "optimizer_state_dict": optimizer.state_dict(),
    "scheduler_state_dict": scheduler.state_dict(),
    "loss": loss,
    "best_acc": best_acc,
}
torch.save(checkpoint, "checkpoint.pth")

# 恢复检查点
checkpoint = torch.load("checkpoint.pth", map_location=device)
model.load_state_dict(checkpoint["model_state_dict"])
optimizer.load_state_dict(checkpoint["optimizer_state_dict"])
scheduler.load_state_dict(checkpoint["scheduler_state_dict"])
start_epoch = checkpoint["epoch"] + 1

保存与加载的注意事项 ​

  1. torch.save(model.state_dict(), ...) 是推荐方式，仅保存参数，不耦合模型结构定义
  2. 加载时需先实例化相同结构的模型，再调用 load_state_dict
  3. map_location=device 参数用于在不同设备间迁移（GPU → CPU 或 CPU → GPU）
  4. 推理前务必调用 model.eval() 关闭 Dropout 和 BatchNorm 的训练行为
  5. 如需在保存后继续训练，务必保存优化器状态和学习率调度器状态

GPU 训练 ​

单 GPU 训练 ​

python
# ── 设备检测 ──────────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using: {device}")
print(f"GPU count: {torch.cuda.device_count()}")
print(f"GPU name: {torch.cuda.get_device_name(0)}")

# ── 将模型和数据移动到 GPU ────────────────────────────

model = ConvNet().to(device)          # 模型移到 GPU
inputs = inputs.to(device)            # 数据移到 GPU
labels = labels.to(device)

# 注意：模型和数据必须在同一设备上，否则会报错

# ── 常见错误处理 ──────────────────────────────────────

# 错误：Expected all tensors to be on the same device
# 检查方式：
assert next(model.parameters()).device == inputs.device

# 简便的判断方法：在 DataLoader 迭代时统一移动
for batch_x, batch_y in dataloader:
    batch_x = batch_x.to(device)
    batch_y = batch_y.to(device)
    # ... 训练步骤
# ── 设备检测 ──────────────────────────────────────────

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using: {device}")
print(f"GPU count: {torch.cuda.device_count()}")
print(f"GPU name: {torch.cuda.get_device_name(0)}")

# ── 将模型和数据移动到 GPU ────────────────────────────

model = ConvNet().to(device)          # 模型移到 GPU
inputs = inputs.to(device)            # 数据移到 GPU
labels = labels.to(device)

# 注意：模型和数据必须在同一设备上，否则会报错

# ── 常见错误处理 ──────────────────────────────────────

# 错误：Expected all tensors to be on the same device
# 检查方式：
assert next(model.parameters()).device == inputs.device

# 简便的判断方法：在 DataLoader 迭代时统一移动
for batch_x, batch_y in dataloader:
    batch_x = batch_x.to(device)
    batch_y = batch_y.to(device)
    # ... 训练步骤

多 GPU 训练（DataParallel） ​

python
# ── DataParallel（简单方式） ──────────────────────────

if torch.cuda.device_count() > 1:
    print(f"Using {torch.cuda.device_count()} GPUs")
    model = nn.DataParallel(model)    # 自动将 batch 分到多个 GPU

model = model.to(device)

# 使用 DataParallel 后，模型参数名会多出 "module." 前缀
# 保存时如果想兼容单 GPU 加载，可以剥离前缀：
if isinstance(model, nn.DataParallel):
    torch.save(model.module.state_dict(), "model.pth")
else:
    torch.save(model.state_dict(), "model.pth")

# ── DistributedDataParallel（推荐，性能更好） ─────────

# DDP 比 DataParallel 更高效（每个 GPU 独立进程，避免 GIL 瓶颈）
# 适合大规模训练，需要额外的启动配置（此处仅作简介）
# 使用 torchrun 或 torch.distributed.launch 启动
# ── DataParallel（简单方式） ──────────────────────────

if torch.cuda.device_count() > 1:
    print(f"Using {torch.cuda.device_count()} GPUs")
    model = nn.DataParallel(model)    # 自动将 batch 分到多个 GPU

model = model.to(device)

# 使用 DataParallel 后，模型参数名会多出 "module." 前缀
# 保存时如果想兼容单 GPU 加载，可以剥离前缀：
if isinstance(model, nn.DataParallel):
    torch.save(model.module.state_dict(), "model.pth")
else:
    torch.save(model.state_dict(), "model.pth")

# ── DistributedDataParallel（推荐，性能更好） ─────────

# DDP 比 DataParallel 更高效（每个 GPU 独立进程，避免 GIL 瓶颈）
# 适合大规模训练，需要额外的启动配置（此处仅作简介）
# 使用 torchrun 或 torch.distributed.launch 启动

GPU 训练性能建议 ​

  1. 使用 pin_memory：DataLoader 设置 pin_memory=True 可加速 CPU → GPU 传输
  2. 使用非阻塞传输：inputs = inputs.to(device, non_blocking=True)
  3. batch_size 调大：GPU 并行计算能力强，批大小通常设 64~512
  4. 使用混合精度训练：减少显存占用，加速训练
python
# 混合精度训练（需要 torch.cuda.amp）
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()
for inputs, labels in dataloader:
    inputs, labels = inputs.to(device), labels.to(device)

    optimizer.zero_grad()

    with autocast():                  # 自动混合精度
        outputs = model(inputs)
        loss = criterion(outputs, labels)

    scaler.scale(loss).backward()     # 缩放损失以防止梯度下溢
    scaler.step(optimizer)
    scaler.update()
# 混合精度训练（需要 torch.cuda.amp）
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()
for inputs, labels in dataloader:
    inputs, labels = inputs.to(device), labels.to(device)

    optimizer.zero_grad()

    with autocast():                  # 自动混合精度
        outputs = model(inputs)
        loss = criterion(outputs, labels)

    scaler.scale(loss).backward()     # 缩放损失以防止梯度下溢
    scaler.step(optimizer)
    scaler.update()

常用工具 ​

TensorBoard 可视化 ​

TensorBoard 用于监控训练过程中的损失、准确率、模型结构等指标。

python
from torch.utils.tensorboard import SummaryWriter

# ── 创建日志写入器 ────────────────────────────────────

writer = SummaryWriter(log_dir="runs/experiment_1")

# ── 记录标量指标 ──────────────────────────────────────

for epoch in range(num_epochs):
    train_loss = ...   # 训练损失
    train_acc = ...    # 训练准确率
    test_acc = ...     # 测试准确率

    writer.add_scalar("Loss/train", train_loss, epoch)
    writer.add_scalar("Accuracy/train", train_acc, epoch)
    writer.add_scalar("Accuracy/test", test_acc, epoch)

# ── 记录模型结构 ──────────────────────────────────────

dummy_input = torch.randn(1, 3, 32, 32).to(device)
writer.add_graph(model, dummy_input)

# ── 记录图像 ──────────────────────────────────────────

images, labels = next(iter(train_loader))
grid = torchvision.utils.make_grid(images)
writer.add_image("images", grid, 0)

# ── 记录直方图（参数分布） ────────────────────────────

for name, param in model.named_parameters():
    writer.add_histogram(name, param, epoch)

# ── 关闭写入器 ────────────────────────────────────────

writer.close()

# 启动 TensorBoard： tensorboard --logdir=runs
# 浏览器访问 http://localhost:6006
from torch.utils.tensorboard import SummaryWriter

# ── 创建日志写入器 ────────────────────────────────────

writer = SummaryWriter(log_dir="runs/experiment_1")

# ── 记录标量指标 ──────────────────────────────────────

for epoch in range(num_epochs):
    train_loss = ...   # 训练损失
    train_acc = ...    # 训练准确率
    test_acc = ...     # 测试准确率

    writer.add_scalar("Loss/train", train_loss, epoch)
    writer.add_scalar("Accuracy/train", train_acc, epoch)
    writer.add_scalar("Accuracy/test", test_acc, epoch)

# ── 记录模型结构 ──────────────────────────────────────

dummy_input = torch.randn(1, 3, 32, 32).to(device)
writer.add_graph(model, dummy_input)

# ── 记录图像 ──────────────────────────────────────────

images, labels = next(iter(train_loader))
grid = torchvision.utils.make_grid(images)
writer.add_image("images", grid, 0)

# ── 记录直方图（参数分布） ────────────────────────────

for name, param in model.named_parameters():
    writer.add_histogram(name, param, epoch)

# ── 关闭写入器 ────────────────────────────────────────

writer.close()

# 启动 TensorBoard： tensorboard --logdir=runs
# 浏览器访问 http://localhost:6006

torchvision ​

torchvision 是 PyTorch 官方的计算机视觉工具库。

python
import torchvision
import torchvision.transforms as transforms

# ── 数据集 ────────────────────────────────────────────
# MNIST, FashionMNIST, CIFAR-10/100, ImageNet, SVHN, etc.

# ── 图像预处理（transforms） ──────────────────────────

transform = transforms.Compose([
    transforms.Resize((224, 224)),           # 调整尺寸
    transforms.CenterCrop(196),              # 中心裁剪
    transforms.RandomHorizontalFlip(),       # 随机翻转
    transforms.RandomRotation(15),           # 随机旋转
    transforms.ColorJitter(0.2, 0.2),        # 颜色扰动
    transforms.ToTensor(),                   # 转为 Tensor
    transforms.Normalize(mean=[0.5], std=[0.5]),  # 标准化
])

# ── 预训练模型 ────────────────────────────────────────
# models.resnet18(), models.vgg16(), models.densenet121()
# models.efficientnet_b0(), models.mobilenet_v3_small()

# ── 工具函数 ──────────────────────────────────────────
# torchvision.utils.make_grid()   # 将多个图像拼接为网格
# torchvision.utils.save_image()  # 保存张量为图像文件
import torchvision
import torchvision.transforms as transforms

# ── 数据集 ────────────────────────────────────────────
# MNIST, FashionMNIST, CIFAR-10/100, ImageNet, SVHN, etc.

# ── 图像预处理（transforms） ──────────────────────────

transform = transforms.Compose([
    transforms.Resize((224, 224)),           # 调整尺寸
    transforms.CenterCrop(196),              # 中心裁剪
    transforms.RandomHorizontalFlip(),       # 随机翻转
    transforms.RandomRotation(15),           # 随机旋转
    transforms.ColorJitter(0.2, 0.2),        # 颜色扰动
    transforms.ToTensor(),                   # 转为 Tensor
    transforms.Normalize(mean=[0.5], std=[0.5]),  # 标准化
])

# ── 预训练模型 ────────────────────────────────────────
# models.resnet18(), models.vgg16(), models.densenet121()
# models.efficientnet_b0(), models.mobilenet_v3_small()

# ── 工具函数 ──────────────────────────────────────────
# torchvision.utils.make_grid()   # 将多个图像拼接为网格
# torchvision.utils.save_image()  # 保存张量为图像文件

tqdm 进度条 ​

python
from tqdm import tqdm
import time

# ── 简单进度条 ────────────────────────────────────────

for epoch in range(num_epochs):
    loop = tqdm(train_loader, desc=f"Epoch {epoch+1}/{num_epochs}")
    for inputs, labels in loop:
        # ... 训练代码 ...

        # 更新进度条显示信息
        loop.set_postfix(loss=loss.item(), acc=accuracy)

# ── 嵌套进度条（epoch + batch） ───────────────────────

from tqdm import trange

for epoch in trange(num_epochs, desc="Training"):
    for inputs, labels in tqdm(train_loader, desc=f"Epoch {epoch+1}", leave=False):
        # ... 训练代码 ...
        pass
from tqdm import tqdm
import time

# ── 简单进度条 ────────────────────────────────────────

for epoch in range(num_epochs):
    loop = tqdm(train_loader, desc=f"Epoch {epoch+1}/{num_epochs}")
    for inputs, labels in loop:
        # ... 训练代码 ...

        # 更新进度条显示信息
        loop.set_postfix(loss=loss.item(), acc=accuracy)

# ── 嵌套进度条（epoch + batch） ───────────────────────

from tqdm import trange

for epoch in trange(num_epochs, desc="Training"):
    for inputs, labels in tqdm(train_loader, desc=f"Epoch {epoch+1}", leave=False):
        # ... 训练代码 ...
        pass

实用技巧与常见陷阱 ​

调试技巧 ​

python
# ── 启用异常检测（便于发现 NaN 等问题） ───────────────

torch.autograd.set_detect_anomaly(True)

# ── 检查梯度是否为 NaN ────────────────────────────────

for name, param in model.named_parameters():
    if param.grad is not None and torch.isnan(param.grad).any():
        print(f"NaN gradient in {name}")

# ── 梯度裁剪（防止梯度爆炸） ──────────────────────────

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

# ── 查看中间层输出 ────────────────────────────────────

# 方法一：注册 forward hook
activations = {}
def get_activation(name):
    def hook(model, input, output):
        activations[name] = output.detach()
    return hook

model.fc1.register_forward_hook(get_activation("fc1"))
# ── 启用异常检测（便于发现 NaN 等问题） ───────────────

torch.autograd.set_detect_anomaly(True)

# ── 检查梯度是否为 NaN ────────────────────────────────

for name, param in model.named_parameters():
    if param.grad is not None and torch.isnan(param.grad).any():
        print(f"NaN gradient in {name}")

# ── 梯度裁剪（防止梯度爆炸） ──────────────────────────

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

# ── 查看中间层输出 ────────────────────────────────────

# 方法一：注册 forward hook
activations = {}
def get_activation(name):
    def hook(model, input, output):
        activations[name] = output.detach()
    return hook

model.fc1.register_forward_hook(get_activation("fc1"))

常见错误及解决办法 ​

错误原因解决
Expected all tensors on same device模型或数据未正确移到 GPU检查 .to(device) 是否遗漏
CUDA out of memory显存不足降低 batch_size，使用梯度累积
RuntimeError: mat1 and mat2 shapes not aligned张量形状不匹配检查网络层之间的维度
loss.item() 显示 nan梯度爆炸或学习率过大降低学习率，添加梯度裁剪
DataLoader 卡死num_workers 在 Windows 下有问题在 Windows 设置 num_workers=0

可重复性设置 ​

python
def set_seed(seed=42):
    """设置随机种子以确保结果可复现"""
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    np.random.seed(seed)
    # 以下设置会降低性能，仅调试时使用
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

set_seed(42)
def set_seed(seed=42):
    """设置随机种子以确保结果可复现"""
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    np.random.seed(seed)
    # 以下设置会降低性能，仅调试时使用
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

set_seed(42)

学习建议 ​

学习路线 ​

  1. 掌握基础：确保理解 Python、NumPy 和 PyTorch Tensor 的基本操作
  2. 从经典数据集开始：MNIST → Fashion-MNIST → CIFAR-10，逐步增加难度
  3. 先理解训练循环：透彻理解前向传播、损失计算、反向传播、参数更新这一完整流程
  4. 动手实现标准模型：自己实现简单的全连接网络和 CNN，而不是直接调用高层 API
  5. 使用 TensorBoard：养成监控训练过程的习惯，观察损失曲线和准确率变化
  6. 逐步增加复杂度：全连接网络 → 简单 CNN → 残差网络 → 预训练模型微调
  7. 阅读和复现论文：从经典论文（AlexNet、VGG、ResNet）开始，尝试复现核心结果

推荐资源 ​

  • 官方文档：PyTorch Tutorials
  • 书籍：《动手学深度学习》（李沐） - 免费在线版，PyTorch 实现
  • 课程：CS231n（Stanford，计算机视觉）、CS224n（Stanford，NLP）
  • 代码仓库：pytorch/examples、pytorch/vision

避坑指南 ​

  • 不要在验证/测试时忘记调用 model.eval()——Dropout 会灾难性地影响结果
  • 不要在训练时忘记调用 optimizer.zero_grad()——梯度会不断累积
  • 数据增强时，注意只在训练集上使用，验证/测试集仅做标准化
  • 迁移学习时，预训练模型的标准化参数（mean/std）必须与原训练数据一致
  • 保存模型推荐使用 state_dict 而非整个模型，确保兼容性
  • 分布式训练时，注意 batch_size 的含义是 每个 GPU 的 batch_size

下一步学习方向 ​

  • 高级模型架构：Vision Transformer（ViT）、Swin Transformer、ConvNeXt
  • 生成模型：VAE、GAN、Diffusion Model（Stable Diffusion）
  • 自然语言处理：RNN/LSTM → Transformer → BERT → GPT 系列
  • 模型部署：ONNX 导出、TorchScript、TensorRT 加速
  • 强化学习：DQN、PPO、SAC 与 PyTorch 实现

Copyright © 2024-present xiaocao