使用Keras和Kedro构建自编码器时出现AssertionError如何解决

你遇到的报错是Keras计算图断裂导致的，核心有三类问题：

• 预训练阶段输出的pre_train_autoencoder、pre_train_encoder、pre_train_decoder三个模型没有在catalog.yaml中配置对应的TensorFlow序列化存储规则，Kedro默认用内存存储跨节点传递的Keras模型时，会丢失TensorFlow计算图的上下文关联，导致训练节点无法识别模型的张量结构。
• 解码器decoder的结构定义错误，你当前的解码器输入和自编码器输入一样都是原始特征层，不符合解码器接收latent向量输入的设计，破坏了模型计算图的逻辑关联。
• 代码存在参数拼写错误、类型标注不匹配的隐藏问题。

1. 修正自编码器三组件的结构定义 #

修改nodes.py中build_models函数的模型构建部分，确保解码器是独立接收latent向量输入的结构，同时修正参数拼写、返回值类型标注：

from typing import Dict
from tensorflow import keras # 确保导入正确

def build_models(data: pd.DataFrame, n_hidden_layers: int, latent_space_size: int, regularization_strength: float, seed: int) -> Dict[str, keras.Model]:
    assert n_hidden_layers >= 1, "There must be at least 1 hidden layer for the autoencoder"
    
    n_features = data.shape[1]
    tf.random.set_seed(seed)
    # 编码器结构
    input_layer = keras.Input(shape=(n_features,))
    hidden = keras.layers.Dense(n_features, kernel_regularizer=keras.regularizers.l1(regularization_strength))(input_layer)
    hidden = keras.layers.LeakyReLU()(hidden)
    
    for _ in range(n_hidden_layers - 1):
        hidden = keras.layers.Dense(n_features, kernel_regularizer=keras.regularizers.l1(regularization_strength))(hidden)
        hidden = keras.layers.LeakyReLU()(hidden)
    encoded = keras.layers.Dense(latent_space_size, activation="sigmoid")(hidden)
    encoder = keras.models.Model(inputs=input_layer, outputs=encoded, name="encoder")

    # 解码器结构（独立输入为latent向量）
    latent_input = keras.Input(shape=(latent_space_size,))
    hidden_dec = keras.layers.Dense(n_features, kernel_regularizer=keras.regularizers.l1(regularization_strength))(latent_input)
    hidden_dec = keras.layers.LeakyReLU()(hidden_dec)
    for _ in range(n_hidden_layers - 1):
        hidden_dec = keras.layers.Dense(n_features, kernel_regularizer=keras.regularizers.l1(regularization_strength))(hidden_dec)
        hidden_dec = keras.layers.LeakyReLU()(hidden_dec)
    decoded = keras.layers.Dense(n_features, activation="sigmoid")(hidden_dec)
    decoder = keras.models.Model(inputs=latent_input, outputs=decoded, name="decoder")

    # 完整自编码器
    autoencoder = keras.models.Model(inputs=input_layer, outputs=decoder(encoder(input_layer)), name="autoencoder")
    autoencoder.compile(optimizer="adam", loss="mean_absolute_error")

    return dict(
        pre_train_autoencoder=autoencoder,
        pre_train_encoder=encoder,
        pre_train_decoder=decoder
    )

2. 补充预训练模型的catalog配置 #

在catalog.yaml中添加三个预训练模型的序列化配置，确保Kedro跨节点传递时计算图完整：

pre_train_autoencoder:
  type: kedro.extras.datasets.tensorflow.TensorFlowModelDataset
  filepath: data/06_models/pre_train_autoencoder.h5

pre_train_encoder:
  type: kedro.extras.datasets.tensorflow.TensorFlowModelDataset
  filepath: data/06_models/pre_train_encoder.h5

pre_train_decoder:
  type: kedro.extras.datasets.tensorflow.TensorFlowModelDataset
  filepath: data/06_models/pre_train_decoder.h5

3. 可选：优化输入数据类型兼容 #

如果修改后仍有输入兼容问题，可以在train_autoencoder的fit步骤显式将输入转为numpy数组：

history = autoencoder.fit(
    np.asarray(train_x), np.asarray(train_x),
    validation_data=(np.asarray(test_x), np.asarray(test_x)),
    callbacks=callbacks, 
    epochs=epochs,
    batch_size=batch_size
)

内容的提问来源于stack exchange，提问作者João Areias