基于PyTorch融合文本与表格数据的分类模型开发求助

一、原数据集代码的核心问题 #

• 数值特征被错误转为int64类型，后续MLP处理需用浮点型
• __getitem__中错误使用实例属性存储单样本数据，应改用局部变量
• MyCollate混淆了返回值命名，且未返回目标标签；数值特征为固定长度，无需做padding
• 日志token转小写会丢失格式区分度，原格式的交互标识无需统一小写
• 未正确区分文本序列与数值特征的batch处理逻辑

二、修正后的自定义Dataset代码 #

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pad_sequence

class Vocabulary:
    def __init__(self, freq_threshold, max_size):
        # 特殊token映射
        self.itos = {0: "<PAD>", 1: "<SOS>", 2: "<EOS>", 3: "<UNK>"}
        self.stoi = {v: k for k, v in self.itos.items()}
        self.freq_threshold = freq_threshold
        self.max_size = max_size

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

    @staticmethod
    def tokenizer(text):
        # 直接按空格分割，保留原格式不转小写
        return [tok.strip() for tok in text.split(" ") if tok.strip()]

    def build_vocabulary(self, sentence_list):
        frequencies = {}
        idx = 4  # 跳过特殊token的索引

        # 统计token频率
        for sentence in sentence_list:
            for token in self.tokenizer(sentence):
                frequencies[token] = frequencies.get(token, 0) + 1

        # 过滤低频token
        frequencies = {k: v for k, v in frequencies.items() if v >= self.freq_threshold}

        # 限制词汇表最大规模
        if len(frequencies) > self.max_size - idx:
            frequencies = dict(sorted(frequencies.items(), key=lambda x: -x[1])[:self.max_size - idx])

        # 构建词汇映射
        for token in frequencies.keys():
            self.stoi[token] = idx
            self.itos[idx] = token
            idx += 1

    def numericalize(self, text):
        tokenized_text = self.tokenizer(text)
        return [self.stoi.get(token, self.stoi["<UNK>"]) for token in tokenized_text]

class CustomerDataset(Dataset):
    def __init__(self, df, target_col, text_col):
        # 提取数值特征（排除customer_id、目标列、文本列）
        self.numerical_features = torch.tensor(
            df.drop([target_col, text_col, "customer_id"], axis=1).values.astype(float),
            dtype=torch.float32
        )
        # 提取文本日志列
        self.text_logs = df[text_col].tolist()
        # 提取目标标签
        self.targets = torch.tensor(df[target_col].values, dtype=torch.float32).unsqueeze(1)

        # 构建词汇表
        self.vocab = Vocabulary(freq_threshold=1, max_size=100)
        self.vocab.build_vocabulary(self.text_logs)

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

    def __getitem__(self, idx):
        # 处理文本序列：添加<SOS>和<EOS>
        numericalized_text = [self.vocab.stoi["<SOS>"]]
        numericalized_text += self.vocab.numericalize(self.text_logs[idx])
        numericalized_text.append(self.vocab.stoi["<EOS>"])

        return (
            self.numerical_features[idx],
            torch.tensor(numericalized_text, dtype=torch.long),
            self.targets[idx]
        )

class CustomerCollate:
    def __init__(self, pad_idx):
        self.pad_idx = pad_idx

    def __call__(self, batch):
        # 拆分batch中的各个部分
        numerical_features = torch.stack([item[0] for item in batch])
        text_sequences = [item[1] for item in batch]
        targets = torch.stack([item[2] for item in batch])

        # 对文本序列做padding，batch_first=True方便后续LSTM处理
        padded_text = pad_sequence(
            text_sequences,
            batch_first=True,
            padding_value=self.pad_idx
        )

        return numerical_features, padded_text, targets

def get_data_loader(dataset, batch_size, shuffle=True, num_workers=0):
    pad_idx = dataset.vocab.stoi["<PAD>"]
    return DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=shuffle,
        num_workers=num_workers,
        collate_fn=CustomerCollate(pad_idx=pad_idx)
    )

# 测试代码
if __name__ == "__main__":
    import pandas as pd
    data = {'customer_id':[369799, 103508, 294535, 222573, 204286, 254953, 268167, 56201, 168900, 96618],
            'application_log':['web_pdp__click_main_banner web_pdp__click_prod',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub',
            'web_home__click_main_banner web_home__click_prod',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__view_hero web_pdp__hover_index web_pdp__click_sub',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub web_pdp__click_sub',
            'web_pdp__click_main_banner web_pdp__click_prod web_pdp__click_sub',
            'web_home__click_main_banner web_home__click_prod'],
            'var_1':[0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
            'var_2':[0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
            'var_3':[1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
            'var_4':[0, 1, 0, 5, 1, 3, 6, 7, 1, 0],
            'var_5':[0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
            'targets':[1, 1, 0, 1, 1, 1, 1, 1, 1, 1]}
    df = pd.DataFrame(data)
    dataset = CustomerDataset(df, "targets", "application_log")
    loader = get_data_loader(dataset, batch_size=3)
    for num_feat, text_feat, target in loader:
        print("数值特征形状:", num_feat.shape)
        print("文本特征形状:", text_feat.shape)
        print("目标标签形状:", target.shape)
        break

三、多模态分类模型实现（LSTM+MLP） #

class CustomerPurchasePredictor(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_numerical_features, num_layers=2, dropout=0.2):
        super().__init__()
        # 文本特征处理模块
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(
            input_size=embed_dim,
            hidden_size=hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            bidirectional=True,
            dropout=dropout if num_layers > 1 else 0
        )
        # 取双向LSTM的最后一层隐藏状态作为文本嵌入
        self.text_proj = nn.Linear(hidden_dim * 2, hidden_dim)

        # 数值特征处理模块
        self.numerical_proj = nn.Sequential(
            nn.Linear(num_numerical_features, hidden_dim//2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim//2, hidden_dim)
        )

        # 融合后的分类模块
        self.classifier = nn.Sequential(
            nn.Linear(hidden_dim * 2, hidden_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim, 1),
            nn.Sigmoid()
        )

    def forward(self, numerical_features, text_features):
        # 处理文本特征
        embed = self.embedding(text_features)
        lstm_out, _ = self.lstm(embed)
        # 取最后一个时间步的输出
        text_embed = lstm_out[:, -1, :]
        text_embed = self.text_proj(text_embed)

        # 处理数值特征
        num_embed = self.numerical_proj(numerical_features)

        # 融合特征
        fused_features = torch.cat([text_embed, num_embed], dim=1)

        # 分类预测
        output = self.classifier(fused_features)
        return output

# 测试模型
if __name__ == "__main__":
    dataset = CustomerDataset(df, "targets", "application_log")
    vocab_size = len(dataset.vocab)
    num_numerical_feat = dataset.numerical_features.shape[1]
    model = CustomerPurchasePredictor(
        vocab_size=vocab_size,
        embed_dim=64,
        hidden_dim=128,
        num_numerical_features=num_numerical_feat,
        num_layers=2,
        dropout=0.2
    )
    # 取一个batch测试前向传播
    loader = get_data_loader(dataset, batch_size=3)
    num_feat, text_feat, target = next(iter(loader))
    pred = model(num_feat, text_feat)
    print("预测输出形状:", pred.shape)

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