ReferencePython SDK ReferenceNemoguardrailsColangV2 XRuntime

nemoguardrails.colang.v2_x.runtime.runtime

View as Markdown

Module Contents

Classes

NameDescription
RuntimeV2_xRuntime for executing the guardrails.

Functions

NameDescription
convert_decorator_list_to_dictionaryConvert list of decorators to a dictionary merging the parameters of decorators with same name.
create_flow_configs_from_flow_listCreate a flow config dictionary and resolves flow overriding.

Data

log

API

class nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x(
    config: nemoguardrails.rails.llm.config.RailsConfig,
    verbose: bool = False
)

Bases: Runtime

Runtime for executing the guardrails.

async_actions
Dict[str, List] = {}
nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._add_flows_action(
    state: nemoguardrails.colang.v2_x.runtime.statemachine.State,
    args: dict = {}
) -> typing.List[str]
async
nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._get_action_finished_event(
    result: dict,
    kwargs = {}
) -> typing.Dict[str, typing.Any]
staticmethod

Helper to return the ActionFinished event from the result of running a local action.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._get_action_resp(
    action_meta: typing.Dict[str, typing.Any],
    action_name: str,
    kwargs: typing.Dict[str, typing.Any]
) -> typing.Tuple[typing.Union[str, typing.Dict[str, typing.Any]], str]
async

Interact with actions and get response from action-server and system actions.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._get_async_actions_finished_events(
    main_flow_uid: str
) -> typing.Tuple[typing.List[dict], int]
async

Helper to return the ActionFinished events for the local async actions that finished.

Args main_flow_uid: The UID of the main flow.

Returns (action_finished_events, pending_counter) The array of *ActionFinished events and the pending counter

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._init_flow_configs() -> None

Initializes the flow configs based on the config.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._internal_error_action_result(
    message: str
) -> nemoguardrails.actions.actions.ActionResult
staticmethod

Helper to construct an action result for an internal error.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._process_start_action(
    action_name: str,
    action_params: dict,
    context: dict,
    events: typing.List[dict],
    state: nemoguardrails.colang.v2_x.runtime.statemachine.State
) -> typing.Tuple[typing.Any, typing.List[dict], dict]
async

Starts the specified action, waits for it to finish and posts back the result.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._remove_flows_action(
    state: nemoguardrails.colang.v2_x.runtime.statemachine.State,
    args: dict = {}
) -> None
async
nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x._run_action(
    action_name: str,
    start_action_event: dict,
    events_history: typing.List[typing.Union[dict, nemoguardrails.colang.v2_x.runtime.flows.Event]],
    state: nemoguardrails.colang.v2_x.runtime.statemachine.State
) -> dict
async

Runs the locally registered action.

Args action_name: The name of the action to be executed. start_action_event: The event that triggered the action. events_history: The recent history of events that led to the action being triggered.

nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x.generate_events(
    events: typing.List[dict]
) -> typing.List[dict]
async
nemoguardrails.colang.v2_x.runtime.runtime.RuntimeV2_x.process_events(
    events: typing.List[dict],
    state: typing.Union[typing.Optional[dict], nemoguardrails.colang.v2_x.runtime.statemachine.State] = None,
    blocking: bool = False,
    instant_actions: typing.Optional[typing.List[str]] = None
) -> typing.Tuple[typing.List[typing.Dict[str, typing.Any]], nemoguardrails.colang.v2_x.runtime.statemachine.State]
async

Process a sequence of events in a given state.

Runs an “event processing cycle”, i.e., process all input events in the given state, and return the new state and the output events.

The events will be processed one by one, in the input order. If new events are generated as part of the processing, they will be appended to the input events.

By default, a processing cycle only waits for the local actions to finish, i.e, if after processing all the input events, there are local actions in progress, the event processing will wait for them to finish.

In blocking mode, the event processing will also wait for the local async actions.

Parameters:

events
List[dict]

A sequence of events that needs to be processed.

state
Union[Optional[dict], State]Defaults to None

The state that should be used as the starting point. If not provided, a clean state will be used.

blocking
boolDefaults to False

If set, in blocking mode, the processing cycle will wait for all the local async actions as well.

instant_actions
Optional[List[str]]Defaults to None

The name of the actions which should finish instantly, i.e., the start event will not be returned to the user and wait for the finish event.

Returns: Tuple[List[Dict[str, Any]], State]

(output_events, output_state) Returns a sequence of output events and an output state.

nemoguardrails.colang.v2_x.runtime.runtime.convert_decorator_list_to_dictionary(
    decorators: typing.List[nemoguardrails.colang.v2_x.lang.colang_ast.Decorator]
) -> typing.Dict[str, typing.Dict[str, typing.Any]]

Convert list of decorators to a dictionary merging the parameters of decorators with same name.

nemoguardrails.colang.v2_x.runtime.runtime.create_flow_configs_from_flow_list(
    flows: typing.List[nemoguardrails.colang.v2_x.lang.colang_ast.Flow]
) -> typing.Dict[str, nemoguardrails.colang.v2_x.runtime.statemachine.FlowConfig]

Create a flow config dictionary and resolves flow overriding.

nemoguardrails.colang.v2_x.runtime.runtime.log = logging.getLogger(__name__)