nemoguardrails.colang.v2_x.runtime.statemachine | NVIDIA NeMo Guardrails Library Developer Guide

Module Contents

Functions

Name	Description
`_abort_flow`	Abort a flow instance and all its active child flows and decrement number of references of activated flow.
`_add_head_to_event_matching_structures`	-
`_advance_head_front`	Advance all provided heads to the next blocking elements (actionable, matching, head merge)
`_clean_up_state`	Perform a clean up of the state to avoid growing memory footprint.
`_compute_arguments_dict_matching_score`	-
`_compute_event_comparison_score`	Check if the given element matches the given event.
`_compute_event_matching_score`	Check if the element matches with given event.
`_context_log`	-
`_create_event_reference`	-
`_evaluate_arguments`	-
`_finish_flow`	Finish a flow instance and all its active child flows and decrement number of references of activated flow.
`_flow_head_changed`	Callback function that is registered to head position/status changes
`_generate_action_event_from_actionable_element`	Helper to create an outgoing event from the flow head element.
`_generate_umim_event`	-
`_get_all_head_candidates`	Find all heads that are on a potential match with the event.
`_get_eval_context`	-
`_get_flow_state_hierarchy`	-
`_get_readable_flow_state_hierarchy`	-
`_get_reference_activated_flow_instance`	-
`_handle_event_matching`	-
`_is_child_activated_flow`	-
`_is_done_flow`	-
`_is_reference_activated_flow`	-
`_log_action_or_intents`	-
`_process_internal_events_without_default_matchers`	Process internal events that have no default matchers in flows yet.
`_push_internal_event`	-
`_push_left_internal_event`	-
`_remove_head_from_event_matching_structures`	-
`_resolve_action_conflicts`	Resolve all conflicting action conflicts from actionable heads.
`_start_flow`	-
`_update_action_status_by_event`	-
`add_new_flow_instance`	Add a new flow instance to the current state.
`create_flow_instance`	Create a new flow instance that can be added.
`create_internal_event`	Returns an internal event for the provided event data
`create_umim_event`	Returns an outgoing UMIM event for the provided action data
`find_all_active_event_matchers`	Return a list of all active heads that point to an event ‘match’ element.
`get_element_from_head`	Returns the element at the flow head position
`get_event_from_element`	Converts the element into the corresponding event if possible.
`get_event_name_from_element`	Converts the element into the corresponding event name if possible.
`get_flow_config_from_head`	Return the flow config of the flow of the head
`get_flow_state_from_head`	Return the flow state of the flow head
`initialize_flow`	Initialize a flow before it can be used and instantiated.
`initialize_state`	Initialize the state to make it ready for the story start.
`is_action_op_element`	Check if the given element is actionable.
`is_active_flow`	True if flow has started.
`is_inactive_flow`	True if flow is not started.
`is_listening_flow`	True if flow is started or waiting to be started.
`is_match_op_element`	Check if the given element is a match statement.
`run_to_completion`	Compute the next state of the flow-driven system.
`slide`	Try to slide a flow with the provided head.

Data

log

API

nemoguardrails.colang.v2_x.runtime.statemachine._abort_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    matching_scores: typing.List[float],
    deactivate_flow: bool = False,
    context_update: typing.Optional[dict] = None
) -> None

Abort a flow instance and all its active child flows and decrement number of references of activated flow.

nemoguardrails.colang.v2_x.runtime.statemachine._add_head_to_event_matching_structures(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._advance_head_front(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    heads: typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]
) -> typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]

Advance all provided heads to the next blocking elements (actionable, matching, head merge) and returns all heads on actionable elements.

nemoguardrails.colang.v2_x.runtime.statemachine._clean_up_state(
    state: nemoguardrails.colang.v2_x.runtime.flows.State
) -> None

Perform a clean up of the state to avoid growing memory footprint.

nemoguardrails.colang.v2_x.runtime.statemachine._compute_arguments_dict_matching_score(
    args: typing.Any,
    ref_args: typing.Any
) -> float

nemoguardrails.colang.v2_x.runtime.statemachine._compute_event_comparison_score(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event,
    ref_event: nemoguardrails.colang.v2_x.runtime.flows.Event,
    priority: typing.Optional[float] = None
) -> float

Check if the given element matches the given event.

Factors that determine the final score:

match event parameter specificity
flow priority [0.0-1.0]
definition order of flow

Returns: float

1.0: Exact match (all parameters match)

nemoguardrails.colang.v2_x.runtime.statemachine._compute_event_matching_score(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> float

Check if the element matches with given event.

nemoguardrails.colang.v2_x.runtime.statemachine._context_log(
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> str

nemoguardrails.colang.v2_x.runtime.statemachine._create_event_reference(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    element: nemoguardrails.colang.v2_x.lang.colang_ast.SpecOp,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> dict

nemoguardrails.colang.v2_x.runtime.statemachine._evaluate_arguments(
    arguments: dict,
    context: dict
) -> dict

nemoguardrails.colang.v2_x.runtime.statemachine._finish_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    matching_scores: typing.List[float],
    deactivate_flow: bool = False,
    context_update: typing.Optional[dict] = None
) -> None

Finish a flow instance and all its active child flows and decrement number of references of activated flow.

nemoguardrails.colang.v2_x.runtime.statemachine._flow_head_changed(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> None

Callback function that is registered to head position/status changes and will update acceleration data structures.

nemoguardrails.colang.v2_x.runtime.statemachine._generate_action_event_from_actionable_element(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> None

Helper to create an outgoing event from the flow head element.

nemoguardrails.colang.v2_x.runtime.statemachine._generate_umim_event(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> typing.Dict[str, typing.Any]

nemoguardrails.colang.v2_x.runtime.statemachine._get_all_head_candidates(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> typing.List[typing.Tuple[str, str]]

Find all heads that are on a potential match with the event. Returns those heads in a flow hierarchical order.

nemoguardrails.colang.v2_x.runtime.statemachine._get_eval_context(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> dict

nemoguardrails.colang.v2_x.runtime.statemachine._get_flow_state_hierarchy(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state_uid: str
) -> typing.List[str]

nemoguardrails.colang.v2_x.runtime.statemachine._get_readable_flow_state_hierarchy(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state_uid: str
) -> str

nemoguardrails.colang.v2_x.runtime.statemachine._get_reference_activated_flow_instance(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.InternalEvent
) -> typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.FlowState]

nemoguardrails.colang.v2_x.runtime.statemachine._handle_event_matching(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event,
    heads_matching: typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._is_child_activated_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

nemoguardrails.colang.v2_x.runtime.statemachine._is_done_flow(
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

nemoguardrails.colang.v2_x.runtime.statemachine._is_reference_activated_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

nemoguardrails.colang.v2_x.runtime.statemachine._log_action_or_intents(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    matching_scores: typing.List[float]
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._process_internal_events_without_default_matchers(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> typing.Set[str]

Process internal events that have no default matchers in flows yet. Return a set of all the event loop ids that handled the event.

nemoguardrails.colang.v2_x.runtime.statemachine._push_internal_event(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._push_left_internal_event(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.InternalEvent
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._remove_head_from_event_matching_structures(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> bool

nemoguardrails.colang.v2_x.runtime.statemachine._resolve_action_conflicts(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    actionable_heads: typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]
) -> typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]

Resolve all conflicting action conflicts from actionable heads.

nemoguardrails.colang.v2_x.runtime.statemachine._start_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    event_arguments: dict
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine._update_action_status_by_event(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: nemoguardrails.colang.v2_x.runtime.flows.ActionEvent
) -> None

nemoguardrails.colang.v2_x.runtime.statemachine.add_new_flow_instance(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> nemoguardrails.colang.v2_x.runtime.flows.FlowState

Add a new flow instance to the current state.

nemoguardrails.colang.v2_x.runtime.statemachine.create_flow_instance(
    flow_config: nemoguardrails.colang.v2_x.runtime.flows.FlowConfig,
    flow_instance_uid: str,
    flow_hierarchy_position: str,
    event_arguments: typing.Dict[str, typing.Any]
) -> nemoguardrails.colang.v2_x.runtime.flows.FlowState

Create a new flow instance that can be added.

nemoguardrails.colang.v2_x.runtime.statemachine.create_internal_event(
    event_name: str,
    event_args: dict,
    matching_scores: typing.List[float]
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Returns an internal event for the provided event data

nemoguardrails.colang.v2_x.runtime.statemachine.create_umim_event(
    event: nemoguardrails.colang.v2_x.runtime.flows.Event,
    event_args: typing.Dict[str, typing.Any],
    config: typing.Optional[nemoguardrails.rails.llm.config.RailsConfig]
) -> typing.Dict[str, typing.Any]

Returns an outgoing UMIM event for the provided action data

nemoguardrails.colang.v2_x.runtime.statemachine.find_all_active_event_matchers(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    event: typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.Event] = None
) -> typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]

Return a list of all active heads that point to an event ‘match’ element.

nemoguardrails.colang.v2_x.runtime.statemachine.get_element_from_head(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> typing.Optional[nemoguardrails.colang.v2_x.lang.colang_ast.ElementType]

Returns the element at the flow head position

nemoguardrails.colang.v2_x.runtime.statemachine.get_event_from_element(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    element: nemoguardrails.colang.v2_x.lang.colang_ast.SpecOp
) -> nemoguardrails.colang.v2_x.runtime.flows.Event

Converts the element into the corresponding event if possible.

Cases:

Event as member of an action or flow reference: send/match $ref.Finished(args) (This is action/flow specific)
Event as member of an action or flow constructor: send/match UtteranceBotAction(args).Finished(args)
Bare event: send/match UtteranceBotActionFinished(args)

nemoguardrails.colang.v2_x.runtime.statemachine.get_event_name_from_element(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    element: nemoguardrails.colang.v2_x.lang.colang_ast.SpecOp
) -> str

Converts the element into the corresponding event name if possible. See also function get_event_from_element which is very similar but returns the full event including parameters.

nemoguardrails.colang.v2_x.runtime.statemachine.get_flow_config_from_head(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> nemoguardrails.colang.v2_x.runtime.flows.FlowConfig

Return the flow config of the flow of the head

nemoguardrails.colang.v2_x.runtime.statemachine.get_flow_state_from_head(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> nemoguardrails.colang.v2_x.runtime.flows.FlowState

Return the flow state of the flow head

nemoguardrails.colang.v2_x.runtime.statemachine.initialize_flow(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_config: nemoguardrails.colang.v2_x.runtime.flows.FlowConfig
) -> None

Initialize a flow before it can be used and instantiated.

nemoguardrails.colang.v2_x.runtime.statemachine.initialize_state(
    state: nemoguardrails.colang.v2_x.runtime.flows.State
) -> None

Initialize the state to make it ready for the story start.

nemoguardrails.colang.v2_x.runtime.statemachine.is_action_op_element(
    element: nemoguardrails.colang.v2_x.lang.colang_ast.ElementType
) -> bool

Check if the given element is actionable.

nemoguardrails.colang.v2_x.runtime.statemachine.is_active_flow(
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

True if flow has started.

nemoguardrails.colang.v2_x.runtime.statemachine.is_inactive_flow(
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

True if flow is not started.

nemoguardrails.colang.v2_x.runtime.statemachine.is_listening_flow(
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState
) -> bool

True if flow is started or waiting to be started.

nemoguardrails.colang.v2_x.runtime.statemachine.is_match_op_element(
    element: nemoguardrails.colang.v2_x.lang.colang_ast.ElementType
) -> bool

Check if the given element is a match statement.

nemoguardrails.colang.v2_x.runtime.statemachine.run_to_completion(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    external_event: typing.Union[dict, nemoguardrails.colang.v2_x.runtime.flows.Event]
) -> nemoguardrails.colang.v2_x.runtime.flows.State

Compute the next state of the flow-driven system.

nemoguardrails.colang.v2_x.runtime.statemachine.slide(
    state: nemoguardrails.colang.v2_x.runtime.flows.State,
    flow_state: nemoguardrails.colang.v2_x.runtime.flows.FlowState,
    flow_config: nemoguardrails.colang.v2_x.runtime.flows.FlowConfig,
    head: nemoguardrails.colang.v2_x.runtime.flows.FlowHead
) -> typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowHead]

Try to slide a flow with the provided head.

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