<jupyter_start><jupyter_text>MNIST: Custom training: walkthrough- https://www.tensorflow.org/tutorials/customization/custom_training_walkthrough   0. 환경 셋업<jupyter_code>import os
import matplotlib.pyplot as plt

import tensorflow as tf
tf.random.set_seed(42)
print("TensorFlow version: {}".format(tf.__version__))
print("Eager execution: {}".format(tf.executing_eagerly()))
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
  # 텐서플로가 첫 번째 GPU만 사용하도록 제한
  try:
    tf.config.experimental.set_visible_devices(gpus[6], 'GPU')
  except RuntimeError as e:
    # 프로그램 시작시에 접근 가능한 장치가 설정되어야만 합니다
    print(e)<jupyter_output><empty_output><jupyter_text>1. 데이터 준비<jupyter_code>(mnist_images, mnist_labels), (x_test, y_test) = tf.keras.datasets.mnist.load_data(
    path="mnist-%d.npz" % 10
)


print("type: ", type(mnist_images))
print("shape: ", mnist_images.shape)
print("mnist_labels: ", mnist_labels.shape)

print("x_test: ", x_test.shape)
print("y_test: ", y_test.shape)
def create_train_dataset(mnist_images, mnist_labels, batch_size, buffer_size):
    dataset = tf.data.Dataset.from_tensor_slices(
        (tf.cast(mnist_images[..., tf.newaxis] / 255.0, tf.float32), 
         tf.cast(mnist_labels, tf.int64))
    )
    dataset = dataset.repeat(1).shuffle(buffer_size).batch(batch_size)
    print("# of batches : {0}".format(tf.data.experimental.cardinality(dataset).numpy()))

    return dataset
    

def create_test_dataset(mnist_images, mnist_labels, batch_size):
    dataset = tf.data.Dataset.from_tensor_slices(
        (tf.cast(mnist_images[..., tf.newaxis] / 255.0, tf.float32), 
         tf.cast(mnist_labels, tf.int64))
    )
    dataset = dataset.repeat(1).batch(batch_size)
    print("# of batches : {0}".format(tf.data.experimental.cardinality(dataset).numpy()))

    return dataset


batch_size = 256
buffer_size = mnist_images.shape[0]
print("buffer_size: ", buffer_size)
dataset = create_train_dataset(mnist_images, mnist_labels,batch_size, buffer_size)
    
# dataset = dataset.repeat(1).shuffle(buffer_size).batch(batch_size)
# print(tf.data.experimental.cardinality(dataset))
# print(tf.data.experimental.cardinality(dataset).numpy())
dataset
batch_size = x_test.shape[0]
test_ds = create_test_dataset(x_test, y_test,batch_size)
test_ds
# features, labels = next(iter(dataset))
# print(features)<jupyter_output><empty_output><jupyter_text>2. 모델 정의<jupyter_code>mnist_model = tf.keras.Sequential(
    [
        tf.keras.layers.Conv2D(32, [3, 3], activation="relu"),
        tf.keras.layers.Conv2D(64, [3, 3], activation="relu"),
        tf.keras.layers.MaxPooling2D(pool_size=(2, 2)),
        tf.keras.layers.Dropout(0.25),
        tf.keras.layers.Flatten(),
        tf.keras.layers.Dense(128, activation="relu"),
        tf.keras.layers.Dropout(0.5),
        tf.keras.layers.Dense(10, activation="softmax"),
    ]
)<jupyter_output><empty_output><jupyter_text>3. 모델 생성  Loss 오브젝트 정의<jupyter_code>loss = tf.losses.SparseCategoricalCrossentropy()
opt = tf.optimizers.Adam(0.000125 * 8)
checkpoint = tf.train.Checkpoint(model=mnist_model, optimizer=opt)<jupyter_output><empty_output><jupyter_text>Gradient 생성 함수 정의<jupyter_code>@tf.function
def training_step(images, labels, first_batch):
    with tf.GradientTape() as tape:
        probs = mnist_model(images, training=True)
        print("probs: ", probs)
        print("labels: ", labels)        
        
        loss_value = loss(labels, probs)
        
    grads = tape.gradient(loss_value, mnist_model.trainable_variables)
    opt.apply_gradients(zip(grads, mnist_model.trainable_variables))
    
    return loss_value
test_loss = tf.keras.metrics.Mean(name='test_loss')
test_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='test_accuracy')

@tf.function
def test_step(images, labels):
    # training=False is only needed if there are layers with different
    # behavior during training versus inference (e.g. Dropout).
    predictions = mnist_model(images, training=False)
#   print("labels: ", labels)
#   print("predictions: ", predictions)    
    
    t_loss = loss(labels, predictions)

    test_loss(t_loss)
    test_accuracy(labels, predictions)
    return t_loss<jupyter_output><empty_output><jupyter_text>4. 모델 훈련<jupyter_code>EPOCHS = 5
print_interval = 100

for epoch in range(EPOCHS):
    for batch, (images, labels) in enumerate(dataset):
        loss_value = training_step(images, labels, batch == 0)

        if batch % print_interval == 0:
            print("Epoch: %d\t, Step #%d\tLoss: %.6f" %  (epoch, batch, loss_value))
            
    # Reset the metrics at the start of the next epoch
    test_loss.reset_states()
    test_accuracy.reset_states()


    for test_images, test_labels in test_ds:
        test_step(test_images, test_labels)

        print(
        #ㅁ    f'Epoch {epoch + 1}, '
        f'Test Loss: {test_loss.result()}, '
        f'Test Accuracy: {test_accuracy.result() * 100}'
        )

            
# #for epoch in range(EPOCHS):
# n_batch = 1000
# print_interval = 1000
# for batch, (images, labels) in enumerate(dataset.take(n_batch)):
#     loss_value = training_step(images, labels, batch == 0)

#     if batch % print_interval == 0:
#         print("Step #%d\tLoss: %.6f" %  (batch, loss_value))
        
import os
log_dir = 'log'
os.makedirs(log_dir, exist_ok=True)
checkpoint_dir = log_dir
mnist_model.save(os.path.join(checkpoint_dir, "1"))
! rm -rf log_dir<jupyter_output><empty_output><jupyter_text>5. 추론<jupyter_code># Reset the metrics at the start of the next epoch
test_loss.reset_states()
test_accuracy.reset_states()


for test_images, test_labels in test_ds:
    test_step(test_images, test_labels)

    print(
    #ㅁ    f'Epoch {epoch + 1}, '
    f'Test Loss: {test_loss.result()}, '
    f'Test Accuracy: {test_accuracy.result() * 100}'
    )<jupyter_output>Test Loss: 0.03226584568619728, Test Accuracy: 98.94000244140625
{'fraction_non_alphanumeric': 0.13190549039605642, 'fraction_numerical': 0.021927587965323816, 'mean_word_length': 4.2024756852343055, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 28, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '597429', 'license': 'MIT', 'n_tokens_mistral': 2297, 'n_tokens_neox': 2233, 'n_words': 376, 'path': 'code/phase0/working/TF-WalkThrough/Z.REF-MNIST-Training.ipynb', 'repo_name': 'gonsoomoon-ml/SageMaker-Tensorflow-Step-By-Step', 'size': '5883'}