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

A simplified modeling of the CoFlows engine.

Module Contents

Classes

Name	Description
`Action`	The action class groups and manages the action events.
`ActionEvent`	The action event class.
`ActionStatus`	The status of an action.
`Event`	The base event class.
`FlowConfig`	The configuration of a flow.
`FlowHead`	The flow head that points to a certain element in the flow
`FlowHeadStatus`	The status of a flow head.
`FlowState`	The state of a flow.
`FlowStatus`	The status of a flow.
`InteractionLoopType`	The type of the interaction loop.
`InternalEvent`	The internal event class.
`InternalEvents`	All internal event types. This event will not appear in the event stream and have priority over them.
`State`	The state of a flow-driven system.

Data

log

random_seed

API

class nemoguardrails.colang.v2_x.runtime.flows.Action(
    name: str,
    arguments: typing.Dict[str, typing.Any],
    flow_uid: typing.Optional[str] = None
)

The action class groups and manages the action events.

_event_name_map

context

dict = {}

flow_scope_count

= 0

status

ActionStatus = ActionStatus.INITIALIZED

uid

str = new_uuid()

nemoguardrails.colang.v2_x.runtime.flows.Action.__getattr__(
    name
)

nemoguardrails.colang.v2_x.runtime.flows.Action.change_event(
    args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Changes a parameter of a started action.

nemoguardrails.colang.v2_x.runtime.flows.Action.finished_event(
    args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Returns the Finished action event.

nemoguardrails.colang.v2_x.runtime.flows.Action.from_dict(
    d
)

staticmethod

nemoguardrails.colang.v2_x.runtime.flows.Action.from_event(
    event: nemoguardrails.colang.v2_x.runtime.flows.ActionEvent
) -> typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.Action]

classmethod

Returns the action if event name conforms with UMIM convention.

nemoguardrails.colang.v2_x.runtime.flows.Action.get_event(
    name: str,
    arguments: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Returns the corresponding action event.

nemoguardrails.colang.v2_x.runtime.flows.Action.process_event(
    event: nemoguardrails.colang.v2_x.runtime.flows.ActionEvent
) -> None

Processes event and updates action accordingly.

nemoguardrails.colang.v2_x.runtime.flows.Action.start_event(
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Starts the action. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.Action.started_event(
    args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Returns the Started action event.

nemoguardrails.colang.v2_x.runtime.flows.Action.stop_event(
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Stops a started action. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.Action.to_dict()

nemoguardrails.colang.v2_x.runtime.flows.Action.updated_event(
    args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

Returns the Updated parameter action event.

class nemoguardrails.colang.v2_x.runtime.flows.ActionEvent(
    name: str,
    arguments: dict,
    matching_scores: typing.List[float] = list(),
    action_uid: typing.Optional[str] = None,
    action: typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.Action] = None
)

Dataclass

Bases: Event

The action event class.

action

Optional[Action] = None

action_uid

Optional[str] = None

nemoguardrails.colang.v2_x.runtime.flows.ActionEvent.from_umim_event(
    event: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.ActionEvent

classmethod

Creates an event from a flat dictionary.

class nemoguardrails.colang.v2_x.runtime.flows.ActionStatus

Bases: enum.Enum

The status of an action.

FINISHED

= 'finished'

INITIALIZED

= 'initialized'

STARTED

= 'started'

STARTING

= 'starting'

STOPPING

= 'stopping'

class nemoguardrails.colang.v2_x.runtime.flows.Event(
    name: str,
    arguments: dict,
    matching_scores: typing.List[float] = list()
)

Dataclass

The base event class.

arguments

dict

matching_scores

List[float] = field(default_factory=list)

name

str

nemoguardrails.colang.v2_x.runtime.flows.Event.__eq__(
    other: object
) -> bool

nemoguardrails.colang.v2_x.runtime.flows.Event.__getattr__(
    name
)

nemoguardrails.colang.v2_x.runtime.flows.Event.__str__() -> str

nemoguardrails.colang.v2_x.runtime.flows.Event.from_umim_event(
    event: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.Event

classmethod

Creates an event from a flat dictionary.

nemoguardrails.colang.v2_x.runtime.flows.Event.is_equal(
    other: nemoguardrails.colang.v2_x.runtime.flows.Event
) -> bool

Compares two events in terms of their name and arguments.

class nemoguardrails.colang.v2_x.runtime.flows.FlowConfig(
    id: str,
    elements: typing.List[nemoguardrails.colang.v2_x.lang.colang_ast.ElementType],
    parameters: typing.List[nemoguardrails.colang.v2_x.lang.colang_ast.FlowParamDef],
    decorators: typing.Dict[str, typing.Dict[str, typing.Any]] = dict(),
    return_members: typing.List[nemoguardrails.colang.v2_x.lang.colang_ast.FlowReturnMemberDef] = list(),
    element_labels: typing.Dict[str, int] = dict(),
    source_code: typing.Optional[str] = None,
    source_file: typing.Optional[str] = None
)

Dataclass

The configuration of a flow.

decorators

Dict[str, Dict[str, Any]] = field(default_factory=dict)

element_labels

Dict[str, int] = field(default_factory=dict)

elements

List[ElementType]

str

is_override

bool

Return True if flow is marked as override.

loop_id

Optional[str]

Return the interaction loop id if set.

loop_priority

int

Return the interaction loop priority (default: 0).

loop_type

InteractionLoopType

Return the interaction loop type.

parameters

List[FlowParamDef]

return_members

List[FlowReturnMemberDef] = field(default_factory=list)

source_code

Optional[str] = None

source_file

Optional[str] = None

nemoguardrails.colang.v2_x.runtime.flows.FlowConfig.has_meta_tag(
    tag_name: str
) -> bool

Return True if flow is marked with given meta tag, e.g. @meta(llm_exclude=True).

nemoguardrails.colang.v2_x.runtime.flows.FlowConfig.meta_tag(
    tag_name: str
) -> typing.Optional[typing.Any]

Return the parameter of the meta tag or None if it does not exist.

class nemoguardrails.colang.v2_x.runtime.flows.FlowHead(
    uid: str,
    flow_state_uid: str,
    matching_scores: typing.List[float],
    scope_uids: typing.List[str] = list(),
    child_head_uids: typing.List[str] = list(),
    catch_pattern_failure_label: typing.List[str] = list(),
    position_changed_callback: typing.Optional[typing.Callable[[FlowHead], None]] = None,
    status_changed_callback: typing.Optional[typing.Callable[[FlowHead], None]] = None,
    _position: int = 0,
    _status: nemoguardrails.colang.v2_x.runtime.flows.FlowHeadStatus = FlowHeadStatus.ACTIVE
)

Dataclass

The flow head that points to a certain element in the flow

_position

int = 0

_status

FlowHeadStatus = FlowHeadStatus.ACTIVE

catch_pattern_failure_label

List[str] = field(default_factory=list)

child_head_uids

List[str] = field(default_factory=list)

flow_state_uid

str

matching_scores

List[float]

position

int

Return the current position of the head.

position_changed_callback

Optional[Callable[[FlowHead], None]] = None

scope_uids

List[str] = field(default_factory=list)

status

FlowHeadStatus

Return the current status of the head.

status_changed_callback

Optional[Callable[[FlowHead], None]] = None

uid

str

nemoguardrails.colang.v2_x.runtime.flows.FlowHead.__eq__(
    other: typing.Any
) -> bool

nemoguardrails.colang.v2_x.runtime.flows.FlowHead.__hash__() -> int

nemoguardrails.colang.v2_x.runtime.flows.FlowHead.__repr__() -> str

nemoguardrails.colang.v2_x.runtime.flows.FlowHead.__str__() -> str

nemoguardrails.colang.v2_x.runtime.flows.FlowHead.get_child_head_uids(
    state: nemoguardrails.colang.v2_x.runtime.flows.State
) -> typing.List[str]

Return uids of all child heads (recursively).

class nemoguardrails.colang.v2_x.runtime.flows.FlowHeadStatus

Bases: enum.Enum

The status of a flow head.

ACTIVE

= 'active'

INACTIVE

= 'inactive'

MERGING

= 'merging'

class nemoguardrails.colang.v2_x.runtime.flows.FlowState(
    uid: str,
    flow_id: str,
    loop_id: typing.Optional[str],
    hierarchy_position: str,
    heads: typing.Dict[str, nemoguardrails.colang.v2_x.runtime.flows.FlowHead] = dict(),
    scopes: typing.Dict[str, typing.Tuple[typing.List[str], typing.List[str]]] = dict(),
    head_fork_uids: typing.Dict[str, str] = dict(),
    action_uids: typing.List[str] = list(),
    context: dict = dict(),
    priority: float = 1.0,
    arguments: typing.Dict[str, typing.Any] = dict(),
    parent_uid: typing.Optional[str] = None,
    parent_head_uid: typing.Optional[str] = None,
    child_flow_uids: typing.List[str] = list(),
    _status: nemoguardrails.colang.v2_x.runtime.flows.FlowStatus = FlowStatus.WAITING,
    status_updated: datetime.datetime = datetime.now(),
    activated: int = 0,
    new_instance_started: bool = False
)

Dataclass

The state of a flow.

_event_name_map

dict = field(init=False)

_status

FlowStatus = FlowStatus.WAITING

action_uids

List[str] = field(default_factory=list)

activated

int = 0

active_heads

Dict[str, FlowHead]

All active heads of this flow.

arguments

Dict[str, Any] = field(default_factory=dict)

child_flow_uids

List[str] = field(default_factory=list)

context

dict = field(default_factory=dict)

flow_id

str

head_fork_uids

Dict[str, str] = field(default_factory=dict)

heads

Dict[str, FlowHead] = field(default_factory=dict)

hierarchy_position

str

loop_id

Optional[str]

new_instance_started

bool = False

parent_head_uid

Optional[str] = None

parent_uid

Optional[str] = None

priority

float = 1.0

scopes

Dict[str, Tuple[List[str], List[str]]] = field(default_factory=dict)

status

FlowStatus

The status of the flow

status_updated

datetime = datetime.now()

uid

str

nemoguardrails.colang.v2_x.runtime.flows.FlowState.__getattr__(
    name
)

nemoguardrails.colang.v2_x.runtime.flows.FlowState.__post_init__() -> None

nemoguardrails.colang.v2_x.runtime.flows.FlowState.__repr__() -> str

nemoguardrails.colang.v2_x.runtime.flows.FlowState._create_out_event(
    event_type: str,
    matching_scores: typing.List[float],
    args: typing.Optional[typing.Dict[str, typing.Any]]
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

nemoguardrails.colang.v2_x.runtime.flows.FlowState.failed_event(
    matching_scores: typing.List[float],
    args: typing.Optional[typing.Dict[str, typing.Any]] = None
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Returns the flow Failed event.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.finish_event(
    matching_scores: typing.List[float],
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Finishes the flow. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.finished_event(
    matching_scores: typing.List[float],
    args: typing.Optional[typing.Dict[str, typing.Any]] = None
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Returns the flow Finished event.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.get_event(
    name: str,
    arguments: dict,
    matching_scores: typing.Optional[typing.List[float]] = None
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Returns the corresponding action event.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.pause_event(
    matching_scores: typing.List[float],
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Pauses the flow. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.resume_event(
    matching_scores: typing.List[float],
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Resumes the flow. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.start_event(
    matching_scores: typing.List[float],
    args: typing.Optional[dict] = None
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Starts the flow. Takes no arguments.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.started_event(
    matching_scores: typing.List[float],
    args: typing.Optional[typing.Dict[str, typing.Any]] = None
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Returns the flow Started event.

nemoguardrails.colang.v2_x.runtime.flows.FlowState.stop_event(
    matching_scores: typing.List[float],
    _args: dict
) -> nemoguardrails.colang.v2_x.runtime.flows.InternalEvent

Stops the flow. Takes no arguments.

class nemoguardrails.colang.v2_x.runtime.flows.FlowStatus

Bases: enum.Enum

The status of a flow.

FINISHED

= 'finished'

STARTED

= 'started'

STARTING

= 'starting'

STOPPED

= 'stopped'

STOPPING

= 'stopping'

WAITING

= 'waiting'

class nemoguardrails.colang.v2_x.runtime.flows.InteractionLoopType

Bases: enum.Enum

The type of the interaction loop.

NAMED

= 'named'

NEW

= 'new'

PARENT

= 'parent'

class nemoguardrails.colang.v2_x.runtime.flows.InternalEvent(
    name: str,
    arguments: dict,
    matching_scores: typing.List[float] = list(),
    flow: typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.FlowState] = None
)

Dataclass

Bases: Event

The internal event class.

flow

Optional[FlowState] = None

class nemoguardrails.colang.v2_x.runtime.flows.InternalEvents()

All internal event types. This event will not appear in the event stream and have priority over them.

ALL

BOT_ACTION_LOG

= 'BotActionLog'

BOT_INTENT_LOG

= 'BotIntentLog'

FINISH_FLOW

= 'FinishFlow'

FLOW_FAILED

= 'FlowFailed'

FLOW_FINISHED

= 'FlowFinished'

FLOW_STARTED

= 'FlowStarted'

START_FLOW

= 'StartFlow'

STOP_FLOW

= 'StopFlow'

UNHANDLED_EVENT

= 'UnhandledEvent'

USER_ACTION_LOG

= 'UserActionLog'

USER_INTENT_LOG

= 'UserIntentLog'

class nemoguardrails.colang.v2_x.runtime.flows.State(
    flow_states: typing.Dict[str, nemoguardrails.colang.v2_x.runtime.flows.FlowState],
    flow_configs: typing.Dict[str, nemoguardrails.colang.v2_x.runtime.flows.FlowConfig],
    rails_config: typing.Optional['RailsConfig'] = None,
    actions: typing.Dict[str, nemoguardrails.colang.v2_x.runtime.flows.Action] = dict(),
    internal_events: typing.Deque[nemoguardrails.colang.v2_x.runtime.flows.Event] = deque(),
    main_flow_state: typing.Optional[nemoguardrails.colang.v2_x.runtime.flows.FlowState] = None,
    context: typing.Dict[str, typing.Any] = dict(),
    outgoing_events: typing.List[dict] = list(),
    last_events: typing.List[typing.Union[dict, nemoguardrails.colang.v2_x.runtime.flows.Event]] = list(),
    context_updates: dict = dict(),
    flow_id_states: typing.Dict[str, typing.List[nemoguardrails.colang.v2_x.runtime.flows.FlowState]] = dict(),
    event_matching_heads: typing.Dict[str, typing.List[typing.Tuple[str, str]]] = dict(),
    event_matching_heads_reverse_map: typing.Dict[str, str] = dict()
)

Dataclass

The state of a flow-driven system.

actions

Dict[str, Action] = field(default_factory=dict)

context

Dict[str, Any] = field(default_factory=dict)

context_updates

dict = field(default_factory=dict)

event_matching_heads

Dict[str, List[Tuple[str, str]]] = field(default_factory=dict)

event_matching_heads_reverse_map

Dict[str, str] = field(default_factory=dict)

flow_configs

Dict[str, FlowConfig]

flow_id_states

Dict[str, List[FlowState]] = field(default_factory=dict)

flow_states

Dict[str, FlowState]

internal_events

Deque[Event] = field(default_factory=deque)

last_events

List[Union[dict, Event]] = field(default_factory=list)

main_flow_state

Optional[FlowState] = None

outgoing_events

List[dict] = field(default_factory=list)

rails_config

Optional['RailsConfig'] = None

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

nemoguardrails.colang.v2_x.runtime.flows.random_seed = int(time.time())