rasa/core/policies/embedding_policy.py
def _create_attn_cell(self,
                          cell: tf.contrib.rnn.RNNCell,
                          embed_utter: tf.Tensor,
                          embed_prev_action: tf.Tensor,
                          real_length: tf.Tensor,
                          embed_for_no_intent: tf.Tensor,
                          embed_for_no_action: tf.Tensor,
                          embed_for_action_listen: tf.Tensor
                          ) -> tf.contrib.rnn.RNNCell:
        """Wrap cell in attention wrapper with given memory."""

        if self.attn_before_rnn:
            # create attention over previous user input
            num_memory_units_before_rnn = self._num_units(embed_utter)
            attn_mech = self._create_attn_mech(embed_utter, real_length)

            # create mask for empty user input not to pay attention to it
            ignore_mask = tf.reduce_all(tf.equal(tf.expand_dims(
                embed_for_no_intent, 0), embed_utter), -1)

            # do not use attention by location before rnn
            attn_shift_range = 0
        else:
            attn_mech = None
            ignore_mask = None
            num_memory_units_before_rnn = None
            attn_shift_range = None

        if self.attn_after_rnn:
            # create attention over previous bot actions
            attn_mech_after_rnn = self._create_attn_mech(embed_prev_action,
                                                         real_length)

            # create mask for empty bot action or action_listen
            # not to pay attention to them
            ignore_mask_listen = tf.logical_or(
                tf.reduce_all(tf.equal(
                    tf.expand_dims(embed_for_no_action, 0),
                    embed_prev_action), -1),
                tf.reduce_all(tf.equal(
                    tf.expand_dims(embed_for_action_listen, 0),
                    embed_prev_action), -1)
            )

            if attn_mech is not None:
                # if there is another attention mechanism,
                # create a list of attention mechanisms
                attn_mech = [attn_mech, attn_mech_after_rnn]
                ignore_mask = [ignore_mask, ignore_mask_listen]
                attn_shift_range = [attn_shift_range, self.attn_shift_range]
            else:
                attn_mech = attn_mech_after_rnn
                ignore_mask = ignore_mask_listen
                attn_shift_range = self.attn_shift_range

            # this particular attention mechanism is unusual
            # in the sense that its calculated attention vector is directly
            # added to cell output, therefore enabling copy mechanism

            # `index_of_attn_to_copy` is used by `TimeAttentionWrapper`,
            # to know which attention to copy
            index_of_attn_to_copy = -1
        else:
            index_of_attn_to_copy = None

        return TimeAttentionWrapper(
            cell=cell,
            attention_mechanism=attn_mech,
            sequence_len=self._dialogue_len,
            attn_shift_range=attn_shift_range,
            sparse_attention=self.sparse_attention,
            rnn_and_attn_inputs_fn=self.rnn_and_attn_inputs_fn,
            ignore_mask=ignore_mask,
            cell_input_fn=lambda inputs, attention: (
                self.cell_input_fn(inputs, attention,
                                   num_memory_units_before_rnn)),
            index_of_attn_to_copy=index_of_attn_to_copy,
            likelihood_fn=lambda emb_1, emb_2: (
                self._tf_sim(emb_1, emb_2, None)),
            tensor_not_to_copy=embed_for_action_listen,
            output_attention=True,
            alignment_history=True
        )
Wrap cell in attention wrapper with given memory.
Wrap cell in attention wrapper with given memory.
Wrap cell in attention wrapper with given memory .
def _create_attn_cell(self,
                          cell: tf.contrib.rnn.RNNCell,
                          embed_utter: tf.Tensor,
                          embed_prev_action: tf.Tensor,
                          real_length: tf.Tensor,
                          embed_for_no_intent: tf.Tensor,
                          embed_for_no_action: tf.Tensor,
                          embed_for_action_listen: tf.Tensor
                          ) -> tf.contrib.rnn.RNNCell:
        if self.attn_before_rnn:
            num_memory_units_before_rnn = self._num_units(embed_utter)
            attn_mech = self._create_attn_mech(embed_utter, real_length)
            ignore_mask = tf.reduce_all(tf.equal(tf.expand_dims(
                embed_for_no_intent, 0), embed_utter), -1)
            attn_shift_range = 0
        else:
            attn_mech = None
            ignore_mask = None
            num_memory_units_before_rnn = None
            attn_shift_range = None
        if self.attn_after_rnn:
            attn_mech_after_rnn = self._create_attn_mech(embed_prev_action,
                                                         real_length)
            ignore_mask_listen = tf.logical_or(
                tf.reduce_all(tf.equal(
                    tf.expand_dims(embed_for_no_action, 0),
                    embed_prev_action), -1),
                tf.reduce_all(tf.equal(
                    tf.expand_dims(embed_for_action_listen, 0),
                    embed_prev_action), -1)
            )
            if attn_mech is not None:
                attn_mech = [attn_mech, attn_mech_after_rnn]
                ignore_mask = [ignore_mask, ignore_mask_listen]
                attn_shift_range = [attn_shift_range, self.attn_shift_range]
            else:
                attn_mech = attn_mech_after_rnn
                ignore_mask = ignore_mask_listen
                attn_shift_range = self.attn_shift_range
            index_of_attn_to_copy = -1
        else:
            index_of_attn_to_copy = None
        return TimeAttentionWrapper(
            cell=cell,
            attention_mechanism=attn_mech,
            sequence_len=self._dialogue_len,
            attn_shift_range=attn_shift_range,
            sparse_attention=self.sparse_attention,
            rnn_and_attn_inputs_fn=self.rnn_and_attn_inputs_fn,
            ignore_mask=ignore_mask,
            cell_input_fn=lambda inputs, attention: (
                self.cell_input_fn(inputs, attention,
                                   num_memory_units_before_rnn)),
            index_of_attn_to_copy=index_of_attn_to_copy,
            likelihood_fn=lambda emb_1, emb_2: (
                self._tf_sim(emb_1, emb_2, None)),
            tensor_not_to_copy=embed_for_action_listen,
            output_attention=True,
            alignment_history=True
        )
def _create_attn_cell ( self , cell : tf . contrib . rnn . RNNCell , embed_utter : tf . Tensor , embed_prev_action : tf . Tensor , real_length : tf . Tensor , embed_for_no_intent : tf . Tensor , embed_for_no_action : tf . Tensor , embed_for_action_listen : tf . Tensor ) -> tf . contrib . rnn . RNNCell : if self . attn_before_rnn : num_memory_units_before_rnn = self . _num_units ( embed_utter ) attn_mech = self . _create_attn_mech ( embed_utter , real_length ) ignore_mask = tf . reduce_all ( tf . equal ( tf . expand_dims ( embed_for_no_intent , 0 ) , embed_utter ) , - 1 ) attn_shift_range = 0 else : attn_mech = None ignore_mask = None num_memory_units_before_rnn = None attn_shift_range = None if self . attn_after_rnn : attn_mech_after_rnn = self . _create_attn_mech ( embed_prev_action , real_length ) ignore_mask_listen = tf . logical_or ( tf . reduce_all ( tf . equal ( tf . expand_dims ( embed_for_no_action , 0 ) , embed_prev_action ) , - 1 ) , tf . reduce_all ( tf . equal ( tf . expand_dims ( embed_for_action_listen , 0 ) , embed_prev_action ) , - 1 ) ) if attn_mech is not None : attn_mech = [ attn_mech , attn_mech_after_rnn ] ignore_mask = [ ignore_mask , ignore_mask_listen ] attn_shift_range = [ attn_shift_range , self . attn_shift_range ] else : attn_mech = attn_mech_after_rnn ignore_mask = ignore_mask_listen attn_shift_range = self . attn_shift_range index_of_attn_to_copy = - 1 else : index_of_attn_to_copy = None return TimeAttentionWrapper ( cell = cell , attention_mechanism = attn_mech , sequence_len = self . _dialogue_len , attn_shift_range = attn_shift_range , sparse_attention = self . sparse_attention , rnn_and_attn_inputs_fn = self . rnn_and_attn_inputs_fn , ignore_mask = ignore_mask , cell_input_fn = lambda inputs , attention : ( self . cell_input_fn ( inputs , attention , num_memory_units_before_rnn ) ) , index_of_attn_to_copy = index_of_attn_to_copy , likelihood_fn = lambda emb_1 , emb_2 : ( self . _tf_sim ( emb_1 , emb_2 , None ) ) , tensor_not_to_copy = embed_for_action_listen , output_attention = True , alignment_history = True )
Wrap cell in attention wrapper with given memory.
Wrap cell in attention wrapper with given memory .
"""Wrap cell in attention wrapper with given memory.""" # create attention over previous user input # create mask for empty user input not to pay attention to it # do not use attention by location before rnn # create attention over previous bot actions # create mask for empty bot action or action_listen # not to pay attention to them # if there is another attention mechanism, # create a list of attention mechanisms # this particular attention mechanism is unusual # in the sense that its calculated attention vector is directly # added to cell output, therefore enabling copy mechanism # `index_of_attn_to_copy` is used by `TimeAttentionWrapper`, # to know which attention to copy
self
cell
tf.contrib.rnn.RNNCell
embed_utter
tf.Tensor
embed_prev_action
tf.Tensor
real_length
tf.Tensor
embed_for_no_intent
tf.Tensor
embed_for_no_action
tf.Tensor
embed_for_action_listen
tf.Tensor