一天一个Python知识点——Day 154:Keras基础

import matplotlib.pyplot as pltdef keras_philosophy():    """Keras核心设计哲学可视化"""    principles = {        '用户友好': '为人类设计，而非为机器',        '模块化': '神经网络是层的组合游戏',        '易扩展': '新类即新模块，自由生长',        '渐进式': '从三行代码到复杂科研，平滑过渡'    }    fig, ax = plt.subplots(figsize=(10, 6))    ax.axis('off')    y_pos = 0.8    for principle, desc in principles.items():        ax.text(0.1, y_pos, f'• {principle}', fontsize=14, fontweight='bold',               transform=ax.transAxes, color='#E44D2E')        ax.text(0.25, y_pos-0.03, desc, fontsize=12,                transform=ax.transAxes, color='#555555')        y_pos -= 0.15    ax.text(0.5, 0.1, '“从想法到结果的最快路径”',            fontsize=16, style='italic', ha='center',           transform=ax.transAxes, color='#2E86C1')    plt.suptitle('Keras设计哲学', fontsize=18, fontweight='bold')    plt.show()keras_philosophy()

def keras_installation_guide():    """Keras安装与验证指南"""    guide = '''━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━【Keras安装三步法】—— 永远通过TensorFlow使用Keras━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━✅ 第一步：安装TensorFlow（Keras自动包含）────────────────────────────────────────   pip install tensorflow   # 或GPU版本   pip install tensorflow[and-cuda]✅ 第二步：验证安装────────────────────────────────────────   import tensorflow as tf   from tensorflow import keras   from tensorflow.keras import layers   print(f"TensorFlow版本: {tf.__version__}")   print(f"Keras版本: {keras.__version__}")   print(f"GPU是否可用: {tf.config.list_physical_devices('GPU')}")✅ 第三步：快速健康检查────────────────────────────────────────   # 创建一个极简模型测试环境   model = keras.Sequential([layers.Dense(10, input_shape=(5,))])   model.compile(optimizer='adam', loss='mse')   print("✓ Keras环境就绪！")━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━⚠️ 注意：请勿单独安装独立的keras包(pip install keras)   在TensorFlow 2.x中，应始终使用 tf.keras━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━    '''    print(guide)keras_installation_guide()

import tensorflow as tffrom tensorflow.keras import layers, modelsdef sequential_demo():    """Sequential API：层的有序堆叠"""    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    print("【Sequential API】—— 适合：线性堆叠、快速原型")    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    # 方式一：逐层add（适合动态添加）    model_a = models.Sequential()    model_a.add(layers.Dense(64, activation='relu', input_shape=(784,)))    model_a.add(layers.Dropout(0.2))    model_a.add(layers.Dense(10, activation='softmax'))    # 方式二：列表初始化（结构一目了然）    model_b = models.Sequential([        layers.Conv2D(32, (3,3), activation='relu', input_shape=(28,28,1)),        layers.MaxPooling2D((2,2)),        layers.Conv2D(64, (3,3), activation='relu'),        layers.MaxPooling2D((2,2)),        layers.Flatten(),        layers.Dense(128, activation='relu'),        layers.Dropout(0.5),        layers.Dense(10, activation='softmax')    ])    model_b.summary()    return model_b

def functional_demo():    """Functional API：多输入、多输出、非结构拓扑"""    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    print("【Functional API】—— 适合：复杂拓扑、分支、多模态")    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    # 典型场景1：双输入模型（文本+图像融合分类）    from tensorflow.keras import Input    # 定义两个独立的输入分支    text_input = Input(shape=(100,), name='text_input')    image_input = Input(shape=(32,32,3), name='image_input')    # 文本处理分支    text_features = layers.Dense(64, activation='relu')(text_input)    # 图像处理分支    x = layers.Conv2D(32, (3,3), activation='relu')(image_input)    x = layers.GlobalAveragePooling2D()(x)    image_features = layers.Dense(64, activation='relu')(x)    # 特征融合    combined = layers.Concatenate()([text_features, image_features])    # 输出层    output = layers.Dense(1, activation='sigmoid', name='output')(combined)    # 定义模型（指定输入和输出）    multi_input_model = models.Model(        inputs=[text_input, image_input],        outputs=output    )    print("✓ 多输入模型构建完成")    print(f"输入: {multi_input_model.input_names}")    print(f"输出: {multi_input_model.output_names}")    # 典型场景2：残差连接（跳层连接）    print("\n【残差连接示例】")    input_tensor = Input(shape=(32,32,3))    # 主路径    x = layers.Conv2D(32, (3,3), padding='same')(input_tensor)    x = layers.BatchNormalization()(x)    x = layers.ReLU()(x)    x = layers.Conv2D(32, (3,3), padding='same')(x)    x = layers.BatchNormalization()(x)    # 残差连接（恒等映射）    residual = layers.Conv2D(32, (1,1), padding='same')(input_tensor)    output_tensor = layers.add([x, residual])    output_tensor = layers.ReLU()(output_tensor)    resnet_block = models.Model(input_tensor, output_tensor)    print("✓ 残差块构建完成（支持任意层数的跳连）")    return multi_input_model# 执行演示functional_demo()

def subclassing_demo():    """Model Subclassing：完全自定义的前向逻辑"""    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    print("【Model Subclassing】—— 适合：科研创新、控制流、动态架构")    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    class DynamicRouteNetwork(tf.keras.Model):        """根据输入特征动态选择网络路径"""        def __init__(self, num_classes=10):            super().__init__()            # 定义所有可能的模块            self.conv_path = tf.keras.Sequential([                layers.Conv2D(32, 3, activation='relu'),                layers.GlobalAveragePooling2D()            ])            self.dense_path = tf.keras.Sequential([                layers.Dense(128, activation='relu'),                layers.Dense(64, activation='relu')            ])            self.classifier = layers.Dense(num_classes, activation='softmax')            # 可训练的路由权重            self.route_weight = self.add_weight(                name='route_weight',                shape=(1,),                initializer='zeros',                trainable=True            )        def call(self, inputs, training=None):            # 动态决策：根据输入尺寸选择路径            if len(inputs.shape) == 4:  # 4D张量 → 图像                features = self.conv_path(inputs)            else:  # 2D张量 → 向量                features = self.dense_path(inputs)            # 可根据训练状态调整行为            if training:                # 训练时添加随机噪声增强鲁棒性                features = features + tf.random.normal(tf.shape(features)) * 0.1            return self.classifier(features)    model = DynamicRouteNetwork()    # 测试动态路由    img_input = tf.random.normal((1, 32, 32, 3))    vec_input = tf.random.normal((1, 784))    print(f"图像输入 → 输出形状: {model(img_input).shape}")    print(f"向量输入 → 输出形状: {model(vec_input).shape}")    print(f"可训练变量数: {len(model.trainable_variables)}")    return model# 执行演示subclassing_demo()

如果你不确定用哪个 → 用Functional如果你只需要堆叠 → 用Sequential如果你在做科研创新 → 用Subclassing

import tensorflow as tffrom tensorflow.keras import layers, models, datasets, callbacksimport numpy as npimport matplotlib.pyplot as pltclass KerasCompletePipeline:    """Keras完整工作流演示"""    def __init__(self):        self.model = None        self.history = None    def step1_prepare_data(self):        """步骤1：数据加载与增强"""        print("【步骤1】准备CIFAR-10数据集...")        (x_train, y_train), (x_test, y_test) = datasets.cifar10.load_data()        # 归一化        x_train = x_train.astype('float32') / 255.0        x_test = x_test.astype('float32') / 255.0        # 将标签转换为一维        y_train = y_train.squeeze()        y_test = y_test.squeeze()        # 数据增强（Keras预处理层）        data_augmentation = tf.keras.Sequential([            layers.RandomFlip("horizontal"),            layers.RandomRotation(0.1),            layers.RandomZoom(0.1),            layers.RandomContrast(0.1),        ])        # 构建高效数据管道        train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))        train_dataset = train_dataset.shuffle(10000)        train_dataset = train_dataset.batch(64)        train_dataset = train_dataset.map(            lambda x, y: (data_augmentation(x, training=True), y),            num_parallel_calls=tf.data.AUTOTUNE        )        train_dataset = train_dataset.prefetch(tf.data.AUTOTUNE)        test_dataset = tf.data.Dataset.from_tensor_slices((x_test, y_test))        test_dataset = test_dataset.batch(64).prefetch(tf.data.AUTOTUNE)        print(f"训练集: {x_train.shape}, 测试集: {x_test.shape}")        print(f"数据增强已启用: Flip, Rotation, Zoom, Contrast")        return train_dataset, test_dataset, (x_test, y_test)    def step2_build_model(self):        """步骤2：模型构建（Functional API）"""        print("\n【步骤2】构建卷积神经网络...")        # 输入层        inputs = layers.Input(shape=(32, 32, 3))        # 卷积基        x = layers.Conv2D(32, (3,3), padding='same')(inputs)        x = layers.BatchNormalization()(x)        x = layers.ReLU()(x)        x = layers.Conv2D(32, (3,3), padding='same')(x)        x = layers.BatchNormalization()(x)        x = layers.ReLU()(x)        x = layers.MaxPooling2D((2,2))(x)        x = layers.Dropout(0.2)(x)        x = layers.Conv2D(64, (3,3), padding='same')(x)        x = layers.BatchNormalization()(x)        x = layers.ReLU()(x)        x = layers.Conv2D(64, (3,3), padding='same')(x)        x = layers.BatchNormalization()(x)        x = layers.ReLU()(x)        x = layers.MaxPooling2D((2,2))(x)        x = layers.Dropout(0.3)(x)        x = layers.Conv2D(128, (3,3), padding='same')(x)        x = layers.BatchNormalization()(x)        x = layers.ReLU()(x)        x = layers.GlobalAveragePooling2D()(x)        x = layers.Dropout(0.4)(x)        # 分类头        x = layers.Dense(128, activation='relu')(x)        x = layers.Dropout(0.5)(x)        outputs = layers.Dense(10, activation='softmax')(x)        model = models.Model(inputs, outputs)        # 编译模型        model.compile(            optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),            loss='sparse_categorical_crossentropy',            metrics=['accuracy',                     tf.keras.metrics.TopKCategoricalAccuracy(k=3, name='top3_acc')]        )        self.model = model        model.summary()        return model    def step3_train(self, train_dataset, val_dataset):        """步骤3：训练与回调"""        print("\n【步骤3】开始训练...")        # 回调函数配置        callbacks_list = [            # 早停（防止过拟合）            callbacks.EarlyStopping(                monitor='val_loss',                patience=10,                restore_best_weights=True,                verbose=1            ),            # 学习率衰减            callbacks.ReduceLROnPlateau(                monitor='val_loss',                factor=0.5,                patience=5,                min_lr=1e-6,                verbose=1            ),            # 模型检查点            callbacks.ModelCheckpoint(                'cifar10_best.h5',                monitor='val_accuracy',                save_best_only=True,                verbose=1            ),            # TensorBoard日志            callbacks.TensorBoard(                log_dir='./logs',                histogram_freq=1,                write_graph=True            ),            # 自定义回调：打印学习率            callbacks.LambdaCallback(                on_epoch_end=lambda epoch, logs:                 print(f"LR: {self.model.optimizer.lr.numpy():.2e}")            )        ]        history = self.model.fit(            train_dataset,            validation_data=val_dataset,            epochs=50,            callbacks=callbacks_list,            verbose=1        )        self.history = history        return history    def step4_evaluate(self, test_data):        """步骤4：评估与可视化"""        print("\n【步骤4】模型评估...")        x_test, y_test = test_data        test_loss, test_acc, test_top3 = self.model.evaluate(x_test, y_test, verbose=0)        print(f"测试集准确率: {test_acc:.4f}")        print(f"测试集Top-3准确率: {test_top3:.4f}")        # 可视化训练历史        self.plot_history()        # 可视化预测结果        self.plot_predictions(x_test, y_test)        return test_acc    def plot_history(self):        """绘制训练曲线"""        fig, axes = plt.subplots(1, 2, figsize=(12, 4))        # 损失曲线        axes[0].plot(self.history.history['loss'], label='训练损失')        axes[0].plot(self.history.history['val_loss'], label='验证损失')        axes[0].set_xlabel('Epoch')        axes[0].set_ylabel('损失')        axes[0].set_title('训练/验证损失')        axes[0].legend()        axes[0].grid(True, alpha=0.3)        # 准确率曲线        axes[1].plot(self.history.history['accuracy'], label='训练准确率')        axes[1].plot(self.history.history['val_accuracy'], label='验证准确率')        axes[1].set_xlabel('Epoch')        axes[1].set_ylabel('准确率')        axes[1].set_title('训练/验证准确率')        axes[1].legend()        axes[1].grid(True, alpha=0.3)        plt.suptitle('CIFAR-10训练历史', fontsize=14, fontweight='bold')        plt.tight_layout()        plt.show()    def plot_predictions(self, x_test, y_test):        """可视化预测结果"""        class_names = ['飞机', '汽车', '鸟', '猫', '鹿',                      '狗', '青蛙', '马', '船', '卡车']        # 随机选择9张测试图片        indices = np.random.choice(len(x_test), 9, replace=False)        fig, axes = plt.subplots(3, 3, figsize=(9, 9))        for i, ax in enumerate(axes.flat):            idx = indices[i]            img = x_test[idx]            true_label = y_test[idx]            # 预测            pred = self.model.predict(img[np.newaxis, ...], verbose=0)            pred_label = np.argmax(pred)            confidence = np.max(pred)            # 显示图像            ax.imshow(img)            color = 'green' if pred_label == true_label else 'red'            ax.set_title(f'真实:{class_names[true_label]}\n预测:{class_names[pred_label]}({confidence:.2f})',                        color=color, fontsize=10)            ax.axis('off')        plt.suptitle('CIFAR-10预测结果可视化', fontsize=14, fontweight='bold')        plt.tight_layout()        plt.show()    def step5_export(self):        """步骤5：模型导出与部署准备"""        print("\n【步骤5】模型导出...")        # 1. 保存完整模型        self.model.save('cifar10_final.h5')        print("✓ 模型保存为 H5 格式")        # 2. 导出为SavedModel（推荐生产部署）        self.model.export('cifar10_saved_model')        print("✓ 模型导出为 SavedModel 格式")        # 3. 转换为TensorFlow Lite        converter = tf.lite.TFLiteConverter.from_keras_model(self.model)        converter.optimizations = [tf.lite.Optimize.DEFAULT]        tflite_model = converter.convert()        with open('cifar10_model.tflite', 'wb') as f:            f.write(tflite_model)        print(f"✓ TensorFlow Lite模型已导出 (大小: {len(tflite_model)/1024:.1f} KB)")        return True# ===== 执行完整流程 =====print("=" * 60)print("Keras端到端实战：CIFAR-10图像分类")print("=" * 60)pipeline = KerasCompletePipeline()# 流水线执行train_ds, test_ds, raw_test = pipeline.step1_prepare_data()model = pipeline.step2_build_model()history = pipeline.step3_train(train_ds, test_ds)accuracy = pipeline.step4_evaluate(raw_test)pipeline.step5_export()print("\n✅ Keras完整工作流演示完成！")print(f"最佳验证准确率: {max(history.history['val_accuracy']):.4f}")

def pretrained_demo():    """Keras预训练模型：迁移学习实战"""    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    print("【Keras Applications】—— 30秒拥有SOTA模型")    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")    from tensorflow.keras import applications    # 1. 加载预训练模型（不包含分类头）    base_model = applications.ResNet50(        weights='imagenet',        include_top=False,  # 移除分类层        input_shape=(224, 224, 3),        pooling='avg'       # 全局平均池化    )    # 2. 冻结基干（不参与训练）    base_model.trainable = False    # 3. 添加新的分类头    model = tf.keras.Sequential([        base_model,        layers.Dense(256, activation='relu'),        layers.Dropout(0.5),        layers.Dense(10, activation='softmax')  # 新数据集10类    ])    print(f"基干模型: ResNet50 (ImageNet预训练)")    print(f"基干参数量: {base_model.count_params():,} (已冻结)")    print(f"新增头参数量: {model.layers[-1].count_params():,}")    # 4. 解冻部分层进行微调    print("\n【微调阶段】")    base_model.trainable = True    # 冻结前100层，只微调高层特征    for layer in base_model.layers[:100]:        layer.trainable = False    trainable_count = sum([1 for layer in base_model.layers if layer.trainable])    print(f"解冻层数: {trainable_count}/{len(base_model.layers)}")    return model# 执行演示pretrained_demo()

def custom_callbacks_demo():    """Keras回调系统：训练时的智能干预"""    from tensorflow.keras import callbacks    class CustomCallback(tf.keras.callbacks.Callback):        """自定义回调：监控梯度范数、学习率自适应"""        def __init__(self, patience=3):            super().__init__()            self.patience = patience            self.best_weights = None            self.wait = 0        def on_epoch_begin(self, epoch, logs=None):            # epoch开始时触发            print(f"\n▶️ Epoch {epoch+1} 开始")        def on_batch_end(self, batch, logs=None):            # 每批次结束时监控梯度            if batch % 50 == 0:                # 计算梯度范数                grad_norm = 0                for weight in self.model.trainable_weights:                    if weight._grad is not None:                        grad_norm += tf.norm(weight._grad).numpy()                logs = logs or {}                logs['grad_norm'] = grad_norm                print(f"  批次 {batch}: 损失={logs.get('loss'):.4f}, 梯度范数={grad_norm:.2f}")        def on_epoch_end(self, epoch, logs=None):            # 自动保存最佳模型            current_acc = logs.get('val_accuracy')            if current_acc and current_acc > self.best_acc:                self.best_acc = current_acc                self.best_weights = self.model.get_weights()                self.wait = 0                print(f"⭐ 保存最佳模型，准确率: {current_acc:.4f}")            else:                self.wait += 1    # 官方回调全家桶    standard_callbacks = {        '早停': callbacks.EarlyStopping(monitor='val_loss', patience=5),        '学习率衰减': callbacks.ReduceLROnPlateau(monitor='val_loss', factor=0.5),        '模型检查点': callbacks.ModelCheckpoint('model.h5', save_best_only=True),        'TensorBoard': callbacks.TensorBoard(log_dir='./logs'),        '学习率调度': callbacks.LearningRateScheduler(            lambda epoch: 0.001 * 0.95**epoch        ),        'CSV日志': callbacks.CSVLogger('training.log'),        '终止NaN': callbacks.TerminateOnNaN(),        'Lambda回调': callbacks.LambdaCallback(            on_epoch_end=lambda epoch, logs: print(f"Epoch {epoch} done")        )    }    print("Keras回调系统——训练时的智能传感器")    print("=" * 50)    for name, cb in standard_callbacks.items():        print(f"  • {name}: {cb.__class__.__name__}")    return standard_callbackscustom_callbacks_demo()

def ecosystem_positioning():    """Keras vs TensorFlow vs PyTorch：共存而非替代"""    positioning = '''━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━【深度学习框架生态定位】—— 工具是为目标服务的━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━🧱 Keras —— 高级API层（抽象程度最高）   ├─ 设计目标：开发者体验、快速迭代、低认知负荷   ├─ 核心价值：3行代码构建神经网络，20+预训练模型   └─ 典型用户：初学者、数据科学家、产品经理⚙️ TensorFlow —— 工业级引擎（抽象程度中等）   ├─ 设计目标：生产部署、端到端生态、跨平台   ├─ 核心价值：TF Serving, TF Lite, TF.js, XLA编译   └─ 典型用户：ML工程师、后端开发者、企业团队🔧 PyTorch —— 科研手术刀（抽象程度较低）   ├─ 设计目标：研究灵活性、动态图、Pythonic   ├─ 核心价值：易于调试、控制流原生、学术界首选   └─ 典型用户：研究员、博士生、算法创新者━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━【选型决策树】问题1：你希望3小时内跑通第一个模型？  →  Keras问题2：你正在做毕设/发论文？          →  PyTorch问题3：你要部署到手机/浏览器？        →  TensorFlow问题4：你需要同时满足研发和生产？     →  Keras建型 + TF部署━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━    '''    print(positioning)ecosystem_positioning()