Debug della pipeline di addestramento migrata TF2

Visualizza su TensorFlow.org Esegui in Google Colab Visualizza l'origine su GitHub Scarica quaderno

Questo notebook mostra come eseguire il debug della pipeline di addestramento durante la migrazione a TF2. Si compone dei seguenti componenti:

  1. Passaggi consigliati ed esempi di codice per il debug della pipeline di addestramento
  2. Strumenti per il debug
  3. Altre risorse correlate

Un presupposto è che tu abbia codice TF1.x e modelli addestrati per il confronto e desideri creare un modello TF2 che ottenga una precisione di convalida simile.

Questo notebook NON copre i problemi di prestazioni di debug per la velocità di training/inferenza o l'utilizzo della memoria.

Flusso di lavoro di debug

Di seguito è riportato un flusso di lavoro generale per il debug delle pipeline di addestramento TF2. Nota che non è necessario seguire questi passaggi in ordine. È inoltre possibile utilizzare un approccio di ricerca binaria in cui si testa il modello in un passaggio intermedio e si restringe l'ambito del debug.

  1. Correggi gli errori di compilazione e di runtime

  2. Convalida del singolo passaggio in avanti (in una guida separata)

    un. Su un singolo dispositivo CPU

    • Verifica che le variabili vengano create una sola volta
    • Verifica che i conteggi delle variabili, i nomi e le forme corrispondano
    • Reimposta tutte le variabili, verifica l'equivalenza numerica con tutta la casualità disabilitata
    • Allinea la generazione di numeri casuali, controlla l'equivalenza numerica nell'inferenza
    • (Facoltativo) Controllare che i checkpoint siano caricati correttamente e i modelli TF1.x/TF2 generano un output identico

    B. Su un singolo dispositivo GPU/TPU

    C. Con strategie multi-dispositivo

  3. Convalida dell'equivalenza numerica dell'addestramento del modello per alcuni passaggi (esempi di codice disponibili di seguito)

    un. Convalida di un singolo passaggio di addestramento utilizzando dati piccoli e fissi su un singolo dispositivo CPU. In particolare, verificare l'equivalenza numerica per i seguenti componenti

    • calcolo delle perdite
    • metrica
    • tasso di apprendimento
    • calcolo e aggiornamento del gradiente

    B. Controlla le statistiche dopo l'allenamento 3 o più passaggi per verificare i comportamenti dell'ottimizzatore come lo slancio, sempre con dati fissi su un singolo dispositivo CPU

    C. Su un singolo dispositivo GPU/TPU

    D. Con strategie multi-dispositivo (controlla l'introduzione per MultiProcessRunner in basso)

  4. Test di copertura end-to-end su set di dati reali

    un. Controlla i comportamenti di allenamento con TensorBoard

    • utilizzare prima ottimizzatori semplici, ad esempio SGD e semplici strategie di distribuzione, ad esempio tf.distribute.OneDeviceStrategy
    • metriche di formazione
    • metriche di valutazione
    • capire qual è la ragionevole tolleranza per la casualità intrinseca

    B. Verifica l'equivalenza con ottimizzatore avanzato/programmatore del tasso di apprendimento/strategie di distribuzione

    C. Verificare l'equivalenza quando si utilizza la precisione mista

  5. Ulteriori benchmark di prodotto

Impostare

pip uninstall -y -q tensorflow

# Install tf-nightly as the DeterministicRandomTestTool is only available in
# Tensorflow 2.8
pip install -q tf-nightly

Convalida del singolo passaggio in avanti

La convalida del singolo passaggio in avanti, incluso il caricamento al checkpoint, è coperta in una colab diversa.

import sys
import unittest
import numpy as np

import tensorflow as tf
import tensorflow.compat.v1 as v1

Convalida dell'equivalenza numerica dell'addestramento del modello per alcuni passaggi

Imposta la configurazione del modello e prepara un set di dati falso.

params = {
    'input_size': 3,
    'num_classes': 3,
    'layer_1_size': 2,
    'layer_2_size': 2,
    'num_train_steps': 100,
    'init_lr': 1e-3,
    'end_lr': 0.0,
    'decay_steps': 1000,
    'lr_power': 1.0,
}

# make a small fixed dataset
fake_x = np.ones((2, params['input_size']), dtype=np.float32)
fake_y = np.zeros((2, params['num_classes']), dtype=np.int32)
fake_y[0][0] = 1
fake_y[1][1] = 1

step_num = 3

Definire il modello TF1.x.

# Assume there is an existing TF1.x model using estimator API
# Wrap the model_fn to log necessary tensors for result comparison
class SimpleModelWrapper():
  def __init__(self):
    self.logged_ops = {}
    self.logs = {
        'step': [],
        'lr': [],
        'loss': [],
        'grads_and_vars': [],
        'layer_out': []}

  def model_fn(self, features, labels, mode, params):
      out_1 = tf.compat.v1.layers.dense(features, units=params['layer_1_size'])
      out_2 = tf.compat.v1.layers.dense(out_1, units=params['layer_2_size'])
      logits = tf.compat.v1.layers.dense(out_2, units=params['num_classes'])
      loss = tf.compat.v1.losses.softmax_cross_entropy(labels, logits)

      # skip EstimatorSpec details for prediction and evaluation 
      if mode == tf.estimator.ModeKeys.PREDICT:
          pass
      if mode == tf.estimator.ModeKeys.EVAL:
          pass
      assert mode == tf.estimator.ModeKeys.TRAIN

      global_step = tf.compat.v1.train.get_or_create_global_step()
      lr = tf.compat.v1.train.polynomial_decay(
        learning_rate=params['init_lr'],
        global_step=global_step,
        decay_steps=params['decay_steps'],
        end_learning_rate=params['end_lr'],
        power=params['lr_power'])

      optmizer = tf.compat.v1.train.GradientDescentOptimizer(lr)
      grads_and_vars = optmizer.compute_gradients(
          loss=loss,
          var_list=graph.get_collection(
              tf.compat.v1.GraphKeys.TRAINABLE_VARIABLES))
      train_op = optmizer.apply_gradients(
          grads_and_vars,
          global_step=global_step)

      # log tensors
      self.logged_ops['step'] = global_step
      self.logged_ops['lr'] = lr
      self.logged_ops['loss'] = loss
      self.logged_ops['grads_and_vars'] = grads_and_vars
      self.logged_ops['layer_out'] = {
          'layer_1': out_1,
          'layer_2': out_2,
          'logits': logits}

      return tf.estimator.EstimatorSpec(mode, loss=loss, train_op=train_op)

  def update_logs(self, logs):
    for key in logs.keys():
      model_tf1.logs[key].append(logs[key])

La seguente classe v1.keras.utils.DeterministicRandomTestTool fornisce un gestore di contesto scope() che può fare in modo che le operazioni casuali con stato utilizzino lo stesso seme su entrambi i grafici/sessioni TF1 e l'esecuzione desiderosa,

Lo strumento fornisce due modalità di test:

  1. constant che utilizza lo stesso seme per ogni singola operazione non importa quante volte sia stata chiamata e,
  2. num_random_ops che utilizza il numero di operazioni casuali stateful osservate in precedenza come seme dell'operazione.

Questo vale sia per le operazioni casuali con stato utilizzate per la creazione e l'inizializzazione delle variabili, sia per le operazioni casuali con stato utilizzate nel calcolo (come per i livelli di eliminazione).

random_tool = v1.keras.utils.DeterministicRandomTestTool(mode='num_random_ops')

WARNING:tensorflow:From /tmp/ipykernel_26769/2689227634.py:1: The name tf.keras.utils.DeterministicRandomTestTool is deprecated. Please use tf.compat.v1.keras.utils.DeterministicRandomTestTool instead.

Eseguire il modello TF1.x in modalità grafico. Raccogli le statistiche per i primi 3 passaggi di allenamento per il confronto dell'equivalenza numerica.

with random_tool.scope():
  graph = tf.Graph()
  with graph.as_default(), tf.compat.v1.Session(graph=graph) as sess:
    model_tf1 = SimpleModelWrapper()
    # build the model
    inputs = tf.compat.v1.placeholder(tf.float32, shape=(None, params['input_size']))
    labels = tf.compat.v1.placeholder(tf.float32, shape=(None, params['num_classes']))
    spec = model_tf1.model_fn(inputs, labels, tf.estimator.ModeKeys.TRAIN, params)
    train_op = spec.train_op

    sess.run(tf.compat.v1.global_variables_initializer())
    for step in range(step_num):
      # log everything and update the model for one step
      logs, _ = sess.run(
          [model_tf1.logged_ops, train_op],
          feed_dict={inputs: fake_x, labels: fake_y})
      model_tf1.update_logs(logs)

/tmpfs/src/tf_docs_env/lib/python3.7/site-packages/ipykernel_launcher.py:14: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead.
  
/tmpfs/src/tf_docs_env/lib/python3.7/site-packages/keras/legacy_tf_layers/core.py:261: UserWarning: `layer.apply` is deprecated and will be removed in a future version. Please use `layer.__call__` method instead.
  return layer.apply(inputs)
/tmpfs/src/tf_docs_env/lib/python3.7/site-packages/ipykernel_launcher.py:15: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead.
  from ipykernel import kernelapp as app
/tmpfs/src/tf_docs_env/lib/python3.7/site-packages/ipykernel_launcher.py:16: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead.
  app.launch_new_instance()

Definire il modello TF2.

class SimpleModel(tf.keras.Model):
  def __init__(self, params, *args, **kwargs):
    super(SimpleModel, self).__init__(*args, **kwargs)
    # define the model
    self.dense_1 = tf.keras.layers.Dense(params['layer_1_size'])
    self.dense_2 = tf.keras.layers.Dense(params['layer_2_size'])
    self.out = tf.keras.layers.Dense(params['num_classes'])
    learning_rate_fn = tf.keras.optimizers.schedules.PolynomialDecay(
      initial_learning_rate=params['init_lr'],
      decay_steps=params['decay_steps'],
      end_learning_rate=params['end_lr'],
      power=params['lr_power'])  
    self.optimizer = tf.keras.optimizers.SGD(learning_rate_fn)
    self.compiled_loss = tf.keras.losses.CategoricalCrossentropy(from_logits=True)
    self.logs = {
        'lr': [],
        'loss': [],
        'grads': [],
        'weights': [],
        'layer_out': []}

  def call(self, inputs):
    out_1 = self.dense_1(inputs)
    out_2 = self.dense_2(out_1)
    logits = self.out(out_2)
    # log output features for every layer for comparison
    layer_wise_out = {
        'layer_1': out_1,
        'layer_2': out_2,
        'logits': logits}
    self.logs['layer_out'].append(layer_wise_out)
    return logits

  def train_step(self, data):
    x, y = data
    with tf.GradientTape() as tape:
      logits = self(x)
      loss = self.compiled_loss(y, logits)
    grads = tape.gradient(loss, self.trainable_weights)
    # log training statistics
    step = self.optimizer.iterations.numpy()
    self.logs['lr'].append(self.optimizer.learning_rate(step).numpy())
    self.logs['loss'].append(loss.numpy())
    self.logs['grads'].append(grads)
    self.logs['weights'].append(self.trainable_weights)
    # update model
    self.optimizer.apply_gradients(zip(grads, self.trainable_weights))
    return

Esegui il modello TF2 in modalità desiderosa. Raccogli le statistiche per i primi 3 passaggi di allenamento per il confronto dell'equivalenza numerica.

random_tool = v1.keras.utils.DeterministicRandomTestTool(mode='num_random_ops')
with random_tool.scope():
  model_tf2 = SimpleModel(params)
  for step in range(step_num):
    model_tf2.train_step([fake_x, fake_y])

Confronta l'equivalenza numerica per i primi passaggi di formazione.

Puoi anche controllare il taccuino Validazione correttezza ed equivalenza numerica per ulteriori consigli sull'equivalenza numerica.

np.testing.assert_allclose(model_tf1.logs['lr'], model_tf2.logs['lr'])
np.testing.assert_allclose(model_tf1.logs['loss'], model_tf2.logs['loss'])
for step in range(step_num):
  for name in model_tf1.logs['layer_out'][step]:
    np.testing.assert_allclose(
        model_tf1.logs['layer_out'][step][name],
        model_tf2.logs['layer_out'][step][name])

Test unitari

Esistono alcuni tipi di unit test che possono aiutare a eseguire il debug del codice di migrazione.

  1. Convalida del singolo passaggio in avanti
  2. Convalida dell'equivalenza numerica dell'addestramento del modello per alcuni passaggi
  3. Prestazioni di inferenza di benchmark
  4. Il modello addestrato effettua previsioni corrette su punti dati fissi e semplici

Puoi utilizzare @parameterized.parameters per testare modelli con diverse configurazioni. Dettagli con codice di esempio .

Tieni presente che è possibile eseguire le API di sessione e l'esecuzione desiderosa nello stesso test case. I frammenti di codice di seguito mostrano come.

import unittest

class TestNumericalEquivalence(unittest.TestCase):

  # copied from code samples above
  def setup(self):
    # record statistics for 100 training steps
    step_num = 100

    # setup TF 1 model
    random_tool = v1.keras.utils.DeterministicRandomTestTool(mode='num_random_ops')
    with random_tool.scope():
      # run TF1.x code in graph mode with context management
      graph = tf.Graph()
      with graph.as_default(), tf.compat.v1.Session(graph=graph) as sess:
        self.model_tf1 = SimpleModelWrapper()
        # build the model
        inputs = tf.compat.v1.placeholder(tf.float32, shape=(None, params['input_size']))
        labels = tf.compat.v1.placeholder(tf.float32, shape=(None, params['num_classes']))
        spec = self.model_tf1.model_fn(inputs, labels, tf.estimator.ModeKeys.TRAIN, params)
        train_op = spec.train_op

        sess.run(tf.compat.v1.global_variables_initializer())
        for step in range(step_num):
          # log everything and update the model for one step
          logs, _ = sess.run(
              [self.model_tf1.logged_ops, train_op],
              feed_dict={inputs: fake_x, labels: fake_y})
          self.model_tf1.update_logs(logs)

    # setup TF2 model
    random_tool = v1.keras.utils.DeterministicRandomTestTool(mode='num_random_ops')
    with random_tool.scope():
      self.model_tf2 = SimpleModel(params)
      for step in range(step_num):
        self.model_tf2.train_step([fake_x, fake_y])

  def test_learning_rate(self):
    np.testing.assert_allclose(
        self.model_tf1.logs['lr'],
        self.model_tf2.logs['lr'])

  def test_training_loss(self):
    # adopt different tolerance strategies before and after 10 steps
    first_n_step = 10

    # abosolute difference is limited below 1e-5
    # set `equal_nan` to be False to detect potential NaN loss issues
    abosolute_tolerance = 1e-5
    np.testing.assert_allclose(
        actual=self.model_tf1.logs['loss'][:first_n_step],
        desired=self.model_tf2.logs['loss'][:first_n_step],
        atol=abosolute_tolerance,
        equal_nan=False)

    # relative difference is limited below 5%
    relative_tolerance = 0.05
    np.testing.assert_allclose(self.model_tf1.logs['loss'][first_n_step:],
                               self.model_tf2.logs['loss'][first_n_step:],
                               rtol=relative_tolerance,
                               equal_nan=False)

Strumenti di debug

tf.print

tf.print vs print/logging.info

  • Con argomenti configurabili, tf.print può visualizzare ricorsivamente i primi e gli ultimi elementi di ciascuna dimensione per i tensori stampati. Controlla i documenti API per i dettagli.
  • Per l'esecuzione desiderosa, sia print che tf.print stampano il valore del tensore. Ma la print può comportare una copia da dispositivo a host, che può potenzialmente rallentare il tuo codice.
  • Per la modalità grafico che include l'utilizzo all'interno di tf.function , è necessario utilizzare tf.print per stampare il valore del tensore effettivo. tf.print viene compilato in un'operazione nel grafico, mentre print e logging.info registrano solo al momento della traccia, che spesso non è quello che desideri.
  • tf.print supporta anche la stampa di tensori compositi come tf.RaggedTensor e tf.sparse.SparseTensor .
  • Puoi anche utilizzare un callback per monitorare metriche e variabili. Controlla come utilizzare i callback personalizzati con logs dict e attributo self.model .

tf.print vs print all'interno di tf.function

# `print` prints info of tensor object
# `tf.print` prints the tensor value
@tf.function
def dummy_func(num):
  num += 1
  print(num)
  tf.print(num)
  return num

_ = dummy_func(tf.constant([1.0]))

# Output:
# Tensor("add:0", shape=(1,), dtype=float32)
# [2]

Tensor("add:0", shape=(1,), dtype=float32)
[2]

tf.distribute.Strategy

  • Se la tf.function contenente tf.print viene eseguita sui worker, ad esempio quando si utilizza TPUStrategy o ParameterServerStrategy , è necessario controllare i registri del worker/parameter server per trovare i valori stampati.
  • Per print o logging.info , i registri verranno stampati sul coordinatore quando si utilizza ParameterServerStrategy e i registri verranno stampati su STDOUT su worker0 quando si utilizzano le TPU.

tf.keras.Model

  • Quando si utilizzano modelli API sequenziali e funzionali, se si desidera stampare valori, ad esempio input del modello o funzioni intermedie dopo alcuni livelli, sono disponibili le seguenti opzioni.
    1. Scrivi un livello personalizzato che tf.print gli input.
    2. Includere gli output intermedi che si desidera ispezionare negli output del modello.
  • I livelli tf.keras.layers.Lambda hanno limitazioni di (de)serializzazione. Per evitare problemi di caricamento del checkpoint, scrivi invece un livello di sottoclasse personalizzato. Controlla i documenti API per maggiori dettagli.
  • Non è possibile tf.print output intermedi in un tf.keras.callbacks.LambdaCallback se non si ha accesso ai valori effettivi, ma solo agli oggetti tensore simbolici di Keras.

Opzione 1: scrivi un livello personalizzato

class PrintLayer(tf.keras.layers.Layer):
  def call(self, inputs):
    tf.print(inputs)
    return inputs

def get_model():
  inputs = tf.keras.layers.Input(shape=(1,))
  out_1 = tf.keras.layers.Dense(4)(inputs)
  out_2 = tf.keras.layers.Dense(1)(out_1)
  # use custom layer to tf.print intermediate features
  out_3 = PrintLayer()(out_2)
  model = tf.keras.Model(inputs=inputs, outputs=out_3)
  return model

model = get_model()
model.compile(optimizer="adam", loss="mse")
model.fit([1, 2, 3], [0.0, 0.0, 1.0])

[[-0.327884018]
 [-0.109294668]
 [-0.218589336]]
1/1 [==============================] - 0s 280ms/step - loss: 0.6077
<keras.callbacks.History at 0x7f63d46bf190>

Opzione 2: includi gli output intermedi che desideri ispezionare negli output del modello.

Tieni presente che in tal caso potrebbero essere necessarie alcune personalizzazioni per utilizzare Model.fit .

def get_model():
  inputs = tf.keras.layers.Input(shape=(1,))
  out_1 = tf.keras.layers.Dense(4)(inputs)
  out_2 = tf.keras.layers.Dense(1)(out_1)
  # include intermediate values in model outputs
  model = tf.keras.Model(
      inputs=inputs,
      outputs={
          'inputs': inputs,
          'out_1': out_1,
          'out_2': out_2})
  return model

pdb

È possibile utilizzare pdb sia nel terminale che in Colab per ispezionare i valori intermedi per il debug.

Visualizza il grafico con TensorBoard

È possibile esaminare il grafico TensorFlow con TensorBoard . TensorBoard è supportato anche su colab . TensorBoard è un ottimo strumento per visualizzare i riepiloghi. Puoi usarlo per confrontare il tasso di apprendimento, i pesi del modello, la scala del gradiente, le metriche di addestramento/convalida o anche modellare gli output intermedi tra il modello TF1.x e il modello TF2 migrato attraverso il processo di addestramento e vedere se i valori sembrano come previsto.

TensorFlow Profiler

TensorFlow Profiler può aiutarti a visualizzare la sequenza temporale di esecuzione su GPU/TPU. Puoi dare un'occhiata a questa demo Colab per il suo utilizzo di base.

Runner multiprocesso

MultiProcessRunner è uno strumento utile durante il debug con MultiWorkerMirroredStrategy e ParameterServerStrategy. Puoi dare un'occhiata a questo esempio concreto per il suo utilizzo.

In particolare per i casi di queste due strategie, si consiglia di 1) non solo disporre di unit test per coprire il loro flusso, 2) ma anche di tentare di riprodurre gli errori utilizzandolo in unit test per evitare di avviare un vero lavoro distribuito ogni volta che tentano una correzione.