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

A simplified modeling of the CoFlows engine.

Module Contents

Classes

Name	Description
`FlowConfig`	The configuration of a flow.
`FlowState`	The state of a flow.
`FlowStatus`	The status of a flow.
`State`	A state of a flow-driven system.

Functions

Name	Description
`_call_subflow`	Helper to call a subflow.
`_flow_id_has_params`	Check if a flow ID contains arguments.
`_get_flow_params`	Return the arguments in a flow id as a dictionary.
`_is_actionable`	Checks if the given element is actionable.
`_is_match`	Checks if the given element matches the given event.
`_normalize_flow_id`	Normalize the flow id by removing the arguments from the id.
`_record_next_step`	Helper to record the next step.
`_slide_with_subflows`	Slides the provided flow and also calls subflows, if applicable.
`_step_to_event`	Converts a next step from a flow element into an actual event.
`compute_context`	Computes the context given a history of events.
`compute_next_state`	Computes the next state of the flow-driven system.
`compute_next_steps`	Computes the next step in a flow-driven system given a history of events.

API

class nemoguardrails.colang.v1_0.runtime.flows.FlowConfig(
    id: str,
    elements: typing.List[dict],
    priority: float = 1.0,
    is_extension: bool = False,
    is_interruptible: bool = True,
    is_subflow: bool = False,
    allow_multiple: bool = False,
    trigger_event_types: typing.List[str] = (lambda: ['UserIntent', 'Bo...,
    source_code: typing.Optional[str] = None
)

Dataclass

The configuration of a flow.

allow_multiple

bool = False

elements

List[dict]

str

is_extension

bool = False

is_interruptible

bool = True

is_subflow

bool = False

priority

float = 1.0

source_code

Optional[str] = None

trigger_event_types

List[str]

class nemoguardrails.colang.v1_0.runtime.flows.FlowState(
    uid: str,
    flow_id: str,
    head: int,
    status: nemoguardrails.colang.v1_0.runtime.flows.FlowStatus = FlowStatus.ACTIVE
)

Dataclass

The state of a flow.

flow_id

str

head

int

status

FlowStatus = FlowStatus.ACTIVE

uid

str

class nemoguardrails.colang.v1_0.runtime.flows.FlowStatus

Bases: enum.Enum

The status of a flow.

ABORTED

= 'aborted'

ACTIVE

= 'active'

COMPLETED

= 'completed'

INTERRUPTED

= 'interrupted'

class nemoguardrails.colang.v1_0.runtime.flows.State(
    context: dict,
    flow_states: typing.List[nemoguardrails.colang.v1_0.runtime.flows.FlowState],
    flow_configs: typing.Dict[str, nemoguardrails.colang.v1_0.runtime.flows.FlowConfig],
    rails_config: typing.Optional[nemoguardrails.rails.RailsConfig] = None,
    next_step: typing.Optional[dict] = None,
    next_step_by_flow_uid: typing.Optional[str] = None,
    next_step_priority: float = 0.0,
    next_step_comment: typing.Optional[str] = None,
    context_updates: dict = dict()
)

Dataclass

A state of a flow-driven system.

context

dict

context_updates

dict = field(default_factory=dict)

flow_configs

Dict[str, FlowConfig]

flow_states

List[FlowState]

next_step

Optional[dict] = None

next_step_by_flow_uid

Optional[str] = None

next_step_comment

Optional[str] = None

next_step_priority

float = 0.0

rails_config

Optional[RailsConfig] = None

nemoguardrails.colang.v1_0.runtime.flows._call_subflow(
    new_state: nemoguardrails.colang.v1_0.runtime.flows.State,
    flow_state: nemoguardrails.colang.v1_0.runtime.flows.FlowState
) -> typing.Optional[nemoguardrails.colang.v1_0.runtime.flows.FlowState]

Helper to call a subflow.

The head for flow_state is expected to be on a “flow” element.

Parameters:

new_state

State

The current state of the system.

flow_state

FlowState

The state of the current flow.

Returns: Optional[FlowState]

Optional[FlowState]: The state of the subflow, if applicable.

nemoguardrails.colang.v1_0.runtime.flows._flow_id_has_params(
    flow_id: str
) -> bool

Check if a flow ID contains arguments.

A flow ID is considered to contain arguments if it contains both ”(” and ”)”.

Parameters:

flow_id

str

The flow ID to check.

Returns: bool

True if the flow ID contains arguments, False otherwise.

nemoguardrails.colang.v1_0.runtime.flows._get_flow_params(
    flow_id: str
) -> dict

Return the arguments in a flow id as a dictionary.

Parameters:

flow_id

str

The flow id.

Returns: dict

A dictionary of arguments in the flow id.

nemoguardrails.colang.v1_0.runtime.flows._is_actionable(
    element: dict
) -> bool

Checks if the given element is actionable.

Parameters:

element

dict

The element to be checked.

Returns: bool

True if the element is actionable, False otherwise.

nemoguardrails.colang.v1_0.runtime.flows._is_match(
    element: dict,
    event: dict
) -> bool

Checks if the given element matches the given event.

Parameters:

element

dict

The element to be checked for a match.

event

dict

The event to compare against.

Returns: bool

True if the element matches the event, False otherwise.

nemoguardrails.colang.v1_0.runtime.flows._normalize_flow_id(
    flow_id: str
) -> str

Normalize the flow id by removing the arguments from the id.

Example:

flow_id = “flow_id_v1(arg1, arg2)” _normalize_flow_id(flow_id) -> “flow_id_v1”

Parameters:

flow_id

str

The flow id.

nemoguardrails.colang.v1_0.runtime.flows._record_next_step(
    new_state: nemoguardrails.colang.v1_0.runtime.flows.State,
    flow_state: nemoguardrails.colang.v1_0.runtime.flows.FlowState,
    flow_config: nemoguardrails.colang.v1_0.runtime.flows.FlowConfig,
    priority_modifier: float = 1.0
)

Helper to record the next step.

Parameters:

new_state

State

The current state to update.

flow_state

FlowState

The state of the current flow.

flow_config

FlowConfig

The configuration of the current flow.

priority_modifier

floatDefaults to 1.0

Priority modifier. Defaults to 1.0.

nemoguardrails.colang.v1_0.runtime.flows._slide_with_subflows(
    state: nemoguardrails.colang.v1_0.runtime.flows.State,
    flow_state: nemoguardrails.colang.v1_0.runtime.flows.FlowState
) -> typing.Optional[int]

Slides the provided flow and also calls subflows, if applicable.

Parameters:

state

State

The current state of the system.

flow_state

FlowState

The state of the current flow.

Returns: Optional[int]

Optional[int]: The new head position of the flow, if applicable.

nemoguardrails.colang.v1_0.runtime.flows._step_to_event(
    step: dict
) -> dict

Converts a next step from a flow element into an actual event.

Parameters:

step

dict

The next step from a flow element.

Returns: dict

The corresponding event.

nemoguardrails.colang.v1_0.runtime.flows.compute_context(
    history: typing.List[dict]
)

Computes the context given a history of events.

Special context variables:

$last_user_message: the last message sent by the user.
$last_bot_message: the last message sent by the bot.

Parameters:

history

List[dict]

The history of events.

Returns:

The computed context.

nemoguardrails.colang.v1_0.runtime.flows.compute_next_state(
    state: nemoguardrails.colang.v1_0.runtime.flows.State,
    event: dict
) -> nemoguardrails.colang.v1_0.runtime.flows.State

Computes the next state of the flow-driven system.

Currently, this is a very simplified implementation, with the following assumptions:

All flows are singleton i.e. you can’t have multiple instances of the same flow.
Flows can be interrupted by one flow at a time.
Flows are resumed when the interruption flow completes.
No prioritization between flows, the first one that can decide something will be used.

Parameters:

state

State

The current state of the system.

event

dict

The event triggering the computation.

Returns: State

The updated state of the system.

nemoguardrails.colang.v1_0.runtime.flows.compute_next_steps(
    history: typing.List[dict],
    flow_configs: typing.Dict[str, nemoguardrails.colang.v1_0.runtime.flows.FlowConfig],
    rails_config: nemoguardrails.rails.RailsConfig,
    processing_log: typing.List[dict]
) -> typing.List[dict]

Computes the next step in a flow-driven system given a history of events.

Parameters:

history

List[dict]

The history of events.

flow_configs

Dict[str, FlowConfig]

Flow configurations.

rails_config

RailsConfig

Rails configuration.

processing_log

List[dict]

The processing log so far. This will be mutated.

Returns: List[dict]

List[dict]: The list of computed next steps.