重复创建tensorflow自定义模型实例并在循环内训练会产生错误

Question

我已经使用tensorflow here中的示例创建了自定义模型类。然后，我尝试获取自定义模型摘要，因此我搜索了here。但是当我试图在for循环中训练自定义模型时，出现了一个问题。第一次迭代总是成功的，但随后的迭代崩溃，并显示以下错误消息( CustomModel的输入张量必须来自tf.keras.InputReceived: None)

在调试过程中，我发现在第一次迭代中，super(CustomModel, self).__init__()调用Model.__init__()，而不调用Functional.__init__()。之后，super(CustomModel, self).__init__(inputs=self.input_layer, outputs=self.out)调用Functional.__init__()。

但在第二次迭代中，super(CustomModel, self).__init__()会立即调用Functional.__init__()。

我如何在第二次迭代中训练这个自定义模型？下面是我的代码：

import os
import numpy as np

os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'
import tensorflow as tf  # noqa
from tensorflow import keras  # noqa
from enum import Enum, auto  # noqa

BATCH_SIZE = 20


class Algo(Enum):
    linearReg = auto()

class CustomModel(keras.Model):
    def __init__(self, input_shape, algo=Algo.linearReg.value, **kwargs):
        super(CustomModel, self).__init__()
        self.in_shape = input_shape
        self.algo = algo
        # create input layer for model summary
        self.input_layer = keras.layers.Input((self.in_shape[1],))

        self.custom_layer = [keras.layers.experimental.preprocessing.Normalization()]
        if algo == Algo.linearReg.value:
            self.custom_layer.append(keras.layers.Dense(units=1, activation='linear'))

        # get output layer with call method for model summary
        self.out = self.call(self.input_layer)

        # reinitialize with input layer and output
        super(CustomModel, self).__init__(
            inputs=self.input_layer,
            outputs=self.out)

    def call(self, input_tensor, **kwargs):
        x = input_tensor
        for layer in self.custom_layer:
            x = layer(x)
        return x

    def predict(self, x, **kwargs):
        for layer in self.custom_layer:
            x = layer.call(x)
        return x

    def from_config(self, config, custom_objects=None):
        super(CustomModel, self).__init__()

    def get_config(self):
        config = super(CustomModel, self).get_config()
        return config


features, labels = (np.random.sample((100, 2, BATCH_SIZE)), np.random.sample((100, 1, BATCH_SIZE)))
dataset = tf.data.Dataset.from_tensor_slices((features, labels))
x, y = next(iter(dataset))

for i in range(4):
    model = CustomModel(x.shape, Algo.linearReg.value)
    model.summary()
    model.compile(optimizer=keras.optimizers.Ftrl(learning_rate=0.01),
                                                  loss='mse', metrics=['mae'])

    model.fit(dataset, epochs=int(2), verbose=2,
              validation_data=dataset.shuffle(2).take(1))
    y_predict = model.predict(x)

Answer 1

嗯，进一步的分析让我很头疼。但是，当我使用输入和输出重新初始化时，似乎出现了这个问题。所以我找到了替代的解决方案，尽管我不喜欢model.model()部分。

class CustomModel(keras.Model):
    def __init__(self, input_shape, algo=Algo.linearReg.value, **kwargs):

        super(CustomModel, self).__init__()
        self.in_shape = input_shape
        self.algo = algo
        # create input layer for model summary
        self.input_layer = keras.layers.Input((self.in_shape[1],))

        self.custom_layer = [keras.layers.experimental.preprocessing.Normalization()]
        if algo == Algo.linearReg.value:
            self.custom_layer.append(keras.layers.Dense(units=1, activation='linear'))

        # get output layer with call method for model summary
        self.out = self.call(self.input_layer)

    def model(self):
        return keras.Model(inputs=self.input_layer, 
                           outputs=self.call(self.input_layer))

    def call(self, input_tensor, **kwargs):
        for layer in self.custom_layer:
            input_tensor = layer(input_tensor)
        return input_tensor

    def predict(self, input_tensor, **kwargs):
        for layer in self.custom_layer:
            input_tensor = layer.call(input_tensor)
        return input_tensor

    def from_config(self, config, custom_objects=None):
        super(CustomModel, self).__init__()

    def get_config(self):
        config = super(CustomModel, self).get_config()
        return config


features, labels = (np.random.sample((100, 2, BATCH_SIZE)), np.random.sample((100, 1, 
BATCH_SIZE)))
dataset = tf.data.Dataset.from_tensor_slices((features, labels))
x, y = next(iter(dataset))

for i in range(4):
    model = CustomModel(x.shape, Algo.linearReg.value)
    model.model().summary()
    model.compile(optimizer=keras.optimizers.Ftrl(learning_rate=0.01),
              loss='mse', metrics=['mae'])

    model.fit(dataset, epochs=int(2), verbose=2,
          validation_data=dataset.shuffle(2).take(1))
    y_predict = model.predict(x)