Article
数据加载（tf.keras.datasets / tf.data.Dataset）
1. 内置数据集
内置数据集形式均为 ((x_train, y_train), (x_test, y_test))，x 为自变量，y 为因变量。获取数据集时需调用 load_data()。
a. boston_housing — 波士顿房价回归数据集
boston_housing = tf.keras.datasets.boston_housing
(x_train, y_train), (x_test, y_test) = boston_housing.load_data()b. cifar10 — CIFAR10 小图像分类数据集
cifar10 = tf.keras.datasets.cifar10
(x_train, y_train), (x_test, y_test) = cifar10.load_data()c. cifar100 — CIFAR100 小图像分类数据集
cifar100 = tf.keras.datasets.cifar100
(x_train, y_train), (x_test, y_test) = cifar100.load_data()d. fashion_mnist — 时尚 MNIST 数据集
fashion_mnist = tf.keras.datasets.fashion_mnist
(x_train, y_train), (x_test, y_test) = fashion_mnist.load_data()e. imdb — IMDB 情绪分类数据集
imdb = tf.keras.datasets.imdb
(x_train, y_train), (x_test, y_test) = imdb.load_data()f. mnist — MNIST 手写数字数据集
mnist = tf.keras.datasets.mnist
(x_train, y_train), (x_test, y_test) = mnist.load_data()g. reuters — 路透社主题分类数据集
reuters = tf.keras.datasets.reuters
(x_train, y_train), (x_test, y_test) = reuters.load_data()2. 数据集 Dataset 的创建
a. 从输入数据创建数据集
dataset = tf.data.Dataset.from_tensor_slices([1, 2, 3])b. 将 pandas DataFrame 转换为数据集
dataset = tf.data.Dataset.from_tensor_slices([df['x1'].values, df['x2'].values, df['x3'].values, df['y'].values])
dataset = tf.data.Dataset.from_tensor_slices([dict(df), df['y'].values])c. 读取 CSV 文件
dataset = tf.data.experimental.CsvDataset(["file1.txt", "file2.txt"], record_defaults=[0, 0, 0, 0, 0])d. 读取文本文件
dataset = tf.data.TextLineDataset(["file1.txt", "file2.txt"])e. 读取 TFRecord（优先使用，性能较高）
dataset = tf.data.TFRecordDataset(["file1.tfrecords", "file2.tfrecords"])f. 读取二进制文件
dataset = tf.data.FixedLengthRecordDataset(
filenames=['/tmp/fixed_length0.bin', '/tmp/fixed_length1.bin'],
record_bytes=2, header_bytes=6, footer_bytes=6
)g. 创建范围数组
dataset = tf.data.Dataset.range(2)h. 使用 keras.utils 从目录读取
dataset = tf.keras.utils.audio_dataset_from_directory(main_directory, labels='inferred')
dataset = tf.keras.utils.image_dataset_from_directory(main_directory, labels='inferred')
dataset = tf.keras.utils.text_dataset_from_directory(main_directory, labels='inferred')i. 使用 read_file 读取单个文件
tf.io.read_file('test.csv')3. 数据集 Dataset 的拆分与合并
a. 将单个数据集拆分为训练集和测试集
# 定义训练集和验证集比例
train_ratio = 0.8
# 获取数据集的大小
total_size = len(df)
# 计算训练集的大小
train_size = int(total_size * train_ratio)
# 拆分数据集
train_dataset = dataset.take(train_size)
valid_dataset = dataset.skip(train_size)b. 多个数据集以抽样方式合并
# 定义权重
weights = [0.8, 0.2]
# 对多个数据集按权重合并
combined_dataset = tf.data.experimental.sample_from_datasets([dataset1, dataset2], weights)c. 多个数据集串联合并
concat_dataset = dataset1.concatenate(dataset2)d. 多个数据集拉链合并
zip_dataset = tf.data.Dataset.zip((dataset1, dataset2))e. 数据集解析
解析为训练集 — 方法 1（普通函数）：
注意：返回结果将特征列以 tuple 封装，标签列单独一列，外层再用 tuple 封装。常见的另一种形式省略了外层 tuple，但实际上返回的仍然是 tuple。
def parse_row_with_label(*fields):
input1 = tf.stack([str(fields[0])])
input2 = tf.stack([str(fields[1])])
input3 = tf.stack([int(fields[2]), int(fields[3])])
label = fields[4]
# 常见的另一种返回形式：(input1, input2, input3), label
return ((input1, input2, input3), label)
parsed_dataset = dataset.map(parse_row_with_label)解析为训练集 — 方法 2（lambda 函数）：
parse_row_with_label = lambda *fields: (
(tf.stack([str(fields[0])]), tf.stack([str(fields[1])]), tf.stack([int(fields[2]), int(fields[3])])),
fields[4]
)
parsed_dataset = dataset.map(parse_row_with_label)解析为预测集 — 方法 1（普通函数）：
注意：返回结果将特征列以 tuple 封装，外层再用 tuple 封装。预测集的返回形式中，外层必须是
(...,)以防止自动转化为单层 tuple。
def parse_row_without_label(*fields):
input1 = tf.stack([str(fields[0])])
input2 = tf.stack([str(fields[1])])
input3 = tf.stack([int(fields[2]), int(fields[3])])
# 注意预测集的返回形式：外层必须是 (...,)，防止自动转化为单层 tuple
return ((input1, input2, input3),)
predict_dataset = dataset.map(parse_row_without_label)解析为预测集 — 方法 2（lambda 函数）：
parse_row_without_label = lambda *fields: (
(tf.stack([str(fields[0])]), tf.stack([str(fields[1])]), tf.stack([int(fields[2]), int(fields[3])])),
)
predict_dataset = dataset.map(parse_row_without_label)f. 数据集解析完整示例
# train.csv 示例数据：
# id,f1,f2,f3,f4,label
# uid1,teacher,approve,1,100,1
# uid1,teacher,approve,2,100,1
# uid2,teacher,finish,3,90,1
# uid3,teacher,regist,4,80,0
# uid4,teacher,regist,5,70,0
# uid5,worker,finish,1,80,0
# uid6,worker,finish,2,70,0
# uid7,worker,regist,3,70,0
# uid8,student,approve,4,80,1
# uid9,student,approve,5,80,1
# uid10,student,finish,6,80,0
# uid11,student,regist,7,80,0
import tensorflow as tf
import pandas as pd
import numpy as np
# 读取 CSV 文件
df = pd.read_csv("train.csv", header=0)
record_defaults = [[''], [''], [''], [0.0], [0.0]]
# 创建 CSV 数据集，选择需要读取的列
dataset = tf.data.experimental.CsvDataset(
"train.csv",
record_defaults=record_defaults,
header=True,
select_cols=[1, 2, 3, 4, 5]
)
# 解析为训练集（lambda 方式）
# 注意：返回结果将特征列以 tuple 封装，标签列单独一列，外层再用 tuple 封装
# 常见的另一种形式省略了外层 tuple，但实际上返回的仍然是 tuple
parse_row_with_label = lambda *fields: (
(tf.stack([str(fields[0])]), tf.stack([str(fields[1])]), tf.stack([int(fields[2]), int(fields[3])])),
fields[4]
)
parsed_dataset = dataset.map(parse_row_with_label)
# 解析为预测集（lambda 方式）
# 注意返回形式：特征列必须放在一个 tuple 中，外层也是 tuple，因此外层必须是 (...,)
parse_row_without_label = lambda *fields: (
(tf.stack([str(fields[0])]), tf.stack([str(fields[1])]), tf.stack([int(fields[2]), int(fields[3])])),
)
predict_dataset = dataset.map(parse_row_without_label)
# 设置 batch 大小
BATCH_SIZE = 32
# 批处理数据集
batched_dataset = parsed_dataset.batch(BATCH_SIZE)
# 构建模型
input1_layer_col1 = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
input1_layer_col2 = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
input2_layer = tf.keras.layers.Input(shape=(2,), dtype=tf.float32)
lookup_layer1 = tf.keras.layers.StringLookup(vocabulary=['teacher', 'worker', 'student'])(input1_layer_col1)
lookup_layer2 = tf.keras.layers.StringLookup(vocabulary=['regist', 'finish', 'approve'])(input1_layer_col2)
concat_lookup_layer = tf.keras.layers.Concatenate()([lookup_layer1, lookup_layer2])
emb_layer = tf.keras.layers.Embedding(input_dim=2, output_dim=4)(concat_lookup_layer)
flatten_layer = tf.keras.layers.Flatten()(emb_layer)
norm_layer = tf.keras.layers.Normalization()(input2_layer)
concat_layer = tf.keras.layers.Concatenate(axis=-1)([flatten_layer, norm_layer])
dense_layer = tf.keras.layers.Dense(16, activation='relu')(concat_layer)
output = tf.keras.layers.Dense(1, activation='sigmoid')(dense_layer)
model = tf.keras.Model(inputs=(input1_layer_col1, input1_layer_col2, input2_layer), outputs=[output])
model.compile(optimizer='adam', loss='binary_crossentropy')
# 训练模型
model.fit(batched_dataset, epochs=10)
# 预测数据
model.predict(predict_dataset)4. 数据集 Dataset 的查看
a. 查看 tf.data.Dataset 的内容
# 将 Dataset 内部的值转为 NumPy
for i in dataset.as_numpy_iterator():
print(i)
# 查看 Dataset 内部的 Tensor
for i in dataset:
print(i)
# 查看 batch 过的 Dataset（每次遍历取出 batch 个值，使用 Tensor 封装）
for i in dataset.unbatch():
print(i)b. 查看 Dataset 大小（小数据集）
count = 0
for _ in dataset:
count += 1
print(f"dataset size: {count}")c. 查看 Dataset 大小（大数据集）
注意：配置过 repeat 的无限数据集返回
0，通过 map 和 filter 创建的数据集返回-1，通过其他操作构建且编译时无法确定大小的数据集返回-2。
cardinality = tf.data.experimental.cardinality(dataset)
print("Dataset size:", cardinality.numpy())d. 查看数据集结构（元素的结构、类型、形状）
dataset.element_spece. 查看数据集结构（元素内容）
for element in dataset.take(1).as_numpy_iterator():
print(element)5. 数据集 Dataset 的元素类型
a. 单张量类型 — 每个元素都是单独的张量，例如一个图像或一个序列
# 创建一个包含单个张量的数据集
dataset = tf.data.Dataset.from_tensor_slices(tf.random.uniform([10, 28, 28]))
# 遍历数据集并打印数据
for image in dataset:
print(image.shape) # (28, 28)b. 元组类型 — 每个元素包含多个相关张量，例如特征和标签
# 创建一个包含元组的数据集
features = tf.random.uniform([10, 28, 28])
labels = tf.random.uniform([10], maxval=10, dtype=tf.int32)
dataset = tf.data.Dataset.from_tensor_slices((features, labels))
# 遍历数据集并打印数据
for feature, label in dataset:
print(f"Feature shape: {feature.shape}, Label: {label}")c. 字典类型 — 具有命名键的数据，每个键对应特定特征或标签
# 创建一个包含字典的数据集
features = {
'image': tf.random.uniform([10, 28, 28]),
'audio': tf.random.uniform([10, 10000]),
}
labels = tf.random.uniform([10], maxval=10, dtype=tf.int32)
dataset = tf.data.Dataset.from_tensor_slices((features, labels))
# 遍历数据集并打印数据
for feature_dict, label in dataset:
print(f"Image shape: {feature_dict['image'].shape}, Audio shape: {feature_dict['audio'].shape}, Label: {label}")d. 有序字典类型 — 与字典类型类似，但保持插入顺序
from collections import OrderedDict
# 创建一个包含有序字典的数据集
features = OrderedDict([
('image', tf.random.uniform([10, 28, 28])),
('audio', tf.random.uniform([10, 10000])),
])
labels = tf.random.uniform([10], maxval=10, dtype=tf.int32)
dataset = tf.data.Dataset.from_tensor_slices((features, labels))
# 遍历数据集并打印数据
for feature_dict, label in dataset:
print(f"Image shape: {feature_dict['image'].shape}, Audio shape: {feature_dict['audio'].shape}, Label: {label}")6. 数据集 Dataset 的结构
a. 构造 x 形式数据集
import tensorflow as tf
# 创建特征数据
x = tf.random.uniform([10, 28, 28]) # 生成 10 个 28*28 的随机数矩阵，元素值介于 0-1 之间
# 构造数据集
dataset_x = tf.data.Dataset.from_tensor_slices(x)
# 查看数据集的结构
print("Dataset structure (x):", dataset_x.element_spec)b. 构造 (x,) 形式数据集
# 构造数据集 (x,)，注意元组中的逗号
dataset_x_tuple = tf.data.Dataset.from_tensor_slices((x,))
# 查看数据集的结构
print("Dataset structure (x,):", dataset_x_tuple.element_spec)c. 构造 (x, y) 形式数据集
# 创建标签数据
y = tf.random.uniform([10], maxval=10, dtype=tf.int32)
# 构造数据集 (x, y)
dataset_xy = tf.data.Dataset.from_tensor_slices((x, y))
# 查看数据集的结构
print("Dataset structure (x, y):", dataset_xy.element_spec)d. 构造 (x, y, sample_weight) 形式数据集
# 创建样本权重数据
sample_weight = tf.random.uniform([10])
# 构造数据集 (x, y, sample_weight)
dataset_xy_sample_weight = tf.data.Dataset.from_tensor_slices((x, y, sample_weight))
# 查看数据集的结构
print("Dataset structure (x, y, sample_weight):", dataset_xy_sample_weight.element_spec)7. 数据集 Dataset 的转换
数据读取后的常见操作：
a. map 操作 — 对每个元素应用函数
dataset = dataset.map(lambda x: x * 2)b. filter 操作 — 根据条件过滤元素
dataset = dataset.filter(lambda x: x > 0)c. batch 操作 — 数据分批
dataset = dataset.batch(32)d. shuffle 操作 — 随机打乱数据
dataset = dataset.shuffle(buffer_size=10000)8. 数据集 Dataset 的遍历
a. iter 和 next — 使用 Python 迭代器遍历，元素是 Tensor 类型
iterator = iter(dataset)
next_element = next(iterator)
print(next_element.numpy())b. for…in 循环 — 直接使用 for 循环遍历，元素是 Tensor 类型
for element in dataset:
print(element.numpy())9. 数据集 Dataset 的性能优化
a. prefetch 操作 — 提前加载数据，加速训练
dataset = dataset.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)b. cache 操作 — 缓存数据，避免重复读取
dataset = dataset.cache()c. 常用组合 — batch、shuffle、repeat、prefetch
dataset = dataset.batch(200).cache().shuffle(len(dataset)).repeat().prefetch(tf.data.AUTOTUNE)d. interleave 操作 — flat_map 的升级版，对数据集中的元素进行交错（interleave）处理，适用于需要从多个数据源并行读取数据的场景
tf.data.Dataset.interleave(
map_func, # 接受一个输入元素，返回一个数据集
cycle_length, # 并发处理的元素数量
num_parallel_calls, # 并行调用 map_func 的数量，可设为 tf.data.AUTOTUNE
deterministic, # 布尔值，True 按确定性顺序读取，False 可提高性能
block_length # 从每个数据集读取的连续元素数量，默认为 1
)e. 性能优化完整示例
import tensorflow as tf
# 文件名
filenames = [
"/home/lijianping/model_data/file1.csv",
"/home/lijianping/model_data/file2.csv",
"/home/lijianping/model_data/file3.csv",
]
# 根据文件名加载文件
def load_file(filename):
return tf.data.TextLineDataset(filename).skip(1)
# 根据文件名读取 csv 文件并解析
def parse_csv(line):
record_defaults = [[''], [0.0], [0.0], [0.0], [0]]
uid, feature1, feature2, feature3, label = tf.io.decode_csv(line, record_defaults)
features = tf.stack([feature1, feature2, feature3], axis=0)
label = tf.cast(label, tf.float32)
return features, label
# ----- 流式处理版本 1（flat_map）-----
dataset = tf.data.Dataset.from_tensor_slices(filenames)\
.flat_map(load_file)\
.map(parse_csv, num_parallel_calls=tf.data.AUTOTUNE)\
.batch(32)\
.prefetch(tf.data.AUTOTUNE)
# ----- 流式处理版本 2（interleave，性能更优）-----
dataset = tf.data.Dataset.from_tensor_slices(filenames)\
.interleave(load_file, cycle_length=3, num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True, block_length=2)\
.map(parse_csv, num_parallel_calls=tf.data.AUTOTUNE)\
.batch(32)\
.prefetch(tf.data.AUTOTUNE)
# 模型构建
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(3,)),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 模型编译
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 模型训练
model.fit(dataset, epochs=10)
# 模型保存
tf.keras.models.save_model(model, "/home/lijianping/model_files/interleave")
# 模型加载
model = tf.keras.models.load_model("/home/lijianping/model_files/interleave")
# 从 csv 文件中解析特征列（预测用）
def parse_features(line):
record_defaults = [[''], [0.0], [0.0], [0.0], [0]]
uid, feature1, feature2, feature3, label = tf.io.decode_csv(line, record_defaults)
features = tf.stack([feature1, feature2, feature3], axis=0)
return features
# 预测数据集 — 版本 1（flat_map）
dataset_predict = tf.data.Dataset.from_tensor_slices(filenames)\
.flat_map(load_file)\
.map(parse_features, num_parallel_calls=tf.data.AUTOTUNE)\
.batch(32)\
.prefetch(tf.data.AUTOTUNE)
# 预测数据集 — 版本 2（interleave）
dataset_predict = tf.data.Dataset.from_tensor_slices(filenames)\
.interleave(load_file, cycle_length=3, num_parallel_calls=tf.data.AUTOTUNE,
deterministic=True, block_length=2)\
.map(parse_features, num_parallel_calls=tf.data.AUTOTUNE)\
.batch(32)\
.prefetch(tf.data.AUTOTUNE)
# 模型预测
model.predict(dataset_predict)10. 执行模式
默认为 eager 即时执行模式。
a. TensorFlow 1.x 版本中关闭即时执行模式
# 禁用 eager execution
tf.compat.v1.disable_eager_execution()b. TensorFlow 2.x 版本默认启用即时执行模式
# 禁用 eager execution
tf.config.run_functions_eagerly(True)
# 查看当前是否为 eager execution
tf.config.functions_run_eagerly()11. 张量 Tensor 的重构
a. 获取张量形状
tf.shape(input, name=None, out_type=tf.int32)b. 获取张量元素数量
tf.size(input, name=None, out_type=tf.int32)c. 获取张量的阶数
tf.rank(input, name=None)d. 重塑
tf.reshape(tensor, shape, name=None)e. 扩充维度
在指定轴位置增加维度：axis 为正数时在指定轴前增加维度，axis 为负数时在指定轴后增加维度（相当于在内层增加一层括号，增加维度中元素数量为 1）。
典型报错：
ValueError: Input 0 of layer sequential_12 is incompatible with the layer: expected min_ndim=4, found ndim=2— 此时通常需要扩充维度。
image = tf.zeros([10, 10, 3])
tf.expand_dims(image, axis=0) # 维度变为 [1, 10, 10, 3]
tf.expand_dims(image, axis=-1) # 维度变为 [10, 10, 3, 1]f. 缩减维度
减少元素数量为 1 的维度。当要剔除的维度元素个数不为 1 时，会报错 InvalidArgumentError。
image = tf.zeros([1, 10, 1, 10, 1])
tf.squeeze(image) # 剔除所有元素数量为 1 的维度
tf.squeeze(image, axis=[0, 2, 4]) # 剔除指定维度g. 转置
# 将形状为 (3, 4) 的张量转置为 (4, 3)
transposed_tensor = tf.transpose(tensor)h. 堆叠
假设有 N 个张量，形状均为 (A, B, C)。
x = tf.constant([1, 4])
y = tf.constant([2, 5])
z = tf.constant([3, 6])
# 横向堆叠，形成新的形状 (N, A, B, C)：axis=0
tf.stack([x, y, z], axis=0)
# <tf.Tensor: shape=(3, 2), dtype=int32,
# numpy=array([[1, 4],[2, 5],[3, 6]], dtype=int32)>
# 纵向堆叠，形成新的形状 (A, N, B, C)：axis=1
tf.stack([x, y, z], axis=1)
# <tf.Tensor: shape=(2, 3), dtype=int32,
# numpy=array([[1, 2, 3], [4, 5, 6]], dtype=int32)>i. 连接
沿指定轴连接多个张量。
# 顺序索引：axis=0
t1 = [[1, 2, 3], [4, 5, 6]]
t2 = [[7, 8, 9], [10, 11, 12]]
tf.concat([t1, t2], 0)
# <tf.Tensor: shape=(4, 3), dtype=int32,
# numpy=array([[1, 2, 3], [4, 5, 6], [7, 8, 9], [10, 11, 12]], dtype=int32)>
# 倒序索引：axis=-1
t1 = [[[1, 2], [2, 3]], [[4, 4], [5, 3]]]
t2 = [[[7, 4], [8, 4]], [[2, 10], [15, 11]]]
tf.concat([t1, t2], -1)
# <tf.Tensor: shape=(2, 2, 4), dtype=int32>
# 沿新轴 newaxis 连接（等价于 stack）
tf.concat([tf.expand_dims(t, axis) for t in tensors], axis)
# 或
tf.stack(tensors, axis=axis)j. 平铺（tile）
将第一个张量按第二个张量进行平铺复制。
a = tf.constant([[1, 2, 3], [4, 5, 6]], tf.int32)
b = tf.constant([1, 2], tf.int32)
tf.tile(a, b)
# <tf.Tensor: shape=(2, 6), dtype=int32,
# numpy=array([[1, 2, 3, 1, 2, 3], [4, 5, 6, 4, 5, 6]], dtype=int32)>k. 重复（repeat）
沿指定轴对元素进行复制。axis=n，n 越大则元素越靠近 array 的内层。
# 一维张量，axis=0
tf.repeat(['a', 'b', 'c'], repeats=[3, 0, 2], axis=0)
# <tf.Tensor: shape=(5,), dtype=string,
# numpy=array([b'a', b'a', b'a', b'c', b'c'], dtype=object)>
# 二维张量，axis=1
tf.repeat([[1, 2], [3, 4]], repeats=[2, 3], axis=1)
# <tf.Tensor: shape=(5, 2), dtype=int32,
# numpy=array([[1, 2], [1, 2], [3, 4], [3, 4], [3, 4]], dtype=int32)>
# 单个值，axis=None 时结果会被展平
tf.repeat(3, repeats=4)
# <tf.Tensor: shape=(4,), dtype=int32,
# numpy=array([3, 3, 3, 3], dtype=int32)>
# axis 可根据情况默认适配
tf.repeat([[1, 2], [3, 4]], repeats=2)
# <tf.Tensor: shape=(8,), dtype=int32,
# numpy=array([1, 1, 2, 2, 3, 3, 4, 4], dtype=int32)>12. 特定数据类型的处理
注意：
tf.feature_column即将弃用，后续建议改用tf.keras.layers层进行特征预处理。
a. 结构化数据
tf.feature_column 旧版 API：
tf.feature_column.numeric_column— 处理数值型特征（如年龄、收入）
numeric_feature = tf.feature_column.numeric_column("numeric_feature")tf.feature_column.categorical_column_with_vocabulary_list— 通过词汇表处理离散类别型特征
categorical_feature = tf.feature_column.categorical_column_with_vocabulary_list(
"categorical_feature", vocabulary_list=["cat", "dog", "bird"]
)tf.feature_column.embedding_column— 将类别型特征嵌入为密集向量
embedding_feature = tf.feature_column.embedding_column(categorical_feature, dimension=8)tf.feature_column.bucketized_column— 将数值型特征分桶离散化
bucketized_feature = tf.feature_column.bucketized_column(numeric_feature, boundaries=[0, 1, 2, 3])tf.feature_column.sequence_categorical_column_with_identity— 处理序列数据的类别型特征
sequence_categorical_feature = tf.feature_column.sequence_categorical_column_with_identity(
"sequence_categorical_feature", num_buckets=10
)tf.feature_column.indicator_column— 将类别型特征转换为独热编码
indicator_feature = tf.feature_column.indicator_column(categorical_feature)tf.feature_column.crossed_column— 特征交叉组合
crossed_feature = tf.feature_column.crossed_column(
[categorical_feature1, categorical_feature2], hash_bucket_size=1000
)tf.keras.layers 新版 API 替代方案：
tf.keras.layers.Discretization— 对连续数值特征分桶离散化
import numpy as np
import tensorflow as tf
train_x = np.array([10, 20, 30]).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
layer0 = tf.keras.layers.Input(shape=(1,))
layer1 = tf.keras.layers.Discretization(bin_boundaries=[5, 15, 25])(layer0)
output = tf.keras.layers.Dense(1, activation='sigmoid')(layer1)
model = tf.keras.Model(inputs=[layer0], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit([train_x], train_y)
predict = model.predict(train_x)
print(predict)tf.keras.layers.Hashing— 对离散型特征进行哈希处理，映射到固定大小的整数空间
import numpy as np
import tensorflow as tf
train_x = np.array(["t1", 't2', 't3']).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
layer0 = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
# 定义 StringLookup 层并初始化
layer1 = tf.keras.layers.StringLookup()
layer1.adapt(train_x)
# 注意：括号内是 layer1(layer0)
layer2 = tf.keras.layers.Hashing(num_bins=2)(layer1(layer0))
output = tf.keras.layers.Dense(1, activation='sigmoid')(layer2)
model = tf.keras.Model(inputs=[layer0], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit([train_x], train_y)
predict = model.predict(train_x)
print(predict)tf.keras.layers.CategoryEncoding— 将整数类别编码为固定大小向量，支持one_hot和multi_hot
import numpy as np
import tensorflow as tf
train_x = np.array(["t1", 't2', 't3']).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
layer0 = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
layer1 = tf.keras.layers.StringLookup()
layer1.adapt(train_x)
print(layer1.get_vocabulary())
layer2 = tf.keras.layers.CategoryEncoding(
num_tokens=len(layer1.get_vocabulary()), output_mode='one_hot'
)(layer1(layer0))
output = tf.keras.layers.Dense(1, activation='sigmoid')(layer2)
model = tf.keras.Model(inputs=[layer0], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit([train_x], train_y)
predict = model.predict(train_x)
print(predict)tf.keras.layers.StringLookup/tf.keras.layers.IntegerLookup— 将字符串/整数特征映射到整数索引
import numpy as np
import tensorflow as tf
train_x = np.array(["t1", 't2', 't3']).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
layer0 = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
layer1 = tf.keras.layers.StringLookup()
layer1.adapt(train_x)
output = tf.keras.layers.Dense(1, activation='sigmoid')(layer1(layer0))
model = tf.keras.Model(inputs=[layer0], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit([train_x], train_y)
predict = model.predict(train_x)
print(predict)tf.keras.layers.experimental.preprocessing.HashedCrossing— 创建特征交叉组合并用哈希映射到固定维度
import numpy as np
import tensorflow as tf
# 构造示例数据
a_train = np.array(['cat', 'dog', 'fish', 'bird', 'cat']).reshape(-1, 1)
b_train = np.array(['red', 'blue', 'green', 'yellow', 'orange']).reshape(-1, 1)
y_train = np.array([1, 0, 0, 1, 0]).reshape(-1, 1)
input_a = tf.keras.layers.Input(shape=(1,), dtype=tf.string, name='input_a')
input_b = tf.keras.layers.Input(shape=(1,), dtype=tf.string, name='input_b')
crossing_layer = tf.keras.layers.experimental.preprocessing.HashedCrossing(num_bins=10)
crossed_feature = crossing_layer([input_a, input_b])
output = tf.keras.layers.Dense(1, activation='sigmoid')(crossed_feature)
model = tf.keras.Model(inputs=[input_a, input_b], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 构建数据集
dataset = tf.data.Dataset.from_tensor_slices(((a_train, b_train), y_train)).batch(1)
model.fit(dataset, epochs=10, verbose=1)tf.keras.layers.Embedding— 将词汇表中词汇映射到密集向量，需先转换为整数索引
import numpy as np
import tensorflow as tf
train_x = np.array(["t1", "t2", "t3"]).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
input_layer = tf.keras.layers.Input(shape=(1,), dtype=tf.string)
lookup_layer = tf.keras.layers.StringLookup(mask_token=None)
lookup_layer.adapt(train_x)
embedded_layer = tf.keras.layers.Embedding(
input_dim=len(lookup_layer.get_vocabulary()), output_dim=3
)(lookup_layer(input_layer))
flattened_layer = tf.keras.layers.Flatten()(embedded_layer)
output = tf.keras.layers.Dense(1, activation='sigmoid')(flattened_layer)
model = tf.keras.Model(inputs=input_layer, outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit(train_x, train_y, epochs=10)
predict = model.predict(train_x)
print(predict)tf.keras.layers.Normalization— 标准化数值特征（零均值、单位方差）
import numpy as np
import tensorflow as tf
train_x = np.array([1, 2, 3]).reshape(-1, 1)
train_y = np.array([1, 0, 0]).reshape(-1, 1)
input_layer = tf.keras.layers.Input(shape=(1,), dtype=tf.float32)
norm_layer = tf.keras.layers.Normalization()(input_layer)
output = tf.keras.layers.Dense(1, activation='sigmoid')(norm_layer)
model = tf.keras.Model(inputs=input_layer, outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit(train_x, train_y, epochs=10)
predict = model.predict(train_x)
print(predict)tf.feature_column 与 tf.keras.layers 对照表：
| feature_column | Keras 层 |
|---|---|
tf.feature_column.bucketized_column | tf.keras.layers.Discretization |
tf.feature_column.categorical_column_with_hash_bucket | tf.keras.layers.Hashing |
tf.feature_column.categorical_column_with_identity | tf.keras.layers.CategoryEncoding |
tf.feature_column.categorical_column_with_vocabulary_file | tf.keras.layers.StringLookup / IntegerLookup |
tf.feature_column.categorical_column_with_vocabulary_list | tf.keras.layers.StringLookup / IntegerLookup |
tf.feature_column.crossed_column | tf.keras.layers.experimental.preprocessing.HashedCrossing |
tf.feature_column.embedding_column | tf.keras.layers.Embedding |
tf.feature_column.indicator_column | tf.keras.layers.CategoryEncoding（output_mode='one_hot' 或 'multi_hot'） |
tf.feature_column.numeric_column | tf.keras.layers.Normalization |
tf.feature_column.sequence_categorical_column_with_hash_bucket | tf.keras.layers.Hashing |
tf.feature_column.sequence_categorical_column_with_identity | tf.keras.layers.CategoryEncoding |
tf.feature_column.sequence_categorical_column_with_vocabulary_file | tf.keras.layers.StringLookup / IntegerLookup / TextVectorization |
tf.feature_column.sequence_categorical_column_with_vocabulary_list | tf.keras.layers.StringLookup / IntegerLookup / TextVectorization |
tf.feature_column.sequence_numeric_column | tf.keras.layers.Normalization |
tf.feature_column.weighted_categorical_column | tf.keras.layers.CategoryEncoding |
b. 图像数据处理
tf.image.resize— 调整图像大小
resized_image = tf.image.resize(image, size=(new_height, new_width))tf.image.resize_with_crop_or_pad— 裁剪或填充图像以匹配目标大小
cropped_padded_image = tf.image.resize_with_crop_or_pad(image, target_height, target_width)tf.image.crop_to_bounding_box— 从图像中裁剪指定区域
cropped_image = tf.image.crop_to_bounding_box(image, offset_height, offset_width, target_height, target_width)tf.image.flip_left_right/tf.image.flip_up_down— 左右/上下翻转
flipped_left_right_image = tf.image.flip_left_right(image)
flipped_up_down_image = tf.image.flip_up_down(image)tf.image.rot90— 逆时针旋转 90 度
rotated_image = tf.image.rot90(image)tf.image.adjust_brightness/tf.image.adjust_contrast— 调整亮度和对比度
brightened_image = tf.image.adjust_brightness(image, delta)
contrasted_image = tf.image.adjust_contrast(image, contrast_factor)tf.image.adjust_hue/tf.image.adjust_saturation/tf.image.adjust_gamma— 调整色调、饱和度和伽马校正
hue_adjusted_image = tf.image.adjust_hue(image, delta)
saturated_image = tf.image.adjust_saturation(image, saturation_factor)
gamma_corrected_image = tf.image.adjust_gamma(image, gamma)tf.image.per_image_standardization— 对每个图像进行标准化
standardized_image = tf.image.per_image_standardization(image)tf.image.normalize— 对图像进行归一化
normalized_image = tf.image.normalize(image, mean=[mean_r, mean_g, mean_b], std=[std_r, std_g, std_b])tf.image.rgb_to_grayscale— 将 RGB 图像转为灰度图像
grayscale_image = tf.image.rgb_to_grayscale(image)tf.image.random_flip_left_right/tf.image.random_flip_up_down— 随机左右/上下翻转
randomly_flipped_left_right_image = tf.image.random_flip_left_right(image)
randomly_flipped_up_down_image = tf.image.random_flip_up_down(image)c. 字符串处理
tf.strings.join— 连接字符串列表
strings = ["Hello", "TensorFlow"]
joined_string = tf.strings.join(strings, separator=" ")tf.strings.split— 按分隔符切割字符串
string = "Hello, TensorFlow"
split_string = tf.strings.split(string, sep=", ")tf.strings.substr— 截取子串
string = "Hello, TensorFlow"
substring = tf.strings.substr(string, pos=7, len=10)tf.strings.encode/tf.strings.decode— 编码与解码
string = "Hello, TensorFlow"
encoded_string = tf.strings.encode(string, "utf-8")
decoded_string = tf.strings.decode(encoded_string, "utf-8")tf.strings.unicode_encode/tf.strings.unicode_decode— Unicode 编码与解码
string = "你好，TensorFlow"
encoded_string = tf.strings.unicode_encode(string, "utf-8")
decoded_string = tf.strings.unicode_decode(encoded_string, "utf-8")tf.strings.format— 格式化字符串
name = "Alice"
age = 30
formatted_string = tf.strings.format("My name is {} and I'm {} years old.", (name, age))tf.strings.regex_replace— 正则替换
input_string = "abc123def456"
replaced_string = tf.strings.regex_replace(input_string, pattern="[0-9]", rewrite="")tf.strings.strip/tf.strings.unicode_script— 去除首尾空格 / 获取 Unicode 脚本
string = " TensorFlow "
stripped_string = tf.strings.strip(string)
script = tf.strings.unicode_script(string)tf.strings.to_number— 字符串转数字
numeric_string = "123.45"
numeric_value = tf.strings.to_number(numeric_string, out_type=tf.float32)d. 音频数据处理（tf.audio）
tf.audio.decode_wav— 解码 WAV 格式音频文件，返回音频数据和采样率
audio_data, sample_rate = tf.audio.decode_wav(tf.io.read_file("audio_file.wav"))tf.audio.encode_wav— 编码音频数据为 WAV 格式
wav_encoded_data = tf.audio.encode_wav(audio_data, sample_rate)
tf.io.write_file("output_audio.wav", wav_encoded_data)-
tf.audio.decode_audio/tf.audio.encode_audio— 支持处理其他音频格式（类似decode_wav/encode_wav） -
tf.audio.spectrogram— 计算音频信号谱图
spectrogram = tf.audio.spectrogram(audio_data, window_size=1024, stride=64)tf.audio.mfccs_from_log_mel_spectrograms— 计算 MFCC（梅尔频率倒谱系数）
mel_spectrogram = tf.audio.decode_wav(tf.io.read_file("audio_file.wav"))
mfccs = tf.audio.mfccs_from_log_mel_spectrograms(tf.math.log(mel_spectrogram + 1e-6))tf.audio.crop/tf.audio.pad— 裁剪或填充音频信号长度
cropped_audio = tf.audio.crop(audio_data, start=0, stop=50000)
padded_audio = tf.audio.pad(audio_data, paddings=[[0, 10000]])tf.audio.resample— 重新采样音频信号
resampled_audio = tf.audio.resample(audio_data, target_sample_rate=16000)13. 创建张量
a. 使用常数创建张量 tf.constant
import tensorflow as tf
# 创建一个标量 (0-D) 张量
scalar = tf.constant(5)
# 创建一个向量 (1-D) 张量
vector = tf.constant([1, 2, 3])
# 创建一个矩阵 (2-D) 张量
matrix = tf.constant([[1, 2, 3], [4, 5, 6]])b. 从 NumPy 数组创建张量 tf.convert_to_tensor
import numpy as np
# 创建一个 NumPy 数组
numpy_array = np.array([[1, 2, 3], [4, 5, 6]])
# 转换为 TensorFlow 张量
tensor_from_numpy = tf.convert_to_tensor(numpy_array)c. 使用随机数生成张量
# 创建一个正态分布随机张量
random_normal_tensor = tf.random.normal(shape=[2, 3])
# 创建一个均匀分布随机张量
random_uniform_tensor = tf.random.uniform(shape=[2, 3], minval=0, maxval=1)d. 使用特定形状创建张量
# 创建一个全零张量
zeros_tensor = tf.zeros(shape=(2, 3))
# 创建一个全一张量
ones_tensor = tf.ones(shape=(2, 3))e. 使用序列生成张量 tf.range
# 创建一个等差数列张量
range_tensor = tf.range(start=0, limit=10, delta=2)14. 张量类型
所有节点都支持指定
name参数，用于后续可视化时进行区分。
a. 张量 tf.Tensor
tensor_constant = tf.constant([1, 2, 3]) # 创建常量张量
tensor_variable = tf.Variable([4, 5, 6]) # 创建可变张量
print("Constant Tensor:", tensor_constant.numpy()) # 使用 .numpy() 获取张量的值
sum_tensor = tensor_constant + tensor_variable # 加法操作
product_tensor = tensor_constant * tensor_variable # 乘法操作
print("Product Tensor:", product_tensor.numpy())
shape = tensor_constant.shape # 获取张量的形状
dtype = tensor_constant.dtype # 获取数据类型
element = tensor_constant[0].numpy() # 索引操作
subset = tensor_constant[1:3].numpy() # 切片操作
numpy_array = tensor_constant.numpy() # 将张量转换为 NumPy 数组b. 常量 tf.constant — 不可变，创建后不能被修改
a = tf.constant(1) # Tensor: shape=(), dtype=int32 — 标量
b = tf.constant([1, 2, 3]) # Tensor: shape=(3,), dtype=int32
c = tf.constant([[1, 2], [3, 4], [5, 6]]) # Tensor: shape=(3, 2), dtype=int32
d = tf.constant([[[1, 2], [3, 4]], [[5, 6], [7, 8]]]) # Tensor: shape=(2, 2, 2), dtype=int32
e = tf.constant([[[1, 2, 3], [4, 5, 6]]]) # Tensor: shape=(1, 2, 3), dtype=int32c. 维度长度解释
以 [[[1, 2, 3], [4, 5, 6]]] 为例，从左向右看各层括号内的元素：
- 第一个维度：最外层括号内的元素
[[1, 2, 3], [4, 5, 6]]，长度为 1 - 第二个维度：第二层括号内的元素
[1, 2, 3]、[4, 5, 6]，长度为 2 - 第三个维度：第三层括号内的元素
1, 2, 3，长度为 3
综上，形状为 (1, 2, 3)。
d. 变量 tf.Variable — 可变，常用于表示神经网络的权重和偏差
variable_tensor = tf.Variable([1, 2, 3]) # 创建可变张量
print("Initial Variable Tensor:", variable_tensor.numpy())
variable_tensor.assign([4, 5, 6]) # 修改张量的值
print("Modified Variable Tensor:", variable_tensor.numpy())
counter = tf.Variable(0) # 创建可变张量
counter.assign_add(1) # 增加张量的值
print("Counter after increment:", counter.numpy())
counter.assign_sub(1) # 减少张量的值
print("Counter after decrement:", counter.numpy())15. 张量计算
a. 加法 tf.add
a = tf.constant([1, 2, 3])
b = tf.constant([4, 5, 6])
result = tf.add(a, b)b. 减法 tf.subtract
a = tf.constant([1, 2, 3])
b = tf.constant([4, 5, 6])
result = tf.subtract(a, b)c. 乘法 tf.multiply（元素级乘法/点积）
如果形状不同，会尝试使用广播规则进行匹配。
a = tf.constant([1, 2, 3])
b = tf.constant([[4], [5], [6]])
result = tf.multiply(a, b)d. 除法 tf.divide
a = tf.constant([1, 2, 3])
b = tf.constant(4)
result = tf.divide(a, b)e. 幂运算 tf.pow
a = tf.constant([1, 2, 3])
b = tf.constant(4)
result = tf.pow(a, b)f. 指数 tf.exp
a = tf.constant([1, 2, 3])
result = tf.exp(a)g. 矩阵乘法 tf.matmul
a = tf.constant([[1, 2], [3, 4]])
b = tf.constant([[5, 6], [7, 8]])
result = tf.matmul(a, b)16. 文件读取
a. tf.data.Dataset.list_files — 通过模式匹配获取文件列表
dataset = tf.data.Dataset.list_files("/path/*.txt")
for file in dataset.as_numpy_iterator():
print(file.decode("utf-8"))b. tf.io.read_file — 读取文件所有内容到字符串张量
content = tf.io.read_file(file_path)c. tf.data.experimental.CsvDataset — 从 CSV 文件读取，生成内部元素为列表的数据集
record_defaults：csv 文件中各列的默认值header：csv 文件具有文件头时设为Trueselect_cols：指定选择哪些列进入 dataset（数值索引）
dataset = tf.data.experimental.CsvDataset(
file_path,
record_defaults=[[0], [''], [0], ['']],
header=True
)d. tf.data.experimental.make_csv_dataset — 从 CSV 文件读取，生成内部元素为 dict 的数据集
# 创建 CSV 数据集
csv_dataset = tf.data.experimental.make_csv_dataset(
csv_file_path,
batch_size=10, # 每个 batch 的大小
column_names=['uid', 'feature1', 'feature2', 'label'], # 列名列表
column_defaults=[tf.string, tf.float32, tf.float32, tf.int32], # 每列默认类型
label_name='label', # 标签列名，可缺省
na_value='', # 缺失值表示
num_epochs=1, # 读取数据集的次数
shuffle=False, # 是否打乱数据
ignore_errors=True, # 忽略解析错误
header=False, # CSV 文件没有标题行
select_columns=['uid', 'feature1', 'feature2', 'label'] # 选择要读取的列
)e. tf.data.TFRecordDataset — 从 TFRecord 文件中读取序列化的 Example 协议缓冲区
# 读取 TFRecord 文件
def parse_example(example_string):
feature_description = {
'id': tf.io.FixedLenFeature([], tf.int64, default_value=0),
'name': tf.io.FixedLenFeature([], tf.string, default_value=''),
'age': tf.io.FixedLenFeature([], tf.int64, default_value=0),
'city': tf.io.FixedLenFeature([], tf.string, default_value='')
}
example = tf.io.parse_single_example(example_string, feature_description)
return example
# 创建 TFRecord 数据集
dataset = tf.data.TFRecordDataset(file_path)
parsed_dataset = dataset.map(parse_example)
# 将数据集分割成特征和标签
def split_features_and_labels(parsed_record):
# 假设年龄是标签，其余为特征
features = {
'id': parsed_record['id'],
'name': parsed_record['name'],
'city': parsed_record['city']
}
label = parsed_record['age']
return features, label
# 使用 map() 应用 split 函数
labeled_dataset = parsed_dataset.map(split_features_and_labels)
# 遍历带有标签的数据集
for features, label in labeled_dataset:
print(f"Features: {features}")
print(f"Label: {label.numpy()}")17. 文件写入
a. tf.io.write_file — 将字符串张量写入文件
content = tf.constant("Hello, TensorFlow!")
file_path = 'example.txt'
tf.io.write_file(file_path, content)b. tf.io.serialize_tensor — 序列化张量到字节字符串
tensor = tf.constant([1.0, 2.0, 3.0, 4.0, 5.0])
serialized_tensor = tf.io.serialize_tensor(tensor)
print("Serialized tensor:")
print(serialized_tensor.numpy())c. tf.io.parse_tensor — 反序列化字节字符串为张量
deserialized_tensor = tf.io.parse_tensor(serialized_tensor, out_type=tf.float32)
print("Deserialized tensor:")
print(deserialized_tensor.numpy())d. tf.io.TFRecordWriter — 将 Example 协议缓冲区写入 TFRecord 文件
import tensorflow as tf
# 辅助函数：创建 Example 协议缓冲区
def _bytes_feature(value):
"""Returns a bytes_list from a string / byte."""
if isinstance(value, type(tf.constant(0))):
value = value.numpy()
return tf.train.Feature(bytes_list=tf.train.BytesList(value=[value]))
def _int64_feature(value):
"""Returns an int64_list from a bool / enum / int / uint."""
return tf.train.Feature(int64_list=tf.train.Int64List(value=[value]))
def serialize_example(id, name, age, city):
feature = {
'id': _int64_feature(id),
'name': _bytes_feature(name.encode('utf-8')),
'age': _int64_feature(age),
'city': _bytes_feature(city.encode('utf-8')),
}
example_proto = tf.train.Example(features=tf.train.Features(feature=feature))
return example_proto.SerializeToString()
# 写入 TFRecord 文件
def write_tfrecord(file_path, examples):
with tf.io.TFRecordWriter(file_path) as writer:
for example in examples:
writer.write(serialize_example(*example))
# 示例数据
examples = [
(1, 'Alice', 30, 'New York'),
(2, 'Bob', 25, 'San Francisco'),
(3, 'Charlie', 22, 'London')
]
# TFRecord 文件路径
file_path = 'example.tfrecord'
# 写入数据
write_tfrecord(file_path, examples)18. 文件列表
tf.io.matching_files — 返回匹配特定模式的文件名列表（返回二进制字符串）
pattern = '*.txt'
matching_files = tf.io.matching_files(pattern)
files = matching_files.numpy()19. 文件系统操作
a. tf.io.gfile.glob — 与 matching_files 类似，但返回明文字符串
pattern = '*.txt'
matching_files = tf.io.gfile.glob(pattern)b. tf.io.gfile.exists — 检查文件或目录是否存在
tf.io.gfile.exists(path)c. tf.io.gfile.listdir — 列出目录中的文件和子目录
tf.io.gfile.listdir(path)d. tf.io.gfile.makedirs — 创建目录及其父目录
tf.io.gfile.makedirs(path)e. tf.io.gfile.rename — 重命名文件或目录
tf.io.gfile.rename(old_path, new_path)f. tf.io.gfile.remove — 删除文件
tf.io.gfile.remove(path)g. tf.io.gfile.copy — 复制文件
tf.io.gfile.copy(path, other_path)20. 文件读取器
a. tf.data.TextLineDataset — 从文本文件中读取每一行
import tensorflow as tf
# 文件路径
file_path = 'example.txt'
# 使用 tf.data.TextLineDataset 读取文本文件
dataset = tf.data.TextLineDataset(file_path)
# 遍历数据集并打印每一行
for line in dataset:
print(line.numpy().decode('utf-8'))
# 示例：将每一行数据分割成单词列表
def split_words(line):
words = tf.strings.split(line, sep=' ')
return words
# 使用 map() 方法应用 split_words 函数
word_dataset = dataset.map(split_words)
# 遍历单词数据集并打印每个单词
for words in word_dataset:
print(words.numpy().tolist())b. tf.data.FixedLengthRecordDataset — 从固定长度记录文件中读取数据
import tensorflow as tf
# 文件路径
file_path = 'fixed_length_records.bin'
# 固定长度记录的长度（以字节为单位）
record_bytes = 10
# 使用 tf.data.FixedLengthRecordDataset 读取固定长度记录文件
dataset = tf.data.FixedLengthRecordDataset(file_path, record_bytes)
# 遍历数据集并打印每一条记录
for record in dataset:
print(record.numpy())
# 示例：将每一行数据转换为整数列表
def parse_record(record):
# 将字节字符串转换为整数
record_ints = tf.io.decode_raw(record, tf.int32)
return record_ints
# 使用 map() 方法应用 parse_record 函数
parsed_dataset = dataset.map(parse_record)
# 遍历解析后的数据集并打印每个记录
for parsed_record in parsed_dataset:
print(parsed_record.numpy().tolist())21. 数据解析
a. tf.io.decode_csv — 解码 CSV 行为张量
import tensorflow as tf
# CSV 字符串
csv_string = """
1,Alice,30,New York
2,Bob,25,San Francisco
3,Charlie,22,London
"""
# 定义 CSV 文件的列默认值
record_defaults = [[0], [''], [0], ['']]
# 使用 tf.io.decode_csv 解码 CSV 字符串
decoded_records = tf.io.decode_csv(csv_string, record_defaults, field_delim=',')b. tf.io.decode_raw — 将字节字符串解码为整数或浮点数
import tensorflow as tf
# 字节字符串
byte_string = b'\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00\x03\x00\x00\x00\x04\x00\x00\x00\x05'
# 使用 tf.io.decode_raw 解码字节字符串
decoded_integers = tf.io.decode_raw(byte_string, out_type=tf.int32)
decoded_floats = tf.io.decode_raw(byte_string, out_type=tf.float32)
# 打印解码后的整数和浮点数
print("Decoded Integers:")
print(decoded_integers.numpy())
print("Decoded Floats:")
print(decoded_floats.numpy())
# 示例：将解码后的整数和浮点数转换为列表
integer_list = decoded_integers.numpy().tolist()
float_list = decoded_floats.numpy().tolist()
# 打印转换后的列表
print("Integer List:")
print(integer_list)
print("Float List:")
print(float_list)c. tf.io.decode_image — 解码图像文件为张量
import tensorflow as tf
# 图像文件路径
image_path = 'example.jpg'
# 读取图像文件
image_data = tf.io.read_file(image_path)
# 解码图像
image = tf.io.decode_image(image_data, channels=3)
# 打印图像形状
print("Image shape:", image.shape)d. tf.io.parse_single_example — 解析单个序列化的 Example 协议缓冲区
import tensorflow as tf
# 创建 Example 协议缓冲区
example_proto = tf.train.Example(
features=tf.train.Features(
feature={
'int_feature': tf.train.Feature(int64_list=tf.train.Int64List(value=[1, 2, 3])),
'float_feature': tf.train.Feature(float_list=tf.train.FloatList(value=[1.0, 2.0, 3.0])),
'string_feature': tf.train.Feature(bytes_list=tf.train.BytesList(value=['hello'.encode(), 'world'.encode()])),
}
)
)
# 序列化 Example 协议缓冲区
serialized_example = example_proto.SerializeToString()
# 定义特征描述
feature_description = {
'int_feature': tf.io.FixedLenFeature([3], tf.int64),
'float_feature': tf.io.FixedLenFeature([3], tf.float32),
'string_feature': tf.io.FixedLenFeature([2], tf.string),
}
# 解析 Example 协议缓冲区
parsed_example = tf.io.parse_single_example(serialized_example, feature_description)
# 打印解析后的特征
print("Parsed Features:")
print(parsed_example)e. tf.io.parse_example — 批量解析多个序列化的 Example 协议缓冲区
import tensorflow as tf
# 创建多个 Example 协议缓冲区
example_protos = [
tf.train.Example(
features=tf.train.Features(
feature={
'int_feature': tf.train.Feature(int64_list=tf.train.Int64List(value=[i])),
'float_feature': tf.train.Feature(float_list=tf.train.FloatList(value=[i + 0.5])),
'string_feature': tf.train.Feature(bytes_list=tf.train.BytesList(value=[f'hello_{i}'.encode()])),
}
)
) for i in range(3)
]
# 序列化 Example 协议缓冲区
serialized_examples = [proto.SerializeToString() for proto in example_protos]
# 定义特征描述
feature_description = {
'int_feature': tf.io.FixedLenFeature([], tf.int64),
'float_feature': tf.io.FixedLenFeature([], tf.float32),
'string_feature': tf.io.FixedLenFeature([], tf.string),
}
# 解析 Example 协议缓冲区
parsed_examples = tf.io.parse_example(serialized_examples, feature_description)
# 打印解析后的特征
print("Parsed Examples:")
print(parsed_examples)f. tf.io.parse_sequence_example — 解析序列化的 SequenceExample 协议缓冲区
import tensorflow as tf
# 创建 SequenceExample 协议缓冲区
context_features = {
'length': tf.train.Feature(int64_list=tf.train.Int64List(value=[3]))
}
sequence_features = {
'int_feature': tf.train.FeatureList(feature=[tf.train.Feature(int64_list=tf.train.Int64List(value=[i])) for i in range(3)]),
'float_feature': tf.train.FeatureList(feature=[tf.train.Feature(float_list=tf.train.FloatList(value=[i + 0.5])) for i in range(3)]),
'string_feature': tf.train.FeatureList(feature=[tf.train.Feature(bytes_list=tf.train.BytesList(value=[f'hello_{i}'.encode()])) for i in range(3)]),
}
sequence_proto = tf.train.SequenceExample(
context=tf.train.Features(feature=context_features),
feature_lists=tf.train.FeatureLists(feature_list=sequence_features)
)
# 序列化 SequenceExample 协议缓冲区
serialized_sequence_example = sequence_proto.SerializeToString()
# 定义上下文特征描述
context_feature_description = {
'length': tf.io.FixedLenFeature([], tf.int64),
}
# 定义序列特征描述
sequence_feature_description = {
'int_feature': tf.io.FixedLenSequenceFeature([], tf.int64),
'float_feature': tf.io.FixedLenSequenceFeature([], tf.float32),
'string_feature': tf.io.FixedLenSequenceFeature([], tf.string),
}
# 解析
context_parsed, sequence_parsed = tf.io.parse_sequence_example(
serialized_sequence_example,
context_features=context_feature_description,
sequence_features=sequence_feature_description
)22. 编码与解码
a. tf.io.encode_base64 — 将张量编码为 Base64 字符串
import tensorflow as tf
# 创建张量
tensor = tf.constant(b'Hello, World!')
# 编码为 base64 字符串
encoded_tensor = tf.io.encode_base64(tensor)
# 打印编码后的 base64 字符串
print("Encoded Base64 String:")
print(encoded_tensor.numpy())b. tf.io.decode_base64 — 将 Base64 字符串解码为张量
import tensorflow as tf
# base64 字符串
base64_string = b'SGVsbG8sIFdvcmxkIQ=='
# 解码 base64 字符串
decoded_tensor = tf.io.decode_base64(base64_string)
# 打印解码后的张量
print("Decoded Tensor:")
print(decoded_tensor.numpy())c. tf.io.encode_jpeg — 将图像张量编码为 JPEG 格式
import tensorflow as tf
import numpy as np
# 创建一个随机图像张量
image = tf.random.uniform(shape=(100, 100, 3), minval=0, maxval=255, dtype=tf.int32)
# 编码为 JPEG 格式
encoded_image = tf.io.encode_jpeg(image, format='rgb', quality=90)
# 打印编码后的 JPEG 图像数据
print("Encoded JPEG Image Data:")
print(encoded_image.numpy())d. tf.io.decode_jpeg — 将 JPEG 图像解码为张量
import tensorflow as tf
# 读取并解码 JPEG 图像
image_path = 'example.jpg'
jpeg_string = tf.io.read_file(image_path)
# 将 JPEG 字符串解码为图像张量
decoded_image = tf.io.decode_jpeg(jpeg_string)
# 打印解码后的图像张量形状
print("Decoded Image Tensor Shape:")
print(decoded_image.shape)23. 数据格式转换
a. tf.io.serialize_sparse — 将稀疏张量序列化为字节字符串
import tensorflow as tf
# 创建一个稀疏张量
indices = tf.constant([[0, 0], [1, 2]], dtype=tf.int64)
values = tf.constant([1, 2], dtype=tf.float32)
dense_shape = tf.constant([3, 4], dtype=tf.int64)
sparse_tensor = tf.SparseTensor(indices, values, dense_shape)
# 将稀疏张量序列化为字节字符串
serialized_sparse = tf.io.serialize_sparse(sparse_tensor)
# 打印序列化后的字节字符串
print("Serialized Sparse Tensor:")
print(serialized_sparse.numpy())b. tf.io.parse_tensor — 将字节字符串反序列化为张量
import tensorflow as tf
# 序列化的字节字符串
serialized_tensor = b'\n\x0c\x08\x01\x08\x02\x12\x04\x00\x00\x01\x02\x12\x02\x03\x04'
# 将字节字符串反序列化为张量
deserialized_tensor = tf.io.parse_tensor(serialized_tensor, out_type=tf.int32)
# 打印反序列化后的张量
print("Deserialized Tensor:")
print(deserialized_tensor.numpy())模型构建（tf.keras）
1. 输入层
tf.keras.Input(shape=(3,))2. 模型封装
inputs = tf.keras.Input(shape=(None, None, 3))
processed = tf.keras.layers.RandomCrop(width=32, height=32)(inputs)
conv = tf.keras.layers.Conv2D(filters=2, kernel_size=3)(processed)
pooling = tf.keras.layers.GlobalAveragePooling2D()(conv)
feature = tf.keras.layers.Dense(10)(pooling)
full_model = tf.keras.Model(inputs, feature)
backbone = tf.keras.Model(processed, conv)
activations = tf.keras.Model(conv, feature)3. 子模型封装
实际上是封装了一批模型层的函数句柄。
tf.keras.Sequential(...)4. 模型层 tf.keras.layers
| 序号 | 层名 | 说明 |
|---|---|---|
| 1 | Add | 对两个张量逐元素相加 |
| 2 | AlphaDropout | 将 Alpha 衰减应用于输入 |
| 3 | Attention | Dot-product 注意力层（梁式注意力） |
| 4 | Average | 对两个张量逐元素求平均值 |
| 5 | AveragePooling1D / AvgPool1D | 时间数据的平均池化 |
| 6 | AveragePooling2D / AvgPool2D | 空间数据的平均池化 |
| 7 | AveragePooling3D / AvgPool3D | 3D 数据的平均池化 |
| 8 | BatchNormalization | 规范化其输入 |
| 9 | Bidirectional | RNN 的双向包装 |
| 10 | CategoryEncoding | 对整数特征进行编码的预处理层 |
| 11 | CenterCrop | 裁剪图像的预处理层 |
| 12 | Concatenate | 沿某个轴拼接张量 |
| 13 | Conv1D / Convolution1D | 1D 卷积层（时间卷积） |
| 14 | Conv1DTranspose / Convolution1DTranspose | 转置卷积层（反卷积） |
| 15 | Conv2D / Convolution2D | 2D 卷积层（图像空间卷积） |
| 16 | Conv2DTranspose / Convolution2DTranspose | 转置卷积层（反卷积） |
| 17 | Conv3D / Convolution3D | 3D 卷积层（体积空间卷积） |
| 18 | Conv3DTranspose / Convolution3DTranspose | 转置卷积层（反卷积） |
| 19 | ConvLSTM1D | 一维卷积 LSTM |
| 20 | ConvLSTM2D | 2D 卷积 LSTM |
| 21 | ConvLSTM3D | 3D 卷积 LSTM |
| 22 | Cropping1D | 1D 输入裁剪（时间序列） |
| 23 | Cropping2D | 2D 输入裁剪（图片） |
| 24 | Cropping3D | 3D 数据裁剪 |
| 25 | Dense | 常规密集连接 NN 层 |
| 26 | DenseFeatures | 基于 feature_columns 生成密集张量 |
| 27 | DepthwiseConv1D | 深度 1D 卷积 |
| 28 | DepthwiseConv2D | 深度 2D 卷积 |
| 29 | Discretization | 按范围对连续特征分桶 |
| 30 | Dot | 对两个张量逐元素计算点积 |
| 31 | Dropout | 将 Dropout 应用于输入 |
| 32 | ELU | 指数线性单元 |
| 33 | EinsumDense | 使用 tf.einsum 作为支持计算的层 |
| 34 | Embedding | 将正整数（索引）转换为固定大小密集向量 |
| 35 | Flatten | 展平输入，不影响 batch 大小 |
| 36 | GRU / GRUCell | 门控循环单元 |
| 37 | GaussianDropout | 乘性 1-中心高斯噪声 |
| 38 | GaussianNoise | 加性零中心高斯噪声 |
| 39 | GlobalAveragePooling1D / GlobalAvgPool1D | 时间数据全局平均池化 |
| 40 | GlobalAveragePooling2D / GlobalAvgPool2D | 空间数据全局平均池化 |
| 41 | GlobalAveragePooling3D / GlobalAvgPool3D | 3D 数据全局平均池化 |
| 42 | GlobalMaxPooling1D / GlobalMaxPool1D | 1D 时间数据全局最大池化 |
| 43 | GlobalMaxPooling2D / GlobalMaxPool2D | 空间数据全局最大池化 |
| 44 | GlobalMaxPooling3D / GlobalMaxPool3D | 3D 数据全局最大池化 |
| 45 | Hashing | 对分类特征进行哈希分桶 |
| 46 | InputLayer | 网络入口点 |
| 47 | IntegerLookup | 将整数特征映射到连续范围 |
| 48 | LSTM / LSTMCell | 长短期记忆层 |
| 49 | LayerNormalization | 层归一化 |
| 50 | LocallyConnected1D | 1D 本地连接层 |
| 51 | LocallyConnected2D | 2D 本地连接层 |
| 52 | Masking | 通过遮罩值跳过时间步来遮罩序列 |
| 53 | MaxPooling1D / MaxPool1D | 1D 最大池化 |
| 54 | MaxPooling2D / MaxPool2D | 2D 最大池化 |
| 55 | MaxPooling3D / MaxPool3D | 3D 最大池化 |
| 56 | Maximum | 对两个张量逐元素求最大值 |
| 57 | Minimum | 对两个张量逐元素求最小值 |
| 58 | MultiHeadAttention | 多头注意力层 |
| 59 | Multiply | 对两个张量逐元素相乘 |
| 60 | Normalization | 对连续特征进行标准化 |
| 61 | RNN | 循环层基类 |
| 62 | RandomBrightness | 训练期间随机调整亮度 |
| 63 | RandomContrast | 训练期间随机调整对比度 |
| 64 | RandomCrop | 训练期间随机裁剪图像 |
| 65 | RandomFlip | 训练期间随机翻转图像 |
| 66 | RandomHeight | 训练期间随机改变图像高度 |
| 67 | RandomRotation | 训练期间随机旋转图像 |
| 68 | RandomTranslation | 训练期间随机平移图像 |
| 69 | RandomWidth | 训练期间随机改变图像宽度 |
| 70 | RandomZoom | 训练期间随机缩放图像 |
| 71 | RepeatVector | 重复输入 n 次 |
| 72 | Rescaling | 将输入值重新缩放到新范围 |
| 73 | Reshape | 将输入重塑为给定形状 |
| 74 | Resizing | 调整图像大小 |
| 75 | SeparableConv1D / SeparableConvolution1D | 深度可分离 1D 卷积 |
| 76 | SeparableConv2D / SeparableConvolution2D | 深度可分离 2D 卷积 |
| 77 | SimpleRNN / SimpleRNNCell | 全连接 RNN |
| 78 | Softmax | Softmax 激活函数 |
| 79 | SpatialDropout1D | Dropout 空间 1D 版本 |
| 80 | SpatialDropout2D | Dropout 空间 2D 版本 |
| 81 | SpatialDropout3D | Dropout 空间 3D 版本 |
| 82 | StackedRNNCells | 包装器，允许 RNN 单元堆叠为单个单元 |
| 83 | StringLookup | 将字符串特征映射到整数索引 |
| 84 | Subtract | 对两个张量逐元素相减 |
| 85 | TextVectorization | 将文本特征映射到整数序列 |
| 86 | UnitNormalization | 单位归一化层 |
| 87 | UpSampling1D | 1D 上采样 |
| 88 | UpSampling2D | 2D 上采样 |
| 89 | UpSampling3D | 3D 上采样 |
| 90 | ZeroPadding1D | 1D 零填充（时间序列） |
| 91 | ZeroPadding2D | 2D 零填充（图片） |
| 92 | ZeroPadding3D | 3D 零填充（空间或时空） |
自定义神经网络（tf.nn）
该模块提供了许多用于构建神经网络的低级操作。
1. 激活函数
a. tf.nn.relu(features, name=None) — ReLU（Rectified Linear Unit）
import tensorflow as tf
features = tf.constant([-2.0, -1.0, 0.0, 1.0, 2.0])
output = tf.nn.relu(features)
print("ReLU output:", output.numpy())b. tf.nn.sigmoid(x, name=None) — Sigmoid 激活函数
import tensorflow as tf
x = tf.constant([-2.0, -1.0, 0.0, 1.0, 2.0])
output = tf.nn.sigmoid(x)
print("Sigmoid output:", output.numpy())c. tf.nn.tanh(x, name=None) — Tanh 激活函数
import tensorflow as tf
x = tf.constant([-2.0, -1.0, 0.0, 1.0, 2.0])
output = tf.nn.tanh(x)
print("Tanh output:", output.numpy())d. 借助 tf.keras.activations 自定义激活函数
import tensorflow as tf
# 使用 tf.keras.activations 模块定义激活函数
def custom_activation(x):
return tf.keras.activations.relu(x) - tf.keras.activations.sigmoid(x)
input_data = tf.constant([-2.0, -1.0, 0.0, 1.0, 2.0])
custom_activation_output = custom_activation(input_data)
print("Custom Activation output:", custom_activation_output.numpy())e. 使用 tf.keras.activations.get 获取激活函数
import tensorflow as tf
# 通过字符串获取激活函数
activation_function = tf.keras.activations.get("relu")
print(activation_function)
# 通过函数获取激活函数
def custom_activation(x):
return tf.square(x)
custom_activation_function = tf.keras.activations.get(custom_activation)
print(custom_activation_function)f. 常用激活函数说明（tf.nn）
| 函数 | 说明 |
|---|---|
linear(x) | 线性激活函数，即 f(x) = x |
relu(x, alpha=0.0, max_value=None, threshold=0) | ReLU 激活函数，f(x) = max(α·x, x) |
elu(x) | Exponential Linear Unit（ELU），负值部分呈指数衰减 |
selu(x) | Scaled Exponential Linear Unit（SELU），ELU 变体，具有一定归一化效果 |
softplus(x) | Softplus 激活函数，f(x) = log(1 + e^x) |
softsign(x) | Softsign 激活函数，f(x) = x / (1 + |x|) |
sigmoid(x) | Sigmoid 激活函数，f(x) = 1 / (1 + e^(-x)) |
tanh(x) | Hyperbolic Tangent（tanh）激活函数，f(x) = tanh(x) |
hard_sigmoid(x) | Hard Sigmoid 激活函数，f(x) = min(max(α·x + β, 0), 1) |
exponential(x) | Exponential 激活函数，f(x) = e^x |
2. 输入层（tf.keras.Input）
严格意义来讲，输入层不算做隐含层。它是 TensorFlow Keras 中的一个类，用于定义模型的输入层，是函数式 API（tf.keras.models.Model）中的一个重要组成部分，用于指定模型的输入形状和类型。
参数说明：
shape：输入张量的形状，不包括第一个维度（通常代表样本数量）。例如，如果输入是一个二维张量，那么shape可能是(100,)或(28, 28, 3)。dtype：输入张量的数据类型，默认为float32。name：输入张量的名字，默认为None。sparse：是否是稀疏张量，默认为False。ragged：是否是 ragged 张量，默认为False。tensor：如果要重用现有张量，可以指定现有的张量。batch_size：如果要固定批量大小，可以指定批量大小。默认为None，表示批量大小是可变的。
单输入模型示例：
import tensorflow as tf
# 定义输入
input_tensor = tf.keras.Input(shape=(100,), name='input')
# 定义模型的层
x = tf.keras.layers.Dense(64, activation='relu')(input_tensor)
x = tf.keras.layers.Dense(64, activation='relu')(x)
output_tensor = tf.keras.layers.Dense(10, activation='softmax')(x)
# 创建模型
model = tf.keras.models.Model(inputs=input_tensor, outputs=output_tensor)
# 显示模型结构
model.summary()多输入/多输出模型示例：
import tensorflow as tf
# 定义两个输入
text_input = tf.keras.Input(shape=(None,), dtype='int32', name='text_input')
numeric_input = tf.keras.Input(shape=(4,), dtype='float32', name='numeric_input')
# 定义模型的层
embedding_layer = tf.keras.layers.Embedding(1000, 64)(text_input)
x = tf.keras.layers.LSTM(64)(embedding_layer)
x = tf.keras.layers.concatenate([x, numeric_input])
output = tf.keras.layers.Dense(1, activation='sigmoid')(x)
# 创建模型
model = tf.keras.models.Model(inputs=[text_input, numeric_input], outputs=output)
# 显示模型结构
model.summary()3. 池化层
a. 一维最大池化 tf.nn.max_pool
import tensorflow as tf
input_data = tf.random.normal([32, 10, 16]) # (batch_size, sequence_length, input_channels)
ksize = [1, 2, 1] # 池化窗口大小
strides = [1, 2, 1] # 步幅
padding = 'VALID'
max_pool_output = tf.nn.max_pool(input_data, ksize, strides, padding)
print("Max Pooling output shape:", max_pool_output.shape)b. 二维最大池化 tf.nn.max_pool2d
import tensorflow as tf
input_data = tf.random.normal([32, 28, 28, 3]) # (batch_size, height, width, input_channels)
ksize = [1, 2, 2, 1] # 池化窗口大小
strides = [1, 2, 2, 1] # 步幅
padding = 'VALID'
max_pool2d_output = tf.nn.max_pool2d(input_data, ksize, strides, padding)
print("Max Pooling 2D output shape:", max_pool2d_output.shape)c. 自定义池化层（继承 tf.keras.layers.Layer）
import tensorflow as tf
class CustomMaxPooling1DLayer(tf.keras.layers.Layer):
def __init__(self, pool_size, strides=1, padding='VALID'):
super(CustomMaxPooling1DLayer, self).__init__()
self.pool_size = pool_size
self.strides = strides
self.padding = padding
def call(self, inputs):
output = tf.nn.max_pool1d(inputs, ksize=self.pool_size, strides=self.strides, padding=self.padding)
return output
# 使用自定义池化层
custom_max_pooling1d_layer = CustomMaxPooling1DLayer(pool_size=2, strides=2, padding='VALID')4. 卷积层
a. 一维卷积 tf.nn.conv1d
import tensorflow as tf
input_data = tf.random.normal([32, 10, 16]) # (batch_size, sequence_length, input_channels)
filters = tf.random.normal([3, 16, 32]) # (filter_size, input_channels, output_channels)
stride = 1
padding = 'VALID'
conv1d_output = tf.nn.conv1d(input_data, filters, stride, padding)
print("Conv1D output shape:", conv1d_output.shape)b. 二维卷积 tf.nn.conv2d
import tensorflow as tf
input_data = tf.random.normal([32, 28, 28, 3]) # (batch_size, height, width, input_channels)
filters = tf.random.normal([3, 3, 3, 64]) # (filter_height, filter_width, input_channels, output_channels)
strides = [1, 1, 1, 1]
padding = 'VALID'
conv2d_output = tf.nn.conv2d(input_data, filters, strides, padding)
print("Conv2D output shape:", conv2d_output.shape)c. 自定义卷积层（继承 tf.keras.layers.Layer）
import tensorflow as tf
class CustomConv1DLayer(tf.keras.layers.Layer):
def __init__(self, filters, kernel_size, stride=1, padding='VALID', activation='relu'):
super(CustomConv1DLayer, self).__init__()
self.filters = filters
self.kernel_size = kernel_size
self.stride = stride
self.padding = padding
self.activation = tf.keras.activations.get(activation)
def build(self, input_shape):
input_channels = input_shape[-1]
self.filters_shape = [self.kernel_size, input_channels, self.filters]
self.kernel = self.add_weight("kernel", shape=self.filters_shape)
def call(self, inputs):
output = tf.nn.conv1d(inputs, self.kernel, stride=self.stride, padding=self.padding)
output = self.activation(output)
return output
# 使用自定义卷积层
custom_conv1d_layer = CustomConv1DLayer(filters=64, kernel_size=3, stride=1, padding='VALID', activation='relu')5. 全连接层（tf.keras.layers.Dense）
v2 版本改为
tf.keras.layers实现。
a. 全连接层（Dense）
import tensorflow as tf
dense_layer = tf.keras.layers.Dense(units=128, activation='relu')参数说明： units、activation、use_bias、kernel_initializer='glorot_uniform'、bias_initializer='zeros'、kernel_regularizer、bias_regularizer
b. 自定义全连接层（继承 tf.keras.layers.Layer）
import tensorflow as tf
class CustomDenseLayer(tf.keras.layers.Layer):
def __init__(self, units=64, activation='relu'):
super(CustomDenseLayer, self).__init__()
self.units = units
self.activation = tf.keras.activations.get(activation)
def build(self, input_shape):
self.kernel = self.add_weight("kernel", shape=(input_shape[-1], self.units))
self.bias = self.add_weight("bias", shape=(self.units,))
def call(self, inputs):
output = tf.matmul(inputs, self.kernel) + self.bias
output = self.activation(output)
return output
# 使用自定义连接层
custom_dense_layer = CustomDenseLayer(units=128, activation='relu')6. 批归一化
a. tf.keras.layers.BatchNormalization — 批规范化层
import tensorflow as tf
# 定义输入数据的形状
input_shape = (None, 10) # 假设每个样本有 10 个特征
# 创建模型
model_batch_norm = tf.keras.Sequential([
tf.keras.layers.Dense(32, input_shape=input_shape),
tf.keras.layers.BatchNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(16),
tf.keras.layers.BatchNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_batch_norm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_batch_norm.summary()b. tf.keras.layers.GroupNormalization — 组规范化层
# 创建模型
model_group_norm = tf.keras.Sequential([
tf.keras.layers.Dense(32, input_shape=input_shape),
tf.keras.layers.GroupNormalization(groups=2),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(16),
tf.keras.layers.GroupNormalization(groups=2),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_group_norm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_group_norm.summary()c. tf.keras.layers.LayerNormalization — 层规范化层
# 创建模型
model_layer_norm = tf.keras.Sequential([
tf.keras.layers.Dense(32, input_shape=input_shape),
tf.keras.layers.LayerNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(16),
tf.keras.layers.LayerNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_layer_norm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_layer_norm.summary()d. tf.nn.batch_normalization — 底层批归一化操作
import tensorflow as tf
x = tf.random.normal([32, 64]) # (batch_size, features)
mean = tf.constant(0.0)
variance = tf.constant(1.0)
offset = tf.constant(0.0)
scale = tf.constant(1.0)
epsilon = 1e-5
normalized_output = tf.nn.batch_normalization(x, mean, variance, offset, scale, epsilon)
print("Batch Normalization output shape:", normalized_output.shape)e. tf.nn.global_avg_pool2d — 全局平均池化（用于全局平均池化批归一化）
import tensorflow as tf
input_data = tf.random.normal([32, 28, 28, 64]) # (batch_size, height, width, channels)
global_avg_pool_output = tf.nn.global_avg_pool2d(input_data, data_format='NHWC')
print("Global Average Pooling output shape:", global_avg_pool_output.shape)f. tf.function 自定义批归一化操作
import tensorflow as tf
# 自定义批归一化函数
def custom_batch_normalization(x, mean, variance, offset, scale, epsilon=1e-5):
normalized_output = (x - mean) * (scale / tf.sqrt(variance + epsilon)) + offset
return normalized_output
# 使用自定义批归一化函数
input_data = tf.random.normal([32, 64])
mean = tf.constant(0.0)
variance = tf.constant(1.0)
offset = tf.constant(0.0)
scale = tf.constant(1.0)
epsilon = 1e-5
custom_normalized_output = custom_batch_normalization(input_data, mean, variance, offset, scale, epsilon)
print("Custom Batch Normalization output shape:", custom_normalized_output.shape)g. 自定义规范化层（继承 tf.keras.layers.Layer）
import tensorflow as tf
import numpy as np
# 自定义规范化层
class CustomLayerNormalization(tf.keras.layers.Layer):
def __init__(self, axis=-1, epsilon=1e-6, **kwargs):
super(CustomLayerNormalization, self).__init__(**kwargs)
self.axis = axis
self.epsilon = epsilon
def build(self, input_shape):
self.gamma = self.add_weight(name='gamma', shape=input_shape[-1:], initializer='ones', trainable=True)
self.beta = self.add_weight(name='beta', shape=input_shape[-1:], initializer='zeros', trainable=True)
super(CustomLayerNormalization, self).build(input_shape)
def call(self, inputs):
mean = tf.reduce_mean(inputs, axis=self.axis, keepdims=True)
variance = tf.reduce_mean(tf.square(inputs - mean), axis=self.axis, keepdims=True)
std = tf.sqrt(variance + self.epsilon)
normalized = (inputs - mean) / std
output = self.gamma * normalized + self.beta
return output
def get_config(self):
config = {'axis': self.axis, 'epsilon': self.epsilon}
base_config = super(CustomLayerNormalization, self).get_config()
return dict(list(base_config.items()) + list(config.items()))
# 使用自定义规范化层创建模型
input_shape = (None, 10)
model_custom_norm = tf.keras.models.Sequential([
tf.keras.layers.Dense(32, input_shape=input_shape),
CustomLayerNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(16),
CustomLayerNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_custom_norm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_custom_norm.summary()
# 假设我们有一些数据
input_data = np.random.rand(100, 10) # 100 个样本，每个样本 10 个特征
labels = np.random.randint(0, 2, size=(100, 1)) # 100 个标签，每个标签为 0 或 1
# 训练模型
history = model_custom_norm.fit(input_data, labels, epochs=10, validation_split=0.2)7. 嵌入层（tf.keras.layers.Embedding）
a. tf.keras.layers.Embedding 函数
import tensorflow as tf
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
# 假设我们有一些文本数据
texts = [
"I love this movie",
"This is the best film ever",
"I hate it",
"It's a disaster",
"I can't wait to watch this again"
]
# 假设我们有一些标签数据
labels = [1, 1, 0, 0, 1] # 1 表示积极，0 表示消极
# 构建词汇表
tokenizer = Tokenizer(num_words=10000, oov_token="<OOV>") # OOV 表示在训练数据中未曾出现过的词汇
tokenizer.fit_on_texts(texts)
# 将文本转换为整数序列
sequences = tokenizer.texts_to_sequences(texts)
# 对序列进行填充，使所有样本具有相同的长度
padded_sequences = pad_sequences(sequences, padding="post")
# 创建嵌入层
embedding_dim = 16
vocab_size = len(tokenizer.word_index) + 1 # 加上 OOV token
embedding_layer = tf.keras.layers.Embedding(vocab_size, embedding_dim, input_length=padded_sequences.shape[1])b. 自定义嵌入层（继承 tf.keras.layers.Layer）
import tensorflow as tf
import numpy as np
# 定义一个自定义的嵌入层
class CustomEmbedding(tf.keras.layers.Layer):
def __init__(self, vocab_size, embedding_dim, dropout_rate=0.2, **kwargs):
super(CustomEmbedding, self).__init__(**kwargs)
self.vocab_size = vocab_size
self.embedding_dim = embedding_dim
self.dropout_rate = dropout_rate
def build(self, input_shape):
# 创建一个可训练的嵌入矩阵
self.embeddings = self.add_weight(
shape=(self.vocab_size, self.embedding_dim),
initializer='uniform',
trainable=True)
super(CustomEmbedding, self).build(input_shape)
def call(self, inputs):
# 使用嵌入矩阵进行查找
embedded = tf.nn.embedding_lookup(self.embeddings, inputs)
# 添加 dropout 层以减少过拟合
embedded_dropout = tf.nn.dropout(embedded, rate=self.dropout_rate)
return embedded_dropout
def get_config(self):
config = super(CustomEmbedding, self).get_config()
config.update({
'vocab_size': self.vocab_size,
'embedding_dim': self.embedding_dim,
'dropout_rate': self.dropout_rate,
})
return config
# 使用自定义嵌入层创建模型
vocab_size = 10000
embedding_dim = 16
dropout_rate = 0.2
# 假设我们有一些整数索引数据
input_data = np.random.randint(0, vocab_size, size=(100, 10)) # 100 个样本，每个样本 10 个词
# 创建自定义嵌入层实例
custom_embedding_layer = CustomEmbedding(vocab_size=vocab_size, embedding_dim=embedding_dim, dropout_rate=dropout_rate)
# 创建模型
model = tf.keras.Sequential([
custom_embedding_layer,
tf.keras.layers.GlobalAveragePooling1D(),
tf.keras.layers.Dense(24, activation='relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
# 显示模型结构
model.summary()
# 假设我们有一些标签数据
labels = np.random.randint(0, 2, size=(100,)) # 100 个标签，每个标签为 0 或 1
# 训练模型
history = model.fit(input_data, labels, epochs=10, validation_split=0.2)8. 循环层（RNN）
a. tf.keras.layers.SimpleRNN — 简单循环神经网络单元
import tensorflow as tf
# 定义输入数据的形状
input_shape = (None, 10) # 假设每个时间步有 10 个特征
# 创建模型
model_simple_rnn = tf.keras.Sequential([
tf.keras.layers.SimpleRNN(32, return_sequences=True, input_shape=input_shape),
# return_sequences=True：返回整个序列的输出，而非仅最后一个时间步
tf.keras.layers.SimpleRNN(16),
# 未设置 return_sequences，默认只返回最后一个时间步的输出
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_simple_rnn.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_simple_rnn.summary()b. tf.keras.layers.GRU — 门控循环单元（Gated Recurrent Unit）
# 创建模型
model_gru = tf.keras.Sequential([
tf.keras.layers.GRU(32, return_sequences=True, input_shape=input_shape),
# return_sequences=True：返回整个序列的输出
tf.keras.layers.GRU(16),
# 未设置 return_sequences，默认只返回最后一个时间步的输出
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_gru.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_gru.summary()c. tf.keras.layers.LSTM — 长短时记忆单元（Long Short-Term Memory）
# 创建模型
model_lstm = tf.keras.Sequential([
tf.keras.layers.LSTM(32, return_sequences=True, input_shape=input_shape),
# return_sequences=True：返回整个序列的输出
tf.keras.layers.LSTM(16),
# 未设置 return_sequences，默认只返回最后一个时间步的输出
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_lstm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_lstm.summary()d. 自定义循环层（继承 tf.keras.layers.AbstractRNNCell）
import tensorflow as tf
# 自定义循环单元
class SimpleAdditiveRNNCell(tf.keras.layers.AbstractRNNCell):
def __init__(self, units, **kwargs):
super(SimpleAdditiveRNNCell, self).__init__(**kwargs)
self.units = units
self.state_size = units
def build(self, input_shape):
# 创建权重矩阵
self.kernel = self.add_weight(shape=(input_shape[-1] + self.units, self.units),
initializer='uniform',
name='kernel')
super(SimpleAdditiveRNNCell, self).build(input_shape)
def call(self, inputs, states):
prev_output = states[0]
# 将输入和状态连接起来
concatenated = tf.concat([inputs, prev_output], axis=-1)
# 计算新的状态
new_state = tf.nn.tanh(tf.matmul(concatenated, self.kernel))
return new_state, [new_state]
def get_config(self):
config = {'units': self.units}
base_config = super(SimpleAdditiveRNNCell, self).get_config()
return dict(list(base_config.items()) + list(config.items()))
# 使用自定义循环单元创建模型
input_shape = (None, 10) # 假设每个时间步有 10 个特征
units = 32
# 创建模型
model_custom_rnn = tf.keras.models.Sequential([
tf.keras.layers.RNN(SimpleAdditiveRNNCell(units), return_sequences=True, input_shape=input_shape),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_custom_rnn.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_custom_rnn.summary()
# 假设我们有一些数据
input_data = np.random.rand(100, 10, 10) # 100 个样本，每个样本 10 个时间步，每个时间步 10 个特征
labels = np.random.randint(0, 2, size=(100, 10)) # 100 个标签，每个标签对应 10 个时间步的输出
# 训练模型
history = model_custom_rnn.fit(input_data, labels, epochs=10, validation_split=0.2)9. 变换器层（Transformer）
a. 多头注意力层 tf.keras.layers.MultiHeadAttention
import tensorflow as tf
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.preprocessing.sequence import pad_sequences
import numpy as np
# 假设我们有一些文本数据
texts = [
"I love this movie",
"This is the best film ever",
"I hate it",
"It's a disaster",
"I can't wait to watch this again"
]
# 假设我们有一些标签数据
labels = [1, 1, 0, 0, 1] # 1 表示积极，0 表示消极
# 构建词汇表
tokenizer = Tokenizer(num_words=10000, oov_token="<OOV>")
tokenizer.fit_on_texts(texts)
# 将文本转换为整数序列
sequences = tokenizer.texts_to_sequences(texts)
# 对序列进行填充，使所有样本具有相同的长度
padded_sequences = pad_sequences(sequences, padding="post")
# 创建模型
embedding_dim = 16
vocab_size = len(tokenizer.word_index) + 1 # 加上 OOV token
model_transformer = tf.keras.Sequential([
tf.keras.layers.Embedding(vocab_size, embedding_dim, input_length=padded_sequences.shape[1]),
tf.keras.layers.MultiHeadAttention(num_heads=2, key_dim=embedding_dim),
# num_heads：注意力头的数量，key_dim：键向量的维度
tf.keras.layers.GlobalAveragePooling1D(),
tf.keras.layers.Dense(24, activation='relu'),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_transformer.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
# 显示模型结构
model_transformer.summary()
# 准备标签数据
labels = np.array(labels)
# 训练模型
history = model_transformer.fit(padded_sequences, labels, epochs=10, validation_split=0.2)b. 自定义变换器层（继承 tf.keras.layers.Layer）
import tensorflow as tf
import numpy as np
# 自定义变换器层
class CustomTransformerLayer(tf.keras.layers.Layer):
def __init__(self, d_model, num_heads, dropout_rate=0.1, **kwargs):
super(CustomTransformerLayer, self).__init__(**kwargs)
self.d_model = d_model
self.num_heads = num_heads
self.dropout_rate = dropout_rate
def build(self, input_shape):
# 创建多头注意力层
self.multi_head_attention = tf.keras.layers.MultiHeadAttention(
num_heads=self.num_heads, key_dim=self.d_model
)
# 创建位置编码
position_encoding = self.positional_encoding(input_shape[1])
self.position_encoding = tf.keras.layers.Lambda(lambda x: x + position_encoding)
# 创建规范化层
self.layer_norm_1 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
self.layer_norm_2 = tf.keras.layers.LayerNormalization(epsilon=1e-6)
# 创建 Dropout 层
self.dropout_1 = tf.keras.layers.Dropout(self.dropout_rate)
self.dropout_2 = tf.keras.layers.Dropout(self.dropout_rate)
# 创建全连接层
self.dense_1 = tf.keras.layers.Dense(self.d_model, activation='relu')
self.dense_2 = tf.keras.layers.Dense(self.d_model)
super(CustomTransformerLayer, self).build(input_shape)
def call(self, inputs, training=False):
# 多头注意力
attention_output = self.multi_head_attention(inputs, inputs)
attention_output = self.dropout_1(attention_output, training=training)
attention_output = self.layer_norm_1(inputs + attention_output)
# FFN（前馈网络）
ffn_output = self.dense_2(self.dense_1(attention_output))
ffn_output = self.dropout_2(ffn_output, training=training)
ffn_output = self.layer_norm_2(attention_output + ffn_output)
return ffn_output
# 位置编码逻辑
def positional_encoding(self, sequence_length):
position = np.arange(sequence_length)[:, np.newaxis]
div_term = np.exp(np.arange(0, self.d_model, 2) * -(np.log(10000.0) / self.d_model))
angles = position * div_term
pos_encoding = np.concatenate([np.sin(angles), np.cos(angles)], axis=-1)
return tf.cast(pos_encoding[np.newaxis, ...], dtype=tf.float32)
def get_config(self):
config = {'d_model': self.d_model, 'num_heads': self.num_heads, 'dropout_rate': self.dropout_rate}
base_config = super(CustomTransformerLayer, self).get_config()
return dict(list(base_config.items()) + list(config.items()))
# 使用自定义变换器层创建模型
input_shape = (None, 10) # 假设每个样本有 10 个特征
d_model = 32
num_heads = 2
model_custom_transformer = tf.keras.models.Sequential([
tf.keras.layers.Dense(d_model, input_shape=input_shape),
CustomTransformerLayer(d_model, num_heads),
tf.keras.layers.GlobalAveragePooling1D(),
tf.keras.layers.Dense(1, activation='sigmoid')
])
# 编译模型
model_custom_transformer.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
model_custom_transformer.summary()
# 假设我们有一些数据
input_data = np.random.rand(100, 10, 10) # 100 个样本，每个样本 10 个时间步，每个时间步 10 个特征
labels = np.random.randint(0, 2, size=(100, 1)) # 100 个标签，每个标签为 0 或 1
# 训练模型
history = model_custom_transformer.fit(input_data, labels, epochs=10, validation_split=0.2)10. Dropout 层
a. tf.nn.dropout(x, rate, noise_shape=None, seed=None, name=None) — dropout 操作
import tensorflow as tf
x = tf.random.normal([32, 256]) # 输入张量的形状为 (batch_size, features)
dropout_rate = 0.5
dropout_output = tf.nn.dropout(x, rate=dropout_rate)
print("Dropout output shape:", dropout_output.shape)b. tf.nn.alpha_dropout(x, rate, seed=None, name=None) — Alpha Dropout 操作
Alpha Dropout 是一种保持均值和方差的 dropout 变体，常用于与 SELU 激活函数配合。
import tensorflow as tf
x = tf.random.normal([32, 256]) # 输入张量的形状为 (batch_size, features)
alpha_dropout_rate = 0.2
alpha_dropout_output = tf.nn.alpha_dropout(x, rate=alpha_dropout_rate)
print("Alpha Dropout output shape:", alpha_dropout_output.shape)c. tf.function 自定义 dropout 操作
import tensorflow as tf
# 自定义 dropout 函数
def custom_dropout(x, rate, training=True):
if training:
mask = tf.random.uniform(shape=tf.shape(x)) > rate
scaled_x = x / (1.0 - rate)
dropped_out_x = tf.where(mask, x, 0.0)
return dropped_out_x
else:
return x
# 使用自定义 dropout 函数
input_data = tf.random.normal([32, 256]) # 输入张量的形状为 (batch_size, features)
dropout_rate = 0.5
training_mode = True # 指定是否处于训练模式
custom_dropout_output = custom_dropout(input_data, rate=dropout_rate, training=training_mode)
print("Custom Dropout output shape:", custom_dropout_output.shape)11. Lambda 层（tf.keras.layers.Lambda）
Lambda 层可以将任意表达式封装为 Layer，适合简单变换操作。
import tensorflow as tf
import numpy as np
# 创建模型
model_lambda = tf.keras.models.Sequential([
tf.keras.layers.Input(shape=(1,)), # 输入层，假设每个样本只有一个值
tf.keras.layers.Lambda(lambda x: x * 2), # Lambda 层，计算输入的两倍
tf.keras.layers.Lambda(lambda x: x + 3), # Lambda 层，计算输入加三
tf.keras.layers.Lambda(lambda x: x * x) # Lambda 层，计算输入的平方
])
# 显示模型结构
model_lambda.summary()
# 准备数据
input_data = np.array([1, 2, 3, 4, 5]) # 输入数据
# 预测输出
output = model_lambda.predict(input_data)
# 输出结果
print("Predicted Output:", output)12. Softmax
a. tf.nn.softmax(logits, axis=None, name=None) — 计算 softmax
import tensorflow as tf
logits = tf.random.normal([32, 10]) # 输入张量的形状为 (batch_size, num_classes)
softmax_output = tf.nn.softmax(logits, axis=-1)
print("Softmax output shape:", softmax_output.shape)b. tf.nn.log_softmax(logits, axis=None, name=None) — 计算 log-softmax
import tensorflow as tf
logits = tf.random.normal([32, 10]) # 输入张量的形状为 (batch_size, num_classes)
log_softmax_output = tf.nn.log_softmax(logits, axis=-1)
print("LogSoftmax output shape:", log_softmax_output.shape)c. tf.function 自定义 softmax
import tensorflow as tf
# 自定义 softmax 函数
def custom_softmax(logits, axis=None):
exp_logits = tf.exp(logits - tf.reduce_max(logits, axis=axis, keepdims=True))
softmax_output = exp_logits / tf.reduce_sum(exp_logits, axis=axis, keepdims=True)
return softmax_output
# 使用自定义 softmax 函数
logits = tf.random.normal([32, 10]) # 输入张量的形状为 (batch_size, num_classes)
custom_softmax_output = custom_softmax(logits, axis=-1)
print("Custom Softmax output shape:", custom_softmax_output.shape)13. 损失函数
a. tf.nn.sparse_softmax_cross_entropy_with_logits — 多分类交叉熵
import tensorflow as tf
labels = tf.constant([1, 2, 0]) # 真实类别标签
logits = tf.random.normal([3, 5]) # 模型预测的 logits，形状为 (batch_size, num_classes)
loss = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=labels, logits=logits)
print("Cross Entropy Loss:", loss.numpy())b. tf.nn.mse_loss — 均方误差损失函数（回归问题）
import tensorflow as tf
labels = tf.constant([1.0, 2.0, 3.0]) # 真实标签
predictions = tf.constant([1.5, 2.5, 3.5]) # 模型预测的标签
mse_loss = tf.nn.mse_loss(labels, predictions)
print("MSE Loss:", mse_loss.numpy())c. tf.nn.sigmoid_cross_entropy_with_logits — 二分类 Sigmoid 交叉熵
import tensorflow as tf
labels = tf.constant([0.0, 1.0, 1.0]) # 真实标签
logits = tf.random.normal([3]) # 模型预测的 logits，形状为 (batch_size,)
sigmoid_cross_entropy_loss = tf.nn.sigmoid_cross_entropy_with_logits(labels=labels, logits=logits)
print("Sigmoid Cross Entropy Loss:", sigmoid_cross_entropy_loss.numpy())d. tf.function 自定义损失函数
import tensorflow as tf
# 自定义损失函数
def custom_loss(labels, predictions):
# 在这里定义你的损失计算逻辑
loss = tf.reduce_mean(tf.square(labels - predictions))
return loss
# 使用自定义损失函数
labels = tf.constant([1.0, 2.0, 3.0])
predictions = tf.constant([1.5, 2.5, 3.5])
loss = custom_loss(labels, predictions)
print("Custom Loss:", loss.numpy())模型梯度（tf.GradientTape）
1. tf.GradientTape() — 自动微分（推荐方式）
tf.GradientTape() 创建上下文管理器，用于在 TensorFlow 中进行自动微分。在该上下文管理器中的操作会被记录，以便稍后计算某些变量的梯度。这对于神经网络和梯度下降等优化算法非常有用，尤其在误差反向传播的神经网络中，tf.GradientTape 是核心组件之一。
tape.watch(variable)：显式指定 tape 跟踪并记录某个变量的操作。若不使用watch，会自动监视可训练变量（如tf.Variable类型）。tape.gradient()：使用监视的可训练变量信息，计算相对于某个变量的梯度。
import tensorflow as tf
x = tf.constant(3.0)
# 使用 tf.GradientTape 计算梯度
with tf.GradientTape() as tape:
tape.watch(x) # 也可以省略，tf.GradientTape 会自动监视可训练变量
y = x**2
# 计算相对于 x 的梯度
gradient = tape.gradient(y, x)
# 显示梯度值
if gradient is not None:
print("Gradient of y with respect to x:", gradient.numpy())
else:
print("Gradient is None.")2. tf.gradients — 解析法计算梯度（v1 版本）
注意： 1.x 版本需要关闭 eager execution 即时执行模式。2.x 版本默认启用即时执行模式，需使用
tf.GradientTape方式。
import tensorflow as tf
# 禁用 eager execution
tf.compat.v1.disable_eager_execution()
# 定义和运行 TensorFlow 操作
x = tf.constant(3.0)
y = x**2
# 使用 tf.gradients 计算梯度
gradient = tf.gradients(y, x)
# 创建会话并运行计算图
with tf.compat.v1.Session() as sess:
# 初始化变量（如果有的话）
sess.run(tf.compat.v1.global_variables_initializer())
# 获取梯度值
gradient_value = sess.run(gradient)
# 显示梯度值
print("Gradient of y with respect to x:", gradient_value[0])3. tf.test.compute_gradient — 数值方法估计梯度
import tensorflow as tf
# 定义函数 f，其中 x 是输入张量
def f(x):
return x**2
x = tf.constant(3.0)
# 使用数值梯度计算
numeric_gradient = tf.test.compute_gradient(f, [x])
# 转换为列表并提取数值梯度
numeric_gradient_values = list(numeric_gradient)[0][0]
print("Numeric gradient of y with respect to x:", numeric_gradient_values)模型创建
1. Sequential 模型
Sequential 适合构建线性堆叠的网络，即每一层的输出都是下一层的输入。这种方式非常适合初学者，对于大多数标准的深度学习任务已经足够。
import tensorflow as tf
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(100,)),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(10, activation='softmax')
])2. 函数式 API（Functional API）
函数式 API 提供了一种更灵活的方式定义模型，可用于创建具有多个输入或输出、共享层、以及任意连接模式的复杂模型。
import tensorflow as tf
input_tensor = tf.keras.Input(shape=(100,))
x = tf.keras.layers.Dense(64, activation='relu')(input_tensor)
x = tf.keras.layers.Dense(64, activation='relu')(x)
output_tensor = tf.keras.layers.Dense(10, activation='softmax')(x)
model = tf.keras.models.Model(inputs=input_tensor, outputs=output_tensor)3. 子类化 API（Subclassing API）
通过继承 tf.keras.Model 类来定义模型，提供最大灵活性，可以在模型类的方法中直接编写自定义逻辑。
import tensorflow as tf
class MyModel(tf.keras.Model):
def __init__(self):
super(MyModel, self).__init__()
self.dense1 = tf.keras.layers.Dense(64, activation='relu')
self.dense2 = tf.keras.layers.Dense(64, activation='relu')
self.dense3 = tf.keras.layers.Dense(10, activation='softmax')
def call(self, inputs):
x = self.dense1(inputs)
x = self.dense2(x)
return self.dense3(x)
model = MyModel()4. 示例
a. 卷积神经网络（CNN）
model = tf.keras.models.Sequential([
tf.keras.layers.Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=(28, 28, 1)),
tf.keras.layers.MaxPooling2D(pool_size=(2, 2)),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(128, activation='relu'),
tf.keras.layers.Dense(10, activation='softmax')
])b. 循环神经网络（RNN）
model = tf.keras.models.Sequential([
tf.keras.layers.SimpleRNN(32, input_shape=(10, 50)),
tf.keras.layers.Dense(10, activation='softmax')
])c. 使用函数式 API 创建多输入模型
text_input = tf.keras.Input(shape=(None,), dtype='int32')
numeric_input = tf.keras.Input(shape=(4,), dtype='float32')
embedding_layer = tf.keras.layers.Embedding(1000, 64)(text_input)
x = tf.keras.layers.LSTM(64)(embedding_layer)
x = tf.keras.layers.concatenate([x, numeric_input])
output = tf.keras.layers.Dense(1, activation='sigmoid')(x)
model = tf.keras.models.Model(inputs=[text_input, numeric_input], outputs=output)模型改进
1. 批量归一化（Batch Normalization）
批量归一化通过对每一层的激活值进行归一化来稳定训练过程，减少内部协变量偏移问题。
改进的问题：
- 内部协变量偏移：随着训练的进行，前面层的分布变化会影响后面层的学习
- 加速训练：通过归一化输入，可以更快地收敛
import tensorflow as tf
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, input_shape=(100,)),
tf.keras.layers.BatchNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(64),
tf.keras.layers.BatchNormalization(),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(10, activation='softmax')
])2. 跳跃连接（Skip Connections）
跳跃连接将网络中较早层的输出直接传递给后续层，有助于在网络的不同部分之间传递信息，这对于恢复细节非常重要。
改进的问题：
- 梯度消失：深层网络中梯度反向传播时容易消失
- 使网络更容易训练
U-Net 跳跃连接示例：
import tensorflow as tf
def unet_model(input_shape):
inputs = tf.keras.Input(shape=input_shape)
# 编码器
conv1 = tf.keras.layers.Conv2D(64, 3, activation='relu', padding='same')(inputs)
conv1 = tf.keras.layers.Conv2D(64, 3, activation='relu', padding='same')(conv1)
pool1 = tf.keras.layers.MaxPooling2D(pool_size=(2, 2))(conv1)
# 跳跃连接
conv2 = tf.keras.layers.Conv2D(128, 3, activation='relu', padding='same')(pool1)
conv2 = tf.keras.layers.Conv2D(128, 3, activation='relu', padding='same')(conv2)
up1 = tf.keras.layers.UpSampling2D(size=(2, 2))(conv2)
concat1 = tf.keras.layers.concatenate([up1, conv1], axis=-1)
# 解码器
conv3 = tf.keras.layers.Conv2D(64, 3, activation='relu', padding='same')(concat1)
conv3 = tf.keras.layers.Conv2D(64, 3, activation='relu', padding='same')(conv3)
output = tf.keras.layers.Conv2D(1, 1, activation='sigmoid')(conv3)
# 创建模型
model = tf.keras.models.Model(inputs=[inputs], outputs=[output])
return model
# 创建 U-Net 模型
model = unet_model((256, 256, 1))
model.summary()3. 残差连接（Residual Connections）
残差连接通过添加跳跃连接来缓解深层网络中的梯度消失问题，并使网络更容易训练。
改进的问题：
- 梯度消失：深层网络中梯度反向传播时容易消失
- 更容易训练深层网络
ResNet 残差块示例：
import tensorflow as tf
def resnet_block(input_data, filters, strides=1):
x = tf.keras.layers.Conv2D(filters, kernel_size=(3, 3), strides=strides, padding='same')(input_data)
x = tf.keras.layers.BatchNormalization()(x)
x = tf.keras.layers.Activation('relu')(x)
x = tf.keras.layers.Conv2D(filters, kernel_size=(3, 3), padding='same')(x)
x = tf.keras.layers.BatchNormalization()(x)
# 如果需要调整输入的维度，以便能够与残差块的输出相加
if strides > 1:
shortcut = tf.keras.layers.Conv2D(filters, kernel_size=(1, 1), strides=strides, padding='same')(input_data)
shortcut = tf.keras.layers.BatchNormalization()(shortcut)
else:
shortcut = input_data
x = tf.keras.layers.Add()([x, shortcut])
x = tf.keras.layers.Activation('relu')(x)
return x
def resnet_model(input_shape):
inputs = tf.keras.Input(shape=input_shape)
# 初始卷积层
x = tf.keras.layers.Conv2D(64, kernel_size=(7, 7), strides=(2, 2), padding='same')(inputs)
x = tf.keras.layers.BatchNormalization()(x)
x = tf.keras.layers.Activation('relu')(x)
x = tf.keras.layers.MaxPooling2D(pool_size=(3, 3), strides=(2, 2), padding='same')(x)
# 残差块
x = resnet_block(x, 64)
x = resnet_block(x, 64)
x = resnet_block(x, 64)
# 平均池化和全连接层
x = tf.keras.layers.GlobalAveragePooling2D()(x)
output = tf.keras.layers.Dense(10, activation='softmax')(x)
# 创建模型
model = tf.keras.models.Model(inputs=[inputs], outputs=[output])
return model
# 创建 ResNet 模型
model = resnet_model((224, 224, 3))
model.summary()4. 权重初始化（Weight Initialization）
权重初始化方法（如 Xavier 初始化或 He 初始化）可以改善网络的收敛速度，避免梯度消失或爆炸。
改进的问题：
- 梯度消失或爆炸：随机初始化可能导致梯度消失或爆炸
- 更快的收敛速度
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, input_shape=(100,), kernel_initializer='he_normal'),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(64, kernel_initializer='he_normal'),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(10, activation='softmax')
])5. 激活函数（Activation Functions）
选择适当的激活函数（如 ReLU、Leaky ReLU、Swish 等）可以改善网络的训练效果。
改进的问题：
- 梯度消失：某些激活函数（如 sigmoid 或 tanh）会导致梯度消失
- 更好的非线性表达能力
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, input_shape=(100,)),
tf.keras.layers.Activation('relu'), # 或 tf.keras.layers.LeakyReLU(alpha=0.3)
tf.keras.layers.Dense(64),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dense(10, activation='softmax')
])6. 正则化（Regularization）
正则化技术（如 L1、L2 或 Dropout）可以防止过拟合，提高模型的泛化能力。
改进的问题：
- 过拟合：模型在训练集上表现很好，但在测试集上表现不佳
- 泛化能力增强
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, input_shape=(100,), kernel_regularizer=tf.keras.regularizers.l2(0.01)),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dropout(0.5),
tf.keras.layers.Dense(64, kernel_regularizer=tf.keras.regularizers.l2(0.01)),
tf.keras.layers.Activation('relu'),
tf.keras.layers.Dropout(0.5),
tf.keras.layers.Dense(10, activation='softmax')
])7. 数据增强（Data Augmentation）
数据增强通过生成训练数据的多种变体来增加数据多样性，从而提高模型的鲁棒性和泛化能力。
改进的问题：
- 数据不足：少量数据可能导致模型过拟合
- 增强泛化能力
data_augmentation = tf.keras.Sequential([
tf.keras.layers.experimental.preprocessing.RandomFlip("horizontal"),
tf.keras.layers.experimental.preprocessing.RandomRotation(0.1),
tf.keras.layers.experimental.preprocessing.RandomZoom(0.1),
])
model = tf.keras.models.Sequential([
data_augmentation,
tf.keras.layers.Conv2D(32, 3, padding='same', activation='relu', input_shape=(32, 32, 3)),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Conv2D(64, 3, padding='same', activation='relu'),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Conv2D(128, 3, padding='same', activation='relu'),
tf.keras.layers.MaxPooling2D(),
tf.keras.layers.Flatten(),
tf.keras.layers.Dense(128, activation='relu'),
tf.keras.layers.Dense(10, activation='softmax')
])模型编译
model.compile(loss, optimizer, metrics) 用于配置训练时的损失函数、优化器和评估指标。
1. 损失函数（loss）
包路径：tf.keras.losses / tf.losses
| 类名 | 函数名 | 字符串（默认） | 作用 |
|---|---|---|---|
BinaryCrossentropy | binary_crossentropy(...) | binary_crossentropy | 计算真实标签和预测标签之间的交叉熵损失 |
BinaryFocalCrossentropy | binary_focal_crossentropy(...) | binary_focal_crossentropy | 计算真实标签和预测之间的焦点交叉熵损失 |
CategoricalCrossentropy | categorical_crossentropy(...) | categorical_crossentropy | 计算标签和预测之间的交叉熵损失 |
CategoricalHinge | categorical_hinge(...) | categorical_hinge | 计算分类铰链损失 |
CosineSimilarity | cosine_similarity(...) | cosine_similarity | 计算标签和预测之间的余弦相似性 |
Hinge | hinge(...) | hinge | 计算铰链损失 |
Huber | huber(...) | huber | 计算 Huber 损失 |
KLDivergence | KLD(...) / kld(...) / kl_divergence(...) | kld | 计算 Kullback-Leibler 散度损失 |
LogCosh | log_cosh(...) / logcosh(...) | logcosh | 计算预测误差的双曲余弦的对数 |
MeanAbsoluteError | MAE(...) / mae(...) | mae | 计算绝对差的平均值 |
MeanAbsolutePercentageError | MAPE(...) / mape(...) | mape | 计算平均绝对百分比误差 |
MeanSquaredError | MSE(...) / mse(...) | mse | 计算误差平方平均值 |
MeanSquaredLogarithmicError | MSLE(...) / msle(...) | msle | 计算均方对数误差 |
Poisson | poisson(...) | poisson | 计算泊松损失 |
SparseCategoricalCrossentropy | sparse_categorical_crossentropy(...) | sparse_categorical_crossentropy | 计算标签和预测之间的交叉熵损失 |
SquaredHinge | squared_hinge(...) | squared_hinge | 计算平方铰链损失 |
2. 优化器（optimizer）
包路径：tf.keras.optimizers / tf.optimizers。不支持函数句柄，需要实例化。
| 类名 | 字符串 | 作用 |
|---|---|---|
Adadelta | adadelta | 实现 Adadelta 算法的优化器 |
Adagrad | adagrad | 实现 Adagrad 算法的优化器 |
Adam | adam | 实现 Adam 算法的优化器 |
Adamax | adamax | 实现 Adamax 算法的优化器 |
Ftrl | ftrl | 实现 FTRL 算法的优化器 |
Nadam | nadam | 实现 NAdam 算法的优化器 |
RMSprop | rmsprop | 实现 RMSprop 算法的优化器 |
SGD | sgd | 梯度下降（带动量）优化器 |
3. 评估指标（metrics）
包路径：tf.keras.metrics
| 类名 | 函数名 | 作用 |
|---|---|---|
AUC | 近似 ROC 或 PR 曲线的 AUC（曲线下面积） | |
Accuracy | 计算预测等于标签的频率 | |
BinaryAccuracy | binary_accuracy(...) | 计算预测与二进制标签匹配的频率 |
BinaryCrossentropy | binary_crossentropy(...) | 计算标签和预测之间的交叉熵度量 |
BinaryIoU | 计算类 0 和/或 1 的并集交集度量 | |
CategoricalAccuracy | categorical_accuracy(...) | 计算预测与 one-hot 标签匹配的频率 |
CategoricalCrossentropy | categorical_crossentropy(...) | 计算标签和预测之间的交叉熵度量 |
CategoricalHinge | 计算分类铰链度量 | |
CosineSimilarity | 计算标签和预测之间的余弦相似性 | |
FalseNegatives | 计算假阴性数 | |
FalsePositives | 计算假阳性数 | |
Hinge | hinge(...) | 计算铰链度量 |
IoU | 计算特定目标类的并集交集度量 | |
KLDivergence | KLD(...) / kld(...) / kl_divergence(...) | 计算 Kullback-Leibler 散度度量 |
LogCoshError | log_cosh(...) / logcosh(...) | 计算预测误差的双曲余弦的对数 |
Mean | 计算给定值的（加权）平均值 | |
MeanAbsoluteError | MAE(...) / mae(...) | 计算平均绝对误差 |
MeanAbsolutePercentageError | MAPE(...) / mape(...) | 计算平均绝对百分比误差 |
MeanIoU | 计算并集上交集度量的平均值 | |
MeanRelativeError | 通过使用给定值归一化来计算平均相对误差 | |
MeanSquaredError | MSE(...) / mse(...) | 计算均方误差 |
MeanSquaredLogarithmicError | MSLE(...) / msle(...) | 计算均方对数误差 |
MeanTensor | 计算给定张量的元素（加权）平均值 | |
OneHotIoU | 计算独热编码标签的并集交集度量 | |
OneHotMeanIoU | 计算独热编码标签的平均并集交集度量 | |
Poisson | poisson(...) | 计算泊松度量 |
Precision | 计算相对于标签的预测精度 | |
PrecisionAtRecall | 当召回率 >= 指定值时，计算最佳精度 | |
Recall | 计算预测相对于标签的召回率 | |
RecallAtPrecision | 计算精度 >= 指定值时的最佳召回率 | |
RootMeanSquaredError | 计算均方根误差度量 | |
SensitivityAtSpecificity | 当特异性 >= 指定值时，计算最佳灵敏度 | |
SparseCategoricalAccuracy | sparse_categorical_accuracy(...) | 计算预测与整数标签匹配的频率 |
SparseCategoricalCrossentropy | sparse_categorical_crossentropy(...) | 计算标签和预测之间的交叉熵度量 |
SparseTopKCategoricalAccuracy | sparse_top_k_categorical_accuracy(...) | 计算整数目标在前 K 个预测中的频率 |
SpecificityAtSensitivity | 当灵敏度 >= 指定值时，计算最佳特异性 | |
SquaredHinge | squared_hinge(...) | 计算平方铰链度量 |
Sum | 计算给定值的（加权）和 | |
TopKCategoricalAccuracy | top_k_categorical_accuracy(...) | 计算目标在前 K 个预测中的频率 |
TrueNegatives | 计算真负例数 | |
TruePositives | 计算真正例数 |
模型训练
model.fit 用于在固定数量的时间段（数据集上的迭代）训练模型。
model.fit(
x=None, # 数据特征
y=None, # 数据标签
batch_size=None, # 每个批次的样本数量
epochs=1, # 模型在整个数据集上迭代的次数，如果 epochs=2，那么模型将在训练集上迭代两次
verbose='auto', # 日志显示模式。0 表示不显示进度条；1 表示显示进度条；2 表示显示每个 epoch 的结束信息
callbacks=None, # 在训练过程中调用的一系列函数，可以用来执行各种操作，如保存模型权重、早期停止等
validation_split=0.0, # 从训练数据中划分出的一部分数据作为验证数据的比例。例如，如果 validation_split=0.2，那么将会使用 20% 的训练数据作为验证集
validation_data=None, # 显式提供用于验证的数据。可以替代 validation_split 参数，输入形式为元组 (x_val, y_val) 或 (x_val, y_val, val_sample_weights)
shuffle=True, # 是否在每个 epoch 开始时打乱输入数据。如果使用 tf.data.Dataset 作为输入，则此参数会被忽略，因为 Dataset 自身可以被设置为打乱顺序
class_weight=None, # 不同类别的样本权重。这对于处理类别不平衡的数据集很有用
sample_weight=None, # 为每个样本分配的权重。这对于处理某些样本更重要的情况很有用
initial_epoch=0, # 开始训练的 epoch 编号。这对于继续训练一个已经部分训练过的模型很有用
steps_per_epoch=None, # 每轮训练使用的批次数量。如果输入数据是 tf.data.Dataset 并且没有指定此参数，则会遍历整个数据集
validation_steps=None, # 用于验证的批次数量。如果验证数据是 tf.data.Dataset 并且没有指定此参数，则会遍历整个数据集
validation_batch_size=None, # 用于验证的每个批次的样本数量
validation_freq=1, # 指定每多少个 epoch 执行一次验证评估
max_queue_size=10, # 用于生成批量的队列的最大长度
workers=1, # 用于数据生成的进程数
use_multiprocessing=False # 是否使用多进程来生成数据。如果设置为 True，则使用 workers 参数
)model.fit、model.evaluate、model.predict 接受的数据类型：numpy、list、dataset 均可，但必须是列的形式。
1. 处理 numpy 数组
import numpy as np
import tensorflow as tf
# 创建一个 1000 行 20 列的 numpy 数组
x_train = np.random.random((1000, 20))
# 创建一个 1000 行 1 列的 0-1 数组
y_train = np.random.randint(2, size=(1000, 1))
model = tf.keras.models.Sequential([
# 注意第一个隐含层，需要 20 个节点
tf.keras.layers.Dense(64, activation='relu', input_shape=(20,)),
tf.keras.layers.Dense(1, activation='sigmoid')
])
model.compile(optimizer='rmsprop',
loss='binary_crossentropy',
metrics=['accuracy'])
# 将 x 和 y 输入模型，进行训练
model.fit(x_train, y_train, epochs=10, batch_size=32)2. TensorFlow 数据集（tf.data.Dataset）
# 创建一个 dataset 数据集，包含特征和标签
dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(32)
# 此时只需要传递 dataset 即可，不需要额外提供标签 y_train
model.fit(dataset, epochs=10)3. 使用生成器
class CustomGenerator(tf.keras.utils.Sequence):
def __init__(self, x_set, y_set, batch_size):
self.x, self.y = x_set, y_set
self.batch_size = batch_size
def __len__(self):
return len(self.x) // self.batch_size
def __getitem__(self, idx):
batch_x = self.x[idx * self.batch_size:(idx + 1) * self.batch_size]
batch_y = self.y[idx * self.batch_size:(idx + 1) * self.batch_size]
return batch_x, batch_y
gen = CustomGenerator(x_train, y_train, 32)
model.fit(gen, epochs=10)4. 使用 ImageDataGenerator 加载图像数据
from tensorflow.keras.preprocessing.image import ImageDataGenerator
train_datagen = ImageDataGenerator(rescale=1./255)
train_generator = train_datagen.flow_from_directory(
directory='/path/to/train_data',
target_size=(150, 150),
batch_size=32,
class_mode='binary')
model.fit(train_generator, epochs=10)模型验证
model.evaluate(
x=None,
y=None,
batch_size=None, # 每个批次的样本数量
verbose='auto', # 显示进度条的方式（0、1、2）
sample_weight=None, # 每个样本应用不同的权重
steps=None, # 如果 x 是 tf.data.Dataset 并且没有指定 steps，则会遍历整个数据集；如果指定了 steps，则只遍历前 steps 个批量
callbacks=None,
max_queue_size=10,
workers=1,
use_multiprocessing=False,
return_dict=False,
**kwargs
)模型预测
model.predict(
x,
batch_size=None, # 每个批次的样本数量
verbose='auto', # 显示进度条的方式（0、1、2）
steps=None, # 如果 x 是 tf.data.Dataset 并且没有指定 steps，则会遍历整个数据集；如果指定了 steps，则只遍历前 steps 个批量
callbacks=None,
max_queue_size=10,
workers=1,
use_multiprocessing=False
)模型保存与加载
a. tf.saved_model.save — 保存任意类型的 TensorFlow 模型
这种方式非常适合保存自定义模型或者使用 tf.keras API 构建的模型。
tf.saved_model.save(model, # 要保存的模型对象
export_dir='saved_model_path') # 模型保存的目标目录，必需参数b. model.save — 保存 Keras 模型
tf.keras API 提供的模型保存方法，专门用于保存使用 tf.keras 构建的模型。这种方法保存的模型包含了模型架构、权重、优化器状态以及编译时配置的信息。
model.save(filepath='keras_model.h5', # 模型保存的目标文件路径
overwrite=True, # 是否覆盖已存在的文件，默认为 True
include_optimizer=True, # 是否保存优化器的状态，默认为 True。如果设置为 False，则不会保存优化器的状态
save_format=None # 保存的格式，可以是 'h5' 或 'tf'。默认为 'h5'（HDF5 格式），选择 'tf' 会保存为 SavedModel 格式
)注意：
- 如果计划在相同的环境中继续训练模型，建议保存优化器的状态。如果不打算继续训练，可以不保存优化器的状态，以减小模型文件的大小。
- HDF5 格式通常较小且易于在不同平台上移动，而 SavedModel 格式更适合部署和生产环境，因为它支持更多的操作和灵活性。
c. tf.keras.models.load_model — 加载 HDF5 格式的模型
model_h5 = tf.keras.models.load_model('keras_model.h5')如果指定 compile=False，则需要重新编译模型。
注意： 使用
tf.keras.models.save保存的模型应该使用tf.keras.models.load_model加载。
d. tf.saved_model.load — 加载 SavedModel 格式的模型
model_saved_model = tf.saved_model.load('saved_model_path')注意： 使用
tf.saved_model.save保存的模型应该使用tf.saved_model.load加载。
回调函数（tf.keras.callbacks）
使用 list 包裹，传递给 model.fit、model.evaluate、model.predict 的 callbacks 参数。
a. tf.keras.callbacks.ModelCheckpoint — 保存模型权重
在每个 epoch 结束时保存模型权重。可以设置保存条件（例如只保存最佳模型），避免训练过程中保存不必要的模型文件，节省存储空间。
checkpoint_callback = ModelCheckpoint(
filepath='best_model.h5',
monitor='val_loss',
save_best_only=True
)b. tf.keras.callbacks.EarlyStopping — 提前终止训练
当监测的指标停止改进时，提前终止训练。通过设置 patience 参数来控制在没有改进后等待的 epoch 数量，可以避免过拟合并缩短训练时间。
early_stopping_callback = EarlyStopping(
monitor='val_loss',
patience=5,
restore_best_weights=True
)c. tf.keras.callbacks.ReduceLROnPlateau — 动态降低学习率
当监测的指标停止改进时，减少学习率。通过动态调整学习率来帮助模型跳出局部最优解，促进更好的收敛。
reduce_lr_callback = ReduceLROnPlateau(
monitor='val_loss',
factor=0.1,
patience=3,
min_lr=0.0001
)d. tf.keras.callbacks.LearningRateScheduler — 自定义学习率调度
根据预定的时间表改变学习率，可以实现自定义的学习率调度策略，以适应不同的训练阶段。
# 定义学习率调度器函数
def lr_scheduler(epoch, lr):
# 每 10 个 epoch 减少一次学习率
if epoch % 10 == 0 and epoch > 0:
return lr * 0.1
else:
return lr
# 创建 LearningRateScheduler 回调
lr_scheduler_callback = tf.keras.callbacks.LearningRateScheduler(lr_scheduler, verbose=1)e. tf.keras.callbacks.TensorBoard — 可视化训练过程
将日志文件写入 TensorBoard，以便可视化训练过程中的指标。虽然不直接控制训练过程，但可以帮助监视训练过程中的指标，以便更好地理解模型的表现并据此调整训练策略。
# 定义 TensorBoard 回调
log_dir = "logs/fit/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
tensorboard_callback = tf.keras.callbacks.TensorBoard(log_dir=log_dir, histogram_freq=1)f. tf.keras.callbacks.CSVLogger — 保存训练日志为 CSV
将日志信息保存为 CSV 文件。通过记录详细的训练历史，可以分析训练过程中的趋势，有助于调整训练策略。
# 定义 CSVLogger 回调
csv_logger = tf.keras.callbacks.CSVLogger('training.log', separator=',', append=False)g. tf.keras.callbacks.ProgbarLogger — 显示进度条
用于在训练或预测时显示进度条。
# 创建 ProgbarLogger 实例
progbar_logger = tf.keras.callbacks.ProgbarLogger(count_mode='steps')h. 使用样例
将多个回调函数通过 callbacks 参数传递给 model.fit：
history = model.fit(
x_train, y_train,
epochs=100,
batch_size=32,
validation_data=(x_val, y_val),
callbacks=[checkpoint_callback, early_stopping_callback, reduce_lr_callback]
)i. 注意事项
- 回调函数的顺序：回调函数的执行顺序可能会影响它们的行为。例如，如果同时使用
EarlyStopping和ModelCheckpoint，请确保EarlyStopping放在ModelCheckpoint之后，以防止过早停止训练导致的最佳模型丢失。 - 回调函数的组合：可以根据需要组合多种回调函数来实现复杂的功能。
j. 自定义回调函数 — 预测数据时显示进度条
import tensorflow as tf
import numpy as np
from tqdm import tqdm
class PredictionProgressBarCallback(tf.keras.callbacks.Callback):
def __init__(self, verbose=1):
super(PredictionProgressBarCallback, self).__init__()
self.verbose = verbose
self.pbar = None
def on_predict_begin(self, logs=None):
self.pbar = tqdm(total=self.params['steps'], disable=not self.verbose)
def on_predict_batch_end(self, batch, logs=None):
if self.pbar:
self.pbar.update(1)
def on_predict_end(self, logs=None):
if self.pbar:
self.pbar.close()
# 创建一个简单的模型并训练
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(32, activation='relu', input_shape=(10,)),
tf.keras.layers.Dense(1, activation='sigmoid')
])
x_train = np.random.rand(1000, 10)
y_train = np.random.rand(1000)
model.compile(optimizer='adam', loss='binary_crossentropy')
model.fit(x_train, y_train, epochs=10)
# 假设您已经有了一些数据
x_predict = np.random.rand(1000, 10) # 1000 个样本，每个样本 10 个特征
# 创建自定义回调函数实例
callback = PredictionProgressBarCallback()
# 使用 model.predict() 进行预测，并显示进度条
predictions = model.predict(x_predict, callbacks=[callback])分布式训练（tf.distribute.Strategy）
训练步骤
- 创建策略：通过
tf.distribute创建一个可以在所有 GPU 上运行的策略对象。 - 代码同步：
- 对于多机多卡的情况，需要在每台机器上运行相同代码。
- 对于单机多卡的情况，只需要在一个环境中运行代码即可。
- 对于单机单卡或单机 CPU 的情况，同样只需要在一个环境中运行代码即可。
- 数据集准备：数据集被加载并转换成适当格式。
- 模型构建：使用
strategy.scope()确保模型在所有的设备上正确地构建，编译模型并指定优化器、损失函数和度量。 - 训练模型：调用
fit方法开始训练过程。
1. 多机多卡（Multi-worker Multi-GPU）
每个机器上有一个或多个 GPU，使用 tf.distribute.experimental.MultiWorkerMirroredStrategy。
import os
import tensorflow as tf
# 设置环境变量来定义集群
os.environ['TF_CONFIG'] = '''
{
"cluster": {
"worker": ["localhost:12345", "localhost:12346"]
},
"task": {"type": "worker", "index": 0}
}
'''
# 创建 MultiWorkerMirroredStrategy 实例
strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy()
# 构建模型
with strategy.scope():
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(784,)),
tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy'])
# 加载数据集
(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(60000, 784).astype('float32') / 255
y_train = y_train.astype('int64')
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(64)
# 训练模型
model.fit(train_dataset, epochs=3)2. 单机多卡（Single-worker Multi-GPU）
一台机器上有多个 GPU，使用 tf.distribute.MirroredStrategy。
import tensorflow as tf
# 创建 MirroredStrategy 实例
strategy = tf.distribute.MirroredStrategy()
# 查看有多少个设备参与计算
print('Number of devices: %d' % strategy.num_replicas_in_sync)
# 构建模型
with strategy.scope():
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(784,)),
tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy'])
# 加载数据集
(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(60000, 784).astype('float32') / 255
y_train = y_train.astype('int64')
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(64)
# 训练模型
model.fit(train_dataset, epochs=3)3. 单机单卡（Single-worker Single-GPU）
一台机器上只有一个 GPU，使用 tf.distribute.OneDeviceStrategy。
import tensorflow as tf
# 创建 OneDeviceStrategy 实例
strategy = tf.distribute.OneDeviceStrategy('/gpu:0')
# 构建模型
with strategy.scope():
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(784,)),
tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy'])
# 加载数据集
(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(60000, 784).astype('float32') / 255
y_train = y_train.astype('int64')
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(64)
# 训练模型
model.fit(train_dataset, epochs=3)4. 单机单 CPU
没有 GPU 可用，只能使用 CPU，使用 tf.distribute.OneDeviceStrategy。
import tensorflow as tf
# 创建 OneDeviceStrategy 实例
strategy = tf.distribute.OneDeviceStrategy('/cpu:0')
# 构建模型
with strategy.scope():
model = tf.keras.models.Sequential([
tf.keras.layers.Dense(64, activation='relu', input_shape=(784,)),
tf.keras.layers.Dense(10, activation='softmax')
])
model.compile(optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy'])
# 加载数据集
(x_train, y_train), _ = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(60000, 784).astype('float32') / 255
y_train = y_train.astype('int64')
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train)).batch(64)
# 训练模型
model.fit(train_dataset, epochs=3)5. 单机多 CPU
使用 tf.distribute.MirroredStrategy，通过 devices 参数指定多个 CPU。
import tensorflow as tf
from tensorflow import keras
# 创建策略
strategy = tf.distribute.MirroredStrategy(devices=["/cpu:0", "/cpu:1", "/cpu:2", "/cpu:3"])
# 在策略范围内定义模型
with strategy.scope():
# 定义模型
model = keras.Sequential([
keras.layers.Dense(64, activation='relu'),
keras.layers.Dense(10)
])
# 编译模型
model.compile(
optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy']
)
# 准备数据
dataset = tf.data.Dataset.from_tensor_slices(
(tf.random.uniform([1000, 10]), tf.random.uniform([1000], maxval=10, dtype=tf.int32))
)
dataset = dataset.batch(32).repeat()
# 训练模型
model.fit(dataset, epochs=10)6. 多机多 CPU
使用 tf.distribute.MultiWorkerMirroredStrategy。
import tensorflow as tf
from tensorflow import keras
import os
# 设置环境变量
os.environ['TF_CONFIG'] = """
{
'cluster': {
'worker': ['localhost:12345', 'localhost:12346']
},
'task': {'type': 'worker', 'index': 0}
}
"""
# 创建多机多 CPU 策略
strategy = tf.distribute.MultiWorkerMirroredStrategy()
with strategy.scope():
# 定义模型
model = keras.Sequential([
keras.layers.Dense(64, activation='relu'),
keras.layers.Dense(10)
])
# 编译模型
model.compile(
optimizer=tf.keras.optimizers.Adam(),
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy']
)
# 准备数据
dataset = tf.data.Dataset.from_tensor_slices(
(tf.random.uniform([1000, 10]), tf.random.uniform([1000], maxval=10, dtype=tf.int32))
)
dataset = dataset.batch(32)
# 训练模型
model.fit(dataset, epochs=10)启动脚本：
# 在第一台机器上启动
TF_CONFIG='{"cluster": {"worker": ["localhost:12345", "localhost:12346"]}, "task": {"type": "worker", "index": 0}}' python train.py
# 在第二台机器上启动
TF_CONFIG='{"cluster": {"worker": ["localhost:12345", "localhost:12346"]}, "task": {"type": "worker", "index": 1}}' python train.py查看设备信息
1. tf.config 查看设备信息
import tensorflow as tf
# 获取所有物理 CPU 和 GPU
physical_cpus = tf.config.list_physical_devices('CPU')
physical_gpus = tf.config.list_physical_devices('GPU')
# 输出结果和 tf.config.experimental.list_physical_devices 一致
# 获取 CPU 和 GPU 的数量
num_physical_cpus = len(physical_cpus)
num_physical_gpus = len(physical_gpus)
# 输出信息
print("Physical CPUs:", num_physical_cpus)
print("Physical GPUs:", num_physical_gpus)2. tf.test 查看 GPU 是否可用
import tensorflow as tf
# 检查是否有 GPU 可用
has_gpu = tf.test.is_gpu_available()
# 输出信息
if has_gpu:
print("GPU is available.")
else:
print("GPU is not available.")Python 函数转换为 TensorFlow 图（tf.function）
1. @tf.function 装饰器 — 基本用法
将普通的 Python 函数转换为由 TensorFlow 图支持的函数，允许 TensorFlow 自动跟踪张量的操作并执行静态图优化。
import tensorflow as tf
# 定义一个普通的 Python 函数
@tf.function
def my_function(a, b):
return tf.multiply(a, b) + tf.constant(3.0)
# 使用转换后的函数
a = tf.constant(2.0)
b = tf.constant(4.0)
result = my_function(a, b)
print(result.numpy()) # 输出结果2. @tf.function(jit_compile=True) — 启用 XLA JIT 编译
在支持的情况下启用 XLA JIT 编译，以提高性能。
@tf.function(jit_compile=True)
def my_function(a, b):
return tf.multiply(a, b) + tf.constant(3.0)3. @tf.function(input_signature=[...]) — 指定输入签名
允许指定输入签名，从而可以让函数提前知道输入张量的形状和类型，有助于提高效率。
import tensorflow as tf
# 定义一个普通的 Python 函数，并指定输入签名
@tf.function(input_signature=[
tf.TensorSpec(shape=(None,), dtype=tf.float32),
tf.TensorSpec(shape=(None,), dtype=tf.float32)
])
def my_function(a, b):
return tf.multiply(a, b) + tf.constant(3.0)
# 使用转换后的函数
a = tf.constant([2.0, 3.0])
b = tf.constant([4.0, 5.0])
result = my_function(a, b)
print(result.numpy()) # 输出结果报错排查
ValueError: Layer sequential_4 expects 1 input(s), but it received 121 input tensors
这个错误通常意味着模型期望接收一个输入，但实际上收到了多个输入。
可能的错误情形： 数据集的输入数据是字典形式，但模型期望接收一个单一的张量作为输入。
假设有以下数据格式：
| feature1 | feature2 | feature3 | label |
|---|---|---|---|
| 1 | 10 | 100 | 0 |
| 2 | 20 | 200 | 1 |
| 3 | 30 | 300 | 0 |
| 4 | 40 | 400 | 1 |
| 5 | 50 | 500 | 0 |
dataset = tf.data.Dataset.from_tensor_slices((
{
'feature1': [1, 2, 3, 4, 5],
'feature2': [10, 20, 30, 40, 50],
'feature3': [100, 200, 300, 400, 500]
},
[0, 1, 0, 1, 0]
))按照这种方式生成的数据集，每个元素都是一个元组，包含特征部分和标签部分。以第一个元素为例，特征部分的字典和标签部分的值解析后，每个特征转换成一个张量：
处理前的数据格式：
+----------------+ +----+
| feature1: 1 | | 0 |
| feature2: 10 | +----+
| feature3: 100 |
+----------------+即：({'feature1': 1, 'feature2': 10, 'feature3': 100}, 0)
解决方法： 将数据集中的字典转换为单一张量。
def convert_to_tensor(sample_dict, label):
return list(sample_dict.values()), label
dataset = dataset.map(convert_to_tensor)处理后的数据格式：
+------------+ +----+
| [1, 10, 100]| | 0 |
+------------+ +----+即：([1, 10, 100], 0)
深度学习推荐算法
深度学习推荐系统是利用深度学习技术来进行个性化推荐的一种方法。这些模型旨在捕捉用户的行为模式和偏好，以便预测用户可能感兴趣的内容。
1. 双塔模型（Two-Tower Model）
a. 完整版示例代码
import numpy as np
import pandas as pd
from tensorflow.keras.layers import Embedding, Dense, LayerNormalization, Dropout
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
import tensorflow as tf
tf.config.run_functions_eagerly(True)
# 定义特征列
def create_feature_columns():
user_age = tf.feature_column.numeric_column("user_age")
user_city = tf.feature_column.categorical_column_with_vocabulary_list(
'user_city', ['suzhou', 'nanjing', 'nantong']
)
user_city_emb = tf.feature_column.embedding_column(user_city, dimension=8)
item_category = tf.feature_column.categorical_column_with_vocabulary_list(
'item_category', ['books', 'electronics', 'movies']
)
item_category_emb = tf.feature_column.embedding_column(item_category, dimension=8)
user_columns = [user_age, user_city_emb]
item_columns = [item_category_emb]
return user_columns, item_columns
# 构建用户塔
def build_user_tower(inputs, feature_columns):
inputs_layer = tf.keras.layers.DenseFeatures(feature_columns)(inputs)
x = Dense(64, activation='relu')(inputs_layer)
x = LayerNormalization()(x)
x = Dropout(0.2)(x)
output = Dense(32, activation=None)(x)
return output
# 构建项目塔
def build_item_tower(inputs, feature_columns):
inputs_layer = tf.keras.layers.DenseFeatures(feature_columns)(inputs)
x = Dense(64, activation='relu')(inputs_layer)
x = LayerNormalization()(x)
x = Dropout(0.2)(x)
output = Dense(32, activation=None)(x)
return output
# 计算相似度
def compute_similarity(user_output, item_output):
dot_product = tf.reduce_sum(tf.multiply(user_output, item_output), axis=1, keepdims=True)
return dot_product
# 构建模型
class TwoTowerModel(Model):
def __init__(self, user_columns, item_columns):
super(TwoTowerModel, self).__init__()
self.user_tower = build_user_tower
self.item_tower = build_item_tower
self.user_columns = user_columns
self.item_columns = item_columns
@tf.function
def call(self, inputs):
user_inputs, item_inputs = inputs
user_output = self.user_tower(user_inputs, self.user_columns)
item_output = self.item_tower(item_inputs, self.item_columns)
similarity = compute_similarity(user_output, item_output)
return similarity
# 初始化特征列
user_columns, item_columns = create_feature_columns()
# 创建模型实例
model = TwoTowerModel(user_columns, item_columns)
# 编译模型
model.compile(
optimizer=Adam(learning_rate=0.001),
loss='binary_crossentropy',
metrics=['accuracy']
)
# 训练数据
train_data = {
'user_age': np.array([1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4]),
'user_city': np.array([
'suzhou', 'nanjing', 'nantong', 'nantong',
'suzhou', 'nanjing', 'nantong', 'nantong',
'suzhou', 'nanjing', 'nantong', 'nantong'
]),
'item_category': np.array([
'books', 'electronics', 'movies', 'movies',
'books', 'electronics', 'movies', 'movies',
'books', 'electronics', 'movies', 'movies'
]),
'label': np.array([1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0])
}
# 将数据拆分为用户特征、项目特征和标签
user_train_data = {
'user_age': train_data['user_age'],
'user_city': train_data['user_city']
}
item_train_data = {
'item_category': train_data['item_category']
}
labels = train_data['label']
# 模型训练部分：由于有两个模型，x 形式采用嵌套的 dict 形式
history = model.fit(
x=[user_train_data, item_train_data],
y=labels,
epochs=100,
batch_size=4
)b. 简化版代码
import numpy as np
from tensorflow.keras.layers import Embedding, Dense, LayerNormalization, Dropout
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
import tensorflow as tf
tf.config.run_functions_eagerly(True)
# 定义特征列
# 注意定义特征列时的列名需要和训练数据集中的键名相同
user_age = tf.feature_column.numeric_column("user_age")
user_city = tf.feature_column.categorical_column_with_vocabulary_list(
'user_city', ['suzhou', 'nanjing', 'nantong']
)
user_city_emb = tf.feature_column.embedding_column(user_city, dimension=8)
item_category = tf.feature_column.categorical_column_with_vocabulary_list(
'item_category', ['books', 'electronics', 'movies']
)
item_category_emb = tf.feature_column.embedding_column(item_category, dimension=8)
user_columns = [user_age, user_city_emb]
item_columns = [item_category_emb]
# 训练数据
# 注意这里的键名必须与定义特征列时的列名相同
train_data = {
'user_age': np.array([1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4]),
'user_city': np.array([
'suzhou', 'nanjing', 'nantong', 'nantong',
'suzhou', 'nanjing', 'nantong', 'nantong',
'suzhou', 'nanjing', 'nantong', 'nantong'
]),
'item_category': np.array([
'books', 'electronics', 'movies', 'movies',
'books', 'electronics', 'movies', 'movies',
'books', 'electronics', 'movies', 'movies'
]),
'label': np.array([1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0])
}
# 将数据拆分为用户特征、项目特征和标签
user_train_data = {
'user_age': train_data['user_age'],
'user_city': train_data['user_city'],
}
item_train_data = {
'item_category': train_data['item_category']
}
# 构建模型
class TwoTowerModel(Model):
def __init__(self):
super(TwoTowerModel, self).__init__()
# 注意这里的 call 方法仅仅返回了张量，TwoTowerModel 最终会将其处理为具有 compile 方法的模型
@tf.function
def call(self, _):
# 构建用户塔
inputs_layer = tf.keras.layers.DenseFeatures(user_columns)(user_train_data)
x = Dense(64, activation='relu')(inputs_layer)
x = LayerNormalization()(x)
x = Dropout(0.2)(x)
user_output = Dense(32, activation=None)(x)
# 构建项目塔
inputs_layer = tf.keras.layers.DenseFeatures(item_columns)(item_train_data)
x = Dense(64, activation='relu')(inputs_layer)
x = LayerNormalization()(x)
x = Dropout(0.2)(x)
item_output = Dense(32, activation=None)(x)
# 计算相似度
dot_product = tf.reduce_sum(tf.multiply(user_output, item_output), axis=1, keepdims=True)
return dot_product
# 创建模型实例
model = TwoTowerModel()
# 编译模型
model.compile(
optimizer=Adam(learning_rate=0.001),
loss='binary_crossentropy',
metrics=['accuracy']
)
labels = train_data['label']
# 模型训练部分：由于有两个模型，x 形式采用嵌套的 dict 形式
history = model.fit(
x=[user_train_data, item_train_data],
y=labels,
epochs=100,
batch_size=1
)2. Wide & Deep Learning Model
结合线性模型（Wide 部分）与非线性深层神经网络（Deep 部分），能够同时从数据中学习低阶和高阶的特征交互。
a. 模型特性
- 灵活性：
- Wide 部分：通常包含线性模型（如逻辑回归），可以处理大量的稀疏特征，捕捉特征之间的第一阶关系。
- Deep 部分：使用深度神经网络来建模特征之间的高阶交互。
- 强大的表达能力：深度神经网络可以通过多层非线性变换捕捉复杂的数据分布，而线性模型则可以处理高维稀疏数据。
- 易于扩展：Wide 部分可以很容易地扩展到包含大量的特征，尤其是对于稀疏特征。Deep 部分可以通过增加层数和节点数来扩展模型的容量。
- 可解释性与泛化能力：Wide 部分保留了一定程度的可解释性，因为它类似于传统的线性模型。Deep 部分能够更好地泛化到未见过的数据，因为它能够捕捉到数据中的复杂模式。
b. 设计思路
- 捕捉低阶和高阶特征组合：在许多现实世界的应用中，数据往往包含了大量的低阶特征（如用户性别、年龄等）和高阶特征（如用户的兴趣爱好）。Wide 部分能够有效地处理低阶特征，而 Deep 部分则擅长捕捉高阶特征之间的复杂关系。
- 平衡性能与可解释性：对于某些应用场景而言，模型的可解释性是非常重要的。Wide 部分提供了一种透明的方式来看待模型是如何做出决策的，而 Deep 部分则可以提供更高的预测精度。
- 处理稀疏数据：推荐系统中的数据通常是高度稀疏的，Wide 部分能够很好地处理这种稀疏性，而 Deep 部分则能够通过嵌入（Embedding）等技术来减少稀疏特征的维度，从而更好地捕捉潜在的模式。
- 提升模型性能：实践证明，Wide & Deep 模型在很多任务上都比单独使用 Wide 或 Deep 模型表现得更好，因为它们结合了两者的优点，能够在不同类型的特征上表现出色。
c. 示例代码
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Concatenate, Dense, Flatten
# 设定随机种子以确保结果的可重复性
np.random.seed(42)
tf.random.set_seed(42)
# 生成示例数据
num_users = 1000
num_items = 500
num_regions = 10
max_age = 100
# 生成用户 ID、项目 ID、年龄、地区、评分等数据
user_ids = np.random.randint(0, num_users, size=(num_users * num_items))
item_ids = np.random.randint(0, num_items, size=(num_users * num_items))
ages = np.random.randint(18, max_age, size=(num_users * num_items))
regions = np.random.randint(0, num_regions, size=(num_users * num_items))
ratings = np.random.randint(1, 6, size=(num_users * num_items))
# 创建 DataFrame
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'age': ages,
'region': regions,
'rating': ratings
})
# 显示数据样本
print(data.head())
# 定义输入层
user_id_input = Input(shape=(1,), name='user_id')
item_id_input = Input(shape=(1,), name='item_id')
age_input = Input(shape=(1,), name='age')
region_input = Input(shape=(1,), name='region')
# 定义嵌入层
user_embedding = Embedding(input_dim=num_users + 1, output_dim=8, name='user_embedding')(user_id_input)
item_embedding = Embedding(input_dim=num_items + 1, output_dim=8, name='item_embedding')(item_id_input)
region_embedding = Embedding(input_dim=num_regions + 1, output_dim=4, name='region_embedding')(region_input)
# 展平嵌入向量
user_flatten = Flatten()(user_embedding)
item_flatten = Flatten()(item_embedding)
region_flatten = Flatten()(region_embedding)
# 深度部分
deep_inputs = Concatenate()([user_flatten, item_flatten, region_flatten, age_input])
x = Dense(64, activation='relu')(deep_inputs)
x = Dense(32, activation='relu')(x)
deep_output = Dense(1, activation='linear')(x)
# 宽度部分
wide_inputs = Concatenate()([age_input, region_flatten])
wide_output = Dense(1, activation='linear')(wide_inputs)
# 结合 Wide 和 Deep 部分
combined_output = tf.keras.layers.Add()([wide_output, deep_output])
# 创建模型
model = Model(inputs=[user_id_input, item_id_input, age_input, region_input], outputs=combined_output)
# 编译模型
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# 查看模型结构
model.summary()
# 准备训练数据
X_train = {
'user_id': data['user_id'].values,
'item_id': data['item_id'].values,
'age': data['age'].values.reshape(-1, 1),
'region': data['region'].values
}
y_train = data['rating'].values
# 训练模型
history = model.fit(X_train, y_train, epochs=5, batch_size=32, validation_split=0.2)3. DeepFM（Deep Factorization Machine）
同时具备因子分解机（FM）的第一阶和第二阶特征组合的能力，并结合深层神经网络来建模高阶特征组合。
a. 模型特点
- Factorization Machine（FM）：FM 是一种二阶特征交互模型，能够捕捉特征之间的两两交互。FM 能够处理稀疏特征，并且对于每个特征，它通过嵌入向量来表示。
- Deep Neural Network（DNN）：DNN 部分可以捕捉高阶特征交互，通过多层非线性变换来学习特征之间的复杂关系。DNN 能够处理非线性关系，并且通过堆叠多层来增强模型的表达能力。
- 结合两者的优势：DeepFM 模型结合了 FM 和 DNN 的优势，既能够捕捉低阶特征交互（通过 FM），又能够捕捉高阶特征交互（通过 DNN）。输出层将 FM 和 DNN 的部分融合在一起，共同进行预测。
b. 设计思路
DeepFM 模型的设计思路是为了在推荐系统等场景下更有效地捕捉特征之间的交互作用，特别是在处理大规模稀疏数据时。
-
因子分解机（FM）：
- 线性部分：能够捕捉特征本身的贡献。
- 二阶交互部分：能够捕捉特征之间的两两交互作用，对推荐系统任务非常重要。
- 参数高效性：即使在特征空间非常大的情况下，参数数量也是可控的。
-
深度神经网络（DNN）：
- 强大的非线性拟合能力：能够学习到复杂的模式和特征间的高阶交互。
- 自动特征学习：不需要手动设计特征，网络自己会学习到有助于任务完成的特征表示。
- 灵活性：可以通过增加或减少网络层数来调整模型的复杂度。
-
DeepFM 的结合：
- 并行学习：FM 部分负责学习特征之间的二阶交互，DNN 部分则学习高阶交互。
- 互补优势：FM 能够有效捕捉到特征之间的显式交互，而 DNN 则可以捕捉到隐式的、复杂的交互。
- 共享输入：两者都基于相同的输入特征，但通过不同的机制来提取特征的重要性。
- 融合输出：FM 和 DNN 的输出被融合在一起，共同决定最终的预测结果。
c. 示例代码
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Concatenate, Dense, Flatten, Lambda
from tensorflow.keras.regularizers import l2
# 设置随机种子以确保结果的可重复性
np.random.seed(42)
tf.random.set_seed(42)
# 生成示例数据
num_users = 1000
num_items = 500
num_features = 10
max_age = 100
# 生成用户 ID、项目 ID、年龄、其他特征及标签等数据
user_ids = np.random.randint(0, num_users, size=(num_users * num_items))
item_ids = np.random.randint(0, num_items, size=(num_users * num_items))
ages = np.random.randint(18, max_age, size=(num_users * num_items))
features = np.random.randint(0, 2, size=(num_users * num_items, num_features))
labels = np.random.randint(0, 2, size=(num_users * num_items))
# 创建 DataFrame
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'age': ages,
'features': features.tolist(),
'label': labels
})
# 显示数据样本
print(data.head())
# 分离特征
user_id = data['user_id'].values
item_id = data['item_id'].values
age = data['age'].values.reshape(-1, 1)
features = data['features'].apply(np.array).values
# 将 features 转换为 NumPy 数组
features = np.vstack(features)
# 定义输入层
user_id_input = Input(shape=(1,), name='user_id')
item_id_input = Input(shape=(1,), name='item_id')
age_input = Input(shape=(1,), name='age')
features_input = Input(shape=(num_features,), name='features')
# 定义嵌入层
embedding_dim = 8
user_embedding = Embedding(
input_dim=num_users + 1, output_dim=embedding_dim, embeddings_regularizer=l2(1e-6)
)(user_id_input)
item_embedding = Embedding(
input_dim=num_items + 1, output_dim=embedding_dim, embeddings_regularizer=l2(1e-6)
)(item_id_input)
# 展平嵌入向量
user_flatten = Flatten()(user_embedding)
item_flatten = Flatten()(item_embedding)
# 拼接所有特征
all_inputs = Concatenate()([user_flatten, item_flatten, age_input, features_input])
# FM 部分
def fm_layer(inputs):
square_of_sum = tf.square(tf.reduce_sum(inputs, axis=1))
sum_of_square = tf.reduce_sum(tf.square(inputs), axis=1)
fm_output = 0.5 * tf.reduce_sum(square_of_sum - sum_of_square, axis=1, keepdims=True)
return fm_output
# 应用 FM 层
fm_inputs = Concatenate()([user_embedding, item_embedding])
fm_output = Lambda(fm_layer)(fm_inputs)
# DNN 部分
dnn_output = Dense(64, activation='relu')(all_inputs)
dnn_output = Dense(32, activation='relu')(dnn_output)
dnn_output = Dense(1, activation='linear')(dnn_output)
# 结合 FM 和 DNN 部分
combined_output = tf.keras.layers.Add()([fm_output, dnn_output])
# 创建模型
model = Model(
inputs=[user_id_input, item_id_input, age_input, features_input],
outputs=combined_output
)
# 编译模型
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 查看模型结构
model.summary()
# 准备训练数据
X_train = {
'user_id': user_id,
'item_id': item_id,
'age': age,
'features': features
}
# 训练模型
history = model.fit(X_train, labels, epochs=5, batch_size=32, validation_split=0.2)4. Neural Collaborative Filtering（NCF）
使用多层感知机（MLP）来模拟用户-项目之间的相互作用，从而实现协同过滤的效果。
a. 模型特性
- 深度学习模型：NCF 采用深度神经网络来学习用户和物品的潜在表示。
- 协同过滤：通过学习用户-物品之间的历史交互数据来预测用户对未见过的物品的兴趣。
- 非线性映射：利用多层神经网络来捕捉用户和物品之间复杂的非线性关系。
- 灵活的架构：可以很容易地扩展模型的深度和宽度，以适应不同的任务需求。
b. 设计思路
- 用户和物品嵌入：首先将用户和物品的 ID 通过嵌入层（Embedding Layer）映射到一个低维的稠密向量空间。
- 多层神经网络：通过多层全连接网络（Fully Connected Layers）来学习用户和物品的潜在表示，并捕捉它们之间的交互关系。
- 输出层：最终通过一个输出层来预测用户对物品的评分或点击概率。
c. 示例代码
import numpy as np
import pandas as pd
# 设置随机种子以确保结果的可重复性
np.random.seed(42)
# 生成示例数据
num_users = 1000
num_items = 500
num_interactions = num_users * num_items
# 生成用户 ID、物品 ID 和评分
user_ids = np.random.randint(0, num_users, size=num_interactions)
item_ids = np.random.randint(0, num_items, size=num_interactions)
ratings = np.random.randint(1, 6, size=num_interactions)
# 创建 DataFrame
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'rating': ratings
})
# 显示数据样本
print(data.head())
# 分离特征
users = data['user_id'].values
items = data['item_id'].values
ratings = data['rating'].valuesimport tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Concatenate, Dense, Flatten
# 设置随机种子以确保结果的可重复性
tf.random.set_seed(42)
# 定义输入层
user_input = Input(shape=(1,), name='user_input')
item_input = Input(shape=(1,), name='item_input')
# 定义嵌入层
embedding_dim = 8 # 嵌入向量的维度
user_embedding = Embedding(input_dim=num_users + 1, output_dim=embedding_dim, name='user_embedding')(user_input)
item_embedding = Embedding(input_dim=num_items + 1, output_dim=embedding_dim, name='item_embedding')(item_input)
# 展平嵌入向量
user_flatten = Flatten()(user_embedding)
item_flatten = Flatten()(item_embedding)
# 拼接用户和物品的嵌入向量
concatenated = Concatenate()([user_flatten, item_flatten])
# 添加全连接层
fc1 = Dense(64, activation='relu')(concatenated)
fc2 = Dense(32, activation='relu')(fc1)
output = Dense(1, activation='linear')(fc2)
# 创建模型
model = Model(inputs=[user_input, item_input], outputs=output)
# 编译模型
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# 查看模型结构
model.summary()
# 准备训练数据
X_train = {
'user_input': users,
'item_input': items
}
# 训练模型
history = model.fit(X_train, ratings, epochs=5, batch_size=32, validation_split=0.2)5. Matrix Factorization with MLP（MF + MLP）
类似于 NCF，但更专注于使用神经网络来替代传统的矩阵分解方法。
a. 模型特性
- 矩阵分解：通过分解用户-物品评分矩阵来得到用户的偏好和物品的属性。
- 多层感知机：通过多个隐藏层的神经网络来学习用户和物品之间的非线性关系。
- 融合模型：将 MF 和 MLP 的结果融合起来，以获得更好的预测效果。
b. 设计思路
- 用户和物品嵌入：与 NCF 类似，首先将用户和物品 ID 映射到低维向量空间。
- 矩阵分解部分：通过简单的内积操作来估计用户对物品的评分。
- MLP 部分：将用户和物品的嵌入向量拼接后输入到 MLP 中，通过多层全连接网络来学习更复杂的交互模式。
- 结果融合：将 MF 部分和 MLP 部分的结果加权融合，得到最终的预测评分。
c. 示例代码
import numpy as np
import pandas as pd
np.random.seed(42)
num_users = 1000
num_items = 500
num_interactions = num_users * num_items
user_ids = np.random.randint(0, num_users, size=num_interactions)
item_ids = np.random.randint(0, num_items, size=num_interactions)
ratings = np.random.randint(1, 6, size=num_interactions)
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'rating': ratings
})
# 显示数据样本
print(data.head())
users = data['user_id'].values
items = data['item_id'].values
ratings = data['rating'].valuesimport tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Concatenate, Dense, Flatten, Multiply, Add
tf.random.set_seed(42)
# 定义输入层
user_input = Input(shape=(1,), name='user_input')
item_input = Input(shape=(1,), name='item_input')
# 定义嵌入层
embedding_dim = 8 # 嵌入向量的维度
user_embedding = Embedding(input_dim=num_users + 1, output_dim=embedding_dim, name='user_embedding')(user_input)
item_embedding = Embedding(input_dim=num_items + 1, output_dim=embedding_dim, name='item_embedding')(item_input)
# 展平嵌入向量
user_flatten = Flatten()(user_embedding)
item_flatten = Flatten()(item_embedding)
# MF 部分
mf_output = Multiply()([user_flatten, item_flatten])
# MLP 部分
mlp_concat = Concatenate()([user_flatten, item_flatten])
mlp_fc1 = Dense(64, activation='relu')(mlp_concat)
mlp_fc2 = Dense(32, activation='relu')(mlp_fc1)
mlp_output = Dense(1, activation='linear')(mlp_fc2)
# 结果融合
combined_output = Add()([mf_output, mlp_output])
# 创建模型
model = Model(inputs=[user_input, item_input], outputs=combined_output)
# 编译模型
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# 查看模型结构
model.summary()
# 准备训练数据
X_train = {
'user_input': users,
'item_input': items
}
# 训练模型
history = model.fit(X_train, ratings, epochs=5, batch_size=32, validation_split=0.2)6. Deep Cross Network（DCN）
将交叉网络（Cross Network）与深度神经网络相结合，其中交叉网络专门用于显式地建模特征间的高阶交互。
a. 模型特性
- 深度网络（DNN）：用于学习特征的复杂组合。
- 交叉网络（Cross Network）：专门设计用来学习特征间的交互作用。
- 双轨学习：通过并行使用 DNN 和交叉网络，模型能够同时从线性组合和特征交互中学习。
- 高效性：相比于纯深度模型，DCN 能够更有效地利用计算资源。
b. 设计思路
- 交叉网络：通过一系列的权重矩阵和偏置项来模拟多项式特征的交叉。每一层的输出都是前一层输出与输入特征的加权和加上一个偏置项。
- 深度网络：类似于普通的深层神经网络，用于捕捉特征间复杂的非线性关系。
- 融合输出：最终的预测是通过将交叉网络和深度网络的输出连接起来，再经过全连接层得到的。
c. 示例代码
import numpy as np
import pandas as pd
# 随机生成一些数据
np.random.seed(42)
num_samples = 1000
num_features = 10
features = {f'feature_{i}': np.random.rand(num_samples) for i in range(num_features)}
labels = {'click': np.random.randint(0, 2, size=num_samples)}
data = pd.DataFrame({**features, **labels})
# 显示数据样本
print(data.head())import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Dense, Concatenate, Layer
tf.random.set_seed(42)
class CrossLayer(Layer):
def __init__(self, l2_reg=0., **kwargs):
self.l2_reg = l2_reg
super(CrossLayer, self).__init__(**kwargs)
def build(self, input_shape):
dim = int(input_shape[-1][-1])
self.kernel = self.add_weight(name='kernel',
shape=(dim, 1),
initializer='glorot_uniform',
regularizer=tf.keras.regularizers.l2(self.l2_reg),
trainable=True)
self.bias = self.add_weight(name='bias',
shape=(dim,),
initializer='zeros',
trainable=True)
super(CrossLayer, self).build(input_shape)
def call(self, inputs):
x_0, x = inputs
return x_0 * (tf.tensordot(x, self.kernel, axes=1) + 1) + self.bias + x
def compute_output_shape(self, input_shape):
return input_shape
def build_dcn_model(feature_dim, cross_num, dnn_hidden_units, l2_reg_cross=0., l2_reg_dnn=0.):
inputs = Input(shape=(feature_dim,))
x_0 = inputs
x = inputs
# Cross Network
for _ in range(cross_num):
x = CrossLayer(l2_reg=l2_reg_cross)([x_0, x])
# DNN
for unit in dnn_hidden_units:
x = Dense(unit, activation='relu', kernel_regularizer=tf.keras.regularizers.l2(l2_reg_dnn))(x)
# Combine and Output
combined = Concatenate()([x, inputs])
output = Dense(1, activation='sigmoid')(combined)
model = Model(inputs=inputs, outputs=output)
return model
# 模型参数
feature_dim = num_features
cross_num = 2
dnn_hidden_units = [128, 64]
# 构建模型
dcn_model = build_dcn_model(feature_dim, cross_num, dnn_hidden_units)
# 编译模型
dcn_model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
# 显示模型结构
dcn_model.summary()
# 准备训练数据
X_train = data.drop('click', axis=1).values
y_train = data['click'].values
# 训练模型
history = dcn_model.fit(X_train, y_train, epochs=5, batch_size=32, validation_split=0.2)7. Factorization-supported Neural Networks（FNN）
在输入到神经网络之前先进行因子分解，以增强模型的学习能力。
a. 模型特性
- 因子分解：通过因子分解技术来提取用户和项目的低维表示（即嵌入向量），这些向量能够捕获每个用户和项目的基本属性。
- 神经网络：利用神经网络来学习更复杂的非线性特征组合，从而提高模型的预测准确性。
- 减少过拟合：通过引入正则化项和使用 Dropout 等技术来避免过拟合问题。
- 可扩展性：模型可以很容易地扩展到大规模数据集上。
b. 设计思路
- 嵌入层：为每个用户和项目分配一个低维向量表示，通常通过查找表的方式实现。
- 交叉特征：通过计算嵌入向量之间的元素乘积来创建交叉特征，这有助于捕捉用户和项目之间隐含的相互作用。
- 全连接层：将嵌入向量及其交叉特征连接起来，通过多层全连接神经网络来学习高级抽象特征。
- 输出层：最终的输出层通常是单个节点，用于预测评分或点击率等目标变量。
c. 示例代码
import numpy as np
import pandas as pd
np.random.seed(42)
num_users = 100
num_items = 50
num_interactions = 1000
user_ids = np.random.choice(num_users, size=num_interactions)
item_ids = np.random.choice(num_items, size=num_interactions)
ratings = np.random.randint(1, 6, size=num_interactions)
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'rating': ratings
})
# 显示数据样本
print(data.head())import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Embedding, Dense, Flatten, Multiply, concatenate
tf.random.set_seed(42)
# 定义输入层
user_input = Input(shape=(1,), name='user_input')
item_input = Input(shape=(1,), name='item_input')
# 定义嵌入层
embedding_dim = 8 # 嵌入向量的维度
user_embedding = Embedding(input_dim=num_users + 1, output_dim=embedding_dim, name='user_embedding')(user_input)
item_embedding = Embedding(input_dim=num_items + 1, output_dim=embedding_dim, name='item_embedding')(item_input)
# 展平嵌入向量
user_vec = Flatten()(user_embedding)
item_vec = Flatten()(item_embedding)
# 交叉特征
crossed_features = Multiply()([user_vec, item_vec])
# 拼接原始特征与交叉特征
concatenated_features = concatenate([user_vec, item_vec, crossed_features])
# 全连接层
hidden_layer_1 = Dense(64, activation='relu')(concatenated_features)
hidden_layer_2 = Dense(32, activation='relu')(hidden_layer_1)
output_layer = Dense(1, activation='linear')(hidden_layer_2)
# 创建模型
fnn_model = Model(inputs=[user_input, item_input], outputs=output_layer)
# 编译模型
fnn_model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# 查看模型结构
fnn_model.summary()
# 准备训练数据
X_train = {'user_input': user_ids, 'item_input': item_ids}
y_train = ratings
# 训练模型
history = fnn_model.fit(X_train, y_train, epochs=10, batch_size=32, validation_split=0.2)8. AutoInt（Automatic Integration）
利用自注意力机制（Self-Attention Mechanism）自动学习特征间的关系，适用于稀疏数据场景。
a. 模型特性
- 自动特征交互：AutoInt 模型能够自动地识别和学习特征之间的复杂交互，而不需要人工选择交互项。
- 自注意力机制：利用自注意力机制（Self-Attention Mechanism），AutoInt 可以有效地捕捉不同特征之间的依赖关系。
- 灵活性：该模型可以轻松地与其他类型的特征工程方法相结合，提高模型的整体性能。
b. 设计思路
- 特征嵌入：将原始的稀疏特征（如用户 ID、物品 ID 等）转换成密集的低维向量。
- 自注意力层：通过自注意力机制计算特征间的相互影响，从而实现特征交互。
- 堆叠注意力层：可以通过堆叠多个注意力层来进一步增强模型的学习能力。
- 输出层：最后将注意力机制产生的特征交互结果传递给一个全连接层，以进行最终的预测。
c. 示例代码
import numpy as np
import pandas as pd
import tensorflow as tf
from tensorflow.keras.layers import Layer, Embedding, Dense, Dropout, Input
from tensorflow.keras.models import Model
# 生成数据
num_users = 100
num_items = 50
num_interactions = 1000
user_ids = np.random.randint(0, num_users, size=num_interactions)
item_ids = np.random.randint(0, num_items, size=num_interactions)
ratings = np.random.randint(1, 6, size=num_interactions)
data = pd.DataFrame({
'user_id': user_ids,
'item_id': item_ids,
'rating': ratings
})
# 数据准备
user = data['user_id'].values
item = data['item_id'].values
rating = data['rating'].values
# 定义自注意力层
class SelfAttention(Layer):
def __init__(self, embed_dim, num_heads=1, dropout_rate=0.0, **kwargs):
super(SelfAttention, self).__init__(**kwargs)
self.embed_dim = embed_dim
self.num_heads = num_heads
self.dropout_rate = dropout_rate
def build(self, input_shape):
self.q_dense = Dense(self.embed_dim, use_bias=False)
self.k_dense = Dense(self.embed_dim, use_bias=False)
self.v_dense = Dense(self.embed_dim, use_bias=False)
self.dropout = Dropout(self.dropout_rate)
def call(self, inputs, training=None):
q = self.q_dense(inputs)
k = self.k_dense(inputs)
v = self.v_dense(inputs)
attn_scores = tf.matmul(q, k, transpose_b=True) / tf.math.sqrt(
tf.cast(self.embed_dim, tf.float32)
)
attn_weights = tf.nn.softmax(attn_scores, axis=-1)
attn_output = tf.matmul(self.dropout(attn_weights, training=training), v)
return attn_output
# 定义输入层
user_input = Input(shape=(1,), name='user_input')
item_input = Input(shape=(1,), name='item_input')
# 定义嵌入层
embedding_dim = 8 # 嵌入向量的维度
user_embedding = Embedding(input_dim=num_users + 1, output_dim=embedding_dim, name='user_embedding')(user_input)
item_embedding = Embedding(input_dim=num_items + 1, output_dim=embedding_dim, name='item_embedding')(item_input)
# 展平嵌入向量
user_flatten = tf.squeeze(user_embedding, axis=1)
item_flatten = tf.squeeze(item_embedding, axis=1)
# 拼接用户和物品的嵌入向量
concat_embeddings = tf.stack([user_flatten, item_flatten], axis=1)
# 自注意力层
attention_layer = SelfAttention(embed_dim=embedding_dim, num_heads=1)(concat_embeddings)
attention_output = tf.reduce_sum(attention_layer, axis=1)
# 输出层
output = Dense(1, activation='linear')(attention_output)
# 创建模型
model = Model(inputs=[user_input, item_input], outputs=output)
# 编译模型
model.compile(optimizer='adam', loss='mse', metrics=['mae'])
# 训练模型
history = model.fit(
[user, item],
rating,
batch_size=32,
epochs=10,
validation_split=0.2,
verbose=1
)
# 评估模型
evaluation = model.evaluate([test_user, test_item], test_rating, verbose=0)
print(f'Test Loss: {evaluation[0]}, Test MAE: {evaluation[1]}')9. Recurrent Neural Networks（RNN）for Sequences
使用 RNN 来捕捉用户行为的时间序列特性，比如用户的点击流数据。
a. 模型特点
- 记忆效应：RNN 能够记住先前的信息，并将其与当前的信息相结合，从而更好地理解序列数据。
- 循环结构：RNN 具有一个循环的连接结构，允许信息在时间上被传递。
- 短期记忆：由于简单的 RNN 容易受到梯度消失或梯度爆炸问题的影响，它们通常只能记住短期的信息。
- 共享权重：在网络的不同时间步上，RNN 使用相同的权重矩阵，这有助于减少参数的数量并提高模型的泛化能力。
- 梯度消失/爆炸问题：由于长时间依赖的问题，RNN 在训练过程中可能会遇到梯度消失或梯度爆炸问题，这限制了它在长序列上的表现。
- 变种模型：为了克服这些问题，发展出了 LSTM（Long Short-Term Memory）和 GRU（Gated Recurrent Unit）等变种模型。
b. 设计思路
- 循环单元：在每个时间步 t，RNN 接收当前输入 xt 和前一时刻的状态 ht−1，计算当前状态 ht。
- 输出层：状态 ht 可以被进一步处理以生成输出 yt 或者直接用于下一个时间步的输入。
- 训练：通过反向传播算法（Backpropagation Through Time, BPTT）来更新权重。
总结：RNN 的基本思想是在处理序列数据时，将前一时刻的状态信息传递到下一时刻，以便当前时刻可以利用先前的信息。这通常是通过隐藏层的状态实现的，该状态在每个时间步都会更新。
c. 示例代码
import numpy as np
import tensorflow as tf
from tensorflow.keras.preprocessing.text import Tokenizer
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, SimpleRNN
# 示例文本数据
text = "hello world how are you hello world how are you"
# 文本预处理
tokenizer = Tokenizer(char_level=True)
tokenizer.fit_on_texts([text])
encoded = tokenizer.texts_to_sequences([text])[0]
# 定义词汇表大小
vocab_size = len(tokenizer.word_index) + 1
# 序列长度
seq_length = 5
sequences = list()
for i in range(1, len(encoded)):
seq = encoded[i - seq_length:i]
line = ' '.join(map(str, seq))
sequences.append(line)
# 将序列转换为 X, y 数据集
sequences = [list(map(int, s.split())) for s in sequences]
X, y = [], []
for seq in sequences:
if len(seq) != 0:
X.append(seq[:-1])
y.append(seq[-1])
X = np.array(X)
y = to_categorical(y, num_classes=vocab_size)
# 创建 one-hot 编码的 X_train
X_train = np.zeros((len(X), seq_length - 1, vocab_size), dtype=np.int8)
for i in range(len(X)):
for t, word_idx in enumerate(X[i]):
X_train[i, t, word_idx] = 1
# 定义模型
model = Sequential()
model.add(SimpleRNN(50, input_shape=(seq_length - 1, vocab_size)))
model.add(Dense(vocab_size, activation='softmax'))
# 编译模型
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
# 查看模型结构
model.summary()
# 训练模型
model.fit(X_train, y, epochs=200, verbose=2)
# 预测新序列
def generate_text(model, tokenizer, seq_len, seed_text, n_chars):
output_text = []
input_text = seed_text
for _ in range(n_chars):
encoded = tokenizer.texts_to_sequences([input_text])[0]
encoded = tf.one_hot(encoded, depth=vocab_size)
encoded = tf.reshape(encoded, (1, -1, vocab_size))
y_pred = model.predict(encoded, verbose=0)
y_pred = tf.argmax(y_pred, axis=-1)
output_text.append(tokenizer.index_word[int(y_pred)])
input_text += tokenizer.index_word[int(y_pred)]
return ' '.join(output_text)
# 使用模型生成文本
seed_text = "hell"
generated_text = generate_text(model, tokenizer, seq_length, seed_text, 10)
print(generated_text)10. Convolutional Neural Networks（CNN）for Content
利用卷积神经网络来分析文本、图像等内容信息，进而推荐相似的内容给用户。
a. 模型特点
- 局部感知：CNN 中的神经元只与输入的一部分相连，这反映了特征的局部相关性。
- 权值共享：同一特征映射内的神经元共享权重，这样可以减少模型的参数数量，同时捕捉到平移不变性。
- 池化操作：通过池化层（如最大池化或平均池化）降低空间维度，从而减少计算量并控制过拟合。
b. 设计思路
- CNN 的设计灵感来源于生物视觉系统的研究，特别是猫的初级视皮层中发现的细胞对特定区域的刺激响应。
- 其基本组件包括卷积层、激活函数、池化层以及全连接层。
- 卷积层负责提取特征，池化层负责下采样，全连接层则用来分类或其他高层任务。
c. 示例代码
import numpy as np
import tensorflow as tf
from tensorflow.keras import layers, models
import matplotlib.pyplot as plt
from math import ceil
# 设置随机种子以保证结果可复现
np.random.seed(42)
# 定义数据集大小和类别数
num_samples = 1000
num_classes = 2
# 生成两个不同均值的二维高斯分布样本
class_0 = np.random.multivariate_normal([5, 5], [[1, .75], [.75, 1]], num_samples // 2)
class_1 = np.random.multivariate_normal([10, 10], [[1, .75], [.75, 1]], num_samples - num_samples // 2)
# 将两类数据合并
features = np.vstack([class_0, class_1])
labels = np.hstack([np.zeros(num_samples // 2), np.ones(num_samples - num_samples // 2)])
# 打乱数据顺序
indices = np.arange(features.shape[0])
np.random.shuffle(indices)
features = features[indices]
labels = labels[indices]
# 将数据转换为图像形式，这里将其扩展到 28x28 大小
image_size = 28
features = np.reshape(features, (-1, 2))
images = np.zeros((num_samples, image_size, image_size))
for i, point in enumerate(features):
x, y = point * 28 / 15 # 将点映射到图像范围内
x, y = ceil(x), ceil(y)
images[i, x, y] = 1 # 在图像上标记这个点
# 将图像数据归一化
images = images[..., np.newaxis] / 1. # 添加通道维度并归一化
# 划分训练集和测试集
split = int(0.8 * num_samples)
train_images, test_images = images[:split], images[split:]
train_labels, test_labels = labels[:split], labels[split:]
# 构建模型
model = models.Sequential([
layers.Conv2D(32, (3, 3), activation='relu', input_shape=(image_size, image_size, 1)),
layers.MaxPooling2D((2, 2)),
layers.Conv2D(64, (3, 3), activation='relu'),
layers.MaxPooling2D((2, 2)),
layers.Flatten(),
layers.Dense(64, activation='relu'),
layers.Dense(num_classes, activation='softmax')
])
# 编译模型
model.compile(optimizer='adam',
loss='sparse_categorical_crossentropy',
metrics=['accuracy'])
# 训练模型
history = model.fit(train_images, train_labels, epochs=5,
validation_data=(test_images, test_labels))
# 评估模型
test_loss, test_acc = model.evaluate(test_images, test_labels, verbose=2)
print(f'\nTest accuracy: {test_acc}')
# 可视化部分训练样本
plt.figure(figsize=(10, 10))
for i in range(25):
plt.subplot(5, 5, i + 1)
plt.xticks([])
plt.yticks([])
plt.grid(False)
plt.imshow(train_images[i].reshape(image_size, image_size), cmap=plt.cm.binary)
plt.show()11. Graph Convolutional Networks（GCN）
利用图卷积网络来捕捉用户、项目以及其他实体之间的关系，适用于社交网络推荐。
a. 模型特点
- 局部性：GCN 通过邻居节点的信息更新节点的表示，这样可以捕获局部结构特征。
- 参数共享：由于图中的节点具有相似性，GCN 在所有节点上共享同一组权重，类似于 CNN 中的权重共享。
- 稀疏性和高效性：GCN 可以高效地处理大规模稀疏邻接矩阵，这是社交网络和推荐系统中的常见情况。
- 多层结构：GCN 可以构建多层结构，每一层聚合邻居的信息，从而可以捕获不同距离节点的关系。
b. 设计思路
- 定义卷积操作：在图中定义一个卷积操作，使得每个节点能够从其邻居节点那里收集信息。
- 聚合邻居信息：节点将其邻居的信息进行聚合，通常是通过求平均或求和。
- 非线性变换：使用激活函数对聚合后的信息进行非线性变换。
- 重复上述过程：通过多层的迭代，使得信息能够在更大的范围内传播。
c. 示例代码
import numpy as np
import tensorflow as tf
from tensorflow.keras.layers import Layer, Dense
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
# 示例图数据
# 用户数
num_users = 4
# 项目数
num_items = 3
# 特征数
num_features = 5
# 总节点数
num_nodes = num_users + num_items
# 邻接矩阵 A（用户-项目）
adj_matrix = np.array([
[0, 1, 1, 0, 0],
[1, 0, 0, 0, 0],
[0, 1, 0, 0, 0],
[0, 0, 1, 0, 0],
[0, 0, 0, 0, 0],
[0, 0, 0, 0, 1],
[0, 0, 0, 0, 0]
], dtype=np.float32)
# 特征矩阵 X（用户特征）
features = tf.random.uniform([num_nodes, num_features])
# 对邻接矩阵进行归一化
def normalize_adj(adj):
"""Symmetrically normalize adjacency matrix."""
rowsum = np.sum(adj, axis=1)
d_inv_sqrt = np.power(rowsum, -0.5).flatten()
d_inv_sqrt[np.isinf(d_inv_sqrt)] = 0.
d_mat_inv_sqrt = np.diag(d_inv_sqrt)
return d_mat_inv_sqrt.transpose().dot(adj).dot(adj.transpose())
# 归一化邻接矩阵
norm_adj = normalize_adj(adj_matrix)
# 定义 GCN 层
class GCNLayer(Layer):
def __init__(self, output_dim, activation=None, **kwargs):
super(GCNLayer, self).__init__(**kwargs)
self.output_dim = output_dim
self.activation = activation
self.dense = Dense(output_dim, use_bias=False)
def call(self, inputs):
x, adj = inputs
# 图卷积
output = tf.matmul(adj, self.dense(x))
if self.activation is not None:
output = self.activation(output)
return output
# 构建模型
input_features = tf.keras.Input(shape=(num_features,))
input_adj = tf.keras.Input(shape=(num_nodes, num_nodes))
x = GCNLayer(16, activation=tf.nn.relu)([input_features, input_adj])
x = GCNLayer(5, activation=None)([x, input_adj])
output = x
model = Model(inputs=[input_features, input_adj], outputs=output)
# 编译模型
model.compile(optimizer=Adam(learning_rate=0.01), loss='mse')
# 准备训练数据
X_train = features
y_train = features
# 训练模型
history = model.fit([X_train, norm_adj], y_train, epochs=10, verbose=1)
# 测试模型
predictions = model.predict([features, norm_adj])
print(predictions)12. Hybrid Models（混合模型）
结合多种模型的优点，如结合基于内容的过滤和协同过滤的方法。
a. 模型特点
- 互补性：基于内容的过滤可以利用物品的内容信息（如描述、标签等），而协同过滤可以利用用户的历史行为数据。结合两者可以弥补各自的不足。
- 鲁棒性：单一模型可能在某些情况下表现不佳，如冷启动问题（新用户或新物品缺乏历史数据），而混合模型可以更好地应对这些问题。
- 灵活性：可以根据具体应用场景和可用数据灵活调整模型的组成部分。
b. 设计思路
- 基于内容的过滤：通过分析物品的内容特征（如文本描述、标签等）来推荐相似的物品。这种方法适用于新用户或新物品。
- 协同过滤：基于用户的行为数据（如评分、点击等）来发现用户之间的相似性，并推荐相似用户喜欢的物品。这种方法适用于有足够历史数据的场景。
- 组合策略：将两种方法的结果进行组合，可以通过加权平均或其他方式来综合考虑两种推荐结果。
c. 示例代码
import numpy as np
import tensorflow as tf
from tensorflow.keras.layers import Input, Embedding, Dot, Flatten, Concatenate, Dense
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
# 示例数据
# 用户数
num_users = 100
# 物品数
num_items = 50
# 物品内容特征数
num_item_features = 10
# 用户-物品评分矩阵
ratings = np.random.randint(1, 6, size=(num_users, num_items))
# 物品内容特征矩阵
item_features = np.random.rand(num_items, num_item_features)
# 定义物品特征输入
item_input = Input(shape=(num_item_features,))
# 物品特征编码层
item_embedding = Dense(16, activation='relu')(item_input)
item_embedding = Dense(8, activation='relu')(item_embedding)
# 基于内容的过滤模型
content_model = Model(inputs=item_input, outputs=item_embedding)
# 定义用户和物品输入
user_input = Input(shape=(1,))
item_input = Input(shape=(1,))
# 用户和物品嵌入层
user_embedding = Embedding(input_dim=num_users, output_dim=8)(user_input)
item_embedding = Embedding(input_dim=num_items, output_dim=8)(item_input)
# 嵌入层输出形状调整
user_embedding = Flatten()(user_embedding)
item_embedding = Flatten()(item_embedding)
# 协同过滤模型
collaborative_output = Dot(axes=1)([user_embedding, item_embedding])
# 协同过滤模型
collaborative_model = Model(inputs=[user_input, item_input], outputs=collaborative_output)
# 定义混合模型的输入
user_input = Input(shape=(1,))
item_input = Input(shape=(1,))
item_feature_input = Input(shape=(num_item_features,))
# 获取基于内容的特征表示
content_output = content_model(item_feature_input)
# 获取协同过滤的输出
collaborative_output = collaborative_model([user_input, item_input])
# 组合两种输出
combined_output = Concatenate()([content_output, collaborative_output])
# 最终的预测输出
final_output = Dense(1)(combined_output)
# 混合模型
hybrid_model = Model(inputs=[user_input, item_input, item_feature_input], outputs=final_output)
# 编译模型
hybrid_model.compile(optimizer=Adam(learning_rate=0.01), loss='mse')
# 准备训练数据
# 用户-物品对
user_item_pairs = []
ratings_list = []
for user_id in range(num_users):
for item_id in range(num_items):
user_item_pairs.append([user_id, item_id])
ratings_list.append(ratings[user_id, item_id])
user_item_pairs = np.array(user_item_pairs)
ratings_list = np.array(ratings_list)
# 训练模型
history = hybrid_model.fit(
[user_item_pairs[:, 0], user_item_pairs[:, 1], item_features[user_item_pairs[:, 1]]],
ratings_list,
epochs=10,
verbose=1
)
# 测试模型
predictions = hybrid_model.predict(
[user_item_pairs[:, 0], user_item_pairs[:, 1], item_features[user_item_pairs[:, 1]]]
)
print(predictions)