TensorFlow.org에서 보기 | Google Colab에서 실행 | GitHub에서 소스 보기 | 노트북 다운로드 |
!pip install -U tf-hub-nightly
import tensorflow_hub as hub
from tensorflow.keras import layers
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
2022-12-14 22:27:51.649540: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer.so.7'; dlerror: libnvinfer.so.7: cannot open shared object file: No such file or directory 2022-12-14 22:27:51.649637: W tensorflow/compiler/xla/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libnvinfer_plugin.so.7'; dlerror: libnvinfer_plugin.so.7: cannot open shared object file: No such file or directory 2022-12-14 22:27:51.649646: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Cannot dlopen some TensorRT libraries. If you would like to use Nvidia GPU with TensorRT, please make sure the missing libraries mentioned above are installed properly.
시작하기
Keras는 기본 학습 및 평가 루프인 fit()
및 evaluate()
를 제공합니다. 이들의 사용법은 내장 메서드를 사용한 학습 및 평가 가이드에서 다룹니다.
fit()
의 편리함을 그대로 활용하면서 모델의 학습 알고리즘을 사용자 정의하려면(예: fit()
을 사용하여 GAN 학습 진행) Model
클래스를 하위 클래스로 만들고 fit()
중에 반복적으로 호출되는 고유한 train_step()
메서드를 구현합니다. 이 내용은 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)
우리의 학습 루프는 다음과 같습니다.
- Epoch를 반복하는
for
루프를 엽니다. - 각 epoch에 대해 데이터세트를 배치 단위로 반복하는
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 WARNING:tensorflow:5 out of the last 5 calls to <function _BaseOptimizer._update_step_xla at 0x7fbf935c6430> triggered tf.function retracing. Tracing is expensive and the excessive number of tracings could be due to (1) creating @tf.function repeatedly in a loop, (2) passing tensors with different shapes, (3) passing Python objects instead of tensors. For (1), please define your @tf.function outside of the loop. For (2), @tf.function has reduce_retracing=True option that can avoid unnecessary retracing. For (3), please refer to https://www.tensorflow.org/guide/function#controlling_retracing and https://www.tensorflow.org/api_docs/python/tf/function for more details. WARNING:tensorflow:6 out of the last 6 calls to <function _BaseOptimizer._update_step_xla at 0x7fbf935c6430> triggered tf.function retracing. Tracing is expensive and the excessive number of tracings could be due to (1) creating @tf.function repeatedly in a loop, (2) passing tensors with different shapes, (3) passing Python objects instead of tensors. For (1), please define your @tf.function outside of the loop. For (2), @tf.function has reduce_retracing=True option that can avoid unnecessary retracing. For (3), please refer to https://www.tensorflow.org/guide/function#controlling_retracing and https://www.tensorflow.org/api_docs/python/tf/function for more details. Training loss (for one batch) at step 0: 114.2052 Seen so far: 64 samples Training loss (for one batch) at step 200: 1.4316 Seen so far: 12864 samples Training loss (for one batch) at step 400: 1.0915 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.7470 Seen so far: 38464 samples Start of epoch 1 Training loss (for one batch) at step 0: 0.5563 Seen so far: 64 samples Training loss (for one batch) at step 200: 1.1355 Seen so far: 12864 samples Training loss (for one batch) at step 400: 0.8665 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.7169 Seen so far: 38464 samples
메트릭 로우 레벨(low-level) 처리
이 기본 루프에 메트릭 모니터링을 추가해 보겠습니다.
처음부터 작성한 이러한 학습 루프에서 내장 메트릭(또는 사용자가 작성한 메트릭)을 쉽게 재사용할 수 있습니다. 흐름은 다음과 같습니다.
- 루프 시작 시 메트릭 인스턴스화
- 각 배치 후에
metric.update_state()
를 호출 - 메트릭의 현재 값을 표시해야 하는 경우 {code 0}matric.result(){/code 0}를 호출
- 메트릭의 상태를 삭제해야 할 경우(일반적으로 Epoch 종료 시)
metric.reset_states()
를 호출
이 지식을 사용하여 각 Epoch가 끝날 때 검증 데이터의 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: 106.2340 Seen so far: 64 samples Training loss (for one batch) at step 200: 1.4773 Seen so far: 12864 samples Training loss (for one batch) at step 400: 0.9677 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.4914 Seen so far: 38464 samples Training acc over epoch: 0.7020 Validation acc: 0.7567 Time taken: 10.50s Start of epoch 1 Training loss (for one batch) at step 0: 0.7326 Seen so far: 64 samples Training loss (for one batch) at step 200: 0.9143 Seen so far: 12864 samples Training loss (for one batch) at step 400: 0.4277 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.7565 Seen so far: 38464 samples Training acc over epoch: 0.8232 Validation acc: 0.8578 Time taken: 10.33s
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.4228 Seen so far: 64 samples Training loss (for one batch) at step 200: 0.6054 Seen so far: 12864 samples Training loss (for one batch) at step 400: 0.2916 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.5421 Seen so far: 38464 samples Training acc over epoch: 0.8574 Validation acc: 0.8583 Time taken: 1.75s Start of epoch 1 Training loss (for one batch) at step 0: 0.3737 Seen so far: 64 samples Training loss (for one batch) at step 200: 0.5216 Seen so far: 12864 samples Training loss (for one batch) at step 400: 0.3324 Seen so far: 25664 samples Training loss (for one batch) at step 600: 0.2431 Seen so far: 38464 samples Training acc over epoch: 0.8770 Validation acc: 0.8565 Time taken: 1.31s
훨씬 빨라지지 않았나요?
모델에서 추적한 손실의 로우 레벨 처리
레이어 및 모델은 self.add_loss(value)
를 호출하는 레이어로 순방향 전달을 수행하는 동안 생성된 손실을 재귀적으로 추적합니다. Scalar 손실 값의 결과 목록은 순방향 전달 종료 시 속성 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
요약
이제 내장 학습 루프를 사용하고 처음부터 자체적으로 작성하기 위해 알아야 할 모든 것을 알게 되었습니다.
결론적으로 다음은 이 가이드에서 배운 모든 것을 하나로 묶는 간단한 엔드 투 엔드 예제인 MNIST 숫자로 학습한 DCGAN 입니다.
엔드 투 엔드 예제: GAN 학습 루프 처음부터 수행하기
GAN(Generative Adversarial Networks)에 대해 잘 알고 있을 것입니다. GAN은 이미지 학습 데이터세트의 잠재 분포(이미지의 "잠재 공간")를 훈련하여 거의 실제처럼 보이는 새로운 이미지를 생성할 수 있습니다.
GAN은 잠재 공간의 지점을 이미지 공간의 지점으로 매핑하는 "생성기" 모델과 실제 이미지(학습 데이터 세트)와 가짜 이미지(생성기 네트워크의 출력물)를 구별할 수 있는 분류자인 "판별기" 모델의 두 부분으로 구성됩니다.
GAN 학습 루프는 다음과 같습니다.
- 판별기를 훈련합니다.
- 잠재 공간에 무작위 지점 배치를 샘플링합니다.
- "생성기" 모델을 통해 지점을 가짜 이미지로 바꿉니다.
- 실제 이미지 배치를 가져와서 생성된 이미지와 결합합니다.
- 생성된 이미지와 실제 이미지를 분류하기 위해 "판별기" 모델을 훈련합니다.
- 생성기를 훈련합니다.
- 잠재 공간에 무작위 지점을 샘플링합니다.
- "생성기"를 통해 지점을 가짜 이미지로 바꿉니다.
- 실제 이미지 배치를 가져와서 생성된 이미지와 결합합니다.
- "생성기" 모델을 훈련시켜 판별기를 "속이고" 가짜 이미지를 진짜로 분류합니다.
GAN의 작동 방식에 대한 더 자세한 개요는 Deep Learning with Python(Python으로 딥러닝하기)을 참조하세요.
이 학습 루프를 구현해 보겠습니다. 먼저 가짜 숫자와 실제 숫자를 구분하기 위한 판별기를 생성합니다.
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 (Globa (None, 128) 0 lMaxPooling2D) 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
이미지 배치에서 train_step
을 반복적으로 호출하여 GAN을 학습시켜 보겠습니다.
판별기와 생성기가 ConvNet이므로 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.68 adversarial loss at step 0: 0.68
이게 전부입니다! Colab GPU에서 30초 정도만 훈련하면 멋진 가짜 MNIST 숫자를 얻게 됩니다.