PyTorch文本分类模型增量训练验证及TF-IDF适配问题

我正在用PyTorch实现文本分类任务，核心流程如下：

• 加载并处理文本数据
• 使用TF-IDF Vectorizer完成文本向量化
• 构建神经网络，保存TF-IDF Vectorizer与模型用于新数据预测

目前需要每日处理新评论并修正错误分类，当前方案是将标注正确的新评论加入原数据集后全量重训，但该过程耗时且新数据可能在验证阶段丢失。我希望基于人工标注完成的新数据集进行增量训练，但自行编写的代码不确定是否符合预期，现提出两个问题：

1. 如何验证该增量训练方案是否达到预期效果？
2. 当TF-IDF Vectorizer遇到新token时，能否直接使用fit_transform？是否会丢失原有向量器的配置？

原全量训练代码 #

import torch
from torch import nn
from torch.utils.data import Dataset, DataLoader, random_split
from sklearn.preprocessing import LabelEncoder
import polars as pl
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import TfidfVectorizer
import joblib

set1 = (
    pl
    .read_csv(
        "set1.txt",
        separator=";",
        has_header=False,
        new_columns=["text","label"]
    )
)

# 平衡数据集
fear_df = set1.filter(pl.col("label") == "fear")
joy_df = set1.filter(pl.col("label") == "joy").sample(n=2500)
sadness_df = set1.filter(pl.col("label") == "sadness").sample(n=2500)
anger_df = set1.filter(pl.col("label") == "anger")

train_df = pl.concat([fear_df,joy_df,sadness_df,anger_df])

# 标签转数值并划分数据集
label_mapping = {
    "anger": 0,
    "fear": 1,
    "joy": 2,
    "sadness": 3
}

train_mapped = (
    train_df
    .with_columns(
        pl.col("label").replace_strict(label_mapping, default="other").cast(pl.Int16)
    )
)

train_set, pre_Test = train_test_split(train_mapped,
                                    test_size=0.4,
                                    random_state=42,
                                    stratify=train_mapped["label"])

test_set, val_set = train_test_split(pre_Test,
                                    test_size=0.5,
                                    random_state=42,
                                    stratify=pre_Test["label"]) 

# TF-IDF向量化
vectorizer = TfidfVectorizer(max_features=30000, ngram_range=(1, 2))

X_train_tfidf = vectorizer.fit_transform(train_set['text']).toarray()
X_val_tfidf = vectorizer.transform(val_set['text']).toarray()
X_test_tfidf = vectorizer.transform(test_set['text']).toarray()

y_train = train_set['label']
y_val = val_set['label']
y_test = test_set['label']

# 自定义数据集类
class TextDataset(Dataset):
    def __init__(self, texts, labels):
        self.texts = texts
        self.labels = labels
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        return text, label
    
train_dataset = TextDataset(X_train_tfidf, y_train)
val_dataset = TextDataset(X_val_tfidf, y_val)
test_dataset = TextDataset(X_test_tfidf, y_test)

batch_size = 32
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=batch_size)
test_loader = DataLoader(test_dataset, batch_size=batch_size)

# 分类模型定义
class TextClassificationModel(nn.Module):
    def __init__(self, input_dim, num_classes):
        super(TextClassificationModel, self).__init__()
        self.fc1 = nn.Linear(input_dim, 64)
        self.dropout1 = nn.Dropout(0.5)
        self.fc2 = nn.Linear(64, 32)
        self.dropout2 = nn.Dropout(0.5)
        self.fc3 = nn.Linear(32, num_classes)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = self.dropout1(x)
        x = torch.relu(self.fc2(x))
        x = self.dropout2(x)
        x = torch.softmax(self.fc3(x), dim=1)
        return x
    
input_dim = X_train_tfidf.shape[1]
model = TextClassificationModel(input_dim, 4)

# 损失函数与优化器
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adamax(model.parameters())

# 训练循环
num_epochs = 17
best_val_acc = 0.0
best_model_path = "modelbest.pth"

for epoch in range(num_epochs):
    model.train()
    for texts, labels in train_loader:
        texts, labels = texts.float(), labels.long()
        outputs = model(texts)
        loss = criterion(outputs, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

    # 验证环节
    model.eval()
    correct, total = 0, 0
    with torch.no_grad():
        for texts, labels in val_loader:
            texts, labels = texts.float(), labels.long()
            outputs = model(texts)
            _, predicted = torch.max(outputs.data, 1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()
    val_acc = correct / total
    if val_acc > best_val_acc:
        best_val_acc = val_acc
        torch.save(model.state_dict(), best_model_path)

    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}, Val Acc: {val_acc:.4f}')

# 加载最优模型并测试
model.load_state_dict(torch.load(best_model_path))
model.eval()
correct, total = 0, 0
with torch.no_grad():
    for texts, labels in test_loader:
        texts, labels = texts.float(), labels.long()
        outputs = model(texts)
        _, predicted = torch.max(outputs.data, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()
test_acc = correct / total
print(f'Test Acc: {test_acc:.3f}')

# 保存向量器与模型
vectorizer_path = "tfidf_vectorizer.pkl"
joblib.dump(vectorizer, vectorizer_path)

model_path = "text_classification_model.pth"
torch.save(model.state_dict(), model_path)

拟用增量训练代码 #

import torch
import joblib
import polars as pl
from sklearn.model_selection import train_test_split
from torch import nn
from torch.utils.data import Dataset, DataLoader

# 加载预训练的TF-IDF向量器
vectorizer_path = "tfidf_vectorizer.pkl"
vectorizer = joblib.load(vectorizer_path)

input_dim = len(vectorizer.get_feature_names_out())

# 定义模型结构（与原模型一致）
class TextClassificationModel(nn.Module):
    def __init__(self, input_dim, num_classes):
        super(TextClassificationModel, self).__init__()
        self.fc1 = nn.Linear(input_dim, 64)
        self.dropout1 = nn.Dropout(0.5)
        self.fc2 = nn.Linear(64, 32)
        self.dropout2 = nn.Dropout(0.5)
        self.fc3 = nn.Linear(32, num_classes)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = self.dropout1(x)
        x = torch.relu(self.fc2(x))
        x = self.dropout2(x)
        x = torch.softmax(self.fc3(x), dim=1)
        return x
    
# 加载预训练模型
model_path = "text_classification_model.pth"
model = TextClassificationModel(input_dim, 4)
model.load_state_dict(torch.load(model_path))

# 处理新标注数据
label_mapping = {"anger": 0, "fear": 1, "joy": 2, "sadness": 3}
sentiments = ["fear","joy","sadness","anger"]

new_data = (
    pl
    .read_csv(
        "set2.txt",
        separator=";",
        has_header=False,
        new_columns=["text","label"]
    )
    .filter(pl.col("label").is_in(sentiments))
    .with_columns(
        pl.col("label").replace_strict(label_mapping, default="other").cast(pl.Int16)
    )
)
# 用预训练向量器处理新数据
X_new = vectorizer.transform(new_data['text']).toarray()
y_new = new_data['label']

# 自定义数据集类
class TextDataset(Dataset):
    def __init__(self, texts, labels):
        self.texts = texts
        self.labels = labels
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        return text, label

batch_size = 10
   
# 构建新数据的DataLoader
new_train_dataset = TextDataset(X_new, y_new)
new_train_loader = DataLoader(new_train_dataset, batch_size=batch_size, shuffle=True)

# 损失函数与优化器
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adamax(model.parameters())

num_epochs = 5
new_best_model_path = "modelbest.pth"
for epoch in range(num_epochs):
    model.train()
    for texts, labels in new_train_loader:
        texts, labels = texts.float(), labels.long()
        outputs = model(texts)
        loss = criterion(outputs, labels)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        torch.save(model.state_dict(), new_best_model_path)
        
print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')

# 保存增量训练后的模型
new_best_model_path = "new_model.pth"
torch.save(model.state_dict(), new_best_model_path)

问题解答 #

1. 如何验证增量训练方案的效果 #

可以从以下几个维度验证：

• 基准性能对比：将增量训练后的模型与全量重训的模型，在同一测试集（原测试集+新标注数据拆分的测试子集）上对比精度、召回率、F1值等核心指标。如果增量模型性能与全量模型接近甚至持平，说明方案有效。
• 新数据适配性：单独用新标注数据的测试子集评估模型，观察模型对新数据的分类准确率是否提升——这是增量训练的核心目标。
• 稳定性验证：多次重复增量训练流程，观察模型性能的波动情况。如果每次训练后性能都能稳定保持或提升，说明方案鲁棒性达标。
• 效率对比：统计增量训练与全量重训的耗时、CPU/GPU资源占用，确认增量训练确实能节省训练成本。

另外，你的拟用增量代码缺少验证环节，建议从新标注数据中拆分出验证集，训练时跟踪验证准确率，避免模型过拟合新数据。

2. TF-IDF Vectorizer处理新token的正确方式 #

绝对不能直接对新数据使用fit_transform，这会重新训练向量器，完全覆盖原有的词汇表、IDF权重等配置，导致新旧数据的向量维度、权重逻辑不一致，之前训练的模型将无法兼容新数据的向量。

正确的做法是：

• 使用加载好的原有向量器调用transform()处理新数据，新token会被默认忽略，保证向量维度与原有数据一致。
• 如果必须加入新token，需要合并原有词汇表与新token后重新计算IDF权重，但此时模型输入层的维度会增加，需要修改模型输入层的参数（如给fc1层新增对应权重），复杂度较高，仅在新token对分类至关重要时考虑。

你的拟用增量代码中使用vectorizer.transform()处理新数据的逻辑是正确的，建议保持该逻辑。

内容的提问来源于stack exchange，提问作者Simon