NLP项目实战之商品分类项目
·
商品分类项目（大模型工程师入门必读）
写给谁：想成为大模型开发工程师的零基础学习者
目标：读完这个文档，你能完全理解一个完整的深度学习 NLP 项目
目录
- 第一章 项目全景
- 第二章 核心概念预备知识
- 第三章 逐行代码详解 - main.py
- 第四章 逐行代码详解 - config.py
- 第五章 逐行代码详解 - preprocess.py
- 第六章 逐行代码详解 - dataset.py
- 第七章 逐行代码详解 - train.py（核心）
- 第八章 逐行代码详解 - predict.py
- 第九章 逐行代码详解 - evaluate.py
- 第十章 逐行代码详解 - Web服务三件套
- 第十一章 完整流程串讲与数据流图
- 第十二章 大模型工程师知识拓展
- 第十三章 常见问题
- 附录
第一章 项目全景
1.1 这个项目到底干什么？
一句话：让电脑自动判断商品属于什么类别。
举个例子，你给电脑输入：
输入：好奇心钻装纸尿裤L40片9-14kg
输出：母婴
电脑就能告诉你这个商品属于"母婴"类别。
这在工业界叫什么？ → 文本分类（Text Classification），是 NLP（自然语言处理）最基础、最经典的任务。
为什么这对大模型工程师重要？ 因为文本分类是所有 NLP 任务的入门砖。你学会了这个，后面学命名实体识别、文本生成、对话系统等任务都会更容易。
1.2 项目文件结构
product_classification/
│
├── data/ ← 数据
│ ├── raw/ ← 原始数据（人能看懂的文字）
│ │ ├── train.txt ← 训练数据（给AI学习用的题目）
│ │ ├── valid.txt ← 验证数据（学习过程中的模拟考）
│ │ └── test.txt ← 测试数据（最终的高考）
│ └── processed/ ← 处理后的数据（AI能看懂的数字）
│
├── pretrained/ ← 预训练的BERT模型（已经学好中文的AI大脑）
│ ├── config.json ← 模型结构配置（大脑有多少神经元）
│ ├── model.safetensors ← 模型权重文件（大脑的记忆，392MB）
│ ├── tokenizer.json ← 分词器（把文字切成词的工具）
│ ├── tokenizer_config.json ← 分词器配置
│ └── vocab.txt ← 词表（AI认识的所有字，共21128个）
│
├── checkpoints/ ← 训练好的模型（学完后的AI）
│ ├── best/ ← 最佳模型
│ ├── last/ ← 最新检查点（训练中断后可以继续）
│ │ └── checkpoint.pt
│ └── labels.txt ← 类别标签列表（30个类别）
│
├── logs/ ← TensorBoard训练日志
│
├── src/ ← 源代码（你的学习重点！）
│ ├── main.py ← 程序入口（28行）
│ ├── configuration/
│ │ ├── __init__.py ← 空文件（标记这是Python包）
│ │ └── config.py ← 配置文件（26行）
│ ├── process/
│ │ ├── __init__.py ← 空文件
│ │ ├── preprocess.py ← 数据预处理（43行）
│ │ └── dataset.py ← 数据加载（43行）
│ ├── runner/
│ │ ├── __init__.py ← 空文件
│ │ ├── train.py ← 训练（273行，最核心！）
│ │ ├── predict.py ← 预测（62行）
│ │ └── evaluate.py ← 评估（51行）
│ └── web/
│ ├── __init__.py ← 空文件
│ ├── app.py ← Web应用（29行）
│ ├── schema.py ← 数据模型（4行）
│ └── service.py ← 服务层（7行）
│
└── test/
└── fast_api_test.py ← FastAPI学习示例（可忽略）
1.3 数据长什么样？
data/raw/train.txt 的内容（TSV格式，Tab键分隔）：
label text_a
母婴 好奇心钻装纸尿裤L40片9-14kg
蔬菜 基地玉米.
酒饮冲调 240ML*15养元2430六个核桃
玩具 76-022斯伯丁篮球
香烟 红塔山软13mg经典195
label：商品类别（这是AI要学习的"答案"）text_a：商品标题（这是AI要看的"题目"）- 共 30 个类别：3C数码、个护、书籍、乳品、休闲食品、健康、健康食品、办公、宠物、家居、家电、服饰内衣、母婴、水产、水果、汽车用品、清洁、玩具、礼品、粮油速食、美妆、肉禽蛋、蔬菜、运动、酒饮冲调、钟表配饰、鞋靴箱包、餐饮、香烟、鲜花绿植
1.4 项目运行命令
# 在项目根目录下执行（以 src 为工作目录）
# 第1步：数据预处理（把文字变成数字）
python src/main.py preprocess
# 第2步：训练模型（教AI学习）
python src/main.py train
# 第3步：评估模型（给AI考试打分）
python src/main.py evaluate
# 第4步：预测（让AI做题）
python src/main.py predict
# 第5步：启动Web服务（给别人用）
python src/main.py run_app
第二章 核心概念预备知识
在看代码之前，你必须先理解这些概念。我用最通俗的语言解释。
2.1 什么是深度学习？
传统编程：程序员写规则 → 电脑按规则执行
深度学习：程序员给数据 → 电脑自己学规则
类比：
- 传统编程 = 你告诉小孩"有四条腿、会汪汪叫的是狗"
- 深度学习 = 你给小孩看1000张狗的照片和1000张猫的照片，他自己学会分辨
2.2 什么是 BERT？
BERT = 一个已经学过中文的AI大脑（预训练模型）
它的"简历"：
- 全名：Bidirectional Encoder Representations from Transformers（基于Transformer的双向编码器表示）
- 发明者：Google，2018年
- 训练数据：大量中文文本（维基百科、新闻、书籍等）
- 能力：理解中文的词义、语法、上下文
本项目的BERT配置（来自 pretrained/config.json）：
| 配置项 | 值 | 通俗解释 |
|---|---|---|
hidden_size |
768 | 神经元的"宽度"，每个词用768个数字来表示 |
num_hidden_layers |
12 | 有12层，越深理解能力越强 |
num_attention_heads |
12 | 12个"注意力头"，可以同时关注不同方面 |
vocab_size |
21128 | 认识21128个字和词 |
max_position_embeddings |
512 | 最多处理512个字 |
2.3 什么是微调（Fine-tuning）？
预训练（Pre-training）：在大量无标注文本上学习语言（已经做好了，模型在pretrained/里）
↓
微调（Fine-tuning）：在特定任务数据上调整参数（我们要做的）
↓
得到专用模型：能做商品分类的AI
类比：
- 预训练 = 一个人上了大学，学了语文、数学、英语
- 微调 = 他毕业后去公司上班，学习具体业务
2.4 什么是分词（Tokenization）？
原始文本："好奇心钻装纸尿裤"
↓ 分词器
词元序列：["好", "奇", "心", "钻", "装", "纸", "尿", "裤"]
↓ 查词表
数字序列：[1962, 1906, 2552, 7032, 6814, 4989, 2399, 6841]
↓ 加特殊标记
最终输入：[101, 1962, 1906, 2552, 7032, 6814, 4989, 2399, 6841, 102]
[CLS] [SEP]
101=[CLS]标记，BERT用它来表示"整个句子的意思"102=[SEP]标记，表示句子结束
为什么要分词？ 因为AI不认字，只认数字。分词器就是"翻译官"。
2.5 训练的核心循环
┌─────────────────────────────────────────────────────────┐
│ 一个训练步骤 │
│ │
│ 输入数据 → 前向传播 → 得到预测 → 计算损失 │
│ ↓ │
│ 更新参数 ← 优化器更新 ← 反向传播 ← 计算梯度 │
│ ↓ │
│ 重复以上过程，直到损失足够小 │
└─────────────────────────────────────────────────────────┘
通俗解释：
- 前向传播：把题目给AI做，得到答案
- 计算损失：把AI的答案和正确答案比较，算出错了多少
- 反向传播：告诉AI哪里错了，错得有多离谱（梯度）
- 更新参数：AI根据反馈调整自己的"思维方式"（参数）
2.6 关键术语表
| 术语 | 通俗解释 | 为什么重要 |
|---|---|---|
| Epoch | 把所有数据从头到尾看一遍 | 控制训练轮数 |
| Batch | 一次看几条数据 | 内存有限，不能一次全看 |
| Batch Size | 一个batch有多少条数据 | 影响训练速度和效果 |
| Learning Rate | 每次调整参数的幅度 | 太大震荡不收敛，太小学得慢 |
| Loss | AI答错的程度 | 越小越好，训练的目标就是让loss变小 |
| Gradient | 损失对参数的导数 | 告诉参数该往哪个方向调 |
| Optimizer | 优化器（如Adam） | 决定怎么根据梯度更新参数 |
| Backward | 反向传播 | 计算梯度的过程 |
| Forward | 前向传播 | 从输入得到输出的过程 |
| State_dict | 模型参数字典 | 保存/加载模型用的 |
| Checkpoint | 检查点/存档 | 训练中断后可以从这里继续 |
第三章 逐行代码详解 - main.py
文件：src/main.py（28行）
作用：程序的入口，根据命令行参数决定执行哪个功能。
# from sys import argv
#开头的是注释，Python不会执行它- 这行被注释掉了，说明作者之前用
sys.argv来获取命令行参数，后来换成了argparse sys.argv是一种简单但不优雅的方式（需要手动处理错误）- 拓展：在实际工作中，注释掉但不删除的代码很常见，方便以后回溯
from argparse import ArgumentParser
from ... import ...：从某个模块导入某个东西argparse：Python标准库，专门用来解析命令行参数ArgumentParser：参数解析器类- 标准库 = Python自带的库，不需要
pip install - 拓展：
argparse会自动生成--help帮助信息，比sys.argv专业得多
if __name__ == '__main__':
- 这是Python的"入口守卫"
__name__是Python的一个内置变量：- 直接运行这个文件时，
__name__的值是'__main__' - 被别人
import时，__name__的值是文件名（如'main'）
- 直接运行这个文件时，
- 为什么要写这个？ 防止被导入时意外执行代码
- 拓展：每个Python项目的入口文件都应该有这行，这是Python编程的基本规范
# 获取命令行传入的参数，作为脚本名称
# arg = argv[1]
- 这两行是注释，记录了之前的实现方式
argv[1]表示取命令行的第2个参数（第1个是脚本名本身）
parser = ArgumentParser(usage='usage: main.py action')
- 创建一个参数解析器对象
usage='usage: main.py action'：用户输入--help时显示的用法提示- 缩进：Python用缩进来表示代码块，这里缩进4个空格，表示在
if语句内部
parser.add_argument('action', choices=['preprocess', 'train', 'predict', 'evaluate', 'run_app'])
-
添加一个名为
action的位置参数（必须填的参数） -
choices=[...]：限制只能选这几个值之一 -
如果用户输入
python main.py hello，会报错：error: argument action: invalid choice: 'hello' -
拓展：位置参数 vs 可选参数
- 位置参数：
python main.py train（必须填） - 可选参数：
python main.py --epochs 100（可以不填，有默认值）
- 位置参数：
# 解析出当前要执行的操作
action = parser.parse_args().action
parser.parse_args()：解析命令行输入，返回一个 Namespace 对象.action：取出其中action属性的值- 例如：输入
python main.py train，那么action = 'train' - 链式调用：
parser.parse_args().action是两步操作写在一起
match action:
- Python 3.10 新增的模式匹配语法
- 类似其他语言的
switch-case - 根据
action的值，跳转到对应的case分支 - 拓展：如果你的Python版本低于3.10，可以用
if-elif-else代替
case 'preprocess':
from process.preprocess import preprocess
preprocess()
- 当
action == 'preprocess'时执行 - 延迟导入（lazy import）：只在需要时才
import- 好处1：启动快（不加载不需要的模块）
- 好处2：省内存
- 好处3：避免不必要的依赖冲突
from process.preprocess import preprocess：process：文件夹名（Python包）preprocess：文件名（.py 后缀省略）preprocess：函数名
preprocess()：调用这个函数
case 'train':
from runner.train import train
train()
case 'predict':
from runner.predict import predict
predict()
case 'evaluate':
from runner.evaluate import evaluate
evaluate()
case 'run_app':
from web.app import run_app
run_app()
- 其余四个分支，结构完全一样
- 每个分支：导入对应模块 → 调用对应函数
- 注意：
case后面的缩进是8个空格（两层缩进）
拓展知识 - Python 包（Package）：
process/ ← 这是一个Python包（因为有__init__.py）
├── __init__.py ← 标记这是包（可以为空）
├── preprocess.py ← 包里的一个模块
└── dataset.py ← 包里的另一个模块
第四章 逐行代码详解 - config.py
文件：src/configuration/config.py（26行）
作用：集中管理所有配置常量。好处是修改配置时只需要改这一个文件。
from pathlib import Path
-
导入
Path类，用于处理文件路径 -
pathlib是 Python 3 推荐的路径处理库 -
Path vs 字符串路径：
# 字符串方式（容易出错，Windows用\，Linux用/） path = "F:\\PycharmProjects\\data\\train.txt" # Path方式（自动处理，推荐） path = Path("F:/PycharmProjects") / "data" / "train.txt" -
/运算符：Path对象重载了/运算符，用来拼接路径，非常直观
# 1. 目录
ROOT_DIR = Path(__file__).parent.parent.parent
__file__：当前文件的绝对路径- 例如：
F:\PycharmProjects\product_classification\src\configuration\config.py
- 例如：
.parent：获取父目录（上一级文件夹）- 第1个
.parent：...\src\configuration\ - 第2个
.parent：...\src\ - 第3个
.parent：...\product_classification\（项目根目录）
- 第1个
- 所以
ROOT_DIR= 项目根目录的绝对路径 - 拓展：这种方式非常巧妙，不管项目放在哪个路径下，都能自动找到根目录
DATA_DIR = ROOT_DIR / 'data'
- 数据目录：
项目根目录/data/ ROOT_DIR / 'data'等价于Path.joinpath(ROOT_DIR, 'data')
RAW_DATA_DIR = DATA_DIR / 'raw'
- 原始数据目录：
项目根目录/data/raw/
PROCESSED_DATA_DIR = DATA_DIR / 'processed'
- 处理后数据目录：
项目根目录/data/processed/
LOGS_DIR = ROOT_DIR / 'logs'
- 日志目录：
项目根目录/logs/
MODELS_DIR = ROOT_DIR / 'checkpoints'
- 模型保存目录：
项目根目录/checkpoints/
# 2. 文件
RAW_TRAIN_DATA = 'train.txt'
RAW_TEST_DATA = 'test.txt'
RAW_VALID_DATA = 'valid.txt'
- 定义原始数据文件名
- 使用时：
RAW_DATA_DIR / RAW_TRAIN_DATA→data/raw/train.txt
MODEL_NAME = str(ROOT_DIR / 'pretrained')
- 预训练模型的路径（转为字符串）
- Transformers库的
from_pretrained()需要字符串类型的路径 str()：把 Path 对象转换为字符串
LABELS_FILE = 'labels.txt'
- 标签文件名，保存在
checkpoints/labels.txt
# 3. 超参数
BATCH_SIZE = 64
- 每个batch有64条数据
- 为什么是64？ 经验值。太小（如8）训练不稳定，太大（如256）显存不够
- 拓展：常见batch_size：16, 32, 64, 128, 256，通常是2的幂次
LEARNING_RATE = 1e-5
- 学习率 = 0.00001
- 为什么这么小？ BERT微调时通常用小学习率（1e-5到5e-5），避免破坏预训练学到的知识
- 科学计数法：
1e-5= 1 × 10^(-5) = 0.00001
EPOCHS = 100
- 最多训练100轮（配合早停机制，通常会提前结束）
- 1个epoch = 把所有训练数据看一遍
# 模型保存的迭代次数
SAVE_STEPS = 100
- 每训练100个batch，就做一次验证评估 + 保存检查点
拓展 - 超参数调优：
| 超参数 | 常见范围 | 影响 |
|---|---|---|
| BATCH_SIZE | 16-256 | 越大训练越稳，但需要更多显存 |
| LEARNING_RATE | 1e-5 ~ 5e-4 | 越大学得快但可能不稳定 |
| EPOCHS | 10-100 | 配合早停，通常不需要太大 |
第五章 逐行代码详解 - preprocess.py
文件：src/process/preprocess.py（43行）
作用：把人类能看懂的文字数据，转换成AI能看懂的数字数据。
from configuration.config import *
-
import *：导入 config.py 中定义的所有变量 -
导入后可以直接使用
RAW_DATA_DIR、BATCH_SIZE、MODEL_NAME等 -
拓展：
import *在大型项目中可能导致命名冲突，但在配置文件中使用很常见 -
也可以这样写（更明确但更啰嗦）：
from configuration.config import RAW_DATA_DIR, RAW_TRAIN_DATA, RAW_VALID_DATA, RAW_TEST_DATA, BATCH_SIZE, MODEL_NAME, MODELS_DIR, LABELS_FILE, PROCESSED_DATA_DIR
from transformers import AutoTokenizer
AutoTokenizer：HuggingFace的自动分词器类- “Auto” = 自动选择，会根据模型名称自动判断用哪种分词器
- 本项目用的是BERT的分词器（WordPiece）
- 拓展：HuggingFace是AI领域最重要的开源社区，transformers库是做NLP的必备工具
from datasets import load_dataset, ClassLabel
load_dataset：HuggingFace的数据集加载函数ClassLabel：用于把文字标签编码成数字的类- datasets库：HuggingFace专门用于数据处理的库，高效且易用
def preprocess():
- 定义
preprocess函数 - 这个函数封装了整个预处理流程
print('Preprocessing data...')
- 打印提示信息，告诉用户程序在干什么
- 拓展：在实际项目中，应该用
logging模块代替print
# 1. 从文件加载多个数据集（字典）
dataset_dict = load_dataset('csv', delimiter='\t', data_files={
'train': str(RAW_DATA_DIR/RAW_TRAIN_DATA),
'valid': str(RAW_DATA_DIR/RAW_VALID_DATA),
'test': str(RAW_DATA_DIR/RAW_TEST_DATA)
})
-
load_dataset()：加载数据集 -
'csv'：指定数据格式（TSV是CSV的变种，分隔符不同） -
delimiter='\t'：指定分隔符为Tab键（\t是Tab的转义字符） -
data_files={...}：指定文件路径的字典- key：数据集名称（‘train’, ‘valid’, ‘test’）
- value：文件路径的字符串形式
-
str(...)：Path对象转字符串（load_dataset需要字符串路径） -
返回值
dataset_dict是一个DatasetDict对象，结构如下：DatasetDict({ train: Dataset({features: ['label', 'text_a'], num_rows: 72855}) valid: Dataset({features: ['label', 'text_a'], num_rows: 18214}) test: Dataset({features: ['label', 'text_a'], num_rows: 18214}) })
print(dataset_dict)
- 打印数据集结构信息，方便调试
# 2. 处理标签，进行类别编码
# 2.1 提取所有的标签，保存成列表
labels_list = sorted(set(dataset_dict['train']['label']))
dataset_dict['train']：获取训练集（Dataset对象）['label']：获取标签列，返回一个Python列表，如['母婴', '蔬菜', '母婴', ...]set(...)：转为集合，自动去重 →{'母婴', '蔬菜', '玩具', ...}sorted(...)：排序 →['3C数码', '个护', '书籍', ...]- 为什么要排序？ 保证每次运行结果一致。如果不排序，set的顺序是不确定的
- 为什么要从训练集提取？ 训练集应该包含所有类别
print(labels_list)
- 打印类别列表，确认有30个类别
# 2.2 对标签进行编码转换
dataset_dict = dataset_dict.cast_column('label', ClassLabel(names=labels_list))
cast_column('label', ClassLabel(...))：把’label’列从字符串转为数字ClassLabel(names=labels_list)：创建标签编码器，告诉它有哪些类别- 转换结果：
'母婴' → 12，'蔬菜' → 22，等等（按拼音/笔画排序后的索引） - 注意：
cast_column返回新的 DatasetDict，不会修改原来的
print(dataset_dict['train'][:3])
- 打印前3条数据，查看编码后的结果
# 2.3 将标签列表保存成文件
with open(MODELS_DIR / LABELS_FILE, 'w', encoding='utf-8') as f:
f.write('\n'.join(labels_list))
-
MODELS_DIR / LABELS_FILE：路径拼接 →checkpoints/labels.txt -
open(..., 'w', encoding='utf-8')：以写入模式打开文件'w'= write（写入，会覆盖已有内容）'r'= read（只读）'a'= append（追加）encoding='utf-8'：指定编码，支持中文
-
with ... as f:：上下文管理器，自动关闭文件，即使出错也会关闭 -
'\n'.join(labels_list)：用换行符连接列表元素- 例如：
['母婴', '蔬菜', '玩具']→'母婴\n蔬菜\n玩具'
- 例如：
-
写入后的文件内容：
3C数码 个护 书籍 乳品 ...
# 3. 加载分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
- 从预训练模型路径加载分词器
MODEL_NAME在config.py中定义为str(ROOT_DIR / 'pretrained')- 分词器会读取
pretrained/vocab.txt，知道每个字对应什么编号 - 拓展：分词器和模型是配套的，不能混用。BERT模型必须用BERT的分词器
# 4. 处理标题文本，转换为模型输入
def encode(batch):
- 定义一个内部函数
encode，用于处理一批数据 batch是一个字典：{'text_a': ['标题1', '标题2', ...], 'label': [0, 1, ...]}
inputs = tokenizer(batch['text_a'], truncation=True)
- 用分词器对文本进行编码
batch['text_a']：商品标题列表truncation=True：如果文本超过512个token，自动截断- 返回的
inputs是一个字典：input_ids：每个字的编号，如[[101, 1962, 1906, ...], [101, 1823, ...]]attention_mask：注意力掩码，标记哪些是真实token（1），哪些是填充（0）token_type_ids：句子类型ID（单句分类全为0）
inputs['labels'] = batch['label']
- 把标签（数字）也加入inputs字典
- 为什么叫 ‘labels’？ 因为
AutoModelForSequenceClassification默认用这个key来获取标签
return inputs
- 返回处理后的数据
dataset_dict = dataset_dict.map(encode, batched=True, batch_size=BATCH_SIZE, remove_columns=['label', 'text_a'])
map()：对数据集的每一批数据应用encode函数batched=True：批量处理（一次处理一批，比逐条快很多）batch_size=BATCH_SIZE：每批64条remove_columns=['label', 'text_a']：删除原始的文字列（已经不需要了）- 处理后的数据只包含：
input_ids、attention_mask、token_type_ids、labels（全是数字）
print(dataset_dict['train'][:3])
- 打印前3条处理后的数据，确认转换正确
# 5. 保存数据集
dataset_dict.save_to_disk(PROCESSED_DATA_DIR)
- 保存处理后的数据到
data/processed/目录 - 使用Arrow格式（高效的列式存储格式）
- 为什么要保存？ 处理数据很耗时，保存后下次直接加载
if __name__ == '__main__':
preprocess()
- 入口守卫，允许单独运行此文件
第六章 逐行代码详解 - dataset.py
文件：src/process/dataset.py（43行）
作用：提供数据集加载和DataLoader创建功能。
from datasets import load_from_disk
- 从磁盘加载之前保存的数据集（preprocess.py保存的）
from transformers import AutoTokenizer, DataCollatorWithPadding
-
AutoTokenizer：分词器（虽然在这个文件中没有直接使用，但导入了） -
DataCollatorWithPadding：数据整理器，负责填充（padding） -
为什么要填充？ 一个batch里的文本长度不同，但模型需要固定长度的输入
文本1: [好, 奇, 心] → 长度3 文本2: [基, 地, 玉, 米, 啊] → 长度5 填充后: 文本1: [好, 奇, 心, 0, 0] → 长度5（用0填充） 文本2: [基, 地, 玉, 米, 啊] → 长度5
from torch.utils.data import DataLoader
- PyTorch的数据加载器
- DataLoader的功能：
- 分批加载数据（batch）
- 打乱顺序（shuffle）
- 多进程加载（num_workers）
- 调用collate_fn整理数据
from configuration.config import *
- 导入所有配置
# 获取数据集
def get_dataset(dataset_type='train'):
- 定义获取数据集的函数
dataset_type：数据集类型，默认为 ‘train’
data_path = str(PROCESSED_DATA_DIR / dataset_type)
- 拼接路径，如
data/processed/train
dataset = load_from_disk(data_path)
- 从磁盘加载Arrow格式的数据集
return dataset
- 返回 Dataset 对象
def get_dataloader(tokenizer, dataset_type='train'):
- 定义创建DataLoader的函数
- 需要传入tokenizer，因为DataCollatorWithPadding需要它来知道填充用什么token
# 加载数据集
# data_path = str(PROCESSED_DATA_DIR/dataset_type)
# dataset = load_from_disk(data_path)
dataset = get_dataset(dataset_type)
- 注释掉的是之前的写法，现在改用
get_dataset函数（代码复用）
# 定义一个"对齐函数"，做批量填充
collate_fn = DataCollatorWithPadding(
tokenizer,
padding=True,
return_tensors='pt'
)
- 创建数据整理器
tokenizer：分词器，用于获取填充token的ID（通常是0）padding=True：启用填充，把短序列填充到batch中最长序列的长度return_tensors='pt'：返回PyTorch张量（‘pt’ = PyTorch）
# 创建DataLoader
dataloader = DataLoader(
dataset=dataset,
batch_size=BATCH_SIZE,
shuffle=(True if dataset_type == 'train' else False),
collate_fn=collate_fn
)
- 创建DataLoader
dataset：数据集batch_size=BATCH_SIZE：每批64条shuffle=(True if dataset_type == 'train' else False)：- 训练集：
True（打乱顺序，防止模型记住顺序） - 验证/测试集：
False（不打乱，结果可复现）
- 训练集：
collate_fn=collate_fn：数据整理函数
return dataloader
- 返回DataLoader对象
if __name__ == '__main__':
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
train_loader = get_dataloader(tokenizer, dataset_type='train')
- 测试代码：加载分词器，创建训练集DataLoader
for batch in train_loader:
for k, v in batch.items():
print(f"key {k} -> shape: {v.shape}")
break
-
遍历第一个batch，打印每个字段的形状
-
输出示例：
key input_ids -> shape: torch.Size([64, 32]) key attention_mask -> shape: torch.Size([64, 32]) key token_type_ids -> shape: torch.Size([64, 32]) key labels -> shape: torch.Size([64]) -
[64, 32]= 64条数据，每条32个token -
break：只看第一个batch就停
拓展 - DataLoader 的工作流程：
原始数据集 (Dataset)
↓
DataLoader
├─ 1. 打乱顺序（shuffle）
├─ 2. 按batch_size分组
├─ 3. 调用collate_fn填充对齐
└─ 4. 返回一个batch的张量
第七章 逐行代码详解 - train.py（核心）
文件：src/runner/train.py（273行）
作用：这是整个项目最核心的文件！实现了完整的模型训练逻辑。
7.1 导入部分（第1-15行）
# import sys
# print(sys.path)
import time
- 注释掉的
sys导入和print(sys.path)是调试用的 import time：时间模块，用于生成时间戳（日志目录名）
from dataclasses import dataclass
-
dataclass：装饰器，自动生成__init__、__repr__等方法 -
没有dataclass要这样写：
class TrainConfig: def __init__(self, epochs, batch_size, lr, ...): self.epochs = epochs self.batch_size = batch_size self.lr = lr # ... 很多重复代码 -
有了dataclass只要这样写：
@dataclass class TrainConfig: epochs: int = 100 batch_size: int = 64 lr: float = 1e-5 # ... 简洁多了！
import torch
- PyTorch深度学习框架
- 拓展：PyTorch vs TensorFlow
- PyTorch：更Pythonic，动态图，学术界首选
- TensorFlow：Google开发，静态图，工业部署多
- 2024年后，PyTorch在学术界和工业界都占主导
from torch.utils.data import DataLoader
- 数据加载器
from torch.utils.tensorboard import SummaryWriter
- TensorBoard的写入器
- TensorBoard：Google的训练可视化工具，可以画loss曲线、accuracy曲线等
- 怎么用？ 训练完后运行
tensorboard --logdir=logs，然后浏览器打开
from tqdm import tqdm
- 进度条库
tqdm源自阿拉伯语 “taqaddum”（进度）- 效果：
[Train: Epoch-1]: 45%|████████████████ | 450/1000 [01:30<01:50]
from transformers import AutoTokenizer, AutoModelForSequenceClassification, DataCollatorWithPadding
AutoTokenizer：分词器AutoModelForSequenceClassification：自动序列分类模型（在BERT上加分类头）DataCollatorWithPadding：数据整理器
from sklearn.metrics import accuracy_score, f1_score
accuracy_score：准确率 = 正确预测数 / 总数f1_score：F1分数 = 精确率和召回率的调和平均- 拓展：为什么用F1而不是只用accuracy？
- 如果数据不平衡（如90%是A类，10%是B类），全部预测A也有90%准确率，但没有意义
- F1同时考虑精确率和召回率，更全面
from configuration.config import *
from process.dataset import get_dataset
- 导入配置和数据集加载函数
7.2 TrainConfig 数据类（第17-39行）
# 提取所有训练相关的配置，定义成一个数据类
# class TrainConfig:
# def __init__(self, epochs=EPOCHS, lr=LEARNING_RATE, batch_size=BATCH_SIZE, logs_dir=LOGS_DIR, output_dir=MODELS_DIR):
# self.epochs = epochs
# self.lr = lr
# self.batch_size = batch_size
# self.logs_dir = logs_dir
# self.output_dir = output_dir
- 注释掉的是之前的写法（手写
__init__），现在改用@dataclass
@dataclass
class TrainConfig:
@dataclass：自动生成__init__等方法- 定义训练配置类
# 训练超参数
epochs: int = EPOCHS # 训练轮数，默认100
batch_size: int = BATCH_SIZE # 批大小，默认64
lr: float = LEARNING_RATE # 学习率，默认1e-5
save_steps: int = SAVE_STEPS # 每多少步保存，默认100
- 每个属性都有类型注解（
int、float）和默认值 - 默认值来自config.py
# 目录路径
logs_dir: str = LOGS_DIR # 日志目录
output_dir: str = MODELS_DIR # 模型输出目录
# 早停配置
early_stop_metric: str = 'loss' # 默认监控验证损失
early_stop_patience: int = 5 # 连续5次没改进就停止
- 早停：如果模型连续5次评估都没有进步，就停止训练
- 为什么用loss而不是accuracy？ loss是连续值，更敏感
# AMP控制开关
use_amp: bool = True
- AMP（Automatic Mixed Precision）：混合精度训练
- 用float16代替float32计算，速度快、省内存
- 拓展：float16 vs float32
- float32：32位浮点数，精度高，但占用内存大
- float16：16位浮点数，精度低一点，但占用内存小一半，计算速度快
- AMP：关键操作用float32，其他用float16，兼顾精度和速度
7.3 Trainer 类 - 初始化（第42-67行）
# 面向对象的训练方式：包装一个训练器类
class Trainer:
- 定义训练器类，封装训练逻辑
- 面向对象的好处：把相关的方法和数据封装在一起，代码更清晰
def __init__(self, model, train_dataset, valid_dataset, collate_fn, compute_metrics_fn, device, train_config=TrainConfig()):
- 构造函数，创建Trainer对象时调用
- 参数：
model：AI模型train_dataset：训练数据集valid_dataset：验证数据集collate_fn：数据整理函数compute_metrics_fn：计算评估指标的函数device：计算设备（CPU或GPU）train_config=TrainConfig()：训练配置，默认使用TrainConfig的默认值
self.train_config = train_config
- 保存训练配置到实例属性
self：代表当前对象实例
self.model = model.to(device)
- 把模型移到指定设备
model.to(device)：如果device是GPU，就把模型的所有参数移到GPU显存中- 为什么要移到GPU？ GPU有几千个小核心，并行计算能力强，训练速度快10-100倍
self.device = device
- 保存设备信息
self.train_dataset = train_dataset
self.valid_dataset = valid_dataset
self.collate_fn = collate_fn
self.compute_metrics_fn = compute_metrics_fn
- 保存其他参数
# 内部定义属性：训练过程中用到的工具对象
self.optimizer = torch.optim.Adam(self.model.parameters(), lr=self.train_config.lr)
- 创建Adam优化器
self.model.parameters()：获取模型所有可训练参数（权重和偏置）lr=self.train_config.lr：学习率- Adam是什么？ 最常用的优化算法，能自适应调整每个参数的学习率
- 拓展：其他常见优化器
- SGD：随机梯度下降，经典但需要手动调学习率
- AdamW：Adam + 权重衰减，BERT微调推荐用这个
self.writer = SummaryWriter(log_dir=LOGS_DIR / time.strftime("%Y-%m-%d_%H-%M-%S"))
- 创建TensorBoard写入器
LOGS_DIR / time.strftime(...)：日志目录带时间戳，如logs/2026-06-05_14-30-00time.strftime("%Y-%m-%d_%H-%M-%S")：格式化当前时间
# AMP梯度缩放器
self.scaler = torch.amp.GradScaler(device=device, enabled=self.train_config.use_amp)
- 创建梯度缩放器（AMP专用）
- 为什么需要缩放？ float16的数值范围小，小梯度可能变成0（下溢）。缩放器把loss放大，计算完梯度后再缩小回来
# 训练过程中用到的全局变量
# self.min_loss = float('inf')
self.step = 0
self.step：全局步数计数器- 注释掉的
min_loss是之前的实现
self.early_stop_best_score = -float('inf') # 初始最佳评估得分
- 初始化最佳得分为负无穷大
-float('inf')：负无穷大，任何实际得分都比它大- 为什么要用负无穷？ 因为我们要找最大值（score越大越好），第一次比较时任何分数都会超过它
self.early_stop_counter = 0 # 容忍度计数器
- 记录连续多少次没有改进
# 定义检查点保存路径
self.best_model_path = Path(self.train_config.output_dir) / 'best'
- 最优模型保存路径：
checkpoints/best/ - Path：来自config.py的import *（config.py导入了pathlib.Path）
self.checkpoint_path = Path(self.train_config.output_dir) / 'last' / 'checkpoint.pt'
- 检查点保存路径：
checkpoints/last/checkpoint.pt - 最优模型 vs 检查点：
best/：只保存效果最好的模型（用于预测和部署）last/checkpoint.pt：保存完整的训练状态（用于断点续训）
7.4 Trainer 类 - 内部方法（第69-151行）
# 定义内部方法：获取加载器
def _get_loader(self, dataset):
- 以
_开头的方法是"内部方法"（约定俗成，不建议外部调用） - 参数
dataset：数据集对象
# 创建DataLoader
dataloader = DataLoader(
dataset=dataset,
batch_size=self.train_config.batch_size,
shuffle=True,
collate_fn=self.collate_fn,
)
return dataloader
- 创建DataLoader，参数和dataset.py中类似
# 定义内部方法：训练一个批次（一次迭代）
def _train_one_step(self, inputs):
- 这是训练的核心！ 一个step = 训练一个batch
self.model.train()
- 设置模型为训练模式
- 训练模式 vs 评估模式：
model.train()：Dropout层会随机丢弃神经元，BatchNorm用当前batch的统计量model.eval()：Dropout不丢弃，BatchNorm用全局统计量
- Dropout：训练时随机关闭一些神经元，防止过拟合
inputs = { k:v.to(self.device) for k,v in inputs.items() }
-
字典推导式：把所有输入张量移到指定设备（GPU/CPU）
-
等价于：
new_inputs = {} for k, v in inputs.items(): new_inputs[k] = v.to(self.device) inputs = new_inputs
with torch.autocast(device_type=self.device.type, dtype=torch.float16, enabled=self.train_config.use_amp):
torch.autocast：自动混合精度上下文管理器- 在这个代码块内，PyTorch会自动把某些计算转为float16
device_type=self.device.type：设备类型（‘cuda’或’cpu’）dtype=torch.float16：使用16位浮点数enabled=self.train_config.use_amp：是否启用（可以通过配置关闭）
# 前向传播，得到结果
outputs = self.model(**inputs)
- 前向传播：把输入数据送入模型，得到输出
**inputs：字典解包，等价于self.model(input_ids=..., attention_mask=..., labels=...)outputs包含：outputs.loss：损失值outputs.logits：模型的原始输出（每个类别的得分）
loss = outputs.loss
- 获取损失值
- 损失（Loss）：衡量模型预测与真实标签的差距
- 损失越大 = 错得越离谱
- 损失越小 = 预测越准确
# 反向传播
self.scaler.scale(loss).backward()
self.scaler.scale(loss)：把loss放大（AMP的技巧）.backward()：反向传播，计算所有参数的梯度- 梯度 = 损失对参数的导数，表示"参数往哪个方向调能减小损失"
# 更新参数
self.scaler.step(self.optimizer)
- 优化器根据梯度更新参数
- scaler会先取消缩放，再更新
self.scaler.update()
- 更新缩放因子（根据梯度是否溢出自动调整）
# 梯度清零
self.optimizer.zero_grad()
- 必须清零！ PyTorch默认会累加梯度
- 如果不清零，梯度会越来越大，训练会不稳定
return loss.item()
loss.item()：把张量转为Python数字- 返回这个batch的损失值
7.5 Trainer 类 - 早停逻辑（第99-119行）
# 定义内部方法：判断是否早停（含模型保存逻辑）
def _should_early_stop(self, metrics):
- 判断是否应该提前停止训练
# 将指标转换为当前评估得分
metric = metrics[self.train_config.early_stop_metric]
- 获取监控的指标值（默认是loss）
score = -metric if self.train_config.early_stop_metric == 'loss' else metric
- 统一为"越大越好"的逻辑：
- 如果监控loss（越小越好）→ 取负数
- 如果监控accuracy（越大越好）→ 直接用
# 判断模型保存逻辑
if score > self.early_stop_best_score:
- 如果当前得分超过历史最佳
# 如果大于历史最佳得分，就保存最优模型
self.early_stop_best_score = score
self.early_stop_counter = 0
- 更新最佳得分，重置计数器
self.model.save_pretrained(self.best_model_path)
- 保存最优模型到
checkpoints/best/ save_pretrained会保存：模型权重、配置文件
tqdm.write("保存最佳模型...")
- 在进度条上方打印信息（不会破坏进度条显示）
return False
- 返回False = 不停止训练
else:
# 如果没有改进，不保存，并更新容忍计数器
self.early_stop_counter += 1
# 判断是否超过容忍度上限
if self.early_stop_counter >= self.train_config.early_stop_patience:
return True
else:
return False
- 如果没有改进：计数器+1
- 如果计数器 >= 5（容忍度）：返回True（停止训练）
- 否则：返回False（继续训练）
7.6 Trainer 类 - 检查点（第121-151行）
# 定义内部方法：保存检查点
def _save_checkpoint(self):
- 保存完整的训练状态（用于断点续训）
# 定义检查点：字典对象
checkpoint = {
'model_state_dict': self.model.state_dict(),
'optimizer_state_dict': self.optimizer.state_dict(),
'scaler_state_dict': self.scaler.state_dict(),
'step': self.step,
'early_stop_best_score': self.early_stop_best_score,
'early_stop_counter': self.early_stop_counter,
}
- state_dict = 状态字典，保存了对象的所有内部状态
model_state_dict：模型的所有权重参数optimizer_state_dict：优化器的状态（学习率、动量等）scaler_state_dict：梯度缩放器的状态
- 还保存了训练进度和早停状态
# 保存到文件
tqdm.write("保存检查点...")
torch.save(checkpoint, self.checkpoint_path)
torch.save()：把对象保存到文件
# 定义内部方法：加载检查点
def _load_checkpoint(self):
- 加载检查点，实现断点续训
# 查看检查点路径是否存在，如果存在就加载
if self.checkpoint_path.exists():
.exists()：检查文件是否存在
tqdm.write("发现检查点，加载并继续训练...")
# 从文件获取检查点对象（字典）
checkpoint = torch.load(self.checkpoint_path)
torch.load()：从文件加载对象
# 分别恢复所有的状态
self.model.load_state_dict(checkpoint['model_state_dict'])
self.optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
self.scaler.load_state_dict(checkpoint['scaler_state_dict'])
self.step = checkpoint['step']
self.early_stop_best_score = checkpoint['early_stop_best_score']
self.early_stop_counter = checkpoint['early_stop_counter']
- 恢复所有状态，从上次中断的地方继续训练
else:
tqdm.write("没有检测到检查点，从头开始训练...")
- 没有检查点就从头开始
7.7 Trainer 类 - train方法（第153-188行）
# 核心方法：训练
def train(self):
- 训练的主方法
# 加载检查点
self._load_checkpoint()
- 先加载检查点（如果有）
# 获取训练集加载器
train_loader = self._get_loader(self.train_dataset)
- 创建训练数据的DataLoader
# 循环迭代
for epoch in range(self.train_config.epochs):
- 外层循环：遍历每个epoch（最多100轮）
for inputs in tqdm(train_loader, desc=f"[Train: Epoch-{epoch + 1}]"):
- 内层循环：遍历每个batch
tqdm(...)：显示进度条desc=f"[Train: Epoch-{epoch + 1}]"：进度条描述文字- f-string：Python 3.6+ 的字符串格式化，
{epoch + 1}会被替换为实际值
train_loss = self._train_one_step(inputs)
- 训练一个batch，返回损失值
self.step += 1
- 全局步数+1
# 如果迭代了100次，就记录loss、判断是否保存模型
if self.step % self.train_config.save_steps == 0:
%是取余运算符，step % 100 == 0表示step是100的倍数- 即每100步执行一次
# 记录训练loss
tqdm.write(f'[Train: Epoch {epoch + 1} | Step {self.step} | Loss: {train_loss:.6f}]')
- 打印训练信息
:.6f：保留6位小数
self.writer.add_scalar('loss', train_loss, self.step)
- 记录到TensorBoard
'loss'：指标名称train_loss：指标值self.step：横坐标（步数）
# 传入验证集，得到评估指标
metrics = self.evaluate()
- 在验证集上评估模型
metrics_str = ' | '.join([ f'{k}:{v:.4f}' for k, v in metrics.items() ])
- 列表推导式：把metrics字典格式化为字符串列表
' | '.join(...)：用|连接- 结果类似：
'loss:0.1234 | accuracy:0.9567 | f1:0.9556'
tqdm.write(f'[Evaluate]{metrics_str}')
- 打印评估结果
# 早停和保存模型逻辑
if self._should_early_stop(metrics):
tqdm.write('早停!')
return
- 如果应该早停，打印信息并退出train方法
# 定时保存模型（检查点）
self._save_checkpoint()
- 保存检查点
7.8 Trainer 类 - evaluate方法（第190-219行）
# 核心方法：验证评估，返回评价指标，字典形如{'loss': 1.253, 'accuracy': 0.95, 'f1': 0.92}
def evaluate(self) -> dict:
- 评估方法，返回类型是字典（
-> dict是类型注解）
# 获取验证集加载器
dataloader = self._get_loader(self.valid_dataset)
self.model.eval()
- 设置为评估模式（关闭Dropout等）
# 记录总损失
total_loss = 0
# 记录所有数据的预测标签和真实标签
all_preds = []
all_labels = []
- 初始化变量
with torch.no_grad():
- 禁用梯度计算，节省内存和计算
- 为什么？ 评估时不需要反向传播，不需要计算梯度
# 按批次迭代，进行验证
for inputs in tqdm(dataloader, desc='[Evaluation]'):
- 遍历验证集
inputs = { k:v.to(self.device) for k,v in inputs.items() }
- 移到设备
# 前向传播
outputs = self.model(**inputs)
- 前向传播
# 获取并累加损失
total_loss += outputs.loss.item()
- 累加损失
# 获取输出并预测结果
preds = torch.argmax(outputs.logits, dim=-1)
outputs.logits：模型输出，形状[batch_size, num_classes]torch.argmax(dim=-1)：在最后一个维度取最大值的索引- 例如：
[[0.1, 0.8, 0.1], [0.7, 0.2, 0.1]]→[1, 0]
all_preds.extend(preds.tolist())
preds.tolist()：张量转列表.extend()：把列表元素追加到另一个列表
# 获取真实标签
labels = inputs['labels']
all_labels.extend(labels.tolist())
- 收集真实标签
# 评估指标1：平均损失
loss = total_loss / len(dataloader)
- 计算平均损失（总损失 / batch数）
# 评估指标2：调用外部传入的计算函数，得到指标的字典
metrics = self.compute_metrics_fn(all_preds, all_labels)
- 调用外部函数计算accuracy和f1
return {'loss': loss, **metrics}
**metrics：字典解包- 结果：
{'loss': 0.1234, 'accuracy': 0.9567, 'f1': 0.9556}
7.9 train函数（第221-273行）
def train():
- 训练的主入口函数（不是类的方法，是独立的函数）
print("Training...")
- 打印提示
# 1. 定义设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
- 检查是否有GPU
- 有GPU →
'cuda'，没有 →'cpu' torch.cuda.is_available()：检查CUDA（NVIDIA GPU计算平台）是否可用
# 2. 加载分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
- 加载分词器
# 3. 加载分类标签，加入模型配置
with open(MODELS_DIR/LABELS_FILE, 'r', encoding='utf-8') as f:
labels = f.read().splitlines()
- 读取标签文件
f.read()：读取全部内容.splitlines()：按行分割成列表
id2label = { id:label for id, label in enumerate(labels) }
- 字典推导式 + enumerate
enumerate(labels)：同时获取索引和值 →[(0, '3C数码'), (1, '个护'), ...]- 结果：
{0: '3C数码', 1: '个护', 2: '书籍', ...}
label2id = { label:id for id, label in enumerate(labels) }
- 反向映射：
{'3C数码': 0, '个护': 1, '书籍': 2, ...}
# 4. 加载模型
model = AutoModelForSequenceClassification.from_pretrained(
MODEL_NAME,
num_labels=len(labels),
id2label=id2label,
label2id=label2id,
)
- 加载预训练模型并添加分类头
MODEL_NAME：预训练模型路径num_labels=len(labels)：30个类别id2label和label2id：标签映射，保存到模型配置中- 内部原理：
- 加载BERT模型（12层Transformer）
- 取
[CLS]token的输出（768维向量） - 加一个线性层：768 → 30（类别数）
- 输出30个类别的得分
# 5. 数据集和相关函数
train_dataset = get_dataset('train')
valid_dataset = get_dataset('valid')
- 加载训练集和验证集
# 定义一个"对齐函数"，做批量填充
collate_fn = DataCollatorWithPadding(
tokenizer,
padding=True,
return_tensors='pt'
)
- 创建数据整理器
# 定义一个"计算评估指标函数"，得到字典返回
def compute_metrics_fn(predictions, labels) -> dict:
accuracy = accuracy_score(labels, predictions)
f1 = f1_score(labels, predictions, average='weighted')
return {'accuracy': accuracy, 'f1': f1}
- 定义计算指标的函数
average='weighted'：按类别样本数加权平均（处理类别不平衡）
# 7. 定义训练器
trainer = Trainer(
model=model,
train_dataset=train_dataset,
valid_dataset=valid_dataset,
collate_fn=collate_fn,
compute_metrics_fn=compute_metrics_fn,
device=device
)
- 创建Trainer实例
- 注意：没有传
train_config，使用默认值
# 训练
trainer.train()
- 开始训练！
if __name__ == "__main__":
train()
- 入口守卫
第八章 逐行代码详解 - predict.py
文件：src/runner/predict.py（62行）
作用：用训练好的模型进行预测。
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from configuration.config import *
- 导入需要的库
# 实现自定义的预测器
class Predictor:
- 定义预测器类
# 初始化
def __init__(self, model, tokenizer, device):
self.model = model.to(device)
self.tokenizer = tokenizer
self.device = device
- 初始化：保存模型、分词器、设备
# 核心方法：推理预测，传入商品标题文本，得到分类结果
def predict(self, text: str | list[str]):
- 预测方法
text: str | list[str]：类型注解，表示可以是字符串或字符串列表str | list[str]是 Python 3.10+ 的写法，旧版本用Union[str, List[str]]
# 判断输入，统一成列表形式
is_str = isinstance(text, str)
if is_str:
text = [text]
- 如果输入是单个字符串，转为列表
- 为什么要转？ 统一处理，后面的代码只需要处理列表
isinstance(text, str)：检查类型
# 1. 用分词器，得到模型输入（字典）
inputs = self.tokenizer(
text,
padding=True,
truncation=True,
return_tensors='pt',
)
- 对文本进行分词编码
padding=True：填充到相同长度truncation=True：超长截断return_tensors='pt'：返回PyTorch张量
# 2. 前向传播
self.model.eval()
- 设置为评估模式
with torch.no_grad():
- 禁用梯度计算
inputs = {k: v.to(self.device) for k, v in inputs.items()}
outputs = self.model(**inputs)
- 移到设备，前向传播
# 3. 根据预测输出，得到分类标签
preds = torch.argmax(outputs.logits, dim=-1).tolist()
- 取最大值索引 → 预测的类别编号
results = [ self.model.config.id2label[pred] for pred in preds ]
- 把编号转换为标签名称
self.model.config.id2label：模型配置中的编号→标签映射
# 4. 如果是单条文本，转换为一个分类标签返回
if is_str:
return results[0]
return results
- 如果输入是单条字符串，返回单个结果；否则返回列表
# 预测主流程
def predict():
- 预测的主入口函数
# 1. 定义设备
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# 2. 加载训练好的模型
model = AutoModelForSequenceClassification.from_pretrained(MODELS_DIR / 'best')
- 加载最优模型（checkpoints/best/）
# 3. 创建分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
# 4. 创建预测器
predictor = Predictor(model, tokenizer, device)
# 5. 传入测试数据，推理预测
text = "好奇心钻装纸尿裤L40片9-14kg"
result = predictor.predict(text)
print(f"{text}: {result}")
- 测试单条预测
texts = ['瓦伦丁Wurenbacher小麦西柚啤酒500ml*12听整箱装德国原装进口果啤', '911-267遥控车', '640G正航牛奶早餐饼干']
results = predictor.predict(texts)
print(results)
- 测试批量预测
if __name__ == '__main__':
predict()
- 入口守卫
第九章 逐行代码详解 - evaluate.py
文件：src/runner/evaluate.py（51行）
作用：在测试集上评估模型的最终效果。
import torch
from sklearn.metrics import accuracy_score, f1_score
from transformers import AutoTokenizer, AutoModelForSequenceClassification, DataCollatorWithPadding
from configuration.config import *
from process.dataset import get_dataset
from runner.train import Trainer
- 导入需要的库
- 注意：从
train.py导入Trainer类（复用其evaluate方法）
def evaluate():
print("Evaluating model...")
- 评估主函数
# 1. 定义设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 2. 加载分词器
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
# 3. 加载模型
model = AutoModelForSequenceClassification.from_pretrained(MODELS_DIR/'best')
- 加载最优模型
# 4. 数据集和相关函数
test_dataset = get_dataset('test')
- 加载测试集（不是验证集！）
# 定义一个"对齐函数"，做批量填充
collate_fn = DataCollatorWithPadding(
tokenizer,
padding=True,
return_tensors='pt'
)
# 定义一个"计算评估指标函数"，得到字典返回
def compute_metrics_fn(predictions, labels) -> dict:
accuracy = accuracy_score(labels, predictions)
f1 = f1_score(labels, predictions, average='weighted')
return {'accuracy': accuracy, 'f1': f1}
# 5. 定义训练器
trainer = Trainer(
model=model,
train_dataset=None, # 评估时不需要训练集
valid_dataset=test_dataset, # 把测试集当作验证集传入
collate_fn=collate_fn,
compute_metrics_fn=compute_metrics_fn,
device=device
)
- 巧妙复用：把测试集传入
valid_dataset参数，复用 Trainer 的 evaluate 方法
# 评估
metrics = trainer.evaluate()
print("Evaluation finished.")
print(f"评估结果：{metrics}")
- 执行评估并打印结果
if __name__ == "__main__":
evaluate()
- 入口守卫
第十章 逐行代码详解 - Web服务三件套
10.1 schema.py（4行）
文件：src/web/schema.py
作用：定义请求数据的格式（数据验证模型）。
from pydantic import BaseModel
class Product(BaseModel):
title: str
pydantic：数据验证库，FastAPI内置支持BaseModel：Pydantic的基类Product类：定义商品数据格式，只有一个字段title（字符串）- 自动功能：
- 请求缺少
title→ 返回422错误 title不是字符串 → 返回422错误- 自动生成API文档（访问
/docs）
- 请求缺少
10.2 service.py（7行）
文件：src/web/service.py
作用：服务层，分离接口逻辑和业务逻辑。
class ProductService:
# 初始化，传入一个预测器
def __init__(self, predictor):
self.predictor = predictor
# 预测分类
def predict_category(self, title):
return self.predictor.predict(title)
-
为什么要加这一层？ 分层架构：
接口层（app.py） → 处理HTTP请求 服务层（service.py） → 业务逻辑 预测层（predict.py） → 模型推理 -
各层职责单一，便于测试和维护
10.3 app.py（29行）
文件：src/web/app.py
作用：创建FastAPI Web应用，提供HTTP接口。
import torch
import uvicorn
from fastapi import FastAPI
from transformers import AutoModelForSequenceClassification, AutoTokenizer
from configuration.config import *
from runner.predict import Predictor
from web.schema import Product
from web.service import ProductService
uvicorn：ASGI服务器，运行FastAPI应用FastAPI：现代Web框架，自动生成API文档
app = FastAPI()
- 创建FastAPI应用实例
# 创建一个预测器对象
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = AutoModelForSequenceClassification.from_pretrained(MODELS_DIR/'best')
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
- 在全局作用域加载模型（应用启动时只加载一次）
- 为什么放在全局？ 如果放在路由函数里，每次请求都要加载模型，非常慢
predictor = Predictor(model=model, tokenizer=tokenizer, device=device)
- 创建预测器
# 创建服务层对象
service = ProductService(predictor=predictor)
- 创建服务层
@app.post("/predict")
def predict(product: Product):
@app.post("/predict")：定义POST路由product: Product：请求体，FastAPI自动解析JSON并验证
category_label = service.predict_category(product.title)
return {"商品名": product.title, "类别": category_label}
- 调用服务层预测，返回结果
def run_app():
uvicorn.run("web.app:app", host="0.0.0.0", port=8000)
- 启动Web服务器
"web.app:app"：模块路径:变量名host="0.0.0.0"：监听所有网络接口port=8000：端口号
使用方式：
# 启动服务
python src/main.py run_app
# 调用接口
curl -X POST "http://localhost:8000/predict" \
-H "Content-Type: application/json" \
-d '{"title": "好奇心钻装纸尿裤L40片9-14kg"}'
# 返回
# {"商品名":"好奇心钻装纸尿裤L40片9-14kg","类别":"母婴"}
第十一章 完整流程串讲与数据流图
11.1 数据流图
【第1步：预处理 preprocess.py】
train.txt (文字) processed/train (数字)
┌─────────────┐ ┌─────────────────────┐
│ 母婴 │ │ input_ids: [101,...] │
│ 好奇心钻装 │ ──→ │ attention_mask: [1,] │
│ 纸尿裤 │ 分词编码 │ token_type_ids: [0,] │
│ │ │ labels: 12 │
└─────────────┘ └─────────────────────┘
【第2步：训练 train.py】
processed/train DataLoader Trainer
┌──────────┐ ┌──────────┐ ┌────────────────────────┐
│ 数字数据 │ ──→ │ 每次64条 │ ──→│ 前向传播 → loss │
│ │ │ 打乱顺序 │ │ 反向传播 → 梯度 │
│ │ │ 填充对齐 │ │ 更新参数 → 更好的模型 │
└──────────┘ └──────────┘ └────────────────────────┘
│
↓
checkpoints/best/ (最优模型)
【第3步：预测 predict.py】
"好奇心钻装纸尿裤" → 分词器 → BERT模型 → argmax → "母婴"
【第4步：部署 app.py】
POST请求 → FastAPI → Predictor → 返回JSON
11.2 项目架构图
┌─────────────────────────────────────────────────────────┐
│ main.py (入口) │
│ 根据命令行参数，分发到不同功能 │
├─────────────────────────────────────────────────────────┤
│ │
│ preprocess.py train.py evaluate.py │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 读取数据 │ │ 加载模型 │ │ 加载模型 │ │
│ │ 标签编码 │ │ 训练循环 │ │ 测试评估 │ │
│ │ 文本分词 │ │ 早停保存 │ │ 输出指标 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ ↓ ↓ ↓ │
│ ┌──────────────────────────────────────────┐ │
│ │ config.py (配置中心) │ │
│ │ 路径、超参数、文件名等所有配置 │ │
│ └──────────────────────────────────────────┘ │
│ ↓ ↓ ↓ │
│ ┌──────────────────────────────────────────┐ │
│ │ dataset.py (数据加载) │ │
│ │ get_dataset() get_dataloader() │ │
│ └──────────────────────────────────────────┘ │
│ │
│ predict.py ←──── app.py ←──── service.py │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Predictor│ │ FastAPI │ │ 业务逻辑 │ │
│ │ 推理预测 │ │ HTTP接口 │ │ 调用预测 │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
第十二章 大模型工程师知识拓展
12.1 从BERT到GPT：预训练模型的演进
| 模型 | 年份 | 架构 | 特点 | 应用 |
|---|---|---|---|---|
| BERT | 2018 | Encoder | 双向理解 | 分类、NER、问答 |
| GPT-2 | 2019 | Decoder | 单向生成 | 文本生成 |
| T5 | 2019 | Encoder-Decoder | 统一框架 | 翻译、摘要 |
| GPT-3 | 2020 | Decoder | 1750亿参数 | Few-shot学习 |
| ChatGPT | 2022 | Decoder | RLHF对齐 | 对话、通用任务 |
| GPT-4 | 2023 | 多模态 | 图文理解 | 通用AI |
本项目用的是BERT（Encoder架构），适合分类任务。 如果你要做大模型开发，还需要学习：
- Decoder架构（GPT系列）
- RLHF（人类反馈强化学习）
- Prompt Engineering（提示工程）
12.2 BERT的预训练任务
BERT在预训练时做两个任务：
任务1：MLM（Masked Language Model）遮盖语言模型
输入：我 爱 [MASK] 国 的 天安门
输出：我 爱 中 国 的 天安门
随机遮盖15%的词，让模型预测被遮盖的词。
任务2：NSP（Next Sentence Prediction）下一句预测
句子A：我爱中国
句子B：天安门在北京 → IsNext=True
句子A：我爱中国
句子B：今天天气很好 → IsNext=False
判断两句话是否是上下句关系。
12.3 Transformer 架构简介
输入文本
↓
┌─────────────────────────────────────┐
│ Embedding 层 │
│ 词向量 + 位置编码 + 段落编码 │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ Transformer Block ×12 │
│ ┌───────────────────────────────┐ │
│ │ Multi-Head Self-Attention │ │
│ │ （多头自注意力） │ │
│ └───────────────────────────────┘ │
│ ┌───────────────────────────────┐ │
│ │ Feed-Forward Network │ │
│ │ （前馈神经网络） │ │
│ └───────────────────────────────┘ │
│ + Layer Norm + Residual Connection │
└─────────────────────────────────────┘
↓
┌─────────────────────────────────────┐
│ 分类头（线性层） │
│ 768维 → 30个类别 │
└─────────────────────────────────────┘
↓
输出：每个类别的概率
12.4 作为大模型工程师，你还需要学什么？
| 阶段 | 知识点 | 本项目涉及 |
|---|---|---|
| 基础 | Python编程 | 有 |
| 基础 | PyTorch | 有 |
| 基础 | NLP基础（分词、编码） | 有 |
| 进阶 | 预训练模型（BERT） | 有 |
| 进阶 | 微调技术 | 有 |
| 进阶 | 模型部署（FastAPI） | 有 |
| 高级 | Transformer原理 | 部分 |
| 高级 | 大模型训练（分布式） | 无 |
| 高级 | RLHF对齐 | 无 |
| 高级 | Prompt Engineering | 无 |
| 高级 | RAG（检索增强生成） | 无 |
| 高级 | Agent（智能体） | 无 |
12.5 实际面试中可能问到的问题
Q1：这个项目的BERT模型有多少参数？
A：约1.1亿（110M）。计算：vocab_size × hidden_size + num_layers × (4 × hidden_size²) ≈ 110M
Q2：为什么用F1而不是Accuracy？
A：数据可能不平衡。F1同时考虑精确率和召回率，更适合评估不平衡数据集。
Q3：什么是早停？为什么要用？
A：监控验证集指标，连续N次没有改进就停止。防止过拟合——模型在训练集上越来越好，但在新数据上变差。
Q4：如何优化这个模型？
A：
- 换更大的预训练模型（如RoBERTa-large）
- 数据增强（同义词替换、回译）
- 调参（学习率、batch_size）
- 冻结底层参数，只微调顶层
- 使用对抗训练（FGM、PGD）
第十三章 常见问题
Q1：__init__.py 是什么？为什么是空的？
A：告诉Python"这个文件夹是一个Python包"。可以为空，但必须存在。
Q2：为什么训练和预测时用 model.train() 和 model.eval()？
A：某些层在训练和评估时行为不同：
- Dropout：训练时随机丢弃，评估时不丢弃
- BatchNorm：训练时用当前batch的统计量，评估时用全局统计量
Q3：没有GPU能运行吗？
A：能！代码会自动检测并使用CPU。但训练会很慢（可能几小时到几天）。建议用Google Colab免费GPU。
Q4：loss 一直在下降，但accuracy没有提升？
A：正常！loss是连续值，accuracy是离散值。loss下降说明模型在学习，accuracy可能需要loss下降到一定程度才会跳变。
Q5：如何添加新的商品类别？
A：在原始数据中添加新类别样本 → 重新运行preprocess → 重新运行train。
附录
A. 安装依赖
pip install torch torchvision
pip install transformers datasets
pip install scikit-learn tensorboard
pip install fastapi uvicorn pydantic
pip install tqdm
B. 运行命令
cd F:\PycharmProjects\product_classification
# 数据预处理
python src/main.py preprocess
# 训练
python src/main.py train
# 评估
python src/main.py evaluate
# 预测
python src/main.py predict
# 启动Web服务
python src/main.py run_app
# 查看训练日志
tensorboard --logdir=logs
C. 实际运行结果
好奇心钻装纸尿裤L40片9-14kg → 母婴 ✓
瓦伦丁小麦西柚啤酒500ml*12听 → 酒饮冲调 ✓
911-267遥控车 → 玩具 ✓
640G正航牛奶早餐饼干 → 休闲食品 ✓
D. 代码统计
| 文件 | 行数 | 作用 |
|---|---|---|
| main.py | 28 | 程序入口 |
| config.py | 26 | 配置管理 |
| preprocess.py | 43 | 数据预处理 |
| dataset.py | 43 | 数据加载 |
| train.py | 273 | 模型训练（核心） |
| predict.py | 62 | 模型预测 |
| evaluate.py | 51 | 模型评估 |
| app.py | 29 | Web应用 |
| schema.py | 4 | 数据模型 |
| service.py | 7 | 服务层 |
| 总计 | 566 |
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
7
0- 0
已为社区贡献9条内容
所有评论(0)