【Tensorflow】使用Tensorflow自定义模型和训练

• Tensorflow的核心与NumPy非常相似，但具有GPU支持；
• Tensorflow支持分布式计算（跨多个设备和服务器）。
  

像NumPy一样使用TensorFlow

• @运算符是在Python 3.5 中添加的，用于矩阵乘法，等效于 tf.matmul() 函数。

Keras的底层API

• Keras API在keras.backend中有自己的底层API，如果要编写可移植到其他Keras实现中的代码，则应使用这些Keras函数。

from tensorflow import keras
K = keras.backend
K.square(K.transpose(t)) + 10
1
2
3

• 当从NumPy数组创建张量时，需设置 dtype=tf.float32；

定制模型和训练算法

自定义损失函数

• 实现Huber损失：

def huber_fn(y_true, y_pred):
	error = y_true - y_pred
	is_small_error = tf.abs(error) < 1
	squared_loss = tf.square(error) / 2
	linear_loss = tf.abs(error) - 0.5
	return tf.where(is_small_error, squared_loss, linear_loss)
1
2
3
4
5
6

• 编译Keras模型，训练模型：

model.compile(loss=huber_fn, optimizer="nadam")
model.fit(X_train, y_train, [...])
1
2

保存和加载包含自定义组件的模型

• 当加载包含自定义对象的模型时，需要将名称映射到对象。

model = keras.models.load_model("my_model_with_a_custom_loss.h5",custom_objects={"huber_fn": huber_fn})
1

• 创建一个函数，该函数创建已配置的损失函数：

def create_huber(threshold=1.0):
	def huber_fn(y_true, y_pred):
		error = y_true - y_pred
		is_small_error = tf.abs(error) < threshold
		squared_loss = tf.square(error) / 2
		linear_loss = threshold * tf.abs(error) - threshold**2 / 2
		return tf.where(is_small_error, squared_loss, linear_loss)
	return huber_fn
	
model.compile(loss=create_huber(2.0), optimizer="nadam")
1
2
3
4
5
6
7
8
9
10

• 在加载模型的时候必须指定阈值：

model = keras.models.load_model("my_model_with_a_custom_loss.h5",custom_objects={"huber_fn": create_huber(2.0)})
1

• 通过创建 keras.losses.Loss 类的子类，然后实现其 get_config() 方法来解决此问题：

class HuberLoss(keras.losses.Loss):
	def __init__(self, threshold=1.0, **kwargs):
		self.threshold = threshold
		super().__init__(**kwargs)
	def call(self, y_true, y_pred):
		error = y_true - y_pred
		is_small_error = tf.abs(error) < self.threshold
		squared_loss = tf.square(error) / 2
		linear_loss = self.threshold * tf.abs(error) - self.threshold**2 / 2
		return tf.where(is_small_error, squared_loss, linear_loss)
	def get_config(self):
		base_config = super().get_config()
		return {**base_config, "threshold": self.threshold}
1
2
3
4
5
6
7
8
9
10
11
12
13

• 以上父类构造函数处理标准超参数：损失的name和用于聚合单个实例损失的reduction算法。
• get_config() 方法返回一个字典，将每个超参数映射到其值。首先调用父类的get_config() 方法，然后将新的超参数添加到此字典中。
• 可在编译模型时使用此类的任何实例：

model.compile(loss=HuberLoss(2.),optimizer="nadam")
1

• 当保存模型的时候，阈值会同时一起保存。在加载模型时，只需要将类名映射到类本身：

model = keras.models.load_model("my_model_with_a_custom_loss.h5",custom_objects={"HuberLoss": HuberLoss})
1

• 保存模型时，Keras会调用损失实例的 get_config() 方法，并将配置以 JSON 格式保存到 HDF5 文件中。

自定义激活函数、初始化、正则化和约束

• 编写简单的函数进行自定义

def my_softplus(z):
	return tf.math.log(tf.exp(z)+1.0)

def my_glorot_initializer(shape, dtype=tf.float32):
	stddev = tf.sqrt(2. / (shape[0] + shape[1]))
	return tf.random.normal(shape, stddev=stddev, dtype=dtype)

def my_l1_regularizer(weights):
	return tf.reduce_sum(tf.abs(0.01 * weights))

def my_positive_weights(weights): # return value is just tf.nn.relu(weights)
	return tf.where(weights < 0., tf.zeros_like(weights), weights)
1
2
3
4
5
6
7
8
9
10
11
12

• 使用这些自定义函数

layer = keras.layers.Dense(30, activation=my_softplus,
							kernel_initializer=my_glorot_initializer,
							kenel_regularizer=my_l1_regularizer,
							kenel_constraint=my_positive_weights)
1
2
3
4

• 在每个训练步骤中，权重将传递给正则化函数，以计算正则化损失，并将其添加到主要损失中得到用于训练的最终损失。
• 必须为损失、层（包括激活函数）和模型实现 call() 方法，或者为正则化、初始化和约束实现 __call__()方法。

自定义指标

• 可以将创建的损失函数作为指标。

model.compile(loss="mse", optimizer="nadam", metrics=[create_huber(2.0)])
1

• 流式指标（状态指标）是逐批次更新的。

自定义层

• 创建不带任何权重的自定义层：

exponential_layer = keras.layers.Lambda(lambda x: tf.exp(x))
1

• 当要预测的值具有非常不同的标度（例如0.001、10、1000）时，有时会在回归模型的输出层中使用指数层。
• 构建自定义的有状态层（即具有权重的层）：

class MyDense(keras.layers.Layer):
	# 将所有超参数用作参数
	def __init__(self, units, activation=None, **kwargs):
		super().__init__(**kwargs)
		self.units = units
		self.activation = keras.activations.get(activation)
	
	# 创建层的变量
	def build(self, batch_input_shape):
		self.kernel = self.add_weight(
			name="kernel", shape=[batch_input_shape[-1], self.units],
			initializer="glorot_normal")
		self.bias = self.add_weight(
			name="bias", shape=[self.units], initializer="zeros")
		super().build(batch_input_shape) # must be at the end
		# 调用父类方法，告诉keras这一层被构建了，设置 self.built=true
	
	def call(self, X):
	 	return self.activation(X @ self.kernel + self.bias)

	def compute_output_shape(self, batch_input_shape):
		return tf.TensorShape(batch_input_shape.as_list()[:-1] + [self.units])
	
	def get_config(self):
		base_config = super().get_config()
		return {**base_config, "units": self.units,
				"activation": keras.activation.serialize(self.activation) }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

• 创建多输入，多输出的层：

class MyMultiLayer(keras.layers.Layer):
	def call(self, X):
		X1, X2 = X
		return [X1 + X2, X1 * X2, X1 / X2]

	def compute_output_shape(self, batch_input_shape):
		b1, b2 = batch_input_shape
		return [b1, b1, b1]
1
2
3
4
5
6
7
8

• 如果层在训练期间和测试期间需要具有不同的行为，比如，创建一个在训练期间（用于正则化）添加高斯噪声，但在测试期间不执行任何操作：

class MyGaussianNoise(keras.layers.Layer):
	def __init__(self, stddev, **kwargs):
		super().__init__(**kwargs)
		self.stddev = stddev
	
	def call(self, X, training=None):
		if training:
			noise = tf.random.normal(tf.shape(X), stddev=self.stddev)
			return X + noise
		else:
			return X
	
	def compute_output_shape(self, batch_input_shape):
		return batch_input_shape
1
2
3
4
5
6
7
8
9
10
11
12
13
14

自定义模型

• 首先创建一个 ResidualBlock 层：

class ResidualBlock(keras.layers.Layer):
	def __init__(self, n_layers, n_nerons, **kwargs):
		super().__init__(**kwargs)
		self.hidden = [keras.layers.Dense(n_nerons, activation="elu",
										  kenel_initializer="he_normal")
					   for _ in range(n_layers)]
	
	def call(self, inputs):
		Z = inputs
		for layer in self.hidden:
			Z = layer(Z)
		return inputs + Z
1
2
3
4
5
6
7
8
9
10
11
12

• 使用子类API定义模型：

class ResidualRegressor(keras.Model):
	def __init__(self, output_dim, **kwargs):
		super().__init__(**kwargs)
		self.hidden1 = keras.layers.Dense(30, activation="elu",
										  kernel_initializer="he_normal")
		self.block1 = ResidualBlock(2, 30)
		self.block2 = ResidualBlock(2, 30)
		self.out = keras.layers.Dense(output_dim)
	
	def call(self, inputs):
		Z = self.hidden1(inputs)
		for _ in range(1 + 3):
			Z = self.block1(Z)
		Z = self.block2(Z)
		return self.out(Z)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

• 在构造函数中创建层，在call()方法中使用它们。
• 带有自定义重建损失的自定义模型，此自定义模型在上部隐藏层的顶部有辅助输出，与该辅助输出相关的损失称为重建损失。

class ReconstructingRegressor(keras.Model):
	def __init__(self, output_dim, **kwargs):
		super().__init__(**kwargs)
		self.hidden = [keras.layers.Dense(30, activation="selu",
										  kernel_initializer = "lecun_normal")
					   for _ in range(5)]
		self.out = keras.layers.Dense(output_dim)
	
	def build(self, batch_input_shape):
		n_inputs = batch_input_shape[-1]
		self.reconstruct = keras.layers.Dense(n_inputs) # 该层用于重建模型的输入
		super().build(batch_input_shape)
	
	def call(self, inputs):
		Z = inputs
		for layer in self.hidden:
			Z = layer(Z)
		reconstruction = self.reconstruct(Z)
		recon_loss = tf.reduce_mean(tf.square(reconstruction - inputs))
		self.add_loss(0.05 * recon_loss)
		return self.out(Z)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

• 正则化，在输入变化不大时惩罚那些变化很大的激活。