TensorFlow训练CNN时遭遇未知图像格式错误求助

问题详情 #

在Google Colab中构建CNN模型时，调用model_1.fit()触发以下错误：

Epoch 1/5
---------------------------------------------------------------------------
InvalidArgumentError                      Traceback (most recent call last)
<ipython-input-29-a98bc2c91ee1> in <module>
----> 1 history = model_1.fit(train_data, epochs=5, steps_per_epoch=len(train_data), validation_data=test_data, validation_steps=int(0.25 * len(test_data)))

1 frames
/usr/local/lib/python3.7/dist-packages/tensorflow/python/eager/execute.py in quick_execute(op_name, num_outputs, inputs, attrs, ctx, name)
     53     ctx.ensure_initialized()
     54     tensors = pywrap_tfe.TFE_Py_Execute(ctx._handle, device_name, op_name,
---&gt; 55                                         inputs, attrs, num_outputs)
     56   except core._NotOkStatusException as e:
     57     if name is not None:

InvalidArgumentError: Graph execution error:

2 root error(s) found.
  (0) INVALID_ARGUMENT:  Unknown image file format. One of JPEG, PNG, GIF, BMP required.
     [[{{node decode_image/DecodeImage}}]]
     [[IteratorGetNext]]
     [[categorical_crossentropy/softmax_cross_entropy_with_logits/Shape_2/_10]]
  (1) INVALID_ARGUMENT:  Unknown image file format. One of JPEG, PNG, GIF, BMP required.
     [[{{node decode_image/DecodeImage}}]]
     [[IteratorGetNext]]
0 successful operations.
0 derived errors ignored. [Op:__inference_train_function_31356]

已确认数据集所有图片后缀为JPEG，参考过相关Stack Overflow回答但问题未解决，数据集来自谷歌图片搜索。

实现代码 #

数据加载部分

import tensorflow as tf

# 创建训练和测试目录路径
train_dir = 'Dataset/train'
test_dir = 'Dataset/test'
IMG_SIZE = (224,224)
BATCH_SIZE=32

# 设置数据加载器
train_data = tf.keras.preprocessing.image_dataset_from_directory(directory=train_dir,
                                                                            image_size=IMG_SIZE,
                                                                            label_mode='categorical',
                                                                            batch_size=BATCH_SIZE)

test_data = tf.keras.preprocessing.image_dataset_from_directory(directory=test_dir, image_size=IMG_SIZE, batch_size=BATCH_SIZE, label_mode='categorical')

数据增强部分

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
from tensorflow.keras.layers.experimental import preprocessing

data_augmentation = keras.Sequential([
        preprocessing.RandomFlip('horizontal'),
        preprocessing.RandomRotation(0.2),
        preprocessing.RandomZoom(0.2),
        preprocessing.RandomHeight(0.2),
        preprocessing.RandomWidth(0.2),
        # preprocessing.Rescale(1/255.) 保留这个用于ResNet，EfficientNet内置归一化
], name='data_augmentation')

模型构建与训练部分

input_shape = (224,224,3)
base_model = tf.keras.applications.EfficientNetB0(include_top=False)
base_model.trainable=False

# 创建输入层
inputs = layers.Input(shape=input_shape, name='input_layer')
x=data_augmentation(inputs)

# 将增强后的输入传入预训练模型（不训练）
x = base_model(x,training=False)

x = layers.GlobalAveragePooling2D()(x)

# 添加输出层
outputs = layers.Dense(9, activation='softmax', name='output_layer')(x)

# 构建模型
model_1 = keras.Model(inputs,outputs)

# 编译模型
model_1.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])

# 训练模型
history = model_1.fit(train_data, epochs=5, steps_per_epoch=len(train_data), validation_data=test_data, validation_steps=int(0.25 * len(test_data)))

排查与解决步骤 #

1. 验证图片实际格式与完整性 #

有些文件后缀是.jpg，但实际可能是损坏的JPEG、WebP等其他格式，或文件本身已损坏。执行以下代码批量检查：

import os
from PIL import Image

def check_image_format(folder_path):
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            if file.lower().endswith(('.jpg', '.jpeg')):
                file_path = os.path.join(root, file)
                try:
                    with Image.open(file_path) as img:
                        actual_format = img.format
                        if actual_format != 'JPEG':
                            print(f"文件 {file_path} 后缀为JPEG，但实际格式是 {actual_format}")
                except Exception as e:
                    print(f"文件 {file_path} 损坏或无法打开: {str(e)}")

# 检查训练集和测试集
check_image_format(train_dir)
check_image_format(test_dir)

对于输出中标记的问题文件，直接删除或替换为有效JPEG图片。

2. 给数据加载器添加错误跳过机制 #

修改数据加载逻辑，捕获加载错误并跳过无效样本：

def load_image_and_label(file_path, label):
    try:
        img = tf.io.read_file(file_path)
        img = tf.image.decode_jpeg(img, channels=3)
        img = tf.image.resize(img, IMG_SIZE)
        return img, label
    except:
        # 返回无效标签，后续过滤
        return tf.zeros(IMG_SIZE + (3,)), tf.constant(-1, dtype=tf.int32)

# 重新构建训练集
train_dataset = tf.keras.utils.image_dataset_from_directory(
    train_dir,
    image_size=IMG_SIZE,
    label_mode='int',
    batch_size=BATCH_SIZE,
    shuffle=False
)

# 获取文件路径和标签
train_file_paths = train_dataset.file_paths
train_labels = []
for _, labels in train_dataset:
    train_labels.extend(labels.numpy())
train_labels = tf.convert_to_tensor(train_labels)

# 创建新数据集并过滤无效样本
train_data = tf.data.Dataset.from_tensor_slices((train_file_paths, train_labels))
train_data = train_data.map(load_image_and_label, num_parallel_calls=tf.data.AUTOTUNE)
train_data = train_data.filter(lambda x, y: tf.not_equal(y, -1))
# 转换为categorical标签
train_data = train_data.map(lambda x, y: (x, tf.one_hot(y, depth=9)))
train_data = train_data.shuffle(1000).batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)

# 测试集执行相同操作
test_dataset = tf.keras.utils.image_dataset_from_directory(
    test_dir,
    image_size=IMG_SIZE,
    label_mode='int',
    batch_size=BATCH_SIZE,
    shuffle=False
)
test_file_paths = test_dataset.file_paths
test_labels = []
for _, labels in test_dataset:
    test_labels.extend(labels.numpy())
test_labels = tf.convert_to_tensor(test_labels)
test_data = tf.data.Dataset.from_tensor_slices((test_file_paths, test_labels))
test_data = test_data.map(load_image_and_label, num_parallel_calls=tf.data.AUTOTUNE)
test_data = test_data.filter(lambda x, y: tf.not_equal(y, -1))
test_data = test_data.map(lambda x, y: (x, tf.one_hot(y, depth=9)))
test_data = test_data.batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)

3. 删除数据集内的非图片文件 #

数据集可能包含隐藏文件（如.DS_Store）或下载附带的其他文件，这些会被误加载。执行以下代码清理：

def remove_non_image_files(folder_path):
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            if not file.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.bmp')):
                file_path = os.path.join(root, file)
                os.remove(file_path)
                print(f"已删除非图片文件: {file_path}")

remove_non_image_files(train_dir)
remove_non_image_files(test_dir)

4. 手动加载数据集（备选方案） #

如果上述方法仍无效，直接用PIL手动加载图片，完全控制加载流程：

import numpy as np
from PIL import Image

def load_dataset(folder_path, img_size, num_classes):
    images = []
    labels = []
    class_names = sorted(os.listdir(folder_path))
    class_indices = {name: idx for idx, name in enumerate(class_names)}
    
    for class_name in class_names:
        class_dir = os.path.join(folder_path, class_name)
        for file in os.listdir(class_dir):
            file_path = os.path.join(class_dir, file)
            try:
                img = Image.open(file_path).convert('RGB')
                img = img.resize(img_size)
                img = np.array(img)
                images.append(img)
                labels.append(class_indices[class_name])
            except Exception as e:
                print(f"跳过无效文件 {file_path}: {str(e)}")
    
    images = np.array(images)
    labels = tf.keras.utils.to_categorical(labels, num_classes=num_classes)
    return tf.data.Dataset.from_tensor_slices((images, labels)).batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)

# 加载数据集
train_data = load_dataset(train_dir, IMG_SIZE, 9)
test_data = load_dataset(test_dir, IMG_SIZE, 9)

内容的提问来源于stack exchange，提问作者Aayanam Anirudh