深度学习项目实战——手写数字识别项目

摘要

本文将介绍的有关于的paddle的实战的相关的问题，并分析相关的代码的阅读和解释。并扩展有关于的python的有关的语言。介绍了深度学习步骤： １. 数据处理：读取数据 和 预处理操作 ２. 模型设计：网络结构（假设） ３. 训练配置：优化器（寻解算法） ４. 训练过程：循环调用训练过程，包括前向计算 + 计算损失（优化目标) + 后向传播 ５. 保存模型并测试：将训练好的模型保存。

手写数字识别任务

数字识别是计算机从纸质文档、照片或其他来源接收、理解并识别可读的数字的能力，目前比较受关注的是手写数字识别。手写数字识别是一个典型的图像分类问题，已经被广泛应用于汇款单号识别、手写邮政编码识别等领域，大大缩短了业务处理时间，提升了工作效率和质量。

在处理如 图1 所示的手写邮政编码的简单图像分类任务时，可以使用基于MNIST数据集的手写数字识别模型。MNIST是深度学习领域标准、易用的成熟数据集，包含50 000条训练样本和10 000条测试样本。

• 任务输入：一系列手写数字图片，其中每张图片都是28x28的像素矩阵。
• 任务输出：经过了大小归一化和居中处理，输出对应的0~9的数字标签。

构建手写数字识别的神经网络模型

使用飞桨完成手写数字识别模型任务的代码结构如 图2 所示，与使用飞桨完成房价预测模型任务的流程一致，下面的章节中我们将详细介绍每个步骤的具体实现方法和优化思路。

深度学习步骤：源码

1. 数据处理：读取数据 和 预处理操作

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: data_process.py
@time: 2021/3/4 10:21
"""
#导入需要的包
import paddle as paddle
BUF_SIZE = 512
BATCH_SIZE = 128

# 1、读取数据
def load_data():
    train_reader = paddle.batch(
        paddle.reader.shuffle(paddle.dataset.mnist.train(),
                              buf_size=BUF_SIZE),
        batch_size=BATCH_SIZE)
    # 用于训练的数据提供器，每次从缓存中随机读取批次大小的数据
    test_reader = paddle.batch(
        paddle.reader.shuffle(paddle.dataset.mnist.test(),
                              buf_size=BUF_SIZE),
        batch_size=BATCH_SIZE)
    return train_reader, test_reader

2. 模型设计：网络结构（假设）

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: train_model.py.py
@time: 2021/3/4 10:51
"""
import paddle.fluid as fluid

#使用的是默认的卷积层
def multilayer_perceptron(input):
    # 第一个全连接层，激活函数为ReLU
    hidden1 = fluid.layers.fc(input=input, size=200, act='relu')
    # 第二个全连接层，激活函数为ReLU
    hidden2 = fluid.layers.fc(input=hidden1, size=200, act='relu')
    # 以softmax为激活函数的全连接输出层，输出层的大小必须为数字的个数10
    prediction = fluid.layers.fc(input=hidden2, size=10, act='softmax')
    return prediction


# 构建一个不同的神经网络的模型的 使用的参数是不一致的
def multilayer_perceptron_update(input):
    conv_pool_1 = fluid.nets.simple_img_conv_pool(
        input=input,
        filter_size=5,
        num_filters=20,
        pool_size=2,
        pool_stride=2,
        act="relu")
    conv_pool_1 = fluid.layers.batch_norm(conv_pool_1)
    # 第二个卷积-池化层
    # 使用50个5*5的滤波器，池化大小为2，池化步长为2，激活函数为Relu
    conv_pool_2 = fluid.nets.simple_img_conv_pool(
        input=conv_pool_1,
        filter_size=5,
        num_filters=50,
        pool_size=2,
        pool_stride=2,
        act="relu")
    # 以softmax为激活函数的全连接输出层，输出层的大小必须为数字的个数10
    prediction = fluid.layers.fc(input=conv_pool_2, size=10, act='softmax')
    return prediction

3. 训练配置：优化器（寻解算法） 4. 训练过程：循环调用训练过程，包括前向计算 + 计算损失（优化目标) + 后向传播

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: train_model.py.py
@time: 2021/3/4 10:49
"""
import paddle.fluid as fluid
import matplotlib.pyplot as plt
import os
from Paddle_Number.row_data.data_process import load_data
from Paddle_Number.model_build.nn import multilayer_perceptron

BUF_SIZE = 512
BATCH_SIZE = 128
model_save_dir = r"D:/softwaresavfile/Github/machine_learning/Paddle_Number/mode_save/"

# 画图的功能
def draw_train_process(title, iters, costs, accs, label_cost, lable_acc):
    plt.title(title, fontsize=24)
    plt.xlabel("iter", fontsize=20)
    plt.ylabel("cost/acc", fontsize=20)
    plt.plot(iters, costs, color='red', label=label_cost)
    plt.plot(iters, accs, color='green', label=lable_acc)
    plt.legend()
    plt.grid()
    plt.savefig(r"D:\\softwaresavfile\\Github\\machine_learning\\Paddle_Number\\train_model\\train_img\\train_loss")
    plt.show()

# 4、开始训练
def trainer():
    train_reader, test_reader = load_data()
    # 输入的原始图像数据，大小为1*28*28
    image = fluid.layers.data(name='image', shape=[1, 28, 28], dtype='float32')  # 单通道，28*28像素值
    # 标签，名称为label,对应输入图片的类别标签
    label = fluid.layers.data(name='label', shape=[1], dtype='int64')  # 图片标签
    # 获取分类器
    predict = multilayer_perceptron(image)
    # 使用交叉熵损失函数,描述真实样本标签和预测概率之间的差值
    cost = fluid.layers.cross_entropy(input=predict, label=label)
    # 使用类交叉熵函数计算predict和label之间的损失函数
    avg_cost = fluid.layers.mean(cost)
    # 计算分类准确率
    acc = fluid.layers.accuracy(input=predict, label=label)
    # 使用Adam算法进行优化, learning_rate 是学习率(它的大小与网络的训练收敛速度有关系)
    optimizer = fluid.optimizer.AdamOptimizer(learning_rate=0.001)
    opts = optimizer.minimize(avg_cost)
    # 定义使用CPU还是GPU，使用CPU时use_cuda = False,使用GPU时use_cuda = True
    use_cuda = False
    place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()
    # 获取测试程序
    test_program = fluid.default_main_program().clone(for_test=True)
    exe = fluid.Executor(place)
    exe.run(fluid.default_startup_program())
    feeder = fluid.DataFeeder(place=place, feed_list=[image, label])

    all_train_iter = 0
    all_train_iters = []
    all_train_costs = []
    all_train_accs = []

    EPOCH_NUM = 2

    for pass_id in range(EPOCH_NUM):
        # 进行训练
        for batch_id, data in enumerate(train_reader()):  # 遍历train_reader
            train_cost, train_acc = exe.run(program=fluid.default_main_program(),  # 运行主程序
                                            feed=feeder.feed(data),  # 给模型喂入数据
                                            fetch_list=[avg_cost, acc])  # fetch 误差、准确率

            all_train_iter = all_train_iter + BATCH_SIZE
            all_train_iters.append(all_train_iter)

            all_train_costs.append(train_cost[0])
            all_train_accs.append(train_acc[0])

            # 每200个batch打印一次信息  误差、准确率
            if batch_id % 200 == 0:
                print('Pass:%d, Batch:%d, Cost:%0.5f, Accuracy:%0.5f' %
                      (pass_id, batch_id, train_cost[0], train_acc[0]))

        # 进行测试
        test_accs = []
        test_costs = []
        # 每训练一轮 进行一次测试
        for batch_id, data in enumerate(test_reader()):  # 遍历test_reader
            test_cost, test_acc = exe.run(program=test_program,  # 执行训练程序
                                          feed=feeder.feed(data),  # 喂入数据
                                          fetch_list=[avg_cost, acc])  # fetch 误差、准确率
            test_accs.append(test_acc[0])  # 每个batch的准确率
            test_costs.append(test_cost[0])  # 每个batch的误差

        # 求测试结果的平均值
        test_cost = (sum(test_costs) / len(test_costs))  # 每轮的平均误差
        test_acc = (sum(test_accs) / len(test_accs))  # 每轮的平均准确率
        print('Test:%d, Cost:%0.5f, Accuracy:%0.5f' % (pass_id, test_cost, test_acc))

        # 保存模型
        # 如果保存路径不存在就创建
    if not os.path.exists(model_save_dir):
        os.makedirs(model_save_dir)
    print('save models to %s' % (model_save_dir))
    fluid.io.save_inference_model(model_save_dir,  # 保存推理model的路径
                                  ['image'],  # 推理（inference）需要 feed 的数据
                                  [predict],  # 保存推理（inference）结果的 Variables
                                  exe)  # executor 保存 inference model

    print('训练模型保存完成！')
    draw_train_process("training", all_train_iters, all_train_costs, all_train_accs, "trainning cost", "trainning acc")

if __name__ == '__main__':
    trainer()

5. 保存模型并测试：将训练好的模型保存

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: infer_model.py
@time: 2021/3/4 11:01
"""
import numpy as np
import paddle.fluid as fluid
from PIL import Image
import matplotlib.pyplot as plt

use_cuda = True
place = fluid.CUDAPlace(0) if use_cuda else fluid.CPUPlace()

model_save_dir = r"D:\softwaresavfile\Github\machine_learning\手写数学的识别\mode_save\__model__"


# 图片预处理
def load_image(file):
    im = Image.open(file).convert('L')  # 将RGB转化为灰度图像，L代表灰度图像，像素值在0~255之间
    im = im.resize((28, 28), Image.ANTIALIAS)  # resize image with high-quality 图像大小为28*28
    im = np.array(im).reshape(1, 1, 28, 28).astype(np.float32)  # 返回新形状的数组,把它变成一个 numpy 数组以匹配数据馈送格式。
    # print(im)
    im = im / 255.0 * 2.0 - 1.0  # 归一化到【-1~1】之间
    return im


# 预测函数
def predict(infer_path):
    # 查询原来的图像
    # infer_path = '预测文件的路径'
    img = Image.open(infer_path)
    plt.imshow(img)  # 根据数组绘制图像
    plt.show()  # 显示图像

    # 创建Executer
    infer_exe = fluid.Executor(place)
    inference_scope = fluid.core.Scope()
    # 加载数据并开始预测
    plt.imshow(img)  # 根据数组绘制图像
    plt.show()  # 显示图像

    with fluid.scope_guard(inference_scope):
        # 获取训练好的模型
        # 从指定目录中加载 推理model(inference model)
        [inference_program,  # 推理Program
         feed_target_names,  # 是一个str列表，它包含需要在推理 Program 中提供数据的变量的名称。
         fetch_targets] = fluid.io.load_inference_model(model_save_dir,
                                                        # fetch_targets：是一个 Variable 列表，从中我们可以得到推断结果。model_save_dir：模型保存的路径
                                                        infer_exe)  # infer_exe: 运行 inference model的 executor

        img = load_image(infer_path)

        results = infer_exe.run(program=inference_program,  # 运行推测程序
                                feed={feed_target_names[0]: img},  # 喂入要预测的img
                                fetch_list=fetch_targets)  # 得到推测结果,
        # 获取概率最大的label
        lab = np.argsort(results)  # argsort函数返回的是result数组值从小到大的索引值

        print("该图片的预测结果的label为: %d" % lab[0][0][-1])  # -1代表读取数组中倒数第一列


if __name__ == '__main__':
    predict()

构建波士顿房价预测的神经网络模型 １. 数据处理：读取数据 和 预处理操作

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@version: 1.0
@author: xjl
@file: data_process.py
@time: 2021/3/4 14:01
"""
from __future__ import print_function
import paddle
import numpy
import six.moves

def load_data_txt(filename):
    """
    如果想直接从txt文件中读取数据的话，可以参考以下方式。
    :param filename:
    :return:
    """
    feature_names = ['CRIM', 'ZN', 'INDUS', 'CHAS', 'NOX', 'RM', 'AGE', 'DIS', 'RAD', 'TAX', 'PTRATIO', 'B', 'LSTAT',
                     'convert']

    feature_num = len(feature_names)

    data = numpy.fromfile(filename, sep=' ')  # 从文件中读取原始数据

    data = data.reshape(data.shape[0] // feature_num, feature_num)

    maximums, minimums, avgs = data.max(axis=0), data.min(axis=0), data.sum(axis=0) / data.shape[0]

    for i in six.moves.range(feature_num - 1): data[:, i] = (data[:, i] - avgs[i]) / (
            maximums[i] - minimums[i])  # six.moves可以兼容python2和python3

    ratio = 0.8  # 训练集和验证集的划分比例

    offset = int(data.shape[0] * ratio)

    train_data = data[:offset]

    test_data = data[offset:]

    train_reader = paddle.batch(paddle.reader.shuffle(train_data, buf_size=500), batch_size=BATCH_SIZE)

    test_reader = paddle.batch(paddle.reader.shuffle(test_data, buf_size=500), batch_size=BATCH_SIZE)


if __name__ == '__main__':
    BATCH_SIZE = 20

    train_reader = paddle.batch(
        paddle.reader.shuffle(
            paddle.dataset.uci_housing.train(), buf_size=500),
        batch_size=BATCH_SIZE)

    test_reader = paddle.batch(
        paddle.reader.shuffle(
            paddle.dataset.uci_housing.test(), buf_size=500),
        batch_size=BATCH_SIZE)

２. 模型设计：网络结构（假设）

cost = fluid.layers.square_error_cost(input=y_predict, label=y)  # 利用标签数据和输出的预测数据估计方差
avg_loss = fluid.layers.mean(cost)  # 对方差求均值，得到平均损失

３. 训练配置：优化器（寻解算法）

#有些operator在训练和测试之间的操作是不同的，例如batch_norm，使用参数for_test来区分该程序是用来训练还是用来测试
#该api不会删除任何操作符,请在backward和optimization之前使用
test_program = main_program.clone(for_test=True)

sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
sgd_optimizer.minimize(avg_loss)

４. 训练过程：循环调用训练过程，包括前向计算 + 计算损失（优化目标) + 后向传播

import paddle.nn.functional as F
y_preds = []
labels_list = []

def train(model):
    print('start training ... ')
    # 开启模型训练模式
    model.train()
    EPOCH_NUM = 500
    train_num = 0
    optimizer = paddle.optimizer.SGD(learning_rate=0.001, parameters=model.parameters())
    for epoch_id in range(EPOCH_NUM):
        # 在每轮迭代开始之前，将训练数据的顺序随机的打乱
        np.random.shuffle(train_data)
        # 将训练数据进行拆分，每个batch包含20条数据
        mini_batches = [train_data[k: k+BATCH_SIZE] for k in range(0, len(train_data), BATCH_SIZE)]
        for batch_id, data in enumerate(mini_batches):
            features_np = np.array(data[:, :13], np.float32)
            labels_np = np.array(data[:, -1:], np.float32)
            features = paddle.to_tensor(features_np)
            labels = paddle.to_tensor(labels_np)
            #前向计算
            y_pred = model(features)
            cost = F.mse_loss(y_pred, label=labels)
            train_cost = cost.numpy()[0]
            #反向传播
            cost.backward()
            #最小化loss，更新参数
            optimizer.step()
            # 清除梯度
            optimizer.clear_grad()

            if batch_id%30 == 0 and epoch_id%50 == 0:
                print("Pass:%d,Cost:%0.5f"%(epoch_id, train_cost))

            train_num = train_num + BATCH_SIZE
            train_nums.append(train_num)
            train_costs.append(train_cost)

model = Regressor()
train(model)

５. 保存模型并测试：将训练好的模型保存。

#获取预测数据
INFER_BATCH_SIZE = 100

infer_features_np = np.array([data[:13] for data in test_data]).astype("float32")
infer_labels_np = np.array([data[-1] for data in test_data]).astype("float32")

infer_features = paddle.to_tensor(infer_features_np)
infer_labels = paddle.to_tensor(infer_labels_np)
fetch_list = model(infer_features)

sum_cost = 0
for i in range(INFER_BATCH_SIZE):
    infer_result = fetch_list[i][0]
    ground_truth = infer_labels[i]
    if i % 10 == 0:
        print("No.%d: infer result is %.2f,ground truth is %.2f" % (i, infer_result, ground_truth))
    cost = paddle.pow(infer_result - ground_truth, 2)
    sum_cost += cost
mean_loss = sum_cost / INFER_BATCH_SIZE
print("Mean loss is:", mean_loss.numpy())

使用更高级的API

import paddle
paddle.set_default_dtype("float64")

#step1:用高层API定义数据集，无需进行数据处理等，高层API为您一条龙搞定
train_dataset = paddle.text.datasets.UCIHousing(mode='train')
eval_dataset = paddle.text.datasets.UCIHousing(mode='test')

#step2:定义模型
class UCIHousing(paddle.nn.Layer):
    def __init__(self):
        super(UCIHousing, self).__init__()
        self.fc = paddle.nn.Linear(13, 1, None)

    def forward(self, input):
        pred = self.fc(input)
        return pred

#step3:训练模型
model = paddle.Model(UCIHousing())
model.prepare(paddle.optimizer.Adam(parameters=model.parameters()),
              paddle.nn.MSELoss())
model.fit(train_dataset, eval_dataset, epochs=5, batch_size=8, verbose=1)

博文参考