在 CIFAR10 数据集上训练 ResNet。
ResNet v1: Deep Residual Learning for Image Recognition
ResNet v2: Identity Mappings in Deep Residual Networks
| Model | n | 200-epoch accuracy | Original paper accuracy | sec/epoch GTX1080Ti |
|---|---|---|---|---|
| ResNet20 v1 | 3 | 92.16 % | 91.25 % | 35 |
| ResNet32 v1 | 5 | 92.46 % | 92.49 % | 50 |
| ResNet44 v1 | 7 | 92.50 % | 92.83 % | 70 |
| ResNet56 v1 | 9 | 92.71 % | 93.03 % | 90 |
| ResNet110 v1 | 18 | 92.65 % | 93.39+-.16 % | 165 |
| ResNet164 v1 | 27 | - % | 94.07 % | - |
| ResNet1001 v1 | N/A | - % | 92.39 % | - |
| Model | n | 200-epoch accuracy | Original paper accuracy | sec/epoch GTX1080Ti |
|---|---|---|---|---|
| ResNet20 v2 | 2 | - % | - % | --- |
| ResNet32 v2 | N/A | NA % | NA % | NA |
| ResNet44 v2 | N/A | NA % | NA % | NA |
| ResNet56 v2 | 6 | 93.01 % | NA % | 100 |
| ResNet110 v2 | 12 | 93.15 % | 93.63 % | 180 |
| ResNet164 v2 | 18 | - % | 94.54 % | - |
| ResNet1001 v2 | 111 | - % | 95.08+-.14 % | - |
from __future__ import print_function
import keras
from keras.layers import Dense, Conv2D, BatchNormalization, Activation
from keras.layers import AveragePooling2D, Input, Flatten
from keras.optimizers import Adam
from keras.callbacks import ModelCheckpoint, LearningRateScheduler
from keras.callbacks import ReduceLROnPlateau
from keras.preprocessing.image import ImageDataGenerator
from keras.regularizers import l2
from keras import backend as K
from keras.models import Model
from keras.datasets import cifar10
import numpy as np
import os
# 训练参数
batch_size = 32 # 原论文按照 batch_size=128 训练所有的网络
epochs = 200
data_augmentation = True
num_classes = 10
# 减去像素均值可提高准确度
subtract_pixel_mean = True
# 模型参数
# ----------------------------------------------------------------------------
# | | 200-epoch | Orig Paper| 200-epoch | Orig Paper| sec/epoch
# Model | n | ResNet v1 | ResNet v1 | ResNet v2 | ResNet v2 | GTX1080Ti
# |v1(v2)| %Accuracy | %Accuracy | %Accuracy | %Accuracy | v1 (v2)
# ----------------------------------------------------------------------------
# ResNet20 | 3 (2)| 92.16 | 91.25 | ----- | ----- | 35 (---)
# ResNet32 | 5(NA)| 92.46 | 92.49 | NA | NA | 50 ( NA)
# ResNet44 | 7(NA)| 92.50 | 92.83 | NA | NA | 70 ( NA)
# ResNet56 | 9 (6)| 92.71 | 93.03 | 93.01 | NA | 90 (100)
# ResNet110 |18(12)| 92.65 | 93.39+-.16| 93.15 | 93.63 | 165(180)
# ResNet164 |27(18)| ----- | 94.07 | ----- | 94.54 | ---(---)
# ResNet1001| (111)| ----- | 92.39 | ----- | 95.08+-.14| ---(---)
# ---------------------------------------------------------------------------
n = 3
# 模型版本
# Orig paper: version = 1 (ResNet v1), Improved ResNet: version = 2 (ResNet v2)
version = 1
# 从提供的模型参数 n 计算的深度
if version == 1:
depth = n * 6 + 2
elif version == 2:
depth = n * 9 + 2
# 模型名称、深度和版本
model_type = 'ResNet%dv%d' % (depth, version)
# 载入 CIFAR10 数据。
(x_train, y_train), (x_test, y_test) = cifar10.load_data()
# 输入图像维度。
input_shape = x_train.shape[1:]
# 数据标准化。
x_train = x_train.astype('float32') / 255
x_test = x_test.astype('float32') / 255
# 如果使用减去像素均值
if subtract_pixel_mean:
x_train_mean = np.mean(x_train, axis=0)
x_train -= x_train_mean
x_test -= x_train_mean
print('x_train shape:', x_train.shape)
print(x_train.shape[0], 'train samples')
print(x_test.shape[0], 'test samples')
print('y_train shape:', y_train.shape)
# 将类向量转换为二进制类矩阵。
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)
def lr_schedule(epoch):
"""学习率调度
学习率将在 80, 120, 160, 180 轮后依次下降。
他作为训练期间回调的一部分，在每个时期自动调用。
# 参数
epoch (int): 轮次
# 返回
lr (float32): 学习率
"""
lr = 1e-3
if epoch > 180:
lr *= 0.5e-3
elif epoch > 160:
lr *= 1e-3
elif epoch > 120:
lr *= 1e-2
elif epoch > 80:
lr *= 1e-1
print('Learning rate: ', lr)
return lr
def resnet_layer(inputs,
num_filters=16,
kernel_size=3,
strides=1,
activation='relu',
batch_normalization=True,
conv_first=True):
"""2D 卷积批量标准化 - 激活栈构建器
# 参数
inputs (tensor): 从输入图像或前一层来的输入张量
num_filters (int): Conv2D 过滤器数量
kernel_size (int): Conv2D 方形核维度
strides (int): Conv2D 方形步幅维度
activation (string): 激活函数名
batch_normalization (bool): 是否包含批标准化
conv_first (bool): conv-bn-activation (True) 或
bn-activation-conv (False)
# 返回
x (tensor): 作为下一层输入的张量
"""
conv = Conv2D(num_filters,
kernel_size=kernel_size,
strides=strides,
padding='same',
kernel_initializer='he_normal',
kernel_regularizer=l2(1e-4))
x = inputs
if conv_first:
x = conv(x)
if batch_normalization:
x = BatchNormalization()(x)
if activation is not None:
x = Activation(activation)(x)
else:
if batch_normalization:
x = BatchNormalization()(x)
if activation is not None:
x = Activation(activation)(x)
x = conv(x)
return x
def resnet_v1(input_shape, depth, num_classes=10):
"""ResNet 版本 1 模型构建器 [a]
2 x (3 x 3) Conv2D-BN-ReLU 的堆栈
最后一个 ReLU 在快捷连接之后。
在每个阶段的开始，特征图大小由具有 strides=2 的卷积层减半（下采样），
而滤波器的数量加倍。在每个阶段中，这些层具有相同数量的过滤器和相同的特征图尺寸。
特征图尺寸:
stage 0: 32x32, 16
stage 1: 16x16, 32
stage 2: 8x8, 64
参数数量与 [a] 中表 6 接近:
ResNet20 0.27M
ResNet32 0.46M
ResNet44 0.66M
ResNet56 0.85M
ResNet110 1.7M
# 参数
input_shape (tensor): 输入图像张量的尺寸
depth (int): 核心卷积层的数量
num_classes (int): 类别数 (CIFAR10 为 10)
# 返回
model (Model): Keras 模型实例
"""
if (depth - 2) % 6 != 0:
raise ValueError('depth should be 6n+2 (eg 20, 32, 44 in [a])')
# 开始模型定义
num_filters = 16
num_res_blocks = int((depth - 2) / 6)
inputs = Input(shape=input_shape)
x = resnet_layer(inputs=inputs)
# 实例化残差单元的堆栈
for stack in range(3):
for res_block in range(num_res_blocks):
strides = 1
if stack > 0 and res_block == 0: # 第一层但不是第一个栈
strides = 2 # downsample
y = resnet_layer(inputs=x,
num_filters=num_filters,
strides=strides)
y = resnet_layer(inputs=y,
num_filters=num_filters,
activation=None)
if stack > 0 and res_block == 0: # first layer but not first stack
# 线性投影残差快捷键连接，以匹配更改的 dims
x = resnet_layer(inputs=x,
num_filters=num_filters,
kernel_size=1,
strides=strides,
activation=None,
batch_normalization=False)
x = keras.layers.add([x, y])
x = Activation('relu')(x)
num_filters *= 2
# 在顶层加分类器。
# v1 不在最后一个快捷连接 ReLU 后使用 BN
x = AveragePooling2D(pool_size=8)(x)
y = Flatten()(x)
outputs = Dense(num_classes,
activation='softmax',
kernel_initializer='he_normal')(y)
# 实例化模型。
model = Model(inputs=inputs, outputs=outputs)
return model
def resnet_v2(input_shape, depth, num_classes=10):
"""ResNet 版本 2 模型构建器 [b]
(1 x 1)-(3 x 3)-(1 x 1) BN-ReLU-Conv2D 的堆栈，也被称为瓶颈层。
每一层的第一个快捷连接是一个 1 x 1 Conv2D。
第二个及以后的快捷连接是 identity。
在每个阶段的开始，特征图大小由具有 strides=2 的卷积层减半（下采样），
而滤波器的数量加倍。在每个阶段中，这些层具有相同数量的过滤器和相同的特征图尺寸。
特征图尺寸:
conv1 : 32x32, 16
stage 0: 32x32, 64
stage 1: 16x16, 128
stage 2: 8x8, 256
# 参数
input_shape (tensor): 输入图像张量的尺寸
depth (int): 核心卷积层的数量
num_classes (int): 类别数 (CIFAR10 为 10)
# 返回
model (Model): Keras 模型实例
"""
if (depth - 2) % 9 != 0:
raise ValueError('depth should be 9n+2 (eg 56 or 110 in [b])')
# 开始模型定义。
num_filters_in = 16
num_res_blocks = int((depth - 2) / 9)
inputs = Input(shape=input_shape)
# v2 在将输入分离为两个路径前执行带 BN-ReLU 的 Conv2D 操作。
x = resnet_layer(inputs=inputs,
num_filters=num_filters_in,
conv_first=True)
# 实例化残差单元的栈
for stage in range(3):
for res_block in range(num_res_blocks):
activation = 'relu'
batch_normalization = True
strides = 1
if stage == 0:
num_filters_out = num_filters_in * 4
if res_block == 0: # first layer and first stage
activation = None
batch_normalization = False
else:
num_filters_out = num_filters_in * 2
if res_block == 0: # first layer but not first stage
strides = 2 # downsample
# 瓶颈残差单元
y = resnet_layer(inputs=x,
num_filters=num_filters_in,
kernel_size=1,
strides=strides,
activation=activation,
batch_normalization=batch_normalization,
conv_first=False)
y = resnet_layer(inputs=y,
num_filters=num_filters_in,
conv_first=False)
y = resnet_layer(inputs=y,
num_filters=num_filters_out,
kernel_size=1,
conv_first=False)
if res_block == 0:
# 线性投影残差快捷键连接，以匹配更改的 dims
x = resnet_layer(inputs=x,
num_filters=num_filters_out,
kernel_size=1,
strides=strides,
activation=None,
batch_normalization=False)
x = keras.layers.add([x, y])
num_filters_in = num_filters_out
# 在顶层添加分类器
# v2 has BN-ReLU before Pooling
x = BatchNormalization()(x)
x = Activation('relu')(x)
x = AveragePooling2D(pool_size=8)(x)
y = Flatten()(x)
outputs = Dense(num_classes,
activation='softmax',
kernel_initializer='he_normal')(y)
# 实例化模型。
model = Model(inputs=inputs, outputs=outputs)
return model
if version == 2:
model = resnet_v2(input_shape=input_shape, depth=depth)
else:
model = resnet_v1(input_shape=input_shape, depth=depth)
model.compile(loss='categorical_crossentropy',
optimizer=Adam(lr=lr_schedule(0)),
metrics=['accuracy'])
model.summary()
print(model_type)
# 准备模型保存路径。
save_dir = os.path.join(os.getcwd(), 'saved_models')
model_name = 'cifar10_%s_model.{epoch:03d}.h5' % model_type
if not os.path.isdir(save_dir):
os.makedirs(save_dir)
filepath = os.path.join(save_dir, model_name)
# 准备保存模型和学习速率调整的回调。
checkpoint = ModelCheckpoint(filepath=filepath,
monitor='val_acc',
verbose=1,
save_best_only=True)
lr_scheduler = LearningRateScheduler(lr_schedule)
lr_reducer = ReduceLROnPlateau(factor=np.sqrt(0.1),
cooldown=0,
patience=5,
min_lr=0.5e-6)
callbacks = [checkpoint, lr_reducer, lr_scheduler]
# 运行训练，是否数据增强可选。
if not data_augmentation:
print('Not using data augmentation.')
model.fit(x_train, y_train,
batch_size=batch_size,
epochs=epochs,
validation_data=(x_test, y_test),
shuffle=True,
callbacks=callbacks)
else:
print('Using real-time data augmentation.')
# 这将做预处理和实时数据增强。
datagen = ImageDataGenerator(
# 在整个数据集上将输入均值置为 0
featurewise_center=False,
# 将每个样本均值置为 0
samplewise_center=False,
# 将输入除以整个数据集的 std
featurewise_std_normalization=False,
# 将每个输入除以其自身 std
samplewise_std_normalization=False,
# 应用 ZCA 白化
zca_whitening=False,
# ZCA 白化的 epsilon 值
zca_epsilon=1e-06,
# 随机图像旋转角度范围 (deg 0 to 180)
rotation_range=0,
# 随机水平平移图像
width_shift_range=0.1,
# 随机垂直平移图像
height_shift_range=0.1,
# 设置随机裁剪范围
shear_range=0.,
# 设置随机缩放范围
zoom_range=0.,
# 设置随机通道切换范围
channel_shift_range=0.,
# 设置输入边界之外的点的数据填充模式
fill_mode='nearest',
# 在 fill_mode = "constant" 时使用的值
cval=0.,
# 随机翻转图像
horizontal_flip=True,
# 随机翻转图像
vertical_flip=False,
# 设置重缩放因子 (应用在其他任何变换之前)
rescale=None,
# 设置应用在每一个输入的预处理函数
preprocessing_function=None,
# 图像数据格式 "channels_first" 或 "channels_last" 之一
data_format=None,
# 保留用于验证的图像的比例 (严格控制在 0 和 1 之间)
validation_split=0.0)
# 计算大量的特征标准化操作
# (如果应用 ZCA 白化，则计算 std, mean, 和 principal components)。
datagen.fit(x_train)
# 在由 datagen.flow() 生成的批次上拟合模型。
model.fit_generator(datagen.flow(x_train, y_train, batch_size=batch_size),
validation_data=(x_test, y_test),
epochs=epochs, verbose=1, workers=4,
callbacks=callbacks)
# 评估训练模型
scores = model.evaluate(x_test, y_test, verbose=1)
print('Test loss:', scores[0])
print('Test accuracy:', scores[1])