Tensorflow Keras 保留每批的损失

我正在寻找构建 keras 模型架构的最佳实践的建议/示例。

我一直在摸索 Model() 子类和功能模型的各种迭代，但无法连接所有点。

该模型应具有自定义指标和相关损失，其中：在训练期间，按批次计算指标，并在每个时期结束时根据批次计算的平均值计算最终指标/损失。

根据我的想法，我需要一个Custom_Batch_Metric()除了一个之外，还将维护每个批次的指标列表Custom_Final_Metric()这将对每批结果进行平均。我不知道如何实现这个。

例如......每批我想生成一个metric and loss用于 y_true、y_pred 的相关性。在我的纪元结束时，我想对批次相关性进行平均（或找到最大值）。

如果有人能向我推荐有关此类架构的任何书籍，我将非常感激。

一个简单的解决方案是子类化tf.keras.callbacks.Callback并定义on_train_batch_end（或测试）。然后还有on_epoch_end.

class SaveBatchLoss(tf.keras.callbacks.Callback):
    def on_train_batch_end(self, batch, logs=None): 
        batch_end_loss.append(logs['loss'])

import tensorflow as tf
from tensorflow.keras.layers import Dense
from sklearn.datasets import load_iris
import numpy as np

X, y = load_iris(return_X_y=True)
X = X.astype(np.float32)

ds = tf.data.Dataset.from_tensor_slices((X, y)).shuffle(25).batch(8)

model = tf.keras.Sequential([
    Dense(16, activation='relu'),
    Dense(32, activation='relu'),
    Dense(3, activation='softmax')])

model.compile(loss='sparse_categorical_crossentropy', optimizer='adam', 
              metrics=['accuracy'])

batch_end_loss = list()

class SaveBatchLoss(tf.keras.callbacks.Callback):
    def on_train_batch_end(self, batch, logs=None):
        batch_end_loss.append(logs['loss'])

history = model.fit(ds, epochs=10, callbacks=SaveBatchLoss())

batch_end_loss[::20]

[1.2742226123809814,
 0.9069833755493164,
 0.9728888869285583,
 0.9536505937576294,
 0.8957988023757935,
 0.8624499440193176,
 0.7952101826667786,
 0.7765023112297058,
 0.7615134716033936,
 0.7278715968132019]

作为一种更复杂的方法，将损失值附加到每个批次末尾的列表中，以及每个 opoch 末尾的另一个列表中。像这样：

train_loss_per_train_batch = list()
train_loss_per_train_epoch = list()

for epoch in range(1, 25 + 1): 
    train_loss = tf.metrics.Mean()
    train_acc = tf.metrics.SparseCategoricalAccuracy()
    test_loss = tf.metrics.Mean()
    test_acc = tf.metrics.SparseCategoricalAccuracy()

    for x, y in train:
        loss_value, grads = get_grad(model, x, y)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))
        train_loss.update_state(loss_value)
        train_acc.update_state(y, model(x, training=True))
        train_loss_per_train_batch.append(loss_value.numpy())
    
    train_loss_per_train_epoch.append(train_loss.result())

实现此功能的完整训练脚本将是：

import tensorflow as tf
import tensorflow_datasets as tfds

ds = tfds.load('iris', split='train', as_supervised=True)

train = ds.take(125).shuffle(16).batch(4)
test = ds.skip(125).take(25).shuffle(16).batch(4)

class MyModel(tf.keras.Model):
    def __init__(self):
        super(MyModel, self).__init__()
        self.d1 = tf.keras.layers.Dense(16, activation='relu')
        self.d2 = tf.keras.layers.Dense(32, activation='relu')
        self.d3 = tf.keras.layers.Dense(3, activation='softmax')

    def call(self, x, training=None, **kwargs):
        x = self.d1(x)
        x = self.d2(x)
        x = self.d3(x)
        return x

model = MyModel()

loss_object = tf.losses.SparseCategoricalCrossentropy(from_logits=False)


def compute_loss(model, x, y, training):
  out = model(x, training=training)
  loss = loss_object(y_true=y, y_pred=out)
  return loss


def get_grad(model, x, y):
    with tf.GradientTape() as tape:
        loss = compute_loss(model, x, y, training=True)
    return loss, tape.gradient(loss, model.trainable_variables)


optimizer = tf.optimizers.Adam()

verbose = "Epoch {:2d} Loss: {:.3f} TLoss: {:.3f} Acc: {:.2%} TAcc: {:.2%}"

train_loss_per_train_batch = list()
train_loss_per_train_epoch = list()

for epoch in range(1, 25 + 1):
    train_loss = tf.metrics.Mean()
    train_acc = tf.metrics.SparseCategoricalAccuracy()
    test_loss = tf.metrics.Mean()
    test_acc = tf.metrics.SparseCategoricalAccuracy()

    for x, y in train:
        loss_value, grads = get_grad(model, x, y)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))
        train_loss.update_state(loss_value)
        train_acc.update_state(y, model(x, training=True))
        train_loss_per_train_batch.append(loss_value.numpy())

    train_loss_per_train_epoch.append(train_loss.result())

    for x, y in test:
        loss_value, _ = get_grad(model, x, y)
        test_loss.update_state(loss_value)
        test_acc.update_state(y, model(x, training=False))

    print(verbose.format(epoch,
                         train_loss.result(),
                         test_loss.result(),
                         train_acc.result(),
                         test_acc.result()))