Method run

KVCOMM/graph/graph.py:213–259

Execute nodes topologically for `num_rounds` and return final answers.

(
        self,
        inputs: Any,
        num_rounds:int = 1,
        max_tries: int = 3,
        max_time: int = 600,
    )

                out_node.add_successor(in_node,'temporal')

    def run(
        self,
        inputs: Any,
        num_rounds:int = 1,
        max_tries: int = 3,
        max_time: int = 600,
    ) -> List[Any]:
        """Execute nodes topologically for `num_rounds` and return final answers."""
        for round in range(num_rounds):
            self.construct_spatial_connection()
            self.construct_temporal_connection(round)

            in_degree = {node_id: len(node.spatial_predecessors) for node_id, node in self.nodes.items()}
            zero_in_degree_queue = [node_id for node_id, deg in in_degree.items() if deg == 0]

            while zero_in_degree_queue:
                current_node_id = zero_in_degree_queue.pop(0)
                tries = 0
                while tries < max_tries:
                    try:
                        self.nodes[current_node_id].execute(inputs)                                      
                        break
                    except Exception as e:
                        logger.exception(
                            "Error during execution of node {}: {}",
                            current_node_id,
                            e,
                        )
                    tries += 1
                for successor in self.nodes[current_node_id].spatial_successors:
                    if successor.id not in self.nodes.keys():
                        continue
                    in_degree[successor.id] -= 1
                    if in_degree[successor.id] == 0:
                        zero_in_degree_queue.append(successor.id)

            self.update_memory()
        if self.decision_node:
            self.connect_decision_node()
            self.decision_node.execute(inputs)
            final_answers = self.decision_node.outputs
            if len(final_answers) == 0:
                final_answers.append("No answer of the decision node")
        else:
            final_answers = self.nodes[list(self.nodes.keys())[-1]].outputs

        return final_answers

    async def arun(
        self,