TensorFlow 教程

TensorFlow 是 Google 开源的深度学习框架,是目前最流行的 AI 开发工具之一,广泛应用于图像识别、自然语言处理、推荐系统等领域。

本教程包含

基础篇:简介、核心概念、环境搭建、张量操作
Keras 篇:Keras 入门、第一个神经网络、常用层
数据篇:数据处理、tf.data、图像/文本数据处理
训练篇:模型训练、评估、调优、图像/文本分类、回归
部署篇:模型保存加载、转换优化、生产部署、分布式训练

为什么学习 TensorFlow

生态成熟:TF Hub、TensorBoard、TF Lite
Keras API 友好,上手快
生产部署工具链完善
社区与文档丰富

TensorFlow 简介

TensorFlow 于 2015 年开源,是 Google Brain 团队开发的深度学习框架。

核心特性

张量(Tensor):核心数据结构,多维数组
计算图(Graph):高效的数值计算表示
自动微分:自动计算梯度,反向传播开箱即用
Keras API:高层接口,快速搭建模型
跨平台部署:服务器/移动端/浏览器/嵌入式

TensorFlow vs PyTorch

特性	TensorFlow	PyTorch
API 风格	Keras 高层/低层	Pythonic 动态图
调试	较复杂	直观
生产部署	完善(TF Serving/Lite)	TorchServe/ONNX
科研社区	较少	主流

安装

pip install tensorflow        # CPU 版
pip install tensorflow-gpu    # GPU 版(旧版本需要)

验证安装

import tensorflow as tf
print(tf.__version__)   # 2.x

TensorFlow 核心概念

掌握张量、自动微分、Eager Execution 等核心概念是使用 TensorFlow 的基础。

张量(Tensor)

import tensorflow as tf

# 创建张量
t1 = tf.constant([1, 2, 3])          # 一维
t2 = tf.constant([[1, 2], [3, 4]])   # 二维
t3 = tf.zeros((2, 3))                # 全零
t4 = tf.ones((2, 3))                 # 全一
t5 = tf.random.normal((3, 3))        # 随机正态

t2.shape    # (2, 2)
t2.dtype    # tf.int32

Eager Execution(动态图)

# TensorFlow 2 默认即时执行,像 Python 一样直观
a = tf.constant(2.0)
b = tf.constant(3.0)
print(a + b)   # tf.Tensor(5.0)

自动微分

with tf.GradientTape() as tape:
    x = tf.Variable(3.0)
    y = x * x           # y = x^2

grad = tape.gradient(y, x)   # dy/dx = 2x = 6
print(grad)   # tf.Tensor(6.0)

变量(Variable)

w = tf.Variable(0.5)     # 可训练参数
w.assign(0.8)            # 更新值
w.assign_add(0.1)        # w += 0.1

计算流程

数据(张量) → 模型(层) → 损失 → 梯度 → 优化更新

TensorFlow 环境搭建

搭建 TensorFlow 开发环境,包括 Python、GPU 支持和 IDE 配置。

1. 安装 Python

# 推荐 Python 3.9 - 3.12
python --version

2. 创建虚拟环境(推荐)

python -m venv tf_env
source tf_env/bin/activate   # macOS/Linux
# Windows: tf_env\Scripts\activate

3. 安装 TensorFlow

pip install tensorflow

4. GPU 支持(可选)

# 需要 NVIDIA 显卡 + CUDA + cuDNN
nvidia-smi   # 检查显卡驱动
pip install tensorflow[and-cuda]

5. 验证安装

python -c "import tensorflow as tf; print(tf.__version__)"

# 检查 GPU 是否可用
import tensorflow as tf
print(tf.config.list_physical_devices("GPU"))

6. IDE 与工具

VS Code + Python 扩展
Jupyter Notebook / JupyterLab(交互式实验)
TensorBoard(可视化训练)

注意事项

CPU 版即可学习入门,训练大模型再上 GPU
macOS(Apple Silicon)可用 Metal 加速
容器环境推荐官方 Docker 镜像

TensorFlow 张量操作

张量操作是 TensorFlow 编程的基础,包括创建、索引、运算、变形等。

创建张量

import tensorflow as tf

tf.constant(5)                    # 标量
tf.constant([1, 2, 3])            # 向量
tf.constant([[1, 2], [3, 4]])     # 矩阵
tf.zeros([2, 3])
tf.ones([2, 3])
tf.fill([2, 3], 7)
tf.range(0, 10, 2)                # [0 2 4 6 8]
tf.linspace(0, 1, 5)

索引与切片

t = tf.constant([[1, 2, 3], [4, 5, 6]])
t[0]          # 第一行
t[:, 1]       # 第二列
t[1, 2]       # 6
t[0:2, 1:]    # 切片

数学运算

a = tf.constant([1, 2, 3])
b = tf.constant([4, 5, 6])

a + b          # 逐元素加
a * b          # 逐元素乘
tf.reduce_sum(a)     # 求和: 6
tf.reduce_mean(a)    # 均值: 2
tf.reduce_max(a)     # 最大值
tf.matmul(m1, m2)    # 矩阵乘法

形状操作

t = tf.reshape(tf.range(12), (3, 4))   # 重塑
tf.transpose(t)                        # 转置
tf.expand_dims(t, 0)                   # 加维度
tf.squeeze(t)                          # 去维度
tf.concat([a, b], axis=0)              # 拼接

类型转换

tf.cast(t, tf.float32)   # 类型转换
t.numpy()                # 转 numpy 数组
tf.convert_to_tensor(np_arr)  # numpy 转张量

TensorFlow Keras

Keras 是 TensorFlow 的高层 API,提供简洁的模型构建方式,是官方推荐的入门选择。

为什么用 Keras

简单直观:几行代码构建模型
模块化:层、模型、损失、优化器组件化
内置训练循环:fit 方法一键训练
支持迁移学习:预训练模型丰富

三种建模方式

# 1. Sequential(序列模型,简单堆叠)
model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(10, activation="softmax"),
])

# 2. Functional(函数式,复杂结构)
inputs = tf.keras.Input(shape=(784,))
x = tf.keras.layers.Dense(64, activation="relu")(inputs)
outputs = tf.keras.layers.Dense(10, activation="softmax")(x)
model = tf.keras.Model(inputs=inputs, outputs=outputs)

# 3. Subclassing(子类化,完全自定义)
class MyModel(tf.keras.Model):
    def __init__(self):
        super().__init__()
        self.dense1 = tf.keras.layers.Dense(64, activation="relu")
        self.dense2 = tf.keras.layers.Dense(10)
    def call(self, x):
        return self.dense2(self.dense1(x))

编译与训练

model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

model.fit(x_train, y_train, epochs=10, validation_split=0.2)

Keras 第一个神经网络

用 Keras 构建并训练第一个神经网络,识别手写数字(MNIST)。

完整代码

import tensorflow as tf
from tensorflow import keras

# 1. 加载数据
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()

# 2. 预处理
x_train = x_train.reshape(-1, 28 * 28).astype("float32") / 255.0
x_test = x_test.reshape(-1, 28 * 28).astype("float32") / 255.0

# 3. 构建模型
model = keras.Sequential([
    keras.layers.Dense(128, activation="relu"),
    keras.layers.Dense(10, activation="softmax"),
])

# 4. 编译
model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

# 5. 训练
history = model.fit(
    x_train, y_train,
    epochs=5,
    validation_split=0.2,
    verbose=1,
)

# 6. 评估
test_loss, test_acc = model.evaluate(x_test, y_test)
print(f"测试准确率: {test_acc:.4f}")

# 7. 预测
predictions = model.predict(x_test[:5])

关键点

MNIST:28x28 手写数字图片,10 类
展平为 784 维向量输入
归一化到 [0,1] 加速收敛
softmax 输出 10 类概率

Keras 常用层类型

了解常用层,知道何时使用哪种层是设计网络的关键。

全连接层(Dense)

keras.layers.Dense(64, activation="relu")
# 输入输出全连接,最基础的层

卷积层(Conv2D)

keras.layers.Conv2D(
    filters=32,          # 卷积核数量
    kernel_size=(3, 3),  # 卷积核大小
    activation="relu",
    input_shape=(28, 28, 1),  # 图像输入
)
# 用于图像特征提取

池化层

keras.layers.MaxPooling2D(pool_size=(2, 2))
keras.layers.GlobalAveragePooling2D()   # 全局平均池化

循环层(RNN/LSTM)

keras.layers.SimpleRNN(64)
keras.layers.LSTM(128)          # 长短期记忆
keras.layers.GRU(128)           # 门控循环单元
# 用于序列数据(文本/时间序列)

Dropout(防过拟合)

keras.layers.Dropout(0.5)   # 随机丢弃50%神经元

BatchNormalization

keras.layers.BatchNormalization()   # 批量归一化,加速训练

Embedding(嵌入层)

keras.layers.Embedding(vocab_size, embedding_dim)
# 词向量:把离散词 ID 映射为稠密向量

Flatten 与 Reshape

keras.layers.Flatten()    # 展平
keras.layers.Reshape((28, 28))

TensorFlow 数据处理

数据质量决定模型上限,数据预处理是深度学习的核心环节。

数据预处理要点

归一化: 像素/特征缩放到 [0,1] 或标准化
one-hot: 分类标签转独热编码
增强:   图像翻转/旋转/缩放扩充数据
清洗:   缺失值处理、去异常
分批:   小批量训练(Batch)

图像归一化示例

import tensorflow as tf

# 像素值 0-255 → 0-1
x_train = x_train.astype("float32") / 255.0

# 标准化(均值0,标准差1)
mean = tf.reduce_mean(x_train)
std = tf.math.reduce_std(x_train)
x_norm = (x_train - mean) / std

标签处理

# 分类标签 one-hot
y_onehot = tf.one_hot(y_train, depth=10)

# 或保持整数索引,损失用 sparse_categorical_crossentropy
# 配合整型标签使用

图像增强

data_augmentation = tf.keras.Sequential([
    tf.keras.layers.RandomFlip("horizontal"),
    tf.keras.layers.RandomRotation(0.1),
    tf.keras.layers.RandomZoom(0.1),
])

# 训练时应用
augmented = data_augmentation(x_batch, training=True)

数据集划分

# train / validation / test
# 训练集训练,验证集调参,测试集最终评估
x_train, x_val = tf.split(x_train, [int(0.8*len(x_train)), -1])

TensorFlow tf.data API

tf.data 是 TensorFlow 的高性能数据管道 API,用于高效加载和预处理数据。

创建数据集

import tensorflow as tf

# 从数组
dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))

# 从文件
dataset = tf.data.TextLineDataset("data.txt")

# 从生成器
dataset = tf.data.Dataset.from_generator(gen, output_types=(tf.float32, tf.int32))

常用操作

dataset = tf.data.Dataset.from_tensor_slices((x, y))

# 打乱
dataset = dataset.shuffle(buffer_size=10000)

# 分批
dataset = dataset.batch(32)

# 重复(多个 epoch)
dataset = dataset.repeat(10)

# 预取(加速训练)
dataset = dataset.prefetch(tf.data.AUTOTUNE)

完整流水线

def prepare_dataset(x, y, batch_size=32):
    dataset = tf.data.Dataset.from_tensor_slices((x, y))
    dataset = dataset.shuffle(1000)
    dataset = dataset.map(preprocess, num_parallel_calls=tf.data.AUTOTUNE)
    dataset = dataset.batch(batch_size)
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    return dataset

train_ds = prepare_dataset(x_train, y_train)
model.fit(train_ds, epochs=10)

优势

自动并行处理
内存友好(流式读取)
与 Keras fit 无缝集成
适合大数据集

TensorFlow 图像数据处理

图像数据处理的完整流程:加载、解码、预处理、增强。

从目录加载图像

# 目录结构: train/cat/*.jpg train/dog/*.jpg
train_ds = tf.keras.preprocessing.image_dataset_from_directory(
    "train",
    image_size=(224, 224),
    batch_size=32,
    label_mode="int",      # 或 "categorical"
    validation_split=0.2,
    subset="training",
)

图像预处理

def preprocess(image, label):
    # 归一化
    image = tf.cast(image, tf.float32) / 255.0
    return image, label

train_ds = train_ds.map(preprocess)

图像增强

data_augmentation = tf.keras.Sequential([
    tf.keras.layers.RandomFlip("horizontal_and_vertical"),
    tf.keras.layers.RandomRotation(0.2),
    tf.keras.layers.RandomContrast(0.2),
])

读取单张图片

image = tf.io.read_file("img.jpg")
image = tf.image.decode_jpeg(image, channels=3)
image = tf.image.resize(image, [224, 224])
image = tf.expand_dims(image, 0)   # 加 batch 维

常用图像操作

tf.image.resize(image, [224, 224])
tf.image.flip_left_right(image)
tf.image.rot90(image)
tf.image.adjust_brightness(image, 0.2)
tf.image.central_crop(image, 0.8)

TensorFlow 文本数据处理

文本数据需要经过分词、向量化后才能输入神经网络。

文本向量化流程

原始文本 → 清洗 → 分词 → 索引化 → 词嵌入/向量 → 模型输入

TextVectorization 层

import tensorflow as tf

# 1. 定义向量化层
vectorizer = tf.keras.layers.TextVectorization(
    max_tokens=10000,       # 词表大小
    output_mode="int",      # 输出词索引
    output_sequence_length=200,  # 固定序列长度
)

# 2. 适配数据(建立词表)
vectorizer.adapt(train_texts)

# 3. 转换
sequences = vectorizer(texts)   # (batch, 200) 整数索引

Word Embedding(词嵌入)

model = tf.keras.Sequential([
    tf.keras.layers.Embedding(
        input_dim=10000,    # 词表大小
        output_dim=128,     # 嵌入维度
    ),
    tf.keras.layers.GlobalAveragePooling1D(),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

文本清洗示例

import re

def clean_text(text):
    text = text.lower()                      # 小写
    text = re.sub(r"<[^>]+>", "", text)      # 去 HTML
    text = re.sub(r"[^\w\s]", "", text)      # 去标点
    text = re.sub(r"\s+", " ", text)         # 合并空格
    return text

数据管道集成

train_ds = tf.data.Dataset.from_tensor_slices((texts, labels))
train_ds = train_ds.map(lambda t, l: (vectorizer(t), l))
train_ds = train_ds.batch(32).prefetch(tf.data.AUTOTUNE)

TensorFlow 模型训练

模型训练是深度学习核心流程:前向传播、计算损失、反向传播、更新参数。

fit 方法训练

history = model.fit(
    x_train, y_train,
    epochs=10,
    batch_size=32,
    validation_data=(x_val, y_val),
    callbacks=[...],       # 回调
)

回调(Callback)

# 早停(防过拟合)
early_stop = tf.keras.callbacks.EarlyStopping(
    monitor="val_loss",
    patience=3,           # 连续3轮无改善则停止
    restore_best_weights=True,
)

# 学习率衰减
lr_schedule = tf.keras.callbacks.ReduceLROnPlateau(
    factor=0.5, patience=2
)

# 模型检查点
checkpoint = tf.keras.callbacks.ModelCheckpoint(
    "best_model.h5",
    save_best_only=True,
    monitor="val_accuracy",
)

model.fit(..., callbacks=[early_stop, lr_schedule, checkpoint])

自定义训练循环

@tf.function
def train_step(x, y):
    with tf.GradientTape() as tape:
        pred = model(x, training=True)
        loss = loss_fn(y, pred)
    grads = tape.gradient(loss, model.trainable_variables)
    optimizer.apply_gradients(zip(grads, model.trainable_variables))
    return loss

for epoch in range(epochs):
    for x, y in train_ds:
        loss = train_step(x, y)
    print(f"epoch {epoch}: loss={loss:.4f}")

TensorFlow 模型评估

训练后必须评估模型在测试集上的表现,衡量泛化能力。

评估方法

# 1. evaluate(整体评估)
loss, acc = model.evaluate(x_test, y_test)
print(f"测试损失: {loss:.4f}, 准确率: {acc:.4f}")

# 2. predict(逐样本预测)
predictions = model.predict(x_test)
pred_classes = tf.argmax(predictions, axis=1)

评估指标

分类: accuracy / precision / recall / F1
回归: MAE / MSE / RMSE
编译时指定: model.compile(metrics=["accuracy"])

混淆矩阵

from sklearn.metrics import confusion_matrix

cm = confusion_matrix(y_test, pred_classes)
# 对角线是正确分类,便于发现易混淆类别

可视化训练曲线

import matplotlib.pyplot as plt

plt.plot(history.history["accuracy"], label="train")
plt.plot(history.history["val_accuracy"], label="val")
plt.legend()
plt.show()
# 训练高验证低 = 过拟合信号

TensorBoard 可视化

tensorboard_cb = tf.keras.callbacks.TensorBoard(log_dir="logs")
model.fit(..., callbacks=[tensorboard_cb])

# 命令行
tensorboard --logdir logs

评估注意事项

测试集不能参与训练与调参
类别不平衡时用宏平均指标
多次运行取平均更可靠

TensorFlow 模型调优技巧

提升模型性能的系统性方法,从数据、模型、训练三个维度入手。

1. 数据层面

增加数据量/数据增强
归一化/标准化
平衡类别(过采样/欠采样)
去除噪声与错误标签

2. 模型层面

增加容量(更多层/神经元)
添加 Dropout(0.2-0.5) 防过拟合
BatchNormalization 加速收敛
迁移学习:使用预训练模型
合适激活函数:ReLU 默认,输出层按任务选

3. 训练层面

学习率: 1e-3 起步,配合衰减调度
优化器: Adam 默认,收敛后换 SGD+动量
批量大小: 32-128 常用,影响收敛
早停: 监控 val_loss,耐心 3-5 轮
epochs: 用 EarlyStopping 自动确定
梯度裁剪: 防止梯度爆炸

调优流程

1. 建立基线(简单模型跑通)
2. 逐个因素实验(一次只改一个)
3. 监控训练/验证曲线
4. 正则化与数据增强优先
5. 网格搜索/随机搜索找最优超参

过拟合处理

更多数据 + 数据增强
Dropout
L1/L2 正则化
减小模型容量
早停

TensorFlow 图像分类

构建一个 CNN 完成图像分类任务,以 CIFAR-10(10类小图)为例。

数据准备

import tensorflow as tf

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.cifar10.load_data()
x_train = x_train.astype("float32") / 255.0
x_test = x_test.astype("float32") / 255.0
y_train = y_train.flatten()
y_test = y_test.flatten()

构建 CNN

model = tf.keras.Sequential([
    tf.keras.layers.Conv2D(32, (3,3), activation="relu", input_shape=(32,32,3)),
    tf.keras.layers.MaxPooling2D((2,2)),
    tf.keras.layers.Conv2D(64, (3,3), activation="relu"),
    tf.keras.layers.MaxPooling2D((2,2)),
    tf.keras.layers.Conv2D(64, (3,3), activation="relu"),
    tf.keras.layers.Flatten(),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dropout(0.5),
    tf.keras.layers.Dense(10, activation="softmax"),
])

model.compile(
    optimizer="adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)

训练

history = model.fit(
    x_train, y_train,
    epochs=10,
    validation_data=(x_test, y_test),
)

迁移学习(效果更好)

base = tf.keras.applications.MobileNetV2(
    weights="imagenet",
    include_top=False,
    input_shape=(128, 128, 3),
)
base.trainable = False   # 冻结预训练层

model = tf.keras.Sequential([
    base,
    tf.keras.layers.GlobalAveragePooling2D(),
    tf.keras.layers.Dense(10, activation="softmax"),
])

TensorFlow 文本分类

文本分类是 NLP 的经典任务,如情感分析、垃圾邮件检测。

数据与向量化

import tensorflow as tf

# 使用 IMDB 影评情感数据集(0负面 1正面)
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.imdb.load_data(
    num_words=10000
)

# 数据已是整数索引,补齐到固定长度
x_train = tf.keras.preprocessing.sequence.pad_sequences(x_train, maxlen=200)
x_test = tf.keras.preprocessing.sequence.pad_sequences(x_test, maxlen=200)

构建模型

model = tf.keras.Sequential([
    tf.keras.layers.Embedding(10000, 128),
    tf.keras.layers.GlobalAveragePooling1D(),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dropout(0.5),
    tf.keras.layers.Dense(1, activation="sigmoid"),  # 二分类
])

model.compile(
    optimizer="adam",
    loss="binary_crossentropy",
    metrics=["accuracy"],
)

训练与评估

x_val, y_val = x_train[:5000], y_train[:5000]
x_train, y_train = x_train[5000:], y_train[5000:]

history = model.fit(
    x_train, y_train,
    epochs=10,
    batch_size=512,
    validation_data=(x_val, y_val),
)

model.evaluate(x_test, y_test)

LSTM 版本(序列建模)

model = tf.keras.Sequential([
    tf.keras.layers.Embedding(10000, 128),
    tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(64)),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(1, activation="sigmoid"),
])

Transformer 方向

# 更先进方案:BERT 等预训练模型
# pip install transformers
from transformers import TFBertForSequenceClassification

TensorFlow 回归问题

回归预测连续数值,如房价、温度、销量。

数据准备(波士顿房价或自定义)

import tensorflow as tf

# 生成示例数据 y = 3x + 2 + 噪声
x = tf.range(0, 100, dtype=tf.float32)
y = 3 * x + 2 + tf.random.normal((100,), stddev=5)

# 归一化
x = (x - tf.reduce_mean(x)) / tf.math.reduce_std(x)
y = (y - tf.reduce_mean(y)) / tf.math.reduce_std(y)

构建回归模型

model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation="relu", input_shape=(1,)),
    tf.keras.layers.Dense(32, activation="relu"),
    tf.keras.layers.Dense(1),     # 输出层无激活函数
])

model.compile(
    optimizer="adam",
    loss="mse",                    # 均方误差
    metrics=["mae"],               # 平均绝对误差
)

训练与预测

history = model.fit(x, y, epochs=100, verbose=0)

# 预测
x_new = tf.constant([50.0])
x_new_norm = (x_new - x_mean) / x_std
pred = model.predict(x_new_norm)
# 反归一化得到实际预测值

回归常用损失

mse        均方误差(对异常敏感)
mae        平均绝对误差(鲁棒)
huber      结合两者(推荐)

多特征回归

# 输入多个特征(面积、卧室数、位置...)
model = tf.keras.Sequential([
    tf.keras.layers.Dense(128, activation="relu", input_shape=(num_features,)),
    tf.keras.layers.Dense(64, activation="relu"),
    tf.keras.layers.Dense(1),
])

评估指标

R²(决定系数)、MAE、RMSE
用 sklearn: r2_score / mean_absolute_error

TensorFlow 模型保存与加载

保存训练好的模型,方便后续加载使用、迁移和部署。

保存整个模型(HDF5)

# 保存(架构+权重+优化器状态)
model.save("my_model.h5")

# 加载
model = tf.keras.models.load_model("my_model.h5")
model.predict(x_test)

SavedModel 格式(推荐)

# 保存
model.export("saved_model")

# 或
tf.saved_model.save(model, "saved_model")

# 加载
model = tf.keras.models.load_model("saved_model")

仅保存权重

# 保存权重
model.save_weights("weights.h5")

# 加载权重(需先定义相同架构)
model = create_model()
model.load_weights("weights.h5")

训练中自动保存(检查点)

checkpoint = tf.keras.callbacks.ModelCheckpoint(
    "best_model.weights.h5",
    monitor="val_accuracy",
    save_best_only=True,
    save_weights_only=True,
)
model.fit(..., callbacks=[checkpoint])

保存与加载对比

save_weights    只权重,体积小,需重建架构
save("model.h5") 完整模型,单文件方便
SavedModel      生产部署标准格式(与 TF Serving 兼容)

TensorFlow 模型转换与优化

将训练好的模型转换、压缩、加速,适配不同部署环境。

TFLite 转换(移动/嵌入式)

import tensorflow as tf

# 加载模型
model = tf.keras.models.load_model("model.h5")

# 转换为 TFLite
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

# 保存
with open("model.tflite", "wb") as f:
    f.write(tflite_model)

量化(Quantization)

# 动态范围量化(减小模型体积)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_quant = converter.convert()

# 全整数量化(进一步压缩)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = representative_data_gen

模型剪枝与蒸馏

# 剪枝: 移除不重要的权重连接
import tensorflow_model_optimization as tfmot
prune_low_magnitude = tfmot.sparsity.keras.prune_low_magnitude

# 知识蒸馏: 大模型(教师)教小模型(学生)

TFLite 推理

import numpy as np

interpreter = tf.lite.Interpreter(model_path="model.tflite")
interpreter.allocate_tensors()

input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

interpreter.set_tensor(input_details[0]["index"], input_data)
interpreter.invoke()
output = interpreter.get_tensor(output_details[0]["index"])

转换注意事项

量化可能轻微损失精度
确保输入输出张量形状固定
动态维度操作可能不支持转换

TensorFlow 生产环境

把模型部署到生产环境,提供实时推理服务。

TensorFlow Serving

# 1. 模型导出为 SavedModel 格式
model.export("saved_model")

# 2. 启动 TF Serving 容器
docker run -p 8501:8501   -v $(pwd)/saved_model:/models/my_model   -e MODEL_NAME=my_model   tensorflow/serving

# 3. 调用 REST API
curl -X POST http://localhost:8501/v1/models/my_model:predict   -d '{"instances": [[0.1, 0.2, ...]]}'

TFLite 移动端部署

# Android 集成 TFLite
implementation "org.tensorflow:tensorflow-lite:2.x"
implementation "org.tensorflow:tensorflow-lite-support:0.4.x"

// 加载模型并推理
Interpreter interpreter = new Interpreter(loadModelFile(context));

Web 端部署(TensorFlow.js)

// 转 TF.js 格式
import tensorflowjs as tfjs
tfjs.converters.save_keras_model(model, "tfjs_model")

// 浏览器加载
const model = await tf.loadLayersModel("tfjs_model/model.json");

推理优化

批处理请求(Batching)
GPU 推理
量化模型
模型缓存与预加载

监控与版本管理

模型版本控制
推理延迟监控
漂移检测(输入分布变化)
A/B 测试

TensorFlow 分布式训练

大规模数据与模型需要分布式训练,使用多 GPU 或多台机器加速。

分布式策略

MirroredStrategy     单机多 GPU(数据并行)
MultiWorkerMirrored  多机多 GPU
TPUStrategy          TPU 训练
ParameterServer      参数服务器模式

MirroredStrategy 示例

import tensorflow as tf

strategy = tf.distribute.MirroredStrategy()
print(f"使用 {strategy.num_replicas_in_sync} 个 GPU")

with strategy.scope():
    # 模型、优化器必须在 strategy 作用域内创建
    model = create_model()
    model.compile(optimizer="adam", loss="mse")

数据并行原理

每张 GPU 一份模型副本
数据切成多份,每 GPU 训练一部分
梯度同步聚合(AllReduce)
更新全局模型

MultiWorkerMirrored(多机)

# TF_CONFIG 环境变量配置集群
os.environ["TF_CONFIG"] = json.dumps({
    "cluster": {
        "worker": ["host1:2222", "host2:2222"]
    },
    "task": {"type": "worker", "index": 0}
})

strategy = tf.distribute.MultiWorkerMirroredStrategy()

数据流水线配合

# 每副本独立数据流
dataset = dataset.shard(num_replicas, replica_index)
# 或使用 tf.distribute.DistributedDataset 自动分发

何时需要分布式

单卡显存不足
训练时间过长(>天级)
超大数据集/超大模型
注意: 小模型分布式反而变慢(通信开销)

TensorFlow 生态系统

TensorFlow 拥有丰富的周边工具与扩展,覆盖全流程。

核心工具

Keras          高层建模 API
TensorBoard    训练可视化
TF Serving     生产推理服务
TFLite         移动/嵌入式部署
TF.js          浏览器/Node 部署

模型库

TF Hub        预训练模型中心(module)
Keras Applications  内置模型(ResNet/MobileNet等)
Model Garden  官方模型实现

扩展库

TensorFlow Addons   附加功能(优化器/层)
TensorFlow I/O      文件系统与格式支持
TensorFlow Probability  概率编程
TensorFlow Quantum  量子机器学习
TensorFlow Privacy  差分隐私训练
TensorFlow Model Optimization  模型优化

领域生态

TensorFlow Object Detection API  目标检测
MediaPipe          多模态应用框架
TensorFlow Graphics 图形与3D
TF Text            文本处理工具

学习资源

官网教程: tensorflow.org/tutorials
Keras 文档: keras.io
TensorFlow 中文社区
Coursera DeepLearning.AI 课程

生态选择建议

学习优先 Keras + TensorBoard;移动端部署学 TFLite;科研自由探索 PyTorch/TF 均可,掌握核心概念后框架迁移成本低。

TensorFlow 自定义组件

自定义层、损失函数、模型和训练循环,突破标准组件限制。

自定义层

import tensorflow as tf

class MyDense(tf.keras.layers.Layer):
    def __init__(self, units=32):
        super().__init__()
        self.units = units

    def build(self, input_shape):
        self.w = self.add_weight(
            shape=(input_shape[-1], self.units),
            initializer="glorot_uniform",
            trainable=True,
        )
        self.b = self.add_weight(shape=(self.units,), initializer="zeros")

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

自定义损失

def huber_loss(y_true, y_pred, delta=1.0):
    error = y_true - y_pred
    return tf.where(
        tf.abs(error) < delta,
        0.5 * tf.square(error),
        delta * (tf.abs(error) - 0.5 * delta),
    )

model.compile(loss=huber_loss)

自定义指标

class MyAccuracy(tf.keras.metrics.Metric):
    def __init__(self, name="my_acc"):
        super().__init__(name=name)
        self.correct = self.add_weight(name="correct", initializer="zeros")
        self.total = self.add_weight(name="total", initializer="zeros")

    def update_state(self, y_true, y_pred):
        correct = tf.cast(tf.equal(y_true, tf.argmax(y_pred, axis=1)), tf.float32)
        self.correct.assign_add(tf.reduce_sum(correct))
        self.total.assign_add(tf.cast(len(y_true), tf.float32))

    def result(self):
        return self.correct / self.total

自定义回调

class MyCallback(tf.keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        print(f"epoch {epoch}: val_loss={logs.get('val_loss'):.4f}")
        if logs.get("val_acc", 0) > 0.95:
            self.model.stop_training = True

自定义模型与训练循环

# 结合前文:自定义 Model 子类 + GradientTape 训练循环
# 完全控制前向、损失、梯度与更新

第 1 / 24 节