💡 为什么要谈“项目化”?
在平时的学习阶段,我们接触的通常是玩具数据集,比如 MNIST 手写数字识别。这些数据是干
净的,代码写在一个 .ipynb 文件里,跑通了看到准确率就结束了。但是现实中的数据是肮脏、缺
失且不均衡的。项目化要求你考虑如果数据格式变了,你的代码需要重写吗?
很多初学者通过 model.fit() 或直接调用预训练模型能得到不错的结果,但这是一种“虚假的高
手感”。
项目化要求你理解
(1)如何标准化配置管理(YAML/Argparse)?
(2)如何实现模型权重的自动保存与断点续训?
(3)如何记录每一场实验的超参数(Experiment Tracking)?
真正的深度学习实战,只有 20% 的时间在写模型架构,剩下的 80% 都在处理数据流和
构建工程闭环。
如果你写的代码只有你自己能看懂,那它永远无法变成一个产品。当你把模型训练逻辑和推理
逻辑分离后,后端工程师才能轻松地把你的模型封装成 API 接口,交付给用户使用。
💡 如何实现一个简单的项目化?
一个项目化的文件一般包含下面这些内容:
🌻 项目模块划分与功能概述
📂 configs文件夹
Alexnet.yaml
这个结构模仿了 YOLOv5 的配置风格,清晰地划分了数据、模型和训练的超参数。
📂 models文件夹
网络结构文件(model文件)
models/vit.py,定义Alexnet的全部组件和前向传播逻辑。
models/common.py ,里面可以放一些通用的模块
📂utils文件夹
utils/datasets.py 数据集加载
utils/general.py 通用工具函数
utils/metrics.py 评估指标
utils/torch_utils.py PyTorch工具
📂runs文件夹
train/ 存放训练结果
val/ 存放验证结果
📂weights 预训练权重
🚀train.py 网络训练
🔍 val.py 加载训练好的权重,对单个图像或整个测试集进行推理和评估。
🔍 predict.py 预测脚本
代码细节
utils/general.py 通用工具函数
💻 load_yaml(path) 函数的作用
这个函数的作用就是读取一份训练项目的“说明书”或“配置清单”,并将其转换成 Python 能够操作的
字典 (Dictionary) 格式。
🚀 代码调用、输入与输出示例
为了演示这段代码的输入和输出,我们需要假设存在一个 YAML 配置文件,并需要导入 yaml 库。
import yaml import os from pathlib import Path # --- 待解释的函数 --- def load_yaml(path): """加载YAML配置""" # 注意:在实际运行中,需要确保 'yaml' 库已安装 (pip install pyyaml) with open(path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) # --- 演示调用 --- # 1. 创建一个模拟的 YAML 文件(以便代码可以运行) yaml_content = """ project_name: flower_classification epochs: 50 learning_rate: 0.0001 data: train_path: 'data/train' val_path: 'data/val' """ config_path = 'temp_config.yaml' Path(config_path).write_text(yaml_content, encoding='utf-8') # 2. 调用函数 # 输入 (Input): 配置文件路径 print(f"输入路径: {config_path}") config_data = load_yaml(config_path) # 3. 查看输出 # 输出 (Output): 函数返回的 Python 字典 print("\n--- 函数输出 (Python 字典) ---") print(config_data) print("\n--- 输出类型 ---") print(type(config_data)) # 4. 演示如何使用输出数据 print("\n--- 访问输出数据 ---") print(f"项目名称: {config_data['project_name']}") print(f"训练轮数 (Epochs): {config_data['epochs']}") print(f"训练集路径: {config_data['data']['train_path']}") # 清理模拟文件 #os.remove(config_path)
💾 save_checkpoint(...) 函数的作用
这个函数的作用是为您的模型训练进度拍摄“快照” (Snapshot),并确保您能够找到性能最好的那
个版本。简单来说如何不是最好的权重,就使用普通的命名,并把训练的相关信息保存成一个字典。
import torch from pathlib import Path import os # --- 待解释的函数 --- def save_checkpoint(state, path, is_best=False): """保存检查点""" # 确保目录存在 Path(path).parent.mkdir(parents=True, exist_ok=True) # 保存当前的检查点 torch.save(state, path) if is_best: best_path = Path(path).parent / 'best.pt' torch.save(state, best_path) print(f'✅ Best model saved: {best_path}') # --- 函数结束 --- # --- 演示调用 --- # 1. 模拟要保存的状态 (State) mock_state = { 'epoch': 5, 'model': {'weight_01': torch.tensor([1.2, 0.8])}, 'optimizer': 'Adam_state_info' } # 2. 设置保存路径 save_dir = 'mock_runs/train_test' save_path_current = os.path.join(save_dir, 'ckpt_005.pt') best_path_expected = os.path.join(save_dir, 'best.pt') # 3. 第一次调用:保存普通快照 print("--- 第一次调用:普通保存 ---") save_checkpoint(mock_state, save_path_current, is_best=False) print(f"输入状态: epoch {mock_state['epoch']}") print(f"输出文件: {save_path_current}") print("==================================\n") # 4. 第二次调用:保存最佳快照 print("--- 第二次调用:保存最佳模型 ---") # 假设这是第 10 轮,并且是目前的最佳成绩 mock_state['epoch'] = 10 save_checkpoint(mock_state, os.path.join(save_dir, 'ckpt_010.pt'), is_best=True) print(f"输入状态: epoch {mock_state['epoch']}") print(f"输出文件: {os.path.join(save_dir, 'ckpt_010.pt')} 和 {best_path_expected}") # 5. 清理模拟文件 # import shutil # shutil.rmtree(save_dir) # 在实际测试后,可以使用此行清理创建的文件夹
state(模型 / 优化器等状态字典)最终会以 PyTorch 二进制文件(.pt) 的形式保存在你指定的路
径,如果像查看如果想查看文件内容,可以用 torch.load() 加载验证,PyTorch 生成的 .pt 文件中
(比如 best.pt),核心是你传入的 state 字典里的所有内容,但除此之外,文件还包含 PyTorch
自动添加的元信息、序列化格式信息,以及字典中各类数据本身的底层存储信息。
# 加载最佳模型 best_state = torch.load('mock_runs/train_test/best.pt') print(best_state['epoch']) # 输出 10,验证保存正确
🔄 load_checkpoint(...) 函数的作用(大白话解释)
【0这个函数的作用就是让一个中断的训练任务能够从上次停止的地方无缝接续,或者让一个训练
好的模型准备好进行预测。】
🚀 代码调用、输入与输出示例
为了演示其作用,我们假设您已经保存了一个检查点文件 (best.pt),现在要加载它来继续训练。
ckpt = torch.load(path, map_location='cpu')从指定 path 加载保存的序列化文件(通常是 .pth
或 .ckpt 格式),返回一个字典(checkpoint 字典)。ckpt:接收加载后的字典,该字典通常包含
训练过程中保存的关键信息,如 model(模型权重)、optimizer(优化器状态)、epoch(训练轮
次)、loss(损失值)等。
model.load_state_dict(ckpt['model']),将预训练的权重参数加载到模型实例中。ckpt['model']:
检查点字典中键为 model 的值,对应模型的 state_dict(状态字典),包含模型所有可学习参数
(如卷积层权重、全连接层偏置等)。
让当前的 model 实例拥有检查点中保存的权重,实现模型参数恢复(比如恢复到上次训练中断时
的状态,或加载预训练权重)。
如果判断检查点字典中有 optimizer 键,优化器的 load_state_dict 方法,恢复优化器的状态。
ckpt.get('epoch', 0):字典的 get 方法,安全获取 epoch 键的值: 若 ckpt 中存在 epoch 键,返
回对应的值(如 100,表示上次训练到第 100 轮); 若不存在 epoch 键,返回默认值 0。返回上
次训练中断的轮次,方便后续训练从该轮次继续。
import torch import torch.nn as nn import torch.optim as optim from pathlib import Path import os # 模拟一个简单的模型定义 class SimpleModel(nn.Module): def __init__(self): super().__init__() self.fc = nn.Linear(10, 1) def forward(self, x): return self.fc(x) # --- 待解释的函数 (假设已定义) --- def load_checkpoint(path, model, optimizer=None): """加载检查点""" ckpt = torch.load(path, map_location='cpu') model.load_state_dict(ckpt['model']) if optimizer and 'optimizer' in ckpt: optimizer.load_state_dict(ckpt['optimizer']) return ckpt.get('epoch', 0) # --- 函数结束 --- # --- 演示调用 --- # 1. 模拟一个检查点文件 (通常是 save_checkpoint 函数创建的) model_to_save = SimpleModel() optimizer_to_save = optim.Adam(model_to_save.parameters(), lr=0.001) mock_ckpt = { 'epoch': 50, 'model': model_to_save.state_dict(), 'optimizer': optimizer_to_save.state_dict() } ckpt_path = 'mock_runs/best.pt' Path(ckpt_path).parent.mkdir(parents=True, exist_ok=True) torch.save(mock_ckpt, ckpt_path) # 2. 初始化新的模型和优化器 (它们需要被加载) new_model = SimpleModel() new_optimizer = optim.Adam(new_model.parameters(), lr=0.01) # 注意:初始LR不同 print("--- 加载前状态 ---") # 打印新的模型和优化器的初始状态 (它们是随机的/初始化的) print(f"新模型参数的初始值 (部分): {list(new_model.parameters())[0].sum().item():.4f}") print(f"新优化器的初始LR: {new_optimizer.param_groups[0]['lr']:.4f}") print("==================================\n") # 3. 调用函数加载检查点 start_epoch = load_checkpoint(ckpt_path, new_model, new_optimizer) # 4. 查看加载后状态 (输出) print("--- 加载后状态 (函数输出) ---") # 模型的参数已经被检查点中的值覆盖 print(f"加载后模型参数的总和 (应与保存前的总和一致): {list(new_model.parameters())[0].sum().item():.4f}") # 优化器的LR已经被检查点中的值覆盖 (如果检查点中包含LR信息的话) print(f"加载后的优化器LR: {new_optimizer.param_groups[0]['lr']:.4f}") print(f"函数返回的下一轮起始 Epoch: {start_epoch}") # 5. 清理模拟文件 # os.remove(ckpt_path) # import shutil # shutil.rmtree('mock_runs')
--- 加载前状态 --- 新模型参数的初始值 (部分): 0.5234 # 随机值 新优化器的初始LR: 0.0100 ================================== --- 加载后状态 (函数输出) --- 加载后模型参数的总和 (应与保存前的总和一致): 0.3547 # 已经被检查点中的值覆盖 加载后的优化器LR: 0.0010 # 已经被检查点中的值覆盖 函数返回的下一轮起始 Epoch: 50
utils/metrics.py 评估指标
📊 Metrics 类的作用(大白话解释)
这个类的作用就是充当一个计分板。它在每一轮比赛(批次)中记录得分(预测和标签),直到整
个 Epoch结束,然后给出比赛的最终得分(各种指标)。
class Metrics: """指标计算器""" def __init__(self): self.reset() def reset(self): self.preds = [] self.labels = [] def update(self, pred, label): self.preds.extend(pred.cpu().numpy()) self.labels.extend(label.cpu().numpy()) def compute(self): acc = accuracy_score(self.labels, self.preds) p, r, f1, _ = precision_recall_fscore_support( self.labels, self.preds, average='macro', zero_division=0 ) return {'acc': acc, 'precision': p, 'recall': r, 'f1': f1}
(1)__init__是类的构造方法,创建Metrics实例时会自动执行。
构造方法中直接调用self.reset():初始化时就完成指标数据的 “清空”,避免初始状态下出现未
定义的属性。
(2)重置方法(reset)
清空存储预测值和真实标签的列表,为新一轮指标计算做准备(比如每个 epoch 开始时重置
指标)。self.preds:实例属性,列表类型,用于累计所有批次的模型预测值。self.labels:实例属
性,列表类型,用于累计所有批次的真实标签值。每次调用reset(),两个列表都会被重新初始化为
空列表,避免不同轮次(如不同 epoch)的数据相互污染。
(3)数据更新方法(update)
将单个批次的预测值和真实标签添加到累计列表中,是批量训练中最核心的 “数据收集” 步骤。
pred:输入的单个批次预测值,通常是 PyTorch 的Tensor类型(GPU/CPU 张量)。
label:输入的单个批次真实标签,同样为 PyTorch 的Tensor类型。
(4)指标计算方法(compute)
基于累计的所有预测值和真实标签,计算分类任务的核心评估指标,并以字典形式返回结果。
🚀 示例:在训练循环中的使用
每个epoch开始的时候,调用重置方法(reset),一个epoch中的不同的批次调用数据更新方法(update),保存单个批次的信息。一个epoch结束后调用计算函数,计算一个批次后的指标信息。
from sklearn.metrics import accuracy_score, precision_recall_fscore_support import torch # ... Metrics 类的定义放在这里 ... # 初始化指标计算器 train_metrics = Metrics() # 模拟一个 Epoch 的训练循环 for epoch in range(num_epochs): train_metrics.reset() # ❶ 每个 Epoch 开始时重置计分板 # 模拟批次循环 for batch_id, (data, label) in enumerate(train_loader): # ... (模型前向传播、损失计算、反向传播) ... # 假设 model_outputs 是模型的输出 logits pred_labels = torch.argmax(model_outputs, dim=1) # ❷ 在每个批次结束后更新记录 train_metrics.update(pred_labels, label) # ❸ 在 Epoch 结束后计算指标 results = train_metrics.compute() print(f"Epoch {epoch+1} 训练准确率: {results['acc']:.4f}")
📏 AverageMeter 类的作用(大白话解释)
这个类的作用就是充当一个“在线平均值计算器”,它能够高效地在每次迭代(如每个训练批次)中
更新数值,并随时提供当前的平均值**,而不需要等到所有数据收集完毕。
在深度学习训练中,它最常用于计算:每个 Epoch的平均损失 (Average Loss) 或 每个 Epoch 的平
均准确率 (Average Accuracy)。
其中val是最新一次更新的单次值,第 5 个批次的损失值(如 2.3)。avg累计的平均值,前 5 个批
次的平均损失。sum累计的总和,前 5 个批次的损失总和。count累计的样本数 ,已
样本总数。
class AverageMeter: """平均值计算器""" def __init__(self): self.reset() def reset(self): self.val = 0 self.avg = 0 self.sum = 0 self.count = 0 def update(self, val, n=1): self.val = val self.sum += val * n self.count += n self.avg = self.sum / self.count # 1. 实例化平均值计算器(用于统计训练损失) loss_meter = AverageMeter() # 2. 模拟训练过程:遍历5个批次 for batch_idx in range(5): # 模拟每个批次的损失值和样本数 batch_loss = 2.0 + batch_idx * 0.1 # 批次损失:2.0, 2.1, 2.2, 2.3, 2.4 batch_size = 32 # 每个批次32个样本 # 3. 更新损失计算器(传入批次损失 + 批次样本数) loss_meter.update(batch_loss, n=batch_size) # 4. 实时打印当前状态 print(f"批次{batch_idx+1}:") print(f" 本次损失值:{loss_meter.val:.4f}") print(f" 累计平均损失:{loss_meter.avg:.4f}") print(f" 累计样本数:{loss_meter.count}") print("-"*20) # 5. 一轮训练结束后,查看最终平均损失 print(f"本轮训练平均损失:{loss_meter.avg:.4f}") # 6. 下一轮训练前重置计算器 loss_meter.reset() print(f"重置后平均值:{loss_meter.avg}") # 输出:0
🚀 示例:在训练循环中的使用
计算每一个批次的平均损失的。
# 初始化损失平均值计算器 loss_meter = AverageMeter() # 模拟一个 Epoch 的训练循环 for epoch in range(num_epochs): loss_meter.reset() # ❶ 每个 Epoch 开始时清零 for batch_id, (data, label) in enumerate(train_loader): # ... (模型前向传播,计算损失) ... current_loss = loss_function(outputs, label).item() # 假设损失是 0.15 batch_size = data.size(0) # 假设 Batch Size 是 32 # ❷ 在每个批次结束后更新损失计算器 loss_meter.update(current_loss, batch_size) # 随时可以查看当前状态 # print(f"Batch {batch_id}: 当前损失={loss_meter.val:.4f}, 累计平均损失={loss_meter.avg:.4f}") # ❸ 在 Epoch 结束时,输出最终的平均损失 print(f"Epoch {epoch+1} 最终平均损失: {loss_meter.avg:.4f}")
utils/torch_utils.py PyTorch工具
"""PyTorch工具""" import torch import torch.nn as nn def select_device(device=''): """选择设备""" if device and 'cuda' in device and torch.cuda.is_available(): return torch.device(device) return torch.device('cpu') def get_optimizer(model, name='Adam', lr=0.001, weight_decay=0.0001): """创建优化器""" if name == 'SGD': return torch.optim.SGD(model.parameters(), lr, momentum=0.9, weight_decay=weight_decay) elif name == 'Adam': return torch.optim.Adam(model.parameters(), lr, weight_decay=weight_decay) else: raise ValueError(f'Unknown optimizer: {name}') def get_scheduler(optimizer, name='CosineAnnealing', epochs=100, step_size=10, gamma=0.1): """创建学习率调度器""" if name == 'StepLR': return torch.optim.lr_scheduler.StepLR(optimizer, step_size, gamma) elif name == 'CosineAnnealing': return torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, epochs) return None
主文件夹
parser = argparse.ArgumentParser()
🚀 代码作用(大白话解释)
这段代码的作用就像是程序的“启动器”和“控制台”。它让您可以在不修改代码的情况下,通过在命
令行中输入不同的指令来控制程序的行为,例如告诉程序使用哪个配置文件、在哪块硬件上运行等
等。
| 对应的代码 | parser = argparse.ArgumentParser() |
| 大白话: | “我准备好接收你的指令了。” |
| 作用: | 创建了一个参数解析器对象 (parser),这个对象专门负责识别和处理您在命令行中输入的各种参数和选项。 |
| 参数定义 | 对应的作用 | 大白话解释 |
--cfg |
default='configs/flower.yaml', help='config file' |
配置文件指令。 允许您指定要使用哪个 YAML 配置文件来运行训练。默认使用 configs/flower.yaml。 |
--device |
default='', help='cuda device, i.e. 0 or cpu' |
硬件选择指令。 允许您指定程序应该在哪个硬件上运行,例如 cuda:0(使用 GPU)或 cpu(使用中央处理器)。 |
--save-dir |
default='', help='save directory' |
结果保存指令。 允许您指定训练日志、模型权重等结果应该保存到哪个文件夹。 |
| 对应的代码 | opt = parser.parse_args() |
| 大白话: | “把所有指令都收集起来。” |
| 作用: | 当您在命令行中按下 Enter 键后,这个方法就会运行:它读取您输入的所有参数(例如 --device cuda:0),并将它们打包成一个名为 opt 的对象。 |
| 对应的代码 | main(opt) |
| 大白话: | “开始执行程序主体。” |
| 作用: | 调用程序的主函数 (main),并将所有收集到的配置和指令 (opt) 传递给它。主函数随后会根据 opt 中包含的参数来运行训练或预测逻辑。 |
💡 命令行使用示例
假设这段代码保存为 train.py,您可以在命令行中这样启动程序
# 运行示例:使用CPU,并将结果保存到 my_exp/ python train.py --device cpu --save-dir my_exp/ # 运行示例:使用 GPU 0,并指定一个不同的配置文件 python train.py --cfg configs/my_custom.yaml --device cuda:0
train.py
"""训练脚本""" import argparse import torch import torch.nn as nn from pathlib import Path from tqdm import tqdm from models.AlexNet_model import AlexNet from utils.datasets import create_dataloader from utils.general import load_yaml, save_checkpoint from utils.metrics import Metrics, AverageMeter from utils.torch_utils import select_device, get_optimizer, get_scheduler def train_epoch(model, loader, criterion, optimizer, device): """训练一个epoch""" model.train() loss_meter = AverageMeter() metrics = Metrics() pbar = tqdm(loader, desc='Training') for imgs, labels in pbar: imgs, labels = imgs.to(device), labels.to(device) optimizer.zero_grad() outputs = model(imgs) loss = criterion(outputs, labels) loss.backward() optimizer.step() loss_meter.update(loss.item(), imgs.size(0)) metrics.update(outputs.argmax(1), labels) pbar.set_postfix({'loss': f'{loss_meter.avg:.4f}'}) results = metrics.compute() results['loss'] = loss_meter.avg return results def validate(model, loader, criterion, device): """验证""" model.eval() loss_meter = AverageMeter() metrics = Metrics() with torch.no_grad(): for imgs, labels in tqdm(loader, desc='Validating'): imgs, labels = imgs.to(device), labels.to(device) outputs = model(imgs) loss = criterion(outputs, labels) loss_meter.update(loss.item(), imgs.size(0)) metrics.update(outputs.argmax(1), labels) results = metrics.compute() results['loss'] = loss_meter.avg return results def main(opt): # 加载配置 cfg = load_yaml(opt.cfg) device = select_device(opt.device or cfg['device']) save_dir = Path(opt.save_dir or cfg['save_dir']) save_dir.mkdir(parents=True, exist_ok=True) print(f'🌻 Training on {device}') # 数据加载 train_loader, train_set = create_dataloader( cfg['train'], cfg['img_size'], cfg['batch_size'], True, cfg['workers'], augment=True ) val_loader, val_set = create_dataloader( cfg['val'], cfg['img_size'], cfg['batch_size'], False, cfg['workers'], augment=False ) print(f'Train: {len(train_set)}, Val: {len(val_set)}') # 模型 model = AlexNet(cfg['nc'], cfg['dropout']).to(device) criterion = nn.CrossEntropyLoss(label_smoothing=0.1) optimizer = get_optimizer(model, cfg['optimizer'], cfg['lr0'], cfg['weight_decay']) scheduler = get_scheduler(optimizer, cfg['scheduler'], cfg['epochs'], cfg.get('step_size', 10), cfg.get('gamma', 0.1)) # 训练 best_acc = 0 for epoch in range(1, cfg['epochs'] + 1): print(f'\n📊 Epoch {epoch}/{cfg["epochs"]}') train_results = train_epoch(model, train_loader, criterion, optimizer, device) val_results = validate(model, val_loader, criterion, device) if scheduler: scheduler.step() print(f'Train - Loss: {train_results["loss"]:.4f}, Acc: {train_results["acc"]:.4f}') print(f'Val - Loss: {val_results["loss"]:.4f}, Acc: {val_results["acc"]:.4f}') # 保存 is_best = val_results['acc'] > best_acc if is_best: best_acc = val_results['acc'] save_checkpoint({ 'epoch': epoch, 'model': model.state_dict(), 'optimizer': optimizer.state_dict(), 'best_acc': best_acc }, save_dir / 'last.pt', is_best) print(f'\n🎉 Training complete! Best Acc: {best_acc:.4f}') if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--cfg', type=str, default='configs/flower.yaml', help='config file') parser.add_argument('--device', type=str, default='', help='cuda device, i.e. 0 or cpu') parser.add_argument('--save-dir', type=str, default='', help='save directory') opt = parser.parse_args() main(opt)