一、文件分工train_eval_test.py文件用来构建词表、训练、验证、测试。load_dataset.py文件用来读取CSV把文本转成词表ID划分训练集、验证集、测试集并做batch迭代。model.py文件用来定义TextRNN模型Embedding BiLSTM Linearmain.py文件是训练入口加载数据、加载预训练词向量、构建模型、开始训练。predict.py文件是加载训练好的模型做单条文本预测并用Gradio做网页界面。simplifyweibo_4_moods.pkl:保存好的词表字——整数IDsimplifyweibo_4_moods.cvs原始博客情绪数据。embedding_Tencent.npz腾讯预训练词向量供Embedding初始化。TextRNN.ckpt训练过程中保存的最佳模型参数。本次项目中运行的顺序是train_eval_test.py——main.py——predict.py二、核心步骤2.1导入依赖库与常量train_eval_test.py在创建的train_eval_test.py中导入NumPy库用于数值计算再导入PyTorch优化器模块进度条库用来显示读取文件的进度导入pickle用来保存和加载词表导入PyTorch函数式接口后面训练时会用F.cross_entropy导入sklearn的评估指标用来计算准确率、分类报告。定义词表最大容量为4760即最多保留4760个高频字。定义两个特殊符号UNK表示未知字PAD表示补齐符合。2.2构建词表把每个字映射成整数ID定义一个函数build_vocab:file_path:语料文件路径max_size:词表最大容量min_freq:最低字频阈值。接着定义一个匿名函数tokenizer,把字符串x拆成单个字符的列表也就是按“字”切分。再创建一个空字典vocb_dic用来统计每个字出现的次数。以UTF-8编码打开语料文件。初始化行号计数器i0用来跳过第一行表头。因为表中第一行是再继续遍历文件中的每一行并用进度条显示。line line[2:].strip()去掉每行前两个字符因为第一个字符是标签第二个字符是逗号。如果这一行去掉后是空行就跳过。for word in tokenizer(line):对正文按字切分逐个去出每个字。vocab_dic[word] vocab_dic.get(word,0) 1统计这个字出现的次数。如果字典里没有这个字就返回0再加1。vocab_list sorted对字频进行排序。[_ for _ in vocab_dic.items() if _[1] min_freq]取出所有(字次数)并且只保留次数大于min_freq的字。key labda x: x[1]排序依据是元组的第二个元素也就是出现的次数字次数。reverseTrue从大到小排序即高频率的在前。[:max_size]只取前max_size个高频字。vocab_dic {word_count[0]: idx for idx, word_count in enumerate(vocab_list)}根据字出现的次数排序后用排序后的序号作为新的ID。频率最高的字ID为0第二高位1以此类推。最后把UNK和PAD加入词表。用pickle把词表保存到simplifyweibo_4_mood.pkl文件中。return vocab_dic返回构建好的词表。vocab build_vocab(simplifyweibo_4_moods.csv, MAX_VOCAB_SIZE, 3)调用build_vocab从CSV中构建词表最大容量4760最低字频3。2.3把文本转成词表ID划分训练集、验证集、测试集(load_dataset.py)首先导入所需要的库和定义未知字和补齐符合后定义函数load_dataset(path, pad_size70):pathCVS文件路径pad_size统一句子的长度默认为70。contents [ ]接着创建一个空列表用来存放所有样本。vocab pkl.load(open(simplifyweibo_4_moods.pkl,rb))加载上一步生成的词表。tokenizerlambda x:[ y for y in x ]定义按字切分函数。with open(path, r, encodingUTF-8) as f:打开CSV文件。i 0行号计算器用来跳过表头。for line in tqdm(f):遍历每一行。if i 0i 1continue跳过第一行表头。if not line:continue如果是空行跳过。label int(line[0])取出标签line[0]是每行第一个字符转成整数。content line[2:].strip(\n)取出正文去掉前两个字符和换行符。words_line [ ]创建一个空列表用来存放这句话转换后的整数ID。token tokensizer(content)把正文按字切分成列表。seq_len len(token)记录这句话的真实长度。if len(token) pad_size:token.extend([PAD] * (pad_size - len(token))) 如果真实长度小于70就用PAD补齐到70个字。elsetoken token[;pad_size]如果真实长度大于等于70只取前70个字。seq_len pad_size把真实长度设为70.for word in token:遍历补齐或截断后的每一个字。word_line.append(vocab.get(word,vocab.get(UNK)))查词表给词表里的每个字分配一个唯一的数字编号然后把句子里的每个字替换成它对应的数字。如果字步骤词表中就用UNK的ID。contents.append((words_line, int(label),seq_len))把一条样本保存为整数ID列表标签真实长度。random.shuffle(contents) 打乱所有样本的顺序。train_data contents[: int(len(contents) * 0.8)]前80%作为训练集。dev_data contents[int(len(contents) * 0.8): int(len(contents) * 0.9)]80%到90%作为验证集。test_data contents[int(len(contents) * 0.9):] 90%到最后作为测试集。定义一个迭代器类用来按batch取数据。def __init__(self,batches,batch_size,device):初始化方法batches:数据列表batch_size每个batch的大小device数据放到CPU还是GPU。self.batch_size batch_size保存batch大小self.data batches保存全部数据self.device device保存设备self.n_batches len(batches) // batch_size把整个数据集分成多少个完整的批次。self.residue (len(batches) % batch_size ! 0)判断是否有剩余不足一个batch的数据。self.index 0 当前batch的索引从0开始。def _to_tensor(self, datas):把batch内的数据转成张量。x torch.LongTensor([_[0] for _ in datas]).to(self.device)取出每个样本的整数ID列表转成LongTenser放到设备上。原样本保存为(整数ID列表标签真实长度)。y torch.LongTensor([_[1] for _ in datas]).to(self.device)取出每个样本的标签转成LongTensor放到设备上。seq_len torch.LongTensor([_[2] for _ in datas]).to(self.device) 取出每个样本的真实长度转成LongTensor放到设备上。return (x, seq_len), y返回输入标签输入是x seq_len)。def __next__(self):定义迭代器的下一个batch。if self.residue and self.index self.n_batches:如果有剩余数据并且已经取完所有完整的batch。batch self.data[self.index * self.batch_size: len(self.data)]取出剩余的数据作为一个batch。self.index 1索引加一。return self._to_tensor(batch)返回这个batch的张量。elif self.index self.n_batches:如果所有batch都取完了。self.index 0重置索引方便下一个epoch重新开始。raise StopIteration抛出停止迭代异常。否则是常规情况取出当前batch的数据索引加1返回这个batch的张量。def __iter__(self):定义迭代器协议。self.index 0每次for循环开始时索引重置为0.return self返回自身。def __len__(self):定义长度。return self.n_batches (1 if self.residue else 0)返回batch树包括可能的剩余batch。2.4定义TextRNN模型(model.py)导入PyTorch和神经网络模块。class Model(nn.Module):定义模型类继承nn.Module。def __init__(self, embedding_pretrained, n_vocab, embed, num_classes): 初始化方法embedding_pretrained:预训练词向量n_vocab词表大小。embed词向量维度。num_classes分类数。super(Model, self).__init__()调用父类初始化。if embedding_pretrained is not None:如果提供了预训练词向量。self.embedding nn.Embedding.from_pretrained(embedding_pretrained, padding_idxn_vocab -1, freezeFalse)用预训练词向量创建Embedding层。embedding_pretrained 传入预训练权重padding_idxn_vocab - 1 指定PAD的索引是词表最后一个Padding不参与梯度更新freeze False 表示预训练词向量可以微调。elseself.embedding nn.Embedding(n_vocab, embed, padding_idxn_vocab - 1)如果没有预训练词向量随机初始化一个Embedding层。self.lstm nn.LSTM(embed, 128, 3, bidirectionalTrue, batch_firstTrue, dropout0.3)创建一个LSTM层。embed 输入维度是词向量维度128 隐藏层维度是1283 LSTM层数是3bidirectionalTrue 使用双向LSTMdropout0.3在训练过程中每一层或多层之间的输出会有 30% 的神经元被随机“丢弃”即输出置为 0。用来防止模型过拟合。self.fc nn.Linear(128 * 2, num_classes)全连接层输入维度是256输出维度是类别数。def forward(self, x):定义前向传播。x, _ x输入是(x, seq_len)这里只取x。out self.embedding(x)把整数ID转成词向量。out_ self.lstm(out)经过LSTMout是每个时间步的输出。out self.fc(out[:, -1, :])取最后一个时间步的输出经过全连接层得到分类logits。2.5 训练入口加载数据、加载预训练词向量、构建模型、开始训练(main.py)导入PyTorch、Numpy、自己写的load_dataset模块、从model.py导入模型类、从train_eval_test.py导入训练函数。选择设备优先CUDA其次MPS最后CPU。np.random.seed(1)设置NumPy随机种子。torch.manual_seed(1)设置PyTorch随机种子。torch.cuda.manual_seed_all(1)设置所有CUDA设备的随机种子。torch.backends.cudnn.deterministic True让cuDNN使用确定性算法保证结果可复现。vocab, train_data, dev_data, test_data load_dataset.load_dataset(simplifyweibo_4_moods.csv)加载数据集返回词表和三个数据划分。train_iter load_dataset.DatasetIterater(train_data, 128, device)创建训练集迭代器 batch size为128。dev_iter load_dataset.DatasetIterater(dev_data, 128, device)创建验证集迭代器。test_iter load_dataset.DatasetIterater(test_data, 128, device)创建测试集迭代器。embedding_pretrained torch.tensor(np.load(embedding_Tencent.npz)[embeddings].astype(float32))加载腾讯预训练词向量转成float32的PyTorch张量。embed embedding_pretrained.size(1) if embedding_pretrained is not None else 300获取词向量维度如果加载失败就默认为300。class_list [喜悦, 愤怒, 厌恶, 低落]定义四个类别。num_classes len(class_list)类别数为4。model Model(embedding_pretrained, len(vocab), embed, num_classes).to(device)构建模型并移动到设备。train(model, train_iter, dev_iter, test_iter, class_list)调用训练函数开始训练。2.6训练、验证、测试train_eval_test.py中的train/evaluate/test)def train(model, train_iter, dev_iter, test_iter, class_list):定义训练函数参数是模型、训练迭代器、验证迭代器测试迭代器、类别列表。model.train()把模型设置为训练模式。optimizer torch.optim.Adam(model.parameters(), lr1e-3)使用Adam优化器学习率0.001。total_batch 0记录总共训练了多少个batch。dev_best_loss float(inf)初始化验证集最佳loss为正无穷。last_improve 0记录最后一次验证集loss提升时的batch数。flag False早停标志。epoch5训练5轮。for epoch in range(epochs):遍历每个epoch。for i, (trains, labels) in enumerate(train_iter):遍历训练集的每个batch。outputs model(trains)前向传播得到预测结果。loss F.cross_entropy(outputs, labels)计算交叉熵损失。model.zero_grad()清空梯度。loss.backward()反向传播。optimizer.step()更新参数。if total_batch % 1000 0:每1000个batch做一次验证。predic torch.max(outputs.data, 1)[1].cpu()取预测类别。train_acc metrics.accuracy_score(labels.data.cpu(), predic)计算当前batch的训练准确率。dev_acc, dev_loss evaluate(class_list, model, dev_iter)在验证集上评估。if dev_loss dev_best_loss:如果验证集 loss 更低。dev_best_loss dev_loss更新最佳 loss。torch.save(model.state_dict(), TextRNN.ckpt)保存当前模型参数。last_improve total_batch记录最后一次提升的 batch。msg IterL {0:6}, Train Loss: {1:5.2}, Train Acc: {2:6.2%}, Val Loss: {3:5.2}, Val Acc: {4:6.2%}定义打印格式。model.train()恢复训练模式。total_batch 1总 batch 数加 1。if total_batch - last_improve 3000:如果 3000 个 batch 都没有提升print(No optimization for a long time, auto-stopping...)打印早停信息。flag True设置早停标志。if flag:break如果早停跳出 epoch 循环。test(model, test_iter, class_list)训练结束后在测试集上测试。def evaluate(class_list, model, data_iter, testFalse):定义评估函数。model.eval()设置为评估模式。loss_total 0总loss。predict_all np.array([], dtypeint)所有预测标签。labels_all np.array([], dtypeint)所有真实标签。with torch.no_grad():不计算梯度。for texts, labels in data_iter:遍历数据。outputs model(texts)前向传播。loss F.cross_entropy(outputs, labels)计算loss。loss_total loss.item()累加 loss。labels labels.data.cpu().numpy()真实标签转 NumPy。predict torch.max(outputs.data, 1)[1].cpu().numpy()预测标签转 NumPy。labels_all np.append(labels_all, labels)追加真实标签predict_all np.append(predict_all, predict)追加预测标签。acc metrics.accuracy_score(labels_all, predict_all)计算准确率。if test:如果是测试report metrics.classification_report(labels_all, predict_all, target_namesclass_list, digits4)return acc, loss_total/len(data_iter), reportreturn acc, loss_total / len(data_iter)生成分类报告return acc, loss_total / len(data_iter), report返回准确率、平均 loss、分类报告。return acc, loss_total / len(data_iter)否则只返回准确率和平均 loss。def test(model, test_iter, class_list):定义测试函数。test_acc, test_loss, test_report evaluate(class_list, model, test_iter, testTrue) 在测试集上评估。