神经网络-bp代码详解

发表于 | 更新于 | 分类于 ， | 阅读次数：

一、说明

代码下载：https://github.com/MichalDanielDobrzanski/DeepLearningPython35
环境： python 3.*
文件：network.py
原理：参考上一篇文章

二、代码中文注释

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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# -*- coding: utf-8 -*-
# @Time    : 2019/4/17 22:32
# @Author  : zwenc
# @File    : a.py
# %load network.py
import random
import numpy as np

class Network(object):

    def __init__(self, sizes):
        self.num_layers = len(sizes) # 记录网络层数
        self.sizes = sizes           # 记录网络层，里面还有每一层神经元的数量
        self.biases = [np.random.randn(y, 1) for y in sizes[1:]] # 高斯随机分配偏执
        self.weights = [np.random.randn(y, x)                     # 高斯随机分配权值
                        for x, y in zip(sizes[:-1], sizes[1:])]
    
    def feedforward(self, a):
        """
        sigmoid神经元函数
        :param a: 神经元的输入向量
        :return a: 神经元的输出向量
        """
        for b, w in zip(self.biases, self.weights):
            a = sigmoid(np.dot(w, a)+b)
        return a
    
    def SGD(self, training_data, epochs, mini_batch_size, eta,
            test_data=None):
        """
        :param training_data: 训练 数据输入，数据包括输入数据x,预定输出数据y。并且有很多对
        :param epochs: 全数据迭代次数
        :param mini_batch_size: 每批处理数据的速率
        :param eta: 学习户数
        :param test_data: 测试数据，数据包括输入数据x,预定输出数据y。并且有很多对
        :return:没有返回值
        """
        training_data = list(training_data)  # 开辟一个新的空间幅值training_data，防止修改原数据
        n = len(training_data) # 计算有多少对数据
    
        if test_data: # 判断时候有测试数据
            test_data = list(test_data)  # 开辟新的空间储存测试数据
            n_test = len(test_data)      # 计算测试户数长度
    
        for j in range(epochs):          # 控制迭代次数
            random.shuffle(training_data)  # 随机打乱测试数据
            mini_batches = [               # 按照每次处理的数据数量进行分块
                training_data[k:k+mini_batch_size]
                for k in range(0, n, mini_batch_size)]
            for mini_batch in mini_batches:   # 对分好的快进行处理，每次处理将会更新一次偏置和权值
                self.update_mini_batch(mini_batch, eta)  # 训练主体函数
            if test_data:
                print("Epoch {} : {} / {}".format(j,self.evaluate(test_data),n_test)) # 测试训练结果
            else:
                print("Epoch {} complete".format(j))
    
    def update_mini_batch(self, mini_batch, eta):
        """
        :param mini_batch: 一次训练的数据，数据包括输入数据x,预定输出数据y。并且有很多对
        :param eta: 学习速率
        :return:
        """
        nabla_b = [np.zeros(b.shape) for b in self.biases] # 开辟新的空间储存偏置
        nabla_w = [np.zeros(w.shape) for w in self.weights] # 开辟新的空间储存权值
        for x, y in mini_batch: #处理每一个数据
            # 通过bp算法求得代价函数对b,w的偏导，输出为两者的偏导。其中\delta 和\nabla符号是有意义的，具体百度
            delta_nabla_b, delta_nabla_w = self.backprop(x, y)
    
            # 将 得到的b,w偏导，累加储存起来，后面做个平均
            nabla_b = [nb+dnb for nb, dnb in zip(nabla_b, delta_nabla_b)]
            nabla_w = [nw+dnw for nw, dnw in zip(nabla_w, delta_nabla_w)]
    
        # 对，就是在这里做个平均，再根据学习速率计算出新的w,b
        self.weights = [w-(eta/len(mini_batch))*nw
                        for w, nw in zip(self.weights, nabla_w)]
        self.biases = [b-(eta/len(mini_batch))*nb
                       for b, nb in zip(self.biases, nabla_b)]
    
    def backprop(self, x, y):
        """
        :param x: 输入训练数据
        :param y: 正确是数据
        :return:NULL
        """
        nabla_b = [np.zeros(b.shape) for b in self.biases] # 开辟新的b,w储存空间，类似于new一个空间
        nabla_w = [np.zeros(w.shape) for w in self.weights]
        # feedforward
        activation = x    #做一个变量存，先暂时保存了x的值
        activations = [x] #这是一个list，先保存了x的，放在最前面
        zs = []           #这是一个list，用来保存s型神经元计算后的值
    
        # 正序输出
        for b, w in zip(self.biases, self.weights):
            # w是一个二维向量,每一维表示一个目标神经元，activation一维向量（动态变化）,dot表示向量乘
            z = np.dot(w, activation)+b
            # z保存到zs之中,后面bp要用到
            zs.append(z)
            # s型神经元的输出
            activation = sigmoid(z)
            # a保存到activations中，后面bp要用到
            activations.append(activation)
    
        #得到输出层的\delta（latex语法）函数，\delta是误差的意思，参考：https://www.zwenc.cn/a0c0c739/
        #activations[-1]表示输出向量，向量对应元素相乘，一定要是np.array创建的数组才可以这么乘
        delta = self.cost_derivative(activations[-1], y) * \
            sigmoid_prime(zs[-1])
        #只要有\delta函数，根据4大公式后两个，直接得到b
        nabla_b[-1] = delta
        #只要有\delta函数，根据4大公式后两个，直接得到w
        nabla_w[-1] = np.dot(delta, activations[-2].transpose())
    
        #反向传播开始，主要使用到4大公式中的第二个
        for l in range(2, self.num_layers):
            #后面表示为当前层每个神经元的输入，是一个向量。在第一次循环里表示倒数第二层
            z = zs[-l]
            #查看该函数的注释有说明
            sp = sigmoid_prime(z)
            # 使用反向传播规则，在第一次循环中推导倒数第二层的误差
            delta = np.dot(self.weights[-l+1].transpose(), delta) * sp
            #只要有\delta函数，根据4大公式后两个，直接得到b
            nabla_b[-l] = delta
            #只要有\delta函数，根据4大公式后两个，直接得到w
            nabla_w[-l] = np.dot(delta, activations[-l-1].transpose())
        return (nabla_b, nabla_w)
    
    def evaluate(self, test_data):
        # argmax 返回最大值的索引，最大值为对应数字
        test_results = [(np.argmax(self.feedforward(x)), y)
                        for (x, y) in test_data]
        # 统计匹配正确的数量
        return sum(int(x == y) for (x, y) in test_results)
    
    def cost_derivative(self, output_activations, y):
        return (output_activations-y)

# 定义S型函数
def sigmoid(z):
    # 1.0而不是1，就很有讲究，表示计算方式为浮点型
    return 1.0/(1.0+np.exp(-z))

# S型函数对z求导
def sigmoid_prime(z):
    """Derivative of the sigmoid function."""
    return sigmoid(z)*(1-sigmoid(z))

三、测试代码

这里有三个函数是用来解压缩提取数据的，不用管它，直接看最下面三行的执行数据

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
# -*- coding: utf-8 -*-
# @Time    : 2019/4/20 23:09
# @Author  : zwenc
# @File    : test.py
import numpy as np
import gzip
import pickle
import Network

    # 这里有三个函数是用来解压缩提取数据的，不用管它，直接看最下面三行的执行数据def load_data():
    f = gzip.open('mnist.pkl.gz', 'rb')
    training_data, validation_data, test_data = pickle.load(f, encoding="latin1")
    f.close()
    return (training_data, validation_data, test_data)
    
def vectorized_result(j):    
    e = np.zeros((10, 1))
    e[j] = 1.0
    return e
    
def load_data_wrapper():
    tr_d, va_d, te_d = load_data()
    training_inputs = [np.reshape(x, (784, 1)) for x in tr_d[0]]
    training_results = [vectorized_result(y) for y in tr_d[1]]
    training_data = zip(training_inputs, training_results)
    validation_inputs = [np.reshape(x, (784, 1)) for x in va_d[0]]
    validation_data = zip(validation_inputs, va_d[1])
    test_inputs = [np.reshape(x, (784, 1)) for x in te_d[0]]
    test_data = zip(test_inputs, te_d[1])
    return (training_data, validation_data, test_data)
    
if __name__ == "__main__":
    training_data, validation_data, test_data = load_data_wrapper()
    net = Network.Network([784,30,10])
    net.SGD(training_data, 30, 10, 3.0, test_data=test_data)

四、输出

从输出可以明显看到最终的匹配准确率在95%左右，在迭代18次后开始达到稳定。

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
Epoch 0 : 8156 / 10000
Epoch 1 : 8352 / 10000
Epoch 2 : 9309 / 10000
Epoch 3 : 9366 / 10000
Epoch 4 : 9316 / 10000
Epoch 5 : 9414 / 10000
Epoch 6 : 9440 / 10000
Epoch 7 : 9424 / 10000
Epoch 8 : 9438 / 10000
Epoch 9 : 9460 / 10000
Epoch 10 : 9438 / 10000
Epoch 11 : 9448 / 10000
Epoch 12 : 9478 / 10000
Epoch 13 : 9478 / 10000
Epoch 14 : 9478 / 10000
Epoch 15 : 9461 / 10000
Epoch 16 : 9491 / 10000
Epoch 17 : 9509 / 10000
Epoch 18 : 9507 / 10000
Epoch 19 : 9471 / 10000
Epoch 20 : 9493 / 10000
Epoch 21 : 9476 / 10000
Epoch 22 : 9508 / 10000
Epoch 23 : 9490 / 10000
Epoch 24 : 9472 / 10000
Epoch 25 : 9496 / 10000
Epoch 26 : 9475 / 10000
Epoch 27 : 9504 / 10000
Epoch 28 : 9505 / 10000
Epoch 29 : 9511 / 10000

-------------本文结束感谢您的阅读-------------
0%