我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用keras.layers.LSTM。
def create_lstm_autoencoder(input_dim, timesteps, latent_dim): """ Creates an LSTM Autoencoder (VAE). Returns Autoencoder, Encoder, Generator. (All code by fchollet - see reference.) # Arguments input_dim: int. timesteps: int, input timestep dimension. latent_dim: int, latent z-layer shape. # References - [Building Autoencoders in Keras](https://blog.keras.io/building-autoencoders-in-keras.html) """ inputs = Input(shape=(timesteps, input_dim,)) encoded = LSTM(latent_dim)(inputs) decoded = RepeatVector(timesteps)(encoded) decoded = LSTM(input_dim, return_sequences=True)(decoded) sequence_autoencoder = Model(inputs, decoded) encoder = Model(inputs, encoded) autoencoder = Model(inputs, decoded) autoencoder.compile(optimizer='adam', loss='mse') return autoencoder
def __init__(self, input_shape, lr=0.01, n_layers=2, n_hidden=8, rate_dropout=0.2, loss='risk_estimation'): print("initializing..., learing rate %s, n_layers %s, n_hidden %s, dropout rate %s." %(lr, n_layers, n_hidden, rate_dropout)) self.model = Sequential() self.model.add(Dropout(rate=rate_dropout, input_shape=(input_shape[0], input_shape[1]))) for i in range(0, n_layers - 1): self.model.add(LSTM(n_hidden * 4, return_sequences=True, activation='tanh', recurrent_activation='hard_sigmoid', kernel_initializer='glorot_uniform', recurrent_initializer='orthogonal', bias_initializer='zeros', dropout=rate_dropout, recurrent_dropout=rate_dropout)) self.model.add(LSTM(n_hidden, return_sequences=False, activation='tanh', recurrent_activation='hard_sigmoid', kernel_initializer='glorot_uniform', recurrent_initializer='orthogonal', bias_initializer='zeros', dropout=rate_dropout, recurrent_dropout=rate_dropout)) self.model.add(Dense(1, kernel_initializer=initializers.glorot_uniform())) # self.model.add(BatchNormalization(axis=-1, moving_mean_initializer=Constant(value=0.5), # moving_variance_initializer=Constant(value=0.25))) self.model.add(BatchRenormalization(axis=-1, beta_init=Constant(value=0.5))) self.model.add(Activation('relu_limited')) opt = RMSprop(lr=lr) self.model.compile(loss=loss, optimizer=opt, metrics=['accuracy'])
def test(path_test, input_size, hidden_size, batch_size, save_dir, model_name, maxlen): db = read_data(path_test) X = create_sequences(db[:-maxlen], win_size=maxlen, step=maxlen) X = np.reshape(X, (X.shape[0], X.shape[1], input_size)) # build the model: 1 layer LSTM print('Build model...') model = Sequential() model.add(LSTM(hidden_size, return_sequences=False, input_shape=(maxlen, input_size))) model.add(Dense(maxlen)) model.load_weights(save_dir + model_name) model.compile(loss='mse', optimizer='adam') prediction = model.predict(X, batch_size, verbose=1) prediction = prediction.flatten() # prediction_container = np.array(prediction).flatten() Y = db[maxlen:] plt.plot(prediction, label='prediction') plt.plot(Y, label='true') plt.legend() plt.show()
def __init__(self, output_dim, num_senses, num_hyps, use_attention=False, return_attention=False, **kwargs): # Set output_dim in kwargs so that we can pass it along to LSTM's init kwargs['output_dim'] = output_dim self.num_senses = num_senses self.num_hyps = num_hyps self.use_attention = use_attention self.return_attention = return_attention super(OntoAttentionLSTM, self).__init__(**kwargs) # Recurrent would have set the input shape to cause the input dim to be 3. Change it. self.input_spec = [InputSpec(ndim=5)] if self.consume_less == "cpu": # In the LSTM implementation in Keras, consume_less = cpu causes all gates' inputs to be precomputed # and stored in memory. However, this doesn't work with OntoLSTM since the input to the gates is # dependent on the previous timestep's output. warnings.warn("OntoLSTM does not support consume_less = cpu. Changing it to mem.") self.consume_less = "mem" #TODO: Remove this dependency. if K.backend() == "tensorflow" and not self.unroll: warnings.warn("OntoLSTM does not work with unroll=False when backend is TF. Changing it to True.") self.unroll = True
def call(self, x, mask=None): # input_shape = (batch_size, input_length, input_dim). This needs to be defined in build. initial_read_states = self.get_initial_states(x, mask) fake_writer_input = K.expand_dims(initial_read_states[0], dim=1) # (batch_size, 1, output_dim) initial_write_states = self.writer.get_initial_states(fake_writer_input) # h_0 and c_0 of the writer LSTM initial_states = initial_read_states + initial_write_states # last_output: (batch_size, output_dim) # all_outputs: (batch_size, input_length, output_dim) # last_states: # last_memory_state: (batch_size, input_length, output_dim) # last_output # last_writer_ct last_output, all_outputs, last_states = self.loop(x, initial_states, mask) last_memory = last_states[0] if self.return_mode == "last_output": return last_output elif self.return_mode == "all_outputs": return all_outputs else: # return mode is output_and_memory expanded_last_output = K.expand_dims(last_output, dim=1) # (batch_size, 1, output_dim) # (batch_size, 1+input_length, output_dim) return K.concatenate([expanded_last_output, last_memory], axis=1)
def get_initial_states(self, nse_input, input_mask=None): ''' Read input in MMA-NSE will be of shape (batch_size, read_input_length*2, input_dim), a concatenation of the actual input to this NSE and the output from a different NSE. The latter will be used to initialize the shared memory. The former will be passed to the read LSTM and also used to initialize the current memory. ''' input_length = K.shape(nse_input)[1] read_input_length = input_length/2 input_to_read = nse_input[:, :read_input_length, :] initial_shared_memory = K.batch_flatten(nse_input[:, read_input_length:, :]) mem_0 = K.batch_flatten(input_to_read) o_mask = self.reader.compute_mask(input_to_read, input_mask) reader_states = self.reader.get_initial_states(nse_input) initial_states = reader_states + [mem_0, initial_shared_memory] return initial_states, o_mask
def train_model(self): # scale scaler = MinMaxScaler(feature_range=(0, 1)) dataset = scaler.fit_transform(self.data) # split into train and test sets train_size = int(len(dataset) * 0.95) train, test = dataset[0:train_size, :], dataset[train_size:len(dataset), :] look_back = 5 trainX, trainY = self.create_dataset(train, look_back) # reshape input to be [samples, time steps, features] trainX = numpy.reshape(trainX, (trainX.shape[0], 1, trainX.shape[1])) # create and fit the LSTM network model = Sequential() model.add(LSTM(6, input_dim=look_back)) model.add(Dense(1)) model.compile(loss='mean_squared_error', optimizer='adam') model.fit(trainX, trainY, nb_epoch=100, batch_size=1, verbose=2) return model
def rcnn(input_shape, n_classes): """ Input size should be [batch, 1d, ch] = (XXX, 3000, 1) """ model = Sequential(name='RCNN test') model.add(Conv1D (kernel_size = (200), filters = 20, batch_input_shape=input_shape, activation='elu')) model.add(MaxPooling1D(pool_size = (20), strides=(10))) model.add(Conv1D (kernel_size = (20), filters = 200, activation='elu')) model.add(MaxPooling1D(pool_size = (10), strides=(3))) model.add(Conv1D (kernel_size = (20), filters = 200, activation='elu')) model.add(MaxPooling1D(pool_size = (10), strides=(3))) model.add(Dense (512, activation='elu')) model.add(Dense (512, activation='elu')) model.add(Reshape((1,model.output_shape[1]))) model.add(LSTM(256, stateful=True, return_sequences=False)) model.add(Dropout(0.3)) model.add(Dense(n_classes, activation = 'sigmoid')) model.compile(loss='categorical_crossentropy', optimizer=Adadelta()) return model
def create_lstm_layer(self, input_dim): """Create a LSTM layer of a model.""" inp = Input(shape=(input_dim, self.embedding_dim,)) inp_dropout = Dropout(self.ldrop_val)(inp) ker_in = glorot_uniform(seed=self.seed) rec_in = Orthogonal(seed=self.seed) outp = LSTM(self.hidden_dim, input_shape=(input_dim, self.embedding_dim,), kernel_regularizer=None, recurrent_regularizer=None, bias_regularizer=None, activity_regularizer=None, recurrent_dropout=self.recdrop_val, dropout=self.inpdrop_val, kernel_initializer=ker_in, recurrent_initializer=rec_in, return_sequences=True)(inp_dropout) outp_dropout = Dropout(self.dropout_val)(outp) model = Model(inputs=inp, outputs=outp_dropout, name="LSTM_encoder") return model
def create_lstm_layer_1(self, input_dim): """Create a LSTM layer of a model.""" inp = Input(shape=(input_dim, self.embedding_dim,)) inp_drop = Dropout(self.ldrop_val)(inp) ker_in = glorot_uniform(seed=self.seed) rec_in = Orthogonal(seed=self.seed) bioutp = Bidirectional(LSTM(self.hidden_dim, input_shape=(input_dim, self.embedding_dim,), kernel_regularizer=None, recurrent_regularizer=None, bias_regularizer=None, activity_regularizer=None, recurrent_dropout=self.recdrop_val, dropout=self.inpdrop_val, kernel_initializer=ker_in, recurrent_initializer=rec_in, return_sequences=True), merge_mode=None)(inp_drop) dropout_forw = Dropout(self.dropout_val)(bioutp[0]) dropout_back = Dropout(self.dropout_val)(bioutp[1]) model = Model(inputs=inp, outputs=[dropout_forw, dropout_back], name="biLSTM_encoder") return model
def create_lstm_layer_2(self, input_dim): """Create a LSTM layer of a model.""" inp = Input(shape=(input_dim, 2*self.perspective_num,)) inp_drop = Dropout(self.ldrop_val)(inp) ker_in = glorot_uniform(seed=self.seed) rec_in = Orthogonal(seed=self.seed) bioutp = Bidirectional(LSTM(self.aggregation_dim, input_shape=(input_dim, 2*self.perspective_num,), kernel_regularizer=None, recurrent_regularizer=None, bias_regularizer=None, activity_regularizer=None, recurrent_dropout=self.recdrop_val, dropout=self.inpdrop_val, kernel_initializer=ker_in, recurrent_initializer=rec_in, return_sequences=True), merge_mode=None)(inp_drop) dropout_forw = Dropout(self.dropout_val)(bioutp[0]) dropout_back = Dropout(self.dropout_val)(bioutp[1]) model = Model(inputs=inp, outputs=[dropout_forw, dropout_back], name="biLSTM_enc_persp") return model
def bilstm_woatt_model(self): """Define a model with bi-LSTM layers and without attention.""" input_a = Input(shape=(self.max_sequence_length, self.embedding_dim,)) input_b = Input(shape=(self.max_sequence_length, self.embedding_dim,)) lstm_layer = self.create_lstm_layer_last(self.max_sequence_length) lstm_last_a = lstm_layer(input_a) lstm_last_b = lstm_layer(input_b) dist = Lambda(self.cosine_dist, output_shape=self.cosine_dist_output_shape, name="similarity_network")([lstm_last_a, lstm_last_b]) dense = Dense(1, activation='sigmoid', name='similarity_score', kernel_regularizer=None, bias_regularizer=None, activity_regularizer=None)(dist) model = Model([input_a, input_b], dense) return model
def __init__(self, n_classes, vocab_size, max_len, num_units=128, useBiDirection=False, useAttention=False, learning_rate=0.001, dropout=0, embedding_size=300): self.model = Sequential() self.model.add(Embedding(input_dim=vocab_size, output_dim=embedding_size, input_length=max_len)) lstm_model = LSTM(num_units, dropout=dropout) if useBiDirection: lstm_model = Bidirectional(lstm_model) if useAttention: lstm_model = lstm_model print("Attention not implement yet ... ") self.model.add(lstm_model) self.model.add(Dense(n_classes, activation='softmax')) self.model.summary() self.model.compile(loss='categorical_crossentropy', optimizer=Adam(lr=learning_rate), metrics=['accuracy'])
def build_model(look_back: int, batch_size: int=1) -> Sequential: """ The function builds a keras Sequential model :param look_back: number of previous time steps as int :param batch_size: batch_size as int, defaults to 1 :return: keras Sequential model """ model = Sequential() model.add(LSTM(64, activation='relu', batch_input_shape=(batch_size, look_back, 1), stateful=True, return_sequences=False)) model.add(Dense(1, activation='linear')) model.compile(loss='mean_squared_error', optimizer='adam') return model
def __init__(self, rnn_class=LSTM, encoder_dims=50, bidirectional=True, dropout_rate=0.5, **rnn_kwargs): """Creates an RNN model with attention. The attention mechanism is implemented as described in https://www.cs.cmu.edu/~hovy/papers/16HLT-hierarchical-attention-networks.pdf, but without sentence level attention. Args: rnn_class: The type of RNN to use. (Default Value = LSTM) encoder_dims: The number of hidden units of RNN. (Default Value: 50) bidirectional: Whether to use bidirectional encoding. (Default Value = True) **rnn_kwargs: Additional args for building the RNN. """ super(AttentionRNN, self).__init__(dropout_rate) self.rnn_class = rnn_class self.encoder_dims = encoder_dims self.bidirectional = bidirectional self.rnn_kwargs = rnn_kwargs
def create_BiLSTM(wordvecs, lstm_dim=300, output_dim=2, dropout=.5, weights=None, train=True): model = Sequential() if weights != None: model.add(Embedding(len(wordvecs)+1, len(wordvecs['the']), weights=[weights], trainable=train)) else: model.add(Embedding(len(wordvecs)+1, len(wordvecs['the']), trainable=train)) model.add(Dropout(dropout)) model.add(Bidirectional(LSTM(lstm_dim))) model.add(Dropout(dropout)) model.add(Dense(output_dim, activation='softmax')) if output_dim == 2: model.compile('adam', 'binary_crossentropy', metrics=['accuracy']) else: model.compile('adam', 'categorical_crossentropy', metrics=['accuracy']) return model
def print_results(bi, file, out_file, file_type): names, results, std_devs, dim = test_embeddings(bi, file, file_type) rr = [[u'{0:.3f} \u00B1{1:.3f}'.format(r, s) for r, s in zip(result, std_dev)] for result, std_dev in zip(results, std_devs)] table_data = [[name] + result for name, result in zip(names, rr)] table = tabulate.tabulate(table_data, headers=['dataset', 'acc', 'prec', 'rec', 'f1'], tablefmt='simple', floatfmt='.3f') if out_file: with open(out_file, 'a') as f: f.write('\n') if bi: f.write('+++Bidirectional LSTM+++\n') else: f.write('+++LSTM+++\n') f.write(table) f.write('\n') else: print() if bi: print('Bidirectional LSTM') else: print('LSTM') print(table)
def textgenrnn_model(weights_path, num_classes, maxlen=40): ''' Builds the model architecture for textgenrnn and loads the pretrained weights for the model. ''' input = Input(shape=(maxlen,), name='input') embedded = Embedding(num_classes, 100, input_length=maxlen, trainable=True, name='embedding')(input) rnn = LSTM(128, return_sequences=False, name='rnn')(embedded) output = Dense(num_classes, name='output', activation='softmax')(rnn) model = Model(inputs=[input], outputs=[output]) model.load_weights(weights_path, by_name=True) model.compile(loss='categorical_crossentropy', optimizer='nadam') return model
def __init__(self, dict_path: str=None, word_max_length: int=30, language: str = "ru", rnn=LSTM, units1: int=256, units2: int=256, dropout: float=0.2, batch_size=2048, emb_dimension=30): self.rnn = rnn self.dropout = dropout # type: float self.units1 = units1 # type: int self.units2 = units2 # type: int self.language = language # type: str if language == "ru": self.grapheme_alphabet = RU_GRAPHEME_SET elif language == "en": self.grapheme_alphabet = EN_GRAPHEME_SET else: assert False self.dict_path = dict_path # type: str self.word_max_length = word_max_length # type: int self.emb_dimension = emb_dimension self.batch_size = batch_size self.model = None
def __init__(self, **kwargs): """ :param **kwargs: output_dim=4: output dimension of LSTM layer; activation_lstm='tanh': activation function for LSTM layers; activation_dense='relu': activation function for Dense layer; activation_last='sigmoid': activation function for last layer; drop_out=0.2: fraction of input units to drop; np_epoch=10, the number of epoches to train the model. epoch is one forward pass and one backward pass of all the training examples; batch_size=32: number of samples per gradient update. The higher the batch size, the more memory space you'll need; loss='mean_square_error': loss function; optimizer='rmsprop' """ self.output_dim = kwargs.get('output_dim', 8) self.activation_lstm = kwargs.get('activation_lstm', 'relu') self.activation_dense = kwargs.get('activation_dense', 'relu') self.activation_last = kwargs.get('activation_last', 'softmax') # softmax for multiple output self.dense_layer = kwargs.get('dense_layer', 2) # at least 2 layers self.lstm_layer = kwargs.get('lstm_layer', 2) self.drop_out = kwargs.get('drop_out', 0.2) self.nb_epoch = kwargs.get('nb_epoch', 10) self.batch_size = kwargs.get('batch_size', 100) self.loss = kwargs.get('loss', 'categorical_crossentropy') self.optimizer = kwargs.get('optimizer', 'rmsprop')
def prep_model(inputs, N, s0pad, s1pad, c, granlevels=1): # LSTM lstm = LSTM(N, return_sequences=True, implementation=2, kernel_regularizer=l2(c['l2reg']), recurrent_regularizer=l2(c['l2reg']), bias_regularizer=l2(c['l2reg'])) x1 = inputs[0] x2 = inputs[1] h1 = lstm(x1) h2 = lstm(x2) W_x = Dense(N, kernel_initializer='glorot_uniform', use_bias=True, kernel_regularizer=l2(c['l2reg'])) W_h = Dense(N, kernel_initializer='orthogonal', use_bias=True, kernel_regularizer=l2(c['l2reg'])) sigmoid = Activation('sigmoid') a1 = multiply([x1, sigmoid( add([W_x(x1), W_h(h1)]) )]) a2 = multiply([x2, sigmoid( add([W_x(x2), W_h(h2)]) )]) # Averaging avg = Lambda(function=lambda x: K.mean(x, axis=1), output_shape=lambda shape: (shape[0], ) + shape[2:]) gran1 = avg(a1) gran2 = avg(a2) return [gran1, gran2], N
def create_char_lstm_model(self, emb_dim, word_maxlen, vocab_char_size, char_maxlen): from keras.layers import LSTM logger.info('Building character LSTM model') input_char = Input(shape=(char_maxlen, ), name='input_char') char_emb = Embedding( vocab_char_size, emb_dim, mask_zero=True)(input_char) lstm = LSTM( 300, return_sequences=True, dropout=self.dropout, recurrent_dropout=self.recurrent_dropout)(char_emb) dropped = Dropout(0.5)(lstm) mot = MeanOverTime(mask_zero=True)(dropped) densed = Dense(self.num_outputs, name='dense')(mot) output = Activation('sigmoid')(densed) model = Model(inputs=input_char, outputs=output) model.get_layer('dense').bias.set_value(self.bias) logger.info(' Done') return model
def test_conv1d_lstm(self): from keras.layers import Conv1D, LSTM, Dense model = Sequential() # input_shape = (time_step, dimensions) model.add(Conv1D(32,3,padding='same',input_shape=(10,8))) # conv1d output shape = (None, 10, 32) model.add(LSTM(24)) model.add(Dense(1, activation='sigmoid')) input_names = ['input'] output_names = ['output'] spec = keras.convert(model, input_names, output_names).get_spec() self.assertIsNotNone(spec) self.assertTrue(spec.HasField('neuralNetwork')) # Test the inputs and outputs self.assertEquals(len(spec.description.input), len(input_names) + 2) self.assertEquals(len(spec.description.output), len(output_names) + 2) # Test the layer parameters. layers = spec.neuralNetwork.layers self.assertIsNotNone(layers[0].convolution) self.assertIsNotNone(layers[1].simpleRecurrent) self.assertIsNotNone(layers[2].innerProduct)
def test_SimpleLSTMStacked(self): params = dict( input_dims=[1, 1, 1], go_backwards=False, activation='tanh', stateful=False, unroll=False, return_sequences=False, output_dim=1 ), model = Sequential() model.add(LSTM(output_dim=params[0]['output_dim'], input_length=params[0]['input_dims'][1], input_dim=params[0]['input_dims'][2], activation=params[0]['activation'], inner_activation='sigmoid', return_sequences=True, go_backwards=params[0]['go_backwards'], unroll=params[0]['unroll'], )) model.add(LSTM(output_dim=1, activation='tanh', inner_activation='sigmoid', )) relative_error, keras_preds, coreml_preds = simple_model_eval(params, model) for i in range(len(relative_error)): self.assertLessEqual(relative_error[i], 0.01)
def test_many_to_many(self): params = dict( input_dims=[1, 10, 5], go_backwards=False, activation='tanh', # fails with hard_sigmoid stateful=False, unroll=False, return_sequences=True, output_dim=1 ), model = Sequential() model.add(LSTM(output_dim=params[0]['output_dim'], input_shape=(10, 5), activation=params[0]['activation'], inner_activation='sigmoid', return_sequences=True, )) relative_error, keras_preds, coreml_preds = simple_model_eval(params, model) # print relative_error, '\n', keras_preds, '\n', coreml_preds, '\n' for i in range(len(relative_error)): self.assertLessEqual(relative_error[i], 0.01)
def test_classifier_no_name(self): np.random.seed(1988) input_dim = 5 num_hidden = 12 num_classes = 6 input_length = 3 model = Sequential() model.add(LSTM(num_hidden, input_dim=input_dim, input_length=input_length, return_sequences=False)) model.add(Dense(num_classes, activation='softmax')) model.set_weights([np.random.rand(*w.shape) for w in model.get_weights()]) input_names = ['input'] output_names = ['zzzz'] class_labels = ['a', 'b', 'c', 'd', 'e', 'f'] predicted_feature_name = 'pf' coremlmodel = keras_converter.convert(model, input_names, output_names, class_labels=class_labels, predicted_feature_name=predicted_feature_name) inputs = np.random.rand(input_dim) outputs = coremlmodel.predict({'input': inputs}) # this checks that the dictionary got the right name and type self.assertEquals(type(outputs[output_names[0]]), type({'a': 0.5}))
def test_tiny_no_sequence_lstm_zeros_gpu(self): np.random.seed(1988) input_dim = 1 input_length = 1 num_channels = 1 # Define a model model = Sequential() model.add(LSTM(num_channels, input_shape = (input_length, input_dim), implementation = 2, recurrent_activation = 'sigmoid')) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, mode = 'zeros', input_blob = 'data', output_blob = 'output')
def test_small_no_sequence_lstm_random(self): np.random.seed(1988) input_dim = 10 input_length = 1 num_channels = 1 # Define a model model = Sequential() model.add(LSTM(num_channels, input_shape = (input_length, input_dim), implementation = 2, recurrent_activation = 'sigmoid')) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output')
def test_tiny_no_sequence_bidir_random(self, model_precision=_MLMODEL_FULL_PRECISION): np.random.seed(1988) input_dim = 1 input_length = 1 num_channels = 1 num_samples = 1 # Define a model model = Sequential() model.add(Bidirectional(LSTM(num_channels, implementation = 1, recurrent_activation = 'sigmoid'), input_shape=(input_length, input_dim))) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output', model_precision=model_precision)
def test_tiny_no_sequence_bidir_random_gpu(self, model_precision = _MLMODEL_FULL_PRECISION): np.random.seed(1988) input_dim = 1 input_length = 1 num_channels = 1 num_samples = 1 # Define a model model = Sequential() model.add(Bidirectional(LSTM(num_channels, implementation = 2, recurrent_activation = 'sigmoid'), input_shape=(input_length, input_dim))) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output', model_precision=model_precision)
def test_small_no_sequence_bidir_random(self): np.random.seed(1988) input_dim = 10 input_length = 1 num_channels = 1 # Define a model model = Sequential() model.add(Bidirectional(LSTM(num_channels, implementation = 2, recurrent_activation = 'sigmoid'), input_shape=(input_length, input_dim))) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output')
def test_medium_bidir_random_return_seq_false(self): np.random.seed(1988) input_dim = 7 input_length = 5 num_channels = 10 # Define a model model = Sequential() model.add(Bidirectional(LSTM(num_channels, return_sequences=False, implementation=2, recurrent_activation='sigmoid'), input_shape=(input_length, input_dim))) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob='data', output_blob='output')
def test_medium_bidir_random_return_seq_true(self): np.random.seed(1988) input_dim = 7 input_length = 5 num_channels = 10 # Define a model model = Sequential() model.add(Bidirectional(LSTM(num_channels, return_sequences = True, implementation = 2, recurrent_activation = 'sigmoid'), input_shape=(input_length, input_dim))) # Set some random weights model.set_weights([np.random.rand(*w.shape)*0.2-0.1 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output')
def test_tiny_sequence_lstm(self, model_precision=_MLMODEL_FULL_PRECISION): np.random.seed(1988) input_dim = 1 input_length = 2 num_channels = 1 # Define a model model = Sequential() model.add(LSTM(num_channels, input_shape = (input_length, input_dim), implementation = 1, recurrent_activation = 'sigmoid')) # Set some random weights model.set_weights([(np.random.rand(*w.shape)-0.5)*0.2 for w in model.get_weights()]) # Test the keras model self._test_keras_model(model, input_blob = 'data', output_blob = 'output', delta=1e-4, model_precision=model_precision)
def test_tiny_babi_rnn(self): vocab_size = 10 embed_hidden_size = 8 story_maxlen = 5 query_maxlen = 5 input_tensor_1 = Input(shape=(story_maxlen,)) x1 = Embedding(vocab_size, embed_hidden_size)(input_tensor_1) x1 = Dropout(0.3)(x1) input_tensor_2 = Input(shape=(query_maxlen,)) x2 = Embedding(vocab_size, embed_hidden_size)(input_tensor_2) x2 = Dropout(0.3)(x2) x2 = LSTM(embed_hidden_size, return_sequences=False)(x2) x2 = RepeatVector(story_maxlen)(x2) x3 = add([x1, x2]) x3 = LSTM(embed_hidden_size, return_sequences=False)(x3) x3 = Dropout(0.3)(x3) x3 = Dense(vocab_size, activation='softmax')(x3) model = Model(inputs=[input_tensor_1,input_tensor_2], outputs=[x3]) self._test_keras_model(model, one_dim_seq_flags=[True, True])
def __init__(self, nlp, expressions): self.nlp = nlp self.all_expressions = expressions self.num_features = nlp.vocab.vectors_length self.categories = list(set([expression.intent().name for expression in expressions])) intents = [expression.intent().name for expression in expressions] intents_indices = [self.categories.index(intent) for intent in intents] self.x = np.zeros(shape=(len(expressions), DOCUMENT_MAX_NUM_WORDS, self.num_features)).astype('float32') self.__init_x(self.x, expressions) self.y = np_utils.to_categorical(intents_indices) self.model = Sequential() self.model.add(Bidirectional(LSTM(int(DOCUMENT_MAX_NUM_WORDS * 1.5)), input_shape=(DOCUMENT_MAX_NUM_WORDS, self.num_features))) self.model.add(Dropout(0.3)) self.model.add(Dense(len(self.categories), activation='sigmoid')) self.model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
def bidirectional_lstm(len_output): # sequence_input is a matrix of glove vectors (one for each input word) sequence_input = Input( shape=(MAX_SEQUENCE_LENGTH, EMBEDDING_DIM,), dtype='float32') l_lstm = Bidirectional(LSTM(100))(sequence_input) preds = Dense(len_output, activation='softmax')(l_lstm) model = Model(sequence_input, preds) model.compile(loss='categorical_crossentropy', optimizer='rmsprop', metrics=[utils.f1_score, 'categorical_accuracy']) """ model.add(Bidirectional(LSTM(shape['nr_hidden']))) # dropout to avoid overfitting model.add(Dropout(settings['dropout'])) model.add(Dense(shape['nr_class'], activation='sigmoid')) model.compile(optimizer=Adam(lr=settings['lr']), loss='binary_crossentropy', metrics=['accuracy']) """ return model
def get_model_41(params): embedding_weights = pickle.load(open("../data/datasets/train_data/embedding_weights_w2v-google_MSD-AG.pk","rb")) # main sequential model model = Sequential() model.add(Embedding(len(embedding_weights[0]), params['embedding_dim'], input_length=params['sequence_length'], weights=embedding_weights)) #model.add(Dropout(params['dropout_prob'][0], input_shape=(params['sequence_length'], params['embedding_dim']))) model.add(LSTM(2048)) #model.add(Dropout(params['dropout_prob'][1])) model.add(Dense(output_dim=params["n_out"], init="uniform")) model.add(Activation(params['final_activation'])) logging.debug("Output CNN: %s" % str(model.output_shape)) if params['final_activation'] == 'linear': model.add(Lambda(lambda x :K.l2_normalize(x, axis=1))) return model # CRNN Arch for audio
def gen_model(input_dim=10, output_dim=8, batch_size=100): model_LSTM = Sequential() model_LSTM.name = "LSTM" model_LSTM.batch_size = batch_size model_LSTM.input_dim = input_dim model_LSTM.output_dim = output_dim model_LSTM.add(LSTM(input_shape=(None, input_dim), units=units, return_sequences=True)) model_LSTM.add(LSTM(units=units, return_sequences=True)) model_LSTM.add(LSTM(units=output_dim, return_sequences=True)) model_LSTM.add(Activation('sigmoid')) sgd = Adam(lr=lr, clipnorm=clipnorm) model_LSTM.compile(loss='binary_crossentropy', optimizer=sgd, metrics = ['binary_accuracy'], sample_weight_mode="temporal") return model_LSTM
def subj_run(index_embedding, dataset, num_words=5000, embedding_len=100, max_len=50): (x_train, y_train), (x_test, y_test) = ds.load_data(dataset, num_words) x_train = sequence.pad_sequences(x_train, maxlen=max_len) x_test = sequence.pad_sequences(x_test, maxlen=max_len) model = Sequential() model.add(Embedding(num_words, embedding_len, input_length=max_len, weights=[index_embedding])) model.add(LSTM(max_len, dropout=0.5, recurrent_dropout=0.5)) model.add(Dense(1, activation='sigmoid')) model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy']) print(model.summary()) model.fit(x_train, y_train, epochs=4, batch_size=50, verbose=2) score, acc = model.evaluate(x_test, y_test, verbose=0) print('Test score:', score) print('Test accuracy:', acc)
def imdb_run(index_embedding, dataset, num_words=5000, embedding_len=100, max_len=500): (x_train, y_train), (x_test, y_test) = ds.load_data(dataset, num_words) x_train = sequence.pad_sequences(x_train, maxlen=max_len) x_test = sequence.pad_sequences(x_test, maxlen=max_len) model = Sequential() model.add(Embedding(num_words, embedding_len, input_length=max_len, weights=[index_embedding])) model.add(LSTM(100, dropout=0.2, recurrent_dropout=0.2)) model.add(Dense(1, activation='sigmoid')) model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy']) print(model.summary()) model.fit(x_train, y_train, epochs=3, batch_size=64, verbose=2) score, acc = model.evaluate(x_test, y_test, verbose=0) print('Test score:', score) print('Test accuracy:', acc)
def train(X_train, y_train): model = Sequential() model.add(LSTM( lstm_neurons, batch_input_shape=(batch_size, X_train.shape[1], X_train.shape[2]), stateful=True)) model.add(Dense(1)) model.compile(loss='mean_squared_error', optimizer='adam') for i in range(epochs): print 'batch', i+1 model.fit( X_train, y_train, epochs=1, batch_size=batch_size, verbose=2, shuffle=False, validation_split=0.33) model.reset_states() return model
def build(self): dim_data = self.size_of_input_data_dim nb_time_step = self.size_of_input_timesteps financial_time_series_input = Input(shape=(nb_time_step, dim_data)) lstm_layer_1 = LSTM(output_dim=nb_hidden_units, dropout_U=dropout, dropout_W=dropout, inner_activation='sigmoid', W_regularizer=l2(l2_norm_alpha), b_regularizer=l2(l2_norm_alpha), activation='tanh', return_sequences=True) lstm_layer_2 = LSTM(output_dim=nb_hidden_units, dropout_U=dropout, dropout_W=dropout, inner_activation='sigmoid', W_regularizer=l2(l2_norm_alpha), b_regularizer=l2(l2_norm_alpha), activation='tanh', return_sequences=True) h1 = lstm_layer_1(financial_time_series_input) h2 = lstm_layer_2(h1) time_series_predictions = TimeDistributedDense(1)(h2) self.model = Model(financial_time_series_input, time_series_predictions, name="deep rnn for financial time series forecasting")
def train(train_generator,train_size,input_num,dims_num): print("Start Train Job! ") start=time.time() inputs=InputLayer(input_shape=(input_num,dims_num),batch_size=batch_size) layer1=LSTM(128) output=Dense(2,activation="softmax",name="Output") optimizer=Adam() model=Sequential() model.add(inputs) model.add(layer1) model.add(Dropout(0.5)) model.add(output) call=TensorBoard(log_dir=log_dir,write_grads=True,histogram_freq=1) model.compile(optimizer,loss="categorical_crossentropy",metrics=["accuracy"]) model.fit_generator(train_generator,steps_per_epoch=train_size//batch_size,epochs=epochs_num,callbacks=[call]) # model.fit_generator(train_generator, steps_per_epoch=5, epochs=5, callbacks=[call]) model.save(model_dir) end=time.time() print("Over train job in %f s"%(end-start))
def model_cnn(net_layers, input_shape): inp = Input(shape=input_shape) model = inp for cl in net_layers['conv_layers']: model = Conv2D(filters=cl[0], kernel_size=cl[1], activation='relu')(model) if cl[4]: model = MaxPooling2D()(model) if cl[2]: model = BatchNormalization()(model) if cl[3]: model = Dropout(0.2)(model) model = Flatten()(model) for dl in net_layers['dense_layers']: model = Dense(dl[0])(model) model = Activation('relu')(model) if dl[1]: model = BatchNormalization()(model) if dl[2]: model = Dropout(0.2)(model) model = Dense(1)(model) model = Activation('sigmoid')(model) model = Model(inp, model) return model # %% # LSTM architecture # conv_layers -> [(filters, kernel_size, BatchNormaliztion, Dropout, MaxPooling)] # dense_layers -> [(num_neurons, BatchNormaliztion, Dropout)]
def model_lstm(input_shape): inp = Input(shape=input_shape) model = inp if input_shape[0] > 2: model = Conv1D(filters=24, kernel_size=(3), activation='relu')(model) # if input_shape[0] > 0: model = TimeDistributed(Conv1D(filters=24, kernel_size=3, activation='relu'))(model) model = LSTM(16)(model) model = Activation('relu')(model) model = Dropout(0.2)(model) model = Dense(16)(model) model = Activation('relu')(model) model = BatchNormalization()(model) model = Dense(1)(model) model = Activation('sigmoid')(model) model = Model(inp, model) return model # %% # Conv-1D architecture. Just one sample as input
def test(path_test, input_size, hidden_size, batch_size, save_dir, model_name, maxlen): db = read_data(path_test) X = create_sequences(db, maxlen, maxlen) y = create_sequences(db, maxlen, maxlen) X = np.reshape(X, (X.shape[0], X.shape[1], 1)) y = np.reshape(y, (y.shape[0], y.shape[1], 1)) # build the model: 1 layer LSTM print('Build model...') model = Sequential() # "Encode" the input sequence using an RNN, producing an output of HIDDEN_SIZE # note: in a situation where your input sequences have a variable length, # use input_shape=(None, nb_feature). model.add(LSTM(hidden_size, input_shape=(maxlen, input_size))) # For the decoder's input, we repeat the encoded input for each time step model.add(RepeatVector(maxlen)) # The decoder RNN could be multiple layers stacked or a single layer model.add(LSTM(hidden_size, return_sequences=True)) # For each of step of the output sequence, decide which character should be chosen model.add(TimeDistributed(Dense(1))) model.load_weights(save_dir + model_name) model.compile(loss='mae', optimizer='adam') model.summary() prediction = model.predict(X, batch_size, verbose=1, ) prediction = prediction.flatten() # prediction_container = np.array(prediction).flatten() plt.plot(prediction.flatten()[:4000], label='prediction') plt.plot(y.flatten()[maxlen:4000 + maxlen], label='true') plt.legend() plt.show() store_prediction_and_ground_truth(model)
def train_normal_model(path_train, input_size, hidden_size, batch_size, early_stopping_patience, val_percentage, save_dir, model_name, maxlen): if not os.path.exists(save_dir): os.mkdir(save_dir) db = read_data(path_train) train_x = db[:-140] train_y = db[140:] X = create_sequences(train_x, 140, 140) y = create_sequences(train_y, 140, 140) X = np.reshape(X, (X.shape[0], X.shape[1], 1)) # preparing the callbacks check_pointer = callbacks.ModelCheckpoint(filepath=save_dir + model_name, verbose=1, save_best_only=True) early_stop = callbacks.EarlyStopping(patience=early_stopping_patience, verbose=1) # build the model: 1 layer LSTM print('Build model...') model = Sequential() model.add(LSTM(hidden_size, return_sequences=False, input_shape=(maxlen, input_size))) model.add(Dense(140)) model.compile(loss='mse', optimizer='adam') model.summary() model.fit(X, y, batch_size=batch_size, nb_epoch=100, validation_split=val_percentage, callbacks=[check_pointer, early_stop]) return model
def train_normal_model(path_train, input_size, hidden_size, batch_size, early_stopping_patience, val_percentage, save_dir, model_name, maxlen): if not os.path.exists(save_dir): os.mkdir(save_dir) db = read_data(path_train) train_x = db[:-maxlen] train_y = db[maxlen:] X = create_sequences(train_x, maxlen, maxlen) y = create_sequences(train_y, maxlen, maxlen) X = np.reshape(X, (X.shape[0], X.shape[1], 1)) y = np.reshape(y, (y.shape[0], y.shape[1], 1)) # # preparing the callbacks check_pointer = callbacks.ModelCheckpoint(filepath=save_dir + model_name, verbose=1, save_best_only=True) early_stop = callbacks.EarlyStopping(patience=early_stopping_patience, verbose=1) # build the model: 1 layer LSTM print('Build model...') model = Sequential() # "Encode" the input sequence using an RNN, producing an output of HIDDEN_SIZE # note: in a situation where your input sequences have a variable length, # use input_shape=(None, nb_feature). model.add(LSTM(hidden_size, input_shape=(maxlen, input_size))) # For the decoder's input, we repeat the encoded input for each time step model.add(RepeatVector(maxlen)) # The decoder RNN could be multiple layers stacked or a single layer model.add(LSTM(hidden_size, return_sequences=True)) # For each of step of the output sequence, decide which character should be chosen model.add(TimeDistributed(Dense(1))) model.compile(loss='mae', optimizer='adam') model.summary() model.fit(X, y, batch_size=batch_size, nb_epoch=50, validation_split=val_percentage, callbacks=[check_pointer, early_stop]) return model
