Method async_execute

KVCOMM/graph/node.py:167–259

Execute a node asynchronously with optional KV cache handling modes.

(self, input:Any, mode: str = "default", **kwargs)

        return self.outputs

    async def async_execute(self, input:Any, mode: str = "default", **kwargs):
        """Execute a node asynchronously with optional KV cache handling modes."""
        if mode == "default":
            self.outputs = []
            spatial_info:Dict[str,Any] = self.get_spatial_info()
            temporal_info:Dict[str,Any] = self.get_temporal_info()
            tasks = [
                asyncio.create_task(
                    self._async_execute(input, spatial_info, temporal_info, mode=mode, **kwargs)
                )
            ]
            results = await asyncio.gather(*tasks, return_exceptions=False)
            request_uid = input.get("_request_uid") or kwargs.get("request_uid")
            for result in results:
                if not isinstance(result, list):
                    result = [result]
                for item in result:
                    if isinstance(item, GenerationResult):
                        generation = item
                    else:
                        generation = GenerationResult(
                            text=str(item),
                            mode="default",
                            ttft=0.0,
                        )
                    self.outputs.append(generation.text)
                    if request_uid:
                        metrics_recorder.record_agent_output(
                            request_uid=request_uid,
                            agent_id=self.id,
                            agent_name=self.agent_name,
                            agent_role=self.role,
                            generation=generation,
                        )
            return self.outputs

        if mode == "allow_kv_reuse":
            self.outputs = {}
            spatial_info:Dict[str,Any] = self.get_spatial_info(input['task'])
            temporal_info:Dict[str,Any] = self.get_temporal_info(input['task'])
            tasks = [
                asyncio.create_task(
                    self._async_execute(
                        input,
                        spatial_info,
                        temporal_info,
                        mode=mode,
                        **kwargs,
                    )
                )
            ]
            results = await asyncio.gather(*tasks, return_exceptions=False)
            request_uid = input.get("_request_uid") or kwargs.get("request_uid")
            for (message, result) in results:
                if not isinstance(result, list):
                    result = [result]
                if message not in self.outputs:
                    self.outputs[message] = []