 Veja no TensorFlow.org |  Executar no Google Colab |  Ver fonte no GitHub |  Baixar caderno |
Introdução
Quando você está fazendo aprendizado supervisionado, você pode usar fit() e tudo funciona sem problemas.
Quando você precisa para escrever seu próprio loop de formação a partir do zero, você pode usar o GradientTape e assumir o controle de cada pequeno detalhe.
Mas e se você precisa de um algoritmo de treinamento personalizado, mas você ainda quer beneficiar das características convenientes de fit() , tais como chamadas de retorno, suporte embutido de distribuição, ou passo de fusão?
Um princípio básico é de Keras divulgação progressiva de complexidade. Você sempre deve ser capaz de entrar em fluxos de trabalho de nível inferior de maneira gradual. Você não deve cair de um penhasco se a funcionalidade de alto nível não corresponder exatamente ao seu caso de uso. Você deve ser capaz de obter mais controle sobre os pequenos detalhes, mantendo uma quantidade proporcional de conveniência de alto nível.
Quando você precisa para personalizar o que fit() faz, você deve substituir a função etapa de treinamento do Model de classe. Esta é a função que é chamado pelo fit() para cada lote de dados. Você então será capaz de chamada fit() , como de costume - e será executar o seu próprio algoritmo de aprendizagem.
Observe que esse padrão não impede que você crie modelos com a API funcional. Você pode fazer isso se você está construindo Sequential modelos, modelos API funcionais, ou modelos subclasse.
Vamos ver como isso funciona.
Configurar
Requer o TensorFlow 2.2 ou posterior.
import tensorflow as tf
from tensorflow import keras
Um primeiro exemplo simples
Vamos começar com um exemplo simples:
- Nós criamos uma nova classe que subclasses
keras.Model. - Nós apenas substituir o método
train_step(self, data). - Retornamos um dicionário mapeando nomes de métricas (incluindo a perda) para seu valor atual.
O argumento de entrada data é o que é passado para caber como dados de treinamento:
- Se você passar matrizes Numpy, chamando
fit(x, y, ...), entãodataserá o tuple(x, y) - Se você passar um
tf.data.Dataset, chamandofit(dataset, ...), em seguida,dataserão o que é produzido pelodatasetem cada lote.
No corpo do train_step método, vamos implementar uma atualização de formação regular, semelhante ao que você já está familiarizado. Importante, nós calcular a perda através self.compiled_loss , que envolve a perda de função (es) (s) que foram passados para compile() .
Da mesma forma, chamamos self.compiled_metrics.update_state(y, y_pred) para atualizar o estado das métricas que foram passados na compile() , e consulta os resultados de self.metrics no final para recuperar o seu valor atual.
class CustomModel(keras.Model):
def train_step(self, data):
# Unpack the data. Its structure depends on your model and
# on what you pass to `fit()`.
x, y = data
with tf.GradientTape() as tape:
y_pred = self(x, training=True) # Forward pass
# Compute the loss value
# (the loss function is configured in `compile()`)
loss = self.compiled_loss(y, y_pred, regularization_losses=self.losses)
# Compute gradients
trainable_vars = self.trainable_variables
gradients = tape.gradient(loss, trainable_vars)
# Update weights
self.optimizer.apply_gradients(zip(gradients, trainable_vars))
# Update metrics (includes the metric that tracks the loss)
self.compiled_metrics.update_state(y, y_pred)
# Return a dict mapping metric names to current value
return {m.name: m.result() for m in self.metrics}
Vamos tentar isso:
import numpy as np
# Construct and compile an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
# Just use `fit` as usual
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.fit(x, y, epochs=3)
Epoch 1/3 32/32 [==============================] - 1s 2ms/step - loss: 0.9909 - mae: 0.8601 Epoch 2/3 32/32 [==============================] - 0s 2ms/step - loss: 0.4363 - mae: 0.5345 Epoch 3/3 32/32 [==============================] - 0s 2ms/step - loss: 0.2906 - mae: 0.4311 <keras.callbacks.History at 0x7f5ad1ca1090>
Indo para o nível mais baixo
Naturalmente, você poderia simplesmente ignorar passando uma função perda de compile() e, em vez fazer tudo manualmente em train_step . Da mesma forma para métricas.
Aqui está um exemplo de nível inferior, que só usa compile() para configurar o otimizador:
- Começamos criando
Metricinstâncias para acompanhar a nossa perda e uma pontuação MAE. - Implementamos um costume
train_step()que atualiza o estado dessas métricas (chamandoupdate_state()sobre eles), então consultá-los (viaresult()) para retornar o seu valor médio atual, a ser exibido pela barra de progresso e de ser passar para qualquer retorno de chamada. - Note-se que seria preciso chamar
reset_states()sobre nossas métricas entre cada época! Caso contrário chamandoresult()retornaria uma média desde o início da formação, enquanto que normalmente trabalho com médias per-época. Felizmente, o quadro pode fazer isso por nós: basta listar qualquer métrica que você deseja redefinir nametricspropriedade do modelo. O modelo vai chamarreset_states()em qualquer objeto listado aqui no início de cadafit()época ou no início de uma chamada paraevaluate().
loss_tracker = keras.metrics.Mean(name="loss")
mae_metric = keras.metrics.MeanAbsoluteError(name="mae")
class CustomModel(keras.Model):
def train_step(self, data):
x, y = data
with tf.GradientTape() as tape:
y_pred = self(x, training=True) # Forward pass
# Compute our own loss
loss = keras.losses.mean_squared_error(y, y_pred)
# Compute gradients
trainable_vars = self.trainable_variables
gradients = tape.gradient(loss, trainable_vars)
# Update weights
self.optimizer.apply_gradients(zip(gradients, trainable_vars))
# Compute our own metrics
loss_tracker.update_state(loss)
mae_metric.update_state(y, y_pred)
return {"loss": loss_tracker.result(), "mae": mae_metric.result()}
@property
def metrics(self):
# We list our `Metric` objects here so that `reset_states()` can be
# called automatically at the start of each epoch
# or at the start of `evaluate()`.
# If you don't implement this property, you have to call
# `reset_states()` yourself at the time of your choosing.
return [loss_tracker, mae_metric]
# Construct an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
# We don't passs a loss or metrics here.
model.compile(optimizer="adam")
# Just use `fit` as usual -- you can use callbacks, etc.
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.fit(x, y, epochs=5)
Epoch 1/5 32/32 [==============================] - 0s 1ms/step - loss: 1.5969 - mae: 1.1523 Epoch 2/5 32/32 [==============================] - 0s 1ms/step - loss: 0.7352 - mae: 0.7310 Epoch 3/5 32/32 [==============================] - 0s 1ms/step - loss: 0.3830 - mae: 0.4999 Epoch 4/5 32/32 [==============================] - 0s 1ms/step - loss: 0.2809 - mae: 0.4215 Epoch 5/5 32/32 [==============================] - 0s 1ms/step - loss: 0.2590 - mae: 0.4058 <keras.callbacks.History at 0x7f5ad1b62c50>
Apoiar sample_weight & class_weight
Você deve ter notado que nosso primeiro exemplo básico não fez nenhuma menção à ponderação da amostra. Se você quiser apoiar o fit() argumentos sample_weight e class_weight , você simplesmente faça o seguinte:
- Descompacte
sample_weightdodataargumento - Passá-lo para
compiled_loss&compiled_metrics(claro, você também pode simplesmente aplicá-lo manualmente, se você não confiar nacompile()por perdas e métricas) - É isso. Essa é a lista.
class CustomModel(keras.Model):
def train_step(self, data):
# Unpack the data. Its structure depends on your model and
# on what you pass to `fit()`.
if len(data) == 3:
x, y, sample_weight = data
else:
sample_weight = None
x, y = data
with tf.GradientTape() as tape:
y_pred = self(x, training=True) # Forward pass
# Compute the loss value.
# The loss function is configured in `compile()`.
loss = self.compiled_loss(
y,
y_pred,
sample_weight=sample_weight,
regularization_losses=self.losses,
)
# Compute gradients
trainable_vars = self.trainable_variables
gradients = tape.gradient(loss, trainable_vars)
# Update weights
self.optimizer.apply_gradients(zip(gradients, trainable_vars))
# Update the metrics.
# Metrics are configured in `compile()`.
self.compiled_metrics.update_state(y, y_pred, sample_weight=sample_weight)
# Return a dict mapping metric names to current value.
# Note that it will include the loss (tracked in self.metrics).
return {m.name: m.result() for m in self.metrics}
# Construct and compile an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(optimizer="adam", loss="mse", metrics=["mae"])
# You can now use sample_weight argument
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
sw = np.random.random((1000, 1))
model.fit(x, y, sample_weight=sw, epochs=3)
Epoch 1/3 32/32 [==============================] - 0s 2ms/step - loss: 0.1365 - mae: 0.4196 Epoch 2/3 32/32 [==============================] - 0s 2ms/step - loss: 0.1285 - mae: 0.4068 Epoch 3/3 32/32 [==============================] - 0s 2ms/step - loss: 0.1212 - mae: 0.3971 <keras.callbacks.History at 0x7f5ad1ba64d0>
Fornecendo sua própria etapa de avaliação
E se você quiser fazer o mesmo para as chamadas para model.evaluate() ? Então você teria que substituir test_step exatamente da mesma maneira. Aqui está o que parece:
class CustomModel(keras.Model):
def test_step(self, data):
# Unpack the data
x, y = data
# Compute predictions
y_pred = self(x, training=False)
# Updates the metrics tracking the loss
self.compiled_loss(y, y_pred, regularization_losses=self.losses)
# Update the metrics.
self.compiled_metrics.update_state(y, y_pred)
# Return a dict mapping metric names to current value.
# Note that it will include the loss (tracked in self.metrics).
return {m.name: m.result() for m in self.metrics}
# Construct an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(loss="mse", metrics=["mae"])
# Evaluate with our custom test_step
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.evaluate(x, y)
32/32 [==============================] - 0s 1ms/step - loss: 2.7584 - mae: 1.5920 [2.758362054824829, 1.59201979637146]
Resumindo: um exemplo de GAN de ponta a ponta
Vejamos um exemplo de ponta a ponta que aproveita tudo o que você acabou de aprender.
Vamos considerar:
- Uma rede geradora destinada a gerar imagens 28x28x1.
- Uma rede discriminadora destinada a classificar imagens 28x28x1 em duas classes ("falsas" e "reais").
- Um otimizador para cada.
- Uma função de perda para treinar o discriminador.
from tensorflow.keras import layers
# Create the discriminator
discriminator = keras.Sequential(
[
keras.Input(shape=(28, 28, 1)),
layers.Conv2D(64, (3, 3), strides=(2, 2), padding="same"),
layers.LeakyReLU(alpha=0.2),
layers.Conv2D(128, (3, 3), strides=(2, 2), padding="same"),
layers.LeakyReLU(alpha=0.2),
layers.GlobalMaxPooling2D(),
layers.Dense(1),
],
name="discriminator",
)
# Create the generator
latent_dim = 128
generator = keras.Sequential(
[
keras.Input(shape=(latent_dim,)),
# We want to generate 128 coefficients to reshape into a 7x7x128 map
layers.Dense(7 * 7 * 128),
layers.LeakyReLU(alpha=0.2),
layers.Reshape((7, 7, 128)),
layers.Conv2DTranspose(128, (4, 4), strides=(2, 2), padding="same"),
layers.LeakyReLU(alpha=0.2),
layers.Conv2DTranspose(128, (4, 4), strides=(2, 2), padding="same"),
layers.LeakyReLU(alpha=0.2),
layers.Conv2D(1, (7, 7), padding="same", activation="sigmoid"),
],
name="generator",
)
Aqui está uma classe GAN recurso completo, substituindo compile() para usar a sua própria assinatura e implementação de todo o algoritmo de GAN em 17 linhas em train_step :
class GAN(keras.Model):
def __init__(self, discriminator, generator, latent_dim):
super(GAN, self).__init__()
self.discriminator = discriminator
self.generator = generator
self.latent_dim = latent_dim
def compile(self, d_optimizer, g_optimizer, loss_fn):
super(GAN, self).compile()
self.d_optimizer = d_optimizer
self.g_optimizer = g_optimizer
self.loss_fn = loss_fn
def train_step(self, real_images):
if isinstance(real_images, tuple):
real_images = real_images[0]
# Sample random points in the latent space
batch_size = tf.shape(real_images)[0]
random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))
# Decode them to fake images
generated_images = self.generator(random_latent_vectors)
# Combine them with real images
combined_images = tf.concat([generated_images, real_images], axis=0)
# Assemble labels discriminating real from fake images
labels = tf.concat(
[tf.ones((batch_size, 1)), tf.zeros((batch_size, 1))], axis=0
)
# Add random noise to the labels - important trick!
labels += 0.05 * tf.random.uniform(tf.shape(labels))
# Train the discriminator
with tf.GradientTape() as tape:
predictions = self.discriminator(combined_images)
d_loss = self.loss_fn(labels, predictions)
grads = tape.gradient(d_loss, self.discriminator.trainable_weights)
self.d_optimizer.apply_gradients(
zip(grads, self.discriminator.trainable_weights)
)
# Sample random points in the latent space
random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))
# Assemble labels that say "all real images"
misleading_labels = tf.zeros((batch_size, 1))
# Train the generator (note that we should *not* update the weights
# of the discriminator)!
with tf.GradientTape() as tape:
predictions = self.discriminator(self.generator(random_latent_vectors))
g_loss = self.loss_fn(misleading_labels, predictions)
grads = tape.gradient(g_loss, self.generator.trainable_weights)
self.g_optimizer.apply_gradients(zip(grads, self.generator.trainable_weights))
return {"d_loss": d_loss, "g_loss": g_loss}
Vamos testá-lo:
# Prepare the dataset. We use both the training & test MNIST digits.
batch_size = 64
(x_train, _), (x_test, _) = keras.datasets.mnist.load_data()
all_digits = np.concatenate([x_train, x_test])
all_digits = all_digits.astype("float32") / 255.0
all_digits = np.reshape(all_digits, (-1, 28, 28, 1))
dataset = tf.data.Dataset.from_tensor_slices(all_digits)
dataset = dataset.shuffle(buffer_size=1024).batch(batch_size)
gan = GAN(discriminator=discriminator, generator=generator, latent_dim=latent_dim)
gan.compile(
d_optimizer=keras.optimizers.Adam(learning_rate=0.0003),
g_optimizer=keras.optimizers.Adam(learning_rate=0.0003),
loss_fn=keras.losses.BinaryCrossentropy(from_logits=True),
)
# To limit the execution time, we only train on 100 batches. You can train on
# the entire dataset. You will need about 20 epochs to get nice results.
gan.fit(dataset.take(100), epochs=1)
100/100 [==============================] - 3s 11ms/step - d_loss: 0.4031 - g_loss: 0.9305 <keras.callbacks.History at 0x7f5ad1b37c50>
As ideias por trás do aprendizado profundo são simples, então por que sua implementação deveria ser dolorosa?