Dive into DL TF2 --BatchNormalization

  1 import numpy as np
  2 import tensorflow as tf
  3 
  4 class BatchNormalization(tf.keras.layers.Layer):
  5   def __init__(self, decay=0.9, epsilon=1e-5, **kwargs):
  6     self.decay = decay
  7     self.epsilon = epsilon
  8     super(BatchNormalization, self).__init__(**kwargs)
  9 
 10   def build(self, input_shape):
 11     self.gamma = self.add_weight(name=‘game‘, 
 12                                  shape = [input_shape[-1],],
 13                                  initializer = tf.initializers.ones,
 14                                  trainable = True)
 15     self.beta = self.add_weight(name=‘beta‘, 
 16                                 shape=[input_shape[-1], ],
 17                                 initializer = tf.initializers.zeros,
 18                                 trainable=True)
 19     self.moving_mean = self.add_weight(name=‘moving_mean‘,
 20                                        shape=[input_shape[-1],],
 21                                        initializer=tf.initializers.zeros,
 22                                        trainable=False)
 23     self.moving_variance = self.add_weight(name=‘moving_variance‘,
 24                                            shape=[input_shape[-1],],
 25                                            initializer=tf.initializers.ones,
 26                                            trainable=False)
 27     super(BatchNormalization, self).build(input_shape)
 28   
 29   def assign_moving_average(self, variable, value):
 30     """variable = variable * decay + value * (1 - decay)"""
 31     delta = variable * self.decay + value * (1 - self.decay)
 32     return variable.assign(delta)
 33 
 34   @tf.function
 35   def call(self, inputs, training):
 36     if training:
 37       batch_mean, batch_variance = tf.nn.moments(inputs, list(range(len(inputs.shape) - 1)))
 38       mean_update = self.assign_moving_average(self.moving_mean, batch_mean)
 39       variance_update = self.assign_moving_average(self.moving_variance, batch_variance)
 40       self.add_update(mean_update)
 41       self.add_update(variance_update)
 42       mean, variance = batch_mean, batch_variance
 43     else:
 44       mean, variance = self.moving_mean, self.moving_variance
 45     output = tf.nn.batch_normalization(inputs, 
 46                                        mean=mean, 
 47                                        variance=variance,
 48                                        offset=self.beta,
 49                                        scale=self.gamma,
 50                                        variance_epsilon=self.epsilon)
 51     return output
 52   
 53   def compute_output_shape(self, input_shape):
 54     return input_shape
 55 
 56 
 57 #使用批量归一化层的LeNet
 58 net = tf.keras.models.Sequential(
 59     [tf.keras.layers.Conv2D(filters=6, kernel_size=5),
 60      BatchNormalization(),
 61      tf.keras.layers.Activation(‘sigmoid‘),
 62      tf.keras.layers.MaxPool2D(pool_size=2, strides=2),
 63      tf.keras.layers.Conv2D(filters=16, kernel_size=5),
 64      BatchNormalization(),
 65      tf.keras.layers.Activation(‘sigmoid‘),
 66      tf.keras.layers.MaxPool2D(pool_size=2, strides=2),
 67      tf.keras.layers.Flatten(),
 68      tf.keras.layers.Dense(120),
 69      BatchNormalization(),
 70      tf.keras.layers.Activation(‘sigmoid‘),
 71      tf.keras.layers.Dense(84),
 72      BatchNormalization(),
 73      tf.keras.layers.Activation(‘sigmoid‘),
 74      tf.keras.layers.Dense(10, activation=‘softmax‘)]
 75 )
 76 
 77 
 78 #训练修改后的模型
 79 (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
 80 x_train = x_train.reshape((60000, 28, 28, 1)).astype(‘float32‘)/255
 81 x_test = x_test.reshape((10000, 28, 28, 1)).astype(‘float32‘)/255
 82 
 83 net.compile(loss=‘sparse_categorical_crossentropy‘,
 84             optimizer=tf.keras.optimizers.RMSprop(),
 85             metrics=[‘accuracy‘])
 86 
 87 history = net.fit(x_train, y_train, batch_size=64, epochs=5, validation_split=0.2)
 88 
 89 test_scores = net.evaluate(x_test, y_test, verbose=2)
 90 print(‘Test loss:‘, test_scores[0])
 91 print(‘Test accuracy:‘, test_scores[1])
 92 
 93 
 94 #第一个批量归一化层学习到的拉伸参数gamma和偏移参数beta
 95 net.get_layer(index=1).gamma, net.get_layer(index=1).beta
 96 
 97 
 98 #用keras实现使用批量归一化的LeNet
 99 
100 net = tf.keras.models.Sequential()
101 net.add(tf.keras.layers.Conv2D(filters=6, kernel_size=5))
102 net.add(tf.keras.layers.BatchNormalization())
103 net.add(tf.keras.layers.Activation(‘sigmoid‘))
104 net.add(tf.keras.layers.MaxPool2D(pool_size=2, strides=2))
105 net.add(tf.keras.layers.Conv2D(filters=16, kernel_size=5))
106 net.add(tf.keras.layers.BatchNormalization())
107 net.add(tf.keras.layers.Activation(‘sigmoid‘))
108 net.add(tf.keras.layers.MaxPool2D(pool_size=2, strides=2))
109 net.add(tf.keras.layers.Flatten())
110 net.add(tf.keras.layers.Dense(120))
111 net.add(tf.keras.layers.BatchNormalization())
112 net.add(tf.keras.layers.Activation(‘sigmoid‘))
113 net.add(tf.keras.layers.Dense(84))
114 net.add(tf.keras.layers.BatchNormalization())
115 net.add(tf.keras.layers.Activation(‘sigmoid‘))
116 net.add(tf.keras.layers.Dense(10, activation=‘sigmoid‘))
117 
118 
119 #训练修改后的模型
120 (x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()
121 x_train = x_train.reshape((60000, 28, 28, 1)).astype(‘float32‘)/255
122 x_test = x_test.reshape((10000, 28, 28, 1)).astype(‘float32‘)/255
123 
124 net.compile(loss=‘sparse_categorical_crossentropy‘,
125             optimizer=tf.keras.optimizers.RMSprop(),
126             metrics=[‘accuracy‘])
127 
128 history = net.fit(x_train, y_train, batch_size=64, epochs=5, validation_split=0.2)
129 
130 test_scores = net.evaluate(x_test, y_test, verbose=2)
131 print(‘Test loss:‘, test_scores[0])
132 print(‘Test accuracy:‘, test_scores[1])