将上一节使用LSTM实现的评论情感分析模型改为使用GRU，并对比两者的效果，另外也改用RNN实现，对比其效果。
- 项目结构
项目结构如下：
![图片[1]-4.3.7 案例实操（AI智评V2.0）-大赛人网](https://www.dsrw.com/wp-content/uploads/2026/11/image-448.png)
2.完整代码
（1）数据预处理
# process.py
import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import JiebaTokenizer
import config
def process():
print("开始处理数据")
# 1.读取数据
df = pd.read_csv(config.RAW_DATA_DIR / 'online_shopping_10_cats.csv', usecols=['review', 'label'], encoding='utf-8')
# 2.过滤数据
df = df.dropna()
df = df[df['review'].str.strip().ne('')]
# 3.划分数据集
train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
# 4.构建词表
JiebaTokenizer.build_vocab(train_df['review'].tolist(), config.PROCESSED_DATA_DIR / 'vocab.txt')
# 5.构建Tokenizer对象
tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
# 6.构建训练集并保存
train_df['review'] = train_df['review'].apply(lambda x: tokenizer.encode(x, seq_len=config.SEQ_LEN))
train_df.to_json(config.PROCESSED_DATA_DIR / 'indexed_train.jsonl', orient='records', lines=True)
# 7.构建测试集并保存
test_df['review'] = test_df['review'].apply(lambda x: tokenizer.encode(x, seq_len=config.SEQ_LEN))
test_df.to_json(config.PROCESSED_DATA_DIR / 'indexed_test.jsonl', orient='records', lines=True)
print("数据处理完成")
if __name__ == '__main__':
process()
（2）自定义分词器
# tokenizer.py
import jieba
from tqdm import tqdm
jieba.setLogLevel(jieba.logging.WARNING)
class JiebaTokenizer:
unk_token = '<unk>'
pad_token = '<pad>'
@staticmethod
def tokenize(sentence):
"""
分词
:param sentence: 句子
:return: token列表
"""
return jieba.lcut(sentence)
@classmethod
def build_vocab(cls, sentences, vocab_file):
"""
构建并保存词表
:param sentences: 句子列表
:param vocab_file: 词表文件路径
"""
# 1.获取词表
unique_words = set()
for sentence in tqdm(sentences, desc='分词'):
for word in cls.tokenize(sentence):
unique_words.add(word)
vocab_list = [cls.pad_token, cls.unk_token] + list(unique_words)
# 2.保存词表
with open(vocab_file, 'w', encoding='utf-8') as f:
for word in vocab_list:
f.write(word + '\n')
def __init__(self, vocab_list):
"""
初始化tokenizer
:param vocab_list: 词表列表
"""
self.vocab_list = vocab_list # 此表列表(实例属性)
self.vocab_size = len(vocab_list) # 词表大小(实例属性)
self.word2index = {word: index for index, word in enumerate(vocab_list)} # 词到索引(实例属性)
self.index2word = {index: word for index, word in enumerate(vocab_list)} # 索引到词(实例属性)
self.unk_token_index = self.word2index[self.unk_token] # 未知词索引(实例属性)
self.pad_token_index = self.word2index[self.pad_token]
@classmethod
def from_vocab(cls, vocab_file):
"""
加载词表并创建Tokenizer对象
:param vocab_file: 词表文件
:return: tokenizer对象
"""
with open(vocab_file, 'r', encoding='utf-8') as f:
vocab_list = [line[:-1] for line in f.readlines()]
return cls(vocab_list)
def encode(self, sentence, seq_len):
"""
编码
:param sentence: 句子
:param seq_len: 长度
:return: 索引列表
"""
tokens = self.tokenize(sentence)
indexes = [self.word2index.get(token, self.unk_token_index) for token in tokens]
if len(indexes) >= seq_len:
return indexes[:seq_len]
else:
return indexes + [self.pad_token_index] * (seq_len - len(indexes))
（3）自定义数据集
# dataset.py
import torch
from torch.utils.data import Dataset, DataLoader
import pandas as pd
import config
class ReviewAnalyzeDataset(Dataset):
"""
评论情感分析数据集。
"""
def __init__(self, file_path):
"""
初始化数据集。
:param file_path: 数据文件路径（jsonl 格式）
"""
self.data = pd.read_json(file_path, lines=True).to_dict(orient='records')
def __len__(self):
"""
返回数据集大小。
:return: 数据集长度
"""
return len(self.data)
def __getitem__(self, index):
"""
获取单条样本。
:param index: 索引
:return: (input_tensor, target_tensor)
"""
input_tensor = torch.tensor(self.data[index]['review'], dtype=torch.long)
target_tensor = torch.tensor(self.data[index]['label'], dtype=torch.float)
return input_tensor, target_tensor
def get_dataloader(train: bool = True):
"""
获取数据加载器。
:param train: 是否加载训练集
:return: DataLoader
"""
file_name = 'indexed_train.jsonl' if train else 'indexed_test.jsonl'
dataset = ReviewAnalyzeDataset(config.PROCESSED_DATA_DIR / file_name)
return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=True)
if __name__ == '__main__':
dataloader = get_dataloader()
for input_tensor, target_tensor in dataloader:
print(f"输入形状: {input_tensor.shape}, 标签形状: {target_tensor.shape}")
break
（4）模型定义
# model.py
import torch
from torch import nn
import config
from torchinfo import summary
class ReviewAnalyzeModel(nn.Module):
"""
评论情感分析模型：Embedding -> GRU -> Linear
"""
def __init__(self, vocab_size, padding_idx):
"""
初始化模型。
:param vocab_size: 词表大小
:param padding_idx: padding token 的索引
"""
super().__init__()
self.embedding = nn.Embedding(
num_embeddings=vocab_size,
embedding_dim=config.EMBEDDING_DIM,
padding_idx=padding_idx
)
self.gru = nn.GRU(
input_size=config.EMBEDDING_DIM,
hidden_size=config.HIDDEN_DIM,
batch_first=True
)
self.linear = nn.Linear(
in_features=config.HIDDEN_DIM,
out_features=1
)
def forward(self, x):
"""
前向传播。
:param x: 输入索引张量，形状 (batch_size, seq_len)
:return: 输出 logits，形状 (batch_size,)
"""
embed = self.embedding(x) # 嵌入层输出: (batch_size, seq_len, embedding_dim)
gru_output, _ = self.gru(embed) # GRU输出: (batch_size, seq_len, hidden_dim)
final_output = gru_output[:, -1, :] # 取最后时间步输出
logits = self.linear(final_output).squeeze(dim=1) # 线性层 + squeeze: (batch_size,)
return logits
if __name__ == '__main__':
model = ReviewAnalyzeModel(vocab_size=1000, padding_idx=0)
dummy_input = torch.randint(low=0, high=1000, size=(config.BATCH_SIZE, config.SEQ_LEN), dtype=torch.long)
summary(model, input_data=dummy_input)
（5）模型训练
# train.py
import time
import torch
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel
import config
def train_one_epoch(model, dataloader, loss_function, optimizer, device):
"""
单轮训练。
:param model: 模型
:param dataloader: 数据加载器
:param loss_function: 损失函数
:param optimizer: 优化器
:param device: 设备
:return: 平均损失
"""
model.train()
total_loss = 0
for input_tensor, target_tensor in tqdm(dataloader, desc='训练'):
input_tensor = input_tensor.to(device)
target_tensor = target_tensor.to(device)
optimizer.zero_grad()
outputs = model(input_tensor)
loss = loss_function(outputs, target_tensor)
loss.backward()
optimizer.step()
total_loss += loss.item()
return total_loss / len(dataloader)
def train():
"""
模型训练主逻辑。
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
dataloader = get_dataloader(train=True)
tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
model = ReviewAnalyzeModel(vocab_size=tokenizer.vocab_size,
padding_idx=tokenizer.pad_token_index).to(device)
loss_function = torch.nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE)
writer = SummaryWriter(log_dir=config.LOG_DIR / time.strftime('%Y-%m-%d_%H-%M-%S'))
best_loss = float('inf')
for epoch in range(1, config.EPOCHS + 1):
print(f'========== Epoch: {epoch} ==========')
avg_loss = train_one_epoch(model, dataloader, loss_function, optimizer, device)
print(f'Loss: {avg_loss:.4f}')
writer.add_scalar('Loss/Train', avg_loss, epoch)
# 保存最佳模型
if avg_loss < best_loss:
best_loss = avg_loss
torch.save(model.state_dict(), config.MODELS_DIR / 'model.pt')
print('模型保存成功')
if __name__ == '__main__':
train()
（6）模型预测
# predict.py
import torch
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel
import config
def predict_batch(input_tensor, model):
"""
对一个批次输入进行预测。
:param input_tensor: 输入张量 (batch_size, seq_len)
:param model: 模型
:return: 概率列表
"""
model.eval()
with torch.no_grad():
logits = model(input_tensor)
probs = torch.sigmoid(logits)
return probs.tolist()
def predict(user_input: str, model, tokenizer, device):
"""
对单条用户输入进行预测。
:param user_input: 用户输入字符串
:param model: 模型
:param tokenizer: 分词器
:param device: 设备
:return: 概率值
"""
input_indexes = tokenizer.encode(user_input, config.SEQ_LEN)
input_tensor = torch.tensor([input_indexes], dtype=torch.long).to(device)
probs = predict_batch(input_tensor, model)
return probs[0]
def run_predict():
"""
预测交互主逻辑。
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
model = ReviewAnalyzeModel(vocab_size=tokenizer.vocab_size,
padding_idx=tokenizer.pad_token_index).to(device)
model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt'))
print('请输入要预测的评论：（输入 q 或 quit 退出）')
while True:
user_input = input('> ').strip()
if user_input in ['q', 'quit']:
print('退出程序')
break
if not user_input:
print('输入为空，请重新输入')
continue
prob = predict(user_input, model, tokenizer, device)
if prob > 0.5:
print(f'正面评价（置信度：{prob:.2f}）')
else:
print(f'负面评价（置信度：{1 - prob:.2f}）')
if __name__ == '__main__':
run_predict()
（7）模型评估
# evaluate.py
import torch
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel
from dataset import get_dataloader
from predict import predict_batch
import config
def evaluate(model, dataloader, device):
"""
模型评估。
:param model: 模型
:param dataloader: 数据加载器
:param device: 设备
:return: 准确率
"""
model.eval()
total_count = 0
correct_count = 0
for input_tensor, target_tensor in dataloader:
input_tensor = input_tensor.to(device)
target_tensor = target_tensor.tolist()
probs = predict_batch(input_tensor, model)
for prob, target in zip(probs, target_tensor):
pred_label = 1 if prob > 0.5 else 0
if pred_label == target:
correct_count += 1
total_count += 1
return correct_count / total_count
def run_evaluate():
"""
评估主流程。
"""
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
model = ReviewAnalyzeModel(vocab_size=tokenizer.vocab_size,
padding_idx=tokenizer.pad_token_index).to(device)
model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt'))
dataloader = get_dataloader(train=False)
acc = evaluate(model, dataloader, device)
print("========== 评估结果 ==========")
print(f"准确率：{acc:.4f}")
print("=============================")
if __name__ == '__main__':
run_evaluate()
（8）配置文件
4.3.8 存在问题
GRU在简化结构、提高训练效率方面表现优秀，但在超长依赖建模、灵活性和并行计算方面仍存在天然限制。
© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
请登录后查看评论内容