编写机器学习程序，解决机器学习问题，通常包含下面的步骤：① 分析问题，获取训练数据；② 定义模型和待学习的参数；③ 定义学习算法；④ 训练模型；⑤ 验证模型。
案例：某水厂2022年前四个月每月用水量见表2-1。
要求根据表2-1的数据建立月份和用水量之间的近似线性函数。若设月份为x，用水量为y，近似的线性函数为y = f(x)，也就是，要找出一个能使上述数据大体适合的函数关系y = f(x)，来预测该厂下个月的用水量。
该问题属于机器学习中的回归问题，我们需要来编写一个机器学习程序，该程序的输入是表2-1数据，输出是预测模型。
1、分析问题，获取训练数据
月份和用水量之间的函数关系为近似的线性关系，如图2-1所示。
图 2-1 月份和用水量近似线性函数关系
设线性函数为：
这是一个最基础的机器学习问题: 已知x和y , 尝试解得直线的斜率W和偏移量b。
回归问题属于监督式学习，监督学习使用输入（通常表示为 x）和输出（表示为 y）。目标是从成对的输入和输出中学得预测模型的参数，让预测模型的输出尽可能接近实际输出，以便可以根据输入预测输出的值。
TensorFlow采用张量来存储数据，张量可以简单理解为数组，在Python中就是列表对象。1阶张量是1维数组（向量），2阶张量是矩阵，3阶张量是3维数组，n阶张量是n维数组。若张量阶数为0，则张量为标量。
import numpy as np import tensorflow as tf import matplotlib.pyplot as plt ''' 定义数据 数据描述：某厂1~4月份用水量（单位：百吨）的一组数据 X：月份 Y：用水量 ''' X = tf.convert_to_tensor(np.array([1,2,3,4]),dtype=tf.float32) Y = tf.convert_to_tensor(np.array([4.5,4,3,2.5]),dtype=tf.float32)
TensorFlow提供的convert_to_tensor函数可以把numpy的数组数据按指定的数据类型转换为张量。
若需要可视化数据，可以使用matplotlib绘制数据，如图2-2所示。
plt.scatter(X, Y, c="b") plt.show()
图 2-2 XY数据可视化
2、定义模型和待学习的参数
从图2-2可以看出，X（月份）和Y（用水量）近似一元线性关系，可以使用一元线性回归模型作为机器学习的目标函数：
式2.2即为预测函数，输入月份x，输出预测的x月份用水量y。当前函数的参数W（权重）和b（偏移量）还是未知数，函数还不能正常工作。我们需要设计一个学习算法，从已获取的数据中学习到W和b的近似正确值，让函数能够正常工作。
式2.2仅是一个预测函数，还不能称之为预测模型。预测模型封装了预测函数和待学习的参数W和b，模型经过训练后，W和b被赋予近似正确值，模型还能够接收输入，执行被封装的预测函数，并输出预测结果。
TensorFlow提供了Module对象用于封装模型，Module对象为命名的TensorFlow容器，用于封装待训练的张量，可执行的功能函数，与模型相关的数据。下面的代码将式2-2定义为Module对象。
'''
定义待训练的线性模型
f(x) = x*w+b
训练权重w和偏置b，让模型输出接近于实际输出Y
'''
class MyModel(tf.Module):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# 初始化权重值为`5.0`，偏差值为`0.0`
# 实际项目中，应该随机初始化
self.w = tf.Variable(5.0)
self.b = tf.Variable(0.0)
@tf.function
def __call__(self, x):
return self.w * x + self.b
MyModel对象继承tf.Module，在MyModel对象内定义了w和b两个属性，w和b是线性函数的参数，它们的值需要在训练中迭代更新。Tensorflow的Variable()函数返回可训练的张量，并初始化张量的值。
__call__()函数是一个特殊的函数，MyModel对象实例化后会自动调用，在该函数内可以进行线性函数的计算。在函数前加@tf.function修饰，将模型转换为易于部署的TensorFlow 图模型，用于部署模型。
3、定义学习算法
对定义的模型如何进行训练，模型内的参数如何迭代更新，如何提高模型的训练效率和模型正确性，这是学习算法要做的事情。
图 2-3 用水量随着月份变化的趋势
图2-3红色趋势线就是我们要找的f(x)函数图形。从图中可以看出，f(x)没有经过图中所有的点，因为这些点本来就不在同一条直线上，函数f(x)反映了用水量随着月份变化的趋势。
如何确定w和b呢？我们知道使用最小二乘法可以对直线或曲线进行拟合，它使用均方误差（MSE）作为损失函数。
其中y(i)为表中所列的经验数据（用水量），x(i)为月份，f(x:w,b)为（2.1）函数，使用表2-1的数据代入式，我们希望使所有偏差的平方和最小，如何求最小值M呢？可以通过微积分的方法得到，把偏差的平方和看作函数，它有w和b两个变量，求这个函数的最小值。
该函数是二元二次函数，分别求变量w和b的偏导函数，令偏导数为0，M取得最小值。求导后得到下面的方程组：
60.0*w + 20.0*b - 63.0 = 0
20.0*w + 8.0*b - 28.0 = 0
解上面的方程组，就可以得到w和b的值。
实际上在机器学习中，比较常用的回归算法是计算损失函数的梯度，并迭代更新损失函数的参数w和b，直至损失函数的输出收敛到一个很小的值，即实际的y值和预测的y值之间的误差符合人们的预期。
下面的代码定义了训练函数。
def train(model, x, y, learning_rate): with tf.GradientTape() as t: # 可训练张量由GradientTape自动跟踪 current_loss = loss(y, model(x)) # 使用Gradient计算相对于W和b的梯度 dw, db = t.gradient(current_loss, [model.w, model.b]) # 减去由学习率缩放的梯度 model.w.assign_sub(learning_rate * dw) model.b.assign_sub(learning_rate * db) def loss(target_y, predicted_y): return tf.reduce_mean(tf.square(target_y - predicted_y))
参数model是MyModel实例对象，x和y是月份和用水量张量，存储了表2-1数据，learning_rate是学习率，即梯度每次缩放的步长。
GradientTape是TensorFlow用于计算函数梯度的对象，当在Python上下文环境使用它时，它会自动跟踪引起函数梯度变化的可训练张量。该对象的gradient()方法计算在上下文环境记录的函数的梯度dw和db。计算出dw和db后，更新model对象的可训练张量w和b。
assign_sub()是Variable对象更新其值的方法，Variable对象的当前值减去传入的值即为更新的值。assign_add()是Variable对象的当前值加上传入的值即为更新的值。assign()方法为Variable对象赋一个新的值。
loss()函数计算实际target_y值与预测 predicted_y值差平方的均值，target_y和predicted_y均为一维数组的张量。
train()函数仅完成了一次梯度计算，学习算法需要进行多次梯度计算，将误差收敛到符合预期的最小值。
下面的代码定义了循环训练函数。
def training_loop(model, x, y,epochs,ws,bs):
for epoch in epochs:
# 更新模型
train(model, x, y, learning_rate=0.1)
# 在更新之前进行跟踪
ws.append(model.w.numpy())
bs.append(model.b.numpy())
current_loss = loss(y, model(x))
# 若当前误差小于0.001，终止循环
if current_loss < 0.001:
break
print("Epoch %2d: W=%1.2f b=%1.2f, loss=%2.5f" %
(epoch, ws[-1], bs[-1], current_loss))参数model是MyModel实例对象，x和y是月份和用水量张量，epochs为Python可迭代对象，定义循环次数。ws和bs为Python列表对象，存储可训练张量w和b的更新记录。在循环内部定义了终止循环的条件，若当前误差小于0.001，则退出循环。Variable对象的mumpy()方法将张量转换为numpy数据。
4、训练模型
前面定义了模型和学习算法，下一步的工作就是训练模型了。下面的代码段对模型进行训练。
# 实例化模型
model = MyModel()
# 收集W值和b值的历史记录以供以后绘制
Ws, Bs = [], []
epochs = range(1000)
# 开始训练
training_loop(model, X, Y,epochs,Ws,Bs)
# 保存模型
tf.saved_model.save(model,"tmp/model")
# 输出训练后的模型
print("训练后的模型：f(x)=%.2f * x + %.2f" % (Ws[-1],Bs[-1]))上面的代码段实例化MyModel对象，定义Ws和Bs空列表对象，用于存储可训练张量W和b的更新记录，创建epochs可迭代对象，设置循环次数为1000，最后调用training_loop()函数开始模型训练。训练完成输出与模型关联的预测函数。
训练完成的模型保存到Python脚本文件所在目录同级tmp目录下。TF的saved_model对象用于存储和加载模型。
5、验证模型
验证模型是对模型是否能够正确工作的一个评估，验证过程是加载模型，输入验证数据，将模型输出结果和验证数据已知的结果进行比较，若两者误差符合人们的预期，则模型验证通过。
模型训练完成后会以文件方式保存到磁盘，应用TF的saved_model对象可以加载模型文件，然后执行模型文件，输入验证数据，查看输出结果。严格来说，验证数据和训练数据必须是两组不同的数据，计算它们的泛化误差，通过泛化误差才能反映出模型工作的正确性，泛化误差越小，模型越有效。
基于学习需要，我们以训练数据作为验证数据来验证模型，在正式的机器学习编程中，是不允许验证数据和训练数据相同的。
下面是模型的验证代码。
import numpy as np
import tensorflow as tf
X = tf.convert_to_tensor(np.array([1,2,3,4]),dtype=tf.float32)
# 程序入口
if __name__ == '__main__':
mymodel = tf.saved_model.load('tmp/model')
print(mymodel(X))应用TF的saved_mode对象从磁盘加载模型，执行模型并传入X数据，输出结果为预测的用水量。
例2-1程序清单
import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt
'''
定义数据
数据描述：某厂1~4月份用水量（单位：百吨）的一组数据
X：月份
Y：用水量
'''
X = tf.convert_to_tensor(np.array([1,2,3,4]),dtype=tf.float32)
Y = tf.convert_to_tensor(np.array([4.5,4,3,2.5]),dtype=tf.float32)
'''
定义待训练的线性模型
f(x) = x*w+b
训练权重w和偏置b，让模型输出接近于实际输出Y
'''
class MyModel(tf.Module):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# 初始化权重值为`5.0`，偏差值为`0.0`
# 实际项目中，应该随机初始化
self.w = tf.Variable(5.0)
self.b = tf.Variable(0.0)
@tf.function
def __call__(self, x):
return self.w * x + self.b
'''
定义损失函数
损失函数衡量给定输入的模型输出与目标输出的匹配程度，
目的是在训练过程中尽量减少这种差异。
损失函数为MSE（均方误差）
损失函数返回传入训练数据损失值的均值
'''
def loss(target_y, predicted_y):
return tf.reduce_mean(tf.square(target_y - predicted_y))
'''
定义训练模型函数
model：待训练的模型
x：月份张量
y：用水量张量
learning_rate：学习率
'''
def train(model, x, y, learning_rate):
with tf.GradientTape() as t:
# 可训练变量由GradientTape自动跟踪
current_loss = loss(y, model(x))
# 使用GradientTape计算相对于W和b的梯度
dw, db = t.gradient(current_loss, [model.w, model.b])
# 减去由学习率缩放的梯度
model.w.assign_sub(learning_rate * dw)
model.b.assign_sub(learning_rate * db)
'''
模型训练循环
'''
# 定义用于训练的循环
def training_loop(model, x, y,epochs,ws,bs):
for epoch in epochs:
# 更新模型
train(model, x, y, learning_rate=0.1)
# 在更新之前进行跟踪
ws.append(model.w.numpy())
bs.append(model.b.numpy())
current_loss = loss(y, model(x))
# 若当前误差小于0.001，终止循环
if current_loss < 0.001:
break
print("Epoch %2d: W=%1.2f b=%1.2f, loss=%2.5f" %
(epoch, ws[-1], bs[-1], current_loss))
# 程序入口
if __name__ == '__main__':
# 实例化模型
model = MyModel()
# 收集W值和b值的历史记录以供以后绘制
Ws, Bs = [], []
epochs = range(1000)
# 开始训练
training_loop(model, X, Y,epochs,Ws,Bs)
# 保存模型
tf.saved_model.save(model,"tmp/model")
# 输出训练后的模型
print("训练后的模型：f(x)=%.2f * x + %.2f" % (Ws[-1],Bs[-1]))
例2-1模型程序清单
import numpy as np
import tensorflow as tf
X = tf.convert_to_tensor(np.array([1,2,3,4]),dtype=tf.float32)
# 程序入口
if __name__ == '__main__':
mymodel = tf.saved_model.load('tmp/model')
print(mymodel(X))