Method arun

KVCOMM/graph/graph.py:261–418

Asynchronous execution entry supporting standard and KV-reuse workflows.

(
        self,
        input: Dict[str, str],
        num_rounds: int = 1,
        max_tries: int = 3,
        max_time: int = 600,
        *,
        mode: str = "default",
        **kwargs,
    )

        return final_answers

    async def arun(
        self,
        input: Dict[str, str],
        num_rounds: int = 1,
        max_tries: int = 3,
        max_time: int = 600,
        *,
        mode: str = "default",
        **kwargs,
    ) -> List[Any]:
        """Asynchronous execution entry supporting standard and KV-reuse workflows."""
        if mode == "default":
            request_uid = input.setdefault(
                "_request_uid", shortuuid.ShortUUID().random(length=8)
            )
            metrics_recorder.start_request(
                request_uid=request_uid,
                batch_index=input.get("_batch_index"),
                task=input.get("task"),
                execution_mode=mode,
            )
            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:
                        await asyncio.wait_for(
                            self.nodes[current_node_id].async_execute(input),
                            timeout=max_time,
                        )
                        break
                    for successor in self.nodes[current_node_id].spatial_successors:
                        if successor.id not in self.nodes:
                            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()
                await self.decision_node.async_execute(input)
                final_answers = self.decision_node.outputs
                if len(final_answers) == 0:
                    final_answers.append("No answer of the decision node")
                metrics_recorder.finalize_request(request_uid)
            else: