[关闭]
@Pigmon 2017-11-29T12:11:00.000000Z 字数 10702 阅读 982

BP 前馈神经网络简易实现

教案

neuron.py

  1. # -*- coding: utf-8 -*-
  2. """
  3. Created on Sun Apr 09 17:25:53 2017
  5. @author: Yuan Sheng
  6. """
  8. #import random
  9. import numpy as np
  10. import random as rnd
  12. def sigmoid(x):
  13.     return 1. / (1. + np.e ** (-x))
  15. class Neuron:
  16.     "神经元基类"
  18.     # layer: 所在层，第一个输入层为0
  19.     # index: 在当前层的位置，最上面为0
  20.     # gid: 全局位置，从 0 开始
  21.     #   整个网络输入在左，输出在右
  22.     #   从输入神经元算起，按左上到右下依次排列
  23.     #         2
  24.     #   ->0       5   7->
  25.     #         3
  26.     #   ->1       6    8->
  27.     #         4
  28.     layer, index, gid = 0, 0, 0
  29.     # output: 输出值
  30.     # error: 误差项
  31.     output, error = 0.0, 0.0
  33.     def __init__ (self, _layer, _index, _global_index):
  34.         self.layer, self.index, self.gid = _layer, _index, _global_index
  35.         self.bias = rnd.uniform(-1.0, 1.0)
  37.     def sigmoid_derive(self):
  38.         return self.output * (1.0 - self.output)
  40.     def count_out(self, _in_w, _in_vec):
  41.         "计算输出值"       
  42.         if len(_in_w) != len(_in_vec):
  43.             return 0.0
  45.         self.output = sigmoid(self.bias + np.dot(_in_w, _in_vec))        
  46.         return self.output
  48.     def count_error(self, _out_w, _next_layer_error):
  49.         "计算误差项"
  50.         if len(_out_w) != len(_next_layer_error):
  51.             return 0.0
  53.         self.error = self.sigmoid_derive() * np.dot(_out_w, _next_layer_error)
  54.         return self.error
  56.     def update_bais(self, _eta):
  57.         "调整 bias 值"
  58.         self.bias -= _eta * self.error
  60.     def debug(self):
  61.         print ('%d: <%d, %d> b=%f' % (self.gid, self.layer, self.index, self.bias))
  63. class OutputNeuron(Neuron):
  64.     "输出层神经元"
  65.     def __init__ (self, _layer, _index, _global_index):
  66.         Neuron.__init__(self, _layer, _index, _global_index)
  68.     def count_error(self, _label):
  69.         self.error = self.sigmoid_derive() * (self.output - _label)
  70.         #self.error = self.sigmoid_derive() * (_label - self.output)
  71.         return self.error
  73. if __name__ == '__main__':
  74.     nr = Neuron(0, 2, 1)
  75.     nr.count_out(np.array([0.1, 0.2, 0.3]), np.array([3.0, 2.0, 1.0]))
  76.     #nr.Count_Error(np.array([1.0, 2.0]), np.array([2.0, 1.0]))
  77.     #nr.Update_Bais(1.0)
  78.     print (nr.output)
  80.     #onr = OutputNeuron(3, 0)
  81.     #label = 10.0
  83.     #onr.count_error(label)
  84.     #print onr.error

bpnn.py

  1. # -*- coding: utf-8 -*-
  2. """
  3. Created on Sun Apr 09 20:50:49 2017
  5. @author: Yuan Sheng
  6. """
  8. import numpy as np
  9. import random as rnd
  10. from neuron import Neuron, OutputNeuron
  12. class BPNN:
  13.     "BP前馈神经网络"
  15.     # @input_vec_len: 输入层维数
  16.     # @output_vec_len: 输出层维数
  17.     # @hidden_layer_count: 隐藏层数
  18.     # @eta: 学习率，步长
  19.     # @threshold: 全局均方差的阈值
  20.     input_vec_len, output_vec_len, hidden_layer_count, eta, threshold = 2, 1, 0, 0.2, 0.1
  22.     # @hidden_layers_len： 每个隐藏层维数
  23.     # @hidden_neurons：隐藏神经元二维数组
  24.     # @output_neurons：输出神经元一维数组
  25.     # @weights：权重三维数组
  26.     hidden_layers_len, hidden_neurons, output_neurons, weights = [], [], [], []
  28.     # 全局位置到(layer, index)的映射字典; 当前的输出值
  29.     gid_dict = {}
  31.     # -----------------输出变量-----------------
  32.     _correct_rate, _output_vec = 0.0, []
  34.     # -------------------------------------------------------------------------
  35.     def __init__ (self, _in_len, _out_len, _hiddens_len, _learning_rate):
  36.         self.input_vec_len, self.output_vec_len, self.hidden_layer_count, self.hidden_layers_len, self.eta = \
  37.             _in_len, _out_len, len(_hiddens_len), _hiddens_len, _learning_rate
  39.         # 生成隐藏层 Nerons
  40.         cnter = _in_len
  41.         for i in range(0, len(_hiddens_len)):
  42.             cnt, layer_list = _hiddens_len[i], []
  43.             for j in range(0, cnt):
  44.                 layer_list.append(Neuron(i, j, cnter))
  45.                 self.gid_dict[cnter] = (i, j)
  46.                 cnter += 1
  47.             self.hidden_neurons.append(layer_list)
  49.         # 生成输出层 Neurons
  50.         for i in range(0, _out_len):
  51.             self.output_neurons.append(OutputNeuron(self.hidden_layer_count, i, cnter))
  52.             self.gid_dict[cnter] = (self.hidden_layer_count, i)
  53.             cnter += 1
  55.         # 生成权重，例如：
  56.         # 形如下面这样的输入层2，隐藏层为(3, 2)，输出层为2:
  57.         #
  58.         #         2
  59.         #   ->0       5   7->
  60.         #         3
  61.         #   ->1       6    8->
  62.         #         4
  63.         #
  64.         # 权重数组如下所示：
  65.         # [
  66.         #     [ [w_0_2, w_1_2], [w_0_3, w_1_3], [w_0_4, w_1_4] ],
  67.         #     [ [w_2_5, w_3_5, w_4_5], [w_2_6, w_3_6, w_4_6] ],
  68.         #     [ [w_5_7, w_6_7], [w_5_8, w_6_8] ]
  69.         # ]
  70.         #
  71.         # 可以在循环最内部，用pair.append(self.debug_weight_index(cnt_list, i, j, k))调试
  72.         #
  73.         cnt_list = np.append(np.array([_in_len]), np.append(_hiddens_len, _out_len))
  74.         #print cnt_list
  75.         for i in range(0, len(cnt_list)-1):
  76.             layer1_cnt, layer2_cnt, layer_weight = cnt_list[i], cnt_list[i + 1], []
  78.             for j in range(0, layer2_cnt):
  79.                 pair = []
  80.                 for k in range(0, layer1_cnt):
  81.                     pair.append(rnd.uniform(-1.0, 1.0))
  82.                     #pair.append(self.debug_weight_index(cnt_list, i, j, k))
  83.                 layer_weight.append(pair)
  85.             self.weights.append(layer_weight)            
  87.     # -------------------------------------------------------------------------        
  88.     def input_weights_of(self, layer, index):
  89.         """
  90.         得到在第layer层的第index个神经元的输入权重（左侧）列表。
  91.         layer 0 代表第一个隐藏层，index 0代表最上面的神经元
  92.         """
  93.         return self.weights[layer][index]
  95.     # -------------------------------------------------------------------------
  96.     def input_weights_of_gid(self, _gid):
  97.         "通过gid得到输入权重列表"
  98.         return self.input_weights_of(self.gid_dict[_gid][0], self.gid_dict[_gid][1])
  100.     # -------------------------------------------------------------------------    
  101.     def output_weights_of(self, layer, index):
  102.         """
  103.         得到在第layer层的第index个神经元的输出权重（右侧）列表。
  104.         layer 0 代表第一个隐藏层，index 0代表最上面的神经元
  105.         """        
  106.         # 如果是输出层的神经元，返回空数组
  107.         if layer == self.hidden_layer_count:
  108.             return []
  110.         next_layer_weights, ret = self.weights[layer + 1], []
  111.         for group in next_layer_weights:
  112.             ret.append(group[index])
  114.         return ret
  116.     # -------------------------------------------------------------------------        
  117.     def output_weights_of_gid(self, _gid):
  118.         "通过gid得到输出权重列表"
  119.         return self.output_weights_of(self.gid_dict[_gid][0], self.gid_dict[_gid][1])
  121.     # -------------------------------------------------------------------------        
  122.     def pre_layer_outs(self, layer, input_vec):
  123.         """
  124.         得到在第layer层的第index个神经元的<前一层>神经元的值的列表。
  125.         layer 0 代表第一个隐藏层，index 0代表最上面的神经元
  126.         """
  127.         if layer == 0:
  128.             return input_vec
  129.         else:
  130.             pre_layer_neurons, ret = self.hidden_neurons[layer - 1], []
  131.             for neuron in pre_layer_neurons:
  132.                 ret.append(neuron.output)
  133.             return ret
  135.     # -------------------------------------------------------------------------            
  136.     def next_layer_errors(self, layer):
  137.         "得到第layer+1层的神经元的误差项列表"
  139.         # 如果是输出层的神经元，返回空数组
  140.         if layer >= self.hidden_layer_count:
  141.             return []
  143.         next_layer_neurons, ret = [], []
  145.         if layer == self.hidden_layer_count - 1:
  146.             next_layer_neurons = self.output_neurons
  147.         else:
  148.             next_layer_neurons = self.hidden_neurons[layer + 1]
  150.         for neuron in next_layer_neurons:
  151.             ret.append(neuron.error)
  153.         return ret
  155.     # -------------------------------------------------------------------------        
  156.     def forward(self, input_vec):
  157.         "前向传播计算值的过程"
  158.         assert len(input_vec) == self.input_vec_len
  159.         self._output_vec = []
  161.         for i in range(0, len(self.hidden_neurons)):
  162.             hidden_layer = self.hidden_neurons[i]
  163.             for j in range(0, len(hidden_layer)):
  164.                 neuron, input_values, input_weights = \
  165.                     hidden_layer[j], self.pre_layer_outs(i, input_vec), self.input_weights_of(i, j)
  166.                 neuron.count_out(input_weights, input_values)
  168.         for j in range(0, len(self.output_neurons)):
  169.             neuron, input_values, input_weights = self.output_neurons[j], \
  170.                 self.pre_layer_outs(self.hidden_layer_count, input_vec), \
  171.                 self.input_weights_of(self.hidden_layer_count, j)
  173.             neuron.count_out(input_weights, input_values)
  174.             self._output_vec.append(neuron.output)
  176.         return self._output_vec
  178.     # -------------------------------------------------------------------------            
  179.     def backward(self, _result_vec, _label_vec):
  180.         "后向传播误差项"
  181.         assert len(_result_vec) == len(_label_vec)
  183.         # 计算输出层误差项
  184.         for j in range(0, len(self.output_neurons)):
  185.             self.output_neurons[j].count_error(_label_vec[j])
  187.         # 从后向前计算隐藏层误差项
  188.         arr, length = self.hidden_neurons[::-1], len(self.hidden_neurons)
  189.         for i in range(0, length):
  190.             hidden_layer, layer_idx = arr[i], length - i - 1
  191.             for j in range(0, len(hidden_layer)):
  192.                 neuron, output_weights, next_layer_errors = \
  193.                     hidden_layer[j], self.output_weights_of(layer_idx, j), self.next_layer_errors(layer_idx)
  194.                 neuron.count_error(output_weights, next_layer_errors)
  196.     # -------------------------------------------------------------------------                
  197.     def update_param(self, _inpput_vec):
  198.         "调整误差和 bias 的值，可以放到 backward 函数里，单独写出来是为了看起来清晰"
  200.         # 调整输出层的输入权重和 bias
  201.         for j in range(0, len(self.output_neurons)):
  202.             neuron, input_values, input_weights = self.output_neurons[j], \
  203.                 self.pre_layer_outs(self.hidden_layer_count, []), self.input_weights_of(self.hidden_layer_count, j)
  205.             neuron.update_bais(self.eta)
  206.             for i in range(0, len(input_weights)):
  207.                 input_weights[i] -= self.eta * neuron.error * input_values[i]
  208.             # 更新 self.weights
  209.             self.weights[self.hidden_layer_count][j] = input_weights
  211.         # 调整隐藏层的输入权重和 bias
  212.         for i in range(0, len(self.hidden_neurons)):
  213.             hidden_layer = self.hidden_neurons[i]
  214.             for j in range(0, len(hidden_layer)):
  215.                 neuron, input_values, input_weights =\
  216.                     hidden_layer[j], self.pre_layer_outs(i, _inpput_vec), self.input_weights_of(i, j)
  218.                 neuron.update_bais(self.eta)
  219.                 for k in range(0, len(input_weights)):
  220.                     input_weights[k] -= self.eta * neuron.error * input_values[k]
  221.                 # 更新 self.weights
  222.                 self.weights[i][j] = input_weights
  225.     # -------------------------------------------------------------------------
  226.     @staticmethod
  227.     def correct_rate(_vec1, _vec2):
  228.         "正确率"
  229.         assert len(_vec1) == len(_vec2)
  231.         counter = 0        
  232.         for i in range(0, len(_vec1)):
  233.             if abs(_vec1[i] - _vec2[i]) < 0.5:
  234.                 counter += 1
  236.         return float(counter) / float(len(_vec1))
  238.     # -------------------------------------------------------------------------
  239.     def total_error_var(self, _train_set):
  240.         total = 0
  242.         for smpl in _train_set:
  243.             total += (self.predict_labeled([smpl])[0] - smpl[1]) ** 2
  245.         return total / len(_train_set)
  247.     # -------------------------------------------------------------------------
  248.     def predict_labeled(self, _test_set):
  249.         "_test_set 每个元素，传入之前需要格式化成 [[x1, x2, ...], label]"
  251.         assert len(_test_set) > 0 and len(_test_set[0]) == 2
  253.         ret_vec, label_vec = [], []
  254.         for sample in _test_set:
  255.             ret_vec.append(self.forward(sample[0])[0])
  256.             label_vec.append(sample[1])
  258.         self._correct_rate = BPNN.correct_rate(ret_vec, label_vec)
  260.         return ret_vec
  262.     # -------------------------------------------------------------------------
  263.     @staticmethod
  264.     def sigsign(x):
  265.         "目前只测试2分类问题，输入标签只有0和1"
  266.         if x >= 0.5:
  267.             return 1
  268.         else:
  269.             return 0
  271.     # -------------------------------------------------------------------------
  272.     def predict(self, _attr_set):
  273.         "只包含特征的测试集，[[x1, x2, ...], [x1, x2,...], ...]"
  275.         ret_vec = []        
  276.         for point in _attr_set:
  277.             ret_vec.append(self.forward(point))
  278.             #ret_vec.append(BPNN.sigsign(self.forward(point)[0]))
  280.         return ret_vec
  282.     # -------------------------------------------------------------------------    
  283.     def fit(self, _train_set, max_epochs = 20000):
  284.         "_train_set 每个元素，传入之前需要格式化成 [[x1, x2, ...], label]"
  286.         assert len(_train_set) > 0 and len(_train_set[0]) == 2        
  287.         counter, go_through = 0, False
  289.         while (counter <= max_epochs):
  290.             # 退出循环的条件（temp）：所有训练样本方差小于阈值
  291.             if self.total_error_var(_train_set) <= self.threshold:
  292.                 go_through = True
  293.                 break
  295.             # 1. 随机选取一个测试样本
  296.             sample = rnd.choice(_train_set)
  297.             input_vec, label = sample[0], sample[1] 
  298.             # 2. 前向计算值
  299.             y_hat = self.forward(input_vec)
  300.             # 3. 反向传播误差
  301.             self.backward([y_hat], [label])
  302.             # 4. 更新参数
  303.             self.update_param(input_vec)
  305.             counter += 1                
  307.         print (counter)
  308.         return go_through
  310.     # -------------------------------------------------------------------------
  311.     # Debug Functions
  312.     # -------------------------------------------------------------------------
  313.     def debug(self):
  314.         "输出 debug 信息"
  316.         print ('Hidden Layers Length: ' + str(nn.hidden_layers_len))
  317.         print ('Hidden Neurons: ')
  319.         for layer_neurons in self.hidden_neurons:
  320.             for neuron in layer_neurons:
  321.                 print (neuron.debug())
  323.         for neuron in self.output_neurons:
  324.             print (neuron.debug())
  326.         print (self.weights)
  328.     # -------------------------------------------------------------------------       
  329.     def debug_weight_index(self, cnt_list, layer, next_layer_index, current_layer_index):
  330.         """
  331.         测试用，生成权重下标。如连接3和8号神经元的权重返回 'w_3_8'
  332.         layer 是从输入层开始为0， 依次+1
  333.         调用方式：
  334.         在构造函数生成weights的循环最内部 pair.append(self.debug_weight_index(cnt_list, i, j, k))
  335.         """
  337.         # 当前层起始位置
  338.         current_layer_start_index = 0 
  339.         for i in range(0, layer):        
  340.             current_layer_start_index += cnt_list[i]
  341.         # 下一层的起始位置
  342.         next_layer_start_index = current_layer_start_index + cnt_list[layer]
  344.         idx1, idx2 = current_layer_start_index + current_layer_index, \
  345.             next_layer_start_index + next_layer_index
  347.         #return 'w_%d_%d' % (idx1, idx2)
  348.         return float(idx1) + idx2 / 10.0 # temp test
  350. if __name__ == '__main__':
  351.     X = [ [[1, 1], 0], [[3, 3], 1], [[3, 4], 1] ]
  352.     T = [[0, 0], [0, 1], [4, 3], [5, 5]] # 期望输出：[0, 0, 1, 1]
  353.     nn = BPNN(2, 1, [3], 4)
  355.     #X = [ [[0, 0], 1], [[1, 1], 1], [[0, 1], 0], [[1, 0], 0] ]
  356.     #T = [[0, 0], [1, 1], [0, 1], [1, 0]]
  357.     #nn = BPNN(2, 1, [2], 0.7)
  359.     # -----test-----
  360.     print (nn.fit(X))
  361.     #print (nn._correct_rate)
  363.     print (nn.predict(T))
  364.     #nn.debug()
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注