אימון מותאם אישית עם tf.distribute.Strategy

מדריך זה מדגים כיצד להשתמש ב- tf.distribute.Strategy עם לולאות אימון מותאמות אישית. נכשיר מודל CNN פשוט על מערך הנתונים MNIST האופנתי. מערך MNIST האופנתי מכיל 60000 תמונות רכבת בגודל 28 x 28 ו 10000 תמונות בדיקה בגודל 28 x 28.

אנו משתמשים בלולאות אימון מותאמות אישית כדי לאמן את המודל שלנו מכיוון שהם נותנים לנו גמישות ושליטה רבה יותר באימונים. יתר על כן, קל יותר לנקות באגים על המודל ולולאת האימון.

# Import TensorFlow
import tensorflow as tf

# Helper libraries
import numpy as np
import os

print(tf.__version__)

2.3.0

הורד את מערך הנתונים MNIST האופנתי

fashion_mnist = tf.keras.datasets.fashion_mnist

(train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data()

# Adding a dimension to the array -> new shape == (28, 28, 1)
# We are doing this because the first layer in our model is a convolutional
# layer and it requires a 4D input (batch_size, height, width, channels).
# batch_size dimension will be added later on.
train_images = train_images[..., None]
test_images = test_images[..., None]

# Getting the images in [0, 1] range.
train_images = train_images / np.float32(255)
test_images = test_images / np.float32(255)

Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/train-labels-idx1-ubyte.gz
32768/29515 [=================================] - 0s 0us/step
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/train-images-idx3-ubyte.gz
26427392/26421880 [==============================] - 1s 0us/step
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/t10k-labels-idx1-ubyte.gz
8192/5148 [===============================================] - 0s 0us/step
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/t10k-images-idx3-ubyte.gz
4423680/4422102 [==============================] - 1s 0us/step

צור אסטרטגיה להפצת המשתנים והגרף

כיצד פועלת אסטרטגיית tf.distribute.MirroredStrategy ?

כל המשתנים וגרף המודל משוכפלים על העתקים.
הקלט מופץ באופן שווה על פני העתקים.
כל עותק משוכפל מחשב את האובדן והשיפועים עבור הקלט שקיבל.
השיפועים מסונכרנים בין כל העתקים על ידי סיכומם.
לאחר הסנכרון, אותו עדכון נעשה בעותקי המשתנים בכל העתק.

# If the list of devices is not specified in the
# `tf.distribute.MirroredStrategy` constructor, it will be auto-detected.
strategy = tf.distribute.MirroredStrategy()

INFO:tensorflow:Using MirroredStrategy with devices ('/job:localhost/replica:0/task:0/device:GPU:0',)

print ('Number of devices: {}'.format(strategy.num_replicas_in_sync))

Number of devices: 1

הגדרת צינור קלט

ייצא את הגרף והמשתנים לפורמט SavedModel לפלטפורמה-אגנוסטית. לאחר שמירת המודל שלך, תוכל לטעון אותו עם או בלי ההיקף.

BUFFER_SIZE = len(train_images)

BATCH_SIZE_PER_REPLICA = 64
GLOBAL_BATCH_SIZE = BATCH_SIZE_PER_REPLICA * strategy.num_replicas_in_sync

EPOCHS = 10

צור את מערכי הנתונים והפיץ אותם:

train_dataset = tf.data.Dataset.from_tensor_slices((train_images, train_labels)).shuffle(BUFFER_SIZE).batch(GLOBAL_BATCH_SIZE) 
test_dataset = tf.data.Dataset.from_tensor_slices((test_images, test_labels)).batch(GLOBAL_BATCH_SIZE) 

train_dist_dataset = strategy.experimental_distribute_dataset(train_dataset)
test_dist_dataset = strategy.experimental_distribute_dataset(test_dataset)

צור את המודל

צור מודל באמצעות tf.keras.Sequential . אתה יכול גם להשתמש ב- API של Subclassing Model לצורך ביצוע פעולה זו.

def create_model():
  model = tf.keras.Sequential([
      tf.keras.layers.Conv2D(32, 3, activation='relu'),
      tf.keras.layers.MaxPooling2D(),
      tf.keras.layers.Conv2D(64, 3, activation='relu'),
      tf.keras.layers.MaxPooling2D(),
      tf.keras.layers.Flatten(),
      tf.keras.layers.Dense(64, activation='relu'),
      tf.keras.layers.Dense(10)
    ])

  return model

# Create a checkpoint directory to store the checkpoints.
checkpoint_dir = './training_checkpoints'
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt")

הגדר את פונקציית האובדן

בדרך כלל, במכונה אחת עם GPU / CPU אחד, ההפסד מחולק במספר הדוגמאות בקבוצת הקלט.

אז איך צריך לחשב את ההפסד בעת שימוש ב- tf.distribute.Strategy ?

לדוגמא, נניח שיש לך 4 גרפי GPU וגודל אצווה של 64. קבוצה אחת של קלט מופצת על פני העתקים (4 GPU), כאשר כל העתק מקבל קלט בגודל 16.
המודל בכל העתק מבצע מעבר קדימה עם הקלט המתאים לו ומחשב את ההפסד. כעת, במקום לחלק את ההפסד במספר הדוגמאות בקלט המתאים לו (BATCH_SIZE_PER_REPLICA = 16), יש לחלק את ההפסד ב- GLOBAL_BATCH_SIZE (64).

מדוע לעשות זאת?

זה צריך להיעשות מכיוון שלאחר חישוב הדרגתיות על כל העתק, הם מסונכרנים על ידי העתקים על ידי סיכומם .

איך לעשות זאת ב- TensorFlow?

אם אתה כותב לולאת אימונים מותאמת אישית, כמו במדריך זה, עליך לסכם את ההפסדים לדוגמא ולחלק את הסכום ב- GLOBAL_BATCH_SIZE: scale_loss = tf.reduce_sum(loss) * (1. / GLOBAL_BATCH_SIZE) או שאתה יכול להשתמש ב- tf.nn.compute_average_loss אשר לוקח את האובדן לכל דוגמה, משקולות הדוגמה האופציונליות ו- GLOBAL_BATCH_SIZE כארגומנטים ומחזיר את האובדן המוקטן.
אם אתה משתמש בהפסדי רגולציה במודל שלך, עליך לשנות את ערך ההפסד לפי מספר העתקים. אתה יכול לעשות זאת באמצעות הפונקציה tf.nn.scale_regularization_loss .
השימוש ב- tf.reduce_mean אינו מומלץ. פעולה זו מחלקת את ההפסד לפי גודל אצווה משוכפל בפועל שעשוי להשתנות שלב אחר שלב.
צמצום model.compile המידה הזה מתבצעים באופן אוטומטי ב- model.compile model.fit וב- model.fit
אם משתמשים בכיתות tf.keras.losses (כמו בדוגמה שלהלן), יש לציין במפורש את צמצום ההפסד כדי להיות אחד מ- NONE או SUM . AUTO ו- SUM_OVER_BATCH_SIZE אינם מורשים בשימוש עם tf.distribute.Strategy . אסור להשתמש ב- AUTO מכיוון שהמשתמש צריך לחשוב במפורש איזו הפחתה הוא רוצה לוודא שהיא נכונה במקרה המבוזר. SUM_OVER_BATCH_SIZE אסור מכיוון שכרגע הוא יחלק רק לפי גודל אצווה עותק, וישאיר את החלוקה לפי מספר העתקים למשתמש, שעלול להיות קל לפספס. אז במקום זאת אנו מבקשים מהמשתמש לעשות את הצמצום בעצמו במפורש.
אם labels הן רב מימדיות, אז ממוצע per_example_loss פני מספר האלמנטים בכל מדגם. לדוגמה, אם צורת predictions היא (batch_size, H, W, n_classes) labels הן (batch_size, H, W) , תצטרך לעדכן per_example_loss כמו: per_example_loss /= tf.cast(tf.reduce_prod(tf.shape(labels)[1:]), tf.float32)

with strategy.scope():
  # Set reduction to `none` so we can do the reduction afterwards and divide by
  # global batch size.
  loss_object = tf.keras.losses.SparseCategoricalCrossentropy(
      from_logits=True,
      reduction=tf.keras.losses.Reduction.NONE)
  def compute_loss(labels, predictions):
    per_example_loss = loss_object(labels, predictions)
    return tf.nn.compute_average_loss(per_example_loss, global_batch_size=GLOBAL_BATCH_SIZE)

הגדר את המדדים למעקב אחר אובדן ודיוק

מדדים אלה עוקבים אחר אובדן הבדיקה ואימון ודיוק המבחנים. אתה יכול להשתמש ב .result() כדי לקבל את הנתונים הסטטיסטיים שנצברו בכל עת.

with strategy.scope():
  test_loss = tf.keras.metrics.Mean(name='test_loss')

  train_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(
      name='train_accuracy')
  test_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(
      name='test_accuracy')

לולאת אימונים

# model, optimizer, and checkpoint must be created under `strategy.scope`.
with strategy.scope():
  model = create_model()

  optimizer = tf.keras.optimizers.Adam()

  checkpoint = tf.train.Checkpoint(optimizer=optimizer, model=model)

def train_step(inputs):
  images, labels = inputs

  with tf.GradientTape() as tape:
    predictions = model(images, training=True)
    loss = compute_loss(labels, predictions)

  gradients = tape.gradient(loss, model.trainable_variables)
  optimizer.apply_gradients(zip(gradients, model.trainable_variables))

  train_accuracy.update_state(labels, predictions)
  return loss 

def test_step(inputs):
  images, labels = inputs

  predictions = model(images, training=False)
  t_loss = loss_object(labels, predictions)

  test_loss.update_state(t_loss)
  test_accuracy.update_state(labels, predictions)

# `run` replicates the provided computation and runs it
# with the distributed input.
@tf.function
def distributed_train_step(dataset_inputs):
  per_replica_losses = strategy.run(train_step, args=(dataset_inputs,))
  return strategy.reduce(tf.distribute.ReduceOp.SUM, per_replica_losses,
                         axis=None)

@tf.function
def distributed_test_step(dataset_inputs):
  return strategy.run(test_step, args=(dataset_inputs,))

for epoch in range(EPOCHS):
  # TRAIN LOOP
  total_loss = 0.0
  num_batches = 0
  for x in train_dist_dataset:
    total_loss += distributed_train_step(x)
    num_batches += 1
  train_loss = total_loss / num_batches

  # TEST LOOP
  for x in test_dist_dataset:
    distributed_test_step(x)

  if epoch % 2 == 0:
    checkpoint.save(checkpoint_prefix)

  template = ("Epoch {}, Loss: {}, Accuracy: {}, Test Loss: {}, "
              "Test Accuracy: {}")
  print (template.format(epoch+1, train_loss,
                         train_accuracy.result()*100, test_loss.result(),
                         test_accuracy.result()*100))

  test_loss.reset_states()
  train_accuracy.reset_states()
  test_accuracy.reset_states()

WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tensorflow/python/data/ops/multi_device_iterator_ops.py:601: get_next_as_optional (from tensorflow.python.data.ops.iterator_ops) is deprecated and will be removed in a future version.
Instructions for updating:
Use `tf.data.Iterator.get_next_as_optional()` instead.
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
Epoch 1, Loss: 0.50295090675354, Accuracy: 82.1116714477539, Test Loss: 0.3852590322494507, Test Accuracy: 86.5999984741211
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
INFO:tensorflow:Reduce to /job:localhost/replica:0/task:0/device:CPU:0 then broadcast to ('/job:localhost/replica:0/task:0/device:CPU:0',).
Epoch 2, Loss: 0.32958829402923584, Accuracy: 88.20333862304688, Test Loss: 0.3391425311565399, Test Accuracy: 87.6500015258789
Epoch 3, Loss: 0.2872008979320526, Accuracy: 89.57167053222656, Test Loss: 0.2974696457386017, Test Accuracy: 89.31000518798828
Epoch 4, Loss: 0.255713552236557, Accuracy: 90.58499908447266, Test Loss: 0.2988712787628174, Test Accuracy: 89.31999969482422
Epoch 5, Loss: 0.23122134804725647, Accuracy: 91.41667175292969, Test Loss: 0.27742496132850647, Test Accuracy: 89.99000549316406
Epoch 6, Loss: 0.212575763463974, Accuracy: 92.17333221435547, Test Loss: 0.2573488652706146, Test Accuracy: 90.75
Epoch 7, Loss: 0.1963273137807846, Accuracy: 92.77166748046875, Test Loss: 0.2587501108646393, Test Accuracy: 90.66000366210938
Epoch 8, Loss: 0.1779220998287201, Accuracy: 93.46666717529297, Test Loss: 0.267805814743042, Test Accuracy: 90.55999755859375
Epoch 9, Loss: 0.16410504281520844, Accuracy: 93.91333770751953, Test Loss: 0.25632956624031067, Test Accuracy: 91.00999450683594
Epoch 10, Loss: 0.14829590916633606, Accuracy: 94.47833251953125, Test Loss: 0.25820475816726685, Test Accuracy: 91.00999450683594

דברים שיש לציין בדוגמה לעיל:

אנו train_dist_dataset על train_dist_dataset test_dist_dataset באמצעות test_dist_dataset for x in ...
ההפסד המוקטן הוא ערך ההחזר של distributed_train_step . ערך זה נצבר על פני עותקים משוכפלים באמצעות השיחה tf.distribute.Strategy.reduce ואז על פני קבוצות על ידי סיכום ערך ההחזר של שיחות tf.distribute.Strategy.reduce .
tf.keras.Metrics צריך להיות מעודכן בתוך train_step ו- test_step tf.distribute.Strategy.run ידי tf.distribute.Strategy.run . * tf.distribute.Strategy.run מחזיר תוצאות מכל עותק משוכפל מקומי באסטרטגיה, וישנן מספר דרכים לצרוך תוצאה זו. אתה יכול לעשות tf.distribute.Strategy.reduce כדי לקבל ערך מצטבר. ניתן גם לבצע tf.distribute.Strategy.experimental_local_results כדי לקבל את רשימת הערכים הכלולים בתוצאה, אחד לכל עותק משוכפל.

שחזר את המחסום האחרון ובדוק

ניתן לשחזר מודל tf.distribute.Strategy עם tf.distribute.Strategy עם או בלי אסטרטגיה.

eval_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(
      name='eval_accuracy')

new_model = create_model()
new_optimizer = tf.keras.optimizers.Adam()

test_dataset = tf.data.Dataset.from_tensor_slices((test_images, test_labels)).batch(GLOBAL_BATCH_SIZE)

@tf.function
def eval_step(images, labels):
  predictions = new_model(images, training=False)
  eval_accuracy(labels, predictions)

checkpoint = tf.train.Checkpoint(optimizer=new_optimizer, model=new_model)
checkpoint.restore(tf.train.latest_checkpoint(checkpoint_dir))

for images, labels in test_dataset:
  eval_step(images, labels)

print ('Accuracy after restoring the saved model without strategy: {}'.format(
    eval_accuracy.result()*100))

Accuracy after restoring the saved model without strategy: 91.00999450683594

דרכים חלופיות לחזור על מערך נתונים

שימוש באיטרטורים

אם ברצונך לחזור על מספר צעדים נתון ולא דרך מערך הנתונים כולו, תוכל ליצור איטרטור באמצעות שיחת iter המפורש next באיטרציה. אתה יכול לבחור לבצע איטרציה מעל מערך הנתונים הן בתוך tf.function והן מחוצה לו. הנה קטע קטן המדגים איטרציה של מערך הנתונים מחוץ ל- tf.function באמצעות איטרטור.

for _ in range(EPOCHS):
  total_loss = 0.0
  num_batches = 0
  train_iter = iter(train_dist_dataset)

  for _ in range(10):
    total_loss += distributed_train_step(next(train_iter))
    num_batches += 1
  average_train_loss = total_loss / num_batches

  template = ("Epoch {}, Loss: {}, Accuracy: {}")
  print (template.format(epoch+1, average_train_loss, train_accuracy.result()*100))
  train_accuracy.reset_states()

Epoch 10, Loss: 0.12157603353261948, Accuracy: 95.0
Epoch 10, Loss: 0.1367541253566742, Accuracy: 94.6875
Epoch 10, Loss: 0.14902949333190918, Accuracy: 93.90625
Epoch 10, Loss: 0.12149540334939957, Accuracy: 95.625
Epoch 10, Loss: 0.13160167634487152, Accuracy: 94.6875
Epoch 10, Loss: 0.13297739624977112, Accuracy: 95.3125
Epoch 10, Loss: 0.16038034856319427, Accuracy: 94.53125
Epoch 10, Loss: 0.1035340279340744, Accuracy: 96.40625
Epoch 10, Loss: 0.11846740543842316, Accuracy: 95.625
Epoch 10, Loss: 0.09006750583648682, Accuracy: 96.71875

מתבוסס בתוך פונקצית tf

ניתן גם לחזור על כל train_dist_dataset הקלט train_dist_dataset בתוך train_dist_dataset tf באמצעות הפונקציה for x in ... או על ידי יצירת איטרטורים כמו שעשינו לעיל. הדוגמה שלהלן מדגימה עוטפת תקופת אימונים אחת בתפקוד tf. train_dist_dataset על train_dist_dataset בתוך הפונקציה.

@tf.function
def distributed_train_epoch(dataset):
  total_loss = 0.0
  num_batches = 0
  for x in dataset:
    per_replica_losses = strategy.run(train_step, args=(x,))
    total_loss += strategy.reduce(
      tf.distribute.ReduceOp.SUM, per_replica_losses, axis=None)
    num_batches += 1
  return total_loss / tf.cast(num_batches, dtype=tf.float32)

for epoch in range(EPOCHS):
  train_loss = distributed_train_epoch(train_dist_dataset)

  template = ("Epoch {}, Loss: {}, Accuracy: {}")
  print (template.format(epoch+1, train_loss, train_accuracy.result()*100))

  train_accuracy.reset_states()

Epoch 1, Loss: 0.13680464029312134, Accuracy: 94.90499877929688
Epoch 2, Loss: 0.12503673136234283, Accuracy: 95.33499908447266
Epoch 3, Loss: 0.11472766101360321, Accuracy: 95.71333312988281
Epoch 4, Loss: 0.10419528931379318, Accuracy: 96.13500213623047
Epoch 5, Loss: 0.09566374123096466, Accuracy: 96.44833374023438
Epoch 6, Loss: 0.08704081922769547, Accuracy: 96.82499694824219
Epoch 7, Loss: 0.08157625794410706, Accuracy: 96.96333312988281
Epoch 8, Loss: 0.07562965154647827, Accuracy: 97.11000061035156
Epoch 9, Loss: 0.0676642507314682, Accuracy: 97.47999572753906
Epoch 10, Loss: 0.06430575996637344, Accuracy: 97.58333587646484

מעקב אחר אובדן אימונים בין עותקים משוכפלים

אנו לא ממליצים להשתמש ב- tf.metrics.Mean כדי לעקוב אחר אובדן האימונים על פני העתקים שונים, בגלל חישוב קנה המידה של האובדן שמתבצע.

לדוגמה, אם אתה מנהל עבודת הדרכה עם המאפיינים הבאים:

שתי העתקות
שתי דוגמאות מעובדות על כל העתק
ערכי אובדן כתוצאה: [2, 3] ו- [4, 5] בכל העתק
גודל אצווה גלובלי = 4

עם קנה המידה של ההפסד, אתה מחשב את הערך לכל דגימה של אובדן בכל עותק משוכפל על ידי הוספת ערכי ההפסד, ואז מחלק לפי גודל האצווה הגלובלי. במקרה זה: (2 + 3) / 4 = 1.25 ו- (4 + 5) / 4 = 2.25 .

אם אתה משתמש ב- tf.metrics.Mean כדי לעקוב אחר אובדן בשני העתקים, התוצאה שונה. בדוגמה זו, בסופו של דבר אתה מגיע total של 3.50 count של 2, מה שמביא total / count = 1.75 כאשר result() נקראת במדד. הפסד המחושב באמצעות tf.keras.Metrics ידי גורם נוסף השווה למספר העתקים המסונכרנים.

מדריך ודוגמאות

להלן מספר דוגמאות לשימוש באסטרטגיית הפצה עם לולאות אימון מותאמות אישית:

מדריך הדרכה מבוזר
דוגמה של DenseNet באמצעות MirroredStrategy .
דוגמה ל- BERT שהוכשרה באמצעות MirroredStrategy ו- TPUStrategy . דוגמה זו מועילה במיוחד להבנת כיצד להיטען ממחסום וליצור מחסומים תקופתיים במהלך אימונים מבוזרים וכו '.
דוגמה ל- NCF שהוכשרה באמצעות MirroredStrategy שניתן להפעיל באמצעות הדגל keras_use_ctl .
דוגמה ל- NMT שהוכשרה באמצעות MirroredStrategy .

דוגמאות נוספות המופיעות במדריך האסטרטגיה להפצה .

הצעדים הבאים

נסה את ה- tf.distribute.Strategy API החדש בדגמים שלך.
בקר בקטע ביצועים במדריך כדי ללמוד עוד על אסטרטגיות וכלים אחרים שבהם תוכל להשתמש כדי לייעל את הביצועים של דגמי TensorFlow שלך.