基于MNIST构建VAE遇TensorFlow AttributeError问题求助

运行VAE训练代码时出现如下错误：

--------------------------------------------------------------------------- AttributeError                            Traceback (most recent call last)  
1 if __name__ == "__main__":
2     x_train, _, _, _ = load_mnist()
----> 3     autoencoder = train(x_train[:10000], LEARNING_RATE, BATCH_SIZE, EPOCHS)
4     autoencoder.save("model")

5 frames
/usr/local/lib/python3.9/dist-packages/keras/engine/training_utils_v1.py
in collect_per_output_metric_info(metrics, output_names, output_shapes, loss_fns, from_serialized, is_weighted)    1040
metric, output_shape=output_shapes[i], loss_fn=loss_fns[i]    1041
)
-> 1042             metric_fn._from_serialized = from_serialized    1043     1044             # If the metric function is not stateful, we create a stateful

AttributeError: 'method' object has no attribute '_from_serialized'

相关VAE类代码如下：

class VAE:
    """
    VAE represents a Deep Convolutional variational autoencoder architecture
    with mirrored encoder and decoder components.
    """

    def __init__(self,
                 input_shape,
                 conv_filters,
                 conv_kernels,
                 conv_strides,
                 latent_space_dim):
        self.input_shape = input_shape # [28, 28, 1]
        self.conv_filters = conv_filters # [2, 4, 8]
        self.conv_kernels = conv_kernels # [3, 5, 3]
        self.conv_strides = conv_strides # [1, 2, 2]
        self.latent_space_dim = latent_space_dim # 2
        self.reconstruction_loss_weight = 1000

        self.encoder = None
        self.decoder = None
        self.model = None

        self._num_conv_layers = len(conv_filters)
        self._shape_before_bottleneck = None
        self._model_input = None

        self._build()

    def summary(self):
        self.encoder.summary()
        self.decoder.summary()
        self.model.summary()

    def compile(self, learning_rate=0.0001):
        optimizer = Adam(learning_rate=learning_rate)
        self.model.compile(optimizer=optimizer,
                          loss=self._calculate_combined_loss,
                          metrics=[self._calculate_reconstruction_loss,
                                   self._calculate_kl_loss]
                           )

    def train(self, x_train, batch_size, num_epochs):
        self.model.fit(x_train,
                       x_train,
                       batch_size=batch_size,
                       epochs=num_epochs,
                       shuffle=True)

    def save(self, save_folder="."):
        self._create_folder_if_it_doesnt_exist(save_folder)
        self._save_parameters(save_folder)
        self._save_weights(save_folder)

    def load_weights(self, weights_path):
        self.model.load_weights(weights_path)

    def reconstruct(self, images):
        latent_representations = self.encoder.predict(images)
        reconstructed_images = self.decoder.predict(latent_representations)
        return reconstructed_images, latent_representations

    @classmethod
    def load(cls, save_folder="."):
        parameters_path = os.path.join(save_folder, "parameters.pkl")
        with open(parameters_path, "rb") as f:
            parameters = pickle.load(f)
        autoencoder = VAE(*parameters)
        weights_path = os.path.join(save_folder, "weights.h5")
        autoencoder.load_weights(weights_path)
        return autoencoder

    def _calculate_combined_loss(self, y_target, y_predicted):
        reconstruction_loss = self._calculate_reconstruction_loss(y_target, y_predicted)
        kl_loss = self._calculate_kl_loss(y_target, y_predicted)
        combined_loss = self.reconstruction_loss_weight * reconstruction_loss\
                                                         + kl_loss
        return combined_loss

    def _calculate_reconstruction_loss(self, y_target, y_predicted):
        error = y_target - y_predicted
        reconstruction_loss = K.mean(K.square(error), axis=[1, 2, 3])
        return reconstruction_loss

    def _calculate_kl_loss(self, y_target, y_predicted):
        kl_loss = -0.5 * K.sum(1 + self.log_variance - K.square(self.mu) -
                               K.exp(self.log_variance), axis=1)
        return kl_loss
    
    def _create_folder_if_it_doesnt_exist(self, folder):
        if not os.path.exists(folder):
            os.makedirs(folder)

    def _save_parameters(self, save_folder):
        parameters = [
            self.input_shape,
            self.conv_filters,
            self.conv_kernels,
            self.conv_strides,
            self.latent_space_dim
        ]
        save_path = os.path.join(save_folder, "parameters.pkl")
        with open(save_path, "wb") as f:
            pickle.dump(parameters, f)

    def _save_weights(self, save_folder):
        save_path = os.path.join(save_folder, "weights.h5")
        self.model.save_weights(save_path)

    def _build(self):
        self._build_encoder()
        self._build_decoder()
        self._build_autoencoder()

    def _build_autoencoder(self):
        model_input = self._model_input
        model_output = self.decoder(self.encoder(model_input))
        self.model = Model(model_input, model_output, name="autoencoder")

    def _build_decoder(self):
        decoder_input = self._add_decoder_input()
        dense_layer = self._add_dense_layer(decoder_input)
        reshape_layer = self._add_reshape_layer(dense_layer)
        conv_transpose_layers = self._add_conv_transpose_layers(reshape_layer)
        decoder_output = self._add_decoder_output(conv_transpose_layers)
        self.decoder = Model(decoder_input, decoder_output, name="decoder")

    def _add_decoder_input(self):
        return Input(shape=self.latent_space_dim, name="decoder_input")

    def _add_dense_layer(self, decoder_input):
        num_neurons = np.prod(self._shape_before_bottleneck) # [1, 2, 4] -> 8
        dense_layer = Dense(num_neurons, name="decoder_dense")(decoder_input)
        return dense_layer

    def _add_reshape_layer(self, dense_layer):
        return Reshape(self._shape_before_bottleneck)(dense_layer)

    def _add_conv_transpose_layers(self, x):
        """Add conv transpose blocks."""
        # loop through all the conv layers in reverse order and stop at the
        # first layer
        for layer_index in reversed(range(1, self._num_conv_layers)):
            x = self._add_conv_transpose_layer(layer_index, x)
        return x

    def _add_conv_transpose_layer(self, layer_index, x):
        layer_num = self._num_conv_layers - layer_index
        conv_transpose_layer = Conv2DTranspose(
            filters=self.conv_filters[layer_index],
            kernel_size=self.conv_kernels[layer_index],
            strides=self.conv_strides[layer_index],
            padding="same",
            name=f"decoder_conv_transpose_layer_{layer_num}"
        )
        x = conv_transpose_layer(x)
        x = ReLU(name=f"decoder_relu_{layer_num}")(x)
        x = BatchNormalization(name=f"decoder_bn_{layer_num}")(x)
        return x

    def _add_decoder_output(self, x):
        conv_transpose_layer = Conv2DTranspose(
            filters=1,
            kernel_size=self.conv_kernels[0],
            strides=self.conv_strides[0],
            padding="same",
            name=f"decoder_conv_transpose_layer_{self._num_conv_layers}"
        )
        x = conv_transpose_layer(x)
        output_layer = Activation("sigmoid", name="sigmoid_layer")(x)
        return output_layer

    def _build_encoder(self):
        encoder_input = self._add_encoder_input()
        conv_layers = self._add_conv_layers(encoder_input)
        bottleneck = self._add_bottleneck(conv_layers)
        self._model_input = encoder_input
        self.encoder = Model(encoder_input, bottleneck, name="encoder")

    def _add_encoder_input(self):
        return Input(shape=self.input_shape, name="encoder_input")

    def _add_conv_layers(self, encoder_input):
        """Create all convolutional blocks in encoder."""
        x = encoder_input
        for layer_index in range(self._num_conv_layers):
            x = self._add_conv_layer(layer_index, x)
        return x

    def _add_conv_layer(self, layer_index, x):
        """Add a convolutional block to a graph of layers, consisting of
        conv 2d + ReLU + batch normalization.
        """
        layer_number = layer_index + 1
        conv_layer = Conv2D(
            filters=self.conv_filters[layer_index],
            kernel_size=self.conv_kernels[layer_index],
            strides=self.conv_strides[layer_index],
            padding="same",
            name=f"encoder_conv_layer_{layer_number}"
        )
        x = conv_layer(x)
        x = ReLU(name=f"encoder_relu_{layer_number}")(x)
        x = BatchNormalization(name=f"encoder_bn_{layer_number}")(x)
        return x

    def _add_bottleneck(self, x):
        """Flatten data and add bottleneck with Guassian sampling (Dense
        layer).
        """
        self._shape_before_bottleneck = K.int_shape(x)[1:]
        x = Flatten()(x)
        self.mu = Dense(self.latent_space_dim, name="mu")(x)
        self.log_variance = Dense(self.latent_space_dim,
                                  name="log_variance")(x)

        def sample_point_from_normal_distribution(args):
            mu, log_variance = args
            epsilon = K.random_normal(shape=K.shape(self.mu), mean=0.,
                                      stddev=1.)
            sampled_point = mu + K.exp(log_variance / 2) * epsilon
            return sampled_point

        x = Lambda(sample_point_from_normal_distribution,
                   name="encoder_output")([self.mu, self.log_variance])
        return x

已尝试相关修复方案但无效，寻求解决办法。

解决方案 #

问题根源 #

Keras在处理模型metrics参数时，无法直接接受类的实例方法（如self._calculate_reconstruction_loss）。这些方法是绑定到VAE实例的对象，而Keras期望的是无状态函数或标准的Keras Metric对象，实例方法的特殊属性会导致内部处理逻辑出错。

此外，原代码中_calculate_kl_loss直接访问实例的self.mu和self.log_variance，而非通过模型输入输出传递，不符合Keras的计算逻辑，会导致训练时的张量维度不匹配问题。

修复步骤 #

方案1：用Lambda包装实例方法（快速临时修复）

修改compile方法，将实例方法用lambda函数包装，避免直接传入绑定方法：

def compile(self, learning_rate=0.0001):
    optimizer = Adam(learning_rate=learning_rate)
    self.model.compile(optimizer=optimizer,
                      loss=self._calculate_combined_loss,
                      metrics=[
                          lambda y_true, y_pred: self._calculate_reconstruction_loss(y_true, y_pred),
                          lambda y_true, y_pred: self._calculate_kl_loss(y_true, y_pred)
                      ])

方案2：重构模型输出（规范长期修复）

这种方式更符合Keras的设计逻辑，将mu和log_variance作为模型的输出，避免依赖实例属性：

1. 修改编码器的瓶颈层，返回采样点、mu和log_variance：

def _add_bottleneck(self, x):
    self._shape_before_bottleneck = K.int_shape(x)[1:]
    x = Flatten()(x)
    mu = Dense(self.latent_space_dim, name="mu")(x)
    log_variance = Dense(self.latent_space_dim, name="log_variance")(x)

    def sample_point_from_normal_distribution(args):
        mu, log_variance = args
        epsilon = K.random_normal(shape=K.shape(mu), mean=0., stddev=1.)
        sampled_point = mu + K.exp(log_variance / 2) * epsilon
        return sampled_point

    encoder_output = Lambda(sample_point_from_normal_distribution, name="encoder_output")([mu, log_variance])
    # 返回三个输出：采样点、mu、log_variance
    return encoder_output, mu, log_variance

2. 更新编码器构建逻辑，让其返回三个输出：

def _build_encoder(self):
    encoder_input = self._add_encoder_input()
    conv_layers = self._add_conv_layers(encoder_input)
    encoder_output, mu, log_variance = self._add_bottleneck(conv_layers)
    self._model_input = encoder_input
    self.encoder = Model(encoder_input, [encoder_output, mu, log_variance], name="encoder")

3. 修改自动编码器的构建，接收编码器的三个输出：

def _build_autoencoder(self):
    model_input = self._model_input
    encoder_output, mu, log_variance = self.encoder(model_input)
    decoder_output = self.decoder(encoder_output)
    # 模型输出重建图、mu、log_variance
    self.model = Model(model_input, [decoder_output, mu, log_variance], name="autoencoder")

4. 重构compile方法，定义多输出对应的loss和总损失：

def compile(self, learning_rate=0.0001):
    optimizer = Adam(learning_rate=learning_rate)
    
    # 定义总损失：重建损失 + KL损失
    def total_loss(y_true, y_pred):
        recon_loss = K.mean(K.square(y_true - y_pred[0]), axis=[1,2,3])
        kl_loss = -0.5 * K.sum(1 + y_pred[2] - K.square(y_pred[1]) - K.exp(y_pred[2]), axis=1)
        return self.reconstruction_loss_weight * recon_loss + kl_loss

    # 为每个输出指定loss，mu和log_variance无需单独损失
    losses = {
        "decoder": lambda y_true, y_pred: 0.0,
        "mu": lambda y_true, y_pred: 0.0,
        "log_variance": lambda y_true, y_pred: 0.0
    }

    self.model.compile(
        optimizer=optimizer,
        loss=total_loss,
        metrics={"decoder": lambda y_true, y_pred: K.mean(K.square(y_true - y_pred[0]), axis=[1,2,3])}
    )

5. 更新train方法，传入对应多输出的标签：

def train(self, x_train, batch_size, num_epochs):
    # 为mu和log_variance创建dummy标签（无实际意义）
    dummy_mu = np.zeros((len(x_train), self.latent_space_dim))
    dummy_log_var = np.zeros((len(x_train), self.latent_space_dim))
    self.model.fit(
        x_train,
        [x_train, dummy_mu, dummy_log_var],
        batch_size=batch_size,
        epochs=num_epochs,
        shuffle=True
    )

验证修复 #

完成修改后，重新运行训练代码，即可避免_from_serialized属性错误，同时VAE的损失计算逻辑也更规范。

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