一天一个Python知识点——Day 152:TensorFlow基础

import matplotlib.pyplot as pltdef visualize_tensorflow_ecosystem():    """可视化TensorFlow生态系统"""    components = {        'TensorFlow Core': ['张量计算', '自动微分', 'GPU加速'],        'Keras': ['高级API', '模型构建', '快速原型'],        'TensorFlow.js': ['浏览器部署', 'Node.js集成'],        'TensorFlow Lite': ['移动设备', '嵌入式系统', '模型量化'],        'TensorFlow Extended (TFX)': ['生产流水线', '数据验证', '模型分析'],        'TensorFlow Hub': ['预训练模型', '模型重用', '迁移学习'],        'TensorBoard': ['可视化', '模型分析', '实验跟踪'],        'TensorFlow Serving': ['模型部署', 'REST/GRPC', '版本管理']    }    # 创建生态系统图    fig, ax = plt.subplots(figsize=(14, 8))    ax.axis('off')    # 设置位置    positions = {        'TensorFlow Core': (0.5, 0.8),        'Keras': (0.3, 0.6),        'TensorFlow.js': (0.1, 0.4),        'TensorFlow Lite': (0.3, 0.4),        'TensorFlow Extended (TFX)': (0.7, 0.6),        'TensorFlow Hub': (0.5, 0.4),        'TensorBoard': (0.9, 0.6),        'TensorFlow Serving': (0.7, 0.4)    }    # 绘制组件    for component, (x, y) in positions.items():        # 绘制框        box = plt.Rectangle((x-0.1, y-0.05), 0.2, 0.08,                           facecolor='lightblue', edgecolor='blue', alpha=0.8)        ax.add_patch(box)        # 添加文本        ax.text(x, y, component, ha='center', va='center',                 fontsize=10, fontweight='bold')        # 添加功能描述        features = components[component]        feature_text = '\n'.join(features)        ax.text(x, y-0.03, feature_text, ha='center', va='top',                 fontsize=8, fontstyle='italic')    # 添加连接线    connections = [        ('TensorFlow Core', 'Keras'),        ('TensorFlow Core', 'TensorFlow Extended (TFX)'),        ('TensorFlow Core', 'TensorFlow Hub'),        ('TensorFlow Core', 'TensorBoard'),        ('TensorFlow Core', 'TensorFlow Serving'),        ('Keras', 'TensorFlow.js'),        ('Keras', 'TensorFlow Lite')    ]    for start, end in connections:        x1, y1 = positions[start]        x2, y2 = positions[end]        ax.annotate('', xy=(x2, y2-0.05), xytext=(x1, y1+0.05),                   arrowprops=dict(arrowstyle='->', color='gray', alpha=0.6))    ax.set_xlim(0, 1)    ax.set_ylim(0, 1)    ax.set_title('TensorFlow生态系统架构', fontsize=16, fontweight='bold')    plt.show()visualize_tensorflow_ecosystem()

def compare_tensorflow_versions():    """比较TensorFlow不同版本"""    versions_data = {        '版本': ['TensorFlow 1.x', 'TensorFlow 2.0-2.4', 'TensorFlow 2.5+', 'TensorFlow 2.13+'],        '发布时间': ['2015-2018', '2019-2021', '2021-2022', '2023+'],        '主要特性': [            '静态计算图\n需要Session\nAPI较复杂',            'Eager Execution默认\nKeras集成\nAPI简化',            '混合精度训练\n性能优化\n新Keras API',            '统一API\n更好性能\n新硬件支持'        ],        'API风格': [            'tf.placeholder\ntf.Session.run()',            'tf.function装饰器\nKeras Model',            '新Keras层\n更好的分布式',            '更简洁API\n模块化设计'        ],        '推荐用户': [            '遗留项目维护\n图模式专家',            '大多数用户\n从零开始项目',            '性能敏感应用\n大规模训练',            '最新功能需求\n前沿项目'        ]    }    import pandas as pd    df = pd.DataFrame(versions_data)    print("TensorFlow版本对比")    print("=" * 120)    print(df.to_string(index=False))    print("\n升级建议:")    print("1. 新项目: 使用TensorFlow 2.x最新版本")    print("2. 迁移项目: 使用tf_upgrade_v2工具")    print("3. 性能关键: TensorFlow 2.5+ + XLA编译")    print("4. 移动端: TensorFlow Lite + 量化")    return df# 显示版本对比compare_tensorflow_versions()

def setup_tensorflow_environment():    """TensorFlow环境配置指南"""    setups = {        '基础安装': {            '命令': 'pip install tensorflow',            '说明': '安装CPU版本（适合学习和开发）',            '验证代码': '''import tensorflow as tfprint(f"TensorFlow版本: {tf.__version__}")print(f"GPU是否可用: {tf.config.list_physical_devices('GPU')}")            '''        },        'GPU支持': {            '命令': 'pip install tensorflow[and-cuda]',            '说明': '安装CUDA支持的GPU版本（需要NVIDIA GPU）',            '前提条件': [                'NVIDIA GPU (Compute Capability 3.5+)',                'CUDA Toolkit (11.2-11.8)',                'cuDNN SDK (8.1-8.6)'            ]        },        'Docker安装': {            '命令': 'docker pull tensorflow/tensorflow:latest-gpu',            '说明': '使用Docker容器（隔离环境）',            '运行命令': 'docker run -it tensorflow/tensorflow:latest-gpu python'        },        '特定版本': {            '命令': 'pip install tensorflow==2.13.0',            '说明': '安装特定版本（生产环境推荐）'        },        'Jupyter支持': {            '命令': 'pip install tensorflow jupyter',            '说明': '安装Jupyter支持'        }    }    print("TensorFlow安装配置指南")    print("=" * 80)    for setup_type, setup_info in setups.items():        print(f"\n{setup_type}:")        print(f"  命令: {setup_info['命令']}")        print(f"  说明: {setup_info['说明']}")        if '前提条件' in setup_info:            print(f"  前提条件:")            for req in setup_info['前提条件']:                print(f"    • {req}")        if '验证代码' in setup_info:            print(f"  验证代码: {setup_info['验证代码']}")    print("\n" + "=" * 80)    print("\n验证安装（复制运行以下代码）:")    print("""import tensorflow as tf# 打印版本信息print(f"TensorFlow版本: {tf.__version__}")# 检查GPUgpus = tf.config.list_physical_devices('GPU')if gpus:    print(f"找到GPU: {gpus}")    for gpu in gpus:        tf.config.experimental.set_memory_growth(gpu, True)else:    print("未找到GPU，使用CPU")    """)# 显示安装指南setup_tensorflow_environment()

def configure_gpu_for_tensorflow():    """配置TensorFlow GPU使用"""    config_code = '''import tensorflow as tfimport osdef configure_gpu_settings():    """配置GPU设置"""    # 1. 检查可用GPU    gpus = tf.config.list_physical_devices('GPU')    print(f"可用GPU数量: {len(gpus)}")    if not gpus:        print("警告: 未找到GPU，使用CPU运行")        return    # 2. 设置显存增长（避免一次性占用所有显存）    for gpu in gpus:        tf.config.experimental.set_memory_growth(gpu, True)    # 3. 设置可见GPU（多GPU环境）    # tf.config.set_visible_devices(gpus[0], 'GPU')  # 只使用第一个GPU    # 4. 设置逻辑GPU设备（虚拟多个GPU）    # try:    #     tf.config.set_logical_device_configuration(    #         gpus[0],    #         [tf.config.LogicalDeviceConfiguration(memory_limit=2048)] * 2    #     )    #     print("创建了2个逻辑GPU")    # except RuntimeError as e:    #     print(f"配置失败: {e}")    # 5. 设置GPU设备    os.environ["CUDA_VISIBLE_DEVICES"] = "0"  # 使用第一个GPU    # 6. 设置混合精度（加快训练速度）    # policy = tf.keras.mixed_precision.Policy('mixed_float16')    # tf.keras.mixed_precision.set_global_policy(policy)    # print(f"计算精度: {policy}")    print("GPU配置完成")def check_gpu_performance():    """检查GPU性能"""    # 创建测试张量    size = 10000    a = tf.random.normal([size, size])    b = tf.random.normal([size, size])    # 测量矩阵乘法时间    import time    start = time.time()    c = tf.matmul(a, b)    elapsed = time.time() - start    print(f"矩阵乘法 ({size}x{size}) 耗时: {elapsed:.3f}秒")    print(f"结果形状: {c.shape}")    return elapsed# 运行配置configure_gpu_settings()check_gpu_performance()    '''    print("TensorFlow GPU配置与优化")    print("=" * 80)    print(config_code)    print("\n常见GPU问题解决方案:")    print("1. CUDA版本不匹配: 使用 conda install cudatoolkit=11.2 cudnn=8.1")    print("2. 显存不足: 设置 memory_growth=True 或 batch_size更小")    print("3. 多GPU训练: 使用 tf.distribute.MirroredStrategy()")    print("4. 性能优化: 启用XLA编译 tf.config.optimizer.set_jit(True)")# 显示GPU配置configure_gpu_for_tensorflow()

import tensorflow as tfimport numpy as npdef tensor_basics():    """TensorFlow张量基础"""    print("=" * 60)    print("TensorFlow张量基础")    print("=" * 60)    # 1. 创建张量    print("\n1. 创建张量:")    # 标量 (0维张量)    scalar = tf.constant(42)    print(f"标量: {scalar}, 形状: {scalar.shape}, 数据类型: {scalar.dtype}")    # 向量 (1维张量)    vector = tf.constant([1, 2, 3, 4, 5])    print(f"向量: {vector}, 形状: {vector.shape}")    # 矩阵 (2维张量)    matrix = tf.constant([[1, 2], [3, 4], [5, 6]])    print(f"矩阵: {matrix}, 形状: {matrix.shape}")    # 3维张量    tensor_3d = tf.constant([[[1, 2], [3, 4]], [[5, 6], [7, 8]]])    print(f"3维张量: 形状: {tensor_3d.shape}")    # 2. 特殊张量    print("\n2. 特殊张量:")    # 零张量    zeros = tf.zeros([2, 3])    print(f"零张量:\n{zeros}")    # 一张量    ones = tf.ones([3, 2])    print(f"一张量:\n{ones}")    # 单位矩阵    eye = tf.eye(3)    print(f"单位矩阵:\n{eye}")    # 随机张量    random_normal = tf.random.normal([2, 2], mean=0.0, stddev=1.0)    print(f"正态分布随机张量:\n{random_normal}")    random_uniform = tf.random.uniform([2, 2], minval=0, maxval=1)    print(f"均匀分布随机张量:\n{random_uniform}")    # 3. 张量属性    print("\n3. 张量属性:")    tensor = tf.constant([[1, 2, 3], [4, 5, 6]], dtype=tf.float32)    print(f"张量:\n{tensor}")    print(f"形状: {tensor.shape}")    print(f"数据类型: {tensor.dtype}")    print(f"维度数: {tensor.ndim}")    print(f"元素总数: {tf.size(tensor).numpy()}")    print(f"转换为NumPy:\n{tensor.numpy()}")    # 4. 张量操作    print("\n4. 张量操作:")    a = tf.constant([[1, 2], [3, 4]])    b = tf.constant([[5, 6], [7, 8]])    print(f"加法:\n{a + b}")    print(f"乘法:\n{a * b}")    print(f"矩阵乘法:\n{tf.matmul(a, b)}")    # 重塑    original = tf.constant([1, 2, 3, 4, 5, 6])    reshaped = tf.reshape(original, [2, 3])    print(f"重塑前: {original.shape}, 重塑后: {reshaped.shape}")    # 转置    transposed = tf.transpose(reshaped)    print(f"转置: {transposed.shape}")    # 5. 广播    print("\n5. 广播机制:")    x = tf.constant([1, 2, 3])    y = tf.constant([[10], [20], [30]])    print(f"x: {x.shape}, y: {y.shape}")    print(f"x + y:\n{x + y}")    return tensor# 运行张量基础tensor_basics()

def compare_execution_modes():    """比较Eager Execution和图模式"""    print("=" * 80)    print("TensorFlow执行模式比较")    print("=" * 80)    # Eager Execution示例    print("\n1. Eager Execution (即时执行模式):")    print("-" * 40)    eager_code = '''# TensorFlow 2.x默认启用Eager Executionimport tensorflow as tf# 即时计算，立即得到结果x = tf.constant([[1, 2], [3, 4]])y = tf.constant([[5, 6], [7, 8]])# 立即执行操作result = tf.matmul(x, y)print(f"结果:\\n{result}")print(f"立即得到结果，无需Session")# 可以与Python控制流无缝集成if tf.reduce_sum(x) > 5:    print("x的元素和大于5")else:    print("x的元素和小于等于5")    '''    print(eager_code)    # Graph Mode示例    print("\n2. Graph Mode (图模式):")    print("-" * 40)    graph_code = '''# 使用@tf.function装饰器将Python函数转换为计算图import tensorflow as tf@tf.functiondef compute(x, y):    # 这部分代码会被转换为计算图    z = tf.matmul(x, y)    return z# 第一次调用会构建计算图（tracing）x = tf.constant([[1, 2], [3, 4]])y = tf.constant([[5, 6], [7, 8]])result = compute(x, y)print(f"结果:\\n{result}")print("函数被编译为计算图，后续调用更快")# 查看计算图print(f"计算图签名: {compute.pretty_printed_concrete_signatures()}")    '''    print(graph_code)    # 性能对比    print("\n3. 性能对比:")    print("-" * 40)    performance_code = '''import tensorflow as tfimport time# 创建测试数据data = tf.random.normal([1000, 1000])# Eager Executionstart = time.time()for _ in range(10):    result = tf.matmul(data, data)eager_time = time.time() - start# Graph Mode@tf.functiondef compute_graph(data):    return tf.matmul(data, data)# 第一次调用会构建图（较慢）_ = compute_graph(data)start = time.time()for _ in range(10):    result = compute_graph(data)graph_time = time.time() - startprint(f"Eager Execution时间: {eager_time:.3f}秒")print(f"Graph Mode时间: {graph_time:.3f}秒")print(f"加速比: {eager_time/graph_time:.1f}x")# AutoGraph转换print("\\nAutoGraph可以自动转换Python控制流:")print("例如: if, for, while循环会被自动转换为图操作")    '''    print(performance_code)    # 使用建议    print("\n4. 使用建议:")    print("-" * 40)    recommendations = [        ("开发调试", "使用Eager Execution，便于调试和快速迭代"),        ("生产部署", "使用@tf.function将关键函数转换为图模式"),        ("性能关键", "对循环和复杂操作使用Graph Mode"),        ("自定义训练", "训练循环使用Graph Mode，单个步骤使用Eager"),        ("模型导出", "使用SavedModel格式，包含计算图")    ]    for scenario, advice in recommendations:        print(f"• {scenario}: {advice}")    return None# 运行执行模式比较compare_execution_modes()

def tf_data_pipeline():    """使用tf.data构建数据管道"""    print("=" * 80)    print("TensorFlow tf.data API - 高效数据管道")    print("=" * 80)    # 基本数据管道    print("\n1. 基本数据管道构建:")    basic_pipeline = '''import tensorflow as tfimport numpy as np# 创建模拟数据data = np.random.randn(1000, 32, 32, 3).astype(np.float32)labels = np.random.randint(0, 10, 1000)# 方法1: 从NumPy数组创建Datasetdataset = tf.data.Dataset.from_tensor_slices((data, labels))print(f"数据集元素类型: {dataset.element_spec}")# 方法2: 从生成器创建def data_generator():    for i in range(100):        yield (np.random.randn(32, 32, 3), np.random.randint(0, 10))dataset_gen = tf.data.Dataset.from_generator(    data_generator,    output_signature=(        tf.TensorSpec(shape=(32, 32, 3), dtype=tf.float32),        tf.TensorSpec(shape=(), dtype=tf.int32)    ))# 方法3: 从文件创建（CSV示例）# dataset_csv = tf.data.experimental.make_csv_dataset(#     'data.csv', batch_size=32, label_name='label'# )# 数据预处理dataset = dataset.shuffle(buffer_size=1000)  # 打乱数据dataset = dataset.batch(32)  # 批次化dataset = dataset.prefetch(tf.data.AUTOTUNE)  # 预取数据print(f"批次大小: 32")print(f"预取设置: AUTOTUNE (自动调整)")    '''    print(basic_pipeline)    # 高级数据增强    print("\n2. 高级数据增强:")    augmentation_code = '''def augment_images(image, label):    """图像数据增强"""    # 随机左右翻转    image = tf.image.random_flip_left_right(image)    # 随机亮度调整    image = tf.image.random_brightness(image, max_delta=0.2)    # 随机对比度调整    image = tf.image.random_contrast(image, lower=0.8, upper=1.2)    # 随机旋转（通过仿射变换）    angle = tf.random.uniform([], -0.2, 0.2)    image = tf.keras.preprocessing.image.apply_affine_transform(        image.numpy(), theta=angle*180/3.14159, row_axis=0, col_axis=1, channel_axis=2    )    image = tf.convert_to_tensor(image, dtype=tf.float32)    # 归一化到[0,1]    image = tf.clip_by_value(image, 0.0, 1.0)    return image, label# 应用增强dataset_augmented = dataset.map(    augment_images,    num_parallel_calls=tf.data.AUTOTUNE)print("应用了以下增强:")print("• 随机左右翻转")print("• 随机亮度调整")print("• 随机对比度调整")print("• 随机旋转")    '''    print(augmentation_code)    # 性能优化技巧    print("\n3. 性能优化技巧:")    optimization_code = '''def create_optimized_pipeline(data_path, batch_size=32, is_training=True):    """创建优化数据管道"""    # 1. 并行数据加载    dataset = tf.data.Dataset.list_files(data_path + "/*.tfrecord")    # 2. 并行文件读取    dataset = dataset.interleave(        tf.data.TFRecordDataset,        num_parallel_calls=tf.data.AUTOTUNE,        deterministic=False    )    # 3. 解析函数（示例）    def parse_tfrecord(example_proto):        features = {            'image': tf.io.FixedLenFeature([], tf.string),            'label': tf.io.FixedLenFeature([], tf.int64)        }        parsed = tf.io.parse_single_example(example_proto, features)        image = tf.io.decode_image(parsed['image'])        image = tf.cast(image, tf.float32) / 255.0        return image, parsed['label']    # 4. 并行解析    dataset = dataset.map(        parse_tfrecord,        num_parallel_calls=tf.data.AUTOTUNE    )    # 5. 缓存数据（如果内存足够）    dataset = dataset.cache()    if is_training:        # 6. 打乱数据        dataset = dataset.shuffle(buffer_size=1000)        # 7. 重复数据（用于多个epoch）        dataset = dataset.repeat()    # 8. 批次化    dataset = dataset.batch(batch_size)    # 9. 预取（最重要的优化）    dataset = dataset.prefetch(buffer_size=tf.data.AUTOTUNE)    return datasetprint("优化技巧:")print("1. interleave: 并行文件读取")print("2. map with num_parallel_calls: 并行处理")print("3. cache: 缓存数据")print("4. shuffle: 数据打乱")print("5. batch: 批次化")print("6. prefetch: 预取数据（最重要！）")print("7. AUTOTUNE: 自动调整并行度")    '''    print(optimization_code)    # 实际使用示例    print("\n4. 实际使用示例:")    practical_code = '''# 创建和训练模型的完整示例import tensorflow as tf# 1. 创建数据管道def create_mnist_pipeline(batch_size=64):    (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()    # 预处理函数    def preprocess(image, label):        image = tf.cast(image, tf.float32) / 255.0        image = tf.expand_dims(image, -1)  # 添加通道维度        label = tf.cast(label, tf.int32)        return image, label    # 训练集管道    train_ds = tf.data.Dataset.from_tensor_slices((x_train, y_train))    train_ds = train_ds.map(preprocess, num_parallel_calls=tf.data.AUTOTUNE)    train_ds = train_ds.shuffle(10000)    train_ds = train_ds.batch(batch_size)    train_ds = train_ds.prefetch(tf.data.AUTOTUNE)    # 测试集管道    test_ds = tf.data.Dataset.from_tensor_slices((x_test, y_test))    test_ds = test_ds.map(preprocess, num_parallel_calls=tf.data.AUTOTUNE)    test_ds = test_ds.batch(batch_size)    test_ds = test_ds.prefetch(tf.data.AUTOTUNE)    return train_ds, test_ds# 2. 创建模型def create_model():    model = tf.keras.Sequential([        tf.keras.layers.Input(shape=(28, 28, 1)),        tf.keras.layers.Conv2D(32, 3, activation='relu'),        tf.keras.layers.MaxPooling2D(),        tf.keras.layers.Conv2D(64, 3, activation='relu'),        tf.keras.layers.MaxPooling2D(),        tf.keras.layers.Flatten(),        tf.keras.layers.Dense(128, activation='relu'),        tf.keras.layers.Dropout(0.5),        tf.keras.layers.Dense(10, activation='softmax')    ])    model.compile(        optimizer='adam',        loss='sparse_categorical_crossentropy',        metrics=['accuracy']    )    return model# 3. 训练模型train_ds, test_ds = create_mnist_pipeline()model = create_model()print("开始训练...")history = model.fit(    train_ds,    validation_data=test_ds,    epochs=5,    verbose=1)print(f"测试准确率: {history.history['val_accuracy'][-1]:.3f}")    '''    print(practical_code)    return None# 运行tf.data API示例tf_data_pipeline()

def keras_model_building():    """使用Keras API构建深度学习模型"""    print("=" * 80)    print("TensorFlow Keras API - 模型构建")    print("=" * 80)    # Sequential API    print("\n1. Sequential API (顺序模型):")    sequential_code = '''import tensorflow as tf# 方法1: 逐层添加model = tf.keras.Sequential()model.add(tf.keras.layers.Input(shape=(784,)))model.add(tf.keras.layers.Dense(128, activation='relu'))model.add(tf.keras.layers.Dropout(0.2))model.add(tf.keras.layers.Dense(10, activation='softmax'))# 方法2: 列表初始化model = tf.keras.Sequential([    tf.keras.layers.Dense(128, activation='relu', input_shape=(784,)),    tf.keras.layers.Dropout(0.2),    tf.keras.layers.Dense(10, activation='softmax')])print("模型摘要:")model.summary()    '''    print(sequential_code)    # Functional API    print("\n2. Functional API (函数式API):")    functional_code = '''import tensorflow as tf# 定义输入inputs = tf.keras.Input(shape=(28, 28, 1))# 构建网络x = tf.keras.layers.Conv2D(32, 3, activation='relu')(inputs)x = tf.keras.layers.MaxPooling2D()(x)x = tf.keras.layers.Conv2D(64, 3, activation='relu')(x)x = tf.keras.layers.MaxPooling2D()(x)x = tf.keras.layers.Flatten()(x)x = tf.keras.layers.Dense(128, activation='relu')(x)x = tf.keras.layers.Dropout(0.5)(x)# 多个输出classification_output = tf.keras.layers.Dense(10, activation='softmax', name='classification')(x)regression_output = tf.keras.layers.Dense(1, name='regression')(x)# 创建模型model = tf.keras.Model(    inputs=inputs,    outputs=[classification_output, regression_output],    name='multi_output_model')print("函数式API模型摘要:")model.summary()# 绘制模型结构图tf.keras.utils.plot_model(model, 'model.png', show_shapes=True)print("模型结构图已保存为 'model.png'")    '''    print(functional_code)    # Model Subclassing    print("\n3. Model Subclassing (模型子类化):")    subclassing_code = '''import tensorflow as tfclass ResidualBlock(tf.keras.layers.Layer):    """残差块"""    def __init__(self, filters, kernel_size=3, stride=1, **kwargs):        super().__init__(**kwargs)        self.filters = filters        self.kernel_size = kernel_size        self.stride = stride        # 定义层        self.conv1 = tf.keras.layers.Conv2D(            filters, kernel_size, stride=stride, padding='same'        )        self.bn1 = tf.keras.layers.BatchNormalization()        self.conv2 = tf.keras.layers.Conv2D(            filters, kernel_size, padding='same'        )        self.bn2 = tf.keras.layers.BatchNormalization()        # 如果需要调整维度        if stride != 1:            self.shortcut = tf.keras.Sequential([                tf.keras.layers.Conv2D(filters, 1, stride=stride),                tf.keras.layers.BatchNormalization()            ])        else:            self.shortcut = tf.keras.layers.Lambda(lambda x: x)    def call(self, inputs, training=False):        # 残差路径        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 = tf.keras.layers.add([x, shortcut])        x = tf.nn.relu(x)        return x    def get_config(self):        config = super().get_config()        config.update({            'filters': self.filters,            'kernel_size': self.kernel_size,            'stride': self.stride        })        return configclass ResNetModel(tf.keras.Model):    """ResNet模型"""    def __init__(self, num_classes=10, **kwargs):        super().__init__(**kwargs)        # 输入层        self.input_layer = tf.keras.layers.InputLayer(input_shape=(32, 32, 3))        # 初始卷积层        self.conv_initial = tf.keras.layers.Conv2D(64, 7, strides=2, padding='same')        self.bn_initial = tf.keras.layers.BatchNormalization()        self.pool_initial = tf.keras.layers.MaxPooling2D(pool_size=3, strides=2, padding='same')        # 残差块        self.res_blocks = [            ResidualBlock(64, stride=1),            ResidualBlock(128, stride=2),            ResidualBlock(256, stride=2),            ResidualBlock(512, stride=2)        ]        # 全局平均池化        self.global_pool = tf.keras.layers.GlobalAveragePooling2D()        # 输出层        self.dense = tf.keras.layers.Dense(num_classes, activation='softmax')    def call(self, inputs, training=False):        x = self.input_layer(inputs)        x = self.conv_initial(x)        x = self.bn_initial(x, training=training)        x = tf.nn.relu(x)        x = self.pool_initial(x)        # 通过残差块        for block in self.res_blocks:            x = block(x, training=training)        x = self.global_pool(x)        return self.dense(x)# 创建模型model = ResNetModel(num_classes=10)model.build((None, 32, 32, 3))print("自定义ResNet模型摘要:")model.summary()    '''    print(subclassing_code)    # 层和激活函数    print("\n4. 常用层和激活函数:")    layers_code = '''import tensorflow as tf# 常用层示例layers_examples = {    '全连接层': tf.keras.layers.Dense(units=64, activation='relu'),    '卷积层': tf.keras.layers.Conv2D(filters=32, kernel_size=3, activation='relu'),    '循环层': tf.keras.layers.LSTM(units=64, return_sequences=True),    '批归一化': tf.keras.layers.BatchNormalization(),    'Dropout': tf.keras.layers.Dropout(rate=0.5),    '池化层': tf.keras.layers.MaxPooling2D(pool_size=2),    '嵌入层': tf.keras.layers.Embedding(input_dim=1000, output_dim=64),    '注意力层': tf.keras.layers.Attention(),    '展平层': tf.keras.layers.Flatten(),    '全局池化': tf.keras.layers.GlobalAveragePooling2D()}print("常用Keras层:")for name, layer in layers_examples.items():    print(f"  • {name}: {layer}")# 激活函数activations = {    'relu': tf.keras.activations.relu,    'sigmoid': tf.keras.activations.sigmoid,    'tanh': tf.keras.activations.tanh,    'softmax': tf.keras.activations.softmax,    'leaky_relu': tf.keras.layers.LeakyReLU(alpha=0.2),    'elu': tf.keras.activations.elu,    'selu': tf.keras.activations.selu,    'swish': tf.keras.activations.swish}print("\\n常用激活函数:")for name, func in activations.items():    print(f"  • {name}")    '''    print(layers_code)    # 模型编译和训练    print("\n5. 模型编译和训练:")    compile_code = '''import tensorflow as tf# 创建简单模型model = tf.keras.Sequential([    tf.keras.layers.Dense(64, activation='relu', input_shape=(784,)),    tf.keras.layers.Dropout(0.2),    tf.keras.layers.Dense(10, activation='softmax')])# 编译模型model.compile(    # 优化器    optimizer=tf.keras.optimizers.Adam(        learning_rate=0.001,        beta_1=0.9,        beta_2=0.999    ),    # 损失函数    loss=tf.keras.losses.SparseCategoricalCrossentropy(),    # 评估指标    metrics=[        'accuracy',        tf.keras.metrics.Precision(name='precision'),        tf.keras.metrics.Recall(name='recall'),        tf.keras.metrics.AUC(name='auc')    ])print("模型编译完成")print(f"优化器: {model.optimizer}")print(f"损失函数: {model.loss}")print(f"评估指标: {[m.name for m in model.metrics]}")# 创建回调函数callbacks = [    # 早停    tf.keras.callbacks.EarlyStopping(        monitor='val_loss',        patience=10,        restore_best_weights=True    ),    # 学习率调度    tf.keras.callbacks.ReduceLROnPlateau(        monitor='val_loss',        factor=0.5,        patience=5,        min_lr=1e-6    ),    # 模型检查点    tf.keras.callbacks.ModelCheckpoint(        'best_model.h5',        monitor='val_accuracy',        save_best_only=True    ),    # TensorBoard    tf.keras.callbacks.TensorBoard(        log_dir='./logs',        histogram_freq=1    )]print("\\n定义的回调函数:")for callback in callbacks:    print(f"  • {type(callback).__name__}")    '''    print(compile_code)    return None# 运行Keras API示例keras_model_building()

def custom_training_loop():    """自定义训练循环"""    print("=" * 80)    print("TensorFlow自定义训练循环")    print("=" * 80)    # 基础自定义训练    print("\n1. 基础自定义训练循环:")    basic_training_code = '''import tensorflow as tfimport numpy as np# 创建数据def create_dataset():    x = np.random.randn(1000, 10).astype(np.float32)    y = np.random.randint(0, 2, (1000, 1)).astype(np.float32)    dataset = tf.data.Dataset.from_tensor_slices((x, y))    dataset = dataset.shuffle(1000).batch(32)    return dataset# 创建模型class SimpleModel(tf.keras.Model):    def __init__(self):        super().__init__()        self.dense1 = tf.keras.layers.Dense(64, activation='relu')        self.dense2 = tf.keras.layers.Dense(32, activation='relu')        self.dense3 = tf.keras.layers.Dense(1, activation='sigmoid')    def call(self, inputs):        x = self.dense1(inputs)        x = self.dense2(x)        return self.dense3(x)# 初始化model = SimpleModel()loss_fn = tf.keras.losses.BinaryCrossentropy()optimizer = tf.keras.optimizers.Adam(learning_rate=0.001)# 训练和验证指标train_loss = tf.keras.metrics.Mean(name='train_loss')train_accuracy = tf.keras.metrics.BinaryAccuracy(name='train_accuracy')val_loss = tf.keras.metrics.Mean(name='val_loss')val_accuracy = tf.keras.metrics.BinaryAccuracy(name='val_accuracy')# 训练步骤@tf.functiondef train_step(x_batch, y_batch):    with tf.GradientTape() as tape:        # 前向传播        predictions = model(x_batch, training=True)        # 计算损失        loss = loss_fn(y_batch, predictions)    # 计算梯度    gradients = tape.gradient(loss, model.trainable_variables)    # 更新权重    optimizer.apply_gradients(zip(gradients, model.trainable_variables))    # 更新指标    train_loss.update_state(loss)    train_accuracy.update_state(y_batch, predictions)    return loss# 验证步骤@tf.functiondef val_step(x_batch, y_batch):    predictions = model(x_batch, training=False)    loss = loss_fn(y_batch, predictions)    val_loss.update_state(loss)    val_accuracy.update_state(y_batch, predictions)    return loss# 训练循环def train_epoch(dataset, epoch):    print(f"\\n开始第 {epoch+1} 轮训练")    # 重置指标    train_loss.reset_states()    train_accuracy.reset_states()    # 训练    for batch, (x_batch, y_batch) in enumerate(dataset):        loss = train_step(x_batch, y_batch)        if batch % 10 == 0:            print(f"  批次 {batch}: 损失 = {loss:.4f}")    # 打印训练结果    print(f"训练结果 - 损失: {train_loss.result():.4f}, 准确率: {train_accuracy.result():.4f}")# 验证循环def validate_epoch(dataset, epoch):    val_loss.reset_states()    val_accuracy.reset_states()    for x_batch, y_batch in dataset:        val_step(x_batch, y_batch)    print(f"验证结果 - 损失: {val_loss.result():.4f}, 准确率: {val_accuracy.result():.4f}")# 创建数据集train_dataset = create_dataset()val_dataset = create_dataset()# 训练多个epochepochs = 5for epoch in range(epochs):    train_epoch(train_dataset, epoch)    validate_epoch(val_dataset, epoch)    '''    print(basic_training_code)    # 高级训练技巧    print("\n2. 高级训练技巧:")    advanced_training_code = '''import tensorflow as tfclass AdvancedTrainingLoop:    """高级训练循环"""    def __init__(self, model, optimizer, loss_fn):        self.model = model        self.optimizer = optimizer        self.loss_fn = loss_fn        # 指标        self.metrics = {            'train': {                'loss': tf.keras.metrics.Mean(name='train_loss'),                'accuracy': tf.keras.metrics.SparseCategoricalAccuracy(name='train_accuracy')            },            'val': {                'loss': tf.keras.metrics.Mean(name='val_loss'),                'accuracy': tf.keras.metrics.SparseCategoricalAccuracy(name='val_accuracy')            }        }        # 学习率调度器        self.lr_scheduler = tf.keras.optimizers.schedules.ExponentialDecay(            initial_learning_rate=0.001,            decay_steps=1000,            decay_rate=0.96        )        # 梯度累积        self.gradient_accumulation_steps = 4        self.accumulated_gradients = None    def reset_metrics(self, phase='train'):        """重置指标"""        for metric in self.metrics[phase].values():            metric.reset_states()    @tf.function    def compute_gradients(self, x_batch, y_batch, training=True):        """计算梯度"""        with tf.GradientTape() as tape:            predictions = self.model(x_batch, training=training)            loss = self.loss_fn(y_batch, predictions)        return loss, predictions, tape.gradient(loss, self.model.trainable_variables)    def apply_gradients(self, gradients):        """应用梯度（支持梯度累积）"""        if self.accumulated_gradients is None:            self.accumulated_gradients = [tf.zeros_like(g) for g in gradients]        # 累积梯度        self.accumulated_gradients = [            acc_g + g for acc_g, g in zip(self.accumulated_gradients, gradients)        ]        # 如果达到累积步数，应用梯度        if tf.equal(tf.math.mod(self.optimizer.iterations, self.gradient_accumulation_steps), 0):            # 平均梯度            avg_gradients = [g / self.gradient_accumulation_steps for g in self.accumulated_gradients]            # 应用梯度            self.optimizer.apply_gradients(zip(avg_gradients, self.model.trainable_variables))            # 重置累积梯度            self.accumulated_gradients = None    @tf.function    def train_step(self, x_batch, y_batch):        """训练步骤"""        loss, predictions, gradients = self.compute_gradients(x_batch, y_batch, training=True)        # 应用梯度        self.apply_gradients(gradients)        # 更新指标        self.metrics['train']['loss'].update_state(loss)        self.metrics['train']['accuracy'].update_state(y_batch, predictions)        return loss    @tf.function    def val_step(self, x_batch, y_batch):        """验证步骤"""        loss, predictions, _ = self.compute_gradients(x_batch, y_batch, training=False)        # 更新指标        self.metrics['val']['loss'].update_state(loss)        self.metrics['val']['accuracy'].update_state(y_batch, predictions)        return loss    def train_epoch(self, dataset, epoch, verbose=True):        """训练一个epoch"""        self.reset_metrics('train')        for batch, (x_batch, y_batch) in enumerate(dataset):            loss = self.train_step(x_batch, y_batch)            if verbose and batch % 20 == 0:                print(f"  Epoch {epoch+1}, Batch {batch}: Loss = {loss:.4f}")        # 获取指标结果        results = {            'loss': self.metrics['train']['loss'].result().numpy(),            'accuracy': self.metrics['train']['accuracy'].result().numpy()        }        return results    def validate_epoch(self, dataset, epoch):        """验证一个epoch"""        self.reset_metrics('val')        for x_batch, y_batch in dataset:            self.val_step(x_batch, y_batch)        # 获取指标结果        results = {            'loss': self.metrics['val']['loss'].result().numpy(),            'accuracy': self.metrics['val']['accuracy'].result().numpy()        }        return results# 使用示例print("高级训练循环特性:")print("• 梯度累积（支持大batch size）")print("• 学习率调度")print("• 详细的指标跟踪")print("• @tf.function优化性能")    '''    print(advanced_training_code)    # 分布式训练    print("\n3. 分布式训练:")    distributed_code = '''import tensorflow as tfimport numpy as npdef setup_distributed_training(strategy_type='mirrored'):    """设置分布式训练"""    strategies = {        'mirrored': tf.distribute.MirroredStrategy(),  # 单机多GPU        'multi_worker': tf.distribute.MultiWorkerMirroredStrategy(),  # 多机多GPU        'tpu': tf.distribute.TPUStrategy(),  # TPU训练        'parameter_server': tf.distribute.ParameterServerStrategy()  # 参数服务器    }    if strategy_type not in strategies:        print(f"警告: 策略 {strategy_type} 不存在，使用默认策略")        strategy = tf.distribute.MirroredStrategy()    else:        strategy = strategies[strategy_type]    print(f"使用分布式策略: {strategy_type}")    print(f"设备数量: {strategy.num_replicas_in_sync}")    return strategydef create_distributed_model(strategy):    """在策略范围内创建模型"""    with strategy.scope():        # 创建模型        model = tf.keras.Sequential([            tf.keras.layers.Dense(128, activation='relu', input_shape=(784,)),            tf.keras.layers.Dropout(0.2),            tf.keras.layers.Dense(64, activation='relu'),            tf.keras.layers.Dropout(0.2),            tf.keras.layers.Dense(10, activation='softmax')        ])        # 编译模型        model.compile(            optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),            loss=tf.keras.losses.SparseCategoricalCrossentropy(),            metrics=['accuracy']        )    return modeldef create_distributed_dataset(strategy, batch_size_per_replica=32):    """创建分布式数据集"""    # 计算全局batch size    global_batch_size = batch_size_per_replica * strategy.num_replicas_in_sync    # 创建数据集    (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()    x_train = x_train.reshape(-1, 784).astype('float32') / 255.0    x_test = x_test.reshape(-1, 784).astype('float32') / 255.0    # 创建tf.data.Dataset    train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))    train_dataset = train_dataset.shuffle(60000).batch(global_batch_size)    test_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))    test_dataset = test_dataset.batch(global_batch_size)    # 分布式数据集    train_dist_dataset = strategy.experimental_distribute_dataset(train_dataset)    test_dist_dataset = strategy.experimental_distribute_dataset(test_dataset)    return train_dist_dataset, test_dist_dataset# 分布式训练示例print("分布式训练示例:")print("1. 设置分布式策略")print("2. 在策略范围内创建模型")print("3. 创建分布式数据集")print("4. 正常训练（框架自动处理分布式）")    '''    print(distributed_code)    return None# 运行自定义训练循环示例custom_training_loop()

def model_saving_deployment():    """模型保存与部署"""    print("=" * 80)    print("TensorFlow模型保存与部署")    print("=" * 80)    # 模型保存格式    print("\n1. 模型保存格式:")    saving_formats = '''import tensorflow as tfimport numpy as np# 创建并训练一个简单模型def create_and_train_model():    model = tf.keras.Sequential([        tf.keras.layers.Dense(64, activation='relu', input_shape=(10,)),        tf.keras.layers.Dense(32, activation='relu'),        tf.keras.layers.Dense(1, activation='sigmoid')    ])    model.compile(        optimizer='adam',        loss='binary_crossentropy',        metrics=['accuracy']    )    # 生成虚拟数据    x = np.random.randn(100, 10).astype(np.float32)    y = np.random.randint(0, 2, (100, 1)).astype(np.float32)    # 训练    model.fit(x, y, epochs=1, verbose=0)    return modelmodel = create_and_train_model()# 1. SavedModel格式（推荐）print("1. SavedModel格式保存:")model.save('my_model')  # 保存整个模型# 加载SavedModelloaded_model = tf.keras.models.load_model('my_model')print(f"  加载成功: {type(loaded_model)}")# 2. HDF5格式print("\\n2. HDF5格式保存:")model.save('my_model.h5')  # 保存为HDF5文件# 加载HDF5h5_model = tf.keras.models.load_model('my_model.h5')print(f"  加载成功: {type(h5_model)}")# 3. 仅保存权重print("\\n3. 仅保存权重:")model.save_weights('model_weights.h5')# 创建新模型并加载权重new_model = tf.keras.Sequential([    tf.keras.layers.Dense(64, activation='relu', input_shape=(10,)),    tf.keras.layers.Dense(32, activation='relu'),    tf.keras.layers.Dense(1, activation='sigmoid')])new_model.compile(optimizer='adam', loss='binary_crossentropy')new_model.load_weights('model_weights.h5')print(f"  权重加载成功")# 4. Checkpoint格式（训练中保存）print("\\n4. Checkpoint格式:")checkpoint_path = "training/cp-{epoch:04d}.ckpt"checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(    filepath=checkpoint_path,    save_weights_only=True,    verbose=1)print("  检查点回调已创建")    '''    print(saving_formats)    # 模型转换与优化    print("\n2. 模型转换与优化:")    conversion_code = '''import tensorflow as tfdef optimize_and_convert_model(model):    """模型优化与转换"""    print("模型优化与转换流程:")    # 1. 创建示例输入    @tf.function    def serving_fn(inputs):        return model(inputs)    # 获取具体函数（用于优化）    concrete_func = serving_fn.get_concrete_function(        tf.TensorSpec(shape=[None, 10], dtype=tf.float32, name='inputs')    )    # 2. 优化模型（修剪、量化等）    print("1. 模型优化:")    # 修剪（减少模型大小）    pruning_params = {        'pruning_schedule': tf.keras.optimizers.schedules.PolynomialDecay(            initial_sparsity=0.0,            final_sparsity=0.5,            begin_step=0,            end_step=1000        )    }    print(f"  修剪参数: {pruning_params}")    # 3. 转换为TensorFlow Lite（移动端）    print("\\n2. 转换为TensorFlow Lite:")    # 创建转换器    converter = tf.lite.TFLiteConverter.from_concrete_functions([concrete_func])    # 优化选项    converter.optimizations = [tf.lite.Optimize.DEFAULT]    # 量化（减少模型大小，加快推理）    converter.target_spec.supported_types = [tf.float16]  # FP16量化    # 转换为TFLite模型    tflite_model = converter.convert()    # 保存TFLite模型    with open('model.tflite', 'wb') as f:        f.write(tflite_model)    print(f"  TFLite模型已保存: model.tflite")    print(f"  模型大小: {len(tflite_model) / 1024:.1f} KB")    # 4. 转换为ONNX格式（跨框架）    print("\\n3. 转换为ONNX格式:")    print("  使用 tf2onnx 工具:")    print("  python -m tf2onnx.convert --saved-model my_model --output model.onnx")    return tflite_model# 注意：实际转换需要安装相应依赖print("模型转换工具:")print("• TensorFlow Lite Converter: 移动端部署")print("• tf2onnx: 转换为ONNX格式")print("• TensorFlow.js Converter: 网页部署")print("• TensorFlow Serving: 服务器端部署")    '''    print(conversion_code)    # TensorFlow Serving部署    print("\n3. TensorFlow Serving部署:")    serving_code = '''import tensorflow as tfimport numpy as npdef prepare_model_for_serving(model, export_path='serving_model'):    """准备模型用于Serving"""    # 1. 保存为SavedModel格式    tf.saved_model.save(model, export_path)    print(f"模型已保存到: {export_path}")    # 2. 创建签名（定义输入输出）    class ExportModule(tf.Module):        def __init__(self, model):            super().__init__()            self.model = model        @tf.function(input_signature=[            tf.TensorSpec(shape=[None, 10], dtype=tf.float32)        ])        def predict(self, inputs):            return {"predictions": self.model(inputs)}    # 3. 导出模型    module = ExportModule(model)    # 保存带签名的模型    tf.saved_model.save(        module,        export_path + '_signed',        signatures={            'serving_default': module.predict        }    )    print(f"带签名的模型已保存到: {export_path}_signed")    return export_path + '_signed'def test_serving_model(model_path):    """测试Serving模型"""    # 加载模型    loaded = tf.saved_model.load(model_path)    # 获取推理函数    infer = loaded.signatures['serving_default']    # 准备测试数据    test_input = tf.constant(np.random.randn(5, 10).astype(np.float32))    # 推理    predictions = infer(test_input)    print(f"测试输入形状: {test_input.shape}")    print(f"预测结果形状: {predictions['predictions'].shape}")    print(f"预测值: {predictions['predictions'].numpy()}")    return infer# Docker部署命令docker_commands = '''# 1. 拉取TensorFlow Serving镜像docker pull tensorflow/serving# 2. 运行Serving容器docker run -p 8501:8501 \\  --mount type=bind,source=/path/to/serving_model,target=/models/model \\  -e MODEL_NAME=model \\  -t tensorflow/serving# 3. REST API调用curl -d '{"instances": [[0.1, 0.2, ..., 1.0]]}' \\  -X POST http://localhost:8501/v1/models/model:predict# 4. gRPC客户端# 使用tensorflow-serving-api包'''print("TensorFlow Serving部署步骤:")print("1. 保存模型为SavedModel格式")print("2. 使用Docker运行TensorFlow Serving")print("3. 通过REST API或gRPC调用模型")print("\\nDocker命令:")print(docker_commands)    '''    print(serving_code)    return None# 运行模型保存与部署示例model_saving_deployment()

def image_classification_project():    """完整的图像分类项目"""    print("=" * 80)    print("TensorFlow完整图像分类项目")    print("=" * 80)    project_code = '''import tensorflow as tfimport numpy as npimport matplotlib.pyplot as pltimport os# 1. 数据准备def prepare_data():    """准备CIFAR-10数据集"""    print("准备CIFAR-10数据集...")    # 加载数据    (x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()    # 类别名称    class_names = ['airplane', 'automobile', 'bird', 'cat', 'deer',                   'dog', 'frog', 'horse', 'ship', 'truck']    print(f"训练集: {x_train.shape}, {y_train.shape}")    print(f"测试集: {x_test.shape}, {y_test.shape}")    # 归一化    x_train = x_train.astype('float32') / 255.0    x_test = x_test.astype('float32') / 255.0    # 将标签转换为一维    y_train = y_train.reshape(-1)    y_test = y_test.reshape(-1)    return (x_train, y_train), (x_test, y_test), class_names# 2. 数据增强def create_data_augmentation():    """创建数据增强层"""    data_augmentation = tf.keras.Sequential([        tf.keras.layers.RandomFlip("horizontal"),        tf.keras.layers.RandomRotation(0.1),        tf.keras.layers.RandomZoom(0.1),        tf.keras.layers.RandomContrast(0.1),    ])    return data_augmentation# 3. 创建模型def create_model(num_classes=10):    """创建CNN模型"""    # 数据增强    data_augmentation = create_data_augmentation()    # 模型构建    inputs = tf.keras.Input(shape=(32, 32, 3))    # 数据增强    x = data_augmentation(inputs)    # 特征提取    x = tf.keras.layers.Conv2D(32, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.Conv2D(32, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.MaxPooling2D()(x)    x = tf.keras.layers.Dropout(0.2)(x)    x = tf.keras.layers.Conv2D(64, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.Conv2D(64, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.MaxPooling2D()(x)    x = tf.keras.layers.Dropout(0.3)(x)    x = tf.keras.layers.Conv2D(128, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.Conv2D(128, 3, padding='same', activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.MaxPooling2D()(x)    x = tf.keras.layers.Dropout(0.4)(x)    # 分类头    x = tf.keras.layers.Flatten()(x)    x = tf.keras.layers.Dense(128, activation='relu')(x)    x = tf.keras.layers.BatchNormalization()(x)    x = tf.keras.layers.Dropout(0.5)(x)    outputs = tf.keras.layers.Dense(num_classes, activation='softmax')(x)    # 创建模型    model = tf.keras.Model(inputs=inputs, outputs=outputs)    return model# 4. 编译模型def compile_model(model, learning_rate=0.001):    """编译模型"""    optimizer = tf.keras.optimizers.Adam(        learning_rate=learning_rate,        beta_1=0.9,        beta_2=0.999,        epsilon=1e-07    )    model.compile(        optimizer=optimizer,        loss=tf.keras.losses.SparseCategoricalCrossentropy(),        metrics=['accuracy',                 tf.keras.metrics.SparseTopKCategoricalAccuracy(k=3, name='top3_accuracy')]    )    return model# 5. 创建回调def create_callbacks():    """创建训练回调"""    callbacks = [        # 早停        tf.keras.callbacks.EarlyStopping(            monitor='val_loss',            patience=15,            restore_best_weights=True,            verbose=1        ),        # 学习率调度        tf.keras.callbacks.ReduceLROnPlateau(            monitor='val_loss',            factor=0.5,            patience=5,            min_lr=1e-6,            verbose=1        ),        # 模型检查点        tf.keras.callbacks.ModelCheckpoint(            filepath='best_model.h5',            monitor='val_accuracy',            save_best_only=True,            verbose=1        ),        # TensorBoard        tf.keras.callbacks.TensorBoard(            log_dir='./logs',            histogram_freq=1,            write_graph=True,            write_images=True        ),        # CSV日志        tf.keras.callbacks.CSVLogger(            filename='training_log.csv',            separator=',',            append=False        )    ]    return callbacks# 6. 创建数据管道def create_data_pipeline(x_train, y_train, x_test, y_test, batch_size=64):    """创建数据管道"""    # 训练数据    train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))    train_dataset = train_dataset.shuffle(buffer_size=10000)    train_dataset = train_dataset.batch(batch_size)    train_dataset = train_dataset.prefetch(tf.data.AUTOTUNE)    # 验证数据    val_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))    val_dataset = val_dataset.batch(batch_size)    val_dataset = val_dataset.prefetch(tf.data.AUTOTUNE)    return train_dataset, val_dataset# 7. 可视化结果def visualize_results(history, x_test, y_test, model, class_names):    """可视化训练结果"""    # 创建图形    fig, axes = plt.subplots(2, 3, figsize=(15, 10))    # 1. 训练损失和准确率    axes[0, 0].plot(history.history['loss'], label='训练损失')    axes[0, 0].plot(history.history['val_loss'], label='验证损失')    axes[0, 0].set_xlabel('Epoch')    axes[0, 0].set_ylabel('损失')    axes[0, 0].set_title('训练和验证损失')    axes[0, 0].legend()    axes[0, 0].grid(True, alpha=0.3)    axes[0, 1].plot(history.history['accuracy'], label='训练准确率')    axes[0, 1].plot(history.history['val_accuracy'], label='验证准确率')    axes[0, 1].set_xlabel('Epoch')    axes[0, 1].set_ylabel('准确率')    axes[0, 1].set_title('训练和验证准确率')    axes[0, 1].legend()    axes[0, 1].grid(True, alpha=0.3)    # 2. Top-3准确率    if 'top3_accuracy' in history.history:        axes[0, 2].plot(history.history['top3_accuracy'], label='训练Top-3')        axes[0, 2].plot(history.history['val_top3_accuracy'], label='验证Top-3')        axes[0, 2].set_xlabel('Epoch')        axes[0, 2].set_ylabel('Top-3准确率')        axes[0, 2].set_title('Top-3准确率')        axes[0, 2].legend()        axes[0, 2].grid(True, alpha=0.3)    # 3. 学习率    axes[1, 0].plot(history.history.get('lr', [0.001]*len(history.history['loss'])))    axes[1, 0].set_xlabel('Epoch')    axes[1, 0].set_ylabel('学习率')    axes[1, 0].set_title('学习率变化')    axes[1, 0].grid(True, alpha=0.3)    # 4. 混淆矩阵（简化）    from sklearn.metrics import confusion_matrix    import seaborn as sns    # 预测测试集    y_pred = model.predict(x_test, verbose=0)    y_pred_classes = np.argmax(y_pred, axis=1)    # 计算混淆矩阵    cm = confusion_matrix(y_test, y_pred_classes)    # 绘制混淆矩阵    axes[1, 1].imshow(cm, interpolation='nearest', cmap=plt.cm.Blues)    axes[1, 1].set_title('混淆矩阵')    axes[1, 1].set_xlabel('预测标签')    axes[1, 1].set_ylabel('真实标签')    # 5. 示例预测    axes[1, 2].axis('off')    axes[1, 2].text(0.1, 0.9, f'测试准确率: {history.history["val_accuracy"][-1]:.3f}',                     fontsize=12, transform=axes[1, 2].transAxes)    axes[1, 2].text(0.1, 0.8, f'测试损失: {history.history["val_loss"][-1]:.3f}',                     fontsize=12, transform=axes[1, 2].transAxes)    plt.suptitle('CIFAR-10图像分类结果', fontsize=16, fontweight='bold')    plt.tight_layout()    plt.savefig('training_results.png', dpi=150, bbox_inches='tight')    plt.show()    return y_pred_classes# 8. 主函数def main():    """主函数"""    print("开始CIFAR-10图像分类项目")    print("=" * 50)    # 准备数据    (x_train, y_train), (x_test, y_test), class_names = prepare_data()    # 创建数据管道    batch_size = 64    train_dataset, val_dataset = create_data_pipeline(        x_train, y_train, x_test, y_test, batch_size    )    # 创建模型    model = create_model(num_classes=10)    model = compile_model(model, learning_rate=0.001)    # 打印模型摘要    print("\\n模型结构:")    model.summary()    # 创建回调    callbacks = create_callbacks()    # 训练模型    print("\\n开始训练模型...")    epochs = 50    history = model.fit(        train_dataset,        validation_data=val_dataset,        epochs=epochs,        callbacks=callbacks,        verbose=1    )    # 评估模型    print("\\n评估模型...")    test_loss, test_accuracy, test_top3 = model.evaluate(val_dataset, verbose=0)    print(f"测试损失: {test_loss:.4f}")    print(f"测试准确率: {test_accuracy:.4f}")    print(f"测试Top-3准确率: {test_top3:.4f}")    # 可视化结果    print("\\n可视化结果...")    y_pred_classes = visualize_results(history, x_test, y_test, model, class_names)    # 保存模型    print("\\n保存模型...")    model.save('cifar10_model.h5')    print("模型已保存为 'cifar10_model.h5'")    # 保存为SavedModel格式（用于部署）    model.save('cifar10_model_savedmodel', save_format='tf')    print("模型已保存为SavedModel格式: 'cifar10_model_savedmodel'")    print("\\n项目完成！")    return model, history# 运行项目if __name__ == "__main__":    # 注意：实际运行需要较长时间，这里提供代码框架    print("项目代码框架已生成")    print("要运行完整项目，请执行 main() 函数")    '''    print(project_code)    return None# 运行图像分类项目image_classification_project()

def tensorflow_debugging():    """TensorFlow调试工具与技巧"""    print("=" * 80)    print("TensorFlow调试与优化")    print("=" * 80)    # TensorBoard集成    print("\n1. TensorBoard集成:")    tensorboard_code = '''import tensorflow as tfimport datetime# 1. 设置TensorBoard回调log_dir = "logs/fit/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")tensorboard_callback = tf.keras.callbacks.TensorBoard(    log_dir=log_dir,    histogram_freq=1,           # 每多少epoch记录一次直方图    write_graph=True,           # 是否记录计算图    write_images=True,          # 是否记录模型权重图像    update_freq='epoch',        # 'batch'或'epoch'    profile_batch=(10, 20)      # 分析批次范围)# 2. 在训练中使用# model.fit(..., callbacks=[tensorboard_callback])# 3. 启动TensorBoard命令# tensorboard --logdir logs/fitprint("TensorBoard设置:")print(f"日志目录: {log_dir}")print("启动命令: tensorboard --logdir logs/fit")# 4. 自定义标量记录file_writer = tf.summary.create_file_writer(log_dir)@tf.functiondef 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))    # 记录自定义标量    with file_writer.as_default():        tf.summary.scalar('training_loss', loss, step=optimizer.iterations)    return lossprint("\\n自定义指标记录已设置")    '''    print(tensorboard_code)    # 模型分析工具    print("\n2. 模型分析工具:")    profiling_code = '''import tensorflow as tfimport numpy as npdef analyze_model_performance(model, input_shape=(1, 32, 32, 3)):    """分析模型性能"""    # 1. 模型摘要    print("模型摘要:")    model.summary()    # 2. 计算FLOPs（浮点运算次数）    try:        from tensorflow.python.profiler import model_analyzer        from tensorflow.python.profiler import option_builder        # 创建分析配置        profile_opts = option_builder.ProfileOptionBuilder.float_operation()        # 获取FLOPs        flops = model_analyzer.profile(            tf.compat.v1.get_default_graph(),            options=profile_opts        )        print(f"\\n模型FLOPs: {flops.total_float_ops:,}")    except:        print("\\nFLOPs计算需要tensorflow.python.profiler")    # 3. 推理时间分析    print("\\n推理时间分析:")    # 创建测试输入    test_input = tf.random.normal(input_shape)    # 预热    _ = model.predict(test_input, verbose=0)    # 测量推理时间    import time    num_runs = 100    times = []    for _ in range(num_runs):        start = time.time()        _ = model.predict(test_input, verbose=0)        times.append(time.time() - start)    avg_time = np.mean(times) * 1000  # 转换为毫秒    print(f"平均推理时间: {avg_time:.2f} ms")    print(f"FPS: {1000/avg_time:.1f}")    # 4. 内存分析    print("\\n内存分析:")    print(f"可训练参数: {model.count_params():,}")    # 5. 层分析    print("\\n层分析:")    for layer in model.layers:        print(f"{layer.name}: {layer.output_shape} | 参数: {layer.count_params():,}")    return avg_time# 使用示例print("模型分析功能:")print("• 模型摘要和参数统计")print("• FLOPs计算")print("• 推理时间测量")print("• 内存使用分析")    '''    print(profiling_code)    # 常见错误与解决方案    print("\n3. 常见错误与解决方案:")    debugging_tips = '''常见TensorFlow错误及解决方案:1. 形状不匹配错误   错误: "Shapes (x, y) and (a, b) are incompatible"   原因: 层输入输出形状不匹配   解决: 检查模型各层形状，使用 model.summary() 调试2. GPU内存不足   错误: "OOM when allocating tensor"   原因: 批量太大或模型太复杂   解决:      - 减小 batch_size     - 使用混合精度训练: tf.keras.mixed_precision.set_global_policy('mixed_float16')     - 启用内存增长: tf.config.experimental.set_memory_growth(gpu, True)3. 梯度消失/爆炸   现象: 损失变为NaN或非常大   解决:     - 使用梯度裁剪: optimizer = tf.keras.optimizers.Adam(clipvalue=1.0)     - 添加批归一化层     - 使用更稳定的激活函数（如ReLU代替sigmoid）4. 过拟合   现象: 训练准确率高，验证准确率低   解决:     - 增加 Dropout 层     - 添加 L1/L2 正则化     - 使用数据增强     - 使用早停（EarlyStopping）5. 训练速度慢   解决:     - 使用 @tf.function 装饰关键函数     - 启用 XLA 编译: tf.config.optimizer.set_jit(True)     - 使用 tf.data API 并启用 prefetch     - 使用混合精度训练6. 模型不收敛   解决:     - 检查学习率（尝试更小的值）     - 检查数据预处理（是否归一化）     - 检查损失函数是否适合任务     - 检查模型是否足够复杂调试工具:   • tf.debugging.enable_check_numerics(): 检查NaN/Inf   • tf.config.run_functions_eagerly(True): 强制Eager模式调试   • tf.print(): 在图模式中打印张量   • pdb/ipdb: Python调试器性能优化检查表:   □ 使用 @tf.function 装饰训练循环   □ 启用 tf.data.Dataset.prefetch()   □ 使用混合精度训练   □ 批处理输入数据   □ 使用 GPU 并正确配置   □ 启用 XLA 编译    '''    print(debugging_tips)    return None# 运行调试工具tensorflow_debugging()