从”调参玄学”到模型落地，一文搞懂Keras如何让深度学习变得触手可及

# 查看Keras版本
import keras
print(keras.__version__)

pip install tensorflow

import tensorflow as tf
from tensorflow import keras

print(f"TensorFlow版本: {tf.__version__}")
print(f"Keras版本: {keras.__version__}")

pip install keras

pip install tensorflow-gpu

import tensorflow as tf
print("GPU可用:", tf.config.list_physical_devices('GPU'))
print("GPU数量:", len(tf.config.list_physical_devices('GPU')))

# Google Colab中检查环境的完整代码
import sys
print(f"Python版本: {sys.version}")

import tensorflow as tf
print(f"TensorFlow版本: {tf.__version__}")
print(f"Keras版本: {tf.keras.__version__}")

# 检查GPU
gpu_info = !nvidia-smi --query-gpu=name,memory.total,memory.free --format=csv
print("GPU信息:", gpu_info)

# 导入所需库
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# 加载并预处理数据
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()

# 数据归一化到0-1范围，加速收敛
x_train = x_train.astype("float32") / 255
x_test = x_test.astype("float32") / 255

# 将标签转换为分类格式
num_classes = 10
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)

# 创建简单的Sequential模型
model = keras.Sequential([
    layers.Flatten(input_shape=(28, 28)),      # 将28x28图像展平为一维向量
    layers.Dense(128, activation='relu'),       # 全连接层，128个神经元
    layers.Dropout(0.2),                        # Dropout防止过拟合
    layers.Dense(num_classes, activation='softmax')  # 输出层，10个类别
])

# 查看模型结构
model.summary()

# Sequential模型示例
model = keras.Sequential([
    layers.Dense(256, activation='relu', input_shape=(784,)),
    layers.Dense(128, activation='relu'),
    layers.Dense(64, activation='relu'),
    layers.Dense(10, activation='softmax')
])

# Functional API示例：构建一个带有跳跃连接的网络
inputs = keras.Input(shape=(784,))
x = layers.Dense(256, activation='relu')(inputs)
skip = x  # 保存跳跃连接

x = layers.Dense(128, activation='relu')(x)
x = layers.Dense(128, activation='relu')(x)
x = layers.Add()([x, skip])  # 跳跃连接：主路径与跳跃路径相加

outputs = layers.Dense(10, activation='softmax')(x)

model = keras.Model(inputs=inputs, outputs=outputs, name='resnet_like_model')

# 选择指南
情况                          推荐方式
=====================================================================
简单的层堆叠                   Sequential API
多分支网络                     Functional API
共享层                        Functional API  
多输入/多输出                 Functional API
需要访问中间层输出             Functional API
复杂拓扑结构                   Functional API

# 全连接层
layers.Dense(units=128, activation='relu')

# 二维卷积层（图像处理必备）
layers.Conv2D(filters=64, kernel_size=(3, 3), activation='relu', padding='same')

# 池化层
layers.MaxPooling2D(pool_size=(2, 2))
layers.AveragePooling2D(pool_size=(2, 2))

# 循环神经网络层
layers.LSTM(units=128, return_sequences=True)
layers.GRU(units=128)

# 归一化层
layers.BatchNormalization()

# 正则化层
layers.Dropout(rate=0.5)  # 训练时随机丢弃50%的神经元

# 自定义层
class MyCustomLayer(layers.Layer):
    def __init__(self, units, **kwargs):
        super().__init__(**kwargs)
        self.units = units

    def build(self, input_shape):
        # 在第一次调用时创建权重
        self.w = self.add_weight(
            shape=(input_shape[-1], self.units),
            initializer='glorot_uniform',
            trainable=True
        )
        self.b = self.add_weight(
            shape=(self.units,),
            initializer='zeros',
            trainable=True
        )

    def call(self, inputs):
        # 前向传播逻辑
        return tf.matmul(inputs, self.w) + self.b

    def get_config(self):
        # 保存层的配置，用于模型序列化
        config = super().get_config()
        config.update({'units': self.units})
        return config

model.compile(
    optimizer='adam',      # 优化器：决定如何更新权重
    loss='categorical_crossentropy',  # 损失函数：衡量预测与真实的差距
    metrics=['accuracy']   # 评估指标：在训练过程中监控的指标
)

# 常用优化器及其适用场景

# Adam：最常用的默认选择，自适应学习率
optimizer='adam'
# 等价于
optimizer=keras.optimizers.Adam(learning_rate=0.001)

# AdamW：带权重衰减的Adam，正则化效果更好
optimizer=keras.optimizers.AdamW(learning_rate=0.001, weight_decay=0.01)

# SGD：随机梯度下降，常用于大规模图像任务（如ResNet训练）
optimizer=keras.optimizers.SGD(learning_rate=0.01, momentum=0.9)

# RMSprop：适合处理非平稳目标，如RNN任务
optimizer=keras.optimizers.RMSprop(learning_rate=0.001)

# Adagrad：适合处理稀疏数据
optimizer=keras.optimizers.Adagrad(learning_rate=0.01)

# 分类问题
loss='sparse_categorical_crossentropy'    # 标签是整数（0, 1, 2...）
loss='categorical_crossentropy'           # 标签是one-hot编码

# 二分类问题
loss='binary_crossentropy'

# 回归问题
loss='mse'                                # 均方误差
loss='mae'                                # 平均绝对误差
loss='huber'                              # Huber损失，对异常值更鲁棒

# 自定义损失函数
def custom_loss(y_true, y_pred):
    # 自定义逻辑
    return tf.reduce_mean(tf.square(y_true - y_pred))

model.compile(optimizer='adam', loss=custom_loss)

# 常用回调函数
callbacks = [
    # 保存最佳模型
    keras.callbacks.ModelCheckpoint(
        filepath='best_model.keras',
        monitor='val_loss',      # 监控验证集损失
        save_best_only=True,     # 只保存最佳模型
        mode='min'               # 损失越小越好
    ),

    # 早停：防止过拟合
    keras.callbacks.EarlyStopping(
        monitor='val_loss',
        patience=10,            # 等待10个epoch没有改善则停止
        restore_best_weights=True  # 恢复到最佳权重
    ),

    # 学习率调度
    keras.callbacks.ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.5,             # 学习率减半
        patience=5,             # 5个epoch没有改善则降低学习率
        min_lr=1e-7             # 学习率下限
    ),

    # 训练可视化
    keras.callbacks.TensorBoard(
        log_dir='./logs',
        histogram_freq=1        # 每1个epoch记录一次权重分布
    ),

    # 自定义回调
    keras.callbacks.LambdaCallback(
        on_epoch_end=lambda epoch, logs: 
            print(f"Epoch {epoch}: loss={logs['loss']:.4f}")
    )
]

# 在训练时应用回调
history = model.fit(
    x_train, y_train,
    epochs=50,
    batch_size=32,
    validation_split=0.2,
    callbacks=callbacks
)

import os
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# 定义数据路径
# 假设数据目录结构如下：
# data/
#   train/
#     cats/
#     dogs/
#   validation/
#     cats/
#     dogs/

train_dir = 'data/train'
validation_dir = 'data/validation'

# 使用image_dataset_from_directory自动加载数据
# 这个方法会自动从目录结构推断类别
image_size = (180, 180)  # 图片统一 resize 到这个尺寸
batch_size = 32

# 加载训练数据
train_ds = keras.utils.image_dataset_from_directory(
    train_dir,
    image_size=image_size,
    batch_size=batch_size,
    shuffle=True,
    seed=42,  # 设置随机种子保证可复现性
    label_mode='categorical'  # 返回one-hot编码的标签
)

# 加载验证数据
val_ds = keras.utils.image_dataset_from_directory(
    validation_dir,
    image_size=image_size,
    batch_size=batch_size,
    shuffle=False,  # 验证集不需要打乱
    seed=42,
    label_mode='categorical'
)

# 查看数据信息
print("类别:", train_ds.class_names)  # ['cats', 'dogs']
print("训练样本数:", train_ds.cardinality())  # 应该是一个整数

# 数据标准化：确保像素值在合理范围内
# ResNet等预训练模型期望输入是0-1范围或-1到1范围
# 这里使用keras.layers.Rescaling进行缩放
rescale_layer = layers.Rescaling(1./255)

# 数据增强：通过随机变换增加训练样本多样性
data_augmentation = keras.Sequential([
    layers.RandomFlip("horizontal"),      # 随机水平翻转
    layers.RandomRotation(0.1),           # 随机旋转±10%
    layers.RandomZoom(0.1),              # 随机缩放
    layers.RandomContrast(0.1),           # 随机对比度调整
], name="data_augmentation")

# 创建预处理管道
def preprocess_image(image, label):
    """
    数据预处理函数：
    1. 数据增强（仅在训练时应用）
    2. 像素值归一化
    """
    # 数据增强
    image = data_augmentation(image)
    # 归一化
    image = rescale_layer(image)
    return image, label

# 应用预处理
# 使用 tf.data API 进行高效数据管道
AUTOTUNE = tf.data.AUTOTUNE

train_ds = (
    train_ds
    .map(preprocess_image, num_parallel_calls=AUTOTUNE)  # 并行预处理
    .cache()                                            # 缓存数据加速
    .shuffle(buffer_size=1000)                          # 打乱数据
    .prefetch(buffer_size=AUTOTUNE)                     # 预取下一批数据
)

val_ds = (
    val_ds
    .map(preprocess_image, num_parallel_calls=AUTOTUNE)
    .cache()
    .prefetch(buffer_size=AUTOTUNE)
)

# 优化数据加载性能
# 关键技巧：使用 buffered prefetching 避免 I/O 阻塞
print("数据管道配置完成，正在进行模型构建...")

# 使用Functional API构建一个完整的卷积神经网络
def create_cnn_model(input_shape=(180, 180, 3), num_classes=2):
    """
    创建用于猫狗分类的卷积神经网络

    网络结构：
    - 4个卷积块（Conv2D + BatchNorm + MaxPooling）
    - 全连接层
    - Dropout防止过拟合
    """
    inputs = keras.Input(shape=input_shape)

    # 第一个卷积块：64个滤波器
    x = layers.Conv2D(64, (3, 3), padding='same', activation='relu')(inputs)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)

    # 第二个卷积块：128个滤波器
    x = layers.Conv2D(128, (3, 3), padding='same', activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)

    # 第三个卷积块：256个滤波器
    x = layers.Conv2D(256, (3, 3), padding='same', activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)

    # 第四个卷积块：512个滤波器
    x = layers.Conv2D(512, (3, 3), padding='same', activation='relu')(x)
    x = layers.BatchNormalization()(x)
    x = layers.MaxPooling2D((2, 2))(x)

    # 全局池化：代替Flatten，参数更少效果更好
    x = layers.GlobalAveragePooling2D()(x)

    # 全连接层
    x = layers.Dense(256, activation='relu')(x)
    x = layers.Dropout(0.5)(x)  # 训练时随机丢弃50%的神经元

    # 输出层
    if num_classes == 2:
        outputs = layers.Dense(1, activation='sigmoid')(x)
        loss = 'binary_crossentropy'
    else:
        outputs = layers.Dense(num_classes, activation='softmax')(x)
        loss = 'categorical_crossentropy'

    model = keras.Model(inputs=inputs, outputs=outputs)

    # 编译模型
    model.compile(
        optimizer=keras.optimizers.Adam(learning_rate=1e-4),
        loss=loss,
        metrics=['accuracy']
    )

    return model

# 创建模型实例
model = create_cnn_model(input_shape=(180, 180, 3), num_classes=2)
model.summary()

# 定义回调函数
callbacks = [
    # 保存最佳模型
    keras.callbacks.ModelCheckpoint(
        filepath='cat_dog_classifier.keras',
        monitor='val_accuracy',
        save_best_only=True,
        mode='max'
    ),

    # 早停策略
    keras.callbacks.EarlyStopping(
        monitor='val_loss',
        patience=5,
        restore_best_weights=True,
        verbose=1
    ),

    # 学习率衰减
    keras.callbacks.ReduceLROnPlateau(
        monitor='val_loss',
        factor=0.5,
        patience=3,
        min_lr=1e-7,
        verbose=1
    ),
]

# 开始训练
print("开始训练...")
epochs = 30

history = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=epochs,
    callbacks=callbacks,
    verbose=1  # 显示进度条
)

print("训练完成！")

import matplotlib.pyplot as plt

def plot_training_history(history):
    """
    绘制训练过程中的损失和准确率曲线
    帮助诊断模型是否过拟合或欠拟合
    """
    history_dict = history.history

    # 创建图表
    fig, axes = plt.subplots(1, 2, figsize=(14, 5))

    # 绘制损失曲线
    epochs_range = range(1, len(history_dict['loss']) + 1)

    axes[0].plot(epochs_range, history_dict['loss'], 'b-', label='训练损失', linewidth=2)
    axes[0].plot(epochs_range, history_dict['val_loss'], 'r-', label='验证损失', linewidth=2)
    axes[0].set_xlabel('Epoch', fontsize=12)
    axes[0].set_ylabel('Loss', fontsize=12)
    axes[0].set_title('模型损失', fontsize=14, fontweight='bold')
    axes[0].legend(fontsize=11)
    axes[0].grid(True, alpha=0.3)

    # 绘制准确率曲线
    axes[1].plot(epochs_range, history_dict['accuracy'], 'b-', label='训练准确率', linewidth=2)
    axes[1].plot(epochs_range, history_dict['val_accuracy'], 'r-', label='验证准确率', linewidth=2)
    axes[1].set_xlabel('Epoch', fontsize=12)
    axes[1].set_ylabel('Accuracy', fontsize=12)
    axes[1].set_title('模型准确率', fontsize=14, fontweight='bold')
    axes[1].legend(fontsize=11)
    axes[1].grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig('training_history.png', dpi=150, bbox_inches='tight')
    plt.show()

    # 打印最终结果
    final_train_acc = history_dict['accuracy'][-1]
    final_val_acc = history_dict['val_accuracy'][-1]
    print(f"\n最终训练准确率: {final_train_acc:.4f}")
    print(f"最终验证准确率: {final_val_acc:.4f}")
    print(f"过拟合程度: {(final_train_acc - final_val_acc):.4f}")

# 调用可视化函数
plot_training_history(history)

# 加载最佳模型进行评估
best_model = keras.models.load_model('cat_dog_classifier.keras')

# 在测试集上评估
print("在测试集上评估模型...")
test_ds = keras.utils.image_dataset_from_directory(
    'data/test',
    image_size=(180, 180),
    batch_size=32,
    label_mode='categorical'
)

# 预处理测试数据
test_ds = test_ds.map(lambda x, y: (rescale_layer(x), y))

results = best_model.evaluate(test_ds, verbose=1)
print(f"\n测试集结果:")
print(f"损失: {results[0]:.4f}")
print(f"准确率: {results[1]:.4f}")

# 对单张图片进行预测
def predict_single_image(model, image_path):
    """
    对单张图片进行预测
    """
    # 加载图片
    img = keras.preprocessing.image.load_img(
        image_path,
        target_size=(180, 180)
    )

    # 转换为数组
    img_array = keras.preprocessing.image.img_to_array(img)

    # 添加 batch 维度
    img_array = tf.expand_dims(img_array, 0)

    # 归一化
    img_array = img_array / 255.0

    # 预测
    predictions = model.predict(img_array)

    # 解析结果
    class_names = ['猫', '狗']
    predicted_class = class_names[int(predictions[0][0] > 0.5)]
    confidence = float(predictions[0][0]) if predictions[0][0] > 0.5 else 1 - float(predictions[0][0])

    return predicted_class, confidence

# 示例预测
# predicted_class, confidence = predict_single_image(best_model, 'test_images/cat_01.jpg')
# print(f"预测结果: {predicted_class}, 置信度: {confidence:.2%}")

# 常用预训练模型及特点
# ==============================

# ResNet50：经典残差网络，50层深，平衡性能和速度
# VGG16/VGG19：简单结构，但参数量大
# InceptionV3： inception模块设计，计算效率高
# MobileNetV2：专为移动端设计，轻量级
# EfficientNetB0-B7：当前最优的效率-精度平衡

# Keras内置的预训练模型
from tensorflow.keras.applications import (
    ResNet50, ResNet101,
    VGG16, VGG19,
    InceptionV3, InceptionResNetV2,
    MobileNetV2,
    EfficientNetB0, EfficientNetB1, EfficientNetB7
)

def create_transfer_learning_model(input_shape=(224, 224, 3), num_classes=2):
    """
    使用 EfficientNetB0 进行迁移学习

    策略：
    1. 冻结基础模型的所有层（不训练）
    2. 添加自定义分类头
    3. 先训练分类头
    4. 然后解冻基础模型进行微调
    """
    # 加载预训练模型，不包含顶层分类器
    base_model = EfficientNetB0(
        weights='imagenet',           # 使用ImageNet预训练权重
        include_top=False,            # 不包含顶层分类器
        input_shape=input_shape
    )

    # 第一阶段：冻结基础模型，只训练新添加的层
    base_model.trainable = False

    # 创建模型
    inputs = keras.Input(shape=input_shape)

    # 数据增强
    x = layers.Rescaling(1./255)(inputs)

    # 基础模型
    x = base_model(x, training=False)  # 关键：确保BatchNorm层不更新

    # 全局池化
    x = layers.GlobalAveragePooling2D()(x)

    # 添加 Dropout 防止过拟合
    x = layers.Dropout(0.3)(x)

    # 分类头
    x = layers.Dense(256, activation='relu')(x)
    x = layers.Dropout(0.3)(x)

    # 输出层
    if num_classes == 2:
        outputs = layers.Dense(1, activation='sigmoid')(x)
    else:
        outputs = layers.Dense(num_classes, activation='softmax')(x)

    model = keras.Model(inputs, outputs)

    return model, base_model

# 创建迁移学习模型
model, base_model = create_transfer_learning_model()

# 编译（冻结状态下）
model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=1e-3),
    loss='binary_crossentropy',
    metrics=['accuracy']
)

print("第一阶段：训练分类头...")
print("=" * 50)

# 第一阶段：只训练分类头
print("阶段1：训练新添加的层...")
history_phase1 = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=10,
    callbacks=[
        keras.callbacks.EarlyStopping(patience=3, restore_best_weights=True)
    ]
)

# 第二阶段：解冻基础模型，进行微调
print("\n阶段2：解冻基础模型进行微调...")
base_model.trainable = True

# 重要：冻结BatchNorm层
for layer in base_model.layers:
    if isinstance(layer, keras.layers.BatchNormalization):
        layer.trainable = False

# 使用更小的学习率进行微调
model.compile(
    optimizer=keras.optimizers.Adam(learning_rate=1e-5),  # 降低100倍
    loss='binary_crossentropy',
    metrics=['accuracy']
)

# 训练整个模型
history_phase2 = model.fit(
    train_ds,
    validation_data=val_ds,
    epochs=20,
    callbacks=[
        keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True),
        keras.callbacks.ReduceLROnPlateau(factor=0.5, patience=2)
    ]
)

# 方式1：保存为 .keras 格式（Keras 3.0 推荐）
model.save('my_model.keras')

# 方式2：保存为 TensorFlow SavedModel 格式（兼容性好）
model.save('my_saved_model')

# 方式3：仅保存权重
model.save_weights('my_model_weights.weights.h5')

# 加载模型
loaded_model = keras.models.load_model('my_model.keras')

# 验证加载的模型是否正确
test_loss, test_acc = loaded_model.evaluate(test_ds)
print(f"加载模型测试准确率: {test_acc:.4f}")

# 转换为 TensorFlow Lite 格式
converter = tf.lite.TFLiteConverter.from_keras_model(model)

# 可选：量化以减小模型大小（轻微影响精度）
converter.optimizations = [tf.lite.Optimize.DEFAULT]

# 完整整型量化（需要代表性数据集）
# converter.representative_dataset = lambda: train_ds.unbatch().batch(1).take(100)

tflite_model = converter.convert()

# 保存 TFLite 模型
with open('model.tflite', 'wb') as f:
    f.write(tflite_model)

print("TFLite 模型已导出，大小:", os.path.getsize('model.tflite') / 1024 / 1024, "MB")

# 在边缘设备上使用 TFLite 模型
import tflite_runtime.interpreter as tflite

# 加载模型
interpreter = tflite.Interpreter(model_path='model.tflite')
interpreter.allocate_tensors()

# 获取输入输出张量
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# 准备输入数据
input_data = np.array(image, dtype=np.float32)
interpreter.set_tensor(input_details[0]['index'], input_data)

# 执行推理
interpreter.invoke()

# 获取结果
output_data = interpreter.get_tensor(output_details[0]['index'])
predicted_class = '狗' if output_data[0][0] > 0.5 else '猫'

# 问题：训练时报 OOM (Out of Memory) 错误

# 解决方案1：减小 batch size
model.fit(train_ds.batch(16), ...)  # 从32降到16或8

# 解决方案2：使用混合精度训练（需要GPU支持）
from tensorflow.keras import mixed_precision
mixed_precision.set_global_policy('mixed_float16')

# 解决方案3：使用梯度累积
# 在小batch上模拟大批次效果

# 解决方案4：清理内存
import gc
del old_model
gc.collect()
tf.keras.backend.clear_session()

# 过拟合的信号：训练准确率高但验证准确率低

# 解决方案1：增加数据增强
data_augmentation = keras.Sequential([
    layers.RandomFlip("horizontal"),
    layers.RandomFlip("vertical"),    # 添加垂直翻转
    layers.RandomRotation(0.2),       # 增加旋转范围
    layers.RandomZoom(0.2),           # 增加缩放范围
    layers.RandomTranslation(0.1, 0.1),  # 添加平移
])

# 解决方案2：增加 Dropout
layers.Dropout(0.5)  # 从0.3增加到0.5

# 解决方案3：使用正则化
model = keras.Sequential([
    layers.Dense(256, kernel_regularizer=keras.regularizers.l2(0.01), activation='relu'),
    # L2正则化惩罚大的权重
])

# 解决方案4：早停
keras.callbacks.EarlyStopping(patience=10)

# 解决方案5：减少模型容量
layers.Dense(128)  # 从256减少到128

# 问题：训练过程中梯度变得极小或极大

# 解决方案1：使用残差连接
def residual_block(x, filters):
    shortcut = x
    x = layers.Conv2D(filters, (3, 3), padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Activation('relu')(x)
    x = layers.Conv2D(filters, (3, 3), padding='same')(x)
    x = layers.BatchNormalization()(x)
    x = layers.Add()([x, shortcut])  # 跳跃连接
    x = layers.Activation('relu')(x)
    return x

# 解决方案2：使用合适的权重初始化
layers.Dense(128, kernel_initializer='he_normal')

# 解决方案3：使用 Batch Normalization
layers.BatchNormalization()

# 解决方案4：使用梯度裁剪
optimizer=keras.optimizers.Adam(clipnorm=1.0)  # 裁剪梯度范数

# 问题：损失值出现 NaN 或震荡

# 解决方案1：检查数据是否有异常值
print("数据范围:", x_train.min(), x_train.max())

# 解决方案2：调整学习率
optimizer=keras.optimizers.Adam(learning_rate=1e-4)  # 从1e-3降到1e-4

# 解决方案3：使用学习率预热
lr_schedule = keras.optimizers.schedules.CosineDecay(
    initial_learning_rate=1e-4,
    decay_steps=10000,
    alpha=0.1
)
optimizer=keras.optimizers.Adam(learning_rate=lr_schedule)

# 解决方案4：检查损失函数是否正确
# 二分类问题使用 sigmoid + binary_crossentropy
# 多分类问题使用 softmax + categorical_crossentropy

# 解决方案5：梯度检查
for grad, var in zip(gradients, model.trainable_variables):
    if tf.math.is_nan(tf.reduce_mean(grad)):
        print(f"NaN 梯度在层: {var.name}")

# 使用 tf.data 优化数据管道
train_ds = (
    train_ds
    .cache()                           # 将数据集缓存在内存或磁盘
    .shuffle(buffer_size=1000)         # 打乱数据
    .prefetch(buffer_size=AUTOTUNE)    # 预取数据
)

# 使用多线程并行加载
train_ds = train_ds.map(
    preprocess_image,
    num_parallel_calls=tf.data.AUTOTUNE
)

# 使用 prefetch 避免 I/O 瓶颈
# prefetch(n) 会在训练的同时预取 n 个 batch

# 使用混合精度加速训练
from tensorflow.keras import mixed_precision

# 设置策略为混合 float16/float32
mixed_precision.set_global_policy('mixed_float16')

# 某些操作（如 softmax）使用 float32 以保持精度
outputs = layers.Dense(num_classes, activation='softmax', dtype='float32')(x)

# 注意：需要 NVIDIA GPU + TensorCore 支持

# 单机多卡训练
strategy = tf.distribute.MirroredStrategy()
print(f"使用 {strategy.num_replicas_in_sync} 个 GPU")

with strategy.scope():
    model = create_cnn_model()
    model.compile(
        optimizer=keras.optimizers.Adam(learning_rate=0.001 * strategy.num_replicas_in_sync),
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )

# 多机多卡训练需要配置 TF_CONFIG 环境变量

"""
Keras 深度学习项目模板
基于最佳实践构建
"""

import os
import numpy as np
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

# ===========================
# 配置区域
# ===========================
class Config:
    SEED = 42
    IMAGE_SIZE = (224, 224)
    BATCH_SIZE = 32
    EPOCHS = 50
    LEARNING_RATE = 1e-4
    DATA_DIR = './data'
    MODEL_DIR = './models'

    # 设置随机种子保证可复现
    tf.random.set_seed(SEED)
    np.random.seed(SEED)

# ===========================
# 数据加载
# ===========================
def load_data(data_dir, batch_size, image_size, validation_split=0.2):
    """
    加载图像数据集
    """
    train_ds = keras.utils.image_dataset_from_directory(
        data_dir,
        image_size=image_size,
        batch_size=batch_size,
        validation_split=validation_split,
        subset='training',
        seed=Config.SEED
    )

    val_ds = keras.utils.image_dataset_from_directory(
        data_dir,
        image_size=image_size,
        batch_size=batch_size,
        validation_split=validation_split,
        subset='validation',
        seed=Config.SEED
    )

    return train_ds, val_ds

# ===========================
# 数据预处理
# ===========================
def create_preprocessing_pipeline(image_size):
    """
    创建数据预处理管道
    """
    return keras.Sequential([
        layers.Rescaling(1./255),
        layers.RandomFlip('horizontal'),
        layers.RandomRotation(0.1),
    ])

# ===========================
# 模型构建
# ===========================
def build_model(input_shape, num_classes):
    """
    构建深度学习模型
    """
    inputs = keras.Input(shape=input_shape)

    # 预处理
    x = create_preprocessing_pipeline(input_shape[:2])(inputs)

    # 骨干网络
    base_model = keras.applications.EfficientNetB0(
        weights='imagenet',
        include_top=False,
        input_shape=input_shape
    )
    base_model.trainable = False
    x = base_model(x, training=False)

    # 分类头
    x = layers.GlobalAveragePooling2D()(x)
    x = layers.Dropout(0.3)(x)
    x = layers.Dense(256, activation='relu')(x)
    x = layers.Dropout(0.3)(x)
    outputs = layers.Dense(num_classes, activation='softmax')(x)

    return keras.Model(inputs, outputs)

# ===========================
# 训练流程
# ===========================
def train_model(model, train_ds, val_ds, epochs, callbacks=None):
    """
    训练模型
    """
    model.compile(
        optimizer=keras.optimizers.Adam(Config.LEARNING_RATE),
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )

    history = model.fit(
        train_ds,
        validation_data=val_ds,
        epochs=epochs,
        callbacks=callbacks or []
    )

    return history

# ===========================
# 主函数
# ===========================
def main():
    # 加载数据
    train_ds, val_ds = load_data(
        Config.DATA_DIR,
        Config.BATCH_SIZE,
        Config.IMAGE_SIZE
    )

    # 获取类别数
    num_classes = len(train_ds.class_names)

    # 构建模型
    model = build_model(
        input_shape=(*Config.IMAGE_SIZE, 3),
        num_classes=num_classes
    )

    # 定义回调
    callbacks = [
        keras.callbacks.ModelCheckpoint(
            os.path.join(Config.MODEL_DIR, 'best_model.keras'),
            monitor='val_accuracy',
            save_best_only=True
        ),
        keras.callbacks.EarlyStopping(
            monitor='val_loss',
            patience=5,
            restore_best_weights=True
        )
    ]

    # 训练
    history = train_model(model, train_ds, val_ds, Config.EPOCHS, callbacks)

    return model, history

if __name__ == '__main__':
    model, history = main()

核心技能清单
==================================================
1. Keras 环境的安装与配置
2. Sequential API 和 Functional API 的使用
3. 常用层的类型和自定义层的方法
4. 模型编译：优化器、损失函数、评估指标的选择
5. 回调函数的使用：早停、学习率调度、模型保存
6. 图像数据加载与预处理
7. CNN 网络的构建与训练
8. 数据增强技术
9. 迁移学习与预训练模型使用
10. 模型保存、加载与部署