完整代码见:gzh2219/-BERT-AI-1.项目结构2.model.py详细解释2.1class ReviewAnalysisModel(nn.Module)类首先我们看nn.Module是 PyTorch 里所有神经网络的基类。nn是torch.nnModule就是“一层网络 / 一个模型”的统一外壳。继承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_featuresself.bert.config.hidden_size, out_features1)这里面使用的是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_idsinput_ids, attention_maskattention_mask, token_type_idstoken_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_sizeTEST_RATE, stratify_by_columnlabel) # print(dataset_dict)第一句意思如下:CSV 里label本来只是整数0/1。cast_column不改数字只换类型告诉库「这是分类标签」并且约定0代表neg负向1代表正向。第二句的意思是按照标签的比例切开。stratify_by_columnlabel正负样本比例在两份里都保持和全集差不多避免测试集里全是好评、训练集里全是差评。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], paddingmax_length, max_lengthSEQ_LEN, truncationTrue) inputs[labels] batch_example[label] return inputs dataset_dict dataset_dict.map(encode, batchedTrue, 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(), lrLEARNING_RATE) # 定义一个写入器 writer SummaryWriter(log_dirLOGS_DIR / time.strftime(%Y-%m-%d_%H-%M-%S)) # 7. 开始训练 min_loss float(inf) for epoch in range(EPOCHS): print(fEpoch {epoch1}/{EPOCHS}: ) train_loss train_one_epoch(modelmodel, devicedevice, dataloadertrain_loader, loss_fnloss_fn, optimizeroptimizer) print(fTrain 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(trainFalse) # 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, paddingmax_length, max_lengthSEQ_LEN, truncationTrue, return_tensorspt, ) 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, modelmodel, tokenizertokenizer, devicedevice) # 根据概率判断分类 if result 0.5: print(f正向 置信度{result}) else: print(f负向 置信度{1 - result}) if __name__ __main__: run_predict()