自定义损失函数的多输出自编码器TypeError问题求助

开发一款利用待重建灰度图像标签的自编码器时，自定义损失函数触发如下错误：

TypeError: Keras symbolic inputs/outputs do not implement __len__.
You may be trying to pass Keras symbolic inputs/outputs to a TF API
that does not register dispatching, preventing Keras from
automatically converting the API call to a lambda layer in the
Functional Model. This error will also get raised if you try asserting
a symbolic input/output directly.

错误产生于return total_loss语句，已尝试禁用eager执行、替换为tf.math操作，均无效。

代码示例 #

import tensorflow as tf
from tensorflow.keras.callbacks import EarlyStopping
from tensorflow import keras


def joint_loss(imgs_true, imgs_pred, y_true, y_pred, reconstruction_weight, 
               classification_weight): 
  # imgs_true = original images (= keras.input)
  # imgs_pred = reconstructed images of my autoencoder
  # y_true = true labels of my data (= keras.input)
  # y_pred = predicted labels from my bottleneck layer
  # reconstruction_weight/classification_weight = explanation below in "hyperparameters"
  reconstruction_loss = tf.reduce_mean(tf.square(imgs_true - imgs_pred))
  classification_loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(y_true, y_pred))
  total_loss = (tf.math.scalar_mul(reconstruction_weight, reconstruction_loss) + 
            tf.math.scalar_mul(classification_weight, classification_loss))
  
  return total_loss



# define function of the autoencoder that's to be optimized; give back validation loss
def create_and_train_autoencoder(encoding_dim, f1, f2, f3, f4, f5, k1, k2, 
                                 num_epochs, bat_size, learning_rate, 
                 recon_weight, class_weight):
    # explanation for parameters in "hyperparameter" below

    # input for images and labels
    input_img = keras.Input(shape=(64, 128, 1))
    input_label = keras.Input(shape=(1,))

    # Encoding layers
    x = keras.layers.Conv2D(f1, (k1, k2), activation='relu', padding='same')(input_img)
    x = keras.layers.MaxPooling2D((2, 2), padding='same')(x)
    x = keras.layers.Conv2D(f2, (k1, k2), activation='relu', padding='same')(x)
    x = keras.layers.MaxPooling2D((2, 2), padding='same')(x)
    x = keras.layers.Conv2D(f3, (k1, k2), activation='relu', padding='same')(x)
    x = keras.layers.MaxPooling2D((2, 2), padding='same')(x)


    # Bottleneck
    encoded = keras.layers.Conv2D(encoding_dim, (k1, k2), activation='relu', 
                                  padding='same')(x)


    # Define the classification branch
    encoded_flattened = keras.layers.Flatten()(encoded)
    encoded_flattened_dense1 = keras.layers.Dense(f4, activation='relu')(encoded_flattened)
    encoded_flattened_dense2 = keras.layers.Dense(f5, activation='relu')(encoded_flattened_dense1)
    label_output = keras.layers.Dense(1, activation='sigmoid')(encoded_flattened_dense2)


    # Decoding layers
    x = keras.layers.UpSampling2D((2, 2))(encoded)
    x = keras.layers.Conv2D(f3, (k1, k2), activation='relu', padding='same')(x)
    x = keras.layers.UpSampling2D((2, 2))(x)
    x = keras.layers.Conv2D(f2, (k1, k2), activation='relu', padding='same')(x)
    x = keras.layers.UpSampling2D((2, 2))(x)
    x = keras.layers.Conv2D(f1, (k1, k2), activation='relu', padding='same')(x)
    decoded = keras.layers.Conv2D(1, (k1, k2), activation='sigmoid', padding='same')(x)


    # create model and compile
    autoencoder = keras.Model([input_img, input_label], [decoded, label_output])

    autoencoder.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate), 
                    loss=joint_loss(input_img, decoded, input_label, label_output,
                                    recon_weight, class_weight))
    
    
    # Create early stopping callback
    early_stopping = EarlyStopping(monitor='val_loss', patience=15, 
                                   restore_best_weights=True)
    
    
    # fit
    history = autoencoder.fit([x_train, y_train], [x_train, y_train],
                epochs=num_epochs,
                batch_size= bat_size,
                shuffle=True,
                validation_data=([x_test, y_test], [x_test, y_test]),
                callbacks=[early_stopping])

    
    return history.history['val_loss'][-1]


# execute the function
best_autoencoder = create_and_train_autoencoder(
    encoding_dim, f1, f2, f3, f4, f5, k1, k2, num_epochs, bat_size, 
    learning_rate, reconstr_weight, class_weight)

背景信息 #

• 数据集：训练集51张(64,128)灰度图（数组形状(51,64,128)），测试集13张同规格灰度图；标签为[0,1]，训练集标签数组形状(51,)，测试集为(13,)
• 超参数：
  • encoding_dim=14
  • f1=16，f2=f3=8，f4=f5=64
  • k1=k2=3
  • num_epochs=25，bat_size=32
  • learning_rate=0.0001
  • reconstr_weight=0.5，class_weight=0.5
• 环境：TensorFlow 2.7版本，未使用CUDA

错误根源是在compile时直接传递符号张量给自定义损失函数，Keras要求自定义损失函数必须符合loss(y_true, y_pred)的签名格式，多输出模型的损失需适配输出结构。以下是修复步骤：

1. 重构自定义损失函数 #

将损失函数改为带权重参数的闭包，内部函数匹配Keras要求的签名，同时修正分类损失的计算冲突：

def joint_loss(reconstruction_weight, classification_weight):
    def loss(y_true, y_pred):
        # y_true对应模型输入的真实值：[真实图像, 真实标签]
        # y_pred对应模型输出的预测值：[重建图像, 预测标签（logits）]
        imgs_true, labels_true = y_true
        imgs_pred, labels_pred = y_pred
        
        # 计算重建损失
        reconstruction_loss = tf.reduce_mean(tf.square(imgs_true - imgs_pred))
        
        # 计算分类损失（要求输入为logits，需配合分类分支的激活修改）
        classification_loss = tf.reduce_mean(tf.nn.sigmoid_cross_entropy_with_logits(labels_true, labels_pred))
        
        # 加权求和总损失
        total_loss = reconstruction_weight * reconstruction_loss + classification_weight * classification_loss
        return total_loss
    return loss

2. 修改分类分支的激活函数 #

原代码中分类分支用了sigmoid激活，会与sigmoid_cross_entropy_with_logits冲突（该函数要求输入为未经过激活的logits），需改为：

# 替换原label_output的定义
label_output = keras.layers.Dense(1, activation=None)(encoded_flattened_dense2)

3. 调整模型编译方式 #

调用闭包生成损失函数，无需传递符号张量：

autoencoder.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate), 
    loss=joint_loss(recon_weight, class_weight)
)

4. 修正输入输出的形状匹配 #

确保数据集形状与模型输入输出一致：

# 训练时调整数据形状
history = autoencoder.fit(
    [x_train.reshape(-1,64,128,1), y_train.reshape(-1,1)],
    [x_train.reshape(-1,64,128,1), y_train.reshape(-1,1)],
    epochs=num_epochs,
    batch_size=bat_size,
    shuffle=True,
    validation_data=(
        [x_test.reshape(-1,64,128,1), y_test.reshape(-1,1)], 
        [x_test.reshape(-1,64,128,1), y_test.reshape(-1,1)]
    ),
    callbacks=[early_stopping]
)

内容的提问来源于stack exchange，提问作者T.A. Anderson