我们从Python开源项目中，提取了以下27个代码示例，用于说明如何使用keras.backend.rnn()。
def _forward(x, reduce_step, initial_states, U, mask=None): '''Forward recurrence of the linear chain crf.''' def _forward_step(energy_matrix_t, states): alpha_tm1 = states[-1] new_states = reduce_step(K.expand_dims(alpha_tm1, 2) + energy_matrix_t) return new_states[0], new_states U_shared = K.expand_dims(K.expand_dims(U, 0), 0) if mask is not None: mask = K.cast(mask, K.floatx()) mask_U = K.expand_dims(K.expand_dims(mask[:, :-1] * mask[:, 1:], 2), 3) U_shared = U_shared * mask_U inputs = K.expand_dims(x[:, 1:, :], 2) + U_shared inputs = K.concatenate([inputs, K.zeros_like(inputs[:, -1:, :, :])], axis=1) last, values, _ = K.rnn(_forward_step, inputs, initial_states) return last, values
def _backward(gamma, mask): '''Backward recurrence of the linear chain crf.''' gamma = K.cast(gamma, 'int32') def _backward_step(gamma_t, states): y_tm1 = K.squeeze(states[0], 0) y_t = batch_gather(gamma_t, y_tm1) return y_t, [K.expand_dims(y_t, 0)] initial_states = [K.expand_dims(K.zeros_like(gamma[:, 0, 0]), 0)] _, y_rev, _ = K.rnn(_backward_step, gamma, initial_states, go_backwards=True) y = K.reverse(y_rev, 1) if mask is not None: mask = K.cast(mask, dtype='int32') # mask output y *= mask # set masked values to -1 y += -(1 - mask) return y
def _backward(gamma, mask): '''Backward recurrence of the linear chain crf.''' gamma = K.cast(gamma, 'int32') def _backward_step(gamma_t, states): y_tm1 = K.squeeze(states[0], 0) y_t = KC.batch_gather(gamma_t, y_tm1) return y_t, [K.expand_dims(y_t, 0)] initial_states = [K.expand_dims(K.zeros_like(gamma[:, 0, 0]), 0)] _, y_rev, _ = K.rnn(_backward_step, gamma, initial_states, go_backwards=True) y = K.reverse(y_rev, 1) if mask is not None: mask = K.cast(mask, dtype='int32') # mask output y *= mask # set masked values to -1 y += -(1 - mask) return y
def get_output(self, train=False): X = self.get_input(train) def out_step(X_i, states): def in_step(x, in_states): output = K.dot(x, self.W) + self.b return output, [] _, in_outputs, _ = K.rnn(in_step, X_i, initial_states=[], mask=None) return in_outputs, [] _, outputs, _ = K.rnn(out_step, X, initial_states=[], mask=None) outputs = self.activation(outputs) return outputs
def call(self, x, mask=None): print("AttentionDecoder.call") H = x x = K.permute_dimensions(H, (1, 0, 2))[-1, :, :] if self.stateful or self.state_input or len(self.state_outputs) > 0: initial_states = self.states[:] else: initial_states = self.get_initial_states(H) constants = self.get_constants(H) + [H] y_0 = x x = K.repeat(x, self.output_length) initial_states += [y_0] last_output, outputs, states = K.rnn( self.step, x, initial_states, go_backwards=self.go_backwards, mask=mask, constants=constants, unroll=self.unroll, input_length=self.output_length) if self.stateful and not self.state_input: self.updates = zip(self.states, states) self.states_to_transfer = states return outputs
def _forward(x, reduce_step, initial_states, U, mask=None): """Forward recurrence of the linear chain crf.""" def _forward_step(energy_matrix_t, states): alpha_tm1 = states[-1] new_states = reduce_step(K.expand_dims(alpha_tm1, 2) + energy_matrix_t) return new_states[0], new_states U_shared = K.expand_dims(K.expand_dims(U, 0), 0) if mask is not None: mask = K.cast(mask, K.floatx()) mask_U = K.expand_dims(K.expand_dims(mask[:, :-1] * mask[:, 1:], 2), 3) U_shared = U_shared * mask_U inputs = K.expand_dims(x[:, 1:, :], 2) + U_shared inputs = K.concatenate([inputs, K.zeros_like(inputs[:, -1:, :, :])], axis=1) last, values, _ = K.rnn(_forward_step, inputs, initial_states) return last, values
def _backward(gamma, mask): """Backward recurrence of the linear chain crf.""" gamma = K.cast(gamma, 'int32') def _backward_step(gamma_t, states): y_tm1 = K.squeeze(states[0], 0) y_t = batch_gather(gamma_t, y_tm1) return y_t, [K.expand_dims(y_t, 0)] initial_states = [K.expand_dims(K.zeros_like(gamma[:, 0, 0]), 0)] _, y_rev, _ = K.rnn(_backward_step, gamma, initial_states, go_backwards=True) y = K.reverse(y_rev, 1) if mask is not None: mask = K.cast(mask, dtype='int32') # mask output y *= mask # set masked values to -1 y += -(1 - mask) return y
def call(self, inputs, mask=None, initial_state=None, training=None): input_shape = K.int_shape(inputs) constants = self.layer.get_constants(inputs, training=None) preprocessed_input = self.layer.preprocess_input(inputs, training=None) initial_states = self.layer.get_initial_states(inputs) last_output, outputs, states = K.rnn(self.step, preprocessed_input, initial_states, go_backwards=self.layer.go_backwards, mask=mask, constants=constants, unroll=self.layer.unroll, input_length=input_shape[1]) return outputs if self.layer.return_sequences else last_output
def step(self, input_energy_t, states, return_logZ=True): # not in the following `prev_target_val` has shape = (B, F) # where B = batch_size, F = output feature dim # Note: `i` is of float32, due to the behavior of `K.rnn` prev_target_val, i, chain_energy = states[:3] t = K.cast(i[0, 0], dtype='int32') if len(states) > 3: if K.backend() == 'theano': m = states[3][:, t:(t + 2)] else: m = K.tf.slice(states[3], [0, t], [-1, 2]) input_energy_t = input_energy_t * K.expand_dims(m[:, 0]) chain_energy = chain_energy * K.expand_dims(K.expand_dims(m[:, 0] * m[:, 1])) # (1, F, F)*(B, 1, 1) -> (B, F, F) if return_logZ: energy = chain_energy + K.expand_dims(input_energy_t - prev_target_val, 2) # shapes: (1, B, F) + (B, F, 1) -> (B, F, F) new_target_val = K.logsumexp(-energy, 1) # shapes: (B, F) return new_target_val, [new_target_val, i + 1] else: energy = chain_energy + K.expand_dims(input_energy_t + prev_target_val, 2) min_energy = K.min(energy, 1) argmin_table = K.cast(K.argmin(energy, 1), K.floatx()) # cast for tf-version `K.rnn` return argmin_table, [min_energy, i + 1]
def call(self, x, mask=None): input_shape = self.input_spec[0].shape en_seq = x x_input = x[:, input_shape[1]-1, :] x_input = K.repeat(x_input, input_shape[1]) initial_states = self.get_initial_states(x_input) constants = super(PointerLSTM, self).get_constants(x_input) constants.append(en_seq) preprocessed_input = self.preprocess_input(x_input) last_output, outputs, states = K.rnn(self.step, preprocessed_input, initial_states, go_backwards=self.go_backwards, constants=constants, input_length=input_shape[1]) return outputs
def step(self, input_t, states): ''' This method is a step function that updates the memory at each time step and produces a new output vector (Equations 1 to 6 in the paper). The memory_state is flattened because K.rnn requires all states to be of the same shape as the output, because it uses the same mask for the output and the states. Inputs: input_t (batch_size, input_dim) states (list[Tensor]) flattened_mem_tm1 (batch_size, input_length * output_dim) writer_h_tm1 (batch_size, output_dim) writer_c_tm1 (batch_size, output_dim) Outputs: h_t (batch_size, output_dim) flattened_mem_t (batch_size, input_length * output_dim) ''' reader_states, flattened_mem_tm1, writer_states = self.split_states(states) input_mem_shape = K.shape(flattened_mem_tm1) mem_tm1_shape = (input_mem_shape[0], input_mem_shape[1]/self.output_dim, self.output_dim) mem_tm1 = K.reshape(flattened_mem_tm1, mem_tm1_shape) # (batch_size, input_length, output_dim) reader_constants = self.reader.get_constants(input_t) # Does not depend on input_t, see init. reader_states = reader_states[:2] + reader_constants + reader_states[2:] o_t, [_, reader_c_t] = self.reader.step(input_t, reader_states) # o_t, reader_c_t: (batch_size, output_dim) z_t, m_rt = self.summarize_memory(o_t, mem_tm1) c_t = self.compose_memory_and_output([o_t, m_rt]) # Collecting the necessary variables to directly call writer's step function. writer_constants = self.writer.get_constants(c_t) # returns dropouts for W and U (all 1s, see init) writer_states += writer_constants # Making a call to writer's step function, Equation 5 h_t, [_, writer_c_t] = self.writer.step(c_t, writer_states) # h_t, writer_c_t: (batch_size, output_dim) mem_t = self.update_memory(z_t, h_t, mem_tm1) flattened_mem_t = K.batch_flatten(mem_t) return h_t, [o_t, reader_c_t, flattened_mem_t, h_t, writer_c_t]
def loop(self, x, initial_states, mask): # This is a separate method because Ontoaware variants will have to override this to make a call # to changingdim rnn. last_output, all_outputs, last_states = K.rnn(self.step, x, initial_states, mask=mask) return last_output, all_outputs, last_states
def call(self, x, mask=None): # input shape: (nb_samples, time (padded with zeros), input_dim) # note that the .build() method of subclasses MUST define # self.input_spec with a complete input shape. input_shape = self.input_spec[0].shape if K._BACKEND == 'tensorflow': if not input_shape[1]: raise Exception('When using TensorFlow, you should define ' 'explicitly the number of timesteps of ' 'your sequences.\n' 'If your first layer is an Embedding, ' 'make sure to pass it an "input_length" ' 'argument. Otherwise, make sure ' 'the first layer has ' 'an "input_shape" or "batch_input_shape" ' 'argument, including the time axis. ' 'Found input shape at layer ' + self.name + ': ' + str(input_shape)) if self.layer.stateful: initial_states = self.layer.states else: initial_states = self.layer.get_initial_states(x) constants = self.get_constants(x) preprocessed_input = self.layer.preprocess_input(x) last_output, outputs, states = K.rnn(self.step, preprocessed_input, initial_states, go_backwards=self.layer.go_backwards, mask=mask, constants=constants, unroll=self.layer.unroll, input_length=input_shape[1]) if self.layer.stateful: self.updates = [] for i in range(len(states)): self.updates.append((self.layer.states[i], states[i])) if self.layer.return_sequences: return outputs else: return last_output
def call(self, x, mask=None): # input shape: (nb_samples, time (padded with zeros), input_dim) # note that the .build() method of subclasses MUST define # self.input_spec with a complete input shape. input_shape = self.input_spec[0].shape if K._BACKEND == 'tensorflow': if not input_shape[1]: raise Exception('When using TensorFlow, you should define ' 'explicitly the number of timesteps of ' 'your sequences.\n' 'If your first layer is an Embedding, ' 'make sure to pass it an "input_length" ' 'argument. Otherwise, make sure ' 'the first layer has ' 'an "input_shape" or "batch_input_shape" ' 'argument, including the time axis. ' 'Found input shape at layer ' + self.name + ': ' + str(input_shape)) if self.layer.stateful: initial_states = self.layer.states else: initial_states = self.layer.get_initial_states(x) constants = self.get_constants(x) preprocessed_input = self.layer.preprocess_input(x) last_output, outputs, states = K.rnn(self.step, preprocessed_input, initial_states, go_backwards=self.layer.go_backwards, mask=mask, constants=constants, unroll=self.layer.unroll, input_length=input_shape[1]) if self.layer.stateful: self.updates = [] for i in range(len(states)): self.updates.append((self.layer.states[i], states[i])) if self.layer.return_sequences: return outputs else:
def call(self, x, mask=None): input_shape = self.input_spec[0].shape if self.layer.unroll and input_shape[1] is None: raise ValueError('Cannot unroll a RNN if the ' 'time dimension is undefined. \n' '- If using a Sequential model, ' 'specify the time dimension by passing ' 'an `input_shape` or `batch_input_shape` ' 'argument to your first layer. If your ' 'first layer is an Embedding, you can ' 'also use the `input_length` argument.\n' '- If using the functional API, specify ' 'the time dimension by passing a `shape` ' 'or `batch_shape` argument to your Input layer.') initial_states = (self.layer.states if self.layer.stateful else self.layer.get_initial_states(x)) constants = self.get_constants(x) preprocessed_input = self.layer.preprocess_input(x) last_output, outputs, states = K.rnn( self.step, preprocessed_input, initial_states, go_backwards=self.layer.go_backwards, mask=mask, constants=constants, unroll=self.layer.unroll, input_length=input_shape[1]) if self.layer.stateful: updates = [] for i in range(len(states)): updates.append((self.layer.states[i], states[i])) self.add_update(updates, x) return outputs if self.layer.return_sequences else last_output
def call(self, x, mask=None): input_shape = self.input_spec[0].shape if self.layer.unroll and input_shape[1] is None: raise ValueError('Cannot unroll a RNN if the ' 'time dimension is undefined. \n' '- If using a Sequential model, ' 'specify the time dimension by passing ' 'an `input_shape` or `batch_input_shape` ' 'argument to your first layer. If your ' 'first layer is an Embedding, you can ' 'also use the `input_length` argument.\n' '- If using the functional API, specify ' 'the time dimension by passing a `shape` ' 'or `batch_shape` argument to your Input layer.') initial_states = (self.layer.states if self.layer.stateful else self.layer.get_initial_states(x)) constants = self.layer.get_constants(x) preprocessed_input = self.layer.preprocess_input(x) last_output, outputs, states = K.rnn( self.step, preprocessed_input, initial_states, go_backwards=self.layer.go_backwards, mask=mask, constants=constants, unroll=self.layer.unroll, input_length=input_shape[1]) if self.layer.stateful: updates = [] for i in range(len(states)): updates.append((self.layer.states[i], states[i])) self.add_update(updates, x) return outputs if self.layer.return_sequences else last_output
def broadcast_state(self, rnns): if type(rnns) not in [list, tuple]: rnns = [rnns] self.state_outputs += rnns for rnn in rnns: rnn.state_input = self
def get_output(self, train=False): # input shape: (nb_samples, time (padded with zeros), input_dim) X = self.get_input(train) if K._BACKEND == 'tensorflow': if not self.input_shape[1]: raise Exception('When using TensorFlow, you should define ' + 'explicitly the number of timesteps of ' + 'your sequences. Make sure the first layer ' + 'has a "batch_input_shape" argument ' + 'including the samples axis.') mask = self.get_output_mask(train) if self.stateful or self.state_input or len(self.state_outputs) > 0: initial_states = self.states else: initial_states = self.get_initial_states(X) last_output, outputs, states = K.rnn(self.step, X, initial_states, go_backwards=self.go_backwards, masking=mask) n = len(states) if self.stateful and not self.state_input: self.updates = [] self.updates = [] for i in range(n): self.updates.append((self.states[i], states[i])) for o in self.state_outputs: o.updates = [] for i in range(n): o.updates.append((o.states[i], states[i])) if self.return_sequences: return outputs else: return last_output
def call(self, x, mask=None): constants = self.get_constants(x) assert K.ndim(x) == 5 if K._BACKEND == 'tensorflow': if not self.input_shape[1]: raise Exception('When using TensorFlow, you should define ' + 'explicitely the number of timesteps of ' + 'your sequences. Make sure the first layer ' + 'has a "batch_input_shape" argument ' + 'including the samples axis.') if self.stateful: initial_states = self.states else: initial_states = self.get_initial_states(x) last_output, outputs, states = K.rnn(self.step, x, initial_states, go_backwards=self.go_backwards, mask=mask, constants=constants) if self.stateful: self.updates = [] for i in range(len(states)): self.updates.append((self.states[i], states[i])) if self.return_sequences: return outputs else: return last_output
def call(self, x, mask=None): X = K.repeat(x, self.output_length) input_shape = list(self.input_spec[0].shape) input_shape = input_shape[:1] + [self.output_length] + input_shape[1:] self.input_spec = [InputSpec(shape=tuple(input_shape))] if self.stateful or self.state_input or len(self.state_outputs) > 0: initial_states = self.states[:] else: initial_states = self.get_initial_states(X) constants = self.get_constants(X) y_0 = K.permute_dimensions(X, (1, 0, 2))[0, :, :] initial_states += [y_0] last_output, outputs, states = K.rnn(self.step, X, initial_states, go_backwards=self.go_backwards, mask=mask, constants=constants, unroll=self.unroll, input_length=self.output_length) if self.stateful and not self.state_input: self.updates = [] for i in range(2): self.updates.append((self.states[i], states[i])) self.states_to_transfer = states input_shape.pop(1) self.input_spec = [InputSpec(shape=input_shape)] return outputs
def viterbi_decoding(self, X, mask=None): input_energy = self.activation(K.dot(X, self.kernel) + self.bias) if self.use_boundary: input_energy = self.add_boundary_energy(input_energy, mask, self.left_boundary, self.right_boundary) argmin_tables = self.recursion(input_energy, mask, return_logZ=False) argmin_tables = K.cast(argmin_tables, 'int32') # backward to find best path, `initial_best_idx` can be any, as all elements in the last argmin_table are the same argmin_tables = K.reverse(argmin_tables, 1) initial_best_idx = [K.expand_dims(argmin_tables[:, 0, 0])] # matrix instead of vector is required by tf `K.rnn` if K.backend() == 'theano': initial_best_idx = [K.T.unbroadcast(initial_best_idx[0], 1)] def gather_each_row(params, indices): n = K.shape(indices)[0] if K.backend() == 'theano': return params[K.T.arange(n), indices] else: indices = K.transpose(K.stack([K.tf.range(n), indices])) return K.tf.gather_nd(params, indices) def find_path(argmin_table, best_idx): next_best_idx = gather_each_row(argmin_table, best_idx[0][:, 0]) next_best_idx = K.expand_dims(next_best_idx) if K.backend() == 'theano': next_best_idx = K.T.unbroadcast(next_best_idx, 1) return next_best_idx, [next_best_idx] _, best_paths, _ = K.rnn(find_path, argmin_tables, initial_best_idx, input_length=K.int_shape(X)[1], unroll=self.unroll) best_paths = K.reverse(best_paths, 1) best_paths = K.squeeze(best_paths, 2) return K.one_hot(best_paths, self.units)
def call(self, x, mask=None): input_shape = self.input_spec[0].shape # state format: [h(t-1), c(t-1), y(t-1)] #h_0 = K.zeros_like(x[:, 0, :]) #c_0 = K.zeros_like(x[:, 0, :]) h_0 = K.reshape(x, (-1, self.input_dim)) c_0 = K.reshape(x, (-1, self.input_dim)) initial_states = [h_0, c_0] #self.states = [None, None] #initial_states = self.get_initial_states(x) last_output, outputs, states = K.rnn(step_function=self.step, inputs=x, initial_states=initial_states, go_backwards=self.go_backwards, mask=mask, constants=None, unroll=self.unroll, input_length=input_shape[1]) if self.return_sequences: return outputs else: return last_output
def call(self, inputs, mask=None, initial_state=None, training=None): # input shape: `(samples, time (padded with zeros), input_dim)` # note that the .build() method of subclasses MUST define # self.input_spec and self.state_spec with complete input shapes. if isinstance(inputs, list): initial_states = inputs[1:] inputs = inputs[0] elif initial_state is not None: pass elif self.stateful: initial_states = self.states else: initial_states = self.get_initial_states(inputs) if len(initial_states) != len(self.states): raise ValueError('Layer has ' + str(len(self.states)) + ' states but was passed ' + str(len(initial_states)) + ' initial states.') input_shape = K.int_shape(inputs) if self.unroll and input_shape[1] is None: raise ValueError('Cannot unroll a RNN if the ' 'time dimension is undefined. \n' '- If using a Sequential model, ' 'specify the time dimension by passing ' 'an `input_shape` or `batch_input_shape` ' 'argument to your first layer. If your ' 'first layer is an Embedding, you can ' 'also use the `input_length` argument.\n' '- If using the functional API, specify ' 'the time dimension by passing a `shape` ' 'or `batch_shape` argument to your Input layer.') constants = self.get_constants(inputs, training=None) preprocessed_input = self.preprocess_input(inputs, training=None) last_output, outputs, states = K.rnn(self.step, preprocessed_input, initial_states, go_backwards=self.go_backwards, mask=mask, constants=constants, unroll=self.unroll, input_length=input_shape[1]) if self.stateful: updates = [] for i in range(len(states)): updates.append((self.states[i], states[i])) self.add_update(updates, inputs) # Properly set learning phase if 0 < self.dropout < 1: last_output._uses_learning_phase = True outputs._uses_learning_phase = True if self.return_sequences: return outputs else: return last_output
def get_state_transfer_rnn(RNN): '''Converts a given Recurrent sub class (e.g, LSTM, GRU) to its state transferable version. A state transfer RNN can transfer its hidden state to another one of the same type and compatible dimensions. ''' class StateTransferRNN(RNN): def __init__(self, state_input=True, **kwargs): self.state_outputs = [] self.state_input = state_input super(StateTransferRNN, self).__init__(**kwargs) def reset_states(self): stateful = self.stateful self.stateful = stateful or self.state_input or len(self.state_outputs) > 0 if self.stateful: super(StateTransferRNN, self).reset_states() self.stateful = stateful def build(self,input_shape): stateful = self.stateful self.stateful = stateful or self.state_input or len(self.state_outputs) > 0 super(StateTransferRNN, self).build(input_shape) self.stateful = stateful def broadcast_state(self, rnns): rnns = (set if type(rnns) in [list, tuple] else lambda a: {a})(rnns) rnns -= set(self.state_outputs) self.state_outputs.extend(rnns) for rnn in rnns: rnn.state_input = self rnn.updates = getattr(rnn, 'updates', []) rnn.updates.extend(zip(rnn.states, self.states_to_transfer)) def call(self, x, mask=None): last_output, outputs, states = K.rnn( self.step, self.preprocess_input(x), self.states or self.get_initial_states(x), go_backwards=self.go_backwards, mask=mask, constants=self.get_constants(x), unroll=self.unroll, input_length=self.input_spec[0].shape[1]) self.updates = zip(self.states, states) self.states_to_transfer = states return outputs if self.return_sequences else last_output return StateTransferRNN
def recursion(self, input_energy, mask=None, go_backwards=False, return_sequences=True, return_logZ=True, input_length=None): """Forward (alpha) or backward (beta) recursion If `return_logZ = True`, compute the logZ, the normalization constance: \[ Z = \sum_{y1, y2, y3} exp(-E) # energy = \sum_{y1, y2, y3} exp(-(u1' y1 + y1' W y2 + u2' y2 + y2' W y3 + u3' y3)) = sum_{y2, y3} (exp(-(u2' y2 + y2' W y3 + u3' y3)) sum_{y1} exp(-(u1' y1' + y1' W y2))) \] Denote: \[ S(y2) := sum_{y1} exp(-(u1' y1 + y1' W y2)), \] \[ Z = sum_{y2, y3} exp(log S(y2) - (u2' y2 + y2' W y3 + u3' y3)) \] \[ logS(y2) = log S(y2) = log_sum_exp(-(u1' y1' + y1' W y2)) \] Note that: yi's are one-hot vectors u1, u3: boundary energies have been merged If `return_logZ = False`, compute the Viterbi's best path lookup table. """ chain_energy = self.chain_kernel chain_energy = K.expand_dims(chain_energy, 0) # shape=(1, F, F): F=num of output features. 1st F is for t-1, 2nd F for t prev_target_val = K.zeros_like(input_energy[:, 0, :]) # shape=(B, F), dtype=float32 if go_backwards: input_energy = K.reverse(input_energy, 1) if mask is not None: mask = K.reverse(mask, 1) initial_states = [prev_target_val, K.zeros_like(prev_target_val[:, :1])] constants = [chain_energy] if mask is not None: mask2 = K.cast(K.concatenate([mask, K.zeros_like(mask[:, :1])], axis=1), K.floatx()) constants.append(mask2) def _step(input_energy_i, states): return self.step(input_energy_i, states, return_logZ) target_val_last, target_val_seq, _ = K.rnn(_step, input_energy, initial_states, constants=constants, input_length=input_length, unroll=self.unroll) if return_sequences: if go_backwards: target_val_seq = K.reverse(target_val_seq, 1) return target_val_seq else: return target_val_last
