نوشتن حلقه آموزش از ابتدا

مشاهده در TensorFlow.org

در Google Colab اجرا شود

مشاهده منبع در GitHub

دانلود دفترچه یادداشت

برپایی

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

معرفی

Keras آموزش به طور پیش فرض و حلقه ارزیابی، فراهم می کند fit() و evaluate() . استفاده از آنها را در راهنمای تحت پوشش آموزش و ارزیابی با روش ساخته شده است در .

اگر شما می خواهید برای سفارشی کردن الگوریتم یادگیری از مدل خود را در حالی که هنوز اعمال نفوذ راحتی از fit() (به عنوان مثال، برای آموزش یک GAN با استفاده از fit() )، شما می توانید زیر مجموعه Model کلاس و اجرای خود را train_step() روش، که بارها و بارها در طول نام fit() . این است که در راهنمای تحت پوشش سفارشی آنچه در اتفاق می افتد fit() .

حال، اگر می‌خواهید کنترل سطح پایینی بر آموزش و ارزیابی داشته باشید، باید حلقه‌های آموزش و ارزیابی خود را از ابتدا بنویسید. این چیزی است که این راهنما در مورد آن است.

با استفاده از GradientTape : یک مثال اول پایان به پایان

فراخوانی یک مدل در داخل یک GradientTape دامنه شما را قادر به بازیابی شیب وزن تربیت شدنی از لایه با توجه به ارزش از دست دادن. با استفاده از نمونه بهینه ساز، شما می توانید این شیب برای به روز رسانی این متغیرها (که شما می توانید با استفاده از بازیابی استفاده model.trainable_weights ).

بیایید یک مدل ساده MNIST را در نظر بگیریم:

inputs = keras.Input(shape=(784,), name="digits")
x1 = layers.Dense(64, activation="relu")(inputs)
x2 = layers.Dense(64, activation="relu")(x1)
outputs = layers.Dense(10, name="predictions")(x2)
model = keras.Model(inputs=inputs, outputs=outputs)

بیایید آن را با استفاده از گرادیان دسته ای کوچک با یک حلقه آموزشی سفارشی آموزش دهیم.

ابتدا، ما به یک بهینه ساز، یک تابع ضرر و یک مجموعه داده نیاز داریم:

# Instantiate an optimizer.
optimizer = keras.optimizers.SGD(learning_rate=1e-3)
# Instantiate a loss function.
loss_fn = keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# Prepare the training dataset.
batch_size = 64
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()
x_train = np.reshape(x_train, (-1, 784))
x_test = np.reshape(x_test, (-1, 784))

# Reserve 10,000 samples for validation.
x_val = x_train[-10000:]
y_val = y_train[-10000:]
x_train = x_train[:-10000]
y_train = y_train[:-10000]

# Prepare the training dataset.
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(batch_size)

# Prepare the validation dataset.
val_dataset = tf.data.Dataset.from_tensor_slices((x_val, y_val))
val_dataset = val_dataset.batch(batch_size)

Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/mnist.npz
11493376/11490434 [==============================] - 1s 0us/step
11501568/11490434 [==============================] - 1s 0us/step

این حلقه آموزشی ما است:

• ما باز کردن یک for حلقه می کند که بیش از دوره
• برای هر دوره، ما باز کردن یک for حلقه می کند که بیش از مجموعه داده، در دسته
• برای هر دسته ای، ما باز کردن یک GradientTape() دامنه
• در این محدوده، مدل را (گذر به جلو) فراخوانی می کنیم و ضرر را محاسبه می کنیم
• خارج از محدوده، ما گرادیان های وزن مدل را با توجه به از دست دادن بازیابی می کنیم.
• در نهایت از بهینه ساز برای به روز رسانی وزن های مدل بر اساس گرادیان ها استفاده می کنیم

epochs = 2
for epoch in range(epochs):
    print("\nStart of epoch %d" % (epoch,))

    # Iterate over the batches of the dataset.
    for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):

        # Open a GradientTape to record the operations run
        # during the forward pass, which enables auto-differentiation.
        with tf.GradientTape() as tape:

            # Run the forward pass of the layer.
            # The operations that the layer applies
            # to its inputs are going to be recorded
            # on the GradientTape.
            logits = model(x_batch_train, training=True)  # Logits for this minibatch

            # Compute the loss value for this minibatch.
            loss_value = loss_fn(y_batch_train, logits)

        # Use the gradient tape to automatically retrieve
        # the gradients of the trainable variables with respect to the loss.
        grads = tape.gradient(loss_value, model.trainable_weights)

        # Run one step of gradient descent by updating
        # the value of the variables to minimize the loss.
        optimizer.apply_gradients(zip(grads, model.trainable_weights))

        # Log every 200 batches.
        if step % 200 == 0:
            print(
                "Training loss (for one batch) at step %d: %.4f"
                % (step, float(loss_value))
            )
            print("Seen so far: %s samples" % ((step + 1) * batch_size))

Start of epoch 0
Training loss (for one batch) at step 0: 68.7478
Seen so far: 64 samples
Training loss (for one batch) at step 200: 1.9448
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 1.1859
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.6914
Seen so far: 38464 samples

Start of epoch 1
Training loss (for one batch) at step 0: 0.9113
Seen so far: 64 samples
Training loss (for one batch) at step 200: 0.9550
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 0.5139
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.7227
Seen so far: 38464 samples

مدیریت سطح پایین معیارها

بیایید نظارت بر معیارها را به این حلقه اصلی اضافه کنیم.

می‌توانید به راحتی از معیارهای داخلی (یا موارد سفارشی که نوشته‌اید) در چنین حلقه‌های آموزشی که از ابتدا نوشته شده‌اند، دوباره استفاده کنید. این جریان است:

• متریک را در ابتدای حلقه مثال بزنید
• پاسخ metric.update_state() بعد از هر دسته ای
• پاسخ metric.result() هنگامی که شما نیاز به نمایش ارزش فعلی از متریک
• پاسخ metric.reset_states() هنگامی که شما نیاز به پاک کردن دولت از متریک (معمولا در پایان یک عصر)

اجازه دهید با استفاده از این دانش برای محاسبه SparseCategoricalAccuracy در داده ها اعتبار در پایان هر دوره:

# Get model
inputs = keras.Input(shape=(784,), name="digits")
x = layers.Dense(64, activation="relu", name="dense_1")(inputs)
x = layers.Dense(64, activation="relu", name="dense_2")(x)
outputs = layers.Dense(10, name="predictions")(x)
model = keras.Model(inputs=inputs, outputs=outputs)

# Instantiate an optimizer to train the model.
optimizer = keras.optimizers.SGD(learning_rate=1e-3)
# Instantiate a loss function.
loss_fn = keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# Prepare the metrics.
train_acc_metric = keras.metrics.SparseCategoricalAccuracy()
val_acc_metric = keras.metrics.SparseCategoricalAccuracy()

حلقه آموزش و ارزیابی ما در اینجا آمده است:

import time

epochs = 2
for epoch in range(epochs):
    print("\nStart of epoch %d" % (epoch,))
    start_time = time.time()

    # Iterate over the batches of the dataset.
    for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):
        with tf.GradientTape() as tape:
            logits = model(x_batch_train, training=True)
            loss_value = loss_fn(y_batch_train, logits)
        grads = tape.gradient(loss_value, model.trainable_weights)
        optimizer.apply_gradients(zip(grads, model.trainable_weights))

        # Update training metric.
        train_acc_metric.update_state(y_batch_train, logits)

        # Log every 200 batches.
        if step % 200 == 0:
            print(
                "Training loss (for one batch) at step %d: %.4f"
                % (step, float(loss_value))
            )
            print("Seen so far: %d samples" % ((step + 1) * batch_size))

    # Display metrics at the end of each epoch.
    train_acc = train_acc_metric.result()
    print("Training acc over epoch: %.4f" % (float(train_acc),))

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

    # Run a validation loop at the end of each epoch.
    for x_batch_val, y_batch_val in val_dataset:
        val_logits = model(x_batch_val, training=False)
        # Update val metrics
        val_acc_metric.update_state(y_batch_val, val_logits)
    val_acc = val_acc_metric.result()
    val_acc_metric.reset_states()
    print("Validation acc: %.4f" % (float(val_acc),))
    print("Time taken: %.2fs" % (time.time() - start_time))

Start of epoch 0
Training loss (for one batch) at step 0: 88.9958
Seen so far: 64 samples
Training loss (for one batch) at step 200: 2.2214
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 1.3083
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.8282
Seen so far: 38464 samples
Training acc over epoch: 0.7406
Validation acc: 0.8201
Time taken: 6.31s

Start of epoch 1
Training loss (for one batch) at step 0: 0.3276
Seen so far: 64 samples
Training loss (for one batch) at step 200: 0.4819
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 0.5971
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.5862
Seen so far: 38464 samples
Training acc over epoch: 0.8474
Validation acc: 0.8676
Time taken: 5.98s

سرعت بالا به گام آموزش خود را با tf.function

زمان اجرا به طور پیش فرض در TensorFlow 2 است اعدام مشتاق . به این ترتیب، حلقه آموزشی ما در بالا مشتاقانه اجرا می شود.

این برای اشکال زدایی عالی است، اما کامپایل نمودار مزیت عملکردی مشخصی دارد. توصیف محاسبات خود به عنوان یک نمودار استاتیک، چارچوب را قادر می‌سازد تا بهینه‌سازی عملکرد جهانی را اعمال کند. این غیرممکن است زمانی که چارچوب محدود شده است تا با حرص و طمع عملیاتی را پس از دیگری اجرا کند، بدون اینکه از آنچه در آینده می آید اطلاعی داشته باشد.

شما می توانید هر تابعی را که تانسورها را به عنوان ورودی دریافت می کند در یک نمودار استاتیک کامپایل کنید. فقط یک اضافه @tf.function دکوراتور بر روی آن، مثل این:

@tf.function
def train_step(x, y):
    with tf.GradientTape() as tape:
        logits = model(x, training=True)
        loss_value = loss_fn(y, logits)
    grads = tape.gradient(loss_value, model.trainable_weights)
    optimizer.apply_gradients(zip(grads, model.trainable_weights))
    train_acc_metric.update_state(y, logits)
    return loss_value

بیایید همین کار را با مرحله ارزیابی انجام دهیم:

@tf.function
def test_step(x, y):
    val_logits = model(x, training=False)
    val_acc_metric.update_state(y, val_logits)

حالا بیایید حلقه آموزشی خود را با این مرحله آموزشی کامپایل شده دوباره اجرا کنیم:

import time

epochs = 2
for epoch in range(epochs):
    print("\nStart of epoch %d" % (epoch,))
    start_time = time.time()

    # Iterate over the batches of the dataset.
    for step, (x_batch_train, y_batch_train) in enumerate(train_dataset):
        loss_value = train_step(x_batch_train, y_batch_train)

        # Log every 200 batches.
        if step % 200 == 0:
            print(
                "Training loss (for one batch) at step %d: %.4f"
                % (step, float(loss_value))
            )
            print("Seen so far: %d samples" % ((step + 1) * batch_size))

    # Display metrics at the end of each epoch.
    train_acc = train_acc_metric.result()
    print("Training acc over epoch: %.4f" % (float(train_acc),))

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

    # Run a validation loop at the end of each epoch.
    for x_batch_val, y_batch_val in val_dataset:
        test_step(x_batch_val, y_batch_val)

    val_acc = val_acc_metric.result()
    val_acc_metric.reset_states()
    print("Validation acc: %.4f" % (float(val_acc),))
    print("Time taken: %.2fs" % (time.time() - start_time))

Start of epoch 0
Training loss (for one batch) at step 0: 0.7921
Seen so far: 64 samples
Training loss (for one batch) at step 200: 0.7755
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 0.1564
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.3181
Seen so far: 38464 samples
Training acc over epoch: 0.8788
Validation acc: 0.8866
Time taken: 1.59s

Start of epoch 1
Training loss (for one batch) at step 0: 0.5222
Seen so far: 64 samples
Training loss (for one batch) at step 200: 0.4574
Seen so far: 12864 samples
Training loss (for one batch) at step 400: 0.4035
Seen so far: 25664 samples
Training loss (for one batch) at step 600: 0.7561
Seen so far: 38464 samples
Training acc over epoch: 0.8959
Validation acc: 0.9028
Time taken: 1.27s

خیلی سریعتر، اینطور نیست؟

مدیریت سطح پایین تلفات ردیابی شده توسط مدل

لایه ها و مدل های به صورت بازگشتی پیگیری هر گونه ضرر و زیان ایجاد شده در طی پاس رو به جلو توسط لایه هایی که پاسخ self.add_loss(value) . لیست به دست آمده از ارزش از دست دادن اسکالر در دسترس از طریق مالکیت model.losses در پایان از پاس رو به جلو.

اگر می خواهید از این مولفه های ضرر استفاده کنید، باید آنها را جمع آوری کنید و در مرحله تمرینی خود به ضرر اصلی اضافه کنید.

این لایه را در نظر بگیرید که باعث از دست دادن منظم فعالیت می شود:

class ActivityRegularizationLayer(layers.Layer):
    def call(self, inputs):
        self.add_loss(1e-2 * tf.reduce_sum(inputs))
        return inputs

بیایید یک مدل واقعا ساده بسازیم که از آن استفاده کند:

inputs = keras.Input(shape=(784,), name="digits")
x = layers.Dense(64, activation="relu")(inputs)
# Insert activity regularization as a layer
x = ActivityRegularizationLayer()(x)
x = layers.Dense(64, activation="relu")(x)
outputs = layers.Dense(10, name="predictions")(x)

model = keras.Model(inputs=inputs, outputs=outputs)

در اینجا مرحله آموزشی ما اکنون باید چگونه باشد:

@tf.function
def train_step(x, y):
    with tf.GradientTape() as tape:
        logits = model(x, training=True)
        loss_value = loss_fn(y, logits)
        # Add any extra losses created during the forward pass.
        loss_value += sum(model.losses)
    grads = tape.gradient(loss_value, model.trainable_weights)
    optimizer.apply_gradients(zip(grads, model.trainable_weights))
    train_acc_metric.update_state(y, logits)
    return loss_value

خلاصه

اکنون همه چیزهایی را که باید در مورد استفاده از حلقه های آموزشی داخلی و نوشتن حلقه های خود از ابتدا بدانید، می دانید.

برای نتیجه‌گیری، در اینجا یک مثال ساده از انتها به انتها آورده شده است که همه چیزهایی را که در این راهنما آموخته‌اید به هم پیوند می‌دهد: یک DCGAN که بر روی ارقام MNIST آموزش دیده است.

مثال سرتاسری: یک حلقه آموزشی GAN از ابتدا

ممکن است با شبکه‌های متخاصم مولد (GAN) آشنا باشید. GAN ها می توانند با یادگیری توزیع پنهان مجموعه داده آموزشی از تصاویر («فضای پنهان» تصاویر) تصاویر جدیدی تولید کنند که تقریباً واقعی به نظر می رسند.

یک GAN از دو بخش ساخته شده است: یک مدل "مولد" که نقاط موجود در فضای پنهان را به نقاطی در فضای تصویر نگاشت می کند، یک مدل "مشخص کننده"، یک طبقه بندی کننده که می تواند تفاوت بین تصاویر واقعی (از مجموعه داده آموزشی) و جعلی را تشخیص دهد. تصاویر (خروجی شبکه ژنراتور).

یک حلقه آموزشی GAN به شکل زیر است:

1) تمایزگر را آموزش دهید. - دسته ای از نقاط تصادفی را در فضای پنهان نمونه برداری کنید. - از طریق مدل "ژنراتور" نقاط را به تصاویر جعلی تبدیل کنید. - مجموعه ای از تصاویر واقعی را دریافت کنید و آنها را با تصاویر تولید شده ترکیب کنید. - آموزش مدل "مشخص کننده" برای طبقه بندی تصاویر تولید شده در مقابل واقعی.

2) ژنراتور را آموزش دهید. - نمونه برداری از نقاط تصادفی در فضای پنهان. - تبدیل نقاط به تصاویر جعلی از طریق شبکه "ژنراتور". - مجموعه ای از تصاویر واقعی را دریافت کنید و آنها را با تصاویر تولید شده ترکیب کنید. - آموزش مدل "ژنراتور" برای "گول زدن" متمایز کننده و طبقه بندی تصاویر جعلی به عنوان واقعی.

برای یک نمای کلی با جزئیات بیشتری از گانز چگونه کار می کند، و یادگیری عمیق با پایتون .

بیایید این حلقه آموزشی را پیاده سازی کنیم. ابتدا، تفکیک‌کننده‌ای را ایجاد کنید که به‌منظور طبقه‌بندی ارقام جعلی و واقعی است:

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",
)
discriminator.summary()

Model: "discriminator"
_________________________________________________________________
Layer (type)                 Output Shape              Param #   
=================================================================
conv2d (Conv2D)              (None, 14, 14, 64)        640       
_________________________________________________________________
leaky_re_lu (LeakyReLU)      (None, 14, 14, 64)        0         
_________________________________________________________________
conv2d_1 (Conv2D)            (None, 7, 7, 128)         73856     
_________________________________________________________________
leaky_re_lu_1 (LeakyReLU)    (None, 7, 7, 128)         0         
_________________________________________________________________
global_max_pooling2d (Global (None, 128)               0         
_________________________________________________________________
dense_4 (Dense)              (None, 1)                 129       
=================================================================
Total params: 74,625
Trainable params: 74,625
Non-trainable params: 0
_________________________________________________________________

سپس اجازه دهید ایجاد یک شبکه ژنراتور، که تبدیل بردار نهفته به خروجی شکل (28, 28, 1) (به نمایندگی از رقم MNIST):

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",
)

نکته کلیدی اینجاست: حلقه آموزشی. همانطور که می بینید کاملاً ساده است. تابع مرحله آموزش فقط 17 خط طول می کشد.

# Instantiate one optimizer for the discriminator and another for the generator.
d_optimizer = keras.optimizers.Adam(learning_rate=0.0003)
g_optimizer = keras.optimizers.Adam(learning_rate=0.0004)

# Instantiate a loss function.
loss_fn = keras.losses.BinaryCrossentropy(from_logits=True)


@tf.function
def train_step(real_images):
    # Sample random points in the latent space
    random_latent_vectors = tf.random.normal(shape=(batch_size, latent_dim))
    # Decode them to fake images
    generated_images = 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((real_images.shape[0], 1))], axis=0
    )
    # Add random noise to the labels - important trick!
    labels += 0.05 * tf.random.uniform(labels.shape)

    # Train the discriminator
    with tf.GradientTape() as tape:
        predictions = discriminator(combined_images)
        d_loss = loss_fn(labels, predictions)
    grads = tape.gradient(d_loss, discriminator.trainable_weights)
    d_optimizer.apply_gradients(zip(grads, discriminator.trainable_weights))

    # Sample random points in the latent space
    random_latent_vectors = tf.random.normal(shape=(batch_size, 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 = discriminator(generator(random_latent_vectors))
        g_loss = loss_fn(misleading_labels, predictions)
    grads = tape.gradient(g_loss, generator.trainable_weights)
    g_optimizer.apply_gradients(zip(grads, generator.trainable_weights))
    return d_loss, g_loss, generated_images

اجازه دهید آموزش GAN ما، بارها و بارها خواستار train_step در دسته ای از تصاویر.

از آنجایی که متمایز کننده و مولد ما convnets هستند، شما می خواهید این کد را روی یک GPU اجرا کنید.

import os

# 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)

epochs = 1  # In practice you need at least 20 epochs to generate nice digits.
save_dir = "./"

for epoch in range(epochs):
    print("\nStart epoch", epoch)

    for step, real_images in enumerate(dataset):
        # Train the discriminator & generator on one batch of real images.
        d_loss, g_loss, generated_images = train_step(real_images)

        # Logging.
        if step % 200 == 0:
            # Print metrics
            print("discriminator loss at step %d: %.2f" % (step, d_loss))
            print("adversarial loss at step %d: %.2f" % (step, g_loss))

            # Save one generated image
            img = tf.keras.preprocessing.image.array_to_img(
                generated_images[0] * 255.0, scale=False
            )
            img.save(os.path.join(save_dir, "generated_img" + str(step) + ".png"))

        # To limit execution time we stop after 10 steps.
        # Remove the lines below to actually train the model!
        if step > 10:
            break

Start epoch 0
discriminator loss at step 0: 0.69
adversarial loss at step 0: 0.69

خودشه! فقط پس از 30 ثانیه آموزش روی پردازنده گرافیکی Colab، ارقام جعلی MNIST با ظاهری زیبا دریافت خواهید کرد.