如何在tf.keras回调中修改神经网络输入并传递参数到损失函数

1. 报错原因分析 #

你当前写法的错误在于：self.model.layers[0]是Keras的InputLayer层对象，既不是输入数据也不是可运算的张量，自然不能直接和整数做乘法运算，所以抛出了类型不匹配的错误。
另外回调本身没有直接修改fit方法传入训练数据的原生能力，要实现替换输入的需求需要调整实现逻辑。

2. 实现随机替换输入为dummy矩阵的正确方案 #

推荐两种适配性强的方案，按需选择：

方案1：使用自定义训练数据生成器 #

在数据生成阶段判断当前epoch状态，决定是否返回dummy数据，同时记录标记供损失函数使用：

import numpy as np
import tensorflow as tf
from tensorflow import keras

# 全局标记变量，也可封装到生成器类中避免全局污染
current_set_zero_flag = 0
total_epochs = 300

def data_generator(x, y, batch_size):
    global current_set_zero_flag
    for epoch in range(total_epochs):
        # 每个epoch开始前随机生成置零标记
        current_set_zero_flag = np.random.choice([0, 1])
        # 打乱数据集
        idx = np.random.permutation(len(x))
        x_shuffled = x[idx]
        y_shuffled = y[idx]
        for i in range(0, len(x), batch_size):
            batch_x = x_shuffled[i:i+batch_size]
            batch_y = y_shuffled[i:i+batch_size]
            if current_set_zero_flag == 1:
                # 替换为自定义dummy矩阵，这里示例是全零矩阵，可修改为你需要的数值
                batch_x = np.zeros_like(batch_x)
            yield batch_x, batch_y

调用fit时传入生成器即可：

history = model.fit(
    data_generator(x_train, y_train_n, batch_size=10),
    epochs=total_epochs,
    steps_per_epoch=len(x_train)//10,
    validation_split=0.2,
    shuffle=False
)

方案2：模型内部增加Lambda掩码层（更简洁） #

把输入掩码逻辑放到模型结构内部，通过可更新的TF变量控制掩码状态，回调仅负责更新变量值：

# 定义可更新的置零控制变量，设置为不可训练
set_zero_input = tf.Variable(0, trainable=False, dtype=tf.float32)

# 构建模型时在输入层后新增掩码层
input = keras.Input(shape=你的输入维度)
masked_input = keras.layers.Lambda(lambda x: x * (1 - set_zero_input))(input)
# 后面拼接你原本的Autoencoder结构即可
# ... 其余层定义 ...
model = keras.Model(inputs=input, outputs=output)

# 自定义回调仅负责更新控制变量
class MyCustomCallback_zeroing(tf.keras.callbacks.Callback):
    def on_epoch_begin(self, epoch, logs=None):
        set_zero_val = np.random.choice([0, 1])
        set_zero_input.assign(set_zero_val)

调用fit时正常传入回调即可。

3. 标记传递给损失函数的实现 #

直接返回变量的方式不可行，你可以通过TF全局变量的方式实现值的传递，以上述方案2为例，自定义损失函数时直接引用控制变量即可，变量更新后损失计算逻辑会自动同步：

def custom_loss(y_true, y_pred):
    # 直接获取当前置零标记
    zero_flag = set_zero_input
    # 你可以根据标记自定义损失逻辑，示例为置零阶段损失乘以0.5，正常训练阶段用原损失
    base_loss = keras.losses.MSE(y_true, y_pred)
    adjusted_loss = base_loss * (1 - zero_flag * 0.5)
    return adjusted_loss

# 编译模型时指定自定义损失
model.compile(optimizer='adam', loss=custom_loss)

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