神经网络分类任务

数据集获取

又是一个新的专辑，我会在这里带大家一起从实战中学习pytorch框架。本节我们介绍手写字体识别的任务，使用的Mnist数据集是在官网下载的，这种获取方式并不常见，我们只在今天的讲解中用到，所以这段的代码不需要理解。

import torch
from pathlib import Path
import requests
import pickle
import gzip

DATA_PATH = Path("data")
PATH = DATA_PATH / "mnist"
PATH.mkdir(parents=True, exist_ok=True)
URL = "http://deeplearning.net/data/mnist/"
FILENAME = "mnist.pkl.gz"
if not (PATH / FILENAME).exists():
        content = requests.get(URL + FILENAME).content
        (PATH / FILENAME).open("wb").write(content)


with gzip.open((PATH / FILENAME).as_posix(), "rb") as f:
        ((x_train, y_train), (x_valid, y_valid), _) = pickle.load(f, encoding="latin-1") # 拿出训练集和测试集
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

通过以上的代码，我们获得了四个主要的数据部分，x_train代表训练集的数据，它由50000张手写数字的灰度图片构成，每张图片是28*28的矩阵。我们可以用以下代码查看图片信息：

from matplotlib import pyplot
import numpy as np

pyplot.imshow(x_train[1].reshape((28, 28)), cmap="gray")
1
2
3
4

y_train代表训练集的每张图片所代表的数字，x_valid代表测试集的数据，由10000张与x_train相同的图片构成，y_valid代表测试集每张图片代表的数字。
本次任务是利用神经网络对这些图像进行十分类，我们最终会得到每个图像属于0到9中每一个数字的概率：

这其中，三层网络就是做了三次wx+b操作。每个输入数据有748个特征，我们可以规定每做一次wx+b，会获得所有输入数据的多次特征加权。假如我们想要先得到128个特征，然后得到256个特征，最后完成10分类。为了完成这个任务，w1要选取784×128大小的矩阵，b1要选取128×1的矩阵；w2要选取128×256大小的矩阵，b2要选取256×1的矩阵；w3要选取256×10大小的矩阵，b3要选取10×1的矩阵。学习的过程就是不断地更新w和b的过程，学习完成之后，计算机会保存每次学习中产生的分类效果最好的所有w和b阵。

简单的分类实践

了解了神经网络工作的特点，我们做一个简单的网络看看分类是如何进行的。由于样本并不复杂，我们选择batch大小为64，只有一层的网络，损失函数选择交叉删损失。另外在开始测试之前，我们需要把xb和yb的格式转换一下，torch模块操作的对象是Tensor类型，这个类型和我们常见的ndarray（numpy库用于存储矩阵的类型）很像，也可以保存矩阵。但是ndarry类型是不能被torch模块解析的，所以一定要转换一下。我们采用map函数进行类型的转换，该函数使用方法如下:

a=(1,2,3)
b=(4,5,6)
print(a,b,type(a),type(b))
a,b=map(list,(a,b)) # 将a和b转换成list类型
print(a,b,type(a),type(b))
# 输出为：(1, 2, 3) (4, 5, 6)  
		  [1, 2, 3] [4, 5, 6] <class 'list'> <class 'list'>
1
2
3
4
5
6
7

了解这些之后，我们开始写测试函数了：

import torch
# torch用到的数据集数据类型需要置为tensor，适合存储在GPU中
import torch.nn.functional as F # functional里包含经典的损失函数、激活函数等

# map函数是一个映射，可以直接将四个ndarray格式的矩阵映射为torch.tensor格式
x_train, y_train, x_valid, y_valid = map(torch.tensor, (x_train,y_train, x_valid, y_valid))
n, c = x_train.shape # n是样本个数，c是每个样本像素点个数

loss_func = F.cross_entropy # 调用经典损失函数：交叉熵损失
def test_1(xb):
    return xb.mm(weights) + bias # 计算w*(xb)+b
    
bs = 64 # 一次训练使用样本的个数

# 先进性对w的随机初始化
# 784个像素点，10个类别，故w是784*10的矩阵
weights = torch.randn([784, 10], dtype = torch.float,requires_grad = True)
# b对结果影响不大，所以以常数作为初始化
# 要做的分类是10分类，所以b要做成10*1的0矩阵
bias = torch.zeros(10, requires_grad=True)

# 计算预测值和真实值间差异大小
print(loss_func(test_1(xb),yb)) # model(xb)代表预测值， yb代表标签
# 输出为：tensor(13.1795, grad_fn=)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

这个结果的大小反映了学习效果的好坏，但看上去并不直观。另外，因为w矩阵是随机初始化的，所以每一次运行都会得到不一样的输出结果。

创建model简化代码

在上面的测试函数中，我们用到的是nn.functional，然而如果模型有可学习的参数，最好用nn.Module。Model可以通过简单的代码实现深度学习的很多复杂操作。使用之前，我们需要注意以下的要点：

• 须继承nn.Module且在其构造函数中调用nn.Module的构造函数；
• 无需写反向传播函数，nn.Module能够利用autograd自动实现反向传播；
• Module中的可学习参数可以通过named_parameters()或者parameters()返回迭代器。

下面我们构建一个满足条件类：

from torch import nn
bs=64
class Mnist_NN(nn.Module): # 创建新类，继承nn.Module类
    def __init__(self):
        super().__init__() # 调用父类的构造函数
        # 模型会自动进行权重参数随机初始化
        self.hidden1 = nn.Linear(784, 128) # 初始化w1，b1
        self.hidden2 = nn.Linear(128, 256) # 初始化w2，b2
        self.out  = nn.Linear(256, 10) # 初始化w3,b3
        self.dropout = nn.Dropout(0.5) # 随机失活,概率为50%，避免过拟合
    # 前向传播需要自己定义，返向传递会自动进行
    def forward(self, x): # 每个输入是784个特征点，batch选的是64，x即为64*784
        x = F.relu(self.hidden1(x)) # 把x变成中间的结果：64*128（64*784矩阵（输入）乘784*128矩阵（w1）+128*1矩阵（b1））
        x = self.dropout(x) # 随机失活
        x = F.relu(self.hidden2(x)) # 把x变成中间的结果：64*256（64*128矩阵（第一隐层）乘784*128矩阵（w2）+128*1矩阵（b2））
        x = self.dropout(x) # 随机失活
        x = self.out(x)
        return x
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

处理训练集和测试集

我们获取了学习所需用到的数据之后，将其分为两大部分，一部分为训练集，训练集的作用在于找到学习效果最好的w和b，另一部分为测试集，作用在于检验学习的效果好坏。在训练和测试之前还需要经历一轮数据的封装，我们需要把“练习题”（灰度图像）和对应的“答案”（图像代表的数字）全部封装到一起，然后再拆分成64个（一个batch）一组的形式做成一套“试卷”，做过几套试卷（一个epoch）之后，机器就有了一定的识别能力了，这个时候我们就可以对这个学习结果进行测试，测试卷和答案也是用相同的办法封装出来的，并且测试题用到的题型不变，只是题量有所变化（可以取128个图像数据为一组测试题），计算机答题完成后会生成测试卷的分数，根据测试卷的分数就可以判断学习的效果了。这就是计算机学习和检验的全过程。封装用到的模块是TensorDataset和DataLoader，本节我们只做初步了解即可。上代码：

from torch.utils.data import TensorDataset
from torch.utils.data import DataLoader
# 训练集
train_ds = TensorDataset(x_train, y_train) # 把x_train, y_train封装成TensorDataset格式，相当于整理所有题目和答案
# 测试集
valid_ds = TensorDataset(x_valid, y_valid)

def get_data(train_ds, valid_ds, bs): # 继续封装
    train_dl = DataLoader(train_ds, batch_size=bs, shuffle=True) #train_ds转换成DataLoader格式
                                                                 # 把完整的训练集数据（50000个手写数字图像）打包成bs（64）个一组，全部打包之后交给GPU
                                                                 # shuffle=True为打乱数据顺序，训练集需要打乱顺序提升学习效果
                                                                 # 64道题一张卷
    # 测试集不需要打乱顺序
    valid_dl = DataLoader(valid_ds, batch_size=bs *2) # 128道题一张卷
    return train_dl,valid_dl
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

有了封装，自然就要有解封。计算机做题给出的结果是一张图片属于每个分类的概率，因此需要每做一道题就对一次答案，然后重新修订w和b，从而达到更好的分类效果。为了配合这样的学习方式，我们应当将用以学习的卷子每道题的题目和答案先匹配起来存储备用，这个过程就需要解封了。简单地说，解封的目的就是教计算机如何对答案。python给我们提供了zip函数，它可以快速完成对位匹配。大家看了这段代码就可以了解zip函数是怎么运行的了：

a=[1,2,3]
b=[4,5,6]
c=[4,5,6,7]
print(list(zip(a,b)))
print(list(zip(a,c)))
d=zip(a,b)
for i in zip(*d):
    print(i)
# 输出为：[(1, 4), (2, 5), (3, 6)]
#         [(1, 4), (2, 5), (3, 6)]
#         (1, 2, 3)
#         (4, 5, 6)
1
2
3
4
5
6
7
8
9
10
11
12

以上我们了解了计算机学习和检验的形象过程，接下来我们就要用代码实现这个过程了：

训练和测试的代码实现

import numpy as np
# 学习并检验
# steps：数据集迭代次数
# model：定义好的模型
# loss_func：损失函数
# opt：优化器
def fit(steps, model, loss_func, opt, train_dl, valid_dl):
    for step in range(steps): # 遍历每个epoch（epoch由一个或多个Batch组成，
                              # 因为系统会根据epoch的设定值定义epoch包含的batch量
                              # 所以可以理解成它被动地定义一论学习计算机需要做的试卷数量）
        # 一般在训练模型时加上model.train()，这样会正常使用Batch Normalization和 Dropout
        model.train() # 训练，需要更新所有的权重和偏置
                      # 通常每经历一次外循环，损失都会有所减小，epoch设置越大，最后的训练结果也越好
       
    
    
        # 训练每个batch（每个分组，64个数据）
        for xb, yb in train_dl: # xb对应了某一层的输入，yb对应输入每个样本的标签
            loss_batch(model, loss_func, xb, yb, opt) # 靠opt和loss_func来实现w和b的更新
                                                      # 更新参数即可，不需要接收返回值
        
        # 测试的时候一般选择model.eval()，这样就不会使用Batch Normalization和 Dropout
        model.eval()
        with torch.no_grad(): # 测试，不更新权重参数
            losses, nums = zip( # 将当前的损失及编号单独存放
                *[loss_batch(model, loss_func, xb, yb) for xb, yb in valid_dl]
            )
        val_loss = np.sum(np.multiply(losses, nums)) / np.sum(nums)
        # np.sum(np.multiply(losses, nums)) losses和nums对应相乘再相加(两个都是list类型），计算总损失
        # 总损失计算完成后在计算总的平均损失
        print('当前step:'+str(step+1), '验证集损失：'+str(val_loss))
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

优化和计算损失是模型中非常重要的两个部分。所谓优化，就是提供w和b的更新方向。每次学习中，每层w和b的第一次初始化都是随机的，但如果w和b的更新方向也是随机的，那么可想而知学习效果必定是很差的。但是有了优化器提供更新方向，w和b的每次更新就都会比上一次效果更好；计算损失则是一套评分系统，因为每一轮完整的学习下来都需要进行一轮测试，计算机做了测试题后需要对测试结果好坏进行评判，因此就需要一套评分系统。从名字上可以看出，损失值越大，学习效果就越糟糕。上述代码中这两个部分是以函数的形式直接使用的，下面我们给出具体函数：

from torch import optim
# 模型实例化，并给出优化算法
def get_model():
    model = Mnist_NN()
    return model, optim.SGD(model.parameters(), lr=0.001) # SGD：梯度下降，lr：学习率，学习率设置过大可鞥会错过最优解情况
                                                          # model.parameters()：该参数可实现每一层w、b的更新，将传递到其他函数中实现更新

# 计算损失
# 按优化模式更新w和b
def loss_batch(model, loss_func, xb, yb, opt=None):
    loss = loss_func(model(xb), yb) # model(xb)：把输入放入模型之中得到预测值
                                    # yb：真实值

    if opt is not None: # 配合优化模式进行更新和计算
        loss.backward() # 反向传播，计算梯度
        opt.step() # 对w和b进行更新，沿着梯度方向更新学习率（lr）个大小
        opt.zero_grad() # 清空之前的梯度。因为torch默认会对梯队进行累加，即上一次的结果会影响到下一次的迭代方向。
                        # 如果不做清空将会影响每次迭代的独立性

    return loss.item(), len(xb) # 返回loss值及训练的样本数量，样本数量用以计算平均损失
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这样一来整个模型的部件就全了，我们只把它们按顺序调用一下就可以查看学习的效果：

train_dl, valid_dl = get_data(train_ds, valid_ds, bs)
model, opt = get_model()
step=25 # 分成25个epoch
fit(step, model, loss_func, opt, train_dl, valid_dl)
1
2
3
4

明显能看出损失值在下降吧，这就是计算机学习效果在逐渐提升的缘故。只不过这样的结果我们看起来并不够直观，毕竟损失值是根据128次测试中预测正确的概率计算得来的：

看不懂公式的小伙伴也没有关系，只需要知道这个值并不能直观的反映我们模型学习后的准确率究竟是多少就好了。想要得到准确率，我们可以用这样一个函数进行计算：

correct=0
total=0
for xb,yb in valid_dl:
    outputs=model(xb) # output是128*10维的矩阵，对应最后一层神经元，输出的是十个分类的概率
    _,predicted=torch.max(outputs.data,1) # 查看10个结果中哪个结果的概率最大
                                          # 注意比较的内容是每个样本对应概率，所以第二个参数置为1而非0
                                          # 返回是最大的值和对应索引，这个任务中索引即识别结果
#     print(outputs.data)
    total+=yb.size(0)
    correct+=(predicted==yb).sum().item() # (predicted==yb).sum()：统计测试集中预测正确的总数
                                          # item()可以将结果转换成数值

print("第"+str(step)+"次学习后预测的准确率为:"+str(100*correct/total)+"%")
1
2
3
4
5
6
7
8
9
10
11
12
13

就可以获得我们能够理解的准确率信息了：

当然如果不想输出平均损失，可以把这一段代码放到fit函数里，就可以得到以下结果（本次运行中我将优化方法从SGD改为Adam）：

可以看到这样的优化方式比SGD好很多。另外也并不是每一次学习效果都会比之前好，但整体的学习效果确实会随着step的增加而变好的。下面我帮大家把本文的演示、测试代码去除掉，将分类任务的代码实现完整整理出来：

完整代码

import torch
from pathlib import Path
import requests
import pickle
import gzip
import torch.nn.functional as F
from torch.utils.data import TensorDataset,DataLoader
from torch import optim
import numpy as np

DATA_PATH = Path("data")
PATH = DATA_PATH / "mnist"
PATH.mkdir(parents=True, exist_ok=True)
URL = "http://deeplearning.net/data/mnist/"
FILENAME = "mnist.pkl.gz"
if not (PATH / FILENAME).exists():
        content = requests.get(URL + FILENAME).content
        (PATH / FILENAME).open("wb").write(content)


with gzip.open((PATH / FILENAME).as_posix(), "rb") as f:
        ((x_train, y_train), (x_valid, y_valid), _) = pickle.load(f, encoding="latin-1") # 拿出训练集和测试集

loss_func = F.cross_entropy
from torch import nn
bs=64
class Mnist_NN(nn.Module):
    def __init__(self):
        super().__init__() 
        
        self.hidden1 = nn.Linear(784, 128)
        self.hidden2 = nn.Linear(128, 256)
        self.out  = nn.Linear(256, 10)
        self.dropout = nn.Dropout(0.5)
        
    def forward(self, x):
        x = F.relu(self.hidden1(x))
        x = self.dropout(x)
        x = F.relu(self.hidden2(x)) 
        x = self.dropout(x) 
        x = self.out(x)
        return x

x_train, y_train, x_valid, y_valid = map(torch.tensor, (x_train, y_train, x_valid, y_valid))
n, c = x_train.shape

train_ds = TensorDataset(x_train, y_train)

valid_ds = TensorDataset(x_valid, y_valid)

def get_data(train_ds, valid_ds, bs):
    train_dl = DataLoader(train_ds, batch_size=bs, shuffle=True) 

    valid_dl = DataLoader(valid_ds, batch_size=bs *2) # 128道题一张卷
    return train_dl,valid_dl

def get_model():
    model = Mnist_NN()
    return model, optim.Adam(model.parameters(), lr=0.001)
                                                          

def loss_batch(model, loss_func, xb, yb, opt=None):
    loss = loss_func(model(xb), yb)

    if opt is not None:
        loss.backward()
        opt.step()
        opt.zero_grad()

    return loss.item(), len(xb)

def fit(steps, model, loss_func, opt, train_dl, valid_dl):
    for step in range(steps):
        for xb, yb in train_dl:
            loss_batch(model, loss_func, xb, yb, opt)
                                                      
        model.eval()
        with torch.no_grad():
            pass
        
        correct=0
        total=0
        for xb,yb in valid_dl:
            outputs=model(xb)
            _,predicted=torch.max(outputs.data,1)
            total+=yb.size(0)
            correct+=(predicted==yb).sum().item() 
        print("第"+str(step+1)+"次学习后预测的准确率为:"+str(100*correct/total)+"%")
train_dl, valid_dl = get_data(train_ds, valid_ds, bs)
model, opt = get_model()
step=25
fit(step, model, loss_func, opt, train_dl, valid_dl)
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

那么，本节的讲述就到这里了，下节见~