c/c++语言开发共享一小时学会TensorFlow2之Fashion Mnist

目录

• tensorboard
• metrics.mean()
• metrics.accuracy()

  # 将当前时间作为子文件名  current_time = datetime.datetime.now().strftime("%y%m%d-%h%m%s")    # 监听的文件的路径  log_dir = 'logs/' + current_time    # 创建writer  summary_writer = tf.summary.create_file_writer(log_dir)  

  tf.summary.scalar(      name, data, step=none, description=none  )  

  with summary_writer.as_default():      tf.summary.scalar("train-loss", float(cross_entropy), step=step)  

  tf.keras.metrics.mean(      name='mean', dtype=none  )  

  # 准确率表  loss_meter = tf.keras.metrics.mean()  

  tf.keras.metrics.accuracy(      name='accuracy', dtype=none  )  

  # 损失表  acc_meter = tf.keras.metrics.accuracy()  

  # 跟新损失  loss_meter.update_state(cross_entropy)    # 重置  loss_meter.reset_state()  

  def pre_process(x, y):      """      数据预处理      :param x: 特征值      :param y: 目标值      :return: 返回处理好的x, y      """      # 转换x      x = tf.cast(x, tf.float32) / 255      x = tf.reshape(x, [-1, 784])        # 转换y      y = tf.cast(y, dtype=tf.int32)      y = tf.one_hot(y, depth=10)        return x, y  

  def get_data():      """      获取数据      :return: 返回分批完的训练集和测试集      """        # 获取数据      (x_train, y_train), (x_test, y_test) = tf.keras.datasets.fashion_mnist.load_data()        # 分割训练集      train_db = tf.data.dataset.from_tensor_slices((x_train, y_train)).shuffle(60000, seed=0)      train_db = train_db.batch(batch_size).map(pre_process)        # 分割测试集      test_db = tf.data.dataset.from_tensor_slices((x_test, y_test)).shuffle(10000, seed=0)      test_db = test_db.batch(batch_size).map(pre_process)        # 返回      return train_db, test_db  

  def train(epoch, train_db):      """      训练数据      :param train_db: 分批的数据集      :return: 无返回值      """      for step, (x, y) in enumerate(train_db):          with tf.gradienttape() as tape:                # 获取模型输出结果              logits = model(x)                # 计算交叉熵              cross_entropy = tf.losses.categorical_crossentropy(y, logits, from_logits=true)              cross_entropy = tf.reduce_sum(cross_entropy)                # 跟新损失              loss_meter.update_state(cross_entropy)            # 计算梯度          grads = tape.gradient(cross_entropy, model.trainable_variables)            # 跟新参数          optimizer.apply_gradients(zip(grads, model.trainable_variables))            # 每100批调试输出一下误差          if step % 100 == 0:              print("step:", step, "cross_entropy:", loss_meter.result().numpy())                # 重置              loss_meter.reset_state()                # 可视化              with summary_writer.as_default():                  tf.summary.scalar("train-loss", float(cross_entropy), step= epoch * 235 + step)  

  def test(epoch, test_db):      """      测试模型      :param epoch: 轮数      :param test_db: 分批的测试集      :return: 无返回值      """        # 重置      acc_meter.reset_state()        for x, y in test_db:          # 获取模型输出结果          logits = model(x)            # 预测结果          pred = tf.argmax(logits, axis=1)            # 从one_hot编码变回来          y = tf.argmax(y, axis=1)            # 计算准确率          acc_meter.update_state(y, pred)        # 调试输出      print("epoch:", epoch + 1, "accuracy:", acc_meter.result().numpy() * 100, "%", )        # 可视化      with summary_writer.as_default():          tf.summary.scalar("val-acc", acc_meter.result().numpy(), step=epoch * 235)  

  def main():      """      主函数      :return: 无返回值      """        # 获取数据      train_db, test_db = get_data()        # 轮期      for epoch in range(iteration_num):          train(epoch, train_db)          test(epoch, test_db)  

  import datetime  import tensorflow as tf    # 定义超参数  batch_size = 256  # 一次训练的样本数目  learning_rate = 0.001  # 学习率  iteration_num = 20  # 迭代次数    # 优化器  optimizer = tf.keras.optimizers.adam(learning_rate=learning_rate)    # 准确率表  loss_meter = tf.keras.metrics.mean()    # 损失表  acc_meter = tf.keras.metrics.accuracy()    # 可视化  current_time = datetime.datetime.now().strftime("%y%m%d-%h%m%s")  log_dir = 'logs/' + current_time  summary_writer = tf.summary.create_file_writer(log_dir)  # 创建writer    # 模型  model = tf.keras.sequential([      tf.keras.layers.dense(256, activation=tf.nn.relu),      tf.keras.layers.dense(128, activation=tf.nn.relu),      tf.keras.layers.dense(64, activation=tf.nn.relu),      tf.keras.layers.dense(32, activation=tf.nn.relu),      tf.keras.layers.dense(10)  ])    # 调试输出summary  model.build(input_shape=[none, 28 * 28])  print(model.summary())      def pre_process(x, y):      """      数据预处理      :param x: 特征值      :param y: 目标值      :return: 返回处理好的x, y      """      # 转换x      x = tf.cast(x, tf.float32) / 255      x = tf.reshape(x, [-1, 784])        # 转换y      y = tf.cast(y, dtype=tf.int32)      y = tf.one_hot(y, depth=10)        return x, y      def get_data():      """      获取数据      :return: 返回分批完的训练集和测试集      """        # 获取数据      (x_train, y_train), (x_test, y_test) = tf.keras.datasets.fashion_mnist.load_data()        # 分割训练集      train_db = tf.data.dataset.from_tensor_slices((x_train, y_train)).shuffle(60000, seed=0)      train_db = train_db.batch(batch_size).map(pre_process)        # 分割测试集      test_db = tf.data.dataset.from_tensor_slices((x_test, y_test)).shuffle(10000, seed=0)      test_db = test_db.batch(batch_size).map(pre_process)        # 返回      return train_db, test_db      def train(epoch, train_db):      """      训练数据      :param train_db: 分批的数据集      :return: 无返回值      """      for step, (x, y) in enumerate(train_db):          with tf.gradienttape() as tape:                # 获取模型输出结果              logits = model(x)                # 计算交叉熵              cross_entropy = tf.losses.categorical_crossentropy(y, logits, from_logits=true)              cross_entropy = tf.reduce_sum(cross_entropy)                # 跟新损失              loss_meter.update_state(cross_entropy)            # 计算梯度          grads = tape.gradient(cross_entropy, model.trainable_variables)            # 跟新参数          optimizer.apply_gradients(zip(grads, model.trainable_variables))            # 每100批调试输出一下误差          if step % 100 == 0:              print("step:", step, "cross_entropy:", loss_meter.result().numpy())                # 重置              loss_meter.reset_state()                # 可视化              with summary_writer.as_default():                  tf.summary.scalar("train-loss", float(cross_entropy), step=epoch * 235 + step)      def test(epoch, test_db):      """      测试模型      :param epoch: 轮数      :param test_db: 分批的测试集      :return: 无返回值      """        # 重置      acc_meter.reset_state()        for x, y in test_db:          # 获取模型输出结果          logits = model(x)            # 预测结果          pred = tf.argmax(logits, axis=1)            # 从one_hot编码变回来          y = tf.argmax(y, axis=1)            # 计算准确率          acc_meter.update_state(y, pred)        # 调试输出      print("epoch:", epoch + 1, "accuracy:", acc_meter.result().numpy() * 100, "%", )        # 可视化      with summary_writer.as_default():          tf.summary.scalar("val-acc", acc_meter.result().numpy(), step=epoch * 235)      def main():      """      主函数      :return: 无返回值      """        # 获取数据      train_db, test_db = get_data()        # 轮期      for epoch in range(iteration_num):          train(epoch, train_db)          test(epoch, test_db)      if __name__ == "__main__":      main()