项目需求：
1. 业务需求
实现11类食物图像自动分类，解决传统图像分类标注数据不足、人工标注成本高的痛点。仅使用少量有标签食物数据、大量无标签食物数据，通过半监督学习提升模型分类精度，适配轻量化食物识别场景。
2. 技术需求
数据层面：支持有标签数据、无标签数据双数据源读取；实现数据增强、数据集自定义封装；筛选高置信度无标签数据伪标签
模型层面：支持自定义CNN模型、预训练VGG11模型迁移学习；引入半监督训练策略，利用无标签数据辅助模型优化
训练层面：固定随机种子保证实验可复现；实现训练/验证双流程、损失与准确率统计；自动保存最优模型；可视化训练曲线
半监督核心：模型迭代过程中，自动对无标签数据打伪标签，筛选置信度≥0.99的高质量数据加入训练集，持续提升模型性能
项目整体规划：
本项目采用有标签数据监督训练+无标签数据伪标签半监督优化的两阶段训练思路，整体分为6大模块，逻辑闭环、循序渐进：
1.环境与随机种子模块
统一所有随机因子，固定训练结果，确保每次运行实验结果一致，杜绝结果随机性波动，满足科研/实验可复现要求。
2.数据预处理与增强模块
针对训练集做随机裁剪、随机旋转数据增强，提升模型泛化能力；验证集仅做标准化预处理，保证验证结果客观真实。统一图像尺寸为224×224，适配主流CNN模型输入规格。
3.自定义数据集模块
封装三类数据集：有标签训练集、有标签验证集、无标签数据集；自动读取文件夹分层数据、匹配标签，适配食物11分类的文件夹数据格式。
4.半监督伪标签生成模块
核心创新模块！利用当前训练好的模型对无标签数据预测，筛选高置信度样本生成伪标签，构建半监督数据集，将无标签数据转化为可用训练数据。
5.模型构建模块
提供两种模型方案：自定义轻量化CNN、预训练VGG11迁移学习；适配11类食物分类任务，完成特征提取+分类输出全流程。
6.训练、验证与可视化模块
循环迭代训练，融合监督数据+半监督数据更新模型参数；统计每轮训练/验证损失、准确率；自动保存最优模型；最后可视化损失、准确率变化曲线，直观展示模型收敛效果。
项目代码实现：
导包部分：
import random #导入Python随机库，用于控制随机裁剪、随机旋转的随机种子 import torch #Pytorch核心库,负责张量运算、模型搭建、训练迭代 import torch.nn as nn #Pytorch神经网络核心模块，包含卷积、BN、全连接、损失函数等所有网络层 import numpy as np #数值计算库，用于存储图像数组、统计准确率 import os #系统路径库，用于遍历文件夹、拼接图片路径、读取本地数据 import time #计时工具，统计每轮训练耗时 import matplotlib.pyplot as plt #绘图工具，绘制损失、准确率训练曲线 from PIL import Image #Python图像处理库，用于打开、缩放图片 from torch.utils.data import Dataset, DataLoader #Python数据集核心工具 #Dataset:自定义数据集父类 #DataLoader:数据加载器，实现分批、打乱、并行读取数据 from tqdm import tqdm #进度条工具，读取大量图片时显示加载进度，直观看到数据读取状态 from torchvision import transforms #图像预处理、数据增强工具库 from model_util.model import initilalize_model #自定义工具函数，用于快速加载预训练VGG、ResNet等模型全局随机种子固定：
深度学习存在大量随机操作（数据增强、参数初始化、GPU卷积计算），不固定种子，每次训练结果完全不同，无法对比模型效果。
def seed_everything(seed) torch.manual_seed(seed) #固定cpu上Pytorch的随机种子 torch.cuda.maanul_seed(seed) #固定单块GPU的随机种子 torch.cuda.manual_sedd_all(seed) #固定多块GPU的随机种子（兼容多卡训练） torch.backends.cudnn.benchmaek = False #关闭GPU卷积优化算法 #benchmark=True时GPU会自动选最优卷积算法，存在随机性，关闭后杜绝随机 torch.backends.cudnn.deterministic = True #开启GPU确定性计算，保证每次卷积、池化结果完全一致 random.seed(seed) #固定Python原生随机操作（随机旋转、随机裁剪） np.random.seed(sedd) #固定Numpy数组的随机操作 os.environ['PYTHONHASHSEED'] = str(seed) #固定Python哈希随机种子，防止字典、列表遍历顺序随机 seed_everything(0) #设置全局随机种子为0，全程固定所有随机操作超参数与数据增强配置
HW = 224 #定义统一图像尺寸224*224，VGG、ResNet等主流预训练模型均适配该尺寸 train_transform = transforms.Compose( #Compose组合多个预处理操作，按顺序串行执行 [ transforms.ToPILImage(), #将numpy数组格式的图片转为PIL图像格式 transforms.RandomResizedCrop(224), #训练集核心增强：随机裁剪图像再缩放到224，模拟不同拍摄视角，提升泛化能力 transforms.RandomRotation(50), #随机旋转±50度，增强图像多样性，防止模型过拟合 transforms.ToTensor(), #将PIL图像/ numpy数组转为张量 #同时归一化像素值0~255 → 0~1，维度从(H,W,C) → (C,H,W)，适配模型输入 ] ) val_transform = transforms.Compose( #验证集绝对不能做随机增强！否则验证结果失真，无法客观评估模型真实性能，仅做格式转换 [ transforms.ToPILImage(), transforms.ToTensor() ] )主函数
train_path = r"F:\pycharm\beike\classification\food_classification \food-11_sample\training\labeled" val_path = r"F:\pycharm\beike\classification\food_classification \food-11_sample\validation" no_label_path = r"F:\pycharm\beike\classification\food_classification \food-11_sample\training\unlabeled\00" #记录有标签训练数据，验证数据，无标签数据集合的地址 train_set = food_Dataset(train_path, ""train) val_set = food_Dataset(val_path, "val") no_label_set = food_Dataset(no_label_path, "semi") #实例化训练集、验证集、无标签集 train_loader = DataLoader(train_set, batch_size=16, shuffle=True) val_loader = DataLoader(val_set, batch_size=16, shuffle=True) no_lable_loader = DataLoader(no_lable_set, batch_size=16, shuffle=False) #构建dataloader训练和验证集shuffle打乱顺序，无标签集不打乱quebao model = myModel(11)#加载前向模型 lr = 0.001 #设置学习率 loss = nn.CrossEntropyLoss()#分类专用交叉熵损失 optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=1e-1) #AdamW优化器,用于权重衰减，梯度更新 device = "cuda" if torch.cuda.is_available() else "cpu"#选择GPU/CPU save_path = "model_save/best_model.pth"#最优模型保存路径 epoch = 15 #训练总轮数 tres =0.99 #伪标签置信度阈值 train_val(model, train_loader, val_loader, no_lable_loader, device, epochs, optimizer, loss, thres, save_path) #调用训练函数，启动完整训练流程自定义数据集类food_Dataset
class food_Dataset(Dataset): def __init__(self, path, mode="train"):#构造函数初始化数据集path:路径;mode:模式 self.mode = mode #保存当前数据集模式，区分训练、验证、无标签数据 if mode == "semi": #无标签数据模式 self.X = self.read_file(path) #只读取图像数据self.X,无标签self.Y else: #训练/验证模式 self.X, self.Y = self.read_file(path) #同时读取图像和标签 self.Y = torch.LongTensor(self.Y) #分类任务强制要求：标签转为长整型 #CrossEntropyLoss损失函数仅支持LongTensor标签 if mode == "train": #训练集增强：随机裁剪、随机旋转 self.transform = train_transform else: #验证集：只转张量，不做随机增强 self.transform = val_transform def __getitem__(self, itm): #PyTtorch Dataset必须实现方法,当执行dataset[item]时自动调用这个函数 #DataLoader内部就是不停调用这个函数拿单条样本 if self.mode == "semi": return self.transfrom(self.X[item]), self.X[item] #无标签返回增强后的tensor图片, 原始numpy文件 else: return self.transform(self.X[item]), self.Y[item] def __len__(self): return len(self.X) def read_file(self, path): #内部封装文件读取函数 if self.mode == "semi" #semi模式读取无标签数据，只有图片没有标签 file_list = os.listdir(path)#读取path文件夹下所有文件名，返回一个字符串列表 xi = np.zeros((len(file_list), HW, HW, 3), dtype=np.unit8) #np.zeros(shape)创建全0的numpy数组 #shape=(len(file_list), HW, HW, 3)(N,H,W,C)样本数，高度，宽度，通道数 #dtype=np.unit8:像素值类型，0-255,图片像素标准类型 for j, img_name in enumerate(file_list): #enumerate同时拿到下标j和文件名img_name img_path = os.path.join(path, img_name)#拼接路径，得到图片文件完整路径 img = Image.open(img_path)#读取图片文件路径，返回PIL图片对象 img = img.resize((HW, HW))#将图片强制缩放为(224,224) xi[j,...] = img #img是(H,W,C)类型，赋值给xi的第j个样本 print("读到了%d个数据" % len(file_list)) return xi #返回图片nummpy数组xi，没有标签 else: #train或val模式，读取带标签数据集 for i in tqdm(range(11)): #读取11个标签类别，并加上进度条更直观 file_dir = path + "/%02d" % i #拼接当前类别文件夹路径 file_list = os.listdir(file_dir) #读取当前类别下所有图片文件名列表 xi = np.zeros((len(file_list), HW, HW, 3), dtype=np.uint8) yi = np.zeros(len(file_list), dtype=np.uint8) #保存当前类别所有图片的标签，长度为图片数量 for j, img_name in enumerate(file_list): #enumerate同时拿到下标j和文件名img_name img_path = os.path.join(filedir, img_name) #拼接路径，得到图片文件完整路径 img = Image.open(img_path)#读取图片文件路径，返回PIL图片对象 img = img.resize((HW, HW))#将图片强制缩放为(224,224) xi[j,...] = img #img是(H,W,C)类型，赋值给xi的第j个样本 yi[j] = i #当前图片标签 = 类别编号i if i == 0: #第一次循环初始化X，Y X = xi Y = yi else: X = np.concatenate((X, xi), axis=0) Y = np.concatenate((Y, yi), axis=0) #X,Y保存所有图片、标签 #np.concatenate:numpy数组拼接，axis=0表示在0维（样本维度）拼接 print("读到了%d个数据" % len(Y))#读完11个类别所有图片,len(Y)可以表示图片总数 return X, Y #返回全部图片 numpy 数组 X，全部标签 numpy 数组 Y自定义CNN模型myModel
负责完成数据前向传播：卷积和全连接
class myModel(nn.Module): def __init__(self, num_class): #num_calss为输出类别数量,本项目为11 super(myModel, self).__init__()#必须调用执行父类构造，注册所有网络层 self.conv1 = nn.Conv2d(3,64,3,1,1)#卷积 self.bn1 = nn.BatchNorm2d(64)#该批次数据归一化 self.relu = nn.ReLU()#激活函数 self.pool1 = nn.MaxPool2d(2)#池化 #Sequential顺序容器，层按顺序串行执行 self.layer1 = nn.Sequential( nn.Conv2d(64,128,3,1,1)#卷积 nn.BatchNorm2d(128)#该批次数据归一化 nn.ReLU()#激活函数 nn.MaxPool2d(2)#池化 ) self.layer2 = nn.Sequential( nn.Conv2d(128,256,3,1,1)#卷积 nn.BatchNorm2d(256)#该批次数据归一化 nn.ReLU()#激活函数 nn.MaxPool2d(2)#池化 ) self.layer3 = nn.Sequential( nn.Conv2d(256,512,3,1,1)#卷积 nn.BatchNorm2d(512)#该批次数据归一化 nn.ReLU()#激活函数 nn.MaxPool2d(2)#池化 ) self.pool2 = nn.MaxPool2d(2)#最后一次池化 self.fc1 = nn.Linear(25088, 1000)#全连接层输入维度(512*7*7),输出1000维特征 self.relu2 = nn.ReLU() self.fc2 = nn.Linear(1000, num_class)#1000维映射到num_class个类别输出 def __forward(self, x): #x输入张量shape[B,3,224,224] #经过卷积层 x = self.conv1(x) x = self.bn1(x) x = self.relu(x) x = self.pool1(x) x = self.layer1(x) x = self.layer2(x) x = self.layer3(x) x = self.pool2(x) x = x.view(x.size()[0], -1)#将x展平为一维数据 #经过全连接层 x = self.fc1(x) x = self.relu2(x) x = self.fc2(x) return x核心训练函数train_val
def train_val(model, train_loader, val_loader, no_lableloader, device, epochs, optimizer, loss, thres, save_path) model = model.to(device)#将模型移动到指定设备 semi_loader = None #初始化半监督dataloader变量为None #保存所有轮次总训练和验证损失,用于绘图 plt_train_loss = [] plt_val_loss = [] #保存所有轮次总训练和验证准确率以及最大准确率 plt_train_acc = [] plt_val_acc = [] max_acc = 0.0 for epoch in range(epochs): #训练epochs个轮次 #初始化本轮训练、验证、半监督损失 train_loss = 0.0 val_loss = 0.0 semi_loss = 0.0 #初始化本轮训练、验证、半监督正确样本数 train_acc = 0.0 val_acc = 0.0 semi_acc = 0.0 start_time = time.time() #记录epoch开始时间，统计耗时 #开启训练模式，进入训练阶段 model.train() for batch_x, batch_y in train_loader: #遍历有标签训练集dataloader的每一个batch x, target = batch_x.to(device), batch_y.to(device) pred = model(x) #模型前向传播得到预测值 train_bat_loss = loss(pred, target) #计算当前batch的损失 train_bat_loss.backward() #反向传播，计算网络参数梯度 optimizer.step() #优化器更新权重参数 optimizer.zero_grad() #梯度清零，防止下轮次累积 train_loss += train_bat_loss.cpu().item()#累加当前batch损失到总损失。 train_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy()) #如果预测值类别最大可能的值为实际目标分类，则累计正确样本数 plt_train_loss.append(train_loss / train_loader.__len__())#所有批次平均loss plt_train_acc.append(train_acc / train_loader.dataset.__len__()) #所有数据准确率 #半监督训练阶段：将伪标签数据集加入训练 if semi_loader != None: #判断是否存在有效伪标签 for batch_x, batch_y in semi_loader: #分批遍历伪标签数据集 x, target = batch_x.to(device), batch_y.to(device) pred = model(x) semi_bat_loss = loss(pred, target) semi_bat_loss.backward() optimizer.step() optimizer.zero_grad() semi_loss += semi_bat_loss.cpu().item() semi_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy()) print("半监督数据集的训练准确率为", semi_acc / train_loader.dataset.__len__()) #开启验证模式，进入验证阶段： model.evel() with torch.no_grad(): for batch_x, batch_y in val_loader: x, target = batch_x.to(device), batch_y.to(device) pred = model(x) val_bat_loss = loss(pred, target) val_loss += val_bat_loss.cpu().item() val_acc += np.sum(np.argmax(pred.detach().cpu().numpy(), axis=1) == target.cpu().numpy()) plt_val_loss.append(val_loss / val_loader.dataset.__len__()) plt_val_acc.append(val_acc / val_loader.dataset.__len__()) #伪标签更新逻辑+最优模型保存 if epoch % 3 == 0 and plt_val_acc[-1] > 0.6: #每3个epoch更新一次伪标签数据集 #早期模型效果差，等验证集准确率大于0.6才生成伪标签 semi_loader = get_semi_loader(no_lable_loader, model, device, thres) #调用函数生成新的半监督loader if val_acc > max_acc: #保存当前最优模型，更新最优准确率 torch.save(model, save_path) max_acc = val_acc #训练结束，绘制loss曲线 plt.plot(plt_train_loss) plt.plot(plt_val_loss) plt.title("loss") plt.legend(["train", "val"]) plt.show() # 绘制accuracy曲线 plt.plot(plt_train_acc) plt.plot(plt_val_acc) plt.title("acc") plt.legend(["train", "val"]) plt.show()半监督核心模块semiDataset
伪标签数据集类，继承Dataset；用模型预测无标签数据，筛选高置信样本生成训练数据集
class semiDataset(Dataset): def __init__(self, no_lable_loader, model, device, thres=0.99): x, y = self.get_lable(no_lable_loader, model, model, device, thres) #调用函数，推理生成符合条件的图片及其对应伪标签 if x == []: #flag标记数据集无效，不参与训练 self.flag = False else: self.flag = True self.X = np.array(x)#将原图列表转为numpy数组 self.Y = torch.LongTensor(y) #将为标签转为LongTensor,适合交叉熵计算 self.transform = train_transform #伪标签样本训练使用训练集随机增强 def __getitem__(self, item): return self.transform(self.X[item]), self.Y[item] def __len__(self): return len(self.X) def get_lable(self, no_lable_loader, model, deice, thres): model = model.to(device) pred_prob = [] #保存每个样本预测最大置信度 lables = [] #保存每个样本的伪标签 x = [] #保存筛选通过的原始图片numpy数组 y = [] #保存筛选通过样本对应的伪标签 soft = nn.Softmax() #将输出各类别分数转为概率 with torch.no_grad(): # 遍历无标签dataloader；bat_x增强后的tensor图片；_忽略第二个返回值原始图片 for bat_x, _ in no_lable_loader: bat_x = bat_x.to(device) pred = model(bat_x)#前向传播计算各类别预测得分 pred_soft = soft(pred) #将得分转为各类别概率 pred_max, pred_value = pred_soft.max(1) #取1类别维度最大值，记录该最大置信度，以及预测类别编号 pred_prob.extend(pred_max.cpu().numpy().tolist()) labels.extend(pred_value.cpu().numpy().tolist()) # GPU张量→cpu→numpy数组→python列表；extend追加到总列表 for index, prob in enumerate(pred_prob): if prob > thres: #将符合条件的图片及对应标签存入x,y x.append(no_lable_loader.dataset[index][1] y.append(lables[index]) return x, y def get_semi_loader(no_lable_loader, model, device, thres): semiset = semiDataset(no_label_loader, model, device, thres) #实例化伪标签数据集 if semiset.flag == False: return None #没有可信伪标签样本 else: semi_loader = DataLoader(semiset, batch_size=16, shuffle=False) return semi_loader
