TensorFlow 源码剖析 - 最佳实践与案例

本文档提供TensorFlow的实战经验、最佳实践和具体案例，帮助开发者高效使用TensorFlow并避免常见陷阱。

一、框架使用示例

1.1 Eager模式：快速原型开发

Eager模式提供命令式执行，便于调试和快速迭代。

基本使用

import tensorflow as tf

# 启用Eager模式（TF 2.x默认启用）
tf.config.run_functions_eagerly(True)

# 创建Tensor
x = tf.constant([[1.0, 2.0], [3.0, 4.0]])
y = tf.constant([[5.0, 6.0], [7.0, 8.0]])

# 立即执行操作
z = tf.matmul(x, y)
print(z.numpy())  # [[19. 22.]
                  #  [43. 50.]]

# 自动微分
with tf.GradientTape() as tape:
    tape.watch(x)
    z = tf.reduce_sum(x ** 2)

gradient = tape.gradient(z, x)
print(gradient.numpy())  # [[2. 4.]
                         #  [6. 8.]]

适用场景：

算法研究和实验
模型调试
小规模训练

性能考虑：

每个操作都有Python调用开销
无法进行全局优化
适合小批量或调试阶段

1.2 Graph模式：生产部署

Graph模式先构建计算图，然后执行，支持全局优化。

tf.function使用

import tensorflow as tf
import numpy as np

# 使用tf.function装饰器转换为Graph模式
@tf.function
def train_step(x, y, model, optimizer):
    with tf.GradientTape() as tape:
        predictions = model(x, training=True)
        loss = tf.keras.losses.sparse_categorical_crossentropy(y, predictions)
        loss = tf.reduce_mean(loss)
    
    gradients = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    return loss

# 创建模型
model = tf.keras.Sequential([
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(10)
])
optimizer = tf.keras.optimizers.Adam()

# 训练循环
for epoch in range(10):
    for x_batch, y_batch in dataset:
        loss = train_step(x_batch, y_batch, model, optimizer)
    print(f"Epoch {epoch}, Loss: {loss.numpy()}")

tf.function最佳实践：

避免Python副作用

# 错误：Python print不会在每次执行时调用
@tf.function
def bad_function(x):
    print(f"Tracing with {x}")  # 只在trace时打印
    return x + 1

# 正确：使用tf.print
@tf.function
def good_function(x):
    tf.print("Executing with", x)  # 每次执行都打印
    return x + 1

控制重跟踪

# 使用input_signature避免每次输入形状变化都重新trace
@tf.function(input_signature=[
    tf.TensorSpec(shape=[None, 784], dtype=tf.float32),
    tf.TensorSpec(shape=[None], dtype=tf.int32)
])
def train_step(x, y):
    # 函数体
    pass

# 或者使用experimental_relax_shapes
@tf.function(experimental_relax_shapes=True)
def flexible_function(x):
    return tf.reduce_sum(x)

避免在循环中创建变量

# 错误：每次调用都创建新变量
@tf.function
def bad_update(x):
    v = tf.Variable(0.0)  # 错误！
    v.assign_add(x)
    return v

# 正确：在外部创建变量
v = tf.Variable(0.0)

@tf.function
def good_update(x):
    v.assign_add(x)
    return v

1.3 自定义层和模型

自定义Layer

class MyDenseLayer(tf.keras.layers.Layer):
    def __init__(self, units, activation=None):
        super().__init__()
        self.units = units
        self.activation = tf.keras.activations.get(activation)
    
    def build(self, input_shape):
        # 延迟创建权重，自动推断输入维度
        self.w = self.add_weight(
            name='kernel',
            shape=(input_shape[-1], self.units),
            initializer='glorot_uniform',
            trainable=True
        )
        self.b = self.add_weight(
            name='bias',
            shape=(self.units,),
            initializer='zeros',
            trainable=True
        )
        super().build(input_shape)
    
    def call(self, inputs, training=None):
        # 前向传播逻辑
        output = tf.matmul(inputs, self.w) + self.b
        if self.activation is not None:
            output = self.activation(output)
        return output
    
    def get_config(self):
        # 支持序列化
        config = super().get_config()
        config.update({
            'units': self.units,
            'activation': tf.keras.activations.serialize(self.activation)
        })
        return config

要点：

在__init__中定义配置参数
在build中创建权重（自动推断形状）
在call中实现前向逻辑
实现get_config支持序列化

自定义Model

class ResidualBlock(tf.keras.Model):
    def __init__(self, filters, strides=1):
        super().__init__()
        self.conv1 = tf.keras.layers.Conv2D(filters, 3, strides, padding='same')
        self.bn1 = tf.keras.layers.BatchNormalization()
        self.conv2 = tf.keras.layers.Conv2D(filters, 3, padding='same')
        self.bn2 = tf.keras.layers.BatchNormalization()
        
        if strides > 1:
            # 下采样shortcut
            self.shortcut = tf.keras.Sequential([
                tf.keras.layers.Conv2D(filters, 1, strides),
                tf.keras.layers.BatchNormalization()
            ])
        else:
            self.shortcut = lambda x: x
    
    def call(self, inputs, training=None):
        # 主路径
        x = self.conv1(inputs)
        x = self.bn1(x, training=training)
        x = tf.nn.relu(x)
        x = self.conv2(x)
        x = self.bn2(x, training=training)
        
        # 残差连接
        shortcut = self.shortcut(inputs)
        x = x + shortcut
        x = tf.nn.relu(x)
        return x

1.4 数据管道优化

tf.data高效输入

import tensorflow as tf

# 创建数据集
def create_dataset(file_pattern, batch_size, shuffle_buffer=10000):
    # 1. 从文件创建dataset
    dataset = tf.data.Dataset.list_files(file_pattern)
    
    # 2. 并行读取和解析
    dataset = dataset.interleave(
        lambda x: tf.data.TFRecordDataset(x),
        cycle_length=tf.data.AUTOTUNE,  # 自动调优并行度
        num_parallel_calls=tf.data.AUTOTUNE
    )
    
    # 3. 解析样本
    def parse_example(serialized):
        features = tf.io.parse_single_example(
            serialized,
            features={
                'image': tf.io.FixedLenFeature([], tf.string),
                'label': tf.io.FixedLenFeature([], tf.int64)
            }
        )
        image = tf.io.decode_jpeg(features['image'], channels=3)
        image = tf.image.resize(image, [224, 224])
        image = tf.cast(image, tf.float32) / 255.0
        label = features['label']
        return image, label
    
    dataset = dataset.map(
        parse_example,
        num_parallel_calls=tf.data.AUTOTUNE
    )
    
    # 4. 缓存（如果数据集不大）
    if should_cache:
        dataset = dataset.cache()
    
    # 5. 打乱
    dataset = dataset.shuffle(shuffle_buffer)
    
    # 6. 批处理
    dataset = dataset.batch(batch_size)
    
    # 7. 预取
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    
    return dataset

# 使用数据集
train_dataset = create_dataset('train/*.tfrecord', batch_size=32)
model.fit(train_dataset, epochs=10)

性能优化技巧：

并行解析：使用num_parallel_calls=AUTOTUNE
缓存：对于小数据集使用.cache()
预取：使用.prefetch(AUTOTUNE)重叠数据准备和训练
向量化：使用.batch()后再做数据增强（批量处理更高效）

1.5 分布式训练

MultiWorkerMirroredStrategy

import tensorflow as tf
import json
import os

# 配置集群
os.environ['TF_CONFIG'] = json.dumps({
    'cluster': {
        'worker': ['host1:12345', 'host2:12345']
    },
    'task': {'type': 'worker', 'index': 0}
})

# 创建分布式策略
strategy = tf.distribute.MultiWorkerMirroredStrategy()

# 在策略作用域内创建模型
with strategy.scope():
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(128, activation='relu'),
        tf.keras.layers.Dense(10)
    ])
    
    model.compile(
        optimizer=tf.keras.optimizers.Adam(),
        loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
        metrics=['accuracy']
    )

# 准备数据集（每个worker自动分片）
train_dataset = create_dataset('train/*.tfrecord', batch_size=64)

# 训练（自动分布式）
model.fit(train_dataset, epochs=10)

分布式策略选择：

策略	适用场景	特点
MirroredStrategy	单机多GPU	同步训练，参数镜像到每个GPU
MultiWorkerMirroredStrategy	多机多GPU	同步训练，使用AllReduce聚合梯度
TPUStrategy	TPU集群	专为TPU优化
ParameterServerStrategy	大规模异步训练	参数服务器架构，异步更新

二、实战经验

2.1 内存优化

梯度累积

# 当GPU内存不足时，使用梯度累积模拟大batch
class GradientAccumulationModel(tf.keras.Model):
    def __init__(self, model, accumulation_steps=4):
        super().__init__()
        self.model = model
        self.accumulation_steps = accumulation_steps
        self.accumulated_gradients = []
    
    def train_step(self, data):
        x, y = data
        
        # 初始化累积梯度
        if not self.accumulated_gradients:
            self.accumulated_gradients = [
                tf.Variable(tf.zeros_like(v), trainable=False)
                for v in self.model.trainable_variables
            ]
        
        # 累积梯度
        with tf.GradientTape() as tape:
            y_pred = self.model(x, training=True)
            loss = self.compiled_loss(y, y_pred)
            # 除以累积步数，相当于平均
            loss = loss / self.accumulation_steps
        
        gradients = tape.gradient(loss, self.model.trainable_variables)
        for i, grad in enumerate(gradients):
            self.accumulated_gradients[i].assign_add(grad)
        
        # 每accumulation_steps更新一次
        if (self.optimizer.iterations + 1) % self.accumulation_steps == 0:
            self.optimizer.apply_gradients(
                zip(self.accumulated_gradients, self.model.trainable_variables)
            )
            # 清零累积梯度
            for acc_grad in self.accumulated_gradients:
                acc_grad.assign(tf.zeros_like(acc_grad))
        
        self.compiled_metrics.update_state(y, y_pred)
        return {m.name: m.result() for m in self.metrics}

梯度检查点

# 使用gradient checkpointing减少内存占用
from tensorflow.python.keras.layers import Layer
import tensorflow as tf

class CheckpointedLayer(Layer):
    def __init__(self, layer):
        super().__init__()
        self.layer = layer
    
    def call(self, inputs):
        # 使用recompute_grad减少内存
        return tf.recompute_grad(self.layer)(inputs)

# 应用到模型
def create_checkpointed_model():
    model = tf.keras.Sequential()
    for _ in range(10):
        # 对内存消耗大的层使用checkpointing
        layer = tf.keras.layers.Dense(4096, activation='relu')
        model.add(CheckpointedLayer(layer))
    model.add(tf.keras.layers.Dense(10))
    return model

混合精度训练

# 使用FP16加速训练并减少内存
from tensorflow.keras import mixed_precision

# 启用混合精度
policy = mixed_precision.Policy('mixed_float16')
mixed_precision.set_global_policy(policy)

# 创建模型（自动使用FP16计算）
model = tf.keras.Sequential([
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(10)
])

# 使用loss scaling避免数值下溢
optimizer = tf.keras.optimizers.Adam()
optimizer = mixed_precision.LossScaleOptimizer(optimizer)

# 编译模型
model.compile(
    optimizer=optimizer,
    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    metrics=['accuracy']
)

混合精度效果：

训练速度提升1.5-3x（取决于硬件）
内存占用减少约50%
精度损失极小（通常<0.1%）

2.2 性能分析与优化

使用Profiler

import tensorflow as tf

# 启用profiler
tf.profiler.experimental.start('logdir')

# 运行要分析的代码
for step in range(100):
    train_step(x_batch, y_batch)
    
    # 周期性保存trace
    if step % 10 == 0:
        tf.profiler.experimental.trace_export(
            name=f'step_{step}',
            profiler_outdir='logdir'
        )

tf.profiler.experimental.stop()

# 使用TensorBoard查看
# tensorboard --logdir=logdir

Profiler分析重点：

Op时间分布：识别瓶颈操作
GPU利用率：目标>80%
数据输入时间：应远小于计算时间
内存使用：避免OOM

XLA编译加速

# 启用XLA编译
@tf.function(jit_compile=True)
def train_step(x, y):
    with tf.GradientTape() as tape:
        predictions = model(x, training=True)
        loss = loss_fn(y, predictions)
    gradients = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(gradients, model.trainable_variables))
    return loss

# 或者全局启用
tf.config.optimizer.set_jit(True)

XLA适用场景：

计算密集型模型（如Transformer）
较少的Python交互
固定的输入形状

性能提升：

通常10-30%加速
某些模型可达2-3x
首次编译需要额外时间

2.3 调试技巧

启用run_functions_eagerly

# 开发阶段启用Eager执行便于调试
tf.config.run_functions_eagerly(True)

@tf.function
def debug_function(x):
    # 现在可以使用Python debugger
    import pdb; pdb.set_trace()
    y = x + 1
    return y

# 生产环境关闭
tf.config.run_functions_eagerly(False)

数值稳定性检查

# 启用数值检查
tf.debugging.enable_check_numerics()

# 自定义检查
@tf.function
def safe_log(x):
    # 检查输入
    tf.debugging.assert_all_finite(x, "Input contains inf/nan")
    tf.debugging.assert_positive(x, "Log requires positive input")
    
    # 数值稳定的log
    return tf.math.log(x + 1e-7)

形状调试

@tf.function
def debug_shapes(x, y):
    # 打印张量形状
    tf.print("x shape:", tf.shape(x))
    tf.print("y shape:", tf.shape(y))
    
    # 断言形状
    tf.debugging.assert_equal(
        tf.shape(x)[0], tf.shape(y)[0],
        message="Batch sizes must match"
    )
    
    return x + y

三、最佳实践总结

3.1 性能优化清单

数据管道：

✅ 使用tf.data API
✅ 启用AUTOTUNE
✅ 使用prefetch
✅ 并行读取和解析
✅ 缓存小数据集

模型执行：

✅ 使用@tf.function
✅ 避免过度重tracing
✅ 使用input_signature
✅ 启用XLA（适用时）
✅ 使用混合精度

内存优化：

✅ 梯度累积（大模型）
✅ 梯度检查点（深模型）
✅ 合理的batch size
✅ 及时释放不用的变量

3.2 常见陷阱

陷阱1：在tf.function中使用Python状态

# 错误
counter = 0

@tf.function
def bad_function(x):
    global counter
    counter += 1  # 只在trace时执行一次！
    return x + counter

# 正确
counter = tf.Variable(0)

@tf.function
def good_function(x):
    counter.assign_add(1)
    return x + counter

陷阱2：不当的tensor转numpy

# 错误：频繁转换降低性能
@tf.function
def bad_function(x):
    if x.numpy()[0] > 0:  # 错误！破坏图模式
        return x + 1
    return x

# 正确：使用tf操作
@tf.function
def good_function(x):
    return tf.cond(x[0] > 0, lambda: x + 1, lambda: x)

陷阱3：忘记设置training参数

# 错误：推理时BN层行为错误
predictions = model(x)  # training=None，使用默认行为

# 正确
predictions = model(x, training=False)  # 推理模式
predictions = model(x, training=True)   # 训练模式

3.3 代码组织建议

project/
├── data/
│   ├── __init__.py
│   ├── dataset.py        # 数据加载和预处理
│   └── augmentation.py   # 数据增强
├── models/
│   ├── __init__.py
│   ├── base_model.py     # 基类
│   ├── resnet.py         # 具体模型
│   └── layers.py         # 自定义层
├── training/
│   ├── __init__.py
│   ├── trainer.py        # 训练循环
│   ├── callbacks.py      # 自定义回调
│   └── losses.py         # 自定义损失
├── utils/
│   ├── __init__.py
│   ├── config.py         # 配置管理
│   └── metrics.py        # 评估指标
├── train.py              # 训练入口
├── evaluate.py           # 评估脚本
└── requirements.txt

四、具体案例

4.1 图像分类完整流程

import tensorflow as tf
from tensorflow import keras

# 1. 数据准备
def create_image_dataset(data_dir, batch_size=32):
    dataset = keras.preprocessing.image_dataset_from_directory(
        data_dir,
        image_size=(224, 224),
        batch_size=batch_size
    )
    
    # 数据增强
    data_augmentation = keras.Sequential([
        keras.layers.RandomFlip("horizontal"),
        keras.layers.RandomRotation(0.1),
        keras.layers.RandomZoom(0.1),
    ])
    
    # 预处理
    def preprocess(images, labels):
        images = data_augmentation(images, training=True)
        images = keras.applications.resnet50.preprocess_input(images)
        return images, labels
    
    dataset = dataset.map(preprocess, num_parallel_calls=tf.data.AUTOTUNE)
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    return dataset

# 2. 模型定义
def create_model(num_classes):
    base_model = keras.applications.ResNet50(
        include_top=False,
        weights='imagenet',
        input_shape=(224, 224, 3)
    )
    base_model.trainable = False  # 冻结预训练权重
    
    model = keras.Sequential([
        base_model,
        keras.layers.GlobalAveragePooling2D(),
        keras.layers.Dense(256, activation='relu'),
        keras.layers.Dropout(0.5),
        keras.layers.Dense(num_classes, activation='softmax')
    ])
    
    return model

# 3. 训练配置
model = create_model(num_classes=10)

model.compile(
    optimizer=keras.optimizers.Adam(1e-3),
    loss=keras.losses.SparseCategoricalCrossentropy(),
    metrics=[keras.metrics.SparseCategoricalAccuracy()]
)

# 4. 回调函数
callbacks = [
    keras.callbacks.ModelCheckpoint(
        'best_model.h5',
        save_best_only=True,
        monitor='val_loss'
    ),
    keras.callbacks.EarlyStopping(
        patience=5,
        restore_best_weights=True
    ),
    keras.callbacks.ReduceLROnPlateau(
        factor=0.5,
        patience=3
    ),
    keras.callbacks.TensorBoard(log_dir='./logs')
]

# 5. 训练
train_dataset = create_image_dataset('train/', batch_size=32)
val_dataset = create_image_dataset('val/', batch_size=32)

history = model.fit(
    train_dataset,
    validation_data=val_dataset,
    epochs=50,
    callbacks=callbacks
)

# 6. 微调
base_model.trainable = True
model.compile(
    optimizer=keras.optimizers.Adam(1e-5),  # 更小的学习率
    loss=keras.losses.SparseCategoricalCrossentropy(),
    metrics=[keras.metrics.SparseCategoricalAccuracy()]
)

history_fine = model.fit(
    train_dataset,
    validation_data=val_dataset,
    epochs=20,
    callbacks=callbacks
)

# 7. 保存和部署
model.save('final_model')

4.2 自然语言处理：文本分类

import tensorflow as tf
from tensorflow import keras
import tensorflow_text as tf_text

# 1. 文本预处理
def create_text_dataset(texts, labels, batch_size=32):
    dataset = tf.data.Dataset.from_tensor_slices((texts, labels))
    dataset = dataset.shuffle(10000)
    dataset = dataset.batch(batch_size)
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    return dataset

# 2. 使用预训练BERT
class BERTClassifier(keras.Model):
    def __init__(self, num_classes, dropout_rate=0.1):
        super().__init__()
        # 加载预训练BERT
        self.bert = hub.KerasLayer(
            "https://tfhub.dev/tensorflow/bert_en_uncased_L-12_H-768_A-12/4",
            trainable=True
        )
        self.dropout = keras.layers.Dropout(dropout_rate)
        self.classifier = keras.layers.Dense(num_classes, activation='softmax')
    
    def call(self, inputs, training=None):
        # BERT输出
        pooled_output, sequence_output = self.bert(inputs)
        
        # 分类头
        x = self.dropout(pooled_output, training=training)
        output = self.classifier(x)
        return output

# 3. 创建和训练模型
model = BERTClassifier(num_classes=2)

model.compile(
    optimizer=keras.optimizers.Adam(2e-5),  # BERT推荐学习率
    loss=keras.losses.SparseCategoricalCrossentropy(),
    metrics=['accuracy']
)

# 4. 训练
train_dataset = create_text_dataset(train_texts, train_labels)
val_dataset = create_text_dataset(val_texts, val_labels)

model.fit(
    train_dataset,
    validation_data=val_dataset,
    epochs=3  # BERT通常只需少量epoch
)

4.3 强化学习：DQN

import tensorflow as tf
from tensorflow import keras
import numpy as np
from collections import deque
import random

class DQNAgent:
    def __init__(self, state_size, action_size):
        self.state_size = state_size
        self.action_size = action_size
        self.memory = deque(maxlen=2000)
        self.gamma = 0.95  # 折扣因子
        self.epsilon = 1.0  # 探索率
        self.epsilon_decay = 0.995
        self.epsilon_min = 0.01
        self.learning_rate = 0.001
        
        # 主网络和目标网络
        self.model = self._build_model()
        self.target_model = self._build_model()
        self.update_target_model()
    
    def _build_model(self):
        model = keras.Sequential([
            keras.layers.Dense(24, activation='relu', input_shape=(self.state_size,)),
            keras.layers.Dense(24, activation='relu'),
            keras.layers.Dense(self.action_size, activation='linear')
        ])
        model.compile(
            optimizer=keras.optimizers.Adam(self.learning_rate),
            loss='mse'
        )
        return model
    
    def update_target_model(self):
        # 同步目标网络权重
        self.target_model.set_weights(self.model.get_weights())
    
    def remember(self, state, action, reward, next_state, done):
        self.memory.append((state, action, reward, next_state, done))
    
    def act(self, state):
        if np.random.rand() <= self.epsilon:
            return random.randrange(self.action_size)
        
        act_values = self.model.predict(state, verbose=0)
        return np.argmax(act_values[0])
    
    @tf.function
    def _train_step(self, states, targets):
        with tf.GradientTape() as tape:
            predictions = self.model(states, training=True)
            loss = tf.reduce_mean(tf.square(targets - predictions))
        
        gradients = tape.gradient(loss, self.model.trainable_variables)
        self.model.optimizer.apply_gradients(
            zip(gradients, self.model.trainable_variables)
        )
        return loss
    
    def replay(self, batch_size):
        if len(self.memory) < batch_size:
            return
        
        minibatch = random.sample(self.memory, batch_size)
        
        states = np.array([m[0][0] for m in minibatch])
        actions = np.array([m[1] for m in minibatch])
        rewards = np.array([m[2] for m in minibatch])
        next_states = np.array([m[3][0] for m in minibatch])
        dones = np.array([m[4] for m in minibatch])
        
        # 使用目标网络计算Q值
        targets = self.model.predict(states, verbose=0)
        next_q_values = self.target_model.predict(next_states, verbose=0)
        
        for i in range(batch_size):
            if dones[i]:
                targets[i][actions[i]] = rewards[i]
            else:
                targets[i][actions[i]] = rewards[i] + self.gamma * np.max(next_q_values[i])
        
        # 训练
        loss = self._train_step(states, targets)
        
        # 衰减探索率
        if self.epsilon > self.epsilon_min:
            self.epsilon *= self.epsilon_decay
        
        return loss.numpy()

# 使用agent
agent = DQNAgent(state_size=4, action_size=2)

for episode in range(1000):
    state = env.reset()
    state = np.reshape(state, [1, 4])
    
    for time in range(500):
        action = agent.act(state)
        next_state, reward, done, _ = env.step(action)
        next_state = np.reshape(next_state, [1, 4])
        
        agent.remember(state, action, reward, next_state, done)
        state = next_state
        
        if done:
            agent.update_target_model()
            break
        
        if len(agent.memory) > 32:
            agent.replay(32)

五、总结

TensorFlow的高效使用需要：

理解执行模式：Eager用于开发，Graph用于生产
优化数据管道：使用tf.data API和AUTOTUNE
合理使用tf.function：避免常见陷阱
性能分析：使用Profiler识别瓶颈
内存优化：梯度累积、混合精度、检查点
分布式训练：选择合适的Strategy
调试技巧：数值检查、形状断言、Eager模式调试

掌握这些最佳实践能显著提升开发效率和模型性能。

TensorFlow 源码剖析 - 最佳实践与案例#

一、框架使用示例#

1.1 Eager模式：快速原型开发#

基本使用#

1.2 Graph模式：生产部署#

tf.function使用#

1.3 自定义层和模型#

自定义Layer#

自定义Model#

1.4 数据管道优化#

tf.data高效输入#

1.5 分布式训练#

MultiWorkerMirroredStrategy#

二、实战经验#

2.1 内存优化#

梯度累积#

梯度检查点#

混合精度训练#

2.2 性能分析与优化#

使用Profiler#

XLA编译加速#

2.3 调试技巧#

启用run_functions_eagerly#

数值稳定性检查#

形状调试#

三、最佳实践总结#

3.1 性能优化清单#

3.2 常见陷阱#

3.3 代码组织建议#

四、具体案例#

4.1 图像分类完整流程#

4.2 自然语言处理：文本分类#

4.3 强化学习：DQN#

五、总结#