Adding a muplitiply layer to an autoencoder in Keras

Question

I want to add a multiply layer on top of an LSTM autoencoder. The multiply layer should multiply the tensor for a constant value. I wrote the following code which work without the multiply layer. Does anyone know how to adjust and make this working?

import keras
from keras import backend as K
from keras.models import Sequential, Model
from keras.layers import Input, LSTM, RepeatVector, TimeDistributed
from keras.layers.core import Flatten, Dense, Dropout, Lambda
from keras.optimizers import SGD, RMSprop, Adam
from keras import objectives
from keras.engine.topology import Layer
import numpy as np

class LayerKMultiply(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        self.k = Null
        super(LayerKMultiply, self).__init__(**kwargs)

    def build(self, input_shape):
        # Create a trainable weight variable for this layer.
        self.k = self.add_weight(
            name='k',
            shape=(),
            initializer='ones',
            dtype='float32',
            trainable=True,
        )
        super(LayerKMultiply, self).build(input_shape)  # Be sure to call this at the end

    def call(self, x):
        #return K.tf.multiply(self.k, x)
        return self.k * x

    def compute_output_shape(self, input_shape):
        return (input_shape[0], self.output_dim)

    timesteps, input_dim, latent_dim = 10, 3, 32


inputs = Input(shape=(timesteps, input_dim))
encoded = LSTM(latent_dim, return_sequences=False, activation='linear')(inputs)
decoded = RepeatVector(timesteps)(encoded)
decoded = LSTM(input_dim, return_sequences=True, activation='linear')(decoded)
decoded = TimeDistributed(Dense(input_dim, activation='linear'))(decoded)
#decoded = LayerKMultiply(k = 20)(decoded)

sequence_autoencoder = Model(inputs, decoded)
encoder = Model(inputs, encoded)

autoencoder = Model(inputs, decoded)
autoencoder.compile(optimizer='adam', loss='mse')

    X = np.array([[[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10],[1,2,3,4,5,6,7,8,9,10]]])
X = X.reshape(1,10,3)
p = autoencoder.predict(x=X, batch_size=1)
print(p)

I am getting the following error:

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-3-b2f9497bbf47> in <module>()
      7 decoded = LSTM(input_dim, return_sequences=True, activation='linear')(decoded)
      8 decoded = TimeDistributed(Dense(input_dim, activation='linear'))(decoded)
----> 9 decoded = LayerKMultiply(k = 20)(decoded)
     10 
     11 sequence_autoencoder = Model(inputs, decoded)

TypeError: __init__() missing 1 required positional argument: 'output_dim'

EDIT

What I want to achieve is the architecture described in the following images:

https://github.com/mg64ve/SMTDAE/blob/master/images/SMTDAE.png https://github.com/mg64ve/SMTDAE/blob/master/images/REF.png

So in this sense, I believe the multiply layer hase to be before the TimeDistributed and Dense layers. I modified the code as following:

import keras
from keras import backend as K
from keras.models import Sequential, Model
from keras.layers import Input, LSTM, RepeatVector, TimeDistributed
from keras.layers.core import Flatten, Dense, Dropout, Lambda
from keras.optimizers import SGD, RMSprop, Adam
from keras import objectives
from keras.engine.topology import Layer
import numpy as np

class LayerKMultiply(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        self.k = None
        super(LayerKMultiply, self).__init__(**kwargs)

    def build(self, input_shape):
        # Create a trainable weight variable for this layer.
        self.k = self.add_weight(
            name='k',
            shape=(),
            initializer='ones',
            dtype='float32',
            trainable=True,
        )
        super(LayerKMultiply, self).build(input_shape)  # Be sure to call this at the end

    def call(self, x):
        return self.k * x

    def compute_output_shape(self, input_shape):
        return (input_shape[0], input_shape[1], self.output_dim)

timesteps, input_dim, latent_dim = 31, 31, 32


inputs = Input(shape=(timesteps, input_dim))
encoded = LSTM(latent_dim, return_sequences=False, activation='linear')(inputs)
decoded = RepeatVector(timesteps)(encoded)
decoded = LSTM(input_dim, return_sequences=True, activation='linear')(decoded)
decoded = LayerKMultiply(20)(decoded)
decoded = TimeDistributed(Dense(input_dim, activation='linear'))(decoded)

autoencoder = Model(inputs, decoded)

batch_size = 100
X = np.zeros([5000,31,31])
autoencoder.fit(X, X, batch_size = batch_size, epochs=3)

autoencoder.compile(optimizer='adam', loss='mse')

But I am still getting the following error:

InvalidArgumentError: Incompatible shapes: [155,31,31] vs. [100,31,31]

Answer 1

I believe the solution is the following:

import keras
from keras import backend as K
from keras.models import Sequential, Model
from keras.layers import Input, LSTM, RepeatVector, TimeDistributed
from keras.layers.core import Flatten, Dense, Dropout, Lambda
from keras.optimizers import SGD, RMSprop, Adam
from keras import objectives
from keras.engine.topology import Layer
import numpy as np

class LayerKMultiply(Layer):

    def __init__(self, output_dim, **kwargs):
        self.output_dim = output_dim
        self.k = None
        super(LayerKMultiply, self).__init__(**kwargs)

    def build(self, input_shape):
        # Create a trainable weight variable for this layer.
        self.k = self.add_weight(
            name='k',
            shape=(),
            initializer='ones',
            dtype='float32',
            trainable=True,
        )
        super(LayerKMultiply, self).build(input_shape)  # Be sure to call this at the end

    def call(self, x):
        return self.k * x

    def compute_output_shape(self, input_shape):
        return (input_shape[0], input_shape[1], input_shape[2])

Answer 2

You are mixing positional arguments with keyword arguments. When you define a function like def __init__(self, output_dim, **kwargs) output_dim is a positional argument. You need to:

• either pass 20 on its own LayerMultiply(20)(decoded)
• or change def __init__(self, k=10, **kwargs)
• or remove output_dim from definition and use self.output_dim = kwargs['k']

More information here.