一小时学会TensorFlow2之Fashion Mnist

# 将当前时间作为子文件名
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()