ReferencePython SDK ReferenceNemoguardrailsGuardrails

nemoguardrails.guardrails.engine_registry

View as Markdown

Engine registry for IORails engine.

Manages a collection of ModelEngine and APIEngine instances, one per configured model type. Each engine owns its own RetryClient with per-model settings.

Module Contents

Classes

NameDescription
EngineRegistryRegistry of ModelEngine and APIEngine instances for IORails.

Data

_EngineT

log

API

class nemoguardrails.guardrails.engine_registry.EngineRegistry(
    models: list[nemoguardrails.rails.llm.config.Model],
    rails_config_data: nemoguardrails.rails.llm.config.RailsConfigData,
    tracer: typing.Optional[opentelemetry.trace.Tracer] = None,
    metrics_enabled: bool = False,
    content_capture_enabled: bool = False
)

Registry of ModelEngine and APIEngine instances for IORails.

Creates one engine per configured model or API service, keyed by name. Each engine owns its own HTTP client with per-model retry and timeout settings.

_engines
dict[str, BaseEngine] = {}
nemoguardrails.guardrails.engine_registry.EngineRegistry.__aenter__()
async

Async context manager entry: start all engine clients.

nemoguardrails.guardrails.engine_registry.EngineRegistry.__aexit__(
    exc_type,
    exc_val,
    exc_tb
)
async

Async context manager exit: stop all engine clients.

nemoguardrails.guardrails.engine_registry.EngineRegistry._get_engine(
    name: str,
    expected_type: type[nemoguardrails.guardrails.engine_registry._EngineT]
) -> nemoguardrails.guardrails.engine_registry._EngineT

Look up an engine by name, verifying its type.

nemoguardrails.guardrails.engine_registry.EngineRegistry.api_call(
    api_name: str,
    message: dict[str, typing.Any],
    kwargs: typing.Any = {}
) -> dict[str, typing.Any]
async

Route an API request to the named API engine.

Raises:

  • KeyError: If no engine is registered with the given name.
  • TypeError: If the named engine is not an APIEngine.
nemoguardrails.guardrails.engine_registry.EngineRegistry.model_call(
    model_type: str,
    messages: list[dict],
    kwargs: typing.Any = {}
) -> nemoguardrails.types.LLMResponse
async

Route a chat completion request to the named model engine.

Returns the structured LLMResponse from the engine — content, reasoning (when the provider exposes it), usage, finish reason. Callers that only want the assistant text should access .content.

When metrics are enabled, emits gen_ai.client.operation.duration (with error.type on exception) and gen_ai.client.token.usage (one observation each for input and output token types, only when LLMResponse.usage is populated).

Raises:

  • KeyError: If no engine is registered with the given name.
  • TypeError: If the named engine is not a ModelEngine.
nemoguardrails.guardrails.engine_registry.EngineRegistry.start() -> None
async

Start all engine clients. Call this during service startup.

nemoguardrails.guardrails.engine_registry.EngineRegistry.stop() -> None
async

Stop all engine clients. Call this during service shutdown.

nemoguardrails.guardrails.engine_registry.EngineRegistry.stream_model_call(
    model_type: str,
    messages: list[dict],
    kwargs: typing.Any = {}
) -> collections.abc.AsyncGenerator[nemoguardrails.types.LLMResponseChunk, None]
async

Stream chat completion chunks from the named model engine.

Yields LLMResponseChunk objects. The surrounding llm_call_span wraps the full generator lifetime: it opens before the first chunk and closes when the generator exhausts or raises.

When metrics are enabled, emits gen_ai.client.operation.duration for the full stream lifetime (with error.type on exception) and gen_ai.client.token.usage after stream completion using the UsageInfo carried on the terminal SSE chunk (when the provider returns one — controlled by include_usage_in_stream, defaults to True for OpenAI-compatible engines). No token observation is emitted on early consumer cancellation or on provider error mid-stream.

Raises:

  • KeyError: If no engine is registered with the given name.
  • TypeError: If the named engine is not a ModelEngine.
nemoguardrails.guardrails.engine_registry._EngineT = TypeVar('_EngineT', bound=BaseEngine)
 
nemoguardrails.guardrails.engine_registry.log = logging.getLogger(__name__)