在当今数字化的时代，分析用户评论中的情感倾向对于了解产品、服务的口碑等方面有着重要意义。而基于强大的预训练语言模型如 Bert 来进行评论情感分析，能够取得较好的效果。 在本次项目中，我们将展示如何利用 Python 语言结合transformers库，借助 Bert 模型实现对给定评论数据集的情感分类任务。
文章目录
- 一、数据集说明
- 二、模型搭建
- 2.1 导包
- 2.2 Bert模型下载
- 2.3 数据集加载
- 2.4 模型加载
- 2.5 训练及评估
- 2.6 训练结果
- 三、完整代码
一、数据集说明
数据集可以在Aitsuio平台下载，下载链接，格式如下：
每个数据有评论内容和后面的情感分类，我们所要做的就是根据评论内容进行分类。
下载数据集后，可以得到拥有三个不同的数据集文件，分别是train.txt、dev.txt和test.txt，它们各自承担不同的角色。train.txt用于训练模型，dev.txt作为开发集（验证集）在训练过程中帮助我们评估模型在未见过的数据上的表现，进而辅助调整模型的超参数等，test.txt则用于最终测试模型的泛化能力。
content: 酒店还是非常的不错，我预定的是套间，服务非常好，随叫随到，结帐非常快。顺便提一句，这个酒店三楼是一个高级KTV，里面的服务小姐非常的漂亮，有机会去看看。
idtype: 2
二、模型搭建
2.1 导包
首先，我们需要确保相关依赖库已经安装好，像transformers库用于方便地加载和使用 Bert 模型，torch库作为深度学习的基础框架（如果有 GPU 支持还能加速训练过程），以及datasets库辅助我们加载和处理数据集。如果未安装，直接pip安装即可。
from transformers import BertTokenizer, BertForSequenceClassification
from datasets import load_dataset
import torch
from torch.utils.data import DataLoader
import numpy as np
2.2 Bert模型下载
所用到的模型是bert-base-chinese，可以在huggingface或者魔塔社区下载，如果不知道如何下载可以看我之前这篇博客，复制对应的地址即可。LLM/深度学习Linux常用指令与代码（进阶）
2.3 数据集加载
为了加载这些数据，需要自定义了一个名为CommentDataset的数据集类，它继承自torch.utils.data.Dataset，这个类负责将文本数据和对应的标签进行整合，方便后续被数据加载器（DataLoader）使用。
在具体的数据加载过程中，分别读取三个文件中的每一行数据，按照空格分割出评论内容和对应的情感标签（示例中假设标签是整数形式，比如 0 代表负面情感，1 代表正面情感等），然后利用 Bert 的tokenizer对文本进行编码，将文本转化为模型能够接受的输入格式（包括添加input_ids、attention_mask等），最终分别创建出对应的训练集、开发集和测试集的Dataset对象以及相应的DataLoader用于按批次加载数据。数据集加载代码如下：
# 自定义数据集类，继承自torch.utils.data.Dataset
class CommentDataset(torch.utils.data.Dataset):
def __init__(self, encodings, labels):
self.encodings = encodings
self.labels = labels
def __getitem__(self, idx):
item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
item['labels'] = torch.tensor(self.labels[idx])
return item
def __len__(self):
return len(self.labels)
# 加载训练集数据
train_data = []
train_labels = []
with open('Dataset/train.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
train_data.append(text)
train_labels.append(int(label))
# 对训练集进行编码
tokenizer = BertTokenizer.from_pretrained('/root/model/bert-base-chinese')
train_encodings = tokenizer(train_data, truncation=True, padding=True)
train_dataset = CommentDataset(train_encodings, train_labels)
train_loader = DataLoader(train_dataset, batch_size=8, shuffle=True)
# 加载开发集数据
dev_data = []
dev_labels = []
with open('Dataset/dev.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
dev_data.append(text)
dev_labels.append(int(label))
# 对开发集进行编码
dev_encodings = tokenizer(dev_data, truncation=True, padding=True)
dev_dataset = CommentDataset(dev_encodings, dev_labels)
dev_loader = DataLoader(dev_dataset, batch_size=8)
# 加载测试集数据
test_data = []
test_labels = []
with open('Dataset/test.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
test_data.append(text)
test_labels.append(int(label))
# 对测试集进行编码
test_encodings = tokenizer(test_data, truncation=True, padding=True)
test_dataset = CommentDataset(test_encodings, test_labels)
test_loader = DataLoader(test_dataset, batch_size=8)
2.4 模型加载
利用BertForSequenceClassification一行命令即可加载模型。
# 加载预训练的Bert模型用于序列分类，这里假设是二分类（0和1代表不同情感倾向），可以根据实际调整num_labels
model = BertForSequenceClassification.from_pretrained('/root/model/bert-base-chinese', num_labels=len(idset))
2.5 训练及评估
接下来定义一个train函数，一个val函数，一个test函数，分别对应三个文件，然后就可以开始训练和测试啦！
# 训练函数
def train():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
epochs = 3 # 可以根据实际调整训练轮数
for epoch in range(epochs):
model.train()
for batch in train_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
loss = outputs.loss
loss.backward()
optimizer.step()
optimizer.zero_grad()
print(f'Epoch {epoch + 1} completed')
# 在开发集上验证
validate()
# 验证函数（在开发集上查看模型表现辅助调参）
def validate():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.eval()
correct = 0
total = 0
with torch.no_grad():
for batch in dev_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask)
_, predicted = torch.max(outputs.logits, dim=1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
print(f'Validation Accuracy: {correct / total}')
# 评估函数
def evaluate():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.eval()
correct = 0
total = 0
with torch.no_grad():
for batch in test_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask)
_, predicted = torch.max(outputs.logits, dim=1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
print(f'Test Accuracy: {correct / total}')
train()
evaluate()
2.6 训练结果
可以看到，在test上有94%的正确率，还是非常不错的！
Epoch 1 completed
Validation Accuracy: 0.9435662345874171
Epoch 2 completed
Validation Accuracy: 0.9521024343977237
Epoch 3 completed
Validation Accuracy: 0.9524185899462535
Test Accuracy: 0.9485416172634574
三、完整代码
完整代码如下，需要的小伙伴可以自己获取！
with open('Dataset/train.txt', 'r', encoding='utf-8') as f:
train_data = f.read()
with open('Dataset/dev.txt', 'r', encoding='utf-8') as f:
eval_data = f.read()
train_data = train_data.split('\n')
eval_data = eval_data.split('\n')
idset = set()
for data in train_data:
idset.add(data.split('\t')[-1])
for data in eval_data:
idset.add(data.split('\t')[-1])
from transformers import BertTokenizer, BertForSequenceClassification
from datasets import load_dataset
import torch
from torch.utils.data import DataLoader
import numpy as np
# 自定义数据集类，继承自torch.utils.data.Dataset
class CommentDataset(torch.utils.data.Dataset):
def __init__(self, encodings, labels):
self.encodings = encodings
self.labels = labels
def __getitem__(self, idx):
item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()}
item['labels'] = torch.tensor(self.labels[idx])
return item
def __len__(self):
return len(self.labels)
# 加载训练集数据
train_data = []
train_labels = []
with open('Dataset/train.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
train_data.append(text)
train_labels.append(int(label))
# 对训练集进行编码
tokenizer = BertTokenizer.from_pretrained('/root/model/bert-base-chinese')
train_encodings = tokenizer(train_data, truncation=True, padding=True)
train_dataset = CommentDataset(train_encodings, train_labels)
train_loader = DataLoader(train_dataset, batch_size=8, shuffle=True)
# 加载开发集数据
dev_data = []
dev_labels = []
with open('Dataset/dev.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
dev_data.append(text)
dev_labels.append(int(label))
# 对开发集进行编码
dev_encodings = tokenizer(dev_data, truncation=True, padding=True)
dev_dataset = CommentDataset(dev_encodings, dev_labels)
dev_loader = DataLoader(dev_dataset, batch_size=8)
# 加载测试集数据
test_data = []
test_labels = []
with open('Dataset/test.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
text, label = line.split('\t')
test_data.append(text)
test_labels.append(int(label))
# 对测试集进行编码
test_encodings = tokenizer(test_data, truncation=True, padding=True)
test_dataset = CommentDataset(test_encodings, test_labels)
test_loader = DataLoader(test_dataset, batch_size=8)
# 加载预训练的Bert模型用于序列分类，这里假设是二分类（0和1代表不同情感倾向），可以根据实际调整num_labels
model = BertForSequenceClassification.from_pretrained('/root/model/bert-base-chinese', num_labels=len(idset))
# 训练函数
def train():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-5)
epochs = 3 # 可以根据实际调整训练轮数
for epoch in range(epochs):
model.train()
for batch in train_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
loss = outputs.loss
loss.backward()
optimizer.step()
optimizer.zero_grad()
print(f'Epoch {epoch + 1} completed')
# 在开发集上验证
validate()
# 验证函数（在开发集上查看模型表现辅助调参）
def validate():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.eval()
correct = 0
total = 0
with torch.no_grad():
for batch in dev_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask)
_, predicted = torch.max(outputs.logits, dim=1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
print(f'Validation Accuracy: {correct / total}')
# 评估函数
def evaluate():
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model.eval()
correct = 0
total = 0
with torch.no_grad():
for batch in test_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask)
_, predicted = torch.max(outputs.logits, dim=1)
total += labels.size(0)
correct += (predicted == labels).sum().item()
print(f'Test Accuracy: {correct / total}')
if __name__ == "__main__":
train()
evaluate()
