【人工智能-01】编写训练循环

• By Admin
• 3年前

Keras 提供了默认的训练与评估循环 fit() 和 evaluate()。使用内置方法进行训练和评估指南中介绍了它们的用法。如果想要自定义模型的学习算法，同时又能利用 fit() 的便利性（例如，使用 fit() 训练 GAN），则可以将 Model 类子类化并实现自己的 train_step() 方法，此方法可在 fit() 中重复调用。自定义 fit() 的功能指南对此进行了介绍。现在，如果您想对训练和评估进行低级别控制，则应当从头开始编写自己的训练和评估循环。

import tensorflow as tf
#实际的线
TRUE_W = 3.0
TRUE_B = 2.0

NUM_EXAMPLES = 2

#随机向量
x = tf.random.normal(shape=[NUM_EXAMPLES])

#生成噪声
noise = tf.random.normal(shape=[NUM_EXAMPLES])

#计算y
y = x * TRUE_W + TRUE_B + noise





#绘制所有的数据
import matplotlib.pyplot as plt
plt.scatter(x,y,c="b")
plt.show()



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)

  def __call__(self, x):
    return self.w * x + self.b

model = MyModel()

# 列出变量tf.modules的内置变量聚合
print("Variables:", model.variables)

# 验证模型是否有效
assert model(3.0).numpy() == 15.0



#定义损失函数
#损失函数衡量给定输入的模型输出与目标输出的匹配程度。目的是在训练过程中尽量减少这种差异。
#定义标准的L2损失，也称为"均方误差"
#计算整个批次的单个损失值
def loss(target_y,predicted_y):
    return tf.reduce_mean(tf.square(target_y - predicted_y))



plt.scatter(x,y,c="b")
plt.scatter(x,model(x),c="r")
plt.show()

print("Current loss: %1.6f" % loss(model(x), y).numpy())




#定义训练循环
#训练循环按顺序重复执行以下任务：
#发送一批输入值，通过模型生成输出值
#通过比较输出值与输出（标签），来计算损失值
#使用梯度带找到梯度值
#使用这些梯度优化变量
# 给定一个可调用的模型，输入，输出和学习率...
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)



#为了了解训练，可以发送同一批 *x和y*经过循环训练，同时查看w和b的变化情况
model = MyModel()

# 收集W值和b值的历史记录以供以后绘制
Ws, bs = [], []
epochs = range(5)

# 定义用于训练的循环
def training_loop(model, x, y):

  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))
    print(y)
    print(model(x))
    print("Epoch %2d: W=%1.2f b=%1.2f  loss=%2.5f" %
          (epoch, Ws[-1], bs[-1], current_loss))



print("Starting: W=%1.2f b=%1.2f, loss=%2.5f" %
      (model.w, model.b, loss(y, model(x))))

# 开始训练
training_loop(model, x, y)

# 绘制
plt.plot(epochs, Ws, "r",
         epochs, bs, "b")

plt.plot([TRUE_W] * len(epochs), "r--",
         [TRUE_B] * len(epochs), "b--")

plt.legend(["W", "b", "True W", "True b"])
plt.show()





# 可视化训练后的模型如何执行
plt.scatter(x, y, c="b")
plt.scatter(x, model(x), c="r")
plt.show()

print("Current loss: %1.6f" % loss(model(x), y).numpy())



class MyModelKeras(tf.keras.Model):
  def __init__(self, **kwargs):
    super().__init__(**kwargs)
    # 初始化权重为`5.0`，偏差为`0.0`
    # 实际中应该随机初始化该值
    self.w = tf.Variable (5.0)
    self.b = tf.Variable(0.0)

  def __call__(self, x, **kwargs):
    return self.w * x + self.b

keras_model = MyModelKeras()

# 使用Keras模型重新进行循环训练
training_loop(keras_model, x, y)

# 您同样可以使用Keras内置的功能保存检查点（checkpoint）
keras_model.save_weights("my_checkpoint")




keras_model = MyModelKeras()
#编译设置培训参数
keras_model.compile(
    #默认情况下，fit()调用tf.function()。
    #debug时可以关闭这一功能，但现在是打开的
    run_eagerly=False,
    #使用内置优化器，配置为对象
    optimizer=tf.keras.optimizers.SGD(learning_rate=0.1),
    #Keras内置MSE
    #也可以使用损失函数像上面一样进行定义
    loss=tf.keras.losses.mean_squared_error
)





#Keras fit期望批处理数据或完整的数据集作为NumPy数组。NumPy数组分为多个批次，
#默认批次为32。
#这一案例中，为了匹配手写训练循环，我们应该以大小为1000的单批次传递x。
print(x.shape[0])
keras_model.fit(x,y,epochs=10,batch_size=1000)




# 可视化训练后的模型如何执行
plt.scatter(x, y, c="b")
plt.scatter(x, model(x), c="r")
plt.show()

print("Current loss: %1.6f" % loss(model(x), y).numpy())