【深度學習】paddlepaddle——基於卷積神經網絡的手寫字識別案例

  1 # 1、導包
  2 import paddle.fluid as fluid
  3 import paddle
  4 import time
  5 
  6 start = time.time()
  7 
  8 
  9 def test_program(exe, feeder, program, fetch_list, reader):
 10     """
 11     測試進程
 12     :param exe:執行器
 13     :param feeder: 數據與網絡關系
 14     :param program: 測試主進程
 15     :param fetch_list: 需要執行之后返回的損失與准確率
 16     :param reader: 測試reader
 17     :return:
 18     """
 19     # 訓練次數
 20     count = 0
 21     # 整個測試集的總損失
 22     sum_loss = 0
 23     # 整個訓練集的准確率
 24     sum_acc = 0
 25     for test_data in reader():
 26         test_avg_loss_value, test_acc_values = exe.run(
 27             program=program,  # 測試主進程
 28             feed=feeder.feed(test_data),  # 給測試喂數據
 29             fetch_list=fetch_list  # 需要執行之后返回的值
 30         )
 31 
 32         sum_loss += test_avg_loss_value
 33         sum_acc += test_acc_values
 34         count += 1
 35     # 得到整個訓練集的平均損失，與整個訓練集的准確率
 36     test_avg_loss = sum_loss / count
 37     test_acc = sum_acc / count
 38 
 39     return test_avg_loss, test_acc
 40 
 41 
 42 # 2、數據處理---paddlepaddle 自帶的mnist數據已經經過了數據處理
 43 
 44 # 3、定義reader
 45 # paddlepaddle給我們已經定義好了reader，只需要去調用
 46 
 47 # 4、指定訓練場所
 48 place = fluid.CPUPlace()
 49 
 50 # 5、配置網絡
 51 # 特征數據層
 52 image = fluid.layers.data(name="image", shape=[1, 28, 28], append_batch_size=True, dtype="float64")
 53 # 目標數據層
 54 label = fluid.layers.data(name="label", shape=[1], append_batch_size=True, dtype="int64")
 55 # 設計兩個卷積、激活、池化 之后 + fc的卷積神經網絡
 56 conv1 = fluid.nets.simple_img_conv_pool(
 57     input=image,  # 輸入
 58     num_filters=20,  # 卷積核個數
 59     filter_size=3,  # 卷積核大小
 60     pool_size=2,  # 池化大小
 61     pool_stride=2,  # 池化步長
 62     act="relu",  # 激活函數
 63 )
 64 conv2 = fluid.nets.simple_img_conv_pool(
 65     input=conv1,  # 輸入
 66     num_filters=10,  # 卷積核個數
 67     filter_size=5,  # 卷積核大小
 68     pool_size=2,  # 池化大小
 69     pool_stride=2,  # 池化步長
 70     act="relu",  # 激活函數
 71 )
 72 y_predict = fluid.layers.fc(input=conv2, size=10, act="softmax", name="output_layer")
 73 
 74 # 6、損失
 75 # 交叉熵損失
 76 loss = fluid.layers.cross_entropy(input=y_predict, label=label)
 77 # 計算平均損失
 78 avg_loss = fluid.layers.mean(loss)
 79 
 80 # 計算准確率
 81 acc = fluid.layers.accuracy(input=y_predict, label=label)
 82 
 83 # 7、指定優化---sgd隨機梯度下降優化算法
 84 sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.1)
 85 # 指定去優化損失
 86 sgd_optimizer.minimize(avg_loss)
 87 
 88 # 8、指定網絡與數據層的關系
 89 feeder = fluid.DataFeeder(feed_list=[image, label], place=place)
 90 
 91 # 9、構建執行器
 92 # 訓練執行器
 93 exe_train = fluid.Executor(place=place)
 94 # 測試執行器
 95 exe_test = fluid.Executor(place=place)
 96 
 97 # 10、初始化網絡參數
 98 # 初始化參數進程
 99 startup_program = fluid.default_startup_program()
100 exe_train.run(startup_program)
101 # 主進程
102 # 訓練主進程
103 train_main_program = fluid.default_main_program()
104 # 測試主進程
105 test_main_program = train_main_program.clone(for_test=True)
106 
107 # 11、獲取圖片數據
108 # 並不是直接拿到數據就往網絡里面送
109 # 構建一個緩沖區，--打亂順序，--再往網絡里面送
110 # paddle.dataset.mnist.train() ----paddlepaddle的訓練reader
111 # 緩沖區大小buf_size與批次大小batch_size 並沒有多大的關系
112 # 一般設計的時候注意：buf_size 略微需要比batch_size 大一點就可以
113 # 而且batch_size 不能過大
114 # 訓練reader 與測試reader 的batch_size數量必須一致
115 train_reader = paddle.batch(
116     paddle.reader.shuffle(paddle.dataset.mnist.train(), buf_size=50),
117     batch_size=10
118 )
119 test_reader = paddle.batch(
120     paddle.reader.shuffle(paddle.dataset.mnist.test(), buf_size=50),
121     batch_size=10
122 )
123 
124 # 12、訓練
125 # 指定訓練輪數
126 loop_num = 2
127 # 定義的執行次數
128 step = 0
129 
130 flag = False
131 
132 for loop in range(loop_num):
133     print("第%d輪訓練" % loop)
134     # train_data 每批次的數據
135     for train_data in train_reader():
136         # 執行器運行訓練主進程
137         train_avg_loss_value, train_acc_value = exe_train.run(
138             program=train_main_program,  # 訓練主進程
139             feed=feeder.feed(train_data),  # 利用數據層與網絡構建好的關系，將真實的數據喂給網絡
140             fetch_list=[avg_loss, acc]  # 執行之后需要返回的結果的值
141         )
142         # 每隔10步來打印一下損失與准確率
143         if step % 10 == 0 and step != 0:
144             print("第%d次訓練的損失為%f,准確率為%f" % (step, train_avg_loss_value, train_acc_value))
145 
146         step += 1
147 
148         # 每隔100步 去測試集中測試一下訓練效果
149         if step % 100 == 0 and step != 0:
150             test_avg_loss, test_acc = test_program(exe_test,
151                                                    feeder,
152                                                    test_main_program,
153                                                    fetch_list=[avg_loss, acc],
154                                                    reader=test_reader
155                                                    )
156             print("*" * 100)
157             print("測試集的損失為：%f,准確率為：%f" % (test_avg_loss, test_acc))
158             print("*" * 100)
159             if test_avg_loss <= 0.1 and test_acc >= 0.98:
160                 flag = True
161 
162                 print("最終測試集的損失為：%f,准確率為：%f" % (test_avg_loss, test_acc))
163                 end = time.time()
164 
165                 print("運行總時長為：", end - start)
166                 break
167     if flag:
168         break