自定义Keras数据生成器集成训练步数异常问题求助

问题背景 #

• 基于tf.keras.utils.Sequence实现自定义数据生成器CustomVoiceAugmentation，用于加权平均集成模型训练
• 数据集规模：训练集1042样本，验证集298样本，batch size设为64
• 异常现象：
  • train_aug_gen的__len__返回值为16，val_aug_gen返回4，符合ceil(样本数/batch_size)的预期（1042/64≈16.28→16，298/64≈4.65→4）
  • 调用next(train_aug_gen)获取的训练batch长度为2，而非预期的64
  • 使用KerasMember训练时，日志显示训练步数为2/2，而非预期的16/16

可能原因分析 #

1. __getitem__返回结构错误
   Keras要求Sequence的__getitem__(idx)返回格式为(inputs, targets)，如果返回的是未嵌套的多输入/多输出（比如(x1, x2, y1, y2)），Keras会误将外层元素数量识别为batch size。
2. __len__逻辑或训练参数覆盖
   即使__len__计算正确，若训练时手动指定了steps_per_epoch=2，会直接覆盖生成器的长度；也可能__len__实现未正确取整（需确认代码逻辑）。
3. 自定义__next__方法逻辑错误
   若重写了__next__，可能错误固定了返回的样本数量，导致每次仅返回2个样本。
4. KerasMember对生成器适配问题
   第三方集成框架的KerasMember可能强制设置了步数参数，未正确读取生成器的__len__结果。

解决方案 #

1. 修正__getitem__返回结构 #

确保__getitem__返回的是**(输入集合, 标签集合)**，其中输入/标签为单数组或嵌套列表，每个元素的第一维度为batch size（最后一个batch可小于64）。示例代码：

import math
import numpy as np
import tensorflow as tf

class CustomVoiceAugmentation(tf.keras.utils.Sequence):
    def __init__(self, samples, batch_size=64, ...):
        self.samples = samples
        self.batch_size = batch_size
        self.indices = np.arange(len(self.samples))
        # 其他初始化逻辑（如数据增强配置）

    def __getitem__(self, idx):
        # 提取当前batch的样本索引
        batch_idx = self.indices[idx*self.batch_size : (idx+1)*self.batch_size]
        # 加载当前batch的输入和标签
        batch_x = np.array([self.load_input(sample) for sample in self.samples[batch_idx]])
        batch_y = np.array([self.load_label(sample) for sample in self.samples[batch_idx]])
        # 多输入场景示例：return ([batch_x1, batch_x2], batch_y)
        return (batch_x, batch_y)

    def __len__(self):
        return math.ceil(len(self.samples) / self.batch_size)

2. 取消手动设置的steps_per_epoch #

训练时不要手动指定steps_per_epoch或validation_steps，让Keras自动从生成器的__len__方法获取步数：

# 错误示例：手动固定步数
member.train(train_data=train_aug_gen, steps_per_epoch=2, validation_data=val_aug_gen, validation_steps=1)

# 正确示例：依赖生成器自动计算步数
member.train(train_data=train_aug_gen, validation_data=val_aug_gen)

3. 移除或修正自定义__next__方法 #

若生成器重写了__next__，建议删除该方法，使用Sequence默认实现。若必须自定义，确保逻辑正确遍历所有batch：

def __next__(self):
    if self.current_index >= len(self):
        self.current_index = 0
        self.on_epoch_end()  # 处理epoch结束后的洗牌等操作
    batch = self.__getitem__(self.current_index)
    self.current_index += 1
    return batch

4. 验证生成器输出正确性 #

通过以下代码验证生成器的batch结构和总样本数：

# 检查第一个batch的形状
first_batch_x, first_batch_y = train_aug_gen[0]
print(f"First batch input shape: {first_batch_x.shape}")  # 应输出(64, ...)
print(f"First batch label shape: {first_batch_y.shape}")  # 应输出(64, ...)

# 统计生成器输出的总样本数
total_samples = 0
for x, y in train_aug_gen:
    total_samples += x.shape[0]
print(f"Total samples from generator: {total_samples}")  # 应输出1042

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