windlass.providers.guardrails.nemo

nemo

NVIDIA NeMo Guardrails adapter.

NeMo Guardrails brings conversational policy that regexes cannot express: topical rails ("never discuss competitors"), dialogue flows written in Colang, fact-checking rails, and model-based jailbreak detection.

Install with::

pip install "windlass[guardrails]"

Windlass gives NeMo a sensible default configuration so .guardrails('nemo') works out of the box, and lets you point at your own Colang config directory when you outgrow it.

Example

from windlass import Windlass # doctest: +SKIP agent = Windlass.agent().guardrails("nemo", config_path="./rails") # doctest: +SKIP

NeMoGuardrail

NeMoGuardrail(
    *,
    config_path: str | Path | None = None,
    colang: str | None = None,
    yaml_config: str | None = None,
    llm: Any = None,
    on_violation: str = "block",
    stages: tuple[str, ...] = ("input", "output"),
    **config: Any
)

Bases: Guardrail

Guardrail backed by NeMo Guardrails.

Parameters:

Name	Type	Description	Default
config_path	str | Path | None	Directory holding config.yml and *.co files. When omitted, :data:DEFAULT_COLANG and :data:DEFAULT_YAML are used.	None
colang	str | None	Inline Colang content, as an alternative to config_path.	None
yaml_config	str | None	Inline YAML rails configuration.	None
llm	Any	Model NeMo should use for its own rails. Windlass passes its configured model through so both layers agree.	None
on_violation	str	block, redact, warn or allow.	'block'
stages	tuple[str, ...]	Which stages to run at.	('input', 'output')
**config	Any	Forwarded to :class:~windlass.interfaces.guardrail.Guardrail.	{}

Raises:

Type	Description
MissingDependencyError	When nemoguardrails is not installed.
ConfigurationError	When the configuration cannot be loaded.

Performance

NeMo rails involve extra model calls, so expect meaningful added latency per request. Run :class:~windlass.providers.guardrails.rules.RuleGuardrail first for the cheap deterministic checks, and reserve NeMo for policy that genuinely needs a model.

Source code in src\windlass\providers\guardrails\nemo.py

def __init__(
    self,
    *,
    config_path: str | Path | None = None,
    colang: str | None = None,
    yaml_config: str | None = None,
    llm: Any = None,
    on_violation: str = "block",
    stages: tuple[str, ...] = ("input", "output"),
    **config: Any,
) -> None:
    super().__init__(on_violation=on_violation, stages=stages, **config)
    nemo = require("nemoguardrails", extra="guardrails", feature="NVIDIA NeMo Guardrails")
    self._nemo = nemo
    try:
        if config_path:
            path = Path(config_path)
            if not path.is_dir():
                raise ConfigurationError(
                    f"NeMo config directory not found: {path}",
                    hint="Point config_path at a directory containing config.yml.",
                )
            rails_config = nemo.RailsConfig.from_path(str(path))
        else:
            rails_config = nemo.RailsConfig.from_content(
                colang_content=colang or DEFAULT_COLANG,
                yaml_content=yaml_config or DEFAULT_YAML,
            )
        self._rails = nemo.LLMRails(rails_config, llm=_unwrap(llm))
    except ConfigurationError:
        raise
    except Exception as exc:
        raise ConfigurationError(
            f"Could not initialise NeMo Guardrails: {exc}",
            hint="Check your Colang syntax and that any models it references "
            "are configured.",
        ) from exc

native

native() -> Any

Return the underlying LLMRails instance (Level 3 access).

Source code in src\windlass\providers\guardrails\nemo.py

def native(self) -> Any:
    """Return the underlying ``LLMRails`` instance (Level 3 access)."""
    return self._rails

acheck async

acheck(
    content: str, *, stage: str = "input", context: dict[str, Any] | None = None
) -> GuardrailResult

Run the configured rails against content.

A rail that refuses produces a different response than the input; that difference is how a violation is detected, since NeMo's public API returns a message rather than a verdict.

Parameters:

Name	Type	Description	Default
content	str	The text to inspect.	required
stage	str	"input" or "output".	'input'
context	dict[str, Any] | None	Extra signals, forwarded to NeMo as conversation context.	None

Returns:

Type	Description
GuardrailResult	The verdict. When a rail fires, content holds NeMo's refusal
GuardrailResult	message.

Source code in src\windlass\providers\guardrails\nemo.py

async def acheck(
    self,
    content: str,
    *,
    stage: str = "input",
    context: dict[str, Any] | None = None,
) -> GuardrailResult:
    """Run the configured rails against ``content``.

    A rail that refuses produces a different response than the input; that
    difference is how a violation is detected, since NeMo's public API
    returns a message rather than a verdict.

    Args:
        content: The text to inspect.
        stage: ``"input"`` or ``"output"``.
        context: Extra signals, forwarded to NeMo as conversation context.

    Returns:
        The verdict. When a rail fires, ``content`` holds NeMo's refusal
        message.
    """
    try:
        response = await self._rails.generate_async(
            messages=[{"role": "user", "content": content}],
            options={"rails": [stage]} if stage in {"input", "output"} else None,
        )
    except Exception as exc:
        self._log.warning("NeMo Guardrails check failed, allowing content: %s", exc)
        return GuardrailResult(allowed=True, content=content, stage=stage)

    message = response if isinstance(response, str) else (response or {}).get("content", "")
    blocked = bool(message) and _is_refusal(message, content)

    return GuardrailResult(
        allowed=not blocked,
        content=message if blocked else content,
        detections=[{"rule": "nemo_rail", "response": message[:200]}] if blocked else [],
        rule="nemo_rail" if blocked else None,
        stage=stage,
    )