Përmbajtja Overview
- Hyrje
- Setup
- Shembulli i parë i thjeshtë
- Në nivelin më të ulët
- Mbështetje për sample_weight & class_weight
- Bëni hapin tuaj të vlerësimit
- Përmbajtja: një shembull i GAN-it end-to-end
Hyrje
Kur jeni duke bërë mësim të mbikëqyrur, ju mund të përdornifit()
Gjithçka funksionon në mënyrë të qetë.
Kur ju duhet të shkruani rrjedhën tuaj të trajnimit nga e para, ju mund të përdorniGradientTape
Mbani nën kontroll çdo detaj të vogël.
Por çfarë nëse keni nevojë për një algoritëm të personalizuar të trajnimit, por ende dëshironi të përfitoni nga tiparet e përshtatshme tëfit()
, të tilla si thirrje mbrapa, mbështetje e ndërtuar e shpërndarjes, ose fusing hap?
Një nga parimet kryesore të Keras ështëprogressive disclosure of complexityJu gjithmonë duhet të jeni në gjendje të hyni në rrjedhat e punës të nivelit të ulët në një mënyrë graduale.Ju nuk duhet të bini nga një shkëmbinj nëse funksionaliteti i nivelit të lartë nuk përputhet saktësisht me rastin tuaj të përdorimit.Ju duhet të jeni në gjendje të fitoni më shumë kontroll mbi detajet e vogla duke ruajtur një sasi proporcionale të komoditetit të nivelit të lartë.
Kur keni nevojë për të rregulluar atë qëfit()
Do, duhet tëoverride the training step function of the Model
Kjo është funksioni që quhetfit()
për çdo batch të të dhënave. atëherë ju do të jetë në gjendje të telefononifit()
si zakonisht - dhe do të drejtojë algoritmin tuaj të të mësuarit.
Vini re se ky model nuk ju pengon të ndërtoni modele me API-në Funksionale.Sequential
Modelet e API-ve funksionale, ose modelet e nënklasuara.
Le të shohim se si funksionon kjo.
Setup
Kërkon TensorFlow 2.8 ose më të ri.
import tensorflow as tf
from tensorflow import keras
Shembulli i parë i thjeshtë
Le të fillojmë me një shembull të thjeshtë:
- Ne krijojmë një klasë të re që nënklasa keras.Model.
- Ne thjesht mbivlerësojmë metodën train_step (vetë, të dhënat).
- Ne i kthejmë emrat metrikë (duke përfshirë humbjen) në vlerën e tyre të tanishme.
Argumentet e hyrjesdata
është ajo që kalon për t'u përshtatur si të dhënat e trajnimit:
- Nëse kaloni array Numpy, duke thirrur fit(x, y, ...), atëherë të dhënat do të jenë tuple (x, y)
- Nëse kaloni një tf.data.Dataset, duke thirrur fit(dataset, ...), atëherë të dhënat do të jenë ajo që rezulton nga dataset në çdo batch.
Në trupin etrain_step
metodën, ne zbatojmë një përditësim të rregullt të trajnimit, të ngjashëm me atë që ju tashmë jeni të njohur me.we compute the loss via self.compute_loss()
e cila do t’i shkatërrojë ata që kanë humburcompile()
.
Në të njëjtën mënyrë, ne thërrasimmetric.update_state(y, y_pred)
në metrikë ngaself.metrics
, për të përditësuar gjendjen e metrikave që janë kaluar nëcompile()
Dhe ne kërkojmë rezultate ngaself.metrics
në fund për të rikthyer vlerën e tyre të tanishme.
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.compute_loss(y=y, y_pred=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))
# Update metrics (includes the metric that tracks the loss)
for metric in self.metrics:
if metric.name == "loss":
metric.update_state(loss)
else:
metric.update_state(y, y_pred)
# Return a dict mapping metric names to current value
return {m.name: m.result() for m in self.metrics}
Le të provojmë këtë:
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 [==============================] - 3s 2ms/step - loss: 1.6446
Epoch 2/3
32/32 [==============================] - 0s 2ms/step - loss: 0.7554
Epoch 3/3
32/32 [==============================] - 0s 2ms/step - loss: 0.3924
<keras.src.callbacks.History at 0x7fef5c11ba30>
Në nivelin më të ulët
Natyrisht, ju thjesht mund të kaloni kalimin e një funksioni humbje nëcompile()
Në vend të kësaj, bëni gjithçkamanualishtnëtrain_step
Po ashtu edhe për metrologjinë.
Këtu është një shembull i nivelit të ulët, i cili përdor vetëmcompile()
Për të rregulluar optimizimin:
- Ne fillojmë duke krijuar instancat metrike për të ndjekur humbjen tonë dhe një rezultat MAE (në __init__()).
- Ne zbatojmë një train_step() të përshtatur që përditëson statusin e këtyre metrikave (duke thirrur update_state() në to), pastaj i pyet ata (via result()) për të kthyer vlerën e tyre aktuale mesatare, për t'u shfaqur nga bara e progresit dhe për t'u kaluar në çdo thirrje prapa.
- Vini re se ne do të duhet të thërrasim reset_states() në metrikat tona midis çdo epoke! Përndryshe thërrimi rezultat() do të kthente një mesatare që nga fillimi i trajnimit, ndërsa ne zakonisht punojmë me mesatare për epoke. Për fat të mirë, korniza mund ta bëjë këtë për ne: thjesht listoni çdo metrikë që dëshironi të rivendosni në pronën e metrikave të modelit.
class CustomModel(keras.Model):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.loss_tracker = keras.metrics.Mean(name="loss")
self.mae_metric = keras.metrics.MeanAbsoluteError(name="mae")
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
self.loss_tracker.update_state(loss)
self.mae_metric.update_state(y, y_pred)
return {"loss": self.loss_tracker.result(), "mae": self.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 [self.loss_tracker, self.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 pass 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 2ms/step - loss: 0.3240 - mae: 0.4583
Epoch 2/5
32/32 [==============================] - 0s 2ms/step - loss: 0.2416 - mae: 0.3984
Epoch 3/5
32/32 [==============================] - 0s 2ms/step - loss: 0.2340 - mae: 0.3919
Epoch 4/5
32/32 [==============================] - 0s 2ms/step - loss: 0.2274 - mae: 0.3870
Epoch 5/5
32/32 [==============================] - 0s 2ms/step - loss: 0.2197 - mae: 0.3808
<keras.src.callbacks.History at 0x7fef3c130b20>
MbështetjeShembull - peshë
& tëklasë - peshë
Shembull - peshë
klasë - peshë
Ju mund të keni vënë re se shembulli ynë i parë bazë nuk bëri asnjë përmendje të peshimit të mostrës.fit()
Argumentetsample_weight
dheclass_weight
Ju thjesht do të bëni gjënë e mëposhtme:
- Unpack sample_weight nga argumenti i të dhënave
- Kaloni atë në compute_loss & update_state (natyrisht, ju gjithashtu mund ta aplikoni atë manualisht nëse nuk mbështeteni në compile() për humbjet & metrics)
- Kjo është ajo
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.compute_loss(
y=y,
y_pred=y_pred,
sample_weight=sample_weight,
)
# 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()`.
for metric in self.metrics:
if metric.name == "loss":
metric.update_state(loss)
else:
metric.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.1298
Epoch 2/3
32/32 [==============================] - 0s 2ms/step - loss: 0.1179
Epoch 3/3
32/32 [==============================] - 0s 2ms/step - loss: 0.1121
<keras.src.callbacks.History at 0x7fef3c168100>
Bëni hapin tuaj të vlerësimit
Çfarë nëse doni të bëni të njëjtën gjë për thirrjet nëmodel.evaluate()
Atëherë ju do të mbivlerësonitest_step
në të njëjtën mënyrë. ja se si duket:
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.compute_loss(y=y, y_pred=y_pred)
# Update the metrics.
for metric in self.metrics:
if metric.name != "loss":
metric.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: 0.9028
0.9028095006942749
Përmbajtja: një shembull i GAN-it end-to-end
Le të ecim nëpër një shembull nga fundi në fund që përfiton nga çdo gjë që sapo keni mësuar.
Le të konsiderojmë:
- Një rrjet gjenerator ishte projektuar për të gjeneruar 28x28x1 imazhe.
- Një rrjet diskriminues synonte të klasifikonte imazhet 28x28x1 në dy klasa ("fake" dhe "real").
- Një optimizues për secilin.
- Një funksion i humbjes për të trajnuar diskriminuesin.
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",
)
Këtu është një klasë e plotë GAN, overridingcompile()
të përdorë nënshkrimin e vet dhe të zbatojë të gjithë algoritmin GAN në 17 rreshta nëtrain_step
:
class GAN(keras.Model):
def __init__(self, discriminator, generator, latent_dim):
super().__init__()
self.discriminator = discriminator
self.generator = generator
self.latent_dim = latent_dim
self.d_loss_tracker = keras.metrics.Mean(name="d_loss")
self.g_loss_tracker = keras.metrics.Mean(name="g_loss")
def compile(self, d_optimizer, g_optimizer, loss_fn):
super().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))
# Update metrics and return their value.
self.d_loss_tracker.update_state(d_loss)
self.g_loss_tracker.update_state(g_loss)
return {
"d_loss": self.d_loss_tracker.result(),
"g_loss": self.g_loss_tracker.result(),
}
Le ta testojmë atë:
# 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)
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/mnist.npz
11490434/11490434 [==============================] - 0s 0us/step
100/100 [==============================] - 8s 15ms/step - d_loss: 0.4372 - g_loss: 0.8775
<keras.src.callbacks.History at 0x7feee42ff190>
Ideja prapa të mësuarit të thellë janë të thjeshta, kështu që pse zbatimi i tyre duhet të jetë i dhimbshëm?
I publikuar fillimisht në faqen e internetit TensorFlow, ky artikull shfaqet këtu nën një titull të ri dhe është i licencuar nën CC BY 4.0.
I publikuar fillimisht në faqen e internetit TensorFlow, ky artikull shfaqet këtu nën një titull të ri dhe është i licencuar nën CC BY 4.0.
TensorFlow