深度学习文本分类

import torch
import torch.nn as nn

class TextCNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, num_classes, num_filters=100):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        # 三种窗口大小的卷积核,分别捕获 2/3/4-gram
        self.convs = nn.ModuleList([
            nn.Conv1d(embed_dim, num_filters, kernel_size=k)
            for k in [2, 3, 4]
        ])
        self.dropout = nn.Dropout(0.5)
        self.fc = nn.Linear(num_filters * 3, num_classes)

    def forward(self, x):
        # x: (batch, seq_len) → 词 ID 序列
        x = self.embedding(x)             # (batch, seq_len, embed_dim)
        x = x.permute(0, 2, 1)            # Conv1d 要求 (batch, channels, length)

        # 每种卷积核扫描 + ReLU 激活 + 最大池化
        pooled = []
        for conv in self.convs:
            h = torch.relu(conv(x))        # (batch, num_filters, new_length)
            h = h.max(dim=2).values        # (batch, num_filters) ← 取最强信号
            pooled.append(h)

        out = torch.cat(pooled, dim=1)     # 拼接三种卷积的结果
        out = self.dropout(out)
        return self.fc(out)                # (batch, num_classes)

class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(
            embed_dim, hidden_dim,
            batch_first=True,
            bidirectional=True,    # 双向:正着读一遍 + 反着读一遍
        )
        self.dropout = nn.Dropout(0.5)
        self.fc = nn.Linear(hidden_dim * 2, num_classes)  # 双向 → 维度翻倍

    def forward(self, x):
        x = self.embedding(x)              # (batch, seq_len, embed_dim)
        output, (hidden, _) = self.lstm(x)

        # 取双向 LSTM 最后一步的隐藏状态拼接
        hidden = torch.cat([hidden[-2], hidden[-1]], dim=1)
        hidden = self.dropout(hidden)
        return self.fc(hidden)

from torch.utils.data import Dataset, DataLoader

class TextDataset(Dataset):
    def __init__(self, texts, labels, vocab, max_len=128):
        self.labels = labels
        self.max_len = max_len
        self.vocab = vocab
        self.encoded = [self._encode(t) for t in texts]

    def _encode(self, text):
        tokens = list(text)  # 字级别分词(中文最简单的方式)
        ids = [self.vocab.get(t, 1) for t in tokens]  # 1 = UNK
        # 截断或填充到固定长度
        ids = ids[:self.max_len]
        ids += [0] * (self.max_len - len(ids))         # 0 = PAD
        return ids

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        return (
            torch.tensor(self.encoded[idx], dtype=torch.long),
            torch.tensor(self.labels[idx], dtype=torch.long),
        )

# DataLoader 负责分批,打乱,并行加载
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)

model = TextCNN(vocab_size=5000, embed_dim=128, num_classes=2)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss()

for epoch in range(10):
    model.train()
    total_loss = 0

    for batch_x, batch_y in train_loader:
        # 1. 前向传播:输入 → 模型 → 预测
        logits = model(batch_x)

        # 2. 计算损失:预测与真实标签的差距
        loss = criterion(logits, batch_y)

        # 3. 反向传播:计算每个参数对损失的贡献(梯度)
        optimizer.zero_grad()
        loss.backward()

        # 4. 更新参数:沿梯度方向微调参数
        optimizer.step()

        total_loss += loss.item()

    avg_loss = total_loss / len(train_loader)
    print(f"Epoch {epoch+1}, Loss: {avg_loss:.4f}")

from sklearn.metrics import classification_report

def evaluate(model, data_loader):
    model.eval()
    all_preds, all_labels = [], []

    with torch.no_grad():  # 评估时不需要计算梯度,省内存
        for batch_x, batch_y in data_loader:
            logits = model(batch_x)
            preds = logits.argmax(dim=1)
            all_preds.extend(preds.cpu().numpy())
            all_labels.extend(batch_y.cpu().numpy())

    print(classification_report(
        all_labels, all_preds, target_names=["正常", "对骂"]
    ))

# 类别加权示例:对骂样本数是正常的 1/3,给对骂类 3 倍权重
weights = torch.tensor([1.0, 3.0])  # [正常, 对骂]
criterion = nn.CrossEntropyLoss(weight=weights)

# Early stopping 简单实现
best_f1 = 0
patience, wait = 5, 0
for epoch in range(50):
    train_one_epoch(model, train_loader, optimizer, criterion)
    f1 = evaluate_f1(model, val_loader)
    if f1 > best_f1:
        best_f1 = f1
        wait = 0
        torch.save(model.state_dict(), "best_model.pt")  # 保存最优
    else:
        wait += 1
        if wait >= patience:
            print(f"Early stopping at epoch {epoch+1}")
            break

# 完整流程(伪代码级别,突出关键步骤)
# ─────────────────────────────────────

# 1. 加载数据
texts, labels = load_dataset("data/dataset_v1.csv")
vocab = build_vocab(texts, max_size=5000)

# 2. 划分数据集(训练 70% / 验证 15% / 测试 15%)
train_texts, temp_texts, train_labels, temp_labels = train_test_split(
    texts, labels, test_size=0.3, stratify=labels
)
val_texts, test_texts, val_labels, test_labels = train_test_split(
    temp_texts, temp_labels, test_size=0.5, stratify=temp_labels
)

# 3. 构建 DataLoader
train_loader = DataLoader(TextDataset(train_texts, train_labels, vocab), batch_size=32, shuffle=True)
val_loader = DataLoader(TextDataset(val_texts, val_labels, vocab), batch_size=64)
test_loader = DataLoader(TextDataset(test_texts, test_labels, vocab), batch_size=64)

# 4. 初始化模型
model = TextCNN(vocab_size=len(vocab), embed_dim=128, num_classes=2)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.CrossEntropyLoss(weight=torch.tensor([1.0, 3.0]))

# 5. 训练(带 early stopping)
# ... 上面的 early stopping 循环

# 6. 加载最优模型,在测试集上最终评估
model.load_state_dict(torch.load("best_model.pt"))
evaluate(model, test_loader)