TensorFlow：从Keras 一行到生产部署

# python fit_line.py —— CPU 上几秒跑完
import numpy as np
import tensorflow as tf

x = np.linspace(-1, 1, 200).astype("float32")
y = 2 * x + 1 + 0.05 * np.random.randn(200).astype("float32")   # 带噪声的观测

model = tf.keras.Sequential([tf.keras.layers.Dense(1)])            # 一行建模型
model.compile(optimizer="sgd", loss="mse")                        # 三要素配齐
model.fit(x, y, epochs=50, verbose=0)                              # 静默训练
print(model.predict(tf.constant([0.5]), verbose=0))                # 约 [[2.0]]（示意输出）

>>> t = tf.constant([[1.0, 2.0], [3.0, 4.0]])   # 不可变张量
>>> v = tf.Variable([[1.0, 2.0]])              # 可训练变量：模型参数就是它
>>> t.shape, t.dtype
(TensorShape([2, 2]), tf.float32)
>>> t + 10                                      # 运算产生新张量，t 本身不变

>>> x = tf.Variable(3.0)
>>> with tf.GradientTape() as tape:
...     y = x ** 2 + 2 * x                       # y = x² + 2x
>>> dy_dx = tape.gradient(y, x)                 # 导数 2x + 2 在 x=3 处
>>> dy_dx.numpy()
8.0

ds = tf.data.Dataset.from_tensor_slices((x_train, y_train))   # 切成一条条样本
ds = ds.shuffle(10000).batch(128).prefetch(tf.data.AUTOTUNE)  # 打乱 → 组批 → 预取
model.fit(ds, epochs=5)                                       # Dataset 直接当 fit 输入

# 自定义训练循环：结构永远不变——前向、求梯度、更新
optimizer = tf.keras.optimizers.Adam(1e-3)
ds = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(64)

for epoch in range(10):
    for xb, yb in ds:
        with tf.GradientTape() as tape:
            logits = model(xb, training=True)                # 1 前向
            loss = tf.keras.losses.sparse_categorical_crossentropy(yb, logits)
        grads = tape.gradient(loss, model.trainable_variables)           # 2 求梯度
        optimizer.apply_gradients(zip(grads, model.trainable_variables)) # 3 更新
    print(f"epoch {epoch} done")

# fit 自带的进度条、回调、验证集都要自己补；先跑通 fit，再按需下探

class MyModel(tf.keras.Model):            # 子类化：自由度最高的一式
    def __init__(self):
        super().__init__()
        self.dense = tf.keras.layers.Dense(10)

    def call(self, inputs):               # 定义数据怎么流过模型
        return self.dense(inputs)

model = MyModel()
model.build(tf.TensorShape([None, 4]))    # 子类式要先 build 或跑一次才有形状

import tensorflow as tf

# 1) 内置数据集：60000 训练 + 10000 测试，28x28 灰度图，10 类
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.fashion_mnist.load_data()

# 2) 归一化到 0~1：像素 0~255 直接喂会让 loss 又大又不稳
x_train = x_train / 255.0
x_test = x_test / 255.0

# 3) 加通道维：(60000, 28, 28) -> (60000, 28, 28, 1)，Conv2D 要求 4 维输入
x_train = x_train[..., tf.newaxis]
x_test = x_test[..., tf.newaxis]

print(x_train.shape, y_train.shape)   # (60000, 28, 28, 1) (60000,)

# 两轮「卷积 + 池化」提特征，摊平后接全连接出 10 类
inputs = tf.keras.Input(shape=(28, 28, 1))
x = tf.keras.layers.Conv2D(32, 3, activation="relu")(inputs)
x = tf.keras.layers.MaxPooling2D()(x)
x = tf.keras.layers.Conv2D(64, 3, activation="relu")(x)
x = tf.keras.layers.MaxPooling2D()(x)
x = tf.keras.layers.Flatten()(x)
x = tf.keras.layers.Dropout(0.4)(x)                       # 小数据集防过拟合
outputs = tf.keras.layers.Dense(10, activation="softmax")(x)
model = tf.keras.Model(inputs, outputs)
model.summary()                                           # 先看一眼结构与参数量

# 标签是 0~9 整数，用 sparse 前缀的交叉熵
model.compile(
    optimizer=tf.keras.optimizers.Adam(1e-3),
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

callbacks = [
    tf.keras.callbacks.EarlyStopping(monitor="val_accuracy", patience=3,
                                     restore_best_weights=True),   # 不再变好就停
    tf.keras.callbacks.ModelCheckpoint("best_model.keras", monitor="val_accuracy",
                                       save_best_only=True),       # 只存最优
]

history = model.fit(
    x_train, y_train,
    epochs=30,                # 上限开大，早停会自己收手
    batch_size=128,
    validation_split=0.1,     # 留 10% 训练数据当验证集
    callbacks=callbacks,
)

# 测试集评估（fit 从没见过这 10000 张）
test_loss, test_acc = model.evaluate(x_test, y_test, verbose=0)
print(f"测试集准确率：{test_acc:.3f}")        # 0.90 左右（示意输出）

# 单张预测：predict 返回 10 个概率，取最大者当类别
import numpy as np
probs = model.predict(x_test[:3], verbose=0)
print(probs.argmax(axis=1))                  # [9 2 1]（示意输出）

# 全量保存为 .keras：结构 + 权重 + 优化器状态一起带走
model.save("fashion_mnist_cnn.keras")
restored = tf.keras.models.load_model("fashion_mnist_cnn.keras")

# 在第 3 步的 callbacks 列表里追加一项：
callbacks.append(tf.keras.callbacks.TensorBoard(log_dir="logs"))

# 训练开跑后，另开一个终端启动看板（浏览器开 http://localhost:6006）：
# tensorboard --logdir logs —— loss / accuracy 曲线实时刷新

pip uninstall tensorflow-macos     # 2.16 起不再需要，留着会互相打架
pip install --upgrade --force-reinstall tensorflow
pip install tf_keras               # Keras 2 兼容包
export TF_USE_LEGACY_KERAS=1       # 让 tf.keras 指回 Keras 2 行为

# 两个修正都齐了再进模型
x = x.astype("float32") / 255.0     # 归一化
x = x[..., tf.newaxis]              # (N, 28, 28) -> (N, 28, 28, 1)
# 标签是整数就配 sparse_categorical_crossentropy；
# 报 sparse/one-hot 相关错误，八成是损失函数跟标签格式配错了对

import tensorflow as tf
print(tf.config.list_physical_devices("GPU"))
# 空列表 = GPU 没生效。Linux：确认装了 tensorflow[and-cuda]、驱动正常；
# WSL2：2.21.0 有 GPU 识别问题（见 GitHub issue #113541），可先用 2.20.0

# 路线 A：继续训练 / 自己加载推理
model.save("model.keras")
m2 = tf.keras.models.load_model("model.keras")

# 路线 B：上 TF Serving / 服务器部署（只含推理图）
model.export("serving/1")
# Serving 端按目录号加载：serving/1, serving/2 … 模型版本管理天然成型