TF.Keras 中自定义 Scratch 训练中的多输出多类分类

我想训练一个多输出和多类别从头开始分类模型（使用自定义fit()）。我想要一些建议。为了提供学习机会，我在这里更详细地演示整个场景。希望它对任何人都有帮助。

数据集和目标

我正在使用来自的数据here https://www.kaggle.com/c/bengaliai-cv19;这是一个孟加拉语手写字符识别挑战赛，每个样本都有3相互相关的输出以及多个班级每个。请看下图：

在上图中，正如你所看到的，ক্ট্রো由3个组件组成(ক্ট , ো , ‍‍্র)，即字素根, 元音变音符号 and 辅音变音符号分别和一起称为Grapheme。再次字素根也有168不同类别，也与其他类别相同（11 and 7）。增加的复杂性导致~13,000不同的字素变体（与英语的 250 个字素单位相比）。

目标是对字素的组成部分在每个图像中。

初步方法（没有问题）

我实施了一个培训管道here https://www.kaggle.com/ipythonx/keras-grapheme-gridmask-augmix-ensemble，其中使用旧的进行了演示keras (not tf.keras）及其方便的功能，例如model.compile, callbacks等等我定义了一个自定义数据生成器 https://www.kaggle.com/ipythonx/keras-grapheme-gridmask-augmix-ensemble#Grapheme-Data-Generator并定义了如下所示的模型架构。

input_tensor = Input(input_dim)
curr_output = base_model(input_tensor)

oputput1 = Dense(168,  activation='softmax', name='gra') (curr_output)
oputput2 = Dense(11,   activation='softmax', name='vow') (curr_output)
oputput3 = Dense(7,    activation='softmax', name='cons') (curr_output)
output_tensor = [oputput1, oputput2, oputput3]
    
model = Model(input_tensor, output_tensor)

并编译模型如下：

model.compile(

        optimizer = Adam(learning_rate=0.001), 

        loss = {'gra' : 'categorical_crossentropy', 
                'vow' : 'categorical_crossentropy', 
                'cons': 'categorical_crossentropy'},

        loss_weights = {'gra' : 1.0,
                        'vow' : 1.0,
                        'cons': 1.0},

        metrics={'gra' : 'accuracy', 
                 'vow' : 'accuracy', 
                 'cons': 'accuracy'}
    )

正如你所看到的，我可以通过特定的方式清楚地控制每个输出loss, loss_weights, and accuracy。并使用.fit()方法，使用任何方法都是可行的callbacks模型的函数。

新方法（以及一些问题）

现在，我想用新功能重新实现它tf.keras。例如模型子类化 and 定制贴合训练。但是，数据加载器没有变化。该模型定义如下：

    def __init__(self, dim):
        super(Net, self).__init__()
        self.efnet  = EfficientNetB0(input_shape=dim,
                                     include_top = False, 
                                     weights = 'imagenet')
        self.gap     = KL.GlobalAveragePooling2D()
        self.output1 = KL.Dense(168,  activation='softmax', name='gra')
        self.output2 = KL.Dense(11,   activation='softmax', name='vow') 
        self.output3 = KL.Dense(7,    activation='softmax', name='cons') 
    
    def call(self, inputs, training=False):
        x     = self.efnet(inputs)
        x     = self.gap(x)
        y_gra = self.output1(x)
        y_vow = self.output2(x)
        y_con = self.output3(x)
        return [y_gra, y_vow, y_con]

现在我面临的主要问题是正确定义metrics, loss, and loss_weights我的每个输出的函数。然而，我是这样开始的：

optimizer        = tf.keras.optimizers.Adam(learning_rate=0.05)
loss_fn          = tf.keras.losses.CategoricalCrossentropy(from_logits=True)
train_acc_metric = tf.keras.metrics.Accuracy()

@tf.function
def train_step(x, y):
    with tf.GradientTape(persistent=True) as tape:
        logits = model(x, training=True)  # Logits for this minibatch
        train_loss_value = loss_fn(y, logits)

    grads = tape.gradient(train_loss_value, model.trainable_weights)
    optimizer.apply_gradients(zip(grads, model.trainable_weights))
    train_acc_metric.update_state(y, logits)
    return train_loss_value


for epoch in range(2):
    # Iterate over the batches of the dataset.
    for step, (x_batch_train, y_batch_train) in enumerate(train_generator):
        train_loss_value = train_step(x_batch_train, y_batch_train)

    # Reset metrics at the end of each epoch
    train_acc_metric.reset_states()

除了上述设置之外，我还尝试了其他多种方法来处理此类问题案例。例如，我定义了 3 个损失函数和 3 个指标，但工作不正常。这loss/acc became nan类型的东西。

以下是我在这种情况下的几个直接查询：

• 如何定义loss, metrics and loss_weights
• 如何高效利用一切callbacks特征

只是为了学习机会，如果它还有额外的呢？回归类型输出（以及其余的3多输出，这样总共4）；如何在自定义中处理所有这些fit？我参观过这个SO https://stackoverflow.com/questions/59690188/how-do-i-make-a-multi-output-tensorflow-2-0-neural-network-with-two-different-va，给出了不同类型输出的一些提示（classification + regression).

您只需要执行自定义训练循环，但所有操作都需要执行 3 次（如果您还有连续变量，则需要执行 1 次）。这是使用四输出架构的示例：

import tensorflow as tf
import numpy as np

(xtrain, train_target), (xtest, test_target) = tf.keras.datasets.mnist.load_data()

# 10 categories, one for each digit
ytrain1 = tf.keras.utils.to_categorical(train_target, num_classes=10)
ytest1 = tf.keras.utils.to_categorical(test_target, num_classes=10)

# 2 categories, if the digit is odd or not
ytrain2 = tf.keras.utils.to_categorical((train_target % 2 == 0).astype(int), 
                                        num_classes=2)
ytest2 = tf.keras.utils.to_categorical((test_target % 2 == 0).astype(int), 
                                       num_classes=2)

# 4 categories, based on the interval of the digit
ytrain3 = tf.keras.utils.to_categorical(np.digitize(train_target, [3, 6, 8]), 
                                        num_classes=4)
ytest3 = tf.keras.utils.to_categorical(np.digitize(test_target, [3, 6, 8]), 
                                       num_classes=4)

# Regression, the square of the digit
ytrain4 = tf.square(tf.cast(train_target, tf.float32))
ytest4 = tf.square(tf.cast(test_target, tf.float32))

# train dataset
train_ds = tf.data.Dataset. \
    from_tensor_slices((xtrain, ytrain1, ytrain2, ytrain3, ytrain4)). \
    shuffle(32). \
    batch(32).map(lambda a, *rest: (tf.divide(a[..., None], 255), rest)). \
    prefetch(tf.data.experimental.AUTOTUNE)

# test dataset
test_ds = tf.data.Dataset. \
    from_tensor_slices((xtest, ytest1, ytest2, ytest3, ytest4)). \
    shuffle(32). \
    batch(32).map(lambda a, *rest: (tf.divide(a[..., None], 255), rest)). \
    prefetch(tf.data.experimental.AUTOTUNE)


# architecture
class Net(tf.keras.Model):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = tf.keras.layers.Conv2D(filters=16, kernel_size=(3, 3),
                                            strides=(1, 1), input_shape=(28, 28, 1),
                                            activation='relu')
        self.maxp1 = tf.keras.layers.MaxPool2D(pool_size=(2, 2))
        self.conv2 = tf.keras.layers.Conv2D(filters=32, kernel_size=(3, 3),
                                            strides=(1, 1),
                                            activation='relu')
        self.maxp2 = tf.keras.layers.MaxPool2D(pool_size=(2, 2))
        self.conv3 = tf.keras.layers.Conv2D(filters=64, kernel_size=(3, 3),
                                            strides=(1, 1),
                                            activation='relu')
        self.maxp3 = tf.keras.layers.MaxPool2D(pool_size=(2, 2))
        self.gap = tf.keras.layers.Flatten()
        self.dense = tf.keras.layers.Dense(64, activation='relu')
        self.output1 = tf.keras.layers.Dense(10, activation='softmax')
        self.output2 = tf.keras.layers.Dense(2, activation='softmax')
        self.output3 = tf.keras.layers.Dense(4, activation='softmax')
        self.output4 = tf.keras.layers.Dense(1, activation='linear')

    def call(self, inputs, training=False, **kwargs):
        x = self.conv1(inputs)
        x = self.maxp1(x)
        x = self.conv2(x)
        x = self.maxp2(x)
        x = self.conv3(x)
        x = self.maxp3(x)
        x = self.gap(x)
        x = self.dense(x)
        out1 = self.output1(x)
        out2 = self.output2(x)
        out3 = self.output3(x)
        out4 = self.output4(x)
        return out1, out2, out3, out4


model = Net()

optimizer = tf.keras.optimizers.Adam(learning_rate=0.001)

# the three losses
loss_1 = tf.losses.CategoricalCrossentropy()
loss_2 = tf.losses.CategoricalCrossentropy()
loss_3 = tf.losses.CategoricalCrossentropy()
loss_4 = tf.losses.MeanAbsoluteError()

# mean object that keeps track of the train losses
loss_1_train = tf.metrics.Mean(name='tr_loss_1')
loss_2_train = tf.metrics.Mean(name='tr_loss_2')
loss_3_train = tf.metrics.Mean(name='tr_loss_3')
loss_4_train = tf.metrics.Mean(name='tr_loss_4')

# mean object that keeps track of the test losses
loss_1_test = tf.metrics.Mean(name='ts_loss_1')
loss_2_test = tf.metrics.Mean(name='ts_loss_2')
loss_3_test = tf.metrics.Mean(name='ts_loss_3')
loss_4_test = tf.metrics.Mean(name='ts_loss_4')

# accuracies for printout
acc_1_train = tf.metrics.CategoricalAccuracy(name='tr_acc_1')
acc_2_train = tf.metrics.CategoricalAccuracy(name='tr_acc_2')
acc_3_train = tf.metrics.CategoricalAccuracy(name='tr_acc_3')

# accuracies for printout
acc_1_test = tf.metrics.CategoricalAccuracy(name='ts_acc_1')
acc_2_test = tf.metrics.CategoricalAccuracy(name='ts_acc_2')
acc_3_test = tf.metrics.CategoricalAccuracy(name='ts_acc_3')


# custom training loop
@tf.function
def train_step(x, y1, y2, y3, y4):
    with tf.GradientTape(persistent=True) as tape:
        out1, out2, out3, out4 = model(x, training=True)
        loss_1_value = loss_1(y1, out1)
        loss_2_value = loss_2(y2, out2)
        loss_3_value = loss_3(y3, out3)
        loss_4_value = loss_4(y4, out4)

    losses = [loss_1_value, loss_2_value, loss_3_value, loss_4_value]

    # a list of losses is passed
    grads = tape.gradient(losses, model.trainable_variables)

    # gradients are applied
    optimizer.apply_gradients(zip(grads, model.trainable_variables))

    # losses are updated
    loss_1_train(loss_1_value)
    loss_2_train(loss_2_value)
    loss_3_train(loss_3_value)
    loss_4_train(loss_4_value)

    # accuracies are updated
    acc_1_train.update_state(y1, out1)
    acc_2_train.update_state(y2, out2)
    acc_3_train.update_state(y3, out3)


@tf.function
def test_step(x, y1, y2, y3, y4):
    out1, out2, out3, out4 = model(x, training=False)
    loss_1_value = loss_1(y1, out1)
    loss_2_value = loss_2(y2, out2)
    loss_3_value = loss_3(y3, out3)
    loss_4_value = loss_4(y4, out4)

    loss_1_test(loss_1_value)
    loss_2_test(loss_2_value)
    loss_3_test(loss_3_value)
    loss_4_test(loss_4_value)

    acc_1_test.update_state(y1, out1)
    acc_2_test.update_state(y2, out2)
    acc_3_test.update_state(y3, out3)


for epoch in range(5):
    # train step
    for inputs, outputs1, outputs2, outputs3, outputs4 in train_ds:
        train_step(inputs, outputs1, outputs2, outputs3, outputs4)

    # test step
    for inputs, outputs1, outputs2, outputs3, outputs4 in test_ds:
        test_step(inputs, outputs1, outputs2, outputs3, outputs4)

    metrics = [acc_1_train, acc_1_test,
               acc_2_train, acc_2_test,
               acc_3_train, acc_3_test,
               loss_4_train, loss_4_test]

    # printing metrics
    for metric in metrics:
        print(f'{metric.name}:{metric.result():=6.4f}', end=' ')   
    print()

    # resetting the states of the metrics
    loss_1_train.reset_states()
    loss_2_train.reset_states()
    loss_3_train.reset_states()

    loss_1_test.reset_states()
    loss_2_test.reset_states()
    loss_3_test.reset_states()

    acc_1_train.reset_states()
    acc_2_train.reset_states()
    acc_3_train.reset_states()

    acc_1_test.reset_states()
    acc_2_test.reset_states()
    acc_3_test.reset_states()

ts_acc_1:0.9495 ts_acc_2:0.9685 ts_acc_3:0.9589 ts_loss_4:5.5617 
ts_acc_1:0.9628 ts_acc_2:0.9747 ts_acc_3:0.9697 ts_loss_4:4.8953 
ts_acc_1:0.9697 ts_acc_2:0.9758 ts_acc_3:0.9733 ts_loss_4:4.5209 
ts_acc_1:0.9715 ts_acc_2:0.9796 ts_acc_3:0.9745 ts_loss_4:4.2175 
ts_acc_1:0.9742 ts_acc_2:0.9834 ts_acc_3:0.9775 ts_loss_4:3.9825

我不知道如何在自定义训练循环中使用 Keras 回调，也不知道最热门的问题 https://stackoverflow.com/questions/59438904/applying-callbacks-in-a-custom-training-loop-in-tensorflow-2-0关于这个话题。如果您想使用 EarlyStopping，我个人使用一个collections.deque https://stackoverflow.com/a/63458302/10908375，并在最小损失为倒数第 n 时中断。这是一个例子：

from collections import deque
import numpy as np

epochs = 100
early_stopping = 5

loss_hist = deque(maxlen=early_stopping)

for epoch in range(epochs):
    loss_value = np.random.rand()
    loss_hist.append(loss_value)

    print('Last 5 values: ', *np.round(loss_hist, 3))

    if len(loss_hist) == early_stopping and loss_hist.popleft() < min(loss_hist):
        print('Early stopping. No loss decrease in %i epochs.\n' % early_stopping)
        break

Last 5 values:  0.456
Last 5 values:  0.456 0.153
Last 5 values:  0.456 0.153 0.2
Last 5 values:  0.456 0.153 0.2 0.433
Last 5 values:  0.456 0.153 0.2 0.433 0.528
Last 5 values:  0.153 0.2 0.433 0.528 0.349
Early stopping. No loss decrease in 5 epochs.

可以看到，最后一次，最里面的值是最小的，所以验证损失没有增加。这就是停止条件。