使用pytorch搭建LSTM进行文本情感分析

zjsnsnxb

834人浏览 · 2025-02-05 16:01:20

zjsnsnxb · 2025-02-05 16:01:20 发布

文本情感分类也算是序列处理任务的hello world，本文会详细讲解LSTM搭建文本情感分类模型及训练的过程，所有代码复制后在笔者的笔记本上都能直接运行，有问题请留言（pytorch版本:2.5.1）

前言

我在查询资料时发现，网上的文本分类的代码很多都使用torchtext，代码陈旧不说，还有一堆兼容性问题。因此，这里我会使用hugging face的datasets库和spacy库处理数据，避免兼容性问题；然后在此基础上搭建一个LSTM模型对imdb数据集进行文本情感分类，所有代码均可直接复制运行。这篇文章适合想要进一步了解RNN、LSTM等模型的使用的用户

一、基本思路

LSTM是RNN的变体，它处理长序列的能力显著强于RNN
在开始之前首先明确一下基本思路：

数据加载与预处理
- 数据集获取
  获取IMDB电影评论数据集
- 数据集预处理
  把数据集处理为需要的形式
- 分词
  直接处理句子太难了，需要分词
- 建立词汇表
  计算机不能直接处理文本，要通过映射转换为数值，为此要建立一个字典，把每个词映射到唯一的数值
- 构建数据集和加载器
  成批加载数据用于训练或者测试，和经典用法有一些改变
构建模型

使用现成的torch.nn.LSTM搭建模型
训练模型
- 设置损失函数和优化器等
  根据任务不同，可以使用不同的损失函数，这里直接二分类，用torch.nn.BCEWithLogitsLoss就行
- 编写训练循环
- 验证模型。
  对模型简单测试，观察是否达到了理想状况
评估模型

在测试集上测试，计算指标。

二、详细步骤

库安装

要用到pytorch、datasets、spacy三个库，在终端里面执行以下代码：

pip install torch datasets spacy
python -m spacy download en_core_web_sm

如果语言包无法安装，可参考
手动安装和测试Spacy中en_core_web_sm模型
进行手动安装

下面正常新建py文件，先把库导进来：

import torch
import spacy
from datasets import load_dataset  # Hugging Face datasets库
from torch.utils.data import DataLoader
from torch.nn.utils.rnn import pad_sequence, pack_padded_sequence
from torch.utils.data import Dataset
from collections import defaultdict
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("Running On", device)

读入数据
```
# 加载IMDB数据集
dataset = load_dataset("imdb")
train_data = dataset["train"]
test_data = dataset["test"]

# 每条数据以字典形式存储：
# {"text": "This movie is great...", "label": 1 (positive) / 0 (negative)}
```
至此数据集已经加载完了，而且分为了训练集和测试集

如果你不能自动下载imdb数据集，需要手动下载（我是死活自动下载不了）
镜像地址：
stanfordnlp/imdb
```
https://hf-mirror.com/datasets/stanfordnlp/imdb
```
点击右边的三个点，点击Clone repository，即可使用Git拉取到本地

没有Git的话可以用百度云：imdb数据集
提取码：3cpd

下载完成后把
```
dataset = load_dataset("imdb")
```
里面的"imdb"改成你的数据集存储的绝对路径就行

分词

用spacy进行分词。

nlp = spacy.load("en_core_web_sm")
tokenizer = lambda x: [token.text for token in nlp.tokenizer(x)]

这个tokenizer可以把文本分为单独的词和符号，例如：

In[3]: tokenizer('I love it so much!!!')  # 这是例子，不要复制错了
Out[3]: ['I', 'love', 'it', 'so', 'much', '!', '!', '!']

构建词汇表（Vocabulary）

class Vocabulary:
    def __init__(self, tokenizer, min_freq=1):
        self.tokenizer = tokenizer
        self.min_freq = min_freq
        self.word2idx = {"<pad>": 0, "<unk>": 1}
        self.idx2word = {0: "<pad>", 1: "<unk>"}
        self.word_counts = defaultdict(int)

    def build_vocab(self, texts):
        # 统计词频
        for text in texts:
            tokens = self.tokenizer(text)
            for token in tokens:
                self.word_counts[token] += 1

        # 过滤低频词
        idx = 2
        for word, count in self.word_counts.items():
            if count >= self.min_freq:
                self.word2idx[word] = idx
                self.idx2word[idx] = word
                idx += 1

    def numericalize(self, text):
        tokens = self.tokenizer(text)
        return [self.word2idx.get(token, self.word2idx["<unk>"]) for token in tokens]

vocab = Vocabulary(tokenizer, min_freq=2)
vocab.build_vocab([i["text"] for i in train_data])

print(f"词汇表大小: {len(vocab.word2idx)}")

建立词汇表就是统计一下有多少个词，所以对每条数据直接用tokenizer分词以后，用字典统计就行，注意给<pad>和<unk>预留位置（defaultdict就是有默认值的字典）

要注意的是，统计词频的时候只需要统计训练集的数据就行了！因为模型根本不知道自己在推理的时候会遇到哪些词，如果把测试集也给统计进去了，可能对训练造成影响。
另一点就是过滤低频词，因为有些词出现次数太少，很难训练到，而且还占用训练时间，这种词都直接设置为<unk>（即未知），但这个词频阈值不能设置太大

定义数据集类

class TextDataset(Dataset):
    def __init__(self, data, vocab, max_len=256):
        self.data = data
        self.vocab = vocab
        self.max_len = max_len  # 截断长度
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        text = self.data[idx]["text"]
        label = self.data[idx]["label"]
        
        # 文本转索引，截断或填充
        numericalized = self.vocab.numericalize(text)
        if len(numericalized) > self.max_len:
            numericalized = numericalized[:self.max_len]
        
        return {
            "text": torch.tensor(numericalized, dtype=torch.long),
            "label": torch.tensor(label, dtype=torch.float),
            "length": len(numericalized)
        }

# 创建Dataset和DataLoader
train_dataset = TextDataset(train_data, vocab, max_len=256)
test_dataset = TextDataset(test_data, vocab, max_len=256)

def collate_fn(batch):
    texts = [item["text"] for item in batch]
    labels = torch.stack([item["label"] for item in batch])
    lengths = torch.tensor([item["length"] for item in batch])
    
    # 填充文本序列 ,设置padding_value=0可以让它忽略填充值的梯度
    padded_texts = pad_sequence(texts, batch_first=True, padding_value=0)
    return padded_texts, labels, lengths

BATCH_SIZE = 64
train_loader = DataLoader(
    train_dataset, batch_size=BATCH_SIZE, shuffle=True, collate_fn=collate_fn
)
test_loader = DataLoader(
    test_dataset, batch_size=BATCH_SIZE, collate_fn=collate_fn
)

这一部分完成的是数据加载任务

max_len代表截断长度，对于很长的句子，直接截取前一段即可
TextData类是数据集类，和常规用法不同的是，因为每个句子的长度不一样，因此它的__getitem__方法返回的是字典；随之，也需要单独定义collate_fn函数收集一个batch的数据
在collate_fn中，直接用pad_sequence填充一批数据，这样可以通过填0来把数据补齐到同一长度

定义LSTM模型
```
class LSTMClassifier(torch.nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, n_layers, dropout):
        super().__init__()
        self.embedding = torch.nn.Embedding(vocab_size, embedding_dim, padding_idx=0)
        self.lstm = torch.nn.LSTM(
            embedding_dim,
            hidden_dim,
            num_layers=n_layers,
            dropout=dropout,
            bidirectional=False,
            batch_first=True
        )
        self.fc = torch.nn.Linear(hidden_dim, output_dim)
        self.dropout = torch.nn.Dropout(dropout)
    
    def forward(self, text, lengths):
        # text: [batch_size, seq_len]
        embedded = self.dropout(self.embedding(text))  # [batch_size, seq_len, emb_dim]
        
        # 处理变长序列
        packed_embedded = pack_padded_sequence(
            embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
        )
        packed_output, (hidden, cell) = self.lstm(packed_embedded)
        
        # 取最后一个有效时间步的输出
        hidden = self.dropout(hidden[-1, :, :])  # [batch_size, hidden_dim]
        return self.fc(hidden).squeeze(1)  # [batch_size]

# 初始化模型
VOCAB_SIZE = len(vocab.word2idx)
EMBEDDING_DIM = 100
HIDDEN_DIM = 256
OUTPUT_DIM = 1
N_LAYERS = 2
DROPOUT = 0.5

model = LSTMClassifier(VOCAB_SIZE, EMBEDDING_DIM, HIDDEN_DIM, OUTPUT_DIM, N_LAYERS, DROPOUT)
model.to(device)
```
这一部分完成的是模型的定义，流程也比较简单

lstm输出3个值，分别是最后一层在所有batch的所有时间步的输出、所有层在最后一个时间步的输出以及一个记忆单元（不用管），所以直接取第二个值的最后一段（即最后一层在最后一个时间步的输出）就行。

这个结构没什么可讲的，可以直接当模板套用；要注意一点就是，在forward方法里面，这里用了pack_padded_sequence函数，这个函数接受填充好的序列和序列的实际长度，输出去掉填充值以后的数据，这样计算的时候不用算那些填充值，可以提高计算效率。

训练与评估

optimizer = torch.optim.Adam(model.parameters())  # 定义优化器
criterion = torch.nn.BCEWithLogitsLoss()  # 定义损失函数

def train_epoch(model, loader, optimizer, criterion):
    model.train()
    total_loss = 0
    for texts, labels, lengths in loader:
        texts, labels, lengths = texts.to(device), labels.to(device), lengths.to(device)
        optimizer.zero_grad()
        predictions = model(texts, lengths)
        loss = criterion(predictions, labels)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
        print(f"\rStep train Loss: {loss.item():.3f}", end="")
    return total_loss / len(loader)

def evaluate(model, loader, criterion):
    model.eval()
    total_loss = 0
    with torch.no_grad():
        for texts, labels, lengths in loader:
            texts, labels, lengths = texts.to(device), labels.to(device), lengths.to(device)
            predictions = model(texts, lengths)
            loss = criterion(predictions, labels)
            total_loss += loss.item()
    return total_loss / len(loader)

# 训练循环
N_EPOCHS = 10
for epoch in range(N_EPOCHS):
    train_loss = train_epoch(model, train_loader, optimizer, criterion)
    test_loss = evaluate(model, test_loader, criterion)
    print(f"\rEpoch: {epoch+1}, Train Loss: {train_loss:.3f}, Test Loss: {test_loss:.3f}")
    torch.save(model, f"model{epoch}_{train_loss:.3f}.pt")

这部分直接当模板用就行，观测的时候直接用的测试集损失值。和CNN的训练相比就是中间多了一个lengths变量
注意如果不是二分类（01这种）就要换一个损失函数

预测

def predict_sentiment(model, text, vocab, max_len=256):
    model.eval()
    numericalized = vocab.numericalize(text)
    if len(numericalized) > max_len:
        numericalized = numericalized[:max_len]
    length = torch.tensor([len(numericalized)])
    tensor = torch.LongTensor(numericalized).unsqueeze(0).to(device)
    with torch.no_grad():
        prediction = torch.sigmoid(model(tensor, length))
    return prediction.item()

text = "This movie is absolutely fantastic!"
print(f"Positive score: {predict_sentiment(model, text, vocab):.4f}")  # 接近1表示正面

text = "The worst film I have ever seen."
print(f"Positive score: {predict_sentiment(model, text, vocab):.4f}")  # 接近0表示负面

预测的时候模型会输出一个数，通过sigmod函数就能转换为0-1之间的分布，这样就能预测文本的类型。

训练示例

直接运行脚本，随着训练进行，训练损失和测试都逐渐减小
最后预测也达到了目标效果
finish

改进

性能优化

使用预训练词向量（如 GloVe 或 Word2Vec）。

尝试双向 LSTM 结构。
实用技巧

动态调整 max_len 平。

添加早停（Early Stopping）和学习率调度。
注意

这里面只要调整dropout就能控制过拟合问题，不要在没有明显过拟合的情况下随便引入其它抗拟合措施（如L2正则化、权重衰减、标签平滑等），否则很容易反过来欠拟合