> 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.v1_0.runtime.flows

A simplified modeling of the CoFlows engine.

## Module Contents

### Classes

| Name                                                                 | Description                      |
| -------------------------------------------------------------------- | -------------------------------- |
| [`FlowConfig`](#nemoguardrails-colang-v1_0-runtime-flows-FlowConfig) | The configuration of a flow.     |
| [`FlowState`](#nemoguardrails-colang-v1_0-runtime-flows-FlowState)   | The state of a flow.             |
| [`FlowStatus`](#nemoguardrails-colang-v1_0-runtime-flows-FlowStatus) | The status of a flow.            |
| [`State`](#nemoguardrails-colang-v1_0-runtime-flows-State)           | A state of a flow-driven system. |

### Functions

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

### API

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

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

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

**Bases:** `enum.Enum`

The status of a flow.

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

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

The current state of the system.

The state of the current flow.

**Returns:** `Optional[FlowState]`

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

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

The flow ID to check.

**Returns:** `bool`

True if the flow ID contains arguments, False otherwise.

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

Return the arguments in a flow id as a dictionary.

**Parameters:**

The flow id.

**Returns:** `dict`

A dictionary of arguments in the flow id.

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

Checks if the given element is actionable.

**Parameters:**

The element to be checked.

**Returns:** `bool`

True if the element is actionable, False otherwise.

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

Checks if the given element matches the given event.

**Parameters:**

The element to be checked for a match.

The event to compare against.

**Returns:** `bool`

True if the element matches the event, False otherwise.

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

The flow id.

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

The current state to update.

The state of the current flow.

The configuration of the current flow.

Priority modifier. Defaults to 1.0.

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

The current state of the system.

The state of the current flow.

**Returns:** `Optional[int]`

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

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

The next step from a flow element.

**Returns:** `dict`

The corresponding event.

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

The history of events.

**Returns:**

The computed context.

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

The current state of the system.

The event triggering the computation.

**Returns:** `State`

The updated state of the system.

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

The history of events.

Flow configurations.

Rails configuration.

The processing log so far. This will be mutated.

**Returns:** `List[dict]`

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