基于TorchText实现Transformer架构的多标签关联多输出回归任务的技术咨询

嘿，刚好我之前处理过类似的自定义文本任务，来给你一步步拆解解决方案！

先明确两个核心问题的答案 #

(1) 是否可以使用torchtext实现该任务？ #

完全可以！ TorchText本质是帮你搞定文本数据的预处理、批量加载这些脏活累活，不管是分类、生成还是你这种特殊的多输出回归，只要适配好数据格式，它都能胜任。

(2) 应如何定义标签对应的Field？ #

这是你这个任务的核心难点——因为你的标签是「目标词+4维小数向量」的可变长度组合，常规LabelField没法直接处理。我们需要自定义一个Field来解析这种结构，把它转换成模型能识别的张量格式，后面会给你具体代码和思路。

具体实现步骤 #

1. 自定义Field与数据预处理 #

首先，我们需要定义两个Field：一个处理输入句子，一个处理特殊的标签结构。

输入句子的Field（常规操作）

你的句子已经分词好了，所以这个Field只需要负责构建词表、把词转成索引就行：

from torchtext.legacy import data

# 输入句子的Field：sequential=True表示是序列数据，include_lengths=True返回句子长度方便后续处理
TEXT = data.Field(
    sequential=True,
    tokenize=lambda x: x,  # 已分词，直接用原列表
    lower=False,
    include_lengths=True
)

自定义标签Field

你的标签是[(word, [d1,d2,d3,d4]), ...]这种可变长度结构，我们需要自定义Field来：

• 把目标词转换成TEXT词表里的索引（复用同一个词表）
• 把4维小数向量转成张量
• 处理不同样本标签数量不一致的padding问题

import torch

class CustomLabelField(data.Field):
    def __init__(self, **kwargs):
        # 标签是序列结构，不需要自己构建词表（复用TEXT的）
        super().__init__(sequential=True, use_vocab=False, **kwargs)
    
    def preprocess(self, x):
        # x是单个样本的labels列表：[(word, [d1,d2,d3,d4]), ...]
        processed = []
        for word, vec in x:
            # 用TEXT的词表把目标词转成索引
            word_idx = TEXT.vocab.stoi[word]
            # 把小数列表转成float张量
            vec_tensor = torch.tensor(vec, dtype=torch.float32)
            processed.append((word_idx, vec_tensor))
        return processed
    
    def pad(self, minibatch):
        # 处理一个batch内标签数量不一致的情况，用padding填充到最长长度
        max_label_num = max(len(sample) for sample in minibatch)
        padded_batch = []
        
        for sample in minibatch:
            pad_num = max_label_num - len(sample)
            # 填充的索引用0（对应TEXT词表的<PAD>），向量用全0
            padded_sample = sample + [(0, torch.zeros(4, dtype=torch.float32))] * pad_num
            # 把索引和向量分开整理成张量
            word_indices = torch.tensor([w for w, v in padded_sample], dtype=torch.long)
            vecs = torch.stack([v for w, v in padded_sample])
            padded_batch.append((word_indices, vecs))
        
        # 把整个batch的索引和向量分别堆叠成二维、三维张量
        batch_indices = torch.stack([w for w, v in padded_batch])
        batch_vecs = torch.stack([v for w, v in padded_batch])
        return (batch_indices, batch_vecs)

# 初始化标签Field
LABEL = CustomLabelField()

2. 构建数据集与迭代器 #

假设你的数据存在一个列表data_list里（就是你给出的0:、1:对应的字典集合），我们把它转换成TorchText的Dataset格式：

# 定义字段映射：把字典的key对应到我们定义的Field
fields = [('sentence', TEXT), ('labels', LABEL)]

# 生成Dataset实例
dataset = data.Dataset(
    examples=[data.Example.fromdict(item, fields) for item in data_list],
    fields=fields
)

# 划分训练集和验证集（按需调整比例）
train_data, valid_data = dataset.split(split_ratio=0.8)

# 给TEXT构建词表，可选加载预训练词向量（比如GloVe）
TEXT.build_vocab(train_data, vectors="glove.6B.100d")

# 构建迭代器：按句子长度排序，减少padding带来的计算浪费
train_iterator, valid_iterator = data.BucketIterator.splits(
    (train_data, valid_data),
    batch_size=32,
    sort_key=lambda x: len(x.sentence),
    device=torch.device('cuda' if torch.cuda.is_available() else 'cpu')
)

3. 适配Transformer模型 #

你的任务是输入句子，针对每个目标词输出4维回归向量，模型需要做一点小调整：

• 用Transformer Encoder对句子编码，得到每个词的隐藏状态
• 根据标签里的目标词索引，取出对应位置的隐藏状态
• 用线性层把隐藏状态映射到4维输出，用MSE损失做回归

示例模型代码：

import torch.nn as nn
import torch.nn.functional as F

class TransformerRegressor(nn.Module):
    def __init__(self, input_dim, emb_dim, nhead, hidden_dim, nlayers, output_dim, dropout):
        super().__init__()
        # 词嵌入层，可选加载预训练向量
        self.embedding = nn.Embedding(input_dim, emb_dim)
        self.embedding.weight.data.copy_(TEXT.vocab.vectors)
        
        # Transformer编码器
        self.transformer_encoder = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(emb_dim, nhead, hidden_dim, dropout),
            num_layers=nlayers
        )
        # 最后映射到4维输出的线性层
        self.fc = nn.Linear(emb_dim, output_dim)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, text, text_lengths):
        # text形状：[seq_len, batch_size]（TorchText默认序列在前）
        embedded = self.dropout(self.embedding(text))
        # Transformer输入格式刚好是[seq_len, batch_size, emb_dim]
        encoded = self.transformer_encoder(embedded)
        # 转成[batch_size, seq_len, emb_dim]方便后续按索引取词
        encoded = encoded.permute(1, 0, 2)
        return encoded
    
    def predict(self, encoded, target_indices):
        # target_indices形状：[batch_size, num_labels]，每个元素是目标词在句子中的索引
        # 用gather取出对应位置的隐藏状态
        target_indices = target_indices.unsqueeze(-1)
        target_hidden = torch.gather(encoded, 1, target_indices.expand(-1, -1, encoded.size(2)))
        # 映射到4维输出
        output = self.fc(target_hidden)
        return output

4. 训练逻辑示例 #

训练时从迭代器取出batch数据，计算MSE损失即可：

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# 初始化模型参数
INPUT_DIM = len(TEXT.vocab)
EMB_DIM = 100
NHEAD = 2
HIDDEN_DIM = 200
NLAYERS = 2
OUTPUT_DIM = 4
DROPOUT = 0.5

model = TransformerRegressor(INPUT_DIM, EMB_DIM, NHEAD, HIDDEN_DIM, NLAYERS, OUTPUT_DIM, DROPOUT).to(device)

# 回归任务用MSE损失
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())

def train(model, iterator, optimizer, criterion):
    model.train()
    epoch_loss = 0
    for batch in iterator:
        text, text_lengths = batch.sentence
        target_indices, target_vecs = batch.labels
        
        optimizer.zero_grad()
        
        # 得到句子编码
        encoded = model(text, text_lengths)
        # 生成预测结果
        predictions = model.predict(encoded, target_indices)
        
        # 计算损失
        loss = criterion(predictions, target_vecs)
        
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
    
    return epoch_loss / len(iterator)

额外注意事项 #

• 因为你的目标词都是句子里的词，训练前可以检查一下所有目标词都在TEXT的词表里（根据你的数据集特征，这一点已经满足），避免KeyError。
• 计算损失时可以考虑忽略padding部分的影响：在CustomLabelField的pad方法里额外返回一个mask张量，训练时用mask过滤padding的损失，结果会更准确。
• 如果你用的是TorchText 0.12+的新版本，可以改用torchtext.data的新API，核心思路是一致的，只是数据加载的细节略有不同。

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