Vgbhfive's Blog

DL-卷积神经网络CNN实践

Posted on Edited on

简介

该篇主要在于实践卷积神经网络 CNN，使用 MNIST 和 CIFAR10 数据集完成分类任务。

MNIST

MNIST 数据集在之前的实践中使用 TensorFlow 基础实现，而这一篇则是使用卷积神经网络（CNN）实现。

引入依赖

1
2
3
4
5
6
import numpy as np
from tensorflow.keras.datasets import mnist
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Conv2D, Flatten, MaxPooling2D, Dropout
from tensorflow.keras.utils import to_categorical
import matplotlib.pyplot as plt

加载数据并预处理

1
2
3
4
5
6
7
(x_train, y_train), (x_test, y_test) = mnist.load_data()

# 数据预处理
x_train, x_test = x_train / 255.0, x_test / 255.0 # 归一化像素值到0-1
# 将标签转换为 one-hot 编码
y_train = to_categorical(y_train, 10)
y_test = to_categorical(y_test, 10)

构建并编译模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
model = Sequential([
    Conv2D(32, (3, 3), activation='relu', input_shape=(28, 28, 1)),
    MaxPooling2D((2, 2)),
    Conv2D(32, (3, 3), activation='relu'),
    MaxPooling2D((2, 2)),
    # 展平层，将多维的输出展平为一维
    Flatten(),
    # 全连接层，使用128个神经元和ReLU激活函数
    Dense(128, activation='relu'),
    # Dropout层，丢弃一部分神经元，防止过拟合
    Dropout(0.3),
    Dense(10, activation='softmax')
])

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

# 训练模型
model.fit(x_train, y_train, epochs=5, batch_size=64, validation_split=0.1)

mnist_9

模型评估

1
2
3
4
5
6
7
8
9
10
11
12
test_loss, test_acc = model.evaluate(x_test, y_test)
print(f"Test accuracy: {test_acc:.4f}, Test loss: {test_loss:.4f}") # 评估预测结果

# 测试实际效果
start, stop = 10, 11
y_pred = model.predict(x_test[start:stop])
for i in range(start, stop):
    print(i, np.argwhere(y_test[i] == 1)[0][0], np.argmax(y_pred[i-start]))
    plt.title(np.argmax(y_pred[i]))
    plt.imshow(x_test[i], cmap='gray')
    plt.show()
    print('-'*50)

mnist_11
mnist_10

CIFAR10 数据集实战

CIFAR 数据集包含 60k 张十个种类的图片，每张图片尺寸为 32*32*3，三通道色图片，分辨率是 32*32，类别包含：飞机、汽车、鸟类、猫、鹿、狗、青蛙、马、船和卡车，每个类别分别有 6k 张。而 CIFAR10 分类任务是将这些图像正确地分类到它们所属的类别中。

引入依赖

1
2
3
4
5
6
7
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

import tensorflow as tf
from tensorflow.keras import datasets, layers, Sequential, optimizers

加载数据

1
2
3
(x,y), (x_test, y_test) = datasets.cifar10.load_data()

x.shape, y.shape, x_test.shape, y_test.shape

cifar_1

数据预处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 删除y的一个维度
y = tf.squeeze(y, axis=1)
y_test = tf.squeeze(y_test, axis=1)
x.shape, y.shape, x_test.shape, y_test.shape

def preprocess(x, y):
    # [0~1]
    x = 2 * tf.cast(x, dtype=tf.float32) / 255. - 1
    y = tf.cast(y, dtype=tf.int32)
    return x,y

# 构建训练集对象，随机打乱，预处理，批量化
train_db = tf.data.Dataset.from_tensor_slices((x,y))
train_db = train_db.shuffle(1000).map(preprocess).batch(128)

# 构建测试集
test_db = tf.data.Dataset.from_tensor_slices((x_test, y_test))
test_db = test_db.map(preprocess).batch(128)

cifar_2

构建网络模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
conv_net = Sequential([
    # unit 1 - 64个3x3卷积核，输入输出同样大小
    layers.Conv2D(64, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.Conv2D(64, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.MaxPool2D(pool_size=[2,2], strides=2, padding="same"),
    
    # unit 2 - 输出通道提升至128.高宽大小减半
    layers.Conv2D(128, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.Conv2D(128, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.MaxPool2D(pool_size=[2,2], strides=2, padding="same"),
    
    # unit 3 - 输出通道提升至256，高宽大小减半
    layers.Conv2D(256, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.Conv2D(256, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.MaxPool2D(pool_size=[2,2], strides=2, padding="same"),
    
    # unit 4 - 输出通道提升至512，高宽大小减半
    layers.Conv2D(512, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.Conv2D(512, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.MaxPool2D(pool_size=[2,2], strides=2, padding="same"),
    
    # unit 5 - 输出通道提升至512，高宽大小减半
    layers.Conv2D(512, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.Conv2D(512, kernel_size=[3,3], padding="same", activation=tf.nn.relu),
    layers.MaxPool2D(pool_size=[2,2], strides=2, padding="same")
])

fc_net = Sequential([
    layers.Dense(512, activation=tf.nn.relu),
    layers.Dense(256, activation=tf.nn.relu),
    layers.Dense(128, activation=tf.nn.relu),
    layers.Dense(10, activation=None)
])

网络参数信息

1
2
3
4
5
6
7
conv_net.build(input_shape=[None, 32, 32, 3])
fc_net.build(input_shape=[None, 512])
conv_net.summary()
fc_net.summary()
optimizer = optimizers.Adam(learning_rate=0.0001)

variables = conv_net.trainable_variables + fc_net.trainable_variables

cifar_3
cifar_4

训练模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
losses, acces = [], []
for epoch in range(20):
    for step, (x, y) in enumerate(train_db):
        with tf.GradientTape() as tape:
            # [b,32,32,3] -> [b,1,1,512]
            out = conv_net(x)
            # flatten - [b,1,1,512] -> [b,512]
            out = tf.reshape(out, [-1,512])
            # [b,512] -> [b,10]
            logits = fc_net(out)
            # [b] -> [b,10]
            y_onehot = tf.one_hot(y, depth=10)
            # compute loss
            loss = tf.losses.categorical_crossentropy(y_onehot, logits, from_logits=True)
            loss = tf.reduce_mean(loss)
        
        grads = tape.gradient(loss, variables)
        optimizer.apply_gradients(zip(grads, variables))
        
        if step % 100 == 0:
            print(epoch, step, 'loss: ', float(loss))
            losses.append(loss.numpy())
    
    # 计算每轮次训练之后的准确率
    total_num = 0
    total_correct = 0
    for x,y in test_db:
        out = conv_net(x)
        out = tf.reshape(out, [-1,512])
        logits = fc_net(out)
        prob = tf.nn.softmax(logits, axis=1)
        pred = tf.argmax(prob, axis=1)
        pred = tf.cast(pred, dtype=tf.int32) 
        
        correct = tf.cast(tf.equal(pred, y), dtype=tf.int32)
        correct = tf.reduce_sum(correct)
        
        total_num += x.shape[0]
        total_correct += int(correct)
    
    acc = total_correct / total_num
    acces.append(acc)

cifar_train

展示指标

1
2
3
4
5
6
7
8
# acc plt
data = pd.DataFrame()
data['acc'] = acces
sns.lineplot(data=data)
# loss plt
data2 = pd.DataFrame()
data2['loss'] = losses
sns.lineplot(data=data2)

cifar_acc
cifar_loss

CIFAR10 使用 ResNet 分类实战

依旧采用 CIFAR10 数据集，使用 ResNet 残差网络完成分类。

引入依赖

1
2
3
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import datasets, layers, Sequential, optimizers

加载数据

1
(x,y), (x_test, y_test) = datasets.cifar10.load_data()

数据预处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 删除y的一个维度
y = tf.squeeze(y, axis=1)
y_test = tf.squeeze(y_test, axis=1)

def preprocess(x, y):
    # [0~1]
    x = 2 * tf.cast(x, dtype=tf.float32) / 255. - 1
    y = tf.cast(y, dtype=tf.int32)
    return x,y

# 构建训练集对象，随机打乱，预处理，批量化
train_db = tf.data.Dataset.from_tensor_slices((x,y))
train_db = train_db.shuffle(1000).map(preprocess).batch(128)

# 构建测试集
test_db = tf.data.Dataset.from_tensor_slices((x_test, y_test))
test_db = test_db.map(preprocess).batch(128)

ResNet 网络

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class BasicBlock(layers.Layer):
    # 残差模块类
    def __init__(self, filter_num, stride=1):
        super(BasicBlock, self).__init__()
        # 卷积层1
        self.conv1 = layers.Conv2D(filter_num, (3,3), strides=stride, padding="same")
        self.bn1 = layers.BatchNormalization()
        self.relu = layers.Activation('relu')
        # 卷积层2
        self.conv2 = layers.Conv2D(filter_num, (3,3), strides=1, padding="same")
        self.bn2 = layers.BatchNormalization()
        # 当f(x)与x的shape不一致时，需要新建identity(x)卷积层，完成形状转换 
        if stride != 1:
            self.downsample = Sequential()
            self.downsample.add(layers.Conv2D(filter_num, (1,1), strides=stride))
        else:
            self.downsample = lambda x:x

    def call(self, inputs, training=None):
        # 前向传播
        out = self.conv1(inputs)
        out = self.bn1(out)
        out = self.relu(out)
        out = self.conv2(out)
        out = self.bn2(out)
        # 输入通过identity()转换
        identity = self.downsample(inputs)
        # x+f(x)
        output = layers.add([out, identity])
        # 通过激活函数返回
        return tf.nn.relu(output)

class ResNet(keras.Model):
    
    def __init__(self, layer_dims, num_classes=10):
        super(ResNet, self).__init__()
        # 根网络，预处理
        self.stem = Sequential([
            layers.Conv2D(64, (3,3), strides=(1,1)),
            layers.BatchNormalization(),
            layers.Activation('relu'),
            layers.MaxPooling2D(pool_size=(2,2), strides=(1,1), padding="same")
        ])
        # 堆叠4个Block，每个block包含多个BasicBlock，设置步长均一样
        self.layer1 = self.build_resblock(64, layer_dims[0])
        self.layer2 = self.build_resblock(128, layer_dims[1], stride=2)
        self.layer3 = self.build_resblock(256, layer_dims[2], stride=2)
        self.layer4 = self.build_resblock(512, layer_dims[3], stride=2)

        # 通过 Pooling 层将高宽降低为 1*1
        self.avgpool = layers.GlobalAveragePooling2D()
        # 全连接层分类
        self.fc = layers.Dense(num_classes)
    
    def call(self, inputs, training=None):
        # 前向计算函数：通过根网络
        x = self.stem(inputs)
        # 四个模块
        x = self.layer1(x)
        x = self.layer2(x)
        x = self.layer3(x)
        x = self.layer4(x)
        # 池化层
        x = self.avgpool(x)
        # 全连接层
        x = self.fc(x)
        return x
        
    # 通过堆叠通道数逐渐增大的 ResBlock 实现高层特征的提取，通过 build_resblock 可以一次完成多个残差模块的新建
    def build_resblock(self, filter_num, blocks, stride=1):
        res_blocks = Sequential()
        res_blocks.add(BasicBlock(filter_num, stride))

        for _ in range(1, blocks):
            res_blocks.add(BasicBlock(filter_num, stride=1))
        return res_blocks

搭建模型

1
2
3
4
5
6
7
8
9
# 通过调整不同的 ResBlock 的堆叠数量和通道数可以产生不同的 ResNet。
def resnet18():
    # 64-64-128-128-256-256-512-512，共8个ResBlock，每个ResBlock 包含两个卷积层，再加上网络首尾的全连接层，共18层。
    return ResNet([2,2,2,2])

model = resnet18()
model.build(input_shape=(None, 32, 32, 3))
model.summary() # 统计网络参数
optimizer = optimizers.Adam(learning_rate=1e-4) # 构建优化器

cifar2_1

模型训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
losses, acces = [], []

for epoch in range(20):
    for step, (x, y) in enumerate(train_db):
        with tf.GradientTape() as tape:
            logits = model(x)
            # [b] -> [b,10]
            y_onehot = tf.one_hot(y, depth=10)
            # compute loss
            loss = tf.losses.categorical_crossentropy(y_onehot, logits, from_logits=True)
            loss = tf.reduce_mean(loss)
            losses.append(loss)
        
        grads = tape.gradient(loss, model.trainable_variables)
        optimizer.apply_gradients(zip(grads, model.trainable_variables))

        if step % 100 == 0:
            print(epoch, step, 'loss: ', float(loss))

    total_num = 0
    total_correct = 0
    for x,y in test_db:
        logits = model(x)
        prob = tf.nn.softmax(logits, axis=1)
        pred = tf.argmax(prob, axis=1)
        pred = tf.cast(pred, dtype=tf.int32)

        correct = tf.cast(tf.equal(pred, y), dtype=tf.int32)
        correct = tf.reduce_sum(correct)
        total_num += x.shape[0]
        total_correct += int(correct)

    acc = total_correct / total_num
    acces.append(acc)
    print(epoch, 'acc: ', acc)

指标展示

1
2
3
4
5
6
7
8
9
10
11
import pandas as pd
import seaborn as sns

# acc plt
data = pd.DataFrame()
data['acc'] = acces
sns.lineplot(data=data)
# loss plt
data2 = pd.DataFrame()
data2['loss'] = losses
sns.lineplot(data=data2)

总结

在写完好几篇实践项目后就会发现，深度学习的重点在于准备结构化数据和设计网络框架，其余的东西在了解后基本上就是属于墨守成规的内容。

引用

个人备注

此博客内容均为作者学习《TensorFlow深度学习》所做笔记，侵删！
若转作其他用途，请注明来源！