自定义tf.keras.Model接入Keras Generator时call方法的批次处理问题

你完全搞错了Keras对接Sequence生成器的调用逻辑，根本不需要在自定义模型的方法里手动拆解生成器：
model.fit()接收Sequence类生成器时，会全自动完成整套调度流程，和手动写迭代循环没有任何区别，且完全保留生成器逐epoch打乱、逐批次加载的特性：

1. 每个epoch启动时自动调用生成器的on_epoch_end方法，重排样本索引
2. 按批次索引依次调用生成器的__getitem__，拿到单批次的(x_batch, y_batch)张量
3. 把仅当前批次的输入x张量传入模型做前向计算，把对应的y张量交给损失函数做反向传播
   你之前的写法相当于把整个生成器实例直接当成批次张量传给模型，当然会报错。而且你重写了__call__方法，而不是按照Keras规范写call方法，这会直接破坏Keras的内置训练流程。

因为你需要基于真实标签y构造decoder的移位输入（也就是seq2seq的teacher forcing逻辑），而Keras标准流程下call方法只能拿到x、拿不到对应批次的y，有两种适配方案，都不需要打断生成器的自动调度。

方案1：最小改动，把decoder输入构造挪到生成器层（最推荐） #

不需要改模型的训练流程，只需要让生成器在返回批次数据时，把构造好的decoder输入和原有的x打包成模型的输入，模型的call方法只需要正常接收输入张量就行：

1. 修改DataGenerator的__getitem__返回值，把decoder移位输入作为x的一部分：

   def __getitem__(self, index):
       indexes = self.indexes[index * self.batch_size:(index + 1) * self.batch_size] 
       list_IDs_temp = [self.list_IDs[k] for k in indexes]         
       x, y = self.__data_generation(list_IDs_temp)
       # 原有构造decoder输入的逻辑从模型里挪到这里
       dec_inputs = np.concatenate((np.zeros_like(y[:, :1, :]), y[:, :-1, :]), axis=1)  
       dec_inputs = dec_inputs[:, :, -hp.n_mels:] 
       # 返回格式：(模型接收的输入元组, 真实标签y)
       return (x, dec_inputs), y
   

2. 修正MyModel的实现，把原来重写的__call__换成标准的call方法，直接接收打包好的输入：

   class MyModel(tf.keras.Model):
       def __init__(self):
           super(MyModel, self).__init__()
           self.enc = Encoder()
           self.dec1 = Decoder1()
           self.dec2 = Decoder2()
   
       # 前向逻辑全部写在call方法里，入参就是生成器返回的输入元组
       def call(self, inputs, training=None, **kwargs):
           x_train, dec_inputs = inputs
           # 后面的逻辑和你之前写的完全一致
           print("########ENC INPUTS #####")
           print("######################")
           print("#########DEC INPUTS #####")
           print("######################")
           memory = self.enc(x_train)
           y_hat = self.dec1(dec_inputs, memory)
           return y_hat
   

   这套方案下，你原来的生成器逻辑、model.fit的调用代码完全不用改，逐epoch自动打乱、后台加载批次的特性100%保留。

方案2：不想改动生成器的话，重写模型的训练步骤 #

如果你不想动现有生成器的返回格式，那就把需要用到y的逻辑放到自定义的训练步骤里，Keras会自动把生成器拆出来的单批次(x,y)传入这个方法，完全不影响生成器的调度：

class MyModel(tf.keras.Model):
    def __init__(self):
        super(MyModel, self).__init__()
        self.enc = Encoder()
        self.dec1 = Decoder1()
        self.dec2 = Decoder2()

    def call(self, x_train, training=None, **kwargs):
        # 基础前向只做encoder编码
        return self.enc(x_train)

    def train_step(self, data):
        # Keras自动把生成器取到的单批次(x,y)传到这里
        x, y = data
        with tf.GradientTape() as tape:
            # 原有构造decoder输入的逻辑放在这里
            dec_inputs = tf.concat((tf.zeros_like(y[:, :1, :]), y[:, :-1, :]), 1)  
            dec_inputs = dec_inputs[:, :, -hp.n_mels:] 
            # 跑完整前向
            memory = self(x, training=True)
            y_hat = self.dec1(dec_inputs, memory, training=True)
            # 计算损失
            loss = self.compiled_loss(y, y_hat)
        # 计算梯度、更新参数，和Keras内置逻辑完全一致
        gradients = tape.gradient(loss, self.trainable_variables)
        self.optimizer.apply_gradients(zip(gradients, self.trainable_variables))
        # 更新指标
        self.compiled_metrics.update_state(y, y_hat)
        return {m.name: m.result() for m in self.metrics}

这套方案下你连生成器的代码都不用改，model.fit(training_generator, epochs=2)的调用逻辑和原来完全一样。

• 不要在自定义模型内部手动写循环去迭代生成器取数据，这会完全打乱Keras的训练调度，也会让Sequence生成器的多线程数据加载、epoch末尾自动打乱的特性失效
• 所有自定义前向逻辑必须写在call方法里，不要重写__call__，否则会破坏Keras的自动梯度追踪、损失计算、指标统计的全流程
• 生成器返回的x、y维度要和模型的输入输出严格匹配，避免出现维度不匹配的报错

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