一篇文章读懂人工神经网络

1.概述

   人工神经网络

（ANN）

是通过中神经元的信息处理机制开发的解决各种问题的数学模型。神经元又称神经细胞，有细胞体和突起（树突和轴突）组成。动物大脑的神经元处理机制为

树突

将接受到的信息递给细胞体，细胞体经过分析处理通过

轴突

传递给下一个

神经元

。

   BP神经网络的结构也和神经元的结构相似，X

1

、X

2

、Xn用于接收信息相当于树突的作用。隐含层相当于细胞体，用于处理和传递信息。Y

1

、Y

2

、Y

m

是输出层。

2.原理

   一种按照误差逆向传播算法训练的

多层前馈神经网络

，至于模型的特点是信息向前传递，计算出的结果与实际产生的误差反向传递。在向前传递的过程中，输入信号从输入层输入，经过隐含层逐层处理，我们可以把隐含层理解为一个函数，通过输入成一个值的输入，这个值带入隐含层的函数中，最后算出输出层结果。每一层的神经元状态只影响下一神经元状态。如果输出的结果与实际的期望有误差，比如说我们想要他输出的结果为200，但是他计算出的结果为150，在这个时候我们就要根据它的误差调整网络的权值和阈值，从而使得这个神经网络的预测输出不断逼近希望输出。

   在这一模型中，隐含层可以有多层的，输入层通过隐含层多层的计算，最后得出输出层。我们还需要提及权值和阈值这个这两个概念。

权值

指的是我们每一个信号特征输入X

1

、X

2

、Xn，都会给他匹配权值，可以理解为占到隐含层中的一个权重。

阈值

可以理解为偏差看，这个输入值根据权值做了相应的计算，和我们实际想要的值有一个差距，通过阈值进行一个修正。

3.训练步骤

 1.网络初始化

   根据实际的条件确定隐含层的层数、

每层隐藏层的神经元个数

、

学习速率

、神经元激励函数。这个

学习速率

是指我们在每一次进行计算后得出的误差，他是使用一种

梯度下降

法，为了保证我们的这个每一次更新权值和阈值，不错过最优解，一般的取值为0~1之间的小数。

隐含层的节点数

通常需要我们通过一定的经验或者公式来尝试得出。

hidden_floors_num：隐藏层的个数
every_hidden_floor_num：每层隐藏层的神经元个数
learning_rate：学习速率
activation：激活函数
regularization：正则化方式
regularization_rate：正则化比率
total_step：总的训练次数
train_data_path：训练数据路径
model_save_path：模型保存路径

 2.隐含层输出计算

   根据输入向量X，输入层和隐藏层之间的权值。以及隐含层的阈值，计算隐含层的输出。

   这时我们需要通过激励函数，把这个输出值控制在某一个范围之间。常用的激活函数有sigmoid、tanh 。sigmoid函数把输入的结果控制在0到1的范围内。

 3.输出层输出计算

   根据隐含层的输出H，连接权值和阈值，计算BP神经网络的预测输出O。

 4.误差计算

   根据BP神经网络的预测输出O和期望输出Y，计算网络的误差e。

 5.权值更新

   根据神经网络的误差e，使用梯度下降法，更新网络的权值。

 6.阈值更新

   根据神经网络的误差e，更新网络的阈值。

 7.判断算法迭代是否完成

   判断算法迭代是否完成，若没有结束，返回步骤2。

4.案例

   

下面将通过一个案例来实现人工神经网络。该案例主要由生成数据、训练模型、预测数据三个部分组成。

 1.生成数据

# 生成测试数据
import numpy as np
import pandas as pd
# 训练集和验证集样本总个数
sample = 2000
train_data_path = 'train.csv'
validate_data_path = 'validate.csv'
predict_data_path = 'test.csv'

# 构造生成数据的模型
X1 = np.zeros((sample, 1))
X1[:, 0] = np.random.normal(1, 1, sample)
X2 = np.zeros((sample, 1))
X2[:, 0] = np.random.normal(2, 1, sample)
X3 = np.zeros((sample, 1))
X3[:, 0] = np.random.normal(3, 1, sample)

# 模型
Y = 6 * X1 - 3* X2 + X3 * X3 + np.random.normal(0, 0.1, [sample, 1])

# 将所有生成的数据放到data里面
data = np.zeros((sample, 4))
data[:, 0] = X1[:, 0]
data[:, 1] = X2[:, 0]
data[:, 2] = X3[:, 0]
data[:, 3] = Y[:, 0]

# 将data分成测试集和训练集
num_traindata = int(0.8*sample)

# 将训练数据保存
traindata = pd.DataFrame(data[0:num_traindata, :], columns=['x1', 'x2', 'x3', 'y'])
traindata.to_csv(train_data_path, index=False)
print('训练数据保存在: ', train_data_path)

# 将验证数据保存
validate_data = pd.DataFrame(data[num_traindata:, :], columns=['x1', 'x2', 'x3', 'y'])
validate_data.to_csv(validate_data_path, index=False)
print('验证数据保存在: ', validate_data_path)

# 将预测数据保存
predict_data = pd.DataFrame(data[num_traindata:, 0:-1], columns=['x1', 'x2', 'x3'])
predict_data.to_csv(predict_data_path, index=False)
print('预测数据保存在: ', predict_data_path)


   以上代码通过Y = 6 * X1 - 3* X2 + X3 * X3 + np.random.normal(0, 0.1, [sample, 1])，随机生成2000组数据。其中1600组放入train.csv，作为训练集。400组放入validate.csv，作为验证集。输入量为X1、X2、X3，输出量为Y。

 2.训练验证模型

import tensorflow as tf
import pandas as pd
import numpy as np
createVar = locals()

'''
建立一个网络结构可变的BP神经网络通用代码：

在训练时各个参数的意义：
hidden_floors_num：隐藏层的个数
every_hidden_floor_num：每层隐藏层的神经元个数
learning_rate：学习速率
activation：激活函数
regularization：正则化方式
regularization_rate：正则化比率
total_step：总的训练次数
train_data_path：训练数据路径
model_save_path：模型保存路径

利用训练好的模型对验证集进行验证时各个参数的意义：
model_save_path：模型保存路径
validate_data_path：验证集路径
precision：精度

利用训练好的模型进行预测时各个参数的意义：
model_save_path：模型的保存路径
predict_data_path：预测数据路径
predict_result_save_path：预测结果保存路径
'''


# 训练模型全局参数
hidden_floors_num = 1
every_hidden_floor_num = [50]
learning_rate = 0.00001
activation = 'tanh'
regularization = 'L1'
regularization_rate = 0.0001
total_step = 200000
train_data_path = 'train.csv'
model_save_path = 'model/predict_model'

# 利用模型对验证集进行验证返回正确率
model_save_path = 'model/predict_model'
validate_data_path = 'validate.csv'
precision = 0.5

# 利用模型进行预测全局参数
model_save_path = 'model/predict_model'
predict_data_path = 'test.csv'
predict_result_save_path = 'test_predict.csv'


def inputs(train_data_path):
 train_data = pd.read_csv(train_data_path)
 X = np.array(train_data.iloc[:, :-1])
 Y = np.array(train_data.iloc[:, -1:])
 return X, Y


def make_hidden_layer(pre_lay_num, cur_lay_num, floor):
 createVar['w' + str(floor)] = tf.Variable(tf.random_normal([pre_lay_num, cur_lay_num], stddev=1))
 createVar['b' + str(floor)] = tf.Variable(tf.random_normal([cur_lay_num], stddev=1))
 return eval('w'+str(floor)), eval('b'+str(floor))


def initial_w_and_b(all_floors_num):
 # 初始化隐藏层的w, b
 for floor in range(2, hidden_floors_num+3):
 pre_lay_num = all_floors_num[floor-2]
 cur_lay_num = all_floors_num[floor-1]
 w_floor, b_floor = make_hidden_layer(pre_lay_num, cur_lay_num, floor)
 createVar['w' + str(floor)] = w_floor
 createVar['b' + str(floor)] = b_floor


def cal_floor_output(x, floor):
 w_floor = eval('w'+str(floor))
 b_floor = eval('b'+str(floor))
 if activation == 'sigmoid':
 output = tf.sigmoid(tf.matmul(x, w_floor) + b_floor)
 if activation == 'tanh':
 output = tf.tanh(tf.matmul(x, w_floor) + b_floor)
 if activation == 'relu':
 output = tf.nn.relu(tf.matmul(x, w_floor) + b_floor)
 return output


def inference(x):
 output = x
 for floor in range(2, hidden_floors_num+2):
 output = cal_floor_output(output, floor)

 floor = hidden_floors_num+2
 w_floor = eval('w'+str(floor))
 b_floor = eval('b'+str(floor))
 output = tf.matmul(output, w_floor) + b_floor
 return output


def loss(x, y_real):
 y_pre = inference(x)
 if regularization == 'None':
 total_loss = tf.reduce_sum(tf.squared_difference(y_real, y_pre))

 if regularization == 'L1':
 total_loss = 0
 for floor in range(2, hidden_floors_num + 3):
 w_floor = eval('w' + str(floor))
 total_loss = total_loss + tf.contrib.layers.l1_regularizer(regularization_rate)(w_floor)
 total_loss = total_loss + tf.reduce_sum(tf.squared_difference(y_real, y_pre))

 if regularization == 'L2':
 total_loss = 0
 for floor in range(2, hidden_floors_num + 3):
 w_floor = eval('w' + str(floor))
 total_loss = total_loss + tf.contrib.layers.l2_regularizer(regularization_rate)(w_floor)
 total_loss = total_loss + tf.reduce_sum(tf.squared_difference(y_real, y_pre))

 return total_loss


def train(total_loss):
 train_op = tf.train.GradientDescentOptimizer(learning_rate).minimize(total_loss)
 return train_op


# 训练模型
def train_model(hidden_floors_num, every_hidden_floor_num, learning_rate, activation, regularization,
 regularization_rate, total_step, train_data_path, model_save_path):
 file_handle = open('acc.txt', mode='w')
 X, Y = inputs(train_data_path)
 X_dim = X.shape[1]
 all_floors_num = [X_dim] + every_hidden_floor_num + [1]

 # 将参数保存到和model_save_path相同的文件夹下， 恢复模型进行预测时加载这些参数创建神经网络
 temp = model_save_path.split('/')
 model_name = temp[-1]
 parameter_path = ''
 for i in range(len(temp)-1):
 parameter_path = parameter_path + temp[i] + '/'
 parameter_path = parameter_path + model_name + '_parameter.txt'
 with open(parameter_path, 'w') as f:
 f.write("all_floors_num:")
 for i in all_floors_num:
 f.write(str(i) + ' ')
 f.write('\n')
 f.write('activation:')
 f.write(str(activation))

 x = tf.placeholder(dtype=tf.float32, shape=[None, X_dim])
 y_real = tf.placeholder(dtype=tf.float32, shape=[None, 1])
 initial_w_and_b(all_floors_num)
 y_pre = inference(x)
 total_loss = loss(x, y_real)
 train_op = train(total_loss)

 # 记录在训练集上的正确率
 train_accuracy = tf.reduce_mean(tf.cast(tf.abs(y_pre - y_real) < precision, tf.float32))
 print(y_pre)
 # 保存模型
 saver = tf.train.Saver()

 # 在一个会话对象中启动数据流图，搭建流程
 sess = tf.Session()
 init = tf.global_variables_initializer()
 sess.run(init)
 for step in range(total_step):
 sess.run([train_op], feed_dict={x: X[0:, :], y_real: Y[0:, :]})
 if step % 1000 == 0:
 saver.save(sess, model_save_path)
 total_loss_value = sess.run(total_loss, feed_dict={x: X[0:, :], y_real: Y[0:, :]})
 lxacc=sess.run(train_accuracy, feed_dict={x: X, y_real: Y})
 print('train step is ', step, ', total loss value is ', total_loss_value,
 ', train_accuracy', lxacc,
 ', precision is ', precision)

 file_handle.write(str(lxacc)+&quot;\n&quot;)


 saver.save(sess, model_save_path)
 sess.close()


def validate(model_save_path, validate_data_path, precision):
 # **********************根据model_save_path推出模型参数路径, 解析出all_floors_num和activation****************
 temp = model_save_path.split('/')
 model_name = temp[-1]
 parameter_path = ''
 for i in range(len(temp)-1):
 parameter_path = parameter_path + temp[i] + '/'
 parameter_path = parameter_path + model_name + '_parameter.txt'
 with open(parameter_path, 'r') as f:
 lines = f.readlines()

 # 从读取的内容中解析all_floors_num
 temp = lines[0].split(':')[-1].split(' ')
 all_floors_num = []
 for i in range(len(temp)-1):
 all_floors_num = all_floors_num + [int(temp[i])]

 # 从读取的内容中解析activation
 activation = lines[1].split(':')[-1]
 hidden_floors_num = len(all_floors_num) - 2

 # **********************读取验证数据*************************************
 X, Y = inputs(validate_data_path)
 X_dim = X.shape[1]

 # **********************创建神经网络************************************
 x = tf.placeholder(dtype=tf.float32, shape=[None, X_dim])
 y_real = tf.placeholder(dtype=tf.float32, shape=[None, 1])
 initial_w_and_b(all_floors_num)
 y_pre = inference(x)

 # 记录在验证集上的正确率
 validate_accuracy = tf.reduce_mean(tf.cast(tf.abs(y_pre - y_real) < precision, tf.float32))

 sess = tf.Session()
 saver = tf.train.Saver()
 with tf.Session() as sess:
 # 读取模型
 try:
 saver.restore(sess, model_save_path)
 print('模型载入成功！')
 except:
 print('模型不存在，请先训练模型！')
 return
 validate_accuracy_value = sess.run(validate_accuracy, feed_dict={x: X, y_real: Y})
 print('validate_accuracy is ', validate_accuracy_value)

 return validate_accuracy_value


def predict(model_save_path, predict_data_path, predict_result_save_path):
 # **********************根据model_save_path推出模型参数路径, 解析出all_floors_num和activation****************
 temp = model_save_path.split('/')
 model_name = temp[-1]
 parameter_path = ''
 for i in range(len(temp)-1):
 parameter_path = parameter_path + temp[i] + '/'
 parameter_path = parameter_path + model_name + '_parameter.txt'
 with open(parameter_path, 'r') as f:
 lines = f.readlines()

 # 从读取的内容中解析all_floors_num
 temp = lines[0].split(':')[-1].split(' ')
 all_floors_num = []
 for i in range(len(temp)-1):
 all_floors_num = all_floors_num + [int(temp[i])]

 # 从读取的内容中解析activation
 activation = lines[1].split(':')[-1]
 hidden_floors_num = len(all_floors_num) - 2

 # **********************读取预测数据*************************************
 predict_data = pd.read_csv(predict_data_path)
 X = np.array(predict_data.iloc[:, :])
 X_dim = X.shape[1]

 # **********************创建神经网络************************************
 x = tf.placeholder(dtype=tf.float32, shape=[None, X_dim])
 initial_w_and_b(all_floors_num)
 y_pre = inference(x)

 sess = tf.Session()
 saver = tf.train.Saver()
 with tf.Session() as sess:
 # 读取模型
 try:
 saver.restore(sess, model_save_path)
 print('模型载入成功！')
 except:
 print('模型不存在，请先训练模型！')
 return
 y_pre_value = sess.run(y_pre, feed_dict={x: X[0:, :]})

 # 将预测结果写入csv文件
 predict_data_columns = list(predict_data.columns) + ['predict']
 data = np.column_stack([X, y_pre_value])
 result = pd.DataFrame(data, columns=predict_data_columns)
 result.to_csv(predict_result_save_path, index=False)
 print('预测结果保存在：', predict_result_save_path)


if __name__ == '__main__':
 mode = &quot;train&quot;

 if mode == 'train':
 # 训练模型
 train_model(hidden_floors_num, every_hidden_floor_num, learning_rate, activation, regularization,
 regularization_rate, total_step, train_data_path, model_save_path)

 if mode == 'validate':
 # 利用模型对验证集进行正确性测试
 validate(model_save_path, validate_data_path, precision)

 if mode == 'predict':
 # 利用模型进行预测
 predict(model_save_path, predict_data_path, predict_result_save_path)

   训练时需要调节的参数有隐藏层的个数

hidden_floors_num

、每层隐藏层的神经元个数

every_hidden_floor_num

、学习速率

learning_rate

、激活函数

activation

、总的训练次数

total_step

。

   我们使用

阿里云天池实验室PAI DSW

进行模型训练以及验证，总的训练次数设为100000，神经元个数设为10。

模型收敛慢

预测准确率低

   

将总的训练次数设为100000，神经元个数设为50。

模型快速收敛

预测准确率高

   我们可以通过调节

隐藏层的个数

、

每层隐藏层的神经元个数

、

学习速率

、

激活函数

、

总的训练次数

等参数提高模型的准确率。

 3.预测数据

   最后使用训练好的模型predict_model对数据进行预测。