nemoguardrails.actions.action_dispatcher | NVIDIA NeMo Guardrails Library Developer Guide

Module for the calling proper action endpoints based on events received at action server endpoint

Module Contents

Classes

Name	Description
`ActionDispatcher`	-
`AsyncInvokableAction`	-
`RunnableAction`	-
`_RegisteredActions`	Read-only view over registered and lazily declared actions.

Functions

Name	Description
`is_action_file`	Heuristics for determining if a Python file can have actions or not.

Data

LEGACY_UNMANIFESTED_LIBRARY_PACKAGES

RegisteredAction

log

API

class nemoguardrails.actions.action_dispatcher.ActionDispatcher(
    load_all_actions: bool = True,
    config_path: typing.Optional[str] = None,
    import_paths: typing.Optional[typing.List[str]] = None,
    rail_catalog: typing.Optional[nemoguardrails.manifests.RailCatalog] = None
)

_lazy_action_refs

Dict[str, ActionRef] = {}

_manifest_action_names

set[str] = set()

_registered_actions

Dict[str, RegisteredAction] = {}

_registered_actions_view

= _RegisteredActions(self)

registered_actions

Gets the dictionary of registered actions. Returns: dict: A dictionary where keys are action names and values are callable action functions.

nemoguardrails.actions.action_dispatcher.ActionDispatcher._action_names() -> tuple[str, ...]

nemoguardrails.actions.action_dispatcher.ActionDispatcher._find_actions(
    directory
) -> typing.Dict

Loop through all the subdirectories and check for the class with @action decorator and add in action_classes dict.

Parameters:

directory

The directory to search for actions.

Returns: Dict

Dictionary of found actions.

nemoguardrails.actions.action_dispatcher.ActionDispatcher._has_action_name(
    action_name: str
) -> bool

nemoguardrails.actions.action_dispatcher.ActionDispatcher._load_actions_from_module(
    filepath: str
)

staticmethod

Loads the actions from the specified python module.

Parameters:

filepath

str

The path of the Python module.

Returns:

Dictionary of loaded actions.

nemoguardrails.actions.action_dispatcher.ActionDispatcher._load_unmanifested_library_actions(
    library_path: pathlib.Path
) -> None

Load pre-manifest library packages that register actions directly.

Manifested library actions are indexed as lazy refs by _register_manifest_action_refs; only the packages in LEGACY_UNMANIFESTED_LIBRARY_PACKAGES still register eagerly.

nemoguardrails.actions.action_dispatcher.ActionDispatcher._normalize_action_name(
    name: str
) -> str

Normalize the action name to the required format.

nemoguardrails.actions.action_dispatcher.ActionDispatcher._register_action_ref(
    action_ref: nemoguardrails.manifests.ActionRef,
    override: bool = True
) -> None

nemoguardrails.actions.action_dispatcher.ActionDispatcher._register_manifest_action_refs() -> None

nemoguardrails.actions.action_dispatcher.ActionDispatcher._resolve_registered_action(
    action_name: str
) -> typing.Optional[nemoguardrails.actions.action_dispatcher.RegisteredAction]

nemoguardrails.actions.action_dispatcher.ActionDispatcher._sanitize_for_log(
    value: typing.Any
) -> str

staticmethod

Sanitize values before logging to prevent log injection.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.execute_action(
    action_name: str,
    params: typing.Dict[str, typing.Any]
) -> typing.Tuple[typing.Union[typing.Optional[str], typing.Dict[str, typing.Any]], str]

async

Execute a registered action.

Parameters:

action_name

str

The name of the action to execute.

params

Dict[str, Any]

Parameters for the action.

Returns: Tuple[Union[Optional[str], Dict[str, Any]], str]

Tuple[Union[str, Dict[str, Any]], str]: A tuple containing the result and status.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.get_action(
    name: str
) -> typing.Optional[nemoguardrails.actions.action_dispatcher.RegisteredAction]

Get the registered action by name.

Parameters:

name

str

The name of the action.

Returns: Optional[RegisteredAction]

The registered action.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.get_registered_actions() -> typing.List[str]

Get the list of available actions.

Returns: List[str]

List[str]: List of available actions.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.has_registered(
    name: str
) -> bool

Check if an action is registered.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.is_manifest_action(
    name: str
) -> bool

nemoguardrails.actions.action_dispatcher.ActionDispatcher.load_actions_from_path(
    path: pathlib.Path
)

Loads all actions from the specified path.

This method loads all actions from the actions.py file if it exists and all actions inside the actions folder if it exists.

Parameters:

path

str

A string representing the path from which to load actions.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.register_action(
    action: nemoguardrails.actions.action_dispatcher.RegisteredAction,
    name: typing.Optional[str] = None,
    override: bool = True
)

Registers an action with the given name.

Parameters:

action

RegisteredAction

The action function or runnable object.

name

Optional[str]Defaults to None

The name of the action. Defaults to None.

override

boolDefaults to True

If an action already exists, whether it should be overridden or not.

nemoguardrails.actions.action_dispatcher.ActionDispatcher.register_actions(
    actions_obj: typing.Any,
    override: bool = True
)

Registers all the actions from the given object.

Parameters:

actions_obj

any

The object containing actions.

override

boolDefaults to True

If an action already exists, whether it should be overridden or not.

class nemoguardrails.actions.action_dispatcher.AsyncInvokableAction()

Protocol

nemoguardrails.actions.action_dispatcher.AsyncInvokableAction.ainvoke(
    args: typing.Any = (),
    kwargs: typing.Any = {}
) -> typing.Awaitable[typing.Any]

class nemoguardrails.actions.action_dispatcher.RunnableAction()

Protocol

nemoguardrails.actions.action_dispatcher.RunnableAction.run(
    args: typing.Any = (),
    kwargs: typing.Any = {}
) -> typing.Any

class nemoguardrails.actions.action_dispatcher._RegisteredActions(
    dispatcher: nemoguardrails.actions.action_dispatcher.ActionDispatcher
)

Bases: Mapping[str, RegisteredAction]

Read-only view over registered and lazily declared actions.

Iteration, in, and len reflect every registered and lazily declared action name. Subscripting (and therefore dict(view)) resolves a lazy action on demand, importing its module and raising KeyError if it cannot be resolved. values and items deliberately expose only actions that are already resolved, so iterating them never triggers a lazy import or fails on an action whose optional dependency is missing.

nemoguardrails.actions.action_dispatcher._RegisteredActions.__contains__(
    name: object
) -> bool

nemoguardrails.actions.action_dispatcher._RegisteredActions.__getitem__(
    name: str
) -> nemoguardrails.actions.action_dispatcher.RegisteredAction

nemoguardrails.actions.action_dispatcher._RegisteredActions.__iter__() -> collections.abc.Iterator[str]

nemoguardrails.actions.action_dispatcher._RegisteredActions.__len__() -> int

nemoguardrails.actions.action_dispatcher._RegisteredActions.items() -> collections.abc.ItemsView[str, nemoguardrails.actions.action_dispatcher.RegisteredAction]

nemoguardrails.actions.action_dispatcher._RegisteredActions.values() -> collections.abc.ValuesView[nemoguardrails.actions.action_dispatcher.RegisteredAction]

nemoguardrails.actions.action_dispatcher.is_action_file(
    filepath
)

Heuristics for determining if a Python file can have actions or not.

Currently, it only excludes the `init.py files.

nemoguardrails.actions.action_dispatcher.LEGACY_UNMANIFESTED_LIBRARY_PACKAGES: Tuple[str, ...] = ('attention', 'utils')

nemoguardrails.actions.action_dispatcher.RegisteredAction: TypeAlias = Union[Callable[..., Any], Type[Any], AsyncInvokableAction, RunnableAction]

nemoguardrails.actions.action_dispatcher.log = logging.getLogger(__name__)