TF2 마이그레이션된 교육 파이프라인 디버그

TensorFlow.org에서 보기 Google Colab에서 실행 GitHub에서 소스 보기 노트북 다운로드

이 노트북은 TF2로 마이그레이션할 때 훈련 파이프라인을 디버그하는 방법을 보여줍니다. 다음 구성 요소로 구성됩니다.

  1. 교육 파이프라인 디버깅을 위한 제안 단계 및 코드 샘플
  2. 디버깅 도구
  3. 기타 관련 리소스

한 가지 가정은 비교를 위해 TF1.x 코드와 훈련된 모델이 있고 유사한 유효성 검사 정확도를 달성하는 TF2 모델을 구축하려는 것입니다.

이 노트북은 훈련/추론 속도 또는 메모리 사용에 대한 디버깅 성능 문제를 다루지 않습니다 .

디버깅 워크플로

다음은 TF2 훈련 파이프라인을 디버깅하기 위한 일반적인 워크플로입니다. 이 단계를 순서대로 수행할 필요는 없습니다. 중간 단계에서 모델을 테스트하고 디버깅 범위를 좁히는 이진 검색 접근 방식을 사용할 수도 있습니다.

  1. 컴파일 및 런타임 오류 수정

  2. 단일 정방향 패스 검증(별도 가이드 )

    ㅏ. 단일 CPU 장치에서

    • 변수가 한 번만 생성되었는지 확인
    • 변수 개수, 이름 및 모양이 일치하는지 확인
    • 모든 변수 재설정, 모든 임의성이 비활성화된 상태에서 수치적 동등성 확인
    • 난수 생성 정렬, 추론에서 수치적 동등성 확인
    • (선택 사항) 체크포인트가 제대로 로드되고 TF1.x/TF2 모델이 동일한 출력을 생성하는지 확인합니다.

    비. 단일 GPU/TPU 장치에서

    씨. 다중 장치 전략으로

  3. 몇 단계에 대한 모델 훈련 수치 동등성 검증(아래에서 코드 샘플 사용 가능)

    ㅏ. 단일 CPU 장치에서 작고 고정된 데이터를 사용하는 단일 훈련 단계 검증. 특히 다음 구성 요소에 대한 수치 동등성을 확인하십시오.

    • 손실 계산
    • 측정항목
    • 학습률
    • 기울기 계산 및 업데이트

    비. 단일 CPU 장치에 고정 데이터가 있는 모멘텀과 같은 옵티마이저 동작을 확인하기 위해 3단계 이상 훈련 후 통계 확인

    씨. 단일 GPU/TPU 장치에서

    디. 다중 장치 전략 사용(하단에서 MultiProcessRunner 에 대한 소개 확인)

  4. 실제 데이터 세트에 대한 종단 간 커버리지 테스트

    ㅏ. TensorBoard로 학습 동작 확인

    • SGD 및 간단한 배포 전략(예: tf.distribute.OneDeviceStrategy first)과 같은 간단한 최적화 프로그램을 사용합니다.
    • 훈련 지표
    • 평가 지표
    • 고유한 무작위성에 대한 합리적인 허용 오차가 무엇인지 파악합니다.

    비. 고급 옵티마이저/학습률 스케줄러/분산 전략으로 동등성 확인

    씨. 혼합 정밀도 사용 시 동등성 확인

  5. 추가 제품 벤치마크

설정

pip uninstall -y -q tensorflow

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

단일 정방향 패스 검증

체크포인트 로딩을 포함한 단일 정방향 패스 검증은 다른 colab 에서 다룹니다.

import sys
import unittest
import numpy as np

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

몇 단계에 대한 모델 학습 수치 동등성 검증

모델 구성을 설정하고 가짜 데이터 세트를 준비합니다.

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

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

다음 v1.keras.utils.DeterministicRandomTestTool 클래스는 상태 저장 임의 작업이 TF1 그래프/세션과 즉시 실행 모두에서 동일한 시드를 사용하도록 할 수 있는 컨텍스트 관리자 scope() 를 제공합니다.

이 도구는 두 가지 테스트 모드를 제공합니다.

  1. 호출된 횟수에 상관없이 모든 단일 작업에 대해 동일한 시드를 사용하는 constant 및,
  2. 이전에 관찰된 상태 저장 임의 작업의 수를 작업 시드로 사용하는 num_random_ops .

이것은 변수를 생성하고 초기화하는 데 사용되는 상태 저장 임의 작업과 계산에 사용되는 상태 저장 임의 작업(예: 드롭아웃 레이어)에 모두 적용됩니다.

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.

그래프 모드에서 TF1.x 모델을 실행합니다. 수치적 동등성 비교를 위해 처음 3개의 훈련 단계에 대한 통계를 수집합니다.

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

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

Eager 모드에서 TF2 모델을 실행합니다. 수치적 동등성 비교를 위해 처음 3개의 훈련 단계에 대한 통계를 수집합니다.

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

처음 몇 개의 훈련 단계에 대한 수치적 동등성을 비교합니다.

수치적 동등성에 대한 추가 조언을 위해 Validating correctness & 수치적 동등성 노트북 을 확인할 수도 있습니다.

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

단위 테스트

마이그레이션 코드를 디버그하는 데 도움이 되는 몇 가지 유형의 단위 테스트가 있습니다.

  1. 단일 정방향 패스 검증
  2. 몇 단계에 대한 모델 학습 수치 동등성 검증
  3. 벤치마크 추론 성능
  4. 훈련된 모델은 고정된 단순 데이터 포인트에 대해 정확한 예측을 수행합니다.

@parameterized.parameters 를 사용하여 다양한 구성으로 모델을 테스트할 수 있습니다. 코드 샘플이 포함된 세부정보 .

동일한 테스트 사례에서 세션 API 및 즉시 실행을 실행할 수 있습니다. 아래의 코드 조각은 방법을 보여줍니다.

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)

디버깅 도구

tf.print

tf.print 대 인쇄/logging.info

  • 구성 가능한 인수를 사용하여 tf.print 는 인쇄된 텐서에 대한 각 차원의 처음 및 마지막 몇 가지 요소를 재귀적으로 표시할 수 있습니다. 자세한 내용은 API 문서 를 확인하세요.
  • 즉시 실행을 위해 print 와 tf.print 는 모두 텐서의 값을 출력합니다. 그러나 print 에는 잠재적으로 코드 속도를 저하시킬 수 있는 장치-호스트 복사가 포함될 수 있습니다.
  • tf.function 내부 사용을 포함하는 그래프 모드의 경우 실제 텐서 값을 인쇄하려면 tf.print 를 사용해야 합니다. tf.print 는 그래프의 op로 컴파일되는 반면, print 및 logging.info 는 추적 시간에만 기록하며, 이는 종종 원하는 것이 아닙니다.
  • tf.print 는 tf.RaggedTensor 및 tf.sparse.SparseTensor 와 같은 복합 텐서 인쇄도 지원합니다.
  • 콜백을 사용하여 메트릭 및 변수를 모니터링할 수도 있습니다. logs dict 및 self.model 속성 과 함께 사용자 정의 콜백을 사용하는 방법을 확인하십시오.

tf.print 대 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

  • tf.function 가 포함된 tf.print 이 작업자에서 실행되는 경우(예: TPUStrategy 또는 ParameterServerStrategy 사용 시) 작업자/매개변수 서버 로그를 확인하여 인쇄된 값을 찾아야 합니다.
  • print 또는 logging.info 의 경우 ParameterServerStrategy 를 사용할 때 로그가 코디네이터에 인쇄되고 로그가 TPU를 사용할 때 worker0의 STDOUT에 인쇄됩니다.

tf.keras.Model

  • Sequential 및 Functional API 모델을 사용할 때 일부 레이어 뒤에 모델 입력 또는 중간 기능과 같은 값을 인쇄하려는 경우 다음 옵션이 있습니다.
    1. 입력을 tf.print 하는 사용자 정의 레이어 를 작성하십시오.
    2. 모델 출력에 검사하려는 중간 출력을 포함합니다.
  • tf.keras.layers.Lambda 레이어에는 (역)직렬화 제한이 있습니다. 체크포인트 로딩 문제를 피하려면 대신 사용자 정의 서브클래싱된 레이어를 작성하십시오. 자세한 내용은 API 문서 를 확인하세요.
  • 실제 값에 액세스할 수 없고 대신 기호 Keras 텐서 객체에만 액세스할 수 있는 경우 tf.print 에서 중간 출력을 tf.keras.callbacks.LambdaCallback 할 수 없습니다.

옵션 1: 사용자 지정 레이어 작성

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>

옵션 2: 검사하려는 중간 출력을 모델 출력에 포함합니다.

이러한 경우 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

터미널과 Colab 모두에서 pdb 를 사용하여 디버깅을 위해 중간 값을 검사할 수 있습니다.

TensorBoard로 그래프 시각화

TensorBoard를 사용하여 TensorFlow 그래프를 검사할 수 있습니다. TensorBoard는 colab에서도 지원됩니다 . TensorBoard는 요약을 시각화하는 훌륭한 도구입니다. 이를 사용하여 학습 속도, 모델 가중치, 그래디언트 스케일, 학습/검증 지표를 비교하거나 학습 프로세스를 통해 TF1.x 모델과 마이그레이션된 TF2 모델 간의 중간 출력을 모델링하고 값이 예상대로 보이는지 확인할 수 있습니다.

텐서플로우 프로파일러

TensorFlow Profiler 를 사용하면 GPU/TPU에서 실행 타임라인을 시각화할 수 있습니다. 이 Colab 데모 에서 기본적인 사용법을 확인할 수 있습니다.

MultiProcessRunner

MultiProcessRunner 는 MultiWorkerMirroredStrategy 및 ParameterServerStrategy로 디버깅할 때 유용한 도구입니다. 사용법에 대한 이 구체적인 예 를 볼 수 있습니다.

특히 이 두 가지 전략의 경우 1) 흐름을 커버하는 단위 테스트를 가질 뿐만 아니라 2) 시도할 때마다 실제 분산 작업을 시작하지 않도록 단위 테스트에서 이를 사용하여 실패를 재현하는 것이 좋습니다. 수정.