 在 TensorFlow.org 上查看
|
 在 Google Colab 中运行
|
 在 GitHub 上查看源代码
|
 下载笔记本
|
此笔记本演示了如何在迁移到 TensorFlow 2 (TF2) 时调试训练管道。它包含以下组件
- 调试训练管道的建议步骤和代码示例
- 调试工具
- 其他相关资源
一个假设是您拥有 TensorFlow 1 (TF1.x) 代码和训练后的模型以供比较，并且您希望构建一个能够达到类似验证精度的 TF2 模型。
此笔记本 **不** 涵盖调试训练/推理速度或内存使用量的性能问题。
调试工作流程
以下是调试 TF2 训练管道的通用工作流程。请注意，您不需要按顺序执行这些步骤。您也可以使用二分查找方法，在中间步骤中测试模型并缩小调试范围。
修复编译和运行时错误
单次前向传递验证（在单独的 指南 中）
a. 在单个 CPU 设备上
- 验证变量仅创建一次
- 检查变量计数、名称和形状是否匹配
- 重置所有变量，检查所有随机性被禁用时的数值等效性
- 对齐随机数生成，检查推理中的数值等效性
- （可选）检查检查点是否已正确加载，以及 TF1.x/TF2 模型是否生成相同的输出
b. 在单个 GPU/TPU 设备上
c. 使用多设备策略
模型训练数值等效性验证，进行几个步骤（代码示例如下）
a. 使用单个 CPU 设备上的少量固定数据进行单个训练步骤验证。具体来说，检查以下组件的数值等效性
- 损失计算
- 指标
- 学习率
- 梯度计算和更新
b. 检查训练 3 步或更多步后的统计信息，以验证优化器行为（例如动量），仍然使用单个 CPU 设备上的固定数据
c. 在单个 GPU/TPU 设备上
d. 使用多设备策略（查看底部 MultiProcessRunner 的介绍）
在真实数据集上进行端到端收敛测试
a. 使用 TensorBoard 检查训练行为
- 首先使用简单的优化器（例如 SGD）和简单的分布式策略（例如
tf.distribute.OneDeviceStrategy） - 训练指标
- 评估指标
- 找出固有随机性的合理容差
b. 检查使用高级优化器/学习率调度器/分布式策略时的等效性
c. 检查使用混合精度时的等效性
- 首先使用简单的优化器（例如 SGD）和简单的分布式策略（例如
其他产品基准测试
设置
# The `DeterministicRandomTestTool` is only available from Tensorflow 2.8:pip install -q "tensorflow==2.9.*"
单次前向传递验证
单次前向传递验证（包括检查点加载）在不同的 colab 中介绍。
import sys
import unittest
import numpy as np
import tensorflow as tf
import tensorflow.compat.v1 as v1
2024-01-17 02:21:07.536045: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory
模型训练数值等效性验证，进行几个步骤
设置模型配置并准备一个假数据集。
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 类提供了一个上下文管理器 scope()，它可以使有状态随机操作在 TF1 图/会话和急切执行中使用相同的种子，
该工具提供两种测试模式
constant，它对每次操作使用相同的种子，无论它被调用多少次，以及num_random_ops，它使用先前观察到的有状态随机操作的数量作为操作种子。
这适用于用于创建和初始化变量的有状态随机操作，以及计算中使用的有状态随机操作（例如 dropout 层）。
random_tool = v1.keras.utils.DeterministicRandomTestTool(mode='num_random_ops')
WARNING:tensorflow:From /tmpfs/tmp/ipykernel_9596/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)
2024-01-17 02:21:10.121960: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcudart.so.11.0'; dlerror: libcudart.so.11.0: cannot open shared object file: No such file or directory 2024-01-17 02:21:10.122074: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcublas.so.11'; dlerror: libcublas.so.11: cannot open shared object file: No such file or directory 2024-01-17 02:21:10.122150: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcublasLt.so.11'; dlerror: libcublasLt.so.11: cannot open shared object file: No such file or directory 2024-01-17 02:21:10.122222: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcufft.so.10'; dlerror: libcufft.so.10: cannot open shared object file: No such file or directory 2024-01-17 02:21:10.189341: W tensorflow/stream_executor/platform/default/dso_loader.cc:64] Could not load dynamic library 'libcusparse.so.11'; dlerror: libcusparse.so.11: cannot open shared object file: No such file or directory 2024-01-17 02:21:10.189554: W tensorflow/core/common_runtime/gpu/gpu_device.cc:1850] Cannot dlopen some GPU libraries. Please make sure the missing libraries mentioned above are installed properly if you would like to use GPU. Follow the guide at https://tensorflowcn.cn/install/gpu for how to download and setup the required libraries for your platform. Skipping registering GPU devices... /tmpfs/tmp/ipykernel_9596/1984550333.py:14: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead. out_1 = tf.compat.v1.layers.dense(features, units=params['layer_1_size']) /tmpfs/tmp/ipykernel_9596/1984550333.py:15: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead. out_2 = tf.compat.v1.layers.dense(out_1, units=params['layer_2_size']) /tmpfs/tmp/ipykernel_9596/1984550333.py:16: UserWarning: `tf.layers.dense` is deprecated and will be removed in a future version. Please use `tf.keras.layers.Dense` instead. logits = tf.compat.v1.layers.dense(out_2, units=params['num_classes'])
定义 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.legacy.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
在急切模式下运行 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])
比较前几个训练步骤的数值等效性。
您还可以查看 验证正确性和数值等效性笔记本，以获取有关数值等效性的更多建议。
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])
单元测试
有几种类型的单元测试可以帮助调试迁移代码。
- 单次前向传递验证
- 模型训练数值等效性验证，进行几个步骤
- 基准测试推理性能
- 训练后的模型对固定和简单的數據点做出正确的预测
您可以使用 @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
# absolute 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 与 print/logging.info
- 使用可配置参数，
tf.print可以递归地显示每个维度中第一个和最后几个元素的打印张量。查看 API 文档 以获取详细信息。 - 对于急切执行，
print和tf.print都打印张量的值。但是print可能涉及设备到主机的复制，这可能会减慢代码速度。 - 对于图模式，包括在
tf.function中使用，您需要使用tf.print打印实际的张量值。tf.print在图中被编译成一个操作，而print和logging.info仅在跟踪时记录，这通常不是您想要的。 tf.print还支持打印复合张量，例如tf.RaggedTensor和tf.sparse.SparseTensor.- 您还可以使用回调来监控指标和变量。请查看如何使用自定义回调以及 logs dict 和 self.model 属性.
tf.print 与 tf.function 中的 print
# `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.print的tf.function在工作进程上执行，例如使用TPUStrategy或ParameterServerStrategy时，您需要检查工作进程/参数服务器日志以查找打印的值。 - 对于
print或logging.info，使用ParameterServerStrategy时，日志将打印在协调器上，使用 TPU 时，日志将打印在 worker0 的 STDOUT 上。
tf.keras.Model
- 使用 Sequential 和 Functional API 模型时，如果您想打印值，例如模型输入或某些层之后的中間特征，您可以选择以下方法。
tf.keras.layers.Lambda层具有 (反) 序列化限制。为了避免检查点加载问题，请改用自定义子类化层。查看 API 文档 以获取更多详细信息。- 如果您无法访问实际值，而只能访问符号 Keras 张量对象，则无法在
tf.keras.callbacks.LambdaCallback中tf.print中間输出。
选项 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.109294683] [-0.218589365]] 1/1 [==============================] - 0s 273ms/step - loss: 0.6077 <keras.callbacks.History at 0x7effa3fcad30>
选项 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
TensorFlow Profiler 可以帮助您可视化 GPU/TPU 上的执行时间线。您可以查看此 Colab 演示 以了解其基本用法。
MultiProcessRunner
MultiProcessRunner 是使用 MultiWorkerMirroredStrategy 和 ParameterServerStrategy 进行调试时的一个有用工具。您可以查看 此具体示例 以了解其用法。
特别是对于这两种策略，建议您 1) 不仅进行单元测试以涵盖其流程，2) 还要尝试使用它在单元测试中重现故障，以避免每次尝试修复时都启动真正的分布式作业。