> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.nvidia.com/nemo/guardrails/llms.txt.
> For full documentation content, see https://docs.nvidia.com/nemo/guardrails/llms-full.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.nvidia.com/nemo/guardrails/_mcp/server.

# nemoguardrails.colang.v2_x.runtime.flows

A simplified modeling of the CoFlows engine.

## Module Contents

### Classes

| Name                                                                                   | Description                                                                                           |
| -------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------- |
| [`Action`](#nemoguardrails-colang-v2_x-runtime-flows-Action)                           | The action class groups and manages the action events.                                                |
| [`ActionEvent`](#nemoguardrails-colang-v2_x-runtime-flows-ActionEvent)                 | The action event class.                                                                               |
| [`ActionStatus`](#nemoguardrails-colang-v2_x-runtime-flows-ActionStatus)               | The status of an action.                                                                              |
| [`Event`](#nemoguardrails-colang-v2_x-runtime-flows-Event)                             | The base event class.                                                                                 |
| [`FlowConfig`](#nemoguardrails-colang-v2_x-runtime-flows-FlowConfig)                   | The configuration of a flow.                                                                          |
| [`FlowHead`](#nemoguardrails-colang-v2_x-runtime-flows-FlowHead)                       | The flow head that points to a certain element in the flow                                            |
| [`FlowHeadStatus`](#nemoguardrails-colang-v2_x-runtime-flows-FlowHeadStatus)           | The status of a flow head.                                                                            |
| [`FlowState`](#nemoguardrails-colang-v2_x-runtime-flows-FlowState)                     | The state of a flow.                                                                                  |
| [`FlowStatus`](#nemoguardrails-colang-v2_x-runtime-flows-FlowStatus)                   | The status of a flow.                                                                                 |
| [`InteractionLoopType`](#nemoguardrails-colang-v2_x-runtime-flows-InteractionLoopType) | The type of the interaction loop.                                                                     |
| [`InternalEvent`](#nemoguardrails-colang-v2_x-runtime-flows-InternalEvent)             | The internal event class.                                                                             |
| [`InternalEvents`](#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. |
| [`State`](#nemoguardrails-colang-v2_x-runtime-flows-State)                             | The state of a flow-driven system.                                                                    |

### Data

[`log`](#nemoguardrails-colang-v2_x-runtime-flows-log)

[`random_seed`](#nemoguardrails-colang-v2_x-runtime-flows-random_seed)

### API

```python
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.

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

```python
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.

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

Returns the Finished action event.

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

staticmethod

```python
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.

```python
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.

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

Processes event and updates action accordingly.

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

Starts the action. Takes no arguments.

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

Returns the Started action event.

```python
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.

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

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

Returns the Updated parameter action event.

```python
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](#nemoguardrails-colang-v2_x-runtime-flows-Event)

The action event class.

```python
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.

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

**Bases:** `enum.Enum`

The status of an action.

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

Dataclass

The base event class.

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

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

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

```python
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.

```python
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.

```python
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.

Return True if flow is marked as override.

Return the interaction loop id if set.

Return the interaction loop priority (default: 0).

Return the interaction loop type.

```python
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)`.

```python
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.

```python
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

Return the current position of the head.

Return the current status of the head.

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

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

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

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

```python
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).

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

**Bases:** `enum.Enum`

The status of a flow head.

```python
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.

All active heads of this flow.

The status of the flow

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

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

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

```python
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
```

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

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

**Bases:** `enum.Enum`

The status of a flow.

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

**Bases:** `enum.Enum`

The type of the interaction loop.

```python
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](#nemoguardrails-colang-v2_x-runtime-flows-Event)

The internal event class.

```python
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.

```python
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.

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

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