一、文件分工
train_eval_test.py文件用来构建词表、训练、验证、测试。
load_dataset.py文件用来读取CSV，把文本转成词表ID，划分训练集、验证集、测试集，并做batch迭代。
model.py文件用来定义TextRNN模型：Embedding + BiLSTM + Linear
main.py文件是训练入口：加载数据、加载预训练词向量、构建模型、开始训练。
predict.py文件是加载训练好的模型，做单条文本预测，并用Gradio做网页界面。
simplifyweibo_4_moods.pkl:保存好的词表：字——>整数ID
simplifyweibo_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编码打开语料文件。
初始化行号计数器i=0，用来跳过第一行表头。因为表中第一行是
再继续遍历文件中的每一行，并用进度条显示。
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]排序依据是元组的第二个元素，也就是出现的次数（字，次数）。
reverse=True：从大到小排序，即高频率的在前。
[: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_size=70):
pathCVS文件路径；
pad_size统一句子的长度，默认为70。
contents = [ ]接着创建一个空列表，用来存放所有样本。
vocab = pkl.load(open('simplifyweibo_4_moods.pkl','rb'))加载上一步生成的词表。
tokenizer=lambda x:[ y for y in x ]定义按字切分函数。
with open(path, 'r', encoding="UTF-8") as f:打开CSV文件。
i = 0行号计算器，用来跳过表头。
for line in tqdm(f):遍历每一行。
if i == 0：
i += 1
continue跳过第一行表头。
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个字。
else：
token = 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_idx=n_vocab -1, freeze=False)用预训练词向量创建Embedding层。
embedding_pretrained 传入预训练权重；
padding_idx=n_vocab - 1 指定<PAD>的索引是词表最后一个，Padding不参与梯度更新；
freeze = False 表示预训练词向量可以微调。
else：
self.embedding = nn.Embedding(n_vocab, embed, padding_idx=n_vocab - 1)如果没有预训练词向量，随机初始化一个Embedding层。
self.lstm = nn.LSTM(embed, 128, 3, bidirectional=True, batch_first=True, dropout=0.3)创建一个LSTM层。
embed 输入维度是词向量维度；
128 隐藏层维度是128；
3 LSTM层数是3；
bidirectional=True 使用双向LSTM；
dropout=0.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)经过LSTM，out是每个时间步的输出。
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(), lr=1e-3)使用Adam优化器，学习率0.001。
total_batch = 0记录总共训练了多少个batch。
dev_best_loss = float('inf')初始化验证集最佳loss为正无穷。
last_improve = 0记录最后一次验证集loss提升时的batch数。
flag = False早停标志。
epoch=5训练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, test=False):定义评估函数。
model.eval()设置为评估模式。
loss_total = 0总loss。
predict_all = np.array([], dtype=int)所有预测标签。
labels_all = np.array([], dtype=int)所有真实标签。
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_names=class_list, digits=4)
return acc, loss_total/len(data_iter), report
return 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, test=True) 在测试集上评估。
