Feature/python action events

nemoguardrails/actions/action_dispatcher.py

@@ -33,6 +33,8 @@
 
 
 class ActionDispatcher:
+    """Manages the execution and life time of local actions."""
+
     def __init__(
         self,
         load_all_actions: bool = True,
@@ -51,7 +53,8 @@ def __init__(
         """
         log.info("Initializing action dispatcher")
 
-        self._registered_actions = {}
+        # Dictionary with all registered actions
+        self._registered_actions: dict = {}
 
         if load_all_actions:
             # TODO: check for better way to find actions dir path or use constants.py
@@ -87,7 +90,7 @@ def __init__(
                 for import_path in import_paths:
                     self.load_actions_from_path(Path(import_path.strip()))
 
-        log.info(f"Registered Actions :: {sorted(self._registered_actions.keys())}")
+        log.info("Registered Actions :: %s", sorted(self._registered_actions.keys()))
         log.info("Action dispatcher initialized")
 
     @property
@@ -181,7 +184,7 @@ def get_action(self, name: str) -> callable:
 
     async def execute_action(
         self, action_name: str, params: Dict[str, Any]
-    ) -> Tuple[Union[str, Dict[str, Any]], str]:
+    ) -> Tuple[Optional[Union[str, Dict[str, Any]]], str]:
         """Execute a registered action.
 
         Args:
@@ -195,7 +198,7 @@ async def execute_action(
         action_name = self._normalize_action_name(action_name)
 
         if action_name in self._registered_actions:
-            log.info(f"Executing registered action: {action_name}")
+            log.info("Executing registered action: %s", action_name)
             fn = self._registered_actions.get(action_name, None)
 
             # Actions that are registered as classes are initialized lazy, when
@@ -214,7 +217,7 @@ async def execute_action(
                             result = await result
                         else:
                             log.warning(
-                                f"Synchronous action `{action_name}` has been called."
+                                "Synchronous action `%s` has been called.", action_name
                             )
 
                     elif isinstance(fn, Chain):
@@ -256,15 +259,12 @@ async def execute_action(
                     filtered_params = {
                         k: v
                         for k, v in params.items()
-                        if k not in ["state", "events", "llm"]
+                        if k not in ["state", "events", "llm", "event_handler"]
                     }
-                    log.warning(
-                        "Error while execution '%s' with parameters '%s': %s",
-                        action_name,
-                        filtered_params,
-                        e,
+                    msg = (
+                        f"Exception while execution '{action_name}' with parameters '{filtered_params}'",
                     )
-                    log.exception(e)
+                    raise Exception(f"{msg}: {e}") from e
 
         return None, "failed"
 

nemoguardrails/cli/chat.py

@@ -498,7 +498,9 @@ async def _process_input_events():
                 chat_state.input_events = []
             else:
                 chat_state.waiting_user_input = True
-                await enable_input.wait()
+                # NOTE: We should never disable the user input since we can have
+                # async Python actions running in parallel
+                # await enable_input.wait()
 
                 user_message: str = await chat_state.session.prompt_async(
                     HTML("<prompt>\n> </prompt>"),

nemoguardrails/colang/runtime.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import logging
 from abc import abstractmethod
 from typing import Any, Callable, List, Optional, Tuple
@@ -47,7 +48,7 @@ def __init__(self, config: RailsConfig, verbose: bool = False):
 
         # A set of watchers that are notified every time an event is processed.
         # Used mainly for reporting the progress to the CLI.
-        self.watchers = []
+        self.watchers: List = []
 
         # The maximum number of events to be processed in a processing loop
         self.max_events = 500

nemoguardrails/colang/v1_0/runtime/runtime.py

@@ -25,6 +25,7 @@
 from langchain.chains.base import Chain
 
 from nemoguardrails.actions.actions import ActionResult
+from nemoguardrails.actions.llm.utils import LLMCallException
 from nemoguardrails.colang import parse_colang_file
 from nemoguardrails.colang.runtime import Runtime
 from nemoguardrails.colang.v1_0.runtime.flows import (
@@ -360,9 +361,16 @@ async def _process_start_action(self, events: List[dict]) -> List[dict]:
                     kwargs["llm"] = self.registered_action_params[f"{action_name}_llm"]
 
                 log.info("Executing action :: %s", action_name)
-                result, status = await self.action_dispatcher.execute_action(
-                    action_name, kwargs
-                )
+                try:
+                    result, status = await self.action_dispatcher.execute_action(
+                        action_name, kwargs
+                    )
+                except LLMCallException as e:
+                    raise e
+                except Exception as e:
+                    result = None
+                    status = "failed"
+                    log.exception(e)
 
             # If the action execution failed, we return a hardcoded message
             if status == "failed":

nemoguardrails/colang/v2_x/runtime/flows.py

@@ -23,7 +23,18 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
-from typing import Any, Callable, Deque, Dict, List, Optional, Tuple, Union
+from typing import (
+    Any,
+    Callable,
+    ClassVar,
+    Deque,
+    Dict,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+)
 
 from dataclasses_json import dataclass_json
 
@@ -33,7 +44,7 @@
     FlowReturnMemberDef,
 )
 from nemoguardrails.colang.v2_x.runtime.errors import ColangSyntaxError
-from nemoguardrails.utils import new_readable_uuid, new_uuid
+from nemoguardrails.utils import new_event_dict, new_readable_uuid, new_uuid
 
 log = logging.getLogger(__name__)
 
@@ -108,6 +119,15 @@ def from_umim_event(cls, event: dict) -> Event:
         )
         return new_event
 
+    def to_umim_event(self, event_source_uid: Optional[str] = None) -> Dict[str, Any]:
+        """Return a umim event dictionary."""
+        new_event_args = dict(self.arguments)
+        new_event_args.setdefault(
+            "source_uid",
+            event_source_uid if event_source_uid else "NeMoGuardrails-Colang-2.x",
+        )
+        return new_event_dict(self.name, **new_event_args)
+
     # Expose all event parameters as attributes of the event
     def __getattr__(self, name):
         if (
@@ -150,6 +170,20 @@ def from_umim_event(cls, event: dict) -> ActionEvent:
             new_event.action_uid = event["action_uid"]
         return new_event
 
+    def to_umim_event(self, event_source_uid: Optional[str] = None) -> Dict[str, Any]:
+        """Return a umim event dictionary."""
+        new_event_args = dict(self.arguments)
+        new_event_args.setdefault(
+            "source_uid",
+            event_source_uid if event_source_uid else "NeMoGuardrails-Colang-2.x",
+        )
+        if self.action_uid and "action_uid" not in new_event_args:
+            return new_event_dict(
+                self.name, action_uid=self.action_uid, **new_event_args
+            )
+        else:
+            return new_event_dict(self.name, **new_event_args)
+
 
 class ActionStatus(Enum):
     """The status of an action."""
@@ -176,19 +210,43 @@ class Action:
         "Stop": "stop_event",
     }
 
+    # List of umim specific parameters
+    _umim_parameters: ClassVar[List[str]] = [
+        "type",
+        "uid",
+        "event_created_at",
+        "source_uid",
+        "action_uid",
+        "action_info_modality",
+        "action_info_modality_policy",
+        "action_finished_at",
+    ]
+
     @classmethod
     def from_event(cls, event: ActionEvent) -> Optional[Action]:
         """Returns the action if event name conforms with UMIM convention."""
         assert event.action_uid is not None
         for name in cls._event_name_map:
             if name in event.name:
-                action = Action(event.name.replace(name, ""), {})
+                action_name: str
+                if name == "Updated":
+                    index = event.name.find("Action") + 6
+                    action_name = event.name[:index]
+                else:
+                    action_name = event.name.replace(name, "")
+                action = Action(action_name, {})
                 action.uid = event.action_uid
                 action.status = (
                     ActionStatus.STARTED
                     if name != "Finished"
                     else ActionStatus.FINISHED
                 )
+                if name == "Start":
+                    action.start_event_arguments = {
+                        key: event.arguments[key]
+                        for key in event.arguments
+                        if key not in cls._umim_parameters
+                    }
                 return action
         return None
 
@@ -288,7 +346,7 @@ def start_event(self, _args: dict) -> ActionEvent:
     def change_event(self, args: dict) -> ActionEvent:
         """Changes a parameter of a started action."""
         return ActionEvent(
-            name=f"Change{self.name}", arguments=args["arguments"], action_uid=self.uid
+            name=f"Change{self.name}", arguments=args, action_uid=self.uid
         )
 
     def stop_event(self, _args: dict) -> ActionEvent:
@@ -355,7 +413,7 @@ class FlowConfig:
     id: str
 
     # The sequence of elements that compose the flow.
-    elements: List[ElementType]
+    elements: Sequence[ElementType]
 
     # The flow parameters
     parameters: List[FlowParamDef]