完整代码见:gzh2219/-BERT-AI-
1.项目结构
2.model.py详细解释
2.1class ReviewAnalysisModel(nn.Module)类
首先我们看nn.Module，是 PyTorch 里所有神经网络的基类。nn是torch.nn，Module就是“一层网络 / 一个模型”的统一外壳。
继承nn.Module之后，PyTorch 会帮你管三件事：
- 参数：
Linear、Embedding、BERT 权重都会被自动登记，优化器才能更新optimizer = Adam(model.parameters(), ...) - 前向计算：你写
forward()，外面直接model(x)就会调用它 - 设备 / 模式：
model.to("cuda")、model.train()、model.eval()会递归作用到所有子模块。
接着我们看初始化函数:
# BERT层 self.bert = AutoModel.from_pretrained(BERT_NAME) # 全连接层 self.fc = nn.Linear(in_features=self.bert.config.hidden_size, out_features=1)这里面使用的是google-bert/bert-base-chinese这个模型，在订一晚之后我们加一个全连接层实现最后的输出。
再接下来，我们看下是如何实现前向传播的:
# 前向传播，传入x形状 (N, L) def forward(self, input_ids, attention_mask, token_type_ids): # BERT层，output 形状 (N, L, hidden_size) output = self.bert(input_ids=input_ids, attention_mask=attention_mask, token_type_ids=token_type_ids) # 从 output 中提取最终特征，形状 (N, hidden_size) cls_hidden_state = output.pooler_output # 全连接层，输出形状 (N, 1) → (N, ) result = self.fc(cls_hidden_state).squeeze(-1) return result这里面我们需要定义forward，同时需要传入这句话里每个字/词对应词表里的编号（ input_id）;哪些位置是真内容(attention_mask)，哪些是填充（token_type_ids）以及这段字属于第几句话(token_type_ids)。
这三个东西我们再传给bert层得到输出output，其中我们最常用的两项是last_hidden_state以及pooler_output。last_hidden_state指的是每个token对应的向量，而pooler_output指的是整句的汇总向量。再通过一个全连接层得到N个数组成的结果。
3.预处理部分
3.1读取数据文件
dataset_dict = load_dataset("csv", data_files= str(RAW_DATA_DIR/RAW_DATA_FILE)) dataset = dataset_dict["train"]load_dataset("csv", ...)读的是一整份 CSV，还没划分训练/测试。它返回的不是普通dict，而是 Hugging Face 的DatasetDict：外面像字典，里面每个 value 是一个Dataset。
3.2数据提取与数据清洗
# 2. 数据提取和清洗 dataset = dataset.remove_columns(['cat']) dataset = dataset.filter(lambda x: x['review'] is not None) # print(dataset)3.3划分数据集
# 3. 划分数据集（训练和测试的语料） dataset = dataset.cast_column('label', ClassLabel(names=['neg', 'pos'])) dataset_dict = dataset.train_test_split(test_size=TEST_RATE, stratify_by_column='label') # print(dataset_dict)第一句意思如下:CSV 里label本来只是整数：0/1。cast_column不改数字，只换类型：告诉库「这是分类标签」，并且约定：0代表neg负向，1代表正向。第二句的意思是按照标签的比例切开。stratify_by_column='label'：正负样本比例在两份里都保持和全集差不多，避免测试集里全是好评、训练集里全是差评。
print(dataset_dict)会类似：
DatasetDict({ train: Dataset({ features: ['label', 'review'], num_rows: 50218 }) test: Dataset({ features: ['label', 'review'], num_rows: 12555 }) })3.4加载分词器
# 4. 加载分词器 tokenizer = AutoTokenizer.from_pretrained(BERT_NAME)3.5构建数据集
# 5. 构建数据集 def encode(batch_example): inputs = tokenizer(batch_example['review'], padding='max_length', max_length=SEQ_LEN, truncation=True) inputs['labels'] = batch_example['label'] return inputs dataset_dict = dataset_dict.map(encode, batched=True, remove_columns=['review', 'label'])第一行行是在把一批中文评论变成 BERT 能吃的数字，并且全部对齐成长度 128。第二行是单独复制label列。那这里就有一个问题?我们为什么不直接把label也按第一种方式赋值过去呢?答案是tokenizer只吃文本，所以它不会自动带label那一列。
3.6保存数据
dataset_dict.save_to_disk(PROCESSED_DATA_DIR)将数据保存为arrow文件。
4.训练模型部分
4.1按批次处理数据
for batch_dict in tqdm(dataloader, desc="训练"):4.2将输入数据放到device上
targets = batch_dict.pop('labels').to(device).float() inputs = { k: v.to(device) for k, v in batch_dict.items() }4.3训练模型常用步骤
# 前向传播 outputs = model(**inputs) # 计算损失 loss = loss_fn(outputs, targets) # 反向传播，计算梯度 loss.backward() # 更新参数 optimizer.step() # 梯度清零 optimizer.zero_grad() # 累加损失 total_loss += loss.item()4.4整体训练
def train(): # 1. 定义设备 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # 2. 构建数据集加载器 train_loader = get_dataloader() # 4. 定义模型 model = ReviewAnalysisModel() model.to(device) # 5. 定义损失函数 loss_fn = nn.BCEWithLogitsLoss() # 6. 定义优化器 optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE) # 定义一个写入器 writer = SummaryWriter(log_dir=LOGS_DIR / time.strftime("%Y-%m-%d_%H-%M-%S")) # 7. 开始训练 min_loss = float('inf') for epoch in range(EPOCHS): print(f"Epoch {epoch+1}/{EPOCHS}: ") train_loss = train_one_epoch(model=model, device=device, dataloader=train_loader, loss_fn=loss_fn, optimizer=optimizer) print(f"Train Loss: {train_loss}") writer.add_scalar('Loss/train', train_loss, epoch) # 判断是否需要保存模型 if train_loss < min_loss: state_dict = model.state_dict() torch.save(state_dict, MODELS_DIR/BEST_MODEL) min_loss = train_loss print("模型保存成功！") writer.close()5.评估部分
import torch from tqdm import tqdm from predict import predict_batch from config import * from model import ReviewAnalysisModel from dataset import get_dataloader # 评估核心逻辑，返回准确率 def evaluate(model, dataloader, device): model.eval() # 统计个数 total_count = 0 acc_count = 0 with torch.no_grad(): for batch_dict in tqdm(dataloader, desc='评估'): targets = batch_dict.pop('labels').tolist() inputs = {k: v.to(device) for k, v in batch_dict.items()} # 前向传播，直接调用预测方法，得到一组预测概率 batch_proba = predict_batch(model, inputs) # 考察评价指标，统计top1和top5准确个数 for target, proba in zip(targets, batch_proba): total_count += 1 # 将概率转化为预测分类 result = 1 if proba > 0.5 else 0 # 累计预测正确个数 if result == target: acc_count += 1 acc = acc_count / total_count return acc # 运行评估流程 def run_evaluate(): # 1. 定义设备 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') # 3. 加载模型 model = ReviewAnalysisModel().to(device) model.load_state_dict(torch.load(MODELS_DIR / BEST_MODEL)) print("模型加载成功！") # 4. 定义测试集加载器 test_dataloader = get_dataloader(train=False) # 5. 调用评估逻辑 acc = evaluate(model, test_dataloader, device) print("评估结果：") print("预测准确率: {:.6f}%".format(acc * 100)) if __name__ == '__main__': run_evaluate()6.预测部分
import torch from model import ReviewAnalysisModel from config import * from transformers import AutoTokenizer # 预测核心逻辑：模型前向传播，得到一批数据的预测概率 def predict_batch(model, inputs): model.eval() with torch.no_grad(): # 前向传播，得到输出形状 (N, vocab_size) outputs = model(**inputs) # 转换为正类概率，形状 (N, ) proba = torch.sigmoid(outputs) return proba.tolist() # 预测函数：传入一串文本，得到接下来最可能的5个词 def predict(text, model, tokenizer, device): # 1. 处理文本，得到模型输入（字典） inputs = tokenizer( text, padding='max_length', max_length=SEQ_LEN, truncation=True, return_tensors='pt', ) inputs = { k: v.to(device) for k, v in inputs.items() } # 2. 调用核心预测逻辑 result = predict_batch(model, inputs) return result[0] # 调用预测逻辑的应用程序 def run_predict(): # 1. 定义设备 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # 2. 创建分词器 tokenizer = AutoTokenizer.from_pretrained(BERT_NAME) print("词表加载成功！") # 3. 加载模型 model = ReviewAnalysisModel().to(device) state_dict = torch.load(MODELS_DIR/BEST_MODEL) model.load_state_dict(state_dict) print("模型加载成功！") print("欢迎使用AI智评模型！输入 quit 或者 q 退出...") # 核心是一个死循环 while True: # 等待用户输入 user_input_text = input(">") # 如果是 quit 或者 q，就退出 if user_input_text in ['q', 'quit']: print("欢迎下次使用！") break # 如果输入为空字符，继续输入 if user_input_text.strip() == "": print("请输入有效内容...") continue # 调用预测逻辑，得到预测结果 result = predict(user_input_text, model=model, tokenizer=tokenizer, device=device) # 根据概率判断分类 if result > 0.5: print(f"正向 置信度：{result}") else: print(f"负向 置信度：{1 - result}") if __name__ == '__main__': run_predict()
