S 3.3深度学习--卷积神经网络--代码

一·数据预处理Dataset

在操作之前先回顾一下，需要对数据进行一个标签的处理

import osdef train_test_file(root, dir):file_txt = open(dir + '.txt', 'w')path = os.path.join(root, dir)for roots, directories, files in os.walk(path):  # os.list_dir()if len(directories) != 0:dirs = directorieselse:now_dir = roots.split('\\')for file in files:path_1 = os.path.join(roots, file)print(path_1)file_txt.write(path_1 + ' ' + str(dirs.index(now_dir[-1])) + '\n')file_txt.close()

二·数据增强

数据增强(Data Augmentation)：缓解深度学习中数据不足的场景，在图像领域首先得到广泛使用，进而延伸到NLP领域，并在许多任务上取得效果。一个主要的方向是增加训练数据的多样性，从而提高模型泛化能力。

垂直翻转

随机旋转

随机裁剪

颜色变换

PIL（Python Imaging Library）是 Python 中一个强大的图像处理库，现在更常用的是它的分支版本 Pillow（可以看作是 PIL 的升级版和维护版本）。它提供了丰富的图像处理功能，包括打开、保存、裁剪、缩放、旋转图像，以及像素操作、滤镜应用等。

import torch
from torch.utils.data import Dataset, DataLoader  # 用于处理数据集的
import numpy as np
from PIL import Image
from torchvision import transforms  # 对数据进行处理工具 转换
import torch.nn as nn  # 这行是关键，导入nn模块
data_transforms = {'train':transforms.Compose([transforms.Resize([300, 300]),transforms.RandomRotation(45),transforms.CenterCrop(256),transforms.RandomHorizontalFlip(p=0.5),transforms.RandomVerticalFlip(p=0.5),transforms.ColorJitter(brightness=0.2, contrast=0.1, saturation=0.1, hue=0.1),transforms.RandomGrayscale(p=0.1),transforms.ToTensor(),transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])]),'valid':transforms.Compose([transforms.Resize([256, 256]),transforms.ToTensor(),transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])]),
}class food_dataset(Dataset):  # food_dataset是自己创建的类名称，可以改为你需要的名称def __init__(self, file_path, transform=None):  # 类的初始化，解析数据文件txtself.file_path = file_pathself.imgs = []  # 存储图片的路径self.labels = []  # 存储图片的标签结果self.transform = transformwith open(self.file_path) as f:  # 是把train.txt文件中图片的路径保存在 self.imgs，train.txt文件中标签保samples = [x.strip().split(' ') for x in f.readlines()]for img_path, label in samples:self.imgs.append(img_path)  # 图像的路径self.labels.append(label)  # 标签，还不是tensor# 初始化：把图片目录加载到self.def __len__(self):  # 类实例化对象后，可以使用len函数测量对象的个数 ls=[12,3,4,4] len(training_data)return len(self.imgs)# training_data[1]def __getitem__(self, idx):  # 关键，可通过索引的形式获取每一个图片数据及标签image = Image.open(self.imgs[idx])  # 读取到图片数据，还不是tensor，BGRif self.transform:  # 将pil图像数据转换为tensorimage = self.transform(image)  # 图像处理为256*256，转换为tenorlabel = self.labels[idx]  # label还不是tensorlabel = torch.from_numpy(np.array(label, dtype=np.int64))  # label也转换为tensor，return image, labeltraining_data = food_dataset(file_path = './train.txt',transform = data_transforms['train'])
test_data = food_dataset(file_path = './test.txt',transform = data_transforms['valid'])train_dataloader = DataLoader(training_data, batch_size=64,shuffle=True)
test_dataloader = DataLoader(test_data, batch_size=64,shuffle=True)device = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
print(f"Using {device} device")import torch.nn as nn  # 这行是关键，导入nn模块
class CNN(nn.Module):def __init__(self):super(CNN, self).__init__()self.conv1 = nn.Sequential(nn.Conv2d(in_channels=3,out_channels=16,kernel_size=5,stride=1,padding=2,),nn.ReLU(),nn.MaxPool2d(kernel_size=2),)self.conv2 = nn.Sequential(nn.Conv2d(16, 32, 5, 1, 2),nn.ReLU(),nn.Conv2d(32, 32, 5, 1, 2),nn.ReLU(),nn.MaxPool2d(2),)self.conv3 = nn.Sequential(nn.Conv2d(32, 128, 5, 1, 2),nn.ReLU(),)self.out = nn.Linear(128 * 64 * 64, 20)def forward(self, x):x = self.conv1(x)x = self.conv2(x)x = self.conv3(x)x = x.view(x.size(0), -1)output = self.out(x)return outputmodel = CNN().to(device)
print(model)def train(dataloader, model, loss_fn, optimizer):model.train()# batch_size_num = 1for X, y in dataloader:X, y = X.to(device), y.to(device)pred = model(X)loss = loss_fn(pred, y)optimizer.zero_grad()loss.backward()optimizer.step()loss_value = loss.item()# if batch_size_num % 100 == 0:#     print(f"loss: {loss_value}  [number: {batch_size_num}]")# batch_size_num += 1def test(dataLoader, model, loss_fn):size = len(dataLoader.dataset)num_batches = len(dataLoader)model.eval()test_loss, correct = 0, 0with torch.no_grad():for X, y in dataLoader:X, y = X.to(device), y.to(device)pred = model.forward(X)test_loss += loss_fn(pred, y).item()correct += (pred.argmax(1) == y).type(torch.float).sum().item()test_loss /= num_batchescorrect /= sizeprint(f"Test result: \n Accuracy: {100 * correct}%, Avg loss: {test_loss}")loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)epochs = 10
for t in range(epochs):print(f"Epoch {t + 1}\n-------------------------------")train(train_dataloader, model, loss_fn, optimizer)print("Done!")
test(test_dataloader, model, loss_fn)

三·保存最优模型

模型的三个方面：train test 模型的推理

一、整体框架概览

代码的核心目标是：用自定义的卷积神经网络（CNN），对 “食品图像数据集” 进行分类训练，并测试模型性能、保存最优模型。整体流程分为 5 个核心模块：

1. 导入依赖库 → 2. 数据预处理与加载 → 3. 自定义 CNN 网络搭建 → 4. 训练 / 测试函数定义 → 5. 执行训练与测试

二、逐模块详细讲解

1. 导入依赖库

首先导入所有需要的工具库，每个库的作用都已标注：

python

运行

import torch  # PyTorch核心库（张量计算、自动求导等）
from torch.utils.data import Dataset, DataLoader  # 数据集处理核心类（自定义数据集、批量加载）
import numpy as np  # 数值计算库（处理标签的数组转换）
from PIL import Image  # 图像读取库（读取本地图片文件）
from torchvision import transforms  # 图像预处理工具（缩放、旋转、归一化等）
import torch.nn as nn  # PyTorch神经网络核心模块（卷积、线性层、损失函数等）

2. 图像预处理配置（data_transforms）

为了提升模型泛化能力，训练集需要加入 “数据增强”（随机变换避免过拟合），验证 / 测试集只做基础预处理（保证数据一致性）。这里用 transforms.Compose 把多个预处理操作串联成 “流水线”。

代码实现：

python

运行

data_transforms = {'train': transforms.Compose([...]),  # 训练集增强流水线'valid': transforms.Compose([...])   # 测试集基础流水线
}

3. 自定义数据集类（food_dataset）

PyTorch 的 Dataset 是抽象类，需自定义子类实现 3 个核心方法，才能让数据被 DataLoader 批量加载：

• __init__：初始化（读取图片路径和标签、绑定预处理流水线）
• __len__：返回数据集总样本数（让 len(dataset) 生效）
• __getitem__：按索引返回单个样本（图片张量 + 标签张量，让 dataset[idx] 生效）

3.1 关键逻辑解析

1. 标签文件格式：代码假设 train.txt / test.txt 中每行格式为 图片路径 标签（例如 ./images/apple1.jpg 0），通过 split(' ') 拆分路径和标签。
2. 图片读取：用 PIL.Image.open() 读取图片（避免 OpenCV 的 BGR 格式问题，PyTorch 默认适配 PIL 的 RGB）。
3. 标签转换：原始标签是字符串，需先转成 numpy.int64（分类任务标签需为整数），再转成 torch.Tensor（适配 PyTorch 计算）。

代码实现：

python

运行

class food_dataset(Dataset):def __init__(self, file_path, transform=None):self.file_path = file_path  # 标签文件路径（train.txt/test.txt）self.imgs = []  # 存储所有图片的路径self.labels = []  # 存储所有图片的标签（字符串格式，后续转张量）self.transform = transform  # 绑定预处理流水线# 读取标签文件，拆分路径和标签with open(self.file_path) as f:# 每行拆分为 [图片路径, 标签]，strip() 去除换行符/空格samples = [x.strip().split(' ') for x in f.readlines()]for img_path, label in samples:self.imgs.append(img_path)self.labels.append(label)def __len__(self):# 返回数据集总样本数return len(self.imgs)def __getitem__(self, idx):# 按索引idx获取单个样本image = Image.open(self.imgs[idx])  # 读取图片（PIL格式）if self.transform:image = self.transform(image)  # 应用预处理，转为张量# 标签转张量（字符串→numpy.int64→torch.Tensor）label = self.labels[idx]label = torch.from_numpy(np.array(label, dtype=np.int64))return image, label  # 返回（图片张量，标签张量）

4. 初始化数据集与 DataLoader

用自定义的 food_dataset 类创建训练 / 测试数据集，并通过 DataLoader 实现 批量加载、打乱、多线程读取（提升训练效率）。

python

运行

# 1. 创建训练/测试数据集（绑定对应的预处理流水线）
training_data = food_dataset(file_path='./train.txt', transform=data_transforms['train'])
test_data = food_dataset(file_path='./test.txt', transform=data_transforms['valid'])# 2. 创建DataLoader（批量加载工具）
train_dataloader = DataLoader(training_data, batch_size=64,  # 每次加载64个样本（批次大小，根据GPU显存调整）shuffle=True    # 训练集打乱（避免样本顺序影响训练）
)
test_dataloader = DataLoader(test_data, batch_size=64, shuffle=True    # 测试集打乱与否不影响结果，仅为方便查看
)

5. 设备配置（CPU/GPU/MPS）

自动检测可用设备，优先使用 GPU（NVIDIA CUDA），其次是苹果 M 系列芯片的 MPS，最后用 CPU（GPU 能大幅加速训练）：

python

运行

device = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
print(f"Using {device} device")  # 例如输出 "Using cuda device"

6. 自定义 CNN 网络（CNN类）

继承 nn.Module 实现自定义卷积神经网络，网络结构分为 特征提取层（卷积 + 池化） 和 分类层（全连接），核心是 __init__（定义网络层）和 forward（定义前向传播流程）。

6.1 网络结构拆解

假设输入图片尺寸为 3×256×256（C×H×W），各层作用和输出尺寸如下：

注：代码中 self.out2 未被使用（冗余定义），不影响网络运行，但可删除以精简代码。

6.2 代码实现

python

运行

class CNN(nn.Module):def __init__(self):super(CNN, self).__init__()  # 继承nn.Module的初始化# 第一层卷积+池化（conv1）self.conv1 = nn.Sequential(# 卷积层：输入3通道（RGB）→输出16通道，卷积核5×5，步长1， padding=2（保证输入输出H/W一致）nn.Conv2d(in_channels=3, out_channels=16, kernel_size=5, stride=1, padding=2),nn.ReLU(),  # 激活函数（引入非线性，提升模型表达能力）nn.MaxPool2d(kernel_size=2)  # 池化层：2×2池化，H/W缩小为1/2)# 第二层卷积+池化（conv2：两层卷积+一次池化）self.conv2 = nn.Sequential(nn.Conv2d(16, 32, 5, 1, 2),  # 16→32通道nn.ReLU(),nn.Conv2d(32, 32, 5, 1, 2),  # 32→32通道（加深网络，提取更复杂特征）nn.ReLU(),nn.MaxPool2d(2)  # H/W再缩小为1/2)# 第三层卷积（conv3：无池化，保留尺寸）self.conv3 = nn.Sequential(nn.Conv2d(32, 128, 5, 1, 2),  # 32→128通道（进一步提升特征维度）nn.ReLU())# 全连接层（分类层）：输入特征数=128×64×64，输出20类self.out1 = nn.Linear(64 * 64 * 128, 20)self.out2 = nn.Linear(64 * 64 * 128, 20)  # 冗余层，未使用，可删除def forward(self, x):# 前向传播：定义数据在网络中的流动路径x = self.conv1(x)  # 经过conv1x = self.conv2(x)  # 经过conv2x = self.conv3(x)  # 经过conv3x = x.view(x.size(0), -1)  # 展平：(batch, C, H, W) → (batch, C*H*W)output = self.out1(x)  # 经过全连接层，输出分类结果return output# 初始化模型，并移动到指定设备（CPU/GPU）
model = CNN().to(device)
print(model)  # 打印模型结构，验证是否正确

7. 损失函数与优化器

• 损失函数：用 nn.CrossEntropyLoss()，适用于多分类任务（自动包含 Softmax 激活，无需在网络输出层额外添加）。
• 优化器：用 torch.optim.Adam（自适应学习率优化器，收敛速度快于 SGD），学习率 lr=0.1（注意：此学习率可能偏高，实际训练中可调整为 0.001 或 0.0001，避免损失震荡不收敛）。

python

运行

loss_fn = nn.CrossEntropyLoss()  # 多分类交叉熵损失
optimizer = torch.optim.Adam(model.parameters(), lr=0.1)  # Adam优化器，优化模型所有参数

8. 训练函数（train）

训练函数的核心是 “前向传播计算损失 → 反向传播更新参数”，流程如下：

1. 设模型为训练模式（model.train()）：启用 Dropout、BatchNorm 等训练时特有的层。
2. 遍历 DataLoader，获取每个批次的（图片 X，标签 y）。
3. 把 X 和 y 移动到指定设备（与模型同设备）。
4. 前向传播：用 model(X) 计算模型预测值 pred。
5. 计算损失：用 loss_fn(pred, y) 计算预测值与真实标签的差距。
6. 反向传播：
   • optimizer.zero_grad()：清空上一轮的梯度（避免梯度累积）。
   • loss.backward()：自动计算各参数的梯度（链式法则）。
   • optimizer.step()：根据梯度更新模型参数（最小化损失）。

python

运行

def train(dataloader, model, loss_fn, optimizer):model.train()  # 启用训练模式for X, y in dataloader:  # 遍历每个批次X, y = X.to(device), y.to(device)  # 移动数据到设备# 前向传播：计算预测值pred = model(X)  # 等价于 model.forward(X)# 计算损失loss = loss_fn(pred, y)# 反向传播：更新参数optimizer.zero_grad()  # 清空梯度loss.backward()        # 计算梯度optimizer.step()       # 更新参数# （可选）打印每批次损失（代码中注释了，可根据需要启用）loss_value = loss.item()# if batch_size_num % 100 == 0:#     print(f"loss: {loss_value}  [number: {batch_size_num}]")# batch_size_num += 1

9. 测试函数（test）

测试函数的核心是 评估模型在测试集上的性能（准确率、平均损失），并保存 “最优模型”（准确率最高的模型），流程如下：

1. 设模型为评估模式（model.eval()）：关闭 Dropout、固定 BatchNorm 的统计量，保证预测结果稳定。
2. 用 torch.no_grad() 禁用梯度计算（测试阶段无需更新参数，节省内存和时间）。
3. 遍历测试集，计算总损失和正确预测数。
4. 计算 平均损失（总损失 / 测试集批次数）和 准确率（正确预测数 / 测试集总样本数）。
5. 保存最优模型：若当前准确率高于历史最高（best_acc），则更新 best_acc 并保存模型参数（用 torch.save(model.state_dict(), ...)，仅保存参数，文件小、加载快）。

python

运行

best_acc = 0  # 记录历史最高准确率def test(dataLoader, model, loss_fn):global best_acc  # 声明使用全局变量best_accsize = len(dataLoader.dataset)  # 测试集总样本数num_batches = len(dataLoader)   # 测试集批次数model.eval()                    # 启用评估模式test_loss, correct = 0, 0       # 总损失、正确预测数with torch.no_grad():  # 禁用梯度计算（测试阶段无需反向传播）for X, y in dataLoader:X, y = X.to(device), y.to(device)pred = model(X)# 累加损失（注意：loss_fn返回的是单批次损失，需累加后求平均）test_loss += loss_fn(pred, y).item()# 累加正确预测数：pred.argmax(1)取预测概率最大的类别，与y比较correct += (pred.arg