基于Transformer的多实例分类模型构建与优化问题咨询

业务场景与问题概述 #

基于车辆维修时序数据构建多实例分类模型：选取两个时间点t1、t2，将t1时刻在修车辆与t2时刻计划维修的车辆归为一个"bag"，提取t1时刻车辆已维修时长、t1到t2计划开始时长等特征，目标是预测bag中每个车辆是否会结束维修或进入维修，而非仅预测bag内维修车辆的总数。选择Transformer的原因是：

它不对数据的时空关系做任何假设，非常适合处理对象集合

数据生成与预处理 #

数据生成 #

每个bag最多包含5个实例，每个实例有3个特征，生成代码如下：

# Generate data
max_length = 5
x_lst = []
y_lst = []
for _ in range(10):
    num_instances = np.random.randint(2, max_length + 1)
    x_bag = np.random.randint(0, 9, size=(num_instances, 3))
    y_bag = np.random.randint(0, 2, size=(num_instances))
    
    x_lst.append(x_bag)
    y_lst.append(y_bag)

前2个bag的特征与标签（分别包含5个和2个实例）：

x_lst[:2]

[array([[8, 0, 3],
        [8, 1, 0],
        [4, 6, 8],
        [1, 6, 4],
        [7, 4, 6]]),
 array([[5, 8, 4],
        [2, 1, 1]])]

y_lst[:2]

[array([0, 1, 1, 1, 0]), array([0, 0])]

数据填充 #

对特征补0、标签补-1以统一bag长度，代码如下：

x_padded = []
y_padded = []

for x, y in zip(x_lst, y_lst):
    x_p = np.zeros((max_length, 3))
    x_p[:x.shape[0], :x.shape[1]] = x
    x_padded.append(x_p)

    y_p = np.negative(np.ones(max_length))
    y_p[:y.shape[0]] = y
    y_padded.append(y_p)

X = np.stack(x_padded)
y = np.stack(y_padded)

处理后X.shape为(10, 5, 3)，y.shape为(10, 5)。

初始模型问题 #

基于Keras时序Transformer示例做两处修改：输入层后添加Masking层，最后一层Dense设为5（对应bag最大长度）并使用sigmoid激活，模型代码如下：

def transformer_encoder(inputs, head_size, num_heads, ff_dim, dropout=0):
    # Attention and Normalization
    x = layers.MultiHeadAttention(
        key_dim=head_size, num_heads=num_heads, dropout=dropout
    )(inputs, inputs)
    x = layers.Dropout(dropout)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    res = x + inputs

    # Feed Forward Part
    x = layers.Conv1D(filters=ff_dim, kernel_size=1, activation="relu")(res)
    x = layers.Dropout(dropout)(x)
    x = layers.Conv1D(filters=inputs.shape[-1], kernel_size=1)(x)
    x = layers.LayerNormalization(epsilon=1e-6)(x)
    return x + res

def build_model(
    input_shape,
    head_size,
    num_heads,
    ff_dim,
    num_transformer_blocks,
    mlp_units,
    dropout=0,
    mlp_dropout=0,
):
    inputs = keras.Input(shape=input_shape)
    inputs = keras.layers.Masking(mask_value=0)(inputs) # ADDED MASKING LAYER
    x = inputs
    for _ in range(num_transformer_blocks):
        x = transformer_encoder(x, head_size, num_heads, ff_dim, dropout)

    x = layers.GlobalAveragePooling1D(data_format="channels_first")(x)
    for dim in mlp_units:
        x = layers.Dense(dim, activation="relu")(x)
        x = layers.Dropout(mlp_dropout)(x)
    outputs = layers.Dense(5, activation='sigmoid')(x) # CHANGED ACCORDING TO MY OUTPUT
    return keras.Model(inputs, outputs)

input_shape = (5, 3)

model = build_model(
    input_shape,
    head_size=256,
    num_heads=4,
    ff_dim=4,
    num_transformer_blocks=4,
    mlp_units=[128],
    mlp_dropout=0.4,
    dropout=0.25,
)

model.compile(
    loss="binary_crossentropy",
    optimizer=keras.optimizers.Adam(learning_rate=1e-4),
    metrics=["binary_accuracy"],
)
model.summary()

该模型学习效果极差，但改为预测每个bag中True值的数量（使用y.sum(axis=1)和Dense(1)）时，模型学习效果良好。

自定义损失后的问题 #

自定义掩码损失函数后，模型看似开始学习（见下图），但所有非掩码标签均被预测为1（见下图）：

def my_loss_fn(y_true, y_pred):
    mask = tf.cast(tf.math.not_equal(y_true, tf.constant(-1.)), tf.float32)
    y_true, y_pred = tf.expand_dims(y_true, axis=-1), tf.expand_dims(y_pred, axis=-1)
    bce = tf.keras.losses.BinaryCrossentropy(reduction='none')
    return tf.reduce_sum(tf.cast(bce(y_true, y_pred), tf.float32) * mask)

mask = (y_test != -1).astype(int)
pd.DataFrame({'n_labels': mask.sum(axis=1), 'preds': ((preds * mask) >= .5).sum(axis=1)}).plot(figsize=(20, 5))

核心问题分析 #

1. 全局池化丢失实例维度信息：原模型使用GlobalAveragePooling1D将每个bag的所有实例特征平均为单一向量，后续全连接层输出的5个值无法对应到具体实例——模型只能学习bag的整体统计特征，无法捕捉实例级别的差异，因此预测总数有效、单个实例无效。
2. 损失函数设计缺陷：初始使用默认binary_crossentropy时，未过滤标签中-1的无效样本，导致损失计算包含噪声；自定义损失中不必要的维度扩展可能干扰梯度传播，且用求和而非平均计算损失，会让大bag的损失权重过高。
3. 全1预测的诱因：若训练集正样本（1）占比过高，模型会偏向多数类；同时模型无实例级特征支撑预测，只能输出均值类结果。

修正方案 #

1. 重构模型结构，保留实例维度 #

移除全局池化层，直接在Transformer输出后对每个实例做独立预测，确保每个预测对应具体实例特征：

def build_model(
    input_shape,
    head_size,
    num_heads,
    ff_dim,
    num_transformer_blocks,
    dropout=0,
):
    inputs = keras.Input(shape=input_shape)
    x = keras.layers.Masking(mask_value=0)(inputs)
    for _ in range(num_transformer_blocks):
        x = transformer_encoder(x, head_size, num_heads, ff_dim, dropout)
    # 对每个实例输出单个预测值，形状为(batch, max_length, 1)
    outputs = layers.Dense(1, activation='sigmoid')(x)
    # 调整形状与标签匹配：(batch, max_length)
    outputs = layers.Reshape((max_length,))(outputs)
    return keras.Model(inputs, outputs)

Transformer的注意力机制仍能捕捉实例间的依赖关系，同时每个实例的预测都有对应的特征支撑。

2. 优化掩码损失函数 #

采用平均损失替代求和，避免bag大小影响损失权重，同时简化掩码逻辑：

def masked_binary_crossentropy(y_true, y_pred):
    # 生成有效样本掩码
    mask = tf.cast(tf.math.not_equal(y_true, -1.0), tf.float32)
    # 计算每个样本的损失（不做归约）
    bce = tf.keras.losses.BinaryCrossentropy(reduction='none')
    # 仅计算有效样本的平均损失
    loss = bce(y_true, y_pred) * mask
    return tf.reduce_sum(loss) / tf.reduce_sum(mask)

3. 调整训练策略与参数 #

• 自定义评估指标：仅计算有效样本的准确率，避免无效标签干扰评估：

def masked_binary_accuracy(y_true, y_pred):
    mask = tf.cast(tf.math.not_equal(y_true, -1.0), tf.float32)
    y_pred = tf.round(y_pred)
    correct = tf.cast(tf.equal(y_true, y_pred), tf.float32) * mask
    return tf.reduce_sum(correct) / tf.reduce_sum(mask)

编译模型时使用该指标：

model.compile(
    loss=masked_binary_crossentropy,
    optimizer=keras.optimizers.Adam(learning_rate=1e-3),
    metrics=[masked_binary_accuracy],
)

• 平衡数据与调整参数：若训练集类别不平衡，可通过过采样负样本bag或添加类别权重优化；同时调小模型参数（如head_size=32、num_transformer_blocks=2），避免过拟合与参数爆炸。

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