wassname/keras-contrib
1
0
Fork 0
mirror of https://github.com/wassname/keras-contrib.git synced 2026-09-12 12:32:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #174 from infelane/master

Browse Source 
Fixed and extended Dead ReLU detector callback
This commit is contained in:
Andrew Hundt
2017-11-20 20:57:41 -05:00
committed by GitHub
parent b48ed724e7 68acf5e2ee
commit 7f549909e4
3 changed files with 213 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
keras_contrib/callbacks/dead_relu_detector.py
+42 -16
View File
@@ -1,8 +1,6 @@
import numpy as np
import warnings
from keras.callbacks import Callback
from keras.layers import Dense
from keras import backend as K
@@ -13,10 +11,11 @@ class DeadReluDetector(Callback):
# Arguments
x_train: Training dataset to check whether or not neurons fire
verbose: verbosity mode
True means that even a single dead neuron triggers warning
True means that even a single dead neuron triggers a warning message
False means that only significant number of dead neurons (10% or more)
triggers warning
triggers a warning message
"""
def __init__(self, x_train, verbose=False):
super(DeadReluDetector, self).__init__()
self.x_train = x_train
@@ -25,7 +24,8 @@ class DeadReluDetector(Callback):
@staticmethod
def is_relu_layer(layer):
return isinstance(layer, Dense) and layer.get_config()['activation'] == 'relu'
# Should work for all layers with relu activation. Tested for Dense and Conv2D
return 'activation' in layer.get_config() and layer.get_config()['activation'] == 'relu'
def get_relu_activations(self):
model_input = self.model.input
@@ -44,17 +44,43 @@ class DeadReluDetector(Callback):
layer_outputs = [func(list_inputs)[0] for func in funcs]
for layer_index, layer_activations in enumerate(layer_outputs):
if self.is_relu_layer(self.model.layers[layer_index]):
yield [layer_index, layer_activations]
layer_name = self.model.layers[layer_index].name
# layer_weight is a list [W] (+ [b])
layer_weight = self.model.layers[layer_index].get_weights()
# with kernel and bias, the weights are saved as a list [W, b]. If only weights, it is [W]
if type(layer_weight) is not list:
raise ValueError("'Layer_weight' should be a list, but was {}".format(type(layer_weight)))
layer_weight_shape = np.shape(layer_weight[0])
yield [layer_index, layer_activations, layer_name, layer_weight_shape]
def on_epoch_end(self, epoch, logs={}):
for relu_activation in self.get_relu_activations():
layer_index, activation_values = relu_activation
total_neurons = activation_values.shape[-1]
dead_neurons = np.sum(activation_values == 0)
dead_neurons_share = dead_neurons / total_neurons
if (self.verbose and dead_neurons > 0) or dead_neurons_share > self.dead_neurons_share_threshold:
warnings.warn(
'Layer #{} has {} dead neurons ({:.2%})!'
.format(layer_index, dead_neurons, dead_neurons_share),
RuntimeWarning
)
layer_index, activation_values, layer_name, layer_weight_shape = relu_activation
shape_act = activation_values.shape
weight_len = len(layer_weight_shape)
act_len = len(shape_act)
# should work for both Conv and Flat
if K.image_data_format() == 'channels_last':
# features in last axis
axis_filter = -1
else:
# features before the convolution axis, for weight_len the input and output have to be subtracted
axis_filter = -1 - (weight_len - 2)
total_featuremaps = shape_act[axis_filter]
axis = tuple(
i for i in range(act_len) if (i != axis_filter) and (i != (len(shape_act) + axis_filter)))
dead_neurons = np.sum(np.sum(activation_values, axis=axis) == 0)
dead_neurons_share = float(dead_neurons) / float(total_featuremaps)
if (self.verbose and dead_neurons > 0) or dead_neurons_share >= self.dead_neurons_share_threshold:
str_warning = 'Layer {} (#{}) has {} dead neurons ({:.2%})!'.format(layer_name, layer_index,
dead_neurons, dead_neurons_share)
print(str_warning)
tests/keras_contrib/backend/backend_test.py
+3 -2
View File
@@ -157,8 +157,9 @@ class TestBackend(object):
th_var_val = KTH.eval(th_var)
tf_var_val = KTF.eval(tf_var)
assert_allclose(th_mean_val, tf_mean_val, rtol=1e-4)
assert_allclose(th_var_val, tf_var_val, rtol=1e-4)
# absolute tolerance needed when working with zeros
assert_allclose(th_mean_val, tf_mean_val, rtol=1e-4, atol=1e-10)
assert_allclose(th_var_val, tf_var_val, rtol=1e-4, atol=1e-10)
def test_clip(self):
check_single_tensor_operation('clip', (4, 2), min_value=0.4, max_value=0.6)
tests/keras_contrib/callbacks/dead_relu_detector_test.py
+168 -17
View File
@@ -1,40 +1,191 @@
import pytest
import warnings
import numpy as np
import sys
if (sys.version_info > (3, 0)):
from io import StringIO
else:
from StringIO import StringIO
from keras_contrib import callbacks
from keras.models import Sequential
from keras.layers import Dense
from keras.layers import Dense, Conv2D, Flatten
from keras import backend as K
n_out = 11 # with 1 neuron dead, 1/11 is just below the threshold of 10% with verbose = False
def check_print(do_train, expected_warnings, nr_dead=None, perc_dead=None):
"""
Receive stdout to check if correct warning message is delivered
:param nr_dead: int
:param perc_dead: float, 10% should be written as 0.1
"""
saved_stdout = sys.stdout
out = StringIO()
out.flush()
sys.stdout = out # overwrite current stdout
do_train()
stdoutput = out.getvalue().strip() # get prints, can be something like: "Layer dense (#0) has 2 dead neurons (20.00%)!"
str_to_count = "dead neurons"
count = stdoutput.count(str_to_count)
sys.stdout = saved_stdout # restore stdout
out.close()
assert expected_warnings == count
if expected_warnings and (nr_dead is not None):
str_to_check = 'has {} dead'.format(nr_dead)
assert str_to_check in stdoutput, '"{}" not in "{}"'.format(str_to_check, stdoutput)
if expected_warnings and (perc_dead is not None):
str_to_check = 'neurons ({:.2%})!'.format(perc_dead)
assert str_to_check in stdoutput, '"{}" not in "{}"'.format(str_to_check, stdoutput)
def test_DeadDeadReluDetector():
def do_test(weights, expected_warnings, verbose):
with warnings.catch_warnings(record=True) as w:
dataset = np.ones((1, 1, 1)) # data to be fed as training
n_samples = 9
input_shape = (n_samples, 3, 4) # 4 input features
shape_out = (n_samples, 3, n_out) # 11 output features
shape_weights = (4, n_out)
# ignore batch size
input_shape_dense = tuple(input_shape[1:])
def do_test(weights, expected_warnings, verbose, nr_dead=None, perc_dead=None):
def do_train():
dataset = np.ones(input_shape) # data to be fed as training
model = Sequential()
model.add(Dense(10, activation='relu', input_shape=(1, 1), use_bias=False, weights=[weights]))
model.add(Dense(n_out, activation='relu', input_shape=input_shape_dense,
use_bias=False, weights=[weights], name='dense'))
model.compile(optimizer='sgd', loss='categorical_crossentropy')
model.fit(
dataset,
np.ones((1, 1, 10)),
np.ones(shape_out),
batch_size=1,
epochs=1,
callbacks=[callbacks.DeadReluDetector(dataset, verbose=verbose)],
verbose=False
)
assert len(w) == expected_warnings
for warn_item in w:
assert issubclass(warn_item.category, RuntimeWarning)
assert "dead neurons" in str(warn_item.message)
weights_1_dead = np.ones((1, 10)) # weights that correspond to NN with 1/10 neurons dead
check_print(do_train, expected_warnings, nr_dead, perc_dead)
weights_1_dead = np.ones(shape_weights) # weights that correspond to NN with 1/11 neurons dead
weights_2_dead = np.ones(shape_weights) # weights that correspond to NN with 2/11 neurons dead
weights_all_dead = np.zeros(shape_weights) # weights that correspond to all neurons dead
weights_1_dead[:, 0] = 0
weights_2_dead = np.ones((1, 10)) # weights that correspond to NN with 2/10 neurons dead
weights_2_dead[:, 0] = 0
weights_2_dead[:, 1] = 0
weights_2_dead[:, 0:2] = 0
do_test(weights_1_dead, verbose=True, expected_warnings=1)
do_test(weights_1_dead, verbose=True, expected_warnings=1, nr_dead=1, perc_dead=1. / n_out)
do_test(weights_1_dead, verbose=False, expected_warnings=0)
do_test(weights_2_dead, verbose=True, expected_warnings=1)
do_test(weights_2_dead, verbose=True, expected_warnings=1, nr_dead=2, perc_dead=2. / n_out)
# do_test(weights_all_dead, verbose=True, expected_warnings=1, nr_dead=n_out, perc_dead=1.)
def test_DeadDeadReluDetector_bias():
n_samples = 9
input_shape = (n_samples, 4) # 4 input features
shape_weights = (4, n_out)
shape_bias = (n_out, )
shape_out = (n_samples, n_out) # 11 output features
# ignore batch size
input_shape_dense = tuple(input_shape[1:])
def do_test(weights, bias, expected_warnings, verbose, nr_dead=None, perc_dead=None):
def do_train():
dataset = np.ones(input_shape) # data to be fed as training
model = Sequential()
model.add(Dense(n_out, activation='relu', input_shape=input_shape_dense,
use_bias=True, weights=[weights, bias], name='dense'))
model.compile(optimizer='sgd', loss='categorical_crossentropy')
model.fit(
dataset,
np.ones(shape_out),
batch_size=1,
epochs=1,
callbacks=[callbacks.DeadReluDetector(dataset, verbose=verbose)],
verbose=False
)
check_print(do_train, expected_warnings, nr_dead, perc_dead)
weights_1_dead = np.ones(shape_weights) # weights that correspond to NN with 1/11 neurons dead
weights_2_dead = np.ones(shape_weights) # weights that correspond to NN with 2/11 neurons dead
weights_all_dead = np.zeros(shape_weights) # weights that correspond to all neurons dead
weights_1_dead[:, 0] = 0
weights_2_dead[:, 0:2] = 0
bias = np.zeros(shape_bias)
do_test(weights_1_dead, bias, verbose=True, expected_warnings=1, nr_dead=1, perc_dead=1. / n_out)
do_test(weights_1_dead, bias, verbose=False, expected_warnings=0)
do_test(weights_2_dead, bias, verbose=True, expected_warnings=1, nr_dead=2, perc_dead=2. / n_out)
# do_test(weights_all_dead, bias, verbose=True, expected_warnings=1, nr_dead=n_out, perc_dead=1.)
def test_DeadDeadReluDetector_conv():
n_samples = 9
# (5, 5) kernel, 4 input featuremaps and 11 output featuremaps
if K.image_data_format() == 'channels_last':
input_shape = (n_samples, 5, 5, 4)
else:
input_shape = (n_samples, 4, 5, 5)
# ignore batch size
input_shape_conv = tuple(input_shape[1:])
shape_weights = (5, 5, 4, n_out)
shape_out = (n_samples, n_out)
def do_test(weights_bias, expected_warnings, verbose, nr_dead=None, perc_dead=None):
"""
:param perc_dead: as float, 10% should be written as 0.1
"""
def do_train():
dataset = np.ones(input_shape) # data to be fed as training
model = Sequential()
model.add(Conv2D(n_out, (5, 5), activation='relu', input_shape=input_shape_conv,
use_bias=True, weights=weights_bias, name='conv'))
model.add(Flatten()) # to handle Theano's categorical crossentropy
model.compile(optimizer='sgd', loss='categorical_crossentropy')
model.fit(
dataset,
np.ones(shape_out),
batch_size=1,
epochs=1,
callbacks=[callbacks.DeadReluDetector(dataset, verbose=verbose)],
verbose=False
)
check_print(do_train, expected_warnings, nr_dead, perc_dead)
weights_1_dead = np.ones(shape_weights) # weights that correspond to NN with 1/11 neurons dead
weights_1_dead[..., 0] = 0
weights_2_dead = np.ones(shape_weights) # weights that correspond to NN with 2/11 neurons dead
weights_2_dead[..., 0:2] = 0
weights_all_dead = np.zeros(shape_weights) # weights that correspond to NN with all neurons dead
bias = np.zeros((11, ))
weights_bias_1_dead = [weights_1_dead, bias]
weights_bias_2_dead = [weights_2_dead, bias]
weights_bias_all_dead = [weights_all_dead, bias]
do_test(weights_bias_1_dead, verbose=True, expected_warnings=1, nr_dead=1, perc_dead=1. / n_out)
do_test(weights_bias_1_dead, verbose=False, expected_warnings=0)
do_test(weights_bias_2_dead, verbose=True, expected_warnings=1, nr_dead=2, perc_dead=2. / n_out)
# do_test(weights_bias_all_dead, verbose=True, expected_warnings=1, nr_dead=n_out, perc_dead=1.)
if __name__ == '__main__':