BLOG

个人博客,记录学习与生活

TensorFlow使用——模型操作

Published June 20, 2020, 10:43 p.m. by kkk

虽然主要靠使用tf.keras,但掌握一些其它知识可以极大方便模型调整,增大自由度。

TensorFlow是一个强大的科学计算库,并针对大规模机器学习而进行了诸多优化。其核心部分同样依赖于Numpy且有非常相似的操作,但是增加了自动微分、各类优化器、GPU支持、分布式计算、JIT编译器(通过计算图),另外计算图可以打包起来在其它平台使用(如通过Java在Android设备上)

TensorFlow最重要的部分当属tf.keras,另外也有用于数据加载和预处理操作的tf.datatf.io,负责图像处理操作的tf.image,信号处理的tf.signal...

底层上,大多TensorFlow操作都有高效的C++代码进行实现,许多操作还有不同的内核实现(kernels)

数据类型

常量(tf.constant)和变量(tf.Variable)

值得注意的一点是,TensorFlow中,为了性能考虑,不会自动进行数据类型转换,只能靠手动数据转换tf.cast,不然会引发异常InvalidArgumentError

自动微分

通常采用两点函数值除以横坐标值差距这种方式来进行近似(通常为$10^6$),TensorFlow具有自动微分功能

def f(w1, w2):
    return 3 * w1 ** 2 + 2 * w2 * w1

w1, w2 = tf.Variable(5.), tf.Variable(3.)
with tf.GradientTape() as tape: # 记录每次涉及到 变量 的操作
    z = f(w1, w2)
    # 可以使用with tape.stop_recording()来暂停tf.GradientTape()内部的记录操作
gradients = tape.gradient(z, [w1, w2])
# 当调用完gradient()后,tape会被自动擦除,再次调用会报错

对于非变量数据,默认不会进行跟踪记录,调用gradient()后会返回None的结果,如果需要在tf.GradientTape()内进行跟踪,可以进行设置(tape.watch()

c1, c2 = tf.constant(5.), tf.constant(3.)
with tf.GradientTape() as tape:
    tape.watch(c1)
    tape.watch(c2)
    z = f(c1, c2)
gradients = tape.gradient(z, [z1, z2]) # 不加tape.watch()的话,返回的结果是[None, None]

对于矢量,如果想计算单独的梯度,要使用tape.jabobian(),另外还可以计算二阶偏导数(Hessian矩阵)

如果要阻止某部分的梯度的反向传播,可以使用tf.stop_gradient(),如下

def f(w1, w2):
    return 3 * w1 ** 2 + tf.stop_gradient(2 * w1 * w2)
    # 只会在前向传播时计算梯度,反向传播时不会计算后面的部分
with tf . GradientTape() as tape:
    z = f(w1, w2) # same result as without stop_gradient()

gradients = tape.gradient(z, [w1, w2])

计算精度问题(自定义求导)

由于浮点数精度错误原因,自动微分可能会计算出NaN的结果,可以自定义链式求导的部分求导(不通过自动微分),然后通过链式求导乘起来

@tf.custom_gradient
def my_better_softplus(z):
    exp = tf.exp(z)
    def my_softplus_gradients(grad):
        return grad / (1 + 1 / exp)
    return tf.math.log(exp + 1), my_softplus_gradients

TensorFlow函数

可以将普通函数变为TensorFlow函数,通过tf.function,有两种方式,例如对于某函数alpha(),有两种方式:

tf_alpha = tf.function(alpha)

@tf.function
def alpha()
    pass

对于自定义损失函数、评估矩阵、网络层...中的函数,Keras会自动将其转换为TensorFlow函数

TensorFlow图

TensorFlow具有自动生成图功能,首先(AutoGraph),分析Python函数源代码,捕获所有控制流语句;然后(Tracing)调用upgraded函数,传递一个符号张量(不具有任何实际值得张量,只具有名字、数据类型和形状)而不是参数,这个函数将运行在图模式graph mode(与之相对的是常规模式,也叫eager modeeager execution),即每个TensorFlow操作将在图形中添加一个节点来表示自身及其输出张量。

可以通过tf.autograph.to_code(sum_squares.python_function)查看生成函数的源代码,有时候可能对于调式有帮助。


模型自定义

自定义损失函数

对于自定义模型的保存(比如模型的损失函数是自己定义的),重新导入时,要传递一个字典对象custom_object,指明模型中自定义函数名和对应的函数声明。

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

def huber_fn:
    pass

但是上述方式不会保存自定义损失函数的超参数,要想保存超参数,需要子类化Keras的Loss类,按照以下的定义方式之后再结合上面的模型保存方式,可以让自定义模型加载后就能具有正确的损失函数和超参数。

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}

当模型保存时,Keras会调用损失函数实例的get_config方法来将配置以JSON, HDF5的形式保存,当加载模型时,Keras会调用HuberLossfrom_fonfigLoss函数已经实现,无需自行实现),然后创建该类(HuberLoss)实例并传递**config到构造器之中。

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

同损失函数的定义,如果没有超参数,直接在加载模型时加上custom_objects这一字典对象即可,不然需要同上所述利用子类化手段处理(keras.regularizers.Regularizer, keras.constraints.Constraint, keras.initializers.Initializer, keras.layers.Layer等),例如正则化:

class MyL1Regularizer(keras.regularizers.Regularizer):
    def __init__(self, factor):
        self.factor = factor

    def __call__(self, weights):
        return tf.reduce_sum(tf.abs(self.factor*weights))

    def get_config(self):
        return {"factor": self.factor}

注:对于损失函数和模型层(包括激活函数),子类化需要使用call函数,而其他的则使用__call__函数

自定义评估矩阵

评估矩阵不同于损失函数,损失函数需要保证可微分、有梯度,而评估矩阵不用,当然损失函数也可以作为评估矩阵用来使用

class HuberMetric(keras.metrics.Metric):
    def __init__(self, threshold=1.0, **kwargs):
        super().__init__(**kwargs)
        self.threshold = threshold
        self.huber_fn = create_huber(threshold)  # 另外定义的一个huber损失函数
        self.total = self.add_weight("total", initializer="zeros") # 创建多次迭代中跟踪矩阵状态的变量,此处为Huber Loss之和
        self.count = self.add_weight("count", initializer="zeros") # 为一直到当前迭代为止, 的实例数

    def update_state(self, y_true, y_pred, sample_weight=None): # 更新变量值
        metric = self.huber_fn(y_true, y_pred)
        self.total.assign_add(tf.reduce_sum(metric))
        self.count.assign_add(tf.cast(tf.size(y_true), tf.float32))

    def result(self): # 返回最终的计算值
        return self.total / self.count

    def get_config(self): # 确保threshold会跟随模型一起保存
        base_config = super().get_config()
        return {**base_config, "threshold": self.threshold}

自定义网络层

满足自定义特殊网络层的需求,首先是没有权重的网络层(如keras.layers.Flattenkeras.layers.ReLU),自定义无权重网络层最简单的是采用keras.layers.Lambda,如下:

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

创建具有权重的网络层,则需要使用子类化keras.layers.Layer,创建简化版Dense层如下:

class MyDense(keras.layers.Layer):
    def __init__(self, units, activation=None, **kwargs):
        super().__init__(**kwargs)
        self.units = units
        self.activation = keras.activations.get(activation) # 可接受函数、标准的激活函数字符串名或者None

    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], initialize="zeros")
        super().build(batch_input_shape)  # 必须放在最后

    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])
        # TensorFlow中shape是TensorShape类型,可以通过as_list()转为Python列表

    def get_config(self):
        base_config = super().get_config()
        return {**base_config, "units":self.units, "activation":keras.activations.serialize(self.activation)}

创建具有多输入的网络层call方法需要接受一个元组数据,包含所有的输入,但是这类网络层只能用于函数式、子类式API,无法用于序列式API(只能处理单输入单输出情况)

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] # 应该可以处理广播规则

要使网络层在训练时和测试时具有不同的表现(例如Dropout、BatchNormalization),需要在call中传递training参数,以下是在网络层训练中加入高斯噪音(基本同keras.layers.GaussianNoise

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

自定义模型

img

对于图示模型(并无实际意义),可以自定义ResidualBlock模块

class ResidualBlock(keras.layers.Layer):
    def __init__(self, n_layers, n_neurons, **kwargs):
        super().__init__(**kwargs)
        self.hidden = [keras.layers.Dense(n_neurons, activation="elu", kernel_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

接下来定义整个网络模块

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.hidden(inputs)
        for _ in range(1 + 3):
            Z = self.block1(Z)
        Z = self.block2(Z)
        return self.out(Z)

如果要使网络能被保存和被keras.models.load_model()加载,就需要向前面一样,定义get_config函数(在ResidualBlockResidualRegressor中都定义)

基于模型内部的损失函数和评估矩阵

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)
        # 5个Dense隐层和一个输出隐层
        # TODO: check https://github.com/tensorflow/tensorflow/issues/26260
        #self.reconstruction_mean = keras.metrics.Mean(name="reconstruction_error")

    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, training=None):
        Z = inputs
        for layer in self.hidden:
            Z = layer(Z)
        reconstruction = self.reconstruct(Z) # reconstruction loss,reconstruction和输入之间的均方差
        recon_loss = tf.reduce_mean(tf.square(reconstruction - inputs))
        self.add_loss(0.05 * recon_loss)
        #if training:
        #    result = self.reconstruction_mean(recon_loss)
        #    self.add_metric(result)
        return self.out(Z)

自定义训练循环

自定义训练循环,即使对模型的fit()进行调整,提高灵活性,以满足某些特殊的需求。

此时,需要自己选取、自定义各类操作(损失函数、优化器、评估矩阵...),然后创建运行时、运行结束后的打印函数,然后手动实现fit()过程。


Share this post
< Pre: 从浅层到深层神经网络(优化策略) Pos: Word2Vec >
434 comments
Similar posts
Add a new comment