基于迁移学习的分心驾驶员检测CNN模型优化及自定义预处理问询

问题背景 #

任务目标是用TensorFlow Keras的EfficientNetB0（或其他预训练模型）实现State Farm分心驾驶员检测，数据集包含约22.5k张训练图像，均匀划分为10个子类。

初始迁移学习实现的核心代码如下：

IMG_HEIGHT, IMG_WIDTH = 224, 224

# 加载预训练EfficientNetB0，去掉顶层分类层
base_model = EfficientNetB0(weights="imagenet", include_top=False, input_shape=(IMG_HEIGHT, IMG_WIDTH, 3))

# 冻结预训练层
base_model.trainable = False

# 添加自定义分类头
x = base_model.output
x = GlobalAveragePooling2D()(x)  # 全局池化将特征图转为向量
x = Dropout(0.5)(x)  # Dropout防止过拟合
x = Dense(256, activation='relu')(x)  # 全连接层
x = Dropout(0.5)(x)
outputs = Dense(train_generator.num_classes, activation='softmax')(x)

# 定义模型
model = Model(inputs=base_model.input, outputs=outputs)

# 编译模型
model.compile(optimizer=Adam(learning_rate=0.0005), 
              loss="categorical_crossentropy", 
              metrics=["accuracy"])

# 定义回调
early_stopping = tf.keras.callbacks.EarlyStopping(monitor='accuracy', patience=5, restore_best_weights=True)
reduce_lr = tf.keras.callbacks.ReduceLROnPlateau(monitor='accuracy', factor=0.2, patience=3, min_lr=1e-6)

# 训练模型
history = model.fit(
    train_generator,
    epochs=100,
    callbacks=[early_stopping, reduce_lr]
)

配套的训练数据生成器设置为：

train_dir = 'state-farm-distracted-driver-detection\\imgs\\train\\'

train_datagen = ImageDataGenerator(
    rescale=1.0 / 255.0,  # 归一化像素到[0,1]
    rotation_range=20,    # 随机旋转±20度
    width_shift_range=0.2,  # 随机水平偏移20%宽度
    height_shift_range=0.2,  # 随机垂直偏移20%高度
    shear_range=0.2,  # 随机剪切变换
    zoom_range=0.2,   # 随机缩放±20%
    horizontal_flip=True,  # 随机水平翻转
    fill_mode="nearest"  # 填充变换后空白像素
)

train_generator = train_datagen.flow_from_directory(
    train_dir,
    target_size = (224, 224),
    class_mode = 'categorical',
    batch_size = 128
)

但初始实现的训练准确率仅为0.1082（回调触发前）。此前使用包含3组CNN+MaxPooling层、后续接Flatten和Dense层的简单CNN时，训练准确率达0.8031，但测试集表现极差，因此尝试用迁移学习提升准确率和泛化能力。

优化进展与效果 #

更换为EfficientNet官方提供的预处理函数后，模型效果大幅提升：

from tensorflow.keras.applications.efficientnet import preprocess_input
train_datagen = ImageDataGenerator(preprocessing_function=preprocess_input)

测试集数据生成器也采用相同设置。经微调后，模型训练准确率达0.9737，验证准确率达0.83909，泛化能力显著改善。

疑问解答 #

1. 预处理方式导致效果差异的核心原因 #

预训练模型（如EfficientNetB0）是在ImageNet数据集上用特定预处理逻辑训练出来的，并非简单的rescale=1/255归一化：

• EfficientNet的preprocess_input会将像素值归一化到[-1, 1]区间（不同预训练模型的预处理规则可能不同，部分是减去均值后除以标准差），和初始实现的[0,1]归一化完全不匹配。
• 预训练模型的特征提取层是基于匹配该预处理的图像训练的，输入不符合规则时，提取的特征会偏离模型预期的分布，导致后续分类层无法有效学习，这是初始准确率极低的核心原因。
• 初始实现中添加的大量数据增强（旋转、平移等）在冻结预训练层阶段，可能进一步破坏预训练特征的有效性，也是效果差的辅助因素。

2. 能否自定义ImageDataGenerator函数或传入自定义参数 #

完全可以，有三种常见实现方式：

方式一：自定义预处理函数，组合官方预处理与自定义逻辑

可以编写自定义函数，将官方预处理和额外的增强/处理逻辑结合，传给ImageDataGenerator的preprocessing_function参数：

def custom_preprocess(img):
    # 先执行EfficientNet的标准预处理
    img = preprocess_input(img)
    # 添加自定义处理，比如随机调整亮度
    img = tf.keras.layers.RandomBrightness(factor=0.2)(img)
    return img

train_datagen = ImageDataGenerator(preprocessing_function=custom_preprocess)

方式二：同时使用官方预处理与内置增强参数

如果需要保留ImageDataGenerator的内置增强功能，可以同时指定preprocessing_function和增强参数：

train_datagen = ImageDataGenerator(
    preprocessing_function=preprocess_input,
    rotation_range=15,
    width_shift_range=0.15,
    horizontal_flip=True
)

注意：微调预训练模型（解冻部分层）时，适度增强能提升泛化性，但过度增强会破坏预训练特征，需控制增强幅度。

方式三：完全自定义数据生成器

如果ImageDataGenerator的内置功能无法满足需求，可以继承tf.keras.utils.Sequence类，实现完全自定义的加载、增强、预处理流程：

import numpy as np
import tensorflow as tf
import os

class CustomDataGenerator(tf.keras.utils.Sequence):
    def __init__(self, img_dir, target_size=(224,224), batch_size=32, shuffle=True):
        # 加载图像路径和对应标签（适配数据集结构）
        self.img_paths = []
        self.labels = []
        for class_idx, class_name in enumerate(sorted(os.listdir(img_dir))):
            class_dir = os.path.join(img_dir, class_name)
            if not os.path.isdir(class_dir):
                continue
            for img_name in os.listdir(class_dir):
                self.img_paths.append(os.path.join(class_dir, img_name))
                self.labels.append(class_idx)
        self.target_size = target_size
        self.batch_size = batch_size
        self.shuffle = shuffle
        self.on_epoch_end()

    def __len__(self):
        # 返回每个epoch的批次数
        return int(np.ceil(len(self.img_paths) / self.batch_size))

    def __getitem__(self, index):
        # 生成单个批次的数据
        batch_start = index * self.batch_size
        batch_end = min(batch_start + self.batch_size, len(self.img_paths))
        batch_paths = self.img_paths[batch_start:batch_end]
        batch_labels = self.labels[batch_start:batch_end]

        batch_imgs = []
        for path, label in zip(batch_paths, batch_labels):
            # 加载图像
            img = tf.keras.preprocessing.image.load_img(path, target_size=self.target_size)
            img = tf.keras.preprocessing.image.img_to_array(img)
            # 标准预处理
            img = preprocess_input(img)
            # 自定义增强（仅训练阶段）
            if self.shuffle:
                img = tf.keras.layers.RandomFlip("horizontal")(img)
                img = tf.keras.layers.RandomRotation(factor=0.15)(img)
            batch_imgs.append(img)
        
        return np.array(batch_imgs), tf.keras.utils.to_categorical(batch_labels, num_classes=10)

    def on_epoch_end(self):
        # 每个epoch结束后打乱数据
        if self.shuffle:
            idx = np.random.permutation(len(self.img_paths))
            self.img_paths = [self.img_paths[i] for i in idx]
            self.labels = [self.labels[i] for i in idx]

内容的提问来源于stack exchange，提问作者Ishan Singh