Update API

dff/pipeline/pipeline/actor.py

@@ -16,13 +16,9 @@
 
 Overall, the actor acts as a bridge between the user's input and the dialog graph,
 making sure that the conversation follows the expected flow and providing a personalized experience to the user.
-
-Below you can see a diagram of user request processing with Actor.
-Both `request` and `response` are saved to :py:class:`.Context`.
-
-.. figure:: /_static/drawio/dfe/user_actor.png
 """
 import logging
+import asyncio
 from typing import Union, Callable, Optional, Dict, List, Any, ForwardRef
 import copy
 
@@ -34,6 +30,7 @@
 from dff.script.core.script import Script, Node
 from dff.script.core.normalization import normalize_label, normalize_response
 from dff.script.core.keywords import GLOBAL, LOCAL
+from ..service.utils import wrap_sync_function_in_async
 
 logger = logging.getLogger(__name__)
 
@@ -109,51 +106,51 @@ def __init__(
         # NB! The following API is highly experimental and may be removed at ANY time WITHOUT FURTHER NOTICE!!
         self._clean_turn_cache = True
 
-    def __call__(
+    async def __call__(
         self, pipeline: Pipeline, ctx: Optional[Union[Context, dict, str]] = None, *args, **kwargs
     ) -> Union[Context, dict, str]:
         # context init
         ctx = self._context_init(ctx, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.CONTEXT_INIT, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.CONTEXT_INIT, *args, **kwargs)
 
         # get previous node
         ctx = self._get_previous_node(ctx, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.GET_PREVIOUS_NODE, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.GET_PREVIOUS_NODE, *args, **kwargs)
 
         # rewrite previous node
         ctx = self._rewrite_previous_node(ctx, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.REWRITE_PREVIOUS_NODE, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.REWRITE_PREVIOUS_NODE, *args, **kwargs)
 
         # run pre transitions processing
-        ctx = self._run_pre_transitions_processing(ctx, pipeline, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.RUN_PRE_TRANSITIONS_PROCESSING, *args, **kwargs)
+        await self._run_pre_transitions_processing(ctx, pipeline, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.RUN_PRE_TRANSITIONS_PROCESSING, *args, **kwargs)
 
         # get true labels for scopes (GLOBAL, LOCAL, NODE)
-        ctx = self._get_true_labels(ctx, pipeline, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.GET_TRUE_LABELS, *args, **kwargs)
+        ctx = await self._get_true_labels(ctx, pipeline, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.GET_TRUE_LABELS, *args, **kwargs)
 
         # get next node
         ctx = self._get_next_node(ctx, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.GET_NEXT_NODE, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.GET_NEXT_NODE, *args, **kwargs)
 
         ctx.add_label(ctx.framework_states["actor"]["next_label"][:2])
 
         # rewrite next node
         ctx = self._rewrite_next_node(ctx, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.REWRITE_NEXT_NODE, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.REWRITE_NEXT_NODE, *args, **kwargs)
 
         # run pre response processing
-        ctx = self._run_pre_response_processing(ctx, pipeline, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.RUN_PRE_RESPONSE_PROCESSING, *args, **kwargs)
+        await self._run_pre_response_processing(ctx, pipeline, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.RUN_PRE_RESPONSE_PROCESSING, *args, **kwargs)
 
         # create response
-        ctx.framework_states["actor"]["response"] = ctx.framework_states["actor"][
-            "pre_response_processed_node"
-        ].run_response(ctx, pipeline, *args, **kwargs)
-        self._run_handlers(ctx, pipeline, ActorStage.CREATE_RESPONSE, *args, **kwargs)
+        ctx.framework_states["actor"]["response"] = await self.run_response(
+            ctx.framework_states["actor"]["pre_response_processed_node"].response, ctx, pipeline, *args, **kwargs
+        )
+        await self._run_handlers(ctx, pipeline, ActorStage.CREATE_RESPONSE, *args, **kwargs)
         ctx.add_response(ctx.framework_states["actor"]["response"])
 
-        self._run_handlers(ctx, pipeline, ActorStage.FINISH_TURN, *args, **kwargs)
+        await self._run_handlers(ctx, pipeline, ActorStage.FINISH_TURN, *args, **kwargs)
         if self._clean_turn_cache:
             cache_clear()
 
@@ -177,20 +174,20 @@ def _get_previous_node(self, ctx: Context, *args, **kwargs) -> Context:
         ).get(ctx.framework_states["actor"]["previous_label"][1], Node())
         return ctx
 
-    def _get_true_labels(self, ctx: Context, pipeline: Pipeline, *args, **kwargs) -> Context:
+    async def _get_true_labels(self, ctx: Context, pipeline: Pipeline, *args, **kwargs) -> Context:
         # GLOBAL
         ctx.framework_states["actor"]["global_transitions"] = (
             self.script.get(GLOBAL, {}).get(GLOBAL, Node()).transitions
         )
-        ctx.framework_states["actor"]["global_true_label"] = self._get_true_label(
+        global_transitions_coro = self._get_true_label(
             ctx.framework_states["actor"]["global_transitions"], ctx, pipeline, GLOBAL, "global"
         )
 
         # LOCAL
         ctx.framework_states["actor"]["local_transitions"] = (
             self.script.get(ctx.framework_states["actor"]["previous_label"][0], {}).get(LOCAL, Node()).transitions
         )
-        ctx.framework_states["actor"]["local_true_label"] = self._get_true_label(
+        local_transitions_coro = self._get_true_label(
             ctx.framework_states["actor"]["local_transitions"],
             ctx,
             pipeline,
@@ -202,13 +199,18 @@ def _get_true_labels(self, ctx: Context, pipeline: Pipeline, *args, **kwargs) ->
         ctx.framework_states["actor"]["node_transitions"] = ctx.framework_states["actor"][
             "pre_transitions_processed_node"
         ].transitions
-        ctx.framework_states["actor"]["node_true_label"] = self._get_true_label(
+        node_transitions_coro = self._get_true_label(
             ctx.framework_states["actor"]["node_transitions"],
             ctx,
             pipeline,
             ctx.framework_states["actor"]["previous_label"][0],
             "node",
         )
+        (
+            ctx.framework_states["actor"]["global_true_label"],
+            ctx.framework_states["actor"]["local_true_label"],
+            ctx.framework_states["actor"]["node_true_label"],
+        ) = await asyncio.gather(*[global_transitions_coro, local_transitions_coro, node_transitions_coro])
         return ctx
 
     def _get_next_node(self, ctx: Context, *args, **kwargs) -> Context:
@@ -262,25 +264,101 @@ def _overwrite_node(
             overwritten_node.transitions = current_node.transitions
         return overwritten_node
 
-    def _run_pre_transitions_processing(self, ctx: Context, pipeline: Pipeline, *args, **kwargs) -> Context:
-        ctx.framework_states["actor"]["processed_node"] = copy.deepcopy(ctx.framework_states["actor"]["previous_node"])
-        ctx = ctx.framework_states["actor"]["previous_node"].run_pre_transitions_processing(
-            ctx, pipeline, *args, **kwargs
+    async def run_response(
+        self,
+        response: Optional[Union[Message, Callable[..., Message]]],
+        ctx: Context,
+        pipeline: Pipeline,
+        *args,
+        **kwargs,
+    ) -> Context:
+        """
+        Executes the normalized response as an asynchronous function.
+        See the details in the :py:func:`~normalize_response` function of `normalization.py`.
+        """
+        response = normalize_response(response)
+        return await wrap_sync_function_in_async(response, ctx, pipeline, *args, **kwargs)
+
+    async def _run_processing_parallel(
+        self, processing: dict, ctx: Context, pipeline: Pipeline, *args, **kwargs
+    ) -> None:
+        """
+        Execute the processing functions for a particular node simultaneously,
+        independent of the order.
+
+        Picked depending on the value of the :py:class:`.Pipeline`'s `parallelize_processing` flag.
+        """
+        results = await asyncio.gather(
+            *[wrap_sync_function_in_async(func, ctx, pipeline, *args, **kwargs) for func in processing.values()],
+            return_exceptions=True,
         )
+        for exc, (processing_name, processing_func) in zip(results, processing.items()):
+            if isinstance(exc, Exception):
+                logger.error(
+                    f"Exception {exc} for processing_name={processing_name} and processing_func={processing_func}",
+                    exc_info=exc,
+                )
+
+    async def _run_processing_sequential(
+        self, processing: dict, ctx: Context, pipeline: Pipeline, *args, **kwargs
+    ) -> None:
+        """
+        Execute the processing functions for a particular node in-order.
+
+        Picked depending on the value of the :py:class:`.Pipeline`'s `parallelize_processing` flag.
+        """
+        for processing_name, processing_func in processing.items():
+            try:
+                await wrap_sync_function_in_async(processing_func, ctx, pipeline, *args, **kwargs)
+            except Exception as exc:
+                logger.error(
+                    f"Exception {exc} for processing_name={processing_name} and processing_func={processing_func}",
+                    exc_info=exc,
+                )
+
+    async def _run_pre_transitions_processing(self, ctx: Context, pipeline: Pipeline) -> None:
+        """
+        Run `PRE_TRANSITIONS_PROCESSING` functions for a particular node.
+        Pre-transition processing functions can modify the context state
+        before the direction of the next transition is determined depending on that state.
+
+        The execution order depends on the value of the :py:class:`.Pipeline`'s
+        `parallelize_processing` flag.
+        """
+        ctx.framework_states["actor"]["processed_node"] = copy.deepcopy(ctx.framework_states["actor"]["previous_node"])
+        pre_transitions_processing = ctx.framework_states["actor"]["previous_node"].pre_transitions_processing
+
+        if pipeline.parallelize_processing:
+            await self._run_processing_parallel(pre_transitions_processing, ctx, pipeline)
+        else:
+            await self._run_processing_sequential(pre_transitions_processing, ctx, pipeline)
+
         ctx.framework_states["actor"]["pre_transitions_processed_node"] = ctx.framework_states["actor"][
             "processed_node"
         ]
         del ctx.framework_states["actor"]["processed_node"]
-        return ctx
 
-    def _run_pre_response_processing(self, ctx: Context, pipeline: Pipeline, *args, **kwargs) -> Context:
+    async def _run_pre_response_processing(self, ctx: Context, pipeline: Pipeline) -> None:
+        """
+        Run `PRE_RESPONSE_PROCESSING` functions for a particular node.
+        Pre-response processing functions can modify the response before it is
+        returned to the user.
+
+        The execution order depends on the value of the :py:class:`.Pipeline`'s
+        `parallelize_processing` flag.
+        """
         ctx.framework_states["actor"]["processed_node"] = copy.deepcopy(ctx.framework_states["actor"]["next_node"])
-        ctx = ctx.framework_states["actor"]["next_node"].run_pre_response_processing(ctx, pipeline, *args, **kwargs)
+        pre_response_processing = ctx.framework_states["actor"]["next_node"].pre_response_processing
+
+        if pipeline.parallelize_processing:
+            await self._run_processing_parallel(pre_response_processing, ctx, pipeline)
+        else:
+            await self._run_processing_sequential(pre_response_processing, ctx, pipeline)
+
         ctx.framework_states["actor"]["pre_response_processed_node"] = ctx.framework_states["actor"]["processed_node"]
         del ctx.framework_states["actor"]["processed_node"]
-        return ctx
 
-    def _get_true_label(
+    async def _get_true_label(
         self,
         transitions: dict,
         ctx: Context,
@@ -291,10 +369,17 @@ def _get_true_label(
         **kwargs,
     ) -> Optional[NodeLabel3Type]:
         true_labels = []
-        for label, condition in transitions.items():
-            if self.condition_handler(condition, ctx, pipeline, *args, **kwargs):
+
+        cond_booleans = await asyncio.gather(
+            *(
+                self.condition_handler(transition[1], ctx, pipeline, *args, **kwargs)
+                for transition in transitions.items()
+            )
+        )
+        for label, cond_is_true in zip(transitions, cond_booleans):
+            if cond_is_true:
                 if callable(label):
-                    label = label(ctx, pipeline, *args, **kwargs)
+                    label = await wrap_sync_function_in_async(label, ctx, pipeline, *args, **kwargs)
                     # TODO: explicit handling of errors
                     if label is None:
                         continue
@@ -311,8 +396,10 @@ def _get_true_label(
         logger.debug(f"{transition_info} transitions sorted by priority = {true_labels}")
         return true_label
 
-    def _run_handlers(self, ctx, pipeline: Pipeline, actor_stage: ActorStage, *args, **kwargs):
-        [handler(ctx, pipeline, *args, **kwargs) for handler in self.handlers.get(actor_stage, [])]
+    async def _run_handlers(self, ctx, pipeline: Pipeline, actor_stage: ActorStage):
+        stage_handlers = self.handlers.get(actor_stage, [])
+        async_handlers = [wrap_sync_function_in_async(handler, ctx, pipeline) for handler in stage_handlers]
+        await asyncio.gather(*async_handlers)
 
     def _choose_label(
         self, specific_label: Optional[NodeLabel3Type], general_label: Optional[NodeLabel3Type]
@@ -360,7 +447,7 @@ def validate_script(self, pipeline: Pipeline, verbose: bool = True):
             # validate responsing
             response_func = normalize_response(node.response)
             try:
-                response_result = response_func(ctx, pipeline)
+                response_result = asyncio.run(wrap_sync_function_in_async(response_func, ctx, pipeline))
                 if not isinstance(response_result, Message):
                     msg = (
                         "Expected type of response_result is `Message`.\n"
@@ -390,7 +477,7 @@ def validate_script(self, pipeline: Pipeline, verbose: bool = True):
         return error_msgs
 
 
-def default_condition_handler(
+async def default_condition_handler(
     condition: Callable, ctx: Context, pipeline: Pipeline, *args, **kwargs
 ) -> Callable[[Context, Pipeline, Any, Any], bool]:
     """
@@ -400,4 +487,4 @@ def default_condition_handler(
     :param ctx: Context of current condition.
     :param pipeline: Pipeline we use in this condition.
     """
-    return condition(ctx, pipeline, *args, **kwargs)
+    return await wrap_sync_function_in_async(condition, ctx, pipeline, *args, **kwargs)

dff/pipeline/pipeline/pipeline.py

@@ -74,6 +74,9 @@ class Pipeline:
 
         - `_services_pipeline` is a pipeline root :py:class:`~.ServiceGroup` object,
         - `actor` is a pipeline actor, found among services.
+    :param parallelize_processing: This flag determines whether or not the functions
+        defined in the ``PRE_RESPONSE_PROCESSING`` and ``PRE_TRANSITIONS_PROCESSING`` sections
+        of the script should be parallelized over respective groups.
 
     """
 
@@ -94,6 +97,7 @@ def __init__(
         after_handler: Optional[ExtraHandlerBuilder] = None,
         timeout: Optional[float] = None,
         optimization_warnings: bool = False,
+        parallelize_processing: bool = False,
     ):
         self.actor: Actor = None
         self.messenger_interface = CLIMessengerInterface() if messenger_interface is None else messenger_interface
@@ -127,6 +131,8 @@ def __init__(
         if optimization_warnings:
             self._services_pipeline.log_optimization_warnings()
 
+        self.parallelize_processing = parallelize_processing
+
         # NB! The following API is highly experimental and may be removed at ANY time WITHOUT FURTHER NOTICE!!
         self._clean_turn_cache = True
         if self._clean_turn_cache:

dff/pipeline/service/service.py

@@ -124,7 +124,7 @@ async def _run_handler(self, ctx: Context, pipeline: Pipeline):
         else:
             raise Exception(f"Too many parameters required for service '{self.name}' handler: {handler_params}!")
 
-    def _run_as_actor(self, ctx: Context, pipeline: Pipeline):
+    async def _run_as_actor(self, ctx: Context, pipeline: Pipeline):
         """
         Method for running this service if its handler is an `Actor`.
         Catches runtime exceptions.
@@ -133,7 +133,7 @@ def _run_as_actor(self, ctx: Context, pipeline: Pipeline):
         :return: Context, mutated by actor.
         """
         try:
-            ctx = pipeline.actor(pipeline, ctx)
+            ctx = await pipeline.actor(pipeline, ctx)
             self._set_state(ctx, ComponentExecutionState.FINISHED)
         except Exception as exc:
             self._set_state(ctx, ComponentExecutionState.FAILED)
@@ -172,7 +172,7 @@ async def _run(self, ctx: Context, pipeline: Optional[Pipeline] = None) -> Optio
         await self.run_extra_handler(ExtraHandlerType.BEFORE, ctx, pipeline)
 
         if isinstance(self.handler, str) and self.handler == "ACTOR":
-            ctx = self._run_as_actor(ctx, pipeline)
+            ctx = await self._run_as_actor(ctx, pipeline)
         else:
             await self._run_as_service(ctx, pipeline)
 

dff/script/core/context.py

@@ -278,23 +278,5 @@ def current_node(self) -> Optional[Node]:
 
         return node
 
-    def overwrite_current_node_in_processing(self, processed_node: Node):
-        """
-        Set the current node to be `processed_node`.
-        This method only works in processing functions (pre-response and pre-transition).
-
-        The actual current node is not changed.
-
-        :param processed_node: `node` to set as the current node.
-        """
-        is_processing = self.framework_states.get("actor", {}).get("processed_node")
-        if is_processing:
-            self.framework_states["actor"]["processed_node"] = Node.model_validate(processed_node)
-        else:
-            logger.warning(
-                f"The `{self.overwrite_current_node_in_processing.__name__}` "
-                "method can only be called from processing functions (either pre-response or pre-transition)."
-            )
-
 
 Context.model_rebuild()