2

I am building a simple neural network using Keras. It has activity regularization so that the output of the only hidden layer is forced to have small values. Here is the code:

import numpy as np
import math
import keras
from keras.models import Model, Sequential
from keras.layers import Input, Dense, Activation
from keras import regularizers
from keras import backend as K

a=1

def my_regularizer(inputs):
     means=K.mean((inputs),axis=1)
     return a*K.sum(means)**2

x_train=np.random.uniform(low=-1,high=1,size=(200,2))

model=Sequential([
     Dense(20,input_shape=(2,),activity_regularizer=my_regularizer),
     Activation('tanh'),
     Dense(2,),
     Activation('linear')
])

model.compile(optimizer='adam',loss='mean_squared_error')
model.fit(x_train,x_train,epochs=20,validation_split=0.1)

Questions:

1) Currently, parameter a is set at the beginning and it does not change. How can I change the code such that the parameter a is updated after each iteration such that

a_new=f(a_old,input)

where input is the values at the hidden layer and f(.) is an arbitrary function.

2) I want my activity regularizer to be applied after the first activation function tanh is applied. Have I written my code correctly? The term "activity_regularizer=my_regularizer" in

Dense(20,input_sahpe=(2,),activity_regularizer=my_regularizer)

makes me feel that the regularizer is being applied to values before the activation function tanh.

Albert
  • 389
  • 3
  • 17
Sus20200
  • 337
  • 3
  • 13

1 Answers1

2

You can - but first, you need a valid Keras Regularizer object (your function won't work):

class MyActivityRegularizer(Regularizer):
    def __init__(self, a=1):
        self.a = K.variable(a, name='a')

    # gets called at each train iteration
    def __call__(self, x): # your custom function here
        means = K.mean(x, axis=1)
        return self.a * K.sum(means)**2

    def get_config(self): # required class method
        return {"a": float(K.get_value(self.a))}

Next, to work with .fit, you need a custom Keras Callback object (see alternative at bottom):

class ActivityRegularizerScheduler(Callback):
    """ 'on_batch_end' gets automatically called by .fit when finishing
    iterating over a batch. The model, and its attributes, are inherited by 
    'Callback' (except at __init__) and can be accessed via, e.g., self.model """

    def __init__(self, model, update_fn):
        self.update_fn=update_fn
        self.activity_regularizers=_get_activity_regularizers(model)

    def on_batch_end(self, batch, logs=None):
        iteration = K.get_value(self.model.optimizer.iterations)
        new_activity_reg = self.update_fn(iteration) 

        # 'activity_regularizer' references model layer's activity_regularizer (in this 
        # case 'MyActivityRegularizer'), so its attributes ('a') can be set directly
        for activity_regularizer in self.activity_regularizers:
            K.set_value(activity_regularizer.a, new_activity_reg)

def _get_activity_regularizers(model):
    activity_regularizers = []
    for layer in model.layers:
        a_reg = getattr(layer,'activity_regularizer',None)
        if a_reg is not None:
            activity_regularizers.append(a_reg)
    return activity_regularizers

Lastly, you'll need to create your model within the Keras CustomObjectScope - see in full ex. below.

Example usage: 
from keras.layers import Dense
from keras.models import Sequential
from keras.regularizers import Regularizer
from keras.callbacks import Callback
from keras.utils import CustomObjectScope
from keras.optimizers import Adam
import keras.backend as K
import numpy as np

def make_model(my_reg):
    return Sequential([
        Dense(20, activation='tanh', input_shape=(2,), activity_regularizer=my_reg),
        Dense(2,  activation='linear'),
        ])

my_reg = MyActivityRegularizer(a=1)

with CustomObjectScope({'MyActivityRegularizer':my_reg}): # required for Keras to recognize
    model = make_model(my_reg)
opt = Adam(lr=1e-4)
model.compile(optimizer=opt, loss='mse')

x = np.random.randn(320,2) # dummy data
y = np.random.randn(320,2) # dummy labels

update_fn = lambda x: .5 + .4*np.cos(x) #x = number of train updates (optimizer.iterations)
activity_regularizer_scheduler = ActivityRegularizerScheduler(model, update_fn)

model.fit(x,y,batch_size=32,callbacks=[activity_regularizer_scheduler],
          epochs=4,verbose=1)

To TRACK your a and make sure it's changing, you can get its value at, e.g., each epoch end via:

for epoch in range(4):
    model.fit(x,y,batch_size=32,callbacks=[activity_regularizer_scheduler],epochs=1)
    print("Epoch {} activity_regularizer 'a': {}".format(epoch,
            K.get_value(_get_activity_regularizers(model)[0].a)))
# My output:
# Epoch 0 activity_regularizer 'a': 0.7190816402435303
# Epoch 1 activity_regularizer 'a': 0.4982417821884155
# Epoch 2 activity_regularizer 'a': 0.2838689386844635
# Epoch 3 activity_regularizer 'a': 0.8644570708274841

Regarding (2), I'm afraid you're right - the 'tanh' outputs won't be used; you'll need to pass activation='tanh' instead.
Lastly, you can do it without a callback, via train_on_batch - but a drawback is, you'll need to feed data to the model yourself (and shuffle it, etc): 
activity_regularizers = _get_activity_regularizers(model)

for iteration in range(100):
   x, y = get_data()
   model.train_on_batch(x,y)
   iteration = K.get_value(model.optimizer.iterations)

   for activity_regularizer in activity_regularizers:
       K.set_value(activity_regularizer, update_fn(iteration))
OverLordGoldDragon
  • 1
  • 9
  • 53
  • 101