nemoguardrails.guardrails.rails_manager | NVIDIA NeMo Guardrails Library Developer Guide

Rails manager for IORails engine.

Orchestrates input/output safety checks by delegating to RailAction instances. Rails run sequentially by default; the first failing rail short-circuits. When parallel mode is enabled, all rails run concurrently and the first unsafe result cancels remaining rails immediately.

Module Contents

Classes

Name	Description
`RailsManager`	Orchestrates input and output safety checks for IORails.

Data

_ACTION_CLASSES

_TOOL_ACTION_CLASSES

_ToolActionT

log

API

class nemoguardrails.guardrails.rails_manager.RailsManager(
    engine_registry: nemoguardrails.guardrails.engine_registry.EngineRegistry,
    task_manager: nemoguardrails.llm.taskmanager.LLMTaskManager,
    input_flows: list[str],
    output_flows: list[str],
    input_parallel: bool = False,
    output_parallel: bool = False,
    tool_call_flows: typing.Optional[list[str]] = None,
    tool_result_flows: typing.Optional[list[str]] = None,
    tracer: typing.Optional[opentelemetry.trace.Tracer] = None,
    content_capture_enabled: bool = False
)

Orchestrates input and output safety checks for IORails.

Reads the rails configuration to determine which checks are enabled, instantiates the corresponding RailAction for each flow, then runs them sequentially or in parallel.

_actions

dict[str, RailAction] = {}

_tool_call_actions

_tool_result_actions

input_flows

list[str] = list(input_flows)

output_flows

list[str] = list(output_flows)

tool_call_flows

list[str] = list(tool_call_flows or [])

tool_result_flows

list[str] = list(tool_result_flows or [])

nemoguardrails.guardrails.rails_manager.RailsManager._build_tool_actions(
    flows: list[str],
    expected_cls: type[nemoguardrails.guardrails.rails_manager._ToolActionT]
) -> dict[str, nemoguardrails.guardrails.rails_manager._ToolActionT]

Instantiate the tool rails for flows, checking each resolves to expected_cls.

Raises RuntimeError on a duplicate flow, an unknown flow, or a flow that resolves to the wrong direction. Duplicates are rejected because the dispatch keys its coroutine map by flow, so a repeated flow would silently drop a run.

nemoguardrails.guardrails.rails_manager.RailsManager._create_action(
    base_name: str
) -> nemoguardrails.guardrails.rail_action.RailAction

Instantiate the RailAction for a given flow base name.

nemoguardrails.guardrails.rails_manager.RailsManager._enabled_flows(
    configured: list[str],
    enabled: typing.Union[bool, list[str]]
) -> list[str]

staticmethod

Resolve the per-request enable toggle into the configured flows to run.

True (the default) runs every configured flow; False runs none; a list runs only the named flows that are configured, preserving configured order and ignoring unknown names. The two booleans are spelled out as separate cases so a non-empty list is never mistaken for True.

List membership is compared on the normalized flow name (_get_flow_name), the same way _create_action, _build_tool_actions and unsupported_reason do, so a request toggle carrying the canonical rail name matches a configured flow that carries a $model=/(...) suffix instead of silently dropping it (fail-open). Shared by the input, output, and tool rail families.

nemoguardrails.guardrails.rails_manager.RailsManager._run_rail(
    flow: str,
    direction: nemoguardrails.guardrails.guardrails_types.RailDirection,
    messages: list[dict],
    bot_response: typing.Optional[str] = None
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Dispatch a single rail flow to its RailAction instance.

nemoguardrails.guardrails.rails_manager.RailsManager._run_rails_parallel(
    rails: collections.abc.Mapping[str, collections.abc.Coroutine[typing.Any, typing.Any, nemoguardrails.guardrails.guardrails_types.RailResult]],
    direction: nemoguardrails.guardrails.guardrails_types.RailDirection
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Run rail coroutines concurrently, cancelling remaining on first unsafe result.

nemoguardrails.guardrails.rails_manager.RailsManager._run_rails_sequential(
    rails: collections.abc.Mapping[str, collections.abc.Coroutine[typing.Any, typing.Any, nemoguardrails.guardrails.guardrails_types.RailResult]],
    direction: nemoguardrails.guardrails.guardrails_types.RailDirection
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Run rail coroutines sequentially, short-circuiting on first unsafe result.

nemoguardrails.guardrails.rails_manager.RailsManager._run_tool_call_rail(
    flow: str,
    tool_calls: list[nemoguardrails.types.ToolCall],
    toolset: nemoguardrails.guardrails.tool_schema.Toolset
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Dispatch a single tool-call rail to its action, wrapped in an OUTPUT rail span.

nemoguardrails.guardrails.rails_manager.RailsManager._run_tool_result_rail(
    flow: str,
    exchanges: list[nemoguardrails.guardrails.tool_schema.ToolExchange]
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Validate each turn’s results against that turn’s calls, wrapped in an INPUT rail span.

Each exchange is validated independently so call_id linkage stays turn-local; the first unsafe exchange short-circuits.

nemoguardrails.guardrails.rails_manager.RailsManager.are_tool_calls_safe(
    tool_calls: list[nemoguardrails.types.ToolCall],
    llm_params: typing.Optional[dict],
    enabled: typing.Union[bool, list[str]] = True,
    model_type: str = 'main'
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Validate the model’s emitted tool calls (OUTPUT-direction tool rail).

The tool-call counterpart to :meth:is_output_safe: takes the model’s output (tool_calls) plus the request’s declared tools (llm_params) and returns a RailResult.

nemoguardrails.guardrails.rails_manager.RailsManager.are_tool_results_safe(
    messages: list[dict],
    enabled: typing.Union[bool, list[str]] = True,
    model_type: str = 'main'
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Validate incoming tool results (INPUT-direction tool rail).

The tool-result counterpart to :meth:is_input_safe: takes the conversation messages and returns a RailResult. Groups the conversation into per-turn (calls, results) exchanges via the engine adapter and validates each result against its own turn’s calls, so call ids reused across turns (spec-allowed) are not flagged as ambiguous duplicates.

nemoguardrails.guardrails.rails_manager.RailsManager.is_input_safe(
    messages: list[dict],
    enabled: typing.Union[bool, list[str]] = True
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Run the enabled input rails, short-circuiting on the first failure.

The per-request enabled toggle selects which configured input rails run: True (the default) runs all, False runs none, and a list runs only the named flows (matched on the normalized flow name). When parallel mode is enabled, all selected rails run concurrently and the first unsafe result cancels the rest.

nemoguardrails.guardrails.rails_manager.RailsManager.is_output_safe(
    messages: list[dict],
    response: str,
    enabled: typing.Union[bool, list[str]] = True
) -> nemoguardrails.guardrails.guardrails_types.RailResult

async

Run the enabled output rails, short-circuiting on the first failure.

The per-request enabled toggle selects which configured output rails run: True (the default) runs all, False runs none, and a list runs only the named flows (matched on the normalized flow name). When parallel mode is enabled, all selected rails run concurrently and the first unsafe result cancels the rest.

nemoguardrails.guardrails.rails_manager._ACTION_CLASSES: dict[str, type[RailAction]] = {(cls.action_name): cls for cls in [ContentSafetyInputAction, ContentSafetyOutpu...

nemoguardrails.guardrails.rails_manager._TOOL_ACTION_CLASSES: dict[str, type[ToolRailAction]] = {(cls.action_name): cls for cls in [ToolCallRailAction, ToolResultRailAction]}

nemoguardrails.guardrails.rails_manager._ToolActionT = TypeVar('_ToolActionT', bound=ToolRailAction)

nemoguardrails.guardrails.rails_manager.log = logging.getLogger(__name__)