[tensorflow] 线性回归模型实现

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

在这一篇博客中大概讲一下用tensorflow如何实现一个简单的线性回归模型，其中就可能涉及到一些tensorflow的基本概念和操作，然后因为我只是入门了点tensorflow，所以我只能对部分代码给出相关的tensorflow的概念。

线性回归模型的表达式如下：

其中，是权重，是偏置，和则是输入数据和对应的模型预测值。

在tensorflow中，是用图来表示计算的形式的，图中的每个节点称为一个op(即operation)，每个operation获得相关张量(Tensor)后进行数值计算，每个张量就是一个类型化的多维数组，用图表示了计算的形式后并不会计算和得到结果，需要在会话(session)中执行(run)才会进行计算并得到结果。

要实现上面的式子，一般是要先声明好相关参数的张量，代码如下：

# 设置tensorflow图模型的输入 X = tf.placeholder("float") Y = tf.placeholder("float")

上面两句代码声明了模型的输入数据和对应的真实值(或者叫标签)，tf.placeholder()可以理解为为变量设置形参，在运算的时候再进行赋值，函数原型为：

placeholder(dtype, shape=None, name=None):

有了输入数据之后，我们还需要先声明好权值和偏置，代码如下：

# 设置模型权重和偏置 W = tf.Variable(rng.randn(), name="weight") b = tf.Variable(rng.randn(), name="bias")

这两句定义了两个变量，并且将两个变量随机初始化，并设置名称，tf.Variable()函数是tensorflow定义变量的的函数。

有了前面的变量之后，我们需要构造计算的表达式，这里就涉及到tensorflow中的一些基本运算：

tf.add() #相加 # 算术操作符：+ - * / % tf.add(x, y, name=None) # 加法(支持 broadcasting) tf.subtract(x, y, name=None) # 减法 tf.multiply(x, y, name=None) # 乘法 tf.divide(x, y, name=None) # 浮点除法, 返回浮点数(python3 除法) tf.mod(x, y, name=None) # 取余

有了上述基本的算术操作之后构造一个线性回归模型的表达式就简单了：

# Construct a linear model pred = tf.add(tf.multiply(X, W), b)

现在表达式已经有了，在训练的过程中我们还需要一个损失函数来衡量模型的误差并更新参数，线性回归模型比较常用的损失函数是均方误差(mean squared error)：

cost = tf.reduce_sum(tf.pow(pred-Y, 2))/(2*N_samples)

优化函数这里选择最常见的随机梯度下降法(SGD)，这个在tensorflow中是内置的，可以直接调用。

好了，这样基本的架构有了，就需要让训练跑起来，这个时候就需要构造一个会话(Session)，会话(session)拥有并管理TensorFlow 程序运行时的所有资源。当所有计算完成之后需要关闭会话来帮助系统回收资源，否则就可能出现资源泄漏的问题。创建会话的方式有两种：

1、第一种模式需要明确调用会话生成函数和关闭会话函数:

sess = tf.Session() ... sess.close()

2、第一种模式在所有计算完成之后，需要明确调用Session.close 函数来关闭会话并释放资源。然而，当程序因为异常而退出时，关闭会话的函数可能就不会被执行从而导致资源泄漏。为了解决异常退出时资源释放的问题，TensorFlow 可以通过Python 的上下文管理器来使用会话:

with tf.Session() as sess: ...

所以，这里我们采用第二种方式是比较合适的。

接着就是一个迭代过程，在每次迭代开始之后，我们需要把数据填充到前面所声明的两个占位符X和Y中，并执行优化算法：

sess.run(optimizer, feed_dict={X: x, Y: y})

完整的代码如下：

from __future__ import print_functionimport tensorflow as tf import numpy import matplotlib.pyplot as plt import numpy as np import matplotlib.pyplot as plt from sklearn import datasets as skd from sklearn.model_selection import train_test_split rng = numpy.random# 设置初始训练参数 learning_rate = 0.01 training_epochs = 5000 display_step = 50# 从sklearn中加载波士顿房价数据集，该数据集的数据为[506，13]，标签为[506] data = skd.load_boston() print("data.shape = ", data.data.shape) print("target.shape = ", data.target.shape)# 将数据集的数据和标签分离 #X_data = data.data[:,12] X_data = data.data[:,12] Y_data = data.target print("X_data.shape = ", X_data.shape) print("Y_data.shape = ", Y_data.shape) print("X_data[0:20, 12] = ", X_data[0:20]) print("Y_data[0:20, 12] = ", Y_data[0:20])# 将数据和标签分成训练集和测试集 x_train,x_test,y_train,y_test = train_test_split(X_data,Y_data,test_size=0.3,random_state=0) print("x_train.shape = ", x_train.shape) print("x_test.shape = ", x_test.shape) print("y_train.shape = ", y_train.shape) print("y_test.shape = ", y_test.shape)N_sample = x_train.shape[0]# 构建图模型计算的输入 X = tf.placeholder("float") Y = tf.placeholder("float")# 定义权重和偏置 W = tf.Variable(rng.randn(), name="weight") b = tf.Variable(rng.randn(), name="bias")# 构件图模型结构 y = tf.add(tf.multiply(W, X), b)# 定义损失函数 cost = tf.reduce_sum(tf.pow(y-Y, 2)/2/N_sample)# 定义优化函数 optimizer = tf.train.GradientDescentOptimizer(learning_rate=learning_rate).minimize(cost)# 定义初始化器 init = tf.global_variables_initializer()# 在会话中运行图模型计算 with tf.Session() as sess: # 执行初始化 sess.run(init) # 开始迭代训练 for epoch in range(training_epochs): for (x, y) in zip(x_train, y_train): sess.run(optimizer, feed_dict={X: x, Y: y}) if (epoch+1) % display_step == 0: c = sess.run(cost, feed_dict={X: x_train, Y:y_train}) print("Epoch:", '%04d' % (epoch+1), "cost=", "{:.9f}".format(c), "W=", sess.run(W), "b=", sess.run(b)) # 完成训练 print("训练完成") training_cost = sess.run(cost, feed_dict={X: x_train, Y: y_train}) print("训练误差=", training_cost, "W=", sess.run(W), "b=", sess.run(b), '\n') # 画图显示 plt.plot(x_train, y_train, 'ro', label='Original data') plt.plot(x_train, sess.run(W) * x_train + sess.run(b), label='Fitted line') plt.legend() plt.show() # 开始测试数据 cost_test = tf.reduce_sum(tf.pow(y - Y, 2)) / (2 * x_test.shape[0]) testing_cost = sess.run(cost_test, feed_dict={X: x_test, Y: y_test}) print("测试=", testing_cost) print("Absolute mean square loss difference:", abs(training_cost - testing_cost)) plt.plot(x_test, y_test, 'bo', label='Testing data') plt.plot(x_train, sess.run(W) * x_train + sess.run(b), label='Fitted line') plt.legend() plt.show()

这里的数据采用的是sklearn提供的波士顿房价数据，这个数据集是一个506*13维的数据集，我提取了其中的第二轴的第12维的数据来训练和预测，然后将数据集分成训练集和测试集，比例是7：3，代码运行中间训练信息：

训练数据和拟合的表达式如下：

测试数据集和表达式如下：

前面讲的是一元的线性回归模型，下面就讲一下多元线性回归模型的实现。数据依然是sklearn内置的波士顿房价数据，这个数据有506个样本，每个样本有13维特征，跟前面一样，这里先加载数据进来，然后切分为训练集和测试集：

boston = skd.load_boston() X_data = boston.data Y_data = boston.target x_train,x_test,y_train,y_test = train_test_split(X_data,Y_data,test_size=0.3,random_state=0) x_train = scale(x_train) x_test = scale(x_test) y_train = scale(y_train.reshape((-1,1))) y_test = scale(y_test.reshape((-1,1)))

然后，同样的，定义图模型计算的输入和标签：

X = tf.placeholder(tf.float32, [None, 13]) Y = tf.placeholder(tf.float32, [None, 1])

定义权重和偏置：

W = tf.Variable(tf.random_normal([13, 1]),dtype=tf.float32, name="weight") b = tf.Variable(tf.random_normal([1]),dtype=tf.float32, name="bias")

然后定义模型结构表达式和损失函数：

# 注意这里是矩阵相乘，所以要用tf.matmul y = tf.add(tf.matmul(X, W), b) cost = tf.reduce_mean(tf.square(Y-y))

剩下的操作其实就跟前面类似了，不过这里要说一下，如果像保存训练好的模型，那可以这么做：

saver = tf.train.Saver() with tf.Session() as sess: # 训练 saver.save(sess, "file_path/model_name.ckpt")

保存模型之后，如果要加载模型的话，可以这么做：

saver = tf.train.Saver() with tf.Session() as sess: sess.run(tf.global_variables_initializer()) saver.restore(sess,"file_path/")# 注意这里，只要到目录就可以 sess.run(...)

完整的代码如下：

from __future__ import print_function import tensorflow as tf import numpy as np import matplotlib.pyplot as plt from sklearn import datasets as skd from sklearn.model_selection import train_test_split from sklearn.preprocessing import scale rng = numpy.random# 设置初始训练参数 learning_rate = 0.001 training_epochs = 100000 display_step = 50# 从sklearn中加载波士顿房价数据集，该数据集的数据为[506，13]，标签为[506] boston = skd.load_boston() # 将数据集的数据和标签分离 X_data = boston.data Y_data = boston.target print("X_data.shape = ", X_data.shape) print("Y_data.shape = ", Y_data.shape)# 将数据和标签分成训练集和测试集 x_train,x_test,y_train,y_test = train_test_split(X_data,Y_data,test_size=0.3,random_state=0) x_train = scale(x_train) x_test = scale(x_test) y_train = scale(y_train.reshape((-1,1))) y_test = scale(y_test.reshape((-1,1))) print("x_train.shape = ", x_train.shape) print("x_test.shape = ", x_test.shape) print("y_train.shape = ", y_train.shape) print("y_test.shape = ", y_test.shape)N_sample = x_train.shape[0]# 构建图模型计算的输入 X = tf.placeholder(tf.float32, [None, 13]) Y = tf.placeholder(tf.float32, [None, 1])# 定义权重和偏置 #W = tf.Variable(rng.randn(), name="weight") #b = tf.Variable(rng.randn(), name="bias") W = tf.Variable(tf.random_normal([13, 1]),dtype=tf.float32, name="weight") b = tf.Variable(tf.random_normal([1]),dtype=tf.float32, name="bias")# 构件图模型结构 y = tf.add(tf.matmul(X, W), b) #y = tf.matmul(X, W)+b# 定义损失函数 #cost = tf.reduce_sum(tf.pow(y-Y, 2)/2/N_sample) cost = tf.reduce_mean(tf.square(Y-y))# 定义优化函数 optimizer = tf.train.GradientDescentOptimizer(learning_rate=learning_rate).minimize(cost)# 定义初始化器 init = tf.global_variables_initializer()# 创建保存模型 saver = tf.train.Saver()# 在会话中运行图模型计算 with tf.Session() as sess: # 执行初始化 sess.run(init) # 开始迭代训练 for epoch in range(training_epochs): sess.run(optimizer, feed_dict={X: x_train, Y: y_train}) if (epoch+1) % display_step == 0: c = sess.run(cost, feed_dict={X: x_train, Y:y_train}) print("Epoch:", '%04d' % (epoch+1), "cost=", "{:.9f}".format(c), "W=", sess.run(W), "b=", sess.run(b)) # 完成训练 saver.save(sess, "linear_regression/LiR4MultiFeatures.ckpt") print("训练完成") training_cost = sess.run(cost, feed_dict={X: x_train, Y: y_train}) print("训练误差=", training_cost, "W=", sess.run(W), "b=", sess.run(b), '\n') # 画图显示 plt.plot(x_train[:,12], y_train, 'ro', label='Original data') pred = sess.run(y, feed_dict={X: x_train}) plt.plot(x_train[:,12], pred, 'bx', label='predict data') plt.legend() plt.show() # 开始测试数据 cost_test = tf.reduce_sum(tf.pow(y - Y, 2)) / (2 * x_test.shape[0]) testing_cost = sess.run(cost_test, feed_dict={X: x_test, Y: y_test}) print("Testing cost=", testing_cost) print("Absolute mean square loss difference:", abs(training_cost - testing_cost)) plt.plot(x_test[:,12], y_test, 'ro', label='Testing data') pred_test = sess.run(y, feed_dict={X: x_test}) plt.plot(x_test[:,12], pred_test, 'bx', label='predicted data') plt.legend() plt.show()# 加载本地训练好的模型进行预测 model_file=tf.train.latest_checkpoint("linear_regression/") with tf.Session() as sess: sess.run(tf.global_variables_initializer()) saver.restore(sess,model_file) cost_test = tf.reduce_sum(tf.pow(y - Y, 2)) / (2 * x_test.shape[0]) testing_cost = sess.run(cost_test, feed_dict={X: x_test, Y: y_test}) print("Testing cost=", testing_cost) print("Absolute mean square loss difference:", abs(training_cost - testing_cost)) plt.plot(x_test[:,12], y_test, 'bo', label='Testing data') pred_test = sess.run(y, feed_dict={X: x_test}) plt.plot(x_test[:,12], pred_test, 'rx', label='predicted data') plt.legend() plt.show()