rasa/core/training/structures.py
def with_cycles_removed(self) -> 'StoryGraph':
        """Create a graph with the cyclic edges removed from this graph."""

        story_end_checkpoints = self.story_end_checkpoints.copy()
        cyclic_edge_ids = self.cyclic_edge_ids
        # we need to remove the start steps and replace them with steps ending
        # in a special end checkpoint

        # as in python 3.5, dict is not ordered, in order to generate
        # reproducible result with random seed in python 3.5, we have
        # to use OrderedDict
        if sys.version_info >= (3, 6):
            story_steps = {s.id: s for s in self.story_steps}
        else:
            from collections import OrderedDict
            story_steps = OrderedDict([(s.id, s) for s in self.story_steps])

        # collect all overlapping checkpoints
        # we will remove unused start ones
        all_overlapping_cps = set()

        if self.cyclic_edge_ids:
            # we are going to do this in a recursive way. we are going to
            # remove one cycle and then we are going to
            # let the cycle detection run again
            # this is not inherently necessary so if this becomes a performance
            # issue, we can change it. It is actually enough to run the cycle
            # detection only once and then remove one cycle after another, but
            # since removing the cycle is done by adding / removing edges and
            #  nodes
            # the logic is a lot easier if we only need to make sure the
            # change is consistent if we only change one compared to
            # changing all of them.

            for s, e in cyclic_edge_ids:
                cid = utils.generate_id(max_chars=GENERATED_HASH_LENGTH)
                prefix = GENERATED_CHECKPOINT_PREFIX + CHECKPOINT_CYCLE_PREFIX
                # need abbreviations otherwise they are not visualized well
                sink_cp_name = prefix + "SINK_" + cid
                connector_cp_name = prefix + "CONN_" + cid
                source_cp_name = prefix + "SRC_" + cid
                story_end_checkpoints[sink_cp_name] = source_cp_name

                overlapping_cps = self.overlapping_checkpoint_names(
                    story_steps[s].end_checkpoints,
                    story_steps[e].start_checkpoints)

                all_overlapping_cps.update(overlapping_cps)

                # change end checkpoints of starts
                start = story_steps[s].create_copy(use_new_id=False)
                start.end_checkpoints = [cp
                                         for cp in start.end_checkpoints
                                         if cp.name not in overlapping_cps]
                start.end_checkpoints.append(Checkpoint(sink_cp_name))
                story_steps[s] = start

                needs_connector = False

                for k, step in list(story_steps.items()):
                    additional_ends = []
                    for original_cp in overlapping_cps:
                        for cp in step.start_checkpoints:
                            if cp.name == original_cp:
                                if k == e:
                                    cp_name = source_cp_name
                                else:
                                    cp_name = connector_cp_name
                                    needs_connector = True

                                if not self._is_checkpoint_in_list(
                                        cp_name, cp.conditions,
                                        step.start_checkpoints):
                                    # add checkpoint only if it was not added
                                    additional_ends.append(
                                        Checkpoint(cp_name, cp.conditions))

                    if additional_ends:
                        updated = step.create_copy(use_new_id=False)
                        updated.start_checkpoints.extend(additional_ends)
                        story_steps[k] = updated

                if needs_connector:
                    start.end_checkpoints.append(Checkpoint(connector_cp_name))

        # the process above may generate unused checkpoints
        # we need to find them and remove them
        self._remove_unused_generated_cps(story_steps,
                                          all_overlapping_cps,
                                          story_end_checkpoints)

        return StoryGraph(list(story_steps.values()),
                          story_end_checkpoints)
Create a graph with the cyclic edges removed from this graph.
Create a graph with the cyclic edges removed from this graph.
Create a graph with the cyclic edges removed from this graph .
def with_cycles_removed(self) -> 'StoryGraph':
        story_end_checkpoints = self.story_end_checkpoints.copy()
        cyclic_edge_ids = self.cyclic_edge_ids
        if sys.version_info >= (3, 6):
            story_steps = {s.id: s for s in self.story_steps}
        else:
            from collections import OrderedDict
            story_steps = OrderedDict([(s.id, s) for s in self.story_steps])
        all_overlapping_cps = set()
        if self.cyclic_edge_ids:
            for s, e in cyclic_edge_ids:
                cid = utils.generate_id(max_chars=GENERATED_HASH_LENGTH)
                prefix = GENERATED_CHECKPOINT_PREFIX + CHECKPOINT_CYCLE_PREFIX
                sink_cp_name = prefix + "SINK_" + cid
                connector_cp_name = prefix + "CONN_" + cid
                source_cp_name = prefix + "SRC_" + cid
                story_end_checkpoints[sink_cp_name] = source_cp_name
                overlapping_cps = self.overlapping_checkpoint_names(
                    story_steps[s].end_checkpoints,
                    story_steps[e].start_checkpoints)
                all_overlapping_cps.update(overlapping_cps)
                start = story_steps[s].create_copy(use_new_id=False)
                start.end_checkpoints = [cp
                                         for cp in start.end_checkpoints
                                         if cp.name not in overlapping_cps]
                start.end_checkpoints.append(Checkpoint(sink_cp_name))
                story_steps[s] = start
                needs_connector = False
                for k, step in list(story_steps.items()):
                    additional_ends = []
                    for original_cp in overlapping_cps:
                        for cp in step.start_checkpoints:
                            if cp.name == original_cp:
                                if k == e:
                                    cp_name = source_cp_name
                                else:
                                    cp_name = connector_cp_name
                                    needs_connector = True
                                if not self._is_checkpoint_in_list(
                                        cp_name, cp.conditions,
                                        step.start_checkpoints):
                                    additional_ends.append(
                                        Checkpoint(cp_name, cp.conditions))
                    if additional_ends:
                        updated = step.create_copy(use_new_id=False)
                        updated.start_checkpoints.extend(additional_ends)
                        story_steps[k] = updated
                if needs_connector:
                    start.end_checkpoints.append(Checkpoint(connector_cp_name))
        self._remove_unused_generated_cps(story_steps,
                                          all_overlapping_cps,
                                          story_end_checkpoints)
        return StoryGraph(list(story_steps.values()),
                          story_end_checkpoints)
def with_cycles_removed ( self ) -> 'StoryGraph' : story_end_checkpoints = self . story_end_checkpoints . copy ( ) cyclic_edge_ids = self . cyclic_edge_ids if sys . version_info >= ( 3 , 6 ) : story_steps = { s . id : s for s in self . story_steps } else : from collections import OrderedDict story_steps = OrderedDict ( [ ( s . id , s ) for s in self . story_steps ] ) all_overlapping_cps = set ( ) if self . cyclic_edge_ids : for s , e in cyclic_edge_ids : cid = utils . generate_id ( max_chars = GENERATED_HASH_LENGTH ) prefix = GENERATED_CHECKPOINT_PREFIX + CHECKPOINT_CYCLE_PREFIX sink_cp_name = prefix + "SINK_" + cid connector_cp_name = prefix + "CONN_" + cid source_cp_name = prefix + "SRC_" + cid story_end_checkpoints [ sink_cp_name ] = source_cp_name overlapping_cps = self . overlapping_checkpoint_names ( story_steps [ s ] . end_checkpoints , story_steps [ e ] . start_checkpoints ) all_overlapping_cps . update ( overlapping_cps ) start = story_steps [ s ] . create_copy ( use_new_id = False ) start . end_checkpoints = [ cp for cp in start . end_checkpoints if cp . name not in overlapping_cps ] start . end_checkpoints . append ( Checkpoint ( sink_cp_name ) ) story_steps [ s ] = start needs_connector = False for k , step in list ( story_steps . items ( ) ) : additional_ends = [ ] for original_cp in overlapping_cps : for cp in step . start_checkpoints : if cp . name == original_cp : if k == e : cp_name = source_cp_name else : cp_name = connector_cp_name needs_connector = True if not self . _is_checkpoint_in_list ( cp_name , cp . conditions , step . start_checkpoints ) : additional_ends . append ( Checkpoint ( cp_name , cp . conditions ) ) if additional_ends : updated = step . create_copy ( use_new_id = False ) updated . start_checkpoints . extend ( additional_ends ) story_steps [ k ] = updated if needs_connector : start . end_checkpoints . append ( Checkpoint ( connector_cp_name ) ) self . _remove_unused_generated_cps ( story_steps , all_overlapping_cps , story_end_checkpoints ) return StoryGraph ( list ( story_steps . values ( ) ) , story_end_checkpoints )
Create a graph with the cyclic edges removed from this graph.
Create a graph with the cyclic edges removed from this graph .
"""Create a graph with the cyclic edges removed from this graph.""" # we need to remove the start steps and replace them with steps ending # in a special end checkpoint # as in python 3.5, dict is not ordered, in order to generate # reproducible result with random seed in python 3.5, we have # to use OrderedDict # collect all overlapping checkpoints # we will remove unused start ones # we are going to do this in a recursive way. we are going to # remove one cycle and then we are going to # let the cycle detection run again # this is not inherently necessary so if this becomes a performance # issue, we can change it. It is actually enough to run the cycle # detection only once and then remove one cycle after another, but # since removing the cycle is done by adding / removing edges and #  nodes # the logic is a lot easier if we only need to make sure the # change is consistent if we only change one compared to # changing all of them. # need abbreviations otherwise they are not visualized well # change end checkpoints of starts # add checkpoint only if it was not added # the process above may generate unused checkpoints # we need to find them and remove them
self