Функция пользовательских потерь с историческим усреднением Keras

В настоящее время я экспериментирую с генеративными состязательными сетями в Керасе. Как предлагается в этой статье, я хочу использовать функцию исторических усредненных потерь. То есть я хочу наказать за изменение весов сети. Я не знаю, как это грамотно реализовать.

Я реализовал пользовательскую функцию потерь в соответствии с ответом на this пост.

def historical_averaging_wrapper(current_weights, prev_weights):
    def historical_averaging(y_true, y_pred):
        diff = 0
        for i in range(len(current_weights)):
            diff += abs(np.sum(current_weights[i]) + np.sum(prev_weights[i]))
        return K.binary_crossentropy(y_true, y_pred) + diff
    return historical_averaging

Веса сети штрафуются, и веса меняются после каждого пакета данных.

Моя первая идея заключалась в том, чтобы обновлять функцию потерь после каждой партии. Примерно так:

prev_weights = model.get_weights()
for i in range(len(data)/batch_len):
    current_weights = model.get_weights()
    model.compile(loss=historical_averaging_wrapper(current_weights, prev_weights), optimizer='adam')
    model.fit(training_data[i*batch_size:(i+1)*batch_size], training_labels[i*batch_size:(i+1)*batch_size], epochs=1, batch_size=batch_size)
    prev_weights = current_weights

Это разумно? На мой взгляд, такой подход кажется немного «запутанным». Есть ли еще возможность сделать это «поумнее»? Например, обновить функцию потерь в генераторе данных и использовать fit_generator ()? Заранее спасибо.


person MMikkk    schedule 19.07.2019    source источник


Ответы (1)


Функции потерь - это операции на графике с использованием тензоров. Вы можете определить дополнительные тензоры в функции потерь для хранения предыдущих значений. Это пример:

import tensorflow as tf
import tensorflow.keras.backend as K
keras = tf.keras

class HistoricalAvgLoss(object):
  def __init__(self, model):
    # create tensors (initialized to zero) to hold the previous value of the
    # weights
    self.prev_weights = []
    for w in model.get_weights():
      self.prev_weights.append(K.variable(np.zeros(w.shape)))

  def loss(self, y_true, y_pred):
    err = keras.losses.mean_squared_error(y_true, y_pred)
    werr = [K.mean(K.abs(c - p)) for c, p in zip(model.get_weights(), self.prev_weights)]
    self.prev_weights = K.in_train_phase(
        [K.update(p, c) for c, p in zip(model.get_weights(), self.prev_weights)],
        self.prev_weights
    )
    return K.in_train_phase(err + K.sum(werr), err)

Переменная prev_weights содержит предыдущие значения. Обратите внимание, что мы добавили операцию K.update после вычисления ошибок веса.

Образец модели для тестирования:

model = keras.models.Sequential([
    keras.layers.Input(shape=(4,)),
    keras.layers.Dense(8),
    keras.layers.Dense(4),
    keras.layers.Dense(1),
])

loss_obj = HistoricalAvgLoss(model)

model.compile('adam', loss_obj.loss)
model.summary()

Некоторые тестовые данные и целевая функция:

import numpy as np

def test_fn(x):
  return x[0]*x[1] + 2.0 * x[1]**2 + x[2]/x[3] + 3.0 * x[3]

X = np.random.rand(1000, 4)
y = np.apply_along_axis(test_fn, 1, X)

hist = model.fit(X, y, validation_split=0.25, epochs=10)

В моем тесте потери модели со временем уменьшаются.

person Pedro Marques    schedule 20.07.2019
comment
Спасибо за ответ, Педро. Отличная идея использовать класс для отслеживания предыдущих весов. У меня все еще была небольшая проблема. Переменная model в функции потерь не определена. Я реализовал функцию-оболочку, как в постановке задачи, передающей модель. Кажется, это работает. - person MMikkk; 22.07.2019