Déboguer le pipeline de formation migré TF2

Voir sur TensorFlow.org Exécuter dans Google Colab Voir la source sur GitHub Télécharger le cahier

Ce notebook montre comment déboguer le pipeline de formation lors de la migration vers TF2. Il se compose des composants suivants :

  1. Étapes suggérées et exemples de code pour le débogage du pipeline de formation
  2. Outils de débogage
  3. Autres ressources connexes

Une hypothèse est que vous avez du code TF1.x et des modèles entraînés à comparer, et que vous souhaitez créer un modèle TF2 qui atteint une précision de validation similaire.

Ce bloc-notes ne couvre PAS les problèmes de performances de débogage pour la vitesse d'entraînement/d'inférence ou l'utilisation de la mémoire.

Flux de travail de débogage

Vous trouverez ci-dessous un flux de travail général pour le débogage de vos pipelines de formation TF2. Notez que vous n'avez pas besoin de suivre ces étapes dans l'ordre. Vous pouvez également utiliser une approche de recherche binaire dans laquelle vous testez le modèle dans une étape intermédiaire et réduisez la portée du débogage.

  1. Correction des erreurs de compilation et d'exécution

  2. Validation de passage unique (dans un guide séparé)

    une. Sur un seul appareil CPU

    • Vérifier que les variables ne sont créées qu'une seule fois
    • Vérifiez que le nombre, les noms et les formes des variables correspondent
    • Réinitialiser toutes les variables, vérifier l'équivalence numérique avec tout caractère aléatoire désactivé
    • Aligner la génération de nombres aléatoires, vérifier l'équivalence numérique dans l'inférence
    • (Facultatif) Vérifiez que les points de contrôle sont chargés correctement et que les modèles TF1.x/TF2 génèrent une sortie identique

    b. Sur un seul appareil GPU/TPU

    c. Avec des stratégies multi-appareils

  3. Validation de l'équivalence numérique d'entraînement du modèle pour quelques étapes (exemples de code disponibles ci-dessous)

    une. Validation d'une seule étape de formation à l'aide de petites données fixes sur un seul processeur. Plus précisément, vérifiez l'équivalence numérique pour les composants suivants

    • calcul des pertes
    • métrique
    • taux d'apprentissage
    • calcul et mise à jour du gradient

    b. Vérifiez les statistiques après avoir entraîné 3 étapes ou plus pour vérifier les comportements de l'optimiseur comme l'élan, toujours avec des données fixes sur un seul processeur

    c. Sur un seul appareil GPU/TPU

    ré. Avec des stratégies multi-appareils (consultez l'intro de MultiProcessRunner en bas)

  4. Test de couverture de bout en bout sur un jeu de données réel

    une. Vérifier les comportements d'entraînement avec TensorBoard

    • utilisez d'abord des optimiseurs simples, par exemple SGD, et des stratégies de distribution simples, par exemple tf.distribute.OneDeviceStrategy
    • métriques de formation
    • mesures d'évaluation
    • déterminer quelle est la tolérance raisonnable pour le caractère aléatoire inhérent

    b. Vérifier l'équivalence avec l'optimiseur avancé/programmateur de taux d'apprentissage/stratégies de distribution

    c. Vérifier l'équivalence lors de l'utilisation d'une précision mixte

  5. Benchmarks produits supplémentaires

Installer

pip uninstall -y -q tensorflow

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

Validation d'un seul passage vers l'avant

La validation de passage unique, y compris le chargement des points de contrôle, est couverte dans un colab différent.

import sys
import unittest
import numpy as np

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

Validation d'équivalence numérique d'entraînement de modèle pour quelques étapes

Configurez la configuration du modèle et préparez un faux ensemble de données.

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

Définir le modèle 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 classe v1.keras.utils.DeterministicRandomTestTool suivante fournit un gestionnaire de contexte scope() qui peut faire en sorte que les opérations aléatoires avec état utilisent la même graine sur les graphes/sessions TF1 et l'exécution impatiente,

L'outil propose deux modes de test :

  1. constant qui utilise la même graine pour chaque opération, quel que soit le nombre de fois qu'elle a été appelée et,
  2. num_random_ops qui utilise le nombre d'opérations aléatoires avec état précédemment observées comme graine d'opération.

Cela s'applique à la fois aux opérations aléatoires avec état utilisées pour créer et initialiser des variables, et aux opérations aléatoires avec état utilisées dans le calcul (comme pour les couches d'abandon).

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.

Exécutez le modèle TF1.x en mode graphique. Recueillir des statistiques pour les 3 premières étapes de formation pour la comparaison d'équivalence numérique.

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

Définir le modèle 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

Exécutez le modèle TF2 en mode impatient. Recueillir des statistiques pour les 3 premières étapes de formation pour la comparaison d'équivalence numérique.

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

Comparez l'équivalence numérique pour les premières étapes de formation.

Vous pouvez également consulter le cahier Validation de l'exactitude et de l'équivalence numérique pour obtenir des conseils supplémentaires sur l'équivalence numérique.

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

Tests unitaires

Il existe quelques types de tests unitaires qui peuvent aider à déboguer votre code de migration.

  1. Validation d'un seul passage vers l'avant
  2. Validation d'équivalence numérique d'entraînement de modèle pour quelques étapes
  3. Performances d'inférence de référence
  4. Le modèle formé fait des prédictions correctes sur des points de données fixes et simples

Vous pouvez utiliser @parameterized.parameters pour tester des modèles avec différentes configurations. Détails avec exemple de code .

Notez qu'il est possible d'exécuter des API de session et une exécution hâtive dans le même cas de test. Les extraits de code ci-dessous montrent comment.

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)

Outils de débogage

tf.print

tf.print vs print/logging.info

  • Avec des arguments configurables, tf.print peut afficher de manière récursive les premiers et derniers éléments de chaque dimension pour les tenseurs imprimés. Consultez la documentation de l' API pour plus de détails.
  • Pour une exécution rapide, print et tf.print impriment la valeur du tenseur. Mais l' print peut impliquer une copie de périphérique à hôte, ce qui peut potentiellement ralentir votre code.
  • Pour le mode graphique, y compris l'utilisation dans tf.function , vous devez utiliser tf.print pour imprimer la valeur réelle du tenseur. tf.print est compilé dans un op dans le graphique, alors que print et logging.info ne se connectent qu'au moment du traçage, ce qui n'est souvent pas ce que vous voulez.
  • tf.print prend également en charge l'impression de tenseurs composites tels que tf.RaggedTensor et tf.sparse.SparseTensor .
  • Vous pouvez également utiliser un rappel pour surveiller les métriques et les variables. Veuillez vérifier comment utiliser les rappels personnalisés avec les journaux dict et l'attribut self.model .

tf.print vs impression à l'intérieur de 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

  • Si la tf.function contenant tf.print est exécutée sur les nœuds de calcul, par exemple lors de l'utilisation TPUStrategy ou ParameterServerStrategy , vous devez vérifier les journaux du serveur de nœuds de calcul/de paramètres pour trouver les valeurs imprimées.
  • Pour print ou logging.info , les journaux seront imprimés sur le coordinateur lors de l'utilisation de ParameterServerStrategy , et les journaux seront imprimés sur STDOUT sur worker0 lors de l'utilisation de TPU.

tf.keras.Modèle

  • Lorsque vous utilisez des modèles d'API séquentiels et fonctionnels, si vous souhaitez imprimer des valeurs, par exemple, des entrées de modèle ou des fonctionnalités intermédiaires après certaines couches, vous disposez des options suivantes.
    1. Écrivez un calque personnalisé qui tf.print les entrées.
    2. Incluez les sorties intermédiaires que vous souhaitez inspecter dans les sorties du modèle.
  • Les couches tf.keras.layers.Lambda ont des limitations de (dé)sérialisation. Pour éviter les problèmes de chargement des points de contrôle, écrivez plutôt une couche de sous-classe personnalisée. Consultez la documentation de l' API pour plus de détails.
  • Vous ne pouvez pas tf.print sorties intermédiaires dans un tf.keras.callbacks.LambdaCallback si vous n'avez pas accès aux valeurs réelles, mais uniquement aux objets tenseurs Keras symboliques.

Option 1 : écrire un calque personnalisé

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>

Option 2 : incluez les sorties intermédiaires que vous souhaitez inspecter dans les sorties du modèle.

Notez que dans ce cas, vous aurez peut-être besoin de certaines personnalisations pour utiliser 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

l'APB

Vous pouvez utiliser pdb à la fois dans le terminal et Colab pour inspecter les valeurs intermédiaires pour le débogage.

Visualiser le graphique avec TensorBoard

Vous pouvez examiner le graphique TensorFlow avec TensorBoard . TensorBoard est également pris en charge sur colab . TensorBoard est un excellent outil pour visualiser des résumés. Vous pouvez l'utiliser pour comparer le taux d'apprentissage, les poids du modèle, l'échelle de gradient, les métriques d'entraînement/validation, ou même les sorties intermédiaires du modèle entre le modèle TF1.x et le modèle TF2 migré tout au long du processus d'entraînement et voir si les valeurs ressemblent aux attentes.

Profileur TensorFlow

TensorFlow Profiler peut vous aider à visualiser la chronologie d'exécution sur les GPU/TPU. Vous pouvez consulter cette démo Colab pour son utilisation de base.

MultiProcessRunner

MultiProcessRunner est un outil utile lors du débogage avec MultiWorkerMirroredStrategy et ParameterServerStrategy. Vous pouvez jeter un oeil à cet exemple concret pour son utilisation.

Spécifiquement pour les cas de ces deux stratégies, il est recommandé 1) non seulement d'avoir des tests unitaires pour couvrir leur flux, 2) mais aussi d'essayer de reproduire les échecs en l'utilisant dans les tests unitaires pour éviter de lancer de vrais travaux distribués à chaque fois qu'ils tentent une réparation.