Python微信订餐小程序课程视频

https://edu.csdn.net/course/detail/36074

Python实战量化交易理财系统

https://edu.csdn.net/course/detail/35475

目录

前言层次结构概述创建张量计算图自动微分数据输入Numpy构建数据生成器构建数据keras.utils.Sequence搭建模型顺序建模函数式API建模自定义建模损失函数自定义损失函数度量函数函数自定义度量类自定义度量优化器回调函数Tensorboard训练模型model.fittrain_on_batch自定义训练循环评估模型推理模型保存和加载模型Keras保存和加载模型TensorFlow原生方式保存和加载汇总GPU训练指定GPU训练单机多卡训练模型部署tensorflow-servingTensorFlow Lite参考

前言

　　至于为什么写这个教程，首先是为了自己学习做个记录，其次是因为Tensorflow的API写的很好，但是他的教程写的太乱了，不适合新手学习。tensorflow 1 和tensorflow 2 有相似之处但是不兼容，tensorflow 2将keras融合了。TensorFlow™ 是一个采用 数据流图（data flow graphs），用于数值计算的开源软件库。图中得节点(Nodes)表示数学操作，图中的线(edges)则表示在节点间相互联系的多维数据数组，即张量（tensor）。它灵活的架构让你可以在多种平台上展开计算，例如台式计算机中的一个或多个CPU（或GPU），服务器，移动设备等等。

TensorFlow的主要优点：

• 灵活性：支持底层数值计算，C++自定义操作符
• 可移植性：从服务器到PC到手机，从CPU到GPU到TPU
• 分布式计算：分布式并行计算，可指定操作符对应计算设备

层次结构

　　TensorFlow的层次结构从低到高可以分成如下五层：硬件层，内核层，低阶API，中阶API，高阶API。

• 第一层：硬件层，TensorFlow支持CPU、GPU或TPU加入计算资源池。
• 第二层：内核层，为C++实现的内核，kernel可以跨平台分布运行。
• 第三层：低阶API，由Python实现的操作符，提供了封装C++内核的低级API指令。主要包括各种张量操作算子、计算图、自动微分。如tf.Variable，tf.constant，tf.function，tf.GradientTape，tf.nn.softmax...
• 第四层：中阶API，由Python实现的模型组件，对低级API进行了函数封装。主要包括数据管道(tf.data)、特征列(tf.feature_column)、激活函数(tf.nn)、模型层(tf.keras.layers)、损失函数(tf.keras.losses)、评估函数(tf.keras.metrics)、优化器(tf.keras.optimizers)、回调函数(tf.keras.callbacks) 等等。
• 第五层：高阶API，由Python实现的模型成品。主要为tf.keras.models提供的模型的类接口，主要包括：模型的构建（Sequential、functional API、Model子类化）、型的训练（内置fit方法、内置train_on_batch方法、自定义训练循环、单GPU训练模型、多GPU训练模型、TPU训练模型）、模型的部署（tensorflow serving部署模型、使用spark(scala)调用tensorflow模型）。

概述

创建张量

tf.constant(value, dtype=tf.float32)    # 常数
tf.range(start, limit=None, delta=1)    # 生成一个范围内间隔为delta的 张量
tf.linspace(start, stop, num) # 在一个间隔内生成均匀间隔的值
tf.zeros()  # 创建全0张量
tf.ones()  # 创建全1张量
tf.zeros\_like(input)  # 创建和input一样大小的张量
tf.fill(dims, value)    # 创建shape为dim，全为value的张量

tf.random.uniform([5], minval=0, maxval=10)     # 均匀分布随机
tf.random.normal([3, 3], mean=0.0, stddev=1.0)  # 正态分布随机

tf.Variable(initial\_value) # 变量

tf.Variable:

• name：变量的名字，默认情况下，会自动获得唯一的变量名
• trainable：设置为 False 可以关闭梯度。例如，训练计步器就是一个不需要梯度的变量

tf.rank(a)：求矩阵的秩

变量的设备位置

为了提高性能，TensorFlow 会尝试将张量和变量放在与其 dtype 兼容的最快设备上。这意味着如果有 GPU，那么大部分变量都会放置在 GPU 上，不过，我们可以重写变量的位置。

with tf.device('CPU:0'):
 # Create some tensors
  a = tf.Variable([[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]])
 b = tf.constant([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]])
 c = tf.matmul(a, b)

print(c)

使用assign重新分配张量

a.assign([5, 6])  # a = [5, 6]
a.assign\_add([2, 3])  # a = a+[2,3]
a.assign\_sub([7, 9])  # a=a-[7,9]

维度变换

维度变换相关函数主要有 tf.reshape，tf.squeeze，tf.expand_dims，tf.transpose。

• tf.reshape()：改变张量的形状
• tf.squeeze()：减少维度为1的维度
• tf.expand_dims(input, axis)：增加维度
• tf.transpose(a, perm=None)：交换维度

tf.reshape可以改变张量的形状，但是其本质上不会改变张量元素的存储顺序，所以，该操作实际上非常迅速，并且是可逆的。

合并分隔

和numpy类似，可以用tf.concat和tf.stack方法对多个张量进行合并，可以用tf.split方法把一个张量分割成多个张量。

tf.concat和tf.stack有略微的区别，tf.concat是连接，不会增加维度，而tf.stack是堆叠，会增加维度。

a = tf.constant([[1.0, 2.0], [3.0, 4.0]])  # (2,2)
b = tf.constant([[5.0, 6.0], [7.0, 8.0]])  # (2,2)

c = tf.concat([a, b], axis=0)  # (4, 2)
c = tf.stack([a, b], axis=0)  # (2, 2, 2)
d = tf.split(c, 2, axis=0)  # [(1, 2, 2),(1, 2, 2)]

Tensor与Array的转换

c = np.array(b)     # tensor 转 np
c = b.numpy()       # tensor 转 np
tf.convert\_to\_tensor(c)     # np 转 tensor

数学运算

tf.add(a,b)        # 加法 a+b
tf.multiply(a,b)       # 逐元素乘法a*b
tf.matmul(a,b)     # 矩阵乘法a@b

类型转换

tensorflow支持的模型有：tf.float16、tf.float64、tf.int8、tf.int16、tf.int32...

a = tf.constant([2.2, 3.3, 4.4], dtype=tf.float64)
b = tf.cast(a, dtype=tf.float16)    # 类型转换

计算图

　　有三种计算图的构建方式：静态计算图，动态计算图，以及Autograph。在TensorFlow1.0时代，采用的是静态计算图，需要先使用TensorFlow的各种算子创建计算图，然后再开启一个会话Session，显式执行计算图。而在TensorFlow2.0时代，采用的是动态计算图，即每使用一个算子后，该算子会被动态加入到隐含的默认计算图中立即执行得到结果，而无需开启Session。使用动态计算图(Eager Excution)的好处是方便调试程序，它会让TensorFlow代码的表现和Python原生代码的表现一样，写起来就像写numpy一样，各种日志打印，控制流全部都是可以使用的。使用动态计算图的缺点是运行效率相对会低一些。因为使用动态图会有许多次Python进程和TensorFlow的C++进程之间的通信。而静态计算图构建完成之后几乎全部在TensorFlow内核上使用C++代码执行，效率更高。此外静态图会对计算步骤进行一定的优化，剪去和结果无关的计算步骤。

　　如果需要在TensorFlow2.0中使用静态图，可以使用@tf.function装饰器将普通Python函数转换成对应的TensorFlow计算图构建代码。运行该函数就相当于在TensorFlow1.0中用Session执行代码。使用tf.function构建静态图的方式叫做 Autograph。当然Autograph机制能够转换的代码并不是没有任何约束的，有一些编码规范需要遵循，否则可能会转换失败或者不符合预期。

1. 被@tf.function修饰的函数应尽可能使用TensorFlow中的函数而不是Python中的其他函数。例如使用tf.print而不是print，使用tf.range而不是range，使用tf.constant(True)而不是True.
2. 避免在@tf.function修饰的函数内部定义tf.Variable
3. 被@tf.function修饰的函数不可修改该函数外部的列表或字典等数据结构变量

　　计算图由节点(nodes)和线(edges)组成。节点表示操作符Operator，或者称之为算子，线表示计算间的依赖。实线表示有数据传递依赖，传递的数据即张量。虚线通常可以表示控制依赖，即执行先后顺序。

import tensorflow as tf

# 使用autograph构建静态图
@tf.function
def strjoin(x,y):
 z =  tf.strings.join([x,y],separator = " ")
 tf.print(z)
 return z

result = strjoin(tf.constant("hello"),tf.constant("world"))

print(result)

您可以像这样测量静态图和动态图性能差异：

x = tf.random.uniform(shape=[10, 10], minval=-1, maxval=2, dtype=tf.dtypes.int32)

def power(x, y):
 result = tf.eye(10, dtype=tf.dtypes.int32)
 for \_ in range(y):
 result = tf.matmul(x, result)
 return result

print("Eager execution:", timeit.timeit(lambda: power(x, 100), number=1000))        # 2.56378621799

# 将python函数转换为图形
power\_as\_graph = tf.function(power)
print("Graph execution:", timeit.timeit(lambda: power\_as\_graph(x, 100), number=1000))        # 0.683253670

View Code
我们还可以再函数前使用装饰器 @tf.function 调用tf.function，同时也可以使用 tf.config.run_functions_eagerly(True) 关闭Function创建和运行图形的能力。

　　前面在介绍Autograph的编码规范时提到构建Autograph时应该避免在@tf.function修饰的函数内部定义tf.Variable。但是如果在函数外部定义tf.Variable的话，又会显得这个函数有外部变量依赖，封装不够完美。一种简单的思路是定义一个类，并将相关的tf.Variable创建放在类的初始化方法中。而将函数的逻辑放在其他方法中。

class DemoModule(tf.Module):
 def \_\_init\_\_(self, init\_value=tf.constant(0.0), name=None):
 super(DemoModule, self).\_\_init\_\_(name=name)
 with self.name\_scope: # 相当于with tf.name\_scope("demo\_module")
            self.x = tf.Variable(init\_value, dtype=tf.float32, trainable=True)

 @tf.function
 def addprint(self, a):
 with self.name\_scope:
 self.x.assign\_add(a)
 tf.print(self.x)
 return self.x

View Code

自动微分

　　自动微分用于训练神经网络的反向传播非常有用，TensorFlow 会记住在前向传递过程中哪些运算以何种顺序发生。随后，在后向传递期间，以相反的顺序遍历此运算列表来计算梯度。

Tensorflow一般使用tf.GradientTape来记录正向运算过程，然后反向传播自动计算梯度值。

f(x)=ax2+bx+cf(x)=ax2+bx+cf(x)=ax^2+bx+c

x = tf.Variable(0.0,name = "x",dtype = tf.float32)
a = tf.constant(1.0)
b = tf.constant(-2.0)
c = tf.constant(1.0)

with tf.GradientTape() as tape:
 y = a*tf.pow(x,2) + b*x + c

dy\_dx = tape.gradient(y,x)
print(dy\_dx)    # tf.Tensor(-2.0, shape=(), dtype=float32)

对常量张量也可以求导，只不过需要增加watch

with tf.GradientTape() as tape:
 tape.watch([a,b,c])
 y = a*tf.pow(x,2) + b*x + c

dy\_dx,dy\_da,dy\_db,dy\_dc = tape.gradient(y,[x,a,b,c])
print(dy\_da)    # tf.Tensor(0.0, shape=(), dtype=float32)
print(dy\_dc)    # tf.Tensor(1.0, shape=(), dtype=float32)

View Code
可以求二阶导数

with tf.GradientTape() as tape2:
 with tf.GradientTape() as tape1: 
 y = a*tf.pow(x,2) + b*x + c
 dy\_dx = tape1.gradient(y,x) 
dy2\_dx2 = tape2.gradient(dy\_dx,x)

print(dy2\_dx2)    # tf.Tensor(2.0, shape=(), dtype=float32)

View Code
利用梯度和优化器求最小值

# 求f(x) = a*x**2 + b*x + c的最小值
# 使用optimizer.apply\_gradients

x = tf.Variable(0.0,name = "x",dtype = tf.float32)
a = tf.constant(1.0)
b = tf.constant(-2.0)
c = tf.constant(1.0)

optimizer = tf.keras.optimizers.SGD(learning\_rate=0.01)
for \_ in range(1000):
 with tf.GradientTape() as tape:
 y = a*tf.pow(x,2) + b*x + c
 dy\_dx = tape.gradient(y,x)　　# 计算梯度
    optimizer.apply\_gradients(grads\_and\_vars=[(dy\_dx,x)])　　# 根据梯度更新变量

tf.print("y =",y,"; x =",x)

View Code
如果不想被计算梯度：

with tf.GradientTape(watch\_accessed\_variables=False) as tape:
 pass

使用TensorFlow实现神经网络模型的一般流程包括：

1. 准备数据
2. 定义模型
3. 训练模型
4. 评估模型
5. 推理模型
6. 保存模型

数据输入

　　tensorflow支持 Numpy 数组、Pandas DataFrame、Python 生成器、csv文件、文本文件、文件路径、TFrecords文件等方式构建数据管道。如果您的数据很小并且适合内存，我们建议您使用tf.data.Dataset.from_tensor_slices()从Numpy array构建数据管道

• Dataset：如果您有大型数据集并且需要进行分布式训练
• Sequence：如果您有大型数据集并且需要执行大量在 TensorFlow 中无法完成的自定义 Python 端处理（例如，如果您依赖外部库进行数据加载或预处理）
• 通过tfrecords文件方式构建数据管道较为复杂，需要对样本构建tf.Example后压缩成字符串写到tfrecoreds文件，读取后再解析成tf.Example。但tfrecoreds文件的优点是压缩后文件较小，便于网络传播，加载速度较快。

Numpy构建数据

官方推荐使用 tf.data.Dataset.from_tensors() 或 tf.data.Dataset.from_tensor_slices() 创建数据集，Dataset支持一类特殊的操作Trainformation（打乱、生成epoch...等操作）

• data.map(function)：将转换函数映射到数据集每一个元素
• data.batch(batch_size)：构建batch
• data.shuffle(buffer_size)：随机打乱输入数据，从该缓冲区中随机采样元素
• data.repeat()：repeat的功能就是将整个序列重复多次，一般不带参数
• data.prefetch(tf.data.experimental.AUTOTUNE) ：预先取 数据
• data.take()：采样，从开始位置取前几个元素
• ...

features = np.arange(0, 100, dtype=np.int32)    # # (100,)
labels = np.zeros(100, dtype=np.int32)  # (100,)

data = tf.data.Dataset.from\_tensor\_slices((features, labels))   # 创建数据集
data = data.repeat()    # 无限期地补充数据
data = data.shuffle(buffer\_size=100)    # 打乱数据
data = data.batch(batch\_size=4)     # 批量数据
data = data.prefetch(buffer\_size=1)     # 预取批处理(预加载批处理，消耗更快)

for batch\_x, batch\_y in data.take(5):
 print(batch\_x.shape, batch\_y.shape)     # (4,) (4,)
    break

注意：如果你打算多次调用，你可以使用迭代器的方式：

ite\_data = iter(data)
for i in range(5):
batch\_x, batch\_y = next(ite\_data)
print(batch\_x, batch\_y)

for i in range(5):
batch\_x, batch\_y = next(ite\_data)
print(batch\_x, batch\_y)

View Code

提升管道性能

　　训练深度学习模型常常会非常耗时。模型训练的耗时主要来自于两个部分，一部分来自数据准备，另一部分来自参数迭代。参数迭代过程的耗时通常依赖于GPU来提升。而数据准备过程的耗时则可以通过构建高效的数据管道进行提升。

以下是一些构建高效数据管道的建议。

1. 使用 prefetch 方法让数据准备和参数迭代两个过程相互并行。
2. 使用 interleave 方法可以让数据读取过程多进程执行，并将不同来源数据夹在一起。
3. 使用 map 时设置num_parallel_calls 让数据转换过程多进行执行。
4. 使用 cache 方法让数据在第一个epoch后缓存到内存中，仅限于数据集不大情形。
5. 使用 map转换时，先batch，然后采用向量化的转换方法对每个batch进行转换。

生成器构建数据

def generate\_features():
 # 函数生成一个随机字符串
    def random\_string(length):
 return ''.join(random.choice(string.ascii\_letters) for m in range(length))
 # 返回一个随机字符串、一个随机向量和一个随机整数
    yield random\_string(4), np.random.uniform(size=4), random.randint(0, 10)

data = tf.data.Dataset.from\_generator(generate\_features, output\_types=(tf.string, tf.float32, tf.int32))
data = data.repeat()        # 无限期地补充数据
data = data.shuffle(buffer\_size=100)    # 打乱数据
data = data.batch(batch\_size=4)     # 批量数据(将记录聚合在一起)
data = data.prefetch(buffer\_size=1)     # 预取批量(预加载批量以便更快的消耗)

# Display data.
for batch\_str, batch\_vector, batch\_int in data.take(5):
 # (4,) (4, 4) (4,)
    print(batch\_str.shape, batch\_vector.shape, batch\_int.shape)

keras.utils.Sequence

特别是，keras.utils.Sequence该类提供了一个简单的接口来构建 Python 数据生成器，该生成器可以感知多处理并且可以洗牌。

Sequence必须实现两种方法：

• __getitem__：返回一个batch数据
• __len__：整型，返回batch的数量

import tensorflow as tf
from keras.utils.data\_utils import Sequence

class SequenceDataset(Sequence):
 def \_\_init\_\_(self, batch\_size):
 self.input\_data = tf.random.normal((640, 8192, 1))
 self.label\_data = tf.random.normal((640, 8192, 1))
 self.batch\_size = batch\_size

 def \_\_len\_\_(self):
 return int(tf.math.ceil(len(self.input\_data) / float(self.batch\_size)))

 # 每次输出一个batch
    def \_\_getitem\_\_(self, idx):
 batch\_x = self.input\_data[idx * self.batch\_size:(idx + 1) * self.batch\_size]
 batch\_y = self.label\_data[idx * self.batch\_size:(idx + 1) * self.batch\_size]
 return batch\_x, batch\_y

sequence = SequenceDataset(batch\_size=64)

for batch\_idx, (x, y) in enumerate(sequence):
 print(batch\_idx, x.shape, y.shape)  # tf.float32
    # 0 (64, 8192, 1) (64, 8192, 1)
    break

搭建模型

　　深度学习模型一般由各种模型层组合而成，如果这些内置模型层不能够满足需求，我们也可以通过编写tf.keras.Lambda匿名模型层或继承tf.keras.layers.Layer基类构建自定义的模型层。其中tf.keras.Lambda匿名模型层只适用于构造没有学习参数的模型层。

搭建模型有以下3种方式构建模型：

1. Sequential顺序模型：用于简单的层堆栈， 其中每一层恰好有一个输入张量和一个输出张量
2. 函数式API模型：多输入多输出，或者模型需要共享权重，或者模型具有残差连接等非顺序结构，
3. 继承Model基类自定义模型：如果无特定必要，尽可能避免使用Model子类化的方式构建模型，这种方式提供了极大的灵活性，但也有更大的概率出错

顺序建模

model = keras.Sequential([
 layers.Dense(2, activation="relu", name="layer1"),
 layers.Dense(3, activation="relu", name="layer2"),
 layers.Dense(4, name="layer3"),
])

还可以通过add方法创建顺序模型

model = keras.Sequential()
model.add(layers.Dense(2, activation="relu"))
model.add(layers.Dense(3, activation="relu"))
model.add(layers.Dense(4))

因为模型不知道输入shape，所以起初模型没有权重，因此我们需要告知模型输入shape

# 在第一层添加Input
model.add(**keras.Input**(shape=(4,)))
model.add(layers.Dense(2, activation="relu"))

# 或者在第一层添加input\_shape
model.add(layers.Dense(2, activation="relu", **input\_shape**=(4,)))

顺序模型可以配合add 和 model.summary() 在模型的任何位置查看该层的输入输出。

函数式API建模

　　函数式API搭建模型比Sequential更加灵活，可以处理具有非线性拓扑、共享层甚至多个输入或输出的模型。

inputs = keras.Input(shape=(784,))
x = layers.Dense(64, activation="relu")(inputs)
x = layers.Dense(64, activation="relu")(x)
outputs = layers.Dense(10)(x)
model = keras.Model(inputs=inputs, outputs=outputs, name="mnist\_model")
model.summary() # 查看模型摘要

还可以将模型绘制为图形

keras.utils.plot\_model(model, "my\_first\_model\_with\_shape\_info.png", show\_shapes=True)

补充：函数式模型是可以嵌套的

自定义建模

　　在 TensorFlow 中，模型类的继承关系为：

tf.keras.Model > tf.keras.layers.Layer > tf.Module。

　　通常使用Layer类来定义内部计算块，并使用Model类定义外部模型(训练的对象)。

继承tf.Module栗子：

class SequentialModule(tf.Module):
 def \_\_init\_\_(self, name=None):
 super().\_\_init\_\_(name=name)
 self.dense\_1 = Dense(in\_features=3, out\_features=3)
 self.dense\_2 = Dense(in\_features=3, out\_features=2)

 def \_\_call\_\_(self, x):
 x = self.dense\_1(x)
 return self.dense\_2(x)

my\_model = SequentialModule(name="the\_model")

View Code
继承tf.keras.layers.Layer栗子：

class ResBlock(layers.Layer):
 def \_\_init\_\_(self, kernel\_size, **kwargs):
 super(ResBlock, self).\_\_init\_\_(**kwargs)
 self.kernel\_size = kernel\_size

 def build(self, input\_shape):
 self.conv1 = layers.Conv1D(filters=64, kernel\_size=self.kernel\_size,
 activation="relu", padding="same")
 self.conv2 = layers.Conv1D(filters=32, kernel\_size=self.kernel\_size,
 activation="relu", padding="same")
 self.conv3 = layers.Conv1D(filters=input\_shape[-1],
 kernel\_size=self.kernel\_size, activation="relu", padding="same")
 self.maxpool = layers.MaxPool1D(2)
 super(ResBlock, self).build(input\_shape) # 相当于设置self.built = True

    def call(self, inputs):
 x = self.conv1(inputs)
 x = self.conv2(x)
 x = self.conv3(x)
 x = layers.Add()([inputs, x])
 x = self.maxpool(x)
 return x

 # 如果要让自定义的Layer通过Functional API 组合成模型时可以序列化，需要自定义get\_config方法。
    def get\_config(self):
 config = super(ResBlock, self).get\_config()
 config.update({'kernel\_size': self.kernel\_size})
 return config

resblock = ResBlock(kernel\_size=3)
resblock.build(input\_shape=(None, 200, 7))
resblock.compute\_output\_shape(input\_shape=(None, 200, 7))

View Code
继承tf.Module栗子：

class ImdbModel(models.Model):
 def \_\_init\_\_(self):
 super(ImdbModel, self).\_\_init\_\_()

 def build(self, input\_shape):
 self.embedding = layers.Embedding(MAX\_WORDS, 7)
 self.block1 = ResBlock(7)
 self.block2 = ResBlock(5)
 self.dense = layers.Dense(1, activation="sigmoid")
 super(ImdbModel, self).build(input\_shape)

 def call(self, x):
 x = self.embedding(x)
 x = self.block1(x)
 x = self.block2(x)
 x = layers.Flatten()(x)
 x = self.dense(x)
 return (x)

model = ImdbModel()
model.build(input\_shape=(None, 200))
model.compile(optimizer='Nadam', loss='binary\_crossentropy', metrics=['accuracy', "AUC"])

View Code
Model类具有Layer类相同的API，但有以下区别：

• Model类 提供了内置的训练 model.fit() 、评估 model.evaluate() 和预测 model.predict() API
• Model类 可以通过 model.layers 属性公开内层的列表。
• Model类 提供了保存和序列化 API save()、save_weights()...

　　Layer 类对应于“层”，Model 类对应于“模型**”**，如果您想知道“我应该使用Layer类还是Model类？”，请问自己：我需要调用fit()它吗？我需要save() 吗？如果是这样，请与Model。如果不是，请使用Layer。

把Layer 类和Model 类用在一起，吃个栗子：

class Sampling(layers.Layer):
 """使用(z\_mean, z\_log\_var)对z进行采样，z是对一个数字进行编码的向量"""

    def call(self, inputs):
 z\_mean, z\_log\_var = inputs
 batch = tf.shape(z\_mean)[0]
 dim = tf.shape(z\_mean)[1]
 epsilon = tf.keras.backend.random\_normal(shape=(batch, dim))
 return z\_mean + tf.exp(0.5 * z\_log\_var) * epsilon

class Encoder(layers.Layer):
 """将MNIST数字映射为一个三元组(z\_mean, z\_log\_var, z)"""

    def \_\_init\_\_(self, latent\_dim=32, intermediate\_dim=64, name="encoder", **kwargs):
 super(Encoder, self).\_\_init\_\_(name=name, **kwargs)
 self.dense\_proj = layers.Dense(intermediate\_dim, activation="relu")
 self.dense\_mean = layers.Dense(latent\_dim)
 self.dense\_log\_var = layers.Dense(latent\_dim)
 self.sampling = Sampling()

 def call(self, inputs):
 x = self.dense\_proj(inputs)
 z\_mean = self.dense\_mean(x)
 z\_log\_var = self.dense\_log\_var(x)
 z = self.sampling((z\_mean, z\_log\_var))
 return z\_mean, z\_log\_var, z

class Decoder(layers.Layer):
 """将已编码的数字向量z转换回可读的数字"""

    def \_\_init\_\_(self, original\_dim, intermediate\_dim=64, name="decoder", **kwargs):
 super(Decoder, self).\_\_init\_\_(name=name, **kwargs)
 self.dense\_proj = layers.Dense(intermediate\_dim, activation="relu")
 self.dense\_output = layers.Dense(original\_dim, activation="sigmoid")

 def call(self, inputs):
 x = self.dense\_proj(inputs)
 return self.dense\_output(x)

class VariationalAutoEncoder(keras.Model):
 """将编码器和解码器组合成端到端的训练模型。"""

    def \_\_init\_\_(self, original\_dim, intermediate\_dim=64, latent\_dim=32, name="autoencoder", **kwargs):
 super(VariationalAutoEncoder, self).\_\_init\_\_(name=name, **kwargs)
 self.original\_dim = original\_dim
 self.encoder = Encoder(latent\_dim=latent\_dim, intermediate\_dim=intermediate\_dim)
 self.decoder = Decoder(original\_dim, intermediate\_dim=intermediate\_dim)

 def call(self, inputs):
 z\_mean, z\_log\_var, z = self.encoder(inputs)
 reconstructed = self.decoder(z)
 # Add KL divergence regularization loss.
        kl\_loss = -0.5 * tf.reduce\_mean(z\_log\_var - tf.square(z\_mean) - tf.exp(z\_log\_var) + 1)
 self.add\_loss(kl\_loss)
 return reconstructed

View Code
自定义循环训练模型

original\_dim = 784
vae = VariationalAutoEncoder(original\_dim, 64, 32)

optimizer = tf.keras.optimizers.Adam(learning\_rate=1e-3)
mse\_loss\_fn = tf.keras.losses.MeanSquaredError()

loss\_metric = tf.keras.metrics.Mean()

(x\_train, \_), \_ = tf.keras.datasets.mnist.load\_data()
x\_train = x\_train.reshape(60000, 784).astype("float32") / 255

train\_dataset = tf.data.Dataset.from\_tensor\_slices(x\_train)
train\_dataset = train\_dataset.shuffle(buffer\_size=1024).batch(64)

epochs = 2

# Iterate over epochs.
for epoch in range(epochs):
 print("Start of epoch %d" % (epoch,))

 # Iterate over the batches of the dataset.
    for step, x\_batch\_train in enumerate(train\_dataset):
 with tf.GradientTape() as tape:
 reconstructed = vae(x\_batch\_train)
 # Compute reconstruction loss
            loss = mse\_loss\_fn(x\_batch\_train, reconstructed)
 loss += sum(vae.losses)  # Add KLD regularization loss

 grads = tape.gradient(loss, vae.trainable\_weights)
 optimizer.apply\_gradients(zip(grads, vae.trainable\_weights))

 loss\_metric(loss)

 if step % 100 == 0:
 print("step %d: mean loss = %.4f" % (step, loss\_metric.result()))

View Code
内置循环方法

vae = VariationalAutoEncoder(784, 64, 32)

optimizer = tf.keras.optimizers.Adam(learning\_rate=1e-3)

vae.compile(optimizer, loss=tf.keras.losses.MeanSquaredError())
vae.fit(x\_train, x\_train, epochs=2, batch\_size=64)

View Code
函数式API和自定义Model类混搭训练

original\_dim = 784
intermediate\_dim = 64
latent\_dim = 32

# Define encoder model.
original\_inputs = tf.keras.Input(shape=(original\_dim,), name="encoder\_input")
x = layers.Dense(intermediate\_dim, activation="relu")(original\_inputs)
z\_mean = layers.Dense(latent\_dim, name="z\_mean")(x)
z\_log\_var = layers.Dense(latent\_dim, name="z\_log\_var")(x)
z = Sampling()((z\_mean, z\_log\_var))
encoder = tf.keras.Model(inputs=original\_inputs, outputs=z, name="encoder")

# Define decoder model.
latent\_inputs = tf.keras.Input(shape=(latent\_dim,), name="z\_sampling")
x = layers.Dense(intermediate\_dim, activation="relu")(latent\_inputs)
outputs = layers.Dense(original\_dim, activation="sigmoid")(x)
decoder = tf.keras.Model(inputs=latent\_inputs, outputs=outputs, name="decoder")

# Define VAE model.
outputs = decoder(z)
vae = tf.keras.Model(inputs=original\_inputs, outputs=outputs, name="vae")

# Add KL divergence regularization loss.
kl\_loss = -0.5 * tf.reduce\_mean(z\_log\_var - tf.square(z\_mean) - tf.exp(z\_log\_var) + 1)
vae.add\_loss(kl\_loss)

# Train.
optimizer = tf.keras.optimizers.Adam(learning\_rate=1e-3)
vae.compile(optimizer, loss=tf.keras.losses.MeanSquaredError())
vae.fit(x\_train, x\_train, epochs=3, batch\_size=64)
函数式API训练模型

View Code
补充知识：自定义层

如果自定义模型层没有需要被训练的参数，一般推荐使用Lamda层实现。

mypower = layers.Lambda(lambda x:tf.math.pow(x,2))
mypower(tf.range(5))

如果自定义模型层有需要被训练的参数，则可以通过继承Layer基类实现。Layer的子类化一般需要重新实现初始化方法，Build方法和Call方法。如果built = False，调用__call__时会先调用build方法, 再调用call方法。

损失函数

　　一般来说，监督学习的目标函数由损失函数和正则化项组成（Objective = Loss + Regularization）。对于keras模型，目标函数中的正则化项一般在各层中指定，例如使用Dense的 kernel_regularizer 和 bias_regularizer等参数指定权重使用l1或者l2正则化项，此外还可以用kernel_constraint 和 bias_constraint等参数约束权重的取值范围，这也是一种正则化手段。

损失函数在模型编译时候指定。

• 对于回归模型，通常使用的损失函数是平方损失函数 mean_squared_error，简写为 mse，类实现形式为 MeanSquaredError 和 MSE
• 对于二分类模型，通常使用的是二元交叉熵损失函数 binary_crossentropy。
• 对于多分类模型，如果label是one-hot编码的，则使用交叉熵损失函数 categorical_crossentropy。如果label是序号编码的，则需要使用稀疏类别交叉熵损失函数 sparse_categorical_crossentropy。

• 如果有需要，也可以自定义损失函数，自定义损失函数需要接收两个张量y_true，y_pred作为输入参数，并输出一个标量作为损失函数值。

  常见的Loss可以参看Tensorflow的官网：tf.keras.losses

自定义损失函数

自定义损失函数接收两个张量y_true，y_pred作为输入参数，并输出一个标量作为损失函数值。

def focal\_loss(gamma=2., alpha=.25):
 def focal\_loss\_fixed(y\_true, y\_pred):
 pt\_1 = tf.where(tf.equal(y\_true, 1), y\_pred, tf.ones\_like(y\_pred))
 pt\_0 = tf.where(tf.equal(y\_true, 0), y\_pred, tf.zeros\_like(y\_pred))
 loss = -tf.sum(alpha * tf.pow(1. - pt\_1, gamma) * tf.log(1e-07+pt\_1)) \
 -tf.sum((1-alpha) * tf.pow( pt\_0, gamma) * tf.log(1. - pt\_0 + 1e-07))
 return loss
 return focal\_loss\_fixed

也可以对tf.keras.losses.Loss进行子类化，重写call方法实现损失的计算逻辑，从而得到损失函数的类的实现。

class FocalLoss(losses.Loss):
 def \_\_init\_\_(self,gamma=2.0,alpha=0.25):
 self.gamma = gamma
 self.alpha = alpha

 def call(self,y\_true,y\_pred):
 pt\_1 = tf.where(tf.equal(y\_true, 1), y\_pred, tf.ones\_like(y\_pred))
 pt\_0 = tf.where(tf.equal(y\_true, 0), y\_pred, tf.zeros\_like(y\_pred))
 loss = -tf.sum(self.alpha * tf.pow(1. - pt\_1, self.gamma) * tf.log(1e-07+pt\_1)) \
 -tf.sum((1-self.alpha) * tf.pow( pt\_0, self.gamma) * tf.log(1. - pt\_0 + 1e-07))
 return loss

度量函数

　　人们通常会通过度量函数来从另一个方面 评估模型的好坏，度量函数不要求连续可导，

• 编译模型时，可以通过列表形式指定多个评估指标。
• 也可以自定义评估指标。自定义评估指标需要接收两个张量y_true，y_pred作为输入参数，并输出一个标量作为评估值。
• 也可以继承 tf.keras.metrics.Metric自定义度量方法，update_state方法，result方法实现评估指标的计算逻辑，从而得到评估指标的类的实现形式。

Tensorflow内置的评估指标可以常见：tf.keras.metrics

函数自定义度量

　　如果编写函数形式的评估指标，则只能取epoch中各个batch计算的评估指标结果，这个结果通常会偏离拿整个epoch数据一次计算的结果。

@tf.function
def ks(y\_true,y\_pred):
 y\_true = tf.reshape(y\_true,(-1,))
 y\_pred = tf.reshape(y\_pred,(-1,))
 length = tf.shape(y\_true)[0]
 t = tf.math.top\_k(y\_pred,k = length,sorted = False)
 y\_pred\_sorted = tf.gather(y\_pred,t.indices)
 y\_true\_sorted = tf.gather(y\_true,t.indices)
 cum\_positive\_ratio = tf.truediv(
 tf.cumsum(y\_true\_sorted),tf.reduce\_sum(y\_true\_sorted))
 cum\_negative\_ratio = tf.truediv(
 tf.cumsum(1 - y\_true\_sorted),tf.reduce\_sum(1 - y\_true\_sorted))
 ks\_value = tf.reduce\_max(tf.abs(cum\_positive\_ratio - cum\_negative\_ratio)) 
 return ks\_value

View Code

类自定义度量

由于训练的过程通常是分批次训练的，而评估指标要跑完一个epoch才能够得到整体的指标结果。因此，类形式的评估指标更为常见。即需要编写update_state方法在每个batch后更新相关中间变量的状态，编写result方法输出最终指标结果。

class KS(metrics.Metric):

 def \_\_init\_\_(self, name="ks", **kwargs):
 super(KS, self).\_\_init\_\_(name=name, **kwargs)
 self.true\_positives = self.add\_weight(
 name="tp", shape=(101,), initializer="zeros")
 self.false\_positives = self.add\_weight(
 name="fp", shape=(101,), initializer="zeros")

 @tf.function
 def update\_state(self, y\_true, y\_pred):
 y\_true = tf.cast(tf.reshape(y\_true, (-1,)), tf.bool)
 y\_pred = tf.cast(100 * tf.reshape(y\_pred, (-1,)), tf.int32)

 for i in tf.range(0, tf.shape(y\_true)[0]):
 if y\_true[i]:
 self.true\_positives[y\_pred[i]].assign(
 self.true\_positives[y\_pred[i]] + 1.0)
 else:
 self.false\_positives[y\_pred[i]].assign(
 self.false\_positives[y\_pred[i]] + 1.0)
 return (self.true\_positives, self.false\_positives)

 @tf.function
 def result(self):
 cum\_positive\_ratio = tf.truediv(
 tf.cumsum(self.true\_positives), tf.reduce\_sum(self.true\_positives))
 cum\_negative\_ratio = tf.truediv(
 tf.cumsum(self.false\_positives), tf.reduce\_sum(self.false\_positives))
 ks\_value = tf.reduce\_max(tf.abs(cum\_positive\_ratio - cum\_negative\_ratio))
 return ks\_value

y\_true = ...
y\_pred = ...

myks = KS()
myks.update\_state(y\_true, y\_pred)
tf.print(myks.result())

View Code

优化器

　　机器学习界有一群炼丹师，他们每天的日常是：拿来药材（数据），架起八卦炉（模型），点着六味真火（优化算法），就摇着蒲扇等着丹药出炉了。不过，当过厨子的都知道，同样的食材，同样的菜谱，但火候不一样了，这出来的口味可是千差万别。火小了夹生，火大了易糊，火不匀则半生半糊。机器学习也是一样，模型优化算法的选择直接关系到最终模型的性能。有时候效果不好，未必是特征的问题或者模型设计的问题，很可能就是优化算法的问题。

　　深度学习优化算法大概经历了 SGD -> SGDM -> NAG ->Adagrad -> Adadelta(RMSprop) -> Adam -> Nadam 这样的发展历程。model.compile(optimizer=optimizers.SGD(learning_rate=0.01), loss=loss)

• SGD：默认参数为纯SGD, 设置momentum参数不为0实际上变成SGDM, 考虑了一阶动量, 设置 nesterov为True后变成NAG，即 Nesterov Acceleration Gradient，在计算梯度时计算的是向前走一步所在位置的梯度
• Adagrad：考虑了二阶动量，对于不同的参数有不同的学习率，即自适应学习率。缺点是学习率单调下降，可能后期学习速率过慢乃至提前停止学习
• RMSprop：考虑了二阶动量，对于不同的参数有不同的学习率，即自适应学习率，对Adagrad进行了优化，通过指数平滑只考虑一定窗口内的二阶动量
• Adadelta：考虑了二阶动量，与RMSprop类似，但是更加复杂一些，自适应性更强
• Adam：同时考虑了一阶动量和二阶动量，可以看成RMSprop上进一步考虑了Momentum
• Nadam：在Adam基础上进一步考虑了 Nesterov Acceleration

　　对于一般新手炼丹师，优化器直接使用Adam，并使用其默认参数就OK了。一些爱写论文的炼丹师由于追求评估指标效果，可能会偏爱前期使用Adam优化器快速下降，后期使用SGD并精调优化器参数得到更好的结果。此外目前也有一些前沿的优化算法，据称效果比Adam更好，例如LazyAdam, Look-ahead, RAdam, Ranger等。

　　初始化优化器时会创建一个变量optimier.iterations用于记录迭代的次数。因此优化器和tf.Variable一样，一般在@tf.function外创建。

优化器主要使用apply_gradients方法传入变量和对应梯度从而来对给定变量进行迭代，

optimizer = tf.keras.optimizers.SGD(learning\_rate=0.01)

with tf.GradientTape() as tape:
 ...
grads = tape.gradient(loss, model.trainable\_weights)  # 根据损失 求梯度
optimizer.apply\_gradients(zip(grads, model.trainable\_weights))  # 根据梯度 优化模型

或者直接使用minimize方法对目标函数进行迭代优化。

@tf.function
def train(epoch=1000):
 for \_ in tf.range(epoch):
 optimizer.minimize(loss, model.trainable\_weights)
 tf.print("epoch = ", optimizer.iterations)
 return loss

当然，更常见的使用是在编译时将优化器传入model.fit

model.compile(optimizer=optimizers.SGD(learning\_rate=0.01), loss=loss)

回调函数

　　tf.keras的回调函数实际上是一个类，一般是在model.fit时作为参数指定，一般收集一些日志信息，改变学习率等超参数，提前终止训练过程等等。

大部分时候，keras.callbacks子模块中定义的回调函数类已经足够使用了，如果有特定的需要，我们也可以通过对keras.callbacks.Callbacks实施子类化构造自定义的回调函数。

• BaseLogger： 收集每个epoch上metrics平均值，对stateful_metrics参数中的带中间状态的指标直接拿最终值无需对各个batch平均，指标均值结果将添加到logs变量中。该回调函数被所有模型默认添加，且是第一个被添加的。
• History： 将BaseLogger计算的各个epoch的metrics结果记录到history这个dict变量中，并作为model.fit的返回值。该回调函数被所有模型默认添加，在BaseLogger之后被添加。
• EarlyStopping： 当被监控指标在设定的若干个epoch后没有提升，则提前终止训练。
• TensorBoard： 为Tensorboard可视化保存日志信息。支持评估指标，计算图，模型参数等的可视化。
• ModelCheckpoint： 在每个epoch后保存模型。
• ReduceLROnPlateau：如果监控指标在设定的若干个epoch后没有提升，则以一定的因子减少学习率。
• TerminateOnNaN：如果遇到loss为NaN，提前终止训练。
• LearningRateScheduler：学习率控制器。给定学习率lr和epoch的函数关系，根据该函数关系在每个epoch前调整学习率。
• CSVLogger：将每个epoch后的logs结果记录到CSV文件中。
• ProgbarLogger：将每个epoch后的logs结果打印到标准输出流中。

Tensorboard

　　Tensorboard有助于追踪模型训练过程的Scalars、Graphs、Distributions等等

• Scalars：显示损失和指标在每个时期如何变化。 还可以使用它来跟踪训练速度，学习率和其他标量值。
• Graphs：可帮助您可视化模型。
• Distributions 和 Histograms 显示张量随时间的分布。 可以 可视化权重和偏差并验证它们是否以预期的方式变化

在model.fit中使用

　　当使用Model.fit() 函数进行训练时, 添加 tf.keras.callback.TensorBoard 回调函数可确保创建和存储日志

tensorboard\_callback = tf.keras.callbacks.TensorBoard(log\_dir="./event\_file")
model.fit(... 
 callbacks=[tensorboard\_callback])

在自定义框架中使用

summary\_writer = tf.summary.create\_file\_writer("./event\_file")
with summary\_writer.as\_default():
 tf.summary.scalar('train/loss', train\_loss, step=epoch)
 tf.summary.scalar('train/accuracy', train\_accuracy, step=epoch)
 tf.summary.scalar('val/loss', val\_loss, step=epoch)
 tf.summary.scalar('val/accuracy', val\_accuracy, step=epoch)

启动Tensorboard，在当前文件夹中，cmd运行：

tensorboard --logdir "./"

训练模型

训练模型通常有3种方法，

• 内置fit方法：支持对numpy array，tf.data.Dataset以及 Python generator数据进行训练，并且可以通过设置回调函数实现对训练过程的复杂控制逻辑。
• 内置train_on_batch方法
• 自定义训练循环

model.fit

model.fit(
 x=None, y=None, batch\_size=None, epochs=1, verbose='auto',
 callbacks=None, validation\_split=0.0, validation\_data=None, shuffle=True,
 initial\_epoch=0, steps\_per\_epoch=None)

参数：

• x：输入数据，可以是：
  • numpy数组、数组列表(如果模型有多个输入)
  • Tensorflow张量或张量列表(如果模型有多个输入)
  • 如果模型具有命名输入，则将输入名称映射到相应的数组/张量的字典。
  • tf.data数据集，应该返回一个(inputs, targets)或 的元组(inputs, targets, sample_weights)
  • 生成器或keras.utils.Sequence返回(inputs, targets) 或(inputs, targets, sample_weights)。
• y：目标数据。与输入数据一样x，它可以是 Numpy 数组或 TensorFlow 张量。它应该是一致的x（你不能有 Numpy 输入和张量目标，或者相反）。如果x是数据集、生成器或keras.utils.Sequence实例，y则不应指定（因为目标将从 获取x）。
• batch_size：每次梯度更新的样本数。如果未指定，将默认为 32。如果您的数据是数据集、生成器或实例的形式（因为它们生成批次），请不要指定。
• epochs：训练模型的周期数
• verbose：'auto'、0、1 或 2。详细模式。0 =静默，1 = 进度条，2 = 每个 epoch 一行。'auto' 在大多数情况下默认为 1
• callbacks：训练期间调用的回调列表。见tf.keras.callbacks。
• validation_split：在 0 和 1 之间浮动。从训练数据集中分离一部分数据用于验证，并将在每个 epoch 结束时评估该数据的损失和指标
• validation_data：验证数据集
• shuffle：布尔值（是否在每个 epoch 之前对训练数据进行洗牌）
• initial_epoch：整数。开始训练的epoch（对于恢复之前的训练运行很有用）。

返回：history，调用 history.history 可以查看训练期间损失值和度量值的记录

train_on_batch

　　该内置方法相比较fit方法更加灵活，可以不通过回调函数而直接在batch层次上更加精细地控制训练的过程。

for epoch in tf.range(1, epoches + 1):
 for x, y in ds\_train:
 train\_result = model.train\_on\_batch(x, y)

 for x, y in ds\_valid:
 valid\_result = model.test\_on\_batch(x, y, reset\_metrics=False)

自定义训练循环

自定义训练循环无需编译模型，直接利用优化器根据损失函数反向传播迭代参数，拥有最高的灵活性。

训练循环包括按顺序重复执行三个任务：

• 给模型输入batch数据以生成输出
• 通过将输出与标签进行比较来计算损失
• 使用GradientTape计算梯度
• 使用这些梯度优化变量

optimizer = keras.optimizers.SGD(learning\_rate=1e-3)  # 实例化一个优化器
loss\_fn = keras.losses.BinaryCrossentropy()  # 实例化损失函数

train\_loss = keras.metrics.Mean(name='train\_loss')
valid\_loss = keras.metrics.Mean(name='valid\_loss')

train\_metric = keras.metrics.BinaryAccuracy(name='train\_accuracy')
valid\_metric = keras.metrics.BinaryAccuracy(name='valid\_accuracy')

@tf.function
def train\_step(features, labels):
 with tf.GradientTape() as tape:
 logits = model(features, training=True)
 loss\_value = loss\_fn(labels, logits)
 # loss\_value += sum(model.losses) # 添加额外的损失
    grads = tape.gradient(loss\_value, model.trainable\_weights)
 optimizer.apply\_gradients(zip(grads, model.trainable\_weights))

 train\_loss.update\_state(loss\_value)
 train\_metric.update\_state(labels, logits)

@tf.function
def valid\_step(features, labels):
 val\_logits = model(features, training=False)

 loss\_value = loss\_fn(labels, val\_logits)
 valid\_loss.update\_state(loss\_value)
 valid\_metric.update\_state(labels, val\_logits)

epochs = 2
for epoch in range(epochs):
 start\_time = time.time()
 for step, (x\_batch\_train, y\_batch\_train) in enumerate(train\_dataset):
 loss\_value = train\_step(x\_batch\_train, y\_batch\_train)
 # 在每个epoch结束时运行验证循环
    for x\_batch\_val, y\_batch\_val in val\_dataset:
 valid\_step(x\_batch\_val, y\_batch\_val)

 if epoch % 5 == 0:
 print('Epoch={},Loss:{},Accuracy:{},Valid Loss:{},Valid Accuracy:{}'.format(epoch, train\_loss.result(),
 train\_metric.result(),
 valid\_loss.result(),
 valid\_metric.result()))
 train\_loss.reset\_states()
 valid\_loss.reset\_states()
 train\_metric.reset\_states()
 valid\_metric.reset\_states()

 print("运行时间: %.2fs" % (time.time() - start\_time))

评估模型

通过自定义训练循环训练的模型没有经过编译，无法直接使用model.evaluate(ds_valid)方法

model.evaluate(x = x_test,y = y_test)

推理模型

可以使用以下方法:

model.predict(ds_test)

model(x_test)

model.call(x_test)

model.predict_on_batch(x_test)

推荐优先使用model.predict(ds_test)方法，既可以对Dataset，也可以对Tensor使用。

保存和加载模型

　　可以使用Keras方式保存模型，也可以使用TensorFlow原生方式保存。前者仅仅适合使用Python环境恢复模型，后者则可以跨平台进行模型部署。推荐使用后一种方式进行保存。

Keras保存和加载模型

保存的模型包括：

• 模型架构 / 配置
• 模型权重值
• 模型的编译信息（如果 compile() 被调用）
• 优化器及其状态（如果有）（这使您可以在离开的地方重新开始训练）

# 保存模型
model.save('path/to/location') 

del model  #删除现有模型

# 加载模型
model = models.load\_model('path/to/location')

我们可以使用两种格式将整个模型保存到磁盘：

• TensorFlow SavedModel 格式（推荐）：它保存了模型架构、权重和调用函数的跟踪 Tensorflow 子图。这使 Keras 能够恢复内置层和自定义对象。
• Keras H5 格式（较旧的）：包含模型架构、权重值和compile()信息。它是SavedModel的轻量级替代品。
  • 通过model.add_loss()和model.add_metric()添加的外部损失和度量 不会被保存
  • 自定义对象(如自定义层)的计算图不包含在保存的h5文件中。在加载时，Keras需要访问这些对象的Python类/函数来重构模型。

可以通过以下方式保存 H5 格式：

• 传递save_format='h5'给save()

• 以.h5或.keras结尾的文件名传递给save()

  我们还可以只保存模型结构

json\_str = model.to\_json()  # 保存模型结构
model\_json = models.model\_from\_json(json\_str)  # 恢复模型结构

或者只保存模型权重

# 保存模型权重
model.save\_weights(...)

# 恢复模型结构
model\_json = models.model\_from\_json(json\_str)
model\_json.compile(...)

# 加载权重
model\_json.load\_weights(...)

TensorFlow原生方式保存和加载

保存模型结构与模型参数到文件，该方式保存的模型具有跨平台性便于部署

model.save(..., save\_format="tf")   # 保存模型
model\_loaded = tf.keras.models.load\_model(...)  # 加载模型

也可以仅保存权重

model.save\_weights(...,save\_format = "tf")

汇总

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
import time

# 超参数
lr = 1e-3   # 学习率
batch\_size = 64
epochs = 2

# 数据准备
(x\_train, y\_train), (x\_test, y\_test) = keras.datasets.mnist.load\_data()
x\_train = np.reshape(x\_train, (-1, 784))
x\_test = np.reshape(x\_test, (-1, 784))

# 保留1万个样品用于验证
x\_val = x\_train[-10000:]
y\_val = y\_train[-10000:]
x\_train = x\_train[:-10000]
y\_train = y\_train[:-10000]

# 准备训练数据集
train\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_train, y\_train))
train\_dataset = train\_dataset.shuffle(buffer\_size=1024).batch(batch\_size)

# 准备验证数据集
val\_dataset = tf.data.Dataset.from\_tensor\_slices((x\_val, y\_val))
val\_dataset = val\_dataset.batch(batch\_size)

# 搭建模型
inputs = keras.Input(shape=(784,), name="digits")
x1 = layers.Dense(64, activation="relu")(inputs)
x2 = layers.Dense(64, activation="relu")(x1)
outputs = layers.Dense(10, name="predictions")(x2)
model = keras.Model(inputs=inputs, outputs=outputs)

optimizer = keras.optimizers.SGD(learning\_rate=lr)    # 实例化一个优化器
loss\_fn = keras.losses.SparseCategoricalCrossentropy(from\_logits=True)  # 实例化损失函数

train\_metric = keras.metrics.BinaryAccuracy(name='train\_accuracy')
valid\_metric = keras.metrics.BinaryAccuracy(name='valid\_accuracy')

@tf.function
def train\_step(x, y):
 with tf.GradientTape() as tape:
 logits = model(x, training=True)
 loss\_value = loss\_fn(y, logits)
 # loss\_value += sum(model.losses) # 添加额外的损失
    grads = tape.gradient(loss\_value, model.trainable\_weights)
 optimizer.apply\_gradients(zip(grads, model.trainable\_weights))
 train\_metric.update\_state(y, logits)
 return loss\_value

@tf.function
def test\_step(x, y):
 val\_logits = model(x, training=False)
 valid\_metric.update\_state(y, val\_logits)

for epoch in range(epochs):
 start\_time = time.time()
 for step, (x\_batch\_train, y\_batch\_train) in enumerate(train\_dataset):
 loss\_value = train\_step(x\_batch\_train, y\_batch\_train)

 if step % 200 == 0:
 print("训练损失( %d: %.4f" % (step, float(loss\_value)))

 train\_acc = train\_metric.result()
 print("训练精度: %.4f" % (float(train\_acc),))
 train\_metric.reset\_states() # 在每个epoch结束时重置训练指标

    # 在每个epoch结束时运行验证循环
    for x\_batch\_val, y\_batch\_val in val\_dataset:
 test\_step(x\_batch\_val, y\_batch\_val)

 val\_acc = valid\_metric.result()
 valid\_metric.reset\_states()
 print("验证精度: %.4f" % (float(val\_acc),))
 print("运行时间: %.2fs" % (time.time() - start\_time))

View Code

GPU训练

指定GPU训练

深度学习的训练过程常常非常耗时，一个模型训练几个小时是家常便饭，训练几天也是常有的事情，有时候甚至要训练几十天。

训练过程的耗时主要来自于两个部分，一部分来自数据准备，另一部分来自参数更新。

　　数据准备过程可以使用更多进程处理数据来缩减时间。参数更新时间可以应用GPU或者Google的TPU来进行加速。

　　当存在可用的GPU时，如果不特意指定device，tensorflow会自动优先选择使用GPU来创建张量和执行张量计算。但如果是在公司或者学校实验室的服务器环境，存在多个GPU和多个使用者时，为了不让单个同学的任务占用全部GPU资源导致其他同学无法使用（tensorflow默认获取全部GPU的全部内存资源权限，但实际上只使用一个GPU的部分资源），我们通常会在开头增加以下几行代码以控制每个任务使用的GPU编号和显存大小，以便其他同学也能够同时训练模型。

gpus = tf.config.list\_physical\_devices("GPU")
tf.print(gpus)
# [PhysicalDevice(name='/physical\_device:GPU:0', device\_type='GPU'),
# PhysicalDevice(name='/physical\_device:GPU:1', device\_type='GPU'),
# PhysicalDevice(name='/physical\_device:GPU:2', device\_type='GPU'),
# PhysicalDevice(name='/physical\_device:GPU:3', device\_type='GPU')]

if gpus:
 gpu0 = gpus[0]  # 如果有多个GPU，仅使用第0个GPU
    tf.config.experimental.set\_memory\_growth(gpu0, True)  # 设置GPU显存用量按需使用
    # 或者也可以设置GPU显存为固定使用量(例如：4G)
    # tf.config.experimental.set\_virtual\_device\_configuration(gpu0,
    # [tf.config.experimental.VirtualDeviceConfiguration(memory\_limit=4096)])
    tf.config.set\_visible\_devices([gpu0], "GPU")

单机多卡训练

　　TensorFlow 在 tf.distribute.MirroredStrategy 中为我们提供了单机多卡训练策略，使用这种策略时，我们只需实例化一个 MirroredStrategy 策略，并将模型构建的代码放入 strategy.scope() 的上下文环境中：

strategy = tf.distribute.MirroredStrategy()

with strategy.scope():
 # 模型构建代码

可以在参数中指定设备，如：

# 指定只使用第 0、1 号 GPU 参与分布式策略。
strategy = tf.distribute.MirroredStrategy(devices=["/gpu:0", "/gpu:1"])

MirroredStrategy 的步骤如下：

• 训练开始前，该策略在所有 N 个计算设备上均各复制一份完整的模型；
• 每次训练传入一个batch的数据时，将数据分成 N 份，分别传入 N 个计算设备（即数据并行）；
• N 个计算设备使用本地变量（镜像变量）分别计算自己所获得的部分数据的梯度；
• 使用分布式计算的 All-reduce 操作，在计算设备间高效交换梯度数据并进行求和，使得最终每个设备都有了所有设备的梯度之和；
• 使用梯度求和的结果更新本地变量（镜像变量）；
• 当所有设备均更新本地变量后，进行下一轮训练（即该并行策略是同步的）。

默认情况下，TensorFlow 中的 MirroredStrategy 策略使用 NVIDIA NCCL 进行 All-reduce 操作。

tf.distribute.Strategy和model.fit

tf.distribute.Strategy被集成到tf.keras，tf.keras是用于构建和训练模型的高级 API。您可以使用 Model.fit 来无缝分布式训练模型

TensorFlow 分布策略支持所有类型的 Keras 模型——Sequential、Functional和subclassed。以下是您需要在代码中更改的内容：

• 创建实例 tf.distribute.Strategy。
• 在strategy.scope中创建 Keras 模型、优化器和度量

num\_epochs = 5
batch\_size\_per\_replica = 64 # 每个显卡上的batch数
learning\_rate = 0.001

strategy = tf.distribute.MirroredStrategy()
print('Number of devices: %d' % strategy.num\_replicas\_in\_sync)  # 输出设备数量
batch\_size = batch\_size\_per\_replica * strategy.num\_replicas\_in\_sync     # 总batch\_size

dataset = ...

with strategy.scope():
 model = tf.keras.applications.MobileNetV2(weights=None, classes=2)
 model.compile(
 optimizer=tf.keras.optimizers.Adam(learning\_rate=learning\_rate),
 loss=tf.keras.losses.sparse\_categorical\_crossentropy,
 metrics=[tf.keras.metrics.sparse\_categorical\_accuracy]
 )

model.fit(dataset, epochs=num\_epochs)

分布式训练汇总

import os
import tensorflow as tf
import tensorflow\_datasets as tfds

strategy = tf.distribute.MirroredStrategy()
print('设备数量: {}'.format(strategy.num\_replicas\_in\_sync))

epochs = 12
batch\_size\_per\_replica = 64
batch\_size = batch\_size\_per\_replica * strategy.num\_replicas\_in\_sync

# 数据集
def scale(image, label):
 image = tf.cast(image, tf.float32)
 image /= 255

    return image, label

datasets, info = tfds.load(name='mnist', with\_info=True, as\_supervised=True)
mnist\_train, mnist\_test = datasets['train'], datasets['test']
train\_dataset = mnist\_train.map(scale).cache().shuffle(10000).batch(batch\_size)
eval\_dataset = mnist\_test.map(scale).batch(batch\_size)

with strategy.scope():
 model = tf.keras.Sequential([
 tf.keras.layers.Conv2D(32, 3, activation='relu', input\_shape=(28, 28, 1)),
 tf.keras.layers.MaxPooling2D(),
 tf.keras.layers.Flatten(),
 tf.keras.layers.Dense(64, activation='relu'),
 tf.keras.layers.Dense(10)
 ])

 model.compile(loss=tf.keras.losses.SparseCategoricalCrossentropy(from\_logits=True),
 optimizer=tf.keras.optimizers.Adam(),
 metrics=['accuracy'])

checkpoint\_dir = './training\_checkpoints'  # 定义用于存储检查点的检查点目录
checkpoint\_prefix = os.path.join(checkpoint\_dir, "ckpt\_{epoch}")  # 定义检查点文件的名称

# 定义一个函数来衰减学习速率。
def decay(epoch):
 if epoch < 3:
 return 1e-3
    elif epoch >= 3 and epoch < 7:
 return 1e-4
    else:
 return 1e-5

# 定义一个回调函数，用于在每个epoch的末尾打印学习速率
class PrintLR(tf.keras.callbacks.Callback):
 def on\_epoch\_end(self, epoch, logs=None):
 print('\nepoch的 {} 学习率是 {}'.format(epoch + 1, model.optimizer.lr.numpy()))

# 把所有的回调放在一起
callbacks = [tf.keras.callbacks.TensorBoard(log\_dir='./logs'),
 tf.keras.callbacks.ModelCheckpoint(filepath=checkpoint\_prefix,
 save\_weights\_only=True),
 tf.keras.callbacks.LearningRateScheduler(decay),
 PrintLR()]

# 训练和评估
model.fit(train\_dataset, epochs=epochs, callbacks=callbacks)

View Code

保存为SavedModel格式模型，您可以使用或不使用Strategy.scope.

path = 'saved\_model/'
model.save(path, save\_format='tf')

# 加载模型，不Strategy.scope
unreplicated\_model = tf.keras.models.load\_model(path)
unreplicated\_model.compile(
 loss=tf.keras.losses.SparseCategoricalCrossentropy(from\_logits=True),
 optimizer=tf.keras.optimizers.Adam(),
 metrics=['accuracy'])

eval\_loss, eval\_acc = unreplicated\_model.evaluate(eval\_dataset)
print('Eval loss: {}, Eval Accuracy: {}'.format(eval\_loss, eval\_acc))

# 加载模型Strategy.scope
with strategy.scope():
 replicated\_model = tf.keras.models.load\_model(path)
 replicated\_model.compile(loss=tf.keras.losses.SparseCategoricalCrossentropy(from\_logits=True),
 optimizer=tf.keras.optimizers.Adam(),
 metrics=['accuracy'])

 eval\_loss, eval\_acc = replicated\_model.evaluate(eval\_dataset)
 print('Eval loss: {}, Eval Accuracy: {}'.format(eval\_loss, eval\_acc))

View Code

tf.distribute.Strategy和自定义训练循环

1、在mirrored_strategy.scope() 内创建模型和优化器

with mirrored\_strategy.scope():
 model = tf.keras.Sequential([tf.keras.layers.Dense(1, input\_shape=(1,))])
 optimizer = tf.keras.optimizers.SGD()

2、调用 tf.distribute.Strategy.experimental_distribute_dataset 创建分布式数据集

dataset = tf.data.Dataset.from\_tensors(([1.], [1.])).repeat(100).batch(global\_batch\_size)
dist\_dataset = mirrored\_strategy.experimental\_distribute\_dataset(dataset)

3、使用 tf.nn.compute_average_loss 计算损失。tf.nn.compute_average_loss对每个样本损失求和并将总和除以global_batch_size。

def compute\_loss(labels, predictions):
 per\_example\_loss = loss\_object(labels, predictions)
 return tf.nn.**compute\_average\_loss**(per\_example\_loss, global\_batch\_size=global\_batch\_size)

4、将train_step放入 tf.distribute.Strategy.run 中，并传入之前创建的数据集

5、使用 tf.distribute.Strategy.reduce 来聚合 tf.distribute.Strategy.run。tf.distribute.Strategy.run返回每个GPU结果。您还可以tf.distribute.Strategy.experimental_local_results获取结果值列表。

def train\_step(inputs):
 features, labels = inputs

 with tf.GradientTape() as tape:
 predictions = model(features, training=True)
 loss = compute\_loss(labels, predictions)

 gradients = tape.gradient(loss, model.trainable\_variables)
 optimizer.apply\_gradients(zip(gradients, model.trainable\_variables))
 return loss

@tf.function
def distributed\_train\_step(dist\_inputs):
 per\_replica\_losses = **mirrored\_strategy.run**(train\_step, args=(dist\_inputs,))
 return mirrored\_strategy.**reduce**(tf.distribute.**ReduceOp.SUM**, per\_replica\_losses, axis=None)

6、迭代dist_dataset并循环运行训练：

for dist\_inputs in dist\_dataset:
 print(distributed\_train\_step(dist\_inputs))

分布式训练汇总

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
import os

strategy = tf.distribute.MirroredStrategy()     # 实例化MirroredStrategy
print('设备数量: {}'.format(strategy.num\_replicas\_in\_sync))

epochs = 10
batch\_size\_per\_replica = 64  # 每个GPU上得batch数
global\_batch\_size = batch\_size\_per\_replica * strategy.num\_replicas\_in\_sync  # 总batch数

# 数据集
fashion\_mnist = tf.keras.datasets.fashion\_mnist
(train\_images, train\_labels), (test\_images, test\_labels) = fashion\_mnist.load\_data()
train\_images = train\_images[..., None]
test\_images = test\_images[..., None]

# 获取[0,1]范围内的图像
train\_images = train\_images / np.float32(255)
test\_images = test\_images / np.float32(255)

buffer\_size = len(train\_images)

train\_dataset = tf.data.Dataset.from\_tensor\_slices((train\_images, train\_labels)).shuffle(buffer\_size).batch(
 global\_batch\_size)
test\_dataset = tf.data.Dataset.from\_tensor\_slices((test\_images, test\_labels)).batch(global\_batch\_size)

train\_dist\_dataset = strategy.experimental\_distribute\_dataset(train\_dataset)    # 分布式数据集
test\_dist\_dataset = strategy.experimental\_distribute\_dataset(test\_dataset)      # 分布式数据集

# 创建模型
def create\_model():
 model = tf.keras.Sequential([
 tf.keras.layers.Conv2D(32, 3, activation='relu'),
 tf.keras.layers.MaxPooling2D(),
 tf.keras.layers.Conv2D(64, 3, activation='relu'),
 tf.keras.layers.MaxPooling2D(),
 tf.keras.layers.Flatten(),
 tf.keras.layers.Dense(64, activation='relu'),
 tf.keras.layers.Dense(10)
 ])

 return model

# 创建一个检查点目录来存储检查点
checkpoint\_dir = './training\_checkpoints'
checkpoint\_prefix = os.path.join(checkpoint\_dir, "ckpt")

with strategy.scope():
 # 创建训练损失
    # 将reduction设置为“none”，这样我们可以在之后进行reduction，并除以全局batch size
    loss\_object = tf.keras.losses.SparseCategoricalCrossentropy(
 from\_logits=True,
 reduction=tf.keras.losses.Reduction.NONE)

 def compute\_loss(labels, predictions):
 per\_example\_loss = loss\_object(labels, predictions)
 return tf.nn.compute\_average\_loss(per\_example\_loss, global\_batch\_size=global\_batch\_size)

 test\_loss = tf.keras.metrics.Mean(name='test\_loss')     # 创建测试损失
    train\_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='train\_accuracy')  # 训练精度
    test\_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='test\_accuracy')  # 测试精度

    # 模型、优化器和检查点必须创建在 'strategy.scope' 下。
    model = create\_model()
 optimizer = tf.keras.optimizers.Adam()
 checkpoint = tf.train.Checkpoint(optimizer=optimizer, model=model)

# 训练step
def train\_step(inputs):
 images, labels = inputs

 with tf.GradientTape() as tape:
 predictions = model(images, training=True)
 loss = compute\_loss(labels, predictions)

 gradients = tape.gradient(loss, model.trainable\_variables)
 optimizer.apply\_gradients(zip(gradients, model.trainable\_variables))

 train\_accuracy.update\_state(labels, predictions)
 return loss

# 测试step
def test\_step(inputs):
 images, labels = inputs

 predictions = model(images, training=False)
 t\_loss = loss\_object(labels, predictions)

 test\_loss.update\_state(t\_loss)
 test\_accuracy.update\_state(labels, predictions)

# 分布式训练
@tf.function
def distributed\_train\_step(dataset\_inputs):
 per\_replica\_losses = strategy.run(train\_step, args=(dataset\_inputs,))
 return strategy.reduce(tf.distribute.ReduceOp.SUM, per\_replica\_losses, axis=None)

# 分布式测试
@tf.function
def distributed\_test\_step(dataset\_inputs):
 return strategy.run(test\_step, args=(dataset\_inputs,))

for epoch in range(epochs):
 # 训练循环
    total\_loss = 0.0
 num\_batches = 0
 for x in train\_dist\_dataset:
 total\_loss += distributed\_train\_step(x)
 num\_batches += 1
 train\_loss = total\_loss / num\_batches

 # 测试循环
    for x in test\_dist\_dataset:
 distributed\_test\_step(x)

 if epoch % 10 == 0:
 checkpoint.save(checkpoint\_prefix)

 print("Epoch {}, Loss: {}, Accuracy: {}, Test Loss: {}, Test Accuracy: {}".format(epoch + 1, train\_loss,
 train\_accuracy.result() * 100,
 test\_loss.result(),
 test\_accuracy.result() * 100))

 test\_loss.reset\_states()
 train\_accuracy.reset\_states()
 test\_accuracy.reset\_states()

View Code
tf.distribute.Strategy 可以再没有strategy的情况下恢复最新的检查点并测试

test\_dataset = tf.data.Dataset.from\_tensor\_slices((test\_images, test\_labels)).batch(global\_batch\_size)
eval\_accuracy = tf.keras.metrics.SparseCategoricalAccuracy(name='eval\_accuracy')

new\_model = create\_model()
new\_optimizer = tf.keras.optimizers.Adam()

@tf.function
def eval\_step(images, labels):
 predictions = new\_model(images, training=False)
 eval\_accuracy(labels, predictions)

checkpoint = tf.train.Checkpoint(optimizer=new\_optimizer, model=new\_model)
checkpoint.restore(tf.train.latest\_checkpoint(checkpoint\_dir))

for images, labels in test\_dataset:
 eval\_step(images, labels)

print('在没有strategy的情况下恢复保存的模型后的准确性: {}'.format(eval\_accuracy.result() * 100))

View Code

模型部署

tensorflow-serving

TensorFlow Lite

　　TensorFlow Lite 是 TensorFlow 在移动和 IoT 等边缘设备端的解决方案，提供了 Java、Python 和 C++ API 库，可以运行在 Android、iOS 和 Raspberry Pi 等设备上。AI技术在边缘设备上的应用，TFLite 将会是愈发重要的角色。

　　目前 TFLite 只提供了推理功能，在服务器端进行训练后，经过如下简单处理即可部署到边缘设备上。

• 模型转换：由于边缘设备计算等资源有限，使用 TensorFlow 训练好的模型，模型太大、运行效率比较低，不能直接在移动端部署，需要通过相应工具进行转换成适合边缘设备的格式。
• 边缘设备部署：本节以 android 为例，简单介绍如何在 android 应用中部署转化后的模型，完成 Mnist 图片的识别。

参考

【电子书】简单粗暴 TensorFlow 2

【知乎】最全Tensorflow2.0 入门教程持续更新

【和鲸社区】30天吃掉那只TensorFlow2.0 | Github

【书籍】TensorFlow 2深度学习开源书 | PDF下载 提取码：juqs

【bilibili】tensorflow2.0入门与实战 2019年最通俗易懂的课程

【bilibili】神经网络与深度学习——TensorFlow2.0实战【中文课程】

【github】TensorFlow-Examples

【github】TensorFlow-2.x-Tutorials

作者：凌逆战欢迎任何形式的转载，但请务必注明出处。限于本人水平，如果文章和代码有表述不当之处，还请不吝赐教。本文章不做任何商业用途，仅作为自学所用，文章后面会有参考链接，我可能会复制原作者的话，如果介意，我会修改或者删除。

转载请注明：xuhss » Tensorflow 2.x入门教程