如何在model.fit中使用多训练/验证生成器？多输入模型报错排查

问题原因 #

直接用zip()组合多个ImageDataGenerator生成器的方式错误，导致输入数据结构不符合Keras要求：

• 单个ImageDataGenerator生成器（如flow_from_dataframe返回的对象）每次迭代会输出(单输入图像数组, 单输出标签数组)的元组。
• 用zip(gen1, gen2, gen3, gen4)组合后，每个批次的结构变为((x1,y1), (x2,y2), (x3,y3), (x4,y4))，而Keras要求的输入格式是**(多输入列表/元组, 多输出列表/元组)**，即([x1,x2,x3,x4], [y1,y2,y3,y4])，因此触发ValueError。

正确实现方式 #

方法1：自定义包装生成器 #

这是最直接适配ImageDataGenerator的方案，步骤如下：

1. 先为4个数据集分别创建对应的训练/验证生成器，确保所有生成器的batch_size、target_size、预处理逻辑完全一致：

from tensorflow.keras.preprocessing.image import ImageDataGenerator

datagen = ImageDataGenerator(rescale=1./255)

# 美学数据集1的训练生成器（回归任务用class_mode='raw'）
train_gen_aes1 = datagen.flow_from_dataframe(
    dataframe=df_aes1,
    directory='path/to/aesthetic1_images',
    x_col='image_path',
    y_col='aesthetic_score',
    target_size=(224,224),
    batch_size=32,
    class_mode='raw'
)

# 同理创建train_gen_aes2、train_gen_qual1、train_gen_qual2，以及验证生成器val_gen_*

2. 编写包装生成器，将4个生成器的输入和输出分别打包成符合要求的列表：

def combined_generator(*generators):
    while True:
        # 从每个生成器获取批次数据
        batches = [next(gen) for gen in generators]
        # 提取所有输入数组和输出数组
        x_inputs = [batch[0] for batch in batches]
        y_outputs = [batch[1] for batch in batches]
        yield (x_inputs, y_outputs)

3. 生成训练/验证用的组合生成器：

train_combined = combined_generator(train_gen_aes1, train_gen_aes2, train_gen_qual1, train_gen_qual2)
val_combined = combined_generator(val_gen_aes1, val_gen_aes2, val_gen_qual1, val_gen_qual2)

4. 模型训练时，指定steps_per_epoch和validation_steps（取值为单个生成器的样本数//batch_size）：

steps_per_epoch = train_gen_aes1.samples // train_gen_aes1.batch_size
validation_steps = val_gen_aes1.samples // val_gen_aes1.batch_size

model.fit(
    train_combined,
    steps_per_epoch=steps_per_epoch,
    validation_data=val_combined,
    validation_steps=validation_steps,
    epochs=50
)

方法2：用tf.data.Dataset组合生成器 #

如果使用TensorFlow 2.x，也可以通过tf.data.Dataset来包装并组合生成器，代码更简洁且支持自动优化：

import tensorflow as tf

# 将每个ImageDataGenerator生成器转为tf.data.Dataset
ds_aes1 = tf.data.Dataset.from_generator(
    lambda: train_gen_aes1,
    output_types=(tf.float32, tf.float32)
)
ds_aes2 = tf.data.Dataset.from_generator(lambda: train_gen_aes2, output_types=(tf.float32, tf.float32))
ds_qual1 = tf.data.Dataset.from_generator(lambda: train_gen_qual1, output_types=(tf.float32, tf.float32))
ds_qual2 = tf.data.Dataset.from_generator(lambda: train_gen_qual2, output_types=(tf.float32, tf.float32))

# 组合数据集并重构输入输出结构
combined_ds = tf.data.Dataset.zip((ds_aes1, ds_aes2, ds_qual1, ds_qual2)).map(
    lambda aes1, aes2, qual1, qual2: (
        [aes1[0], aes2[0], qual1[0], qual2[0]],  # 多输入列表
        [aes1[1], aes2[1], qual1[1], qual2[1]]   # 多输出列表
    )
)

# 添加预取优化
combined_ds = combined_ds.prefetch(tf.data.AUTOTUNE)

# 验证集同理创建val_combined_ds

训练时直接传入combined_ds，无需指定steps_per_epoch（TensorFlow会自动计算）：

model.fit(
    combined_ds,
    validation_data=val_combined_ds,
    epochs=50
)

注意事项 #

• 所有生成器的batch_size必须完全一致，否则会出现批次维度不匹配的问题。
• 若4个数据集的样本数量不一致，可将样本数裁剪至相同，或在自定义生成器中用itertools.cycle循环补充样本较少的生成器。
• 模型编译时需为每个输出指定对应的损失函数和指标（以回归任务为例）：

model.compile(
    optimizer='adam',
    loss={
        'output_aesthetic1': 'mean_squared_error',
        'output_aesthetic2': 'mean_squared_error',
        'output_quality1': 'mean_squared_error',
        'output_quality2': 'mean_squared_error'
    },
    metrics={
        'output_aesthetic1': ['mean_absolute_error'],
        'output_aesthetic2': ['mean_absolute_error'],
        'output_quality1': ['mean_absolute_error'],
        'output_quality2': ['mean_absolute_error']
    }
)

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