Index

A package to handle the guardrail of the text.

BaseGuardrailEngineConfig

Bases: BaseModel

Base configuration for guardrail engines.

This config determines which types of content the engine will check. Engines can be configured to check input only, output only, or both.

Attributes:

Name	Type	Description
guardrail_mode	GuardrailMode	Specifies what content the engine should check. - DISABLED: Skip this engine entirely - no checks performed - INPUT_ONLY: Check only user input (queries, prompts, context) - OUTPUT_ONLY: Check only system output (LLM responses, generated text) - BOTH: Check both input and output content

DMGuardrailEngine(dm_invoker, engine_config=None)

Bases: DMComponent, GuardrailEngine

Guardrail engine that uses a decision model to evaluate atomic safety policies.

This engine targets Noul-capable decision models. Each policy key is used as the stable question identifier when invoking the model and when reading the answer from DMOutput.answers.

Attributes:

Name	Type	Description
config	DMGuardrailEngineConfig	Engine configuration.
dm_invoker	BaseDMInvoker	Decisions model invoker used for evaluations.
policies	dict[str, DMPolicyConfig]	Loaded and validated policy configurations.
questions	dict[str, DMQuestion]	Mapping of policy keys to DMQuestion.noul instances.

Initialize the DMGuardrailEngine.

Parameters:

Name	Type	Description	Default
dm_invoker	BaseDMInvoker	Pre-built decisions model invoker.	required
engine_config	DMGuardrailEngineConfig | None	Engine policy and template configuration. Defaults to an empty config (bundled policies).	None

Raises:

Type	Description
ValueError	If no policies are configured or an enabled category is unknown.

check_input(content, *, trace_attributes=None, **kwargs) async

Check input content against configured decision-model policies.

Parameters:

Name	Type	Description	Default
content	str	The input text to evaluate for safety.	required
trace_attributes	dict[str, Any] | None	Additional trace attributes. Defaults to None.	None
**kwargs	Any	hyperparameters forwarded to the DM invoker. Defaults to empty.	{}

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	Safe or unsafe result with category, policy, and score metadata.

check_output(content, *, trace_attributes=None, **kwargs) async

Check output content against configured decision-model policies.

Parameters:

Name	Type	Description	Default
content	str	The output text to evaluate for safety.	required
trace_attributes	dict[str, Any] | None	Additional trace attributes. Defaults to None.	None
**kwargs	Any	hyperparameters forwarded to the DM invoker. Defaults to empty.	{}

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	Safe or unsafe result with category, policy, and score metadata.

DMGuardrailEngineConfig

Bases: BaseGuardrailEngineConfig

Configuration for DMGuardrailEngine.

Attributes:

Name	Type	Description
policy_config_path	str | None	Path to a custom YAML policy file. Defaults to None.
policies	dict[str, DMPolicyConfig] | None	In-memory override of YAML policies. Defaults to None.
enabled_categories	list[str] | None	Category names of the loaded policies to check, including their aliases. None enables all policies; an empty list disables all policies. Defaults to None.
default_threshold	float	Probability threshold used when a policy omits one. Defaults to 0.5, tuned for the bundled openrouter/typesafe/jev-1.13 model.
state_template	dict[str, Any] | None	Decisions state template containing a {content} placeholder. Defaults to None.

GuardrailEngine

Bases: Protocol

Base engine interface for guardrail engines.

This protocol defines the contract that all guardrail engines must implement. Engines can check content for safety violations using various techniques such as phrase matching, topic classification, or external API calls.

All engines must be asynchronous to support both sync and async guardrail providers.

Attributes:

Name	Type	Description
config	BaseGuardrailEngineConfig	Engine configuration specifying what content types to check

check_input(content, **kwargs) async

Check input content for safety violations.

This method should implement provider-specific safety checks on user input such as query content, context, or prompts before sending to LLM.

Parameters:

Name	Type	Description	Default
content	str	The input text to evaluate for safety	required
**kwargs		Additional engine-specific parameters (optional)	{}

Returns:

Type	Description
GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe

Raises:

Type	Description
Exception	If the safety check fails due to engine errors

check_output(content, **kwargs) async

Check output content for safety violations.

This method should implement provider-specific safety checks on system output such as LLM responses, generated text, or static messages before showing to users.

Parameters:

Name	Type	Description	Default
content	str	The output text to evaluate for safety	required
**kwargs		Additional engine-specific parameters (optional)	{}

Returns:

Type	Description
GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe

Raises:

Type	Description
Exception	If the safety check fails due to engine errors

GuardrailInput

Bases: BaseModel

Guardrail input data model.

This model represents the input data for guardrail operations. Use this when you need to check both input and output in a single call, or when you want to be explicit about which content to check.

Example

Check user query only

input_only = GuardrailInput(input="Tell me about AI", output=None)

Check LLM response only

output_only = GuardrailInput(input=None, output="AI is artificial intelligence...")

Check both query and response

both = GuardrailInput( input="Tell me about AI", output="AI is artificial intelligence..." )

from_dict(data) classmethod

Create from dictionary format for backward compatibility.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary with input and output keys	required

Returns:

Type	Description
GuardrailInput	GuardrailInput instance

GuardrailManager(engine=None, empty_content_safe=None, error_conservative=None, raise_on_exceptions=None, max_stream_buffer_bytes=None, max_stream_buffer_chunks=None)

Bases: Component

Manage content guardrails using pluggable engines.

This component provides content filtering and safety checks and delegates the provider-specific logic to a GuardrailEngine.

Attributes:

Name	Type	Description
engines	list[GuardrailEngine]	Pluggable engines for guardrail operations. Defaults to a lazily-built DMGuardrailEngine configured with the bundled DM policies when no engine is provided. Provide a custom engine or a list of engines to handle guardrail logic. If a list is provided, engines are executed sequentially in the order they appear.
empty_content_safe	bool | None	Whether empty content should be considered safe. Defaults to None.
error_conservative	bool | None	Whether to mark content as unsafe on errors. Defaults to None.
raise_on_exceptions	tuple[type[Exception], ...]	Tuple of exception types that should propagate to the caller instead of being handled as safe/unsafe. Defaults to ().
max_stream_buffer_bytes	int | None	Maximum cumulative UTF-8 bytes each strict stream segment may buffer. Defaults to 1,048,576.
max_stream_buffer_chunks	int | None	Maximum source chunks each strict stream segment may buffer. Defaults to 10,000.

Initialize the GuardrailManager class.

Parameters:

Name	Type	Description	Default
engine	GuardrailEngine | list[GuardrailEngine] | None	Pluggable engine for guardrail operations. Defaults to None, in which case a DMGuardrailEngine is built lazily via DMGuardrailEngine.from_config (bundled DM policies, model openrouter/typesafe/jev-1.13, credentials from the OPENROUTER_API_KEY environment variable). Provide a custom engine or a list of engines to handle guardrail logic. If a list is provided, engines are executed sequentially in the order they appear.	None
empty_content_safe	bool | None	Whether empty content should be considered safe. Defaults to True.	None
error_conservative	bool | None	Whether to mark content as unsafe on errors. Defaults to True.	None
raise_on_exceptions	tuple[type[Exception], ...] | None	Exception types to propagate to the caller when raised by an engine. Defaults to ().	None
max_stream_buffer_bytes	int | None	Maximum cumulative UTF-8 bytes each strict stream segment may buffer. Defaults to None, in which case the default value is used.	None
max_stream_buffer_chunks	int | None	Maximum source chunks each strict stream segment may buffer. Defaults to None, in which case the default value is used.	None

Raises:

Type	Description
TypeError	If raise_on_exceptions is not a tuple or contains non-Exception subclasses.
ValueError	If buffer limits are not positive, non-bool integers.

engines property

Return configured engines, lazily building the default when none was provided.

Returns:

Type	Description
list[GuardrailEngine]	list[GuardrailEngine]: Active guardrail engines in evaluation order.

check_content(content, engine_kwargs=None) async

Check the content for safety using configured engines.

This method iterates through the configured engines sequentially and returns the first unsafe result found. This "fail-fast" approach ensures efficient processing while maintaining strict safety guarantees.

Engines with DISABLED mode are skipped entirely. For active engines, it checks input content if the engine's guardrail_mode includes input checking, then checks output content if the input passed and the engine's mode includes output checking.

Parameters:

Name	Type	Description	Default
content	str | GuardrailInput	The primary content to check. If str, treated as input-only content.	required
engine_kwargs	dict[str, Any] | None	Arguments to pass to each engine. Defaults to None.	None

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	The result of the content safety check. Returns safe result only if all engines pass all applicable checks.

Raises:

Type	Description
BaseInvokerError	If an LM invoker used by an engine fails.
ValueError	If content is neither str nor GuardrailInput.
Exception	The original engine exception, if its type matches any entry in raise_on_exceptions. The exception propagates unmodified after logger.error fires.

check_input_stream(chunks, engine_kwargs=None) async

Check an input stream through configured engines.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterable[str]	Asynchronous source chunk iterator.	required
engine_kwargs	dict[str, Any] | None	Engine-specific kwargs forwarded to each active engine. Defaults to None.	None

Yields:

Name	Type	Description
GuardrailStreamChunk	AsyncIterator[GuardrailStreamChunk]	Safe source chunks in order, or one terminal unsafe marker when an engine blocks the content.

check_output_stream(chunks, engine_kwargs=None) async

Check an output stream through configured engines.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterable[str]	Asynchronous source chunk iterator.	required
engine_kwargs	dict[str, Any] | None	Engine-specific kwargs forwarded to each active engine. Defaults to None.	None

Yields:

Name	Type	Description
GuardrailStreamChunk	AsyncIterator[GuardrailStreamChunk]	Safe source chunks in order, or one terminal unsafe marker when an engine blocks the content.

GuardrailMode

Bases: StrEnum

Guardrail mode enumeration for guardrail configuration.

GuardrailResult

Bases: BaseModel

Result of a guardrail content safety check.

This model represents the standardized return value for all guardrail operations, providing type safety and validation for safety check results.

Attributes:

Name	Type	Description
is_safe	bool	Whether the content passed the safety check.
reason	str | None	Human-readable reason for rejection if content is not safe. Defaults to None.
filtered_content	str | None	Cleaned/filtered version of content if available. Defaults to None.
category	str | None	Violation category metadata. Defaults to None.
policy	str | None	Policy key that triggered the result. Defaults to None.
score	float | None	Score/probability associated with the result. Defaults to None.

from_dict(data) classmethod

Create from dictionary format for backward compatibility.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary with is_safe, reason, filtered_content, category, policy, and score keys.	required

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	Parsed result instance.

safe(filtered_content=None, score=None) classmethod

Create a safe result.

Parameters:

Name	Type	Description	Default
filtered_content	str | None	Filtered content. Defaults to None.	None
score	float | None	Highest policy score observed. Defaults to None.	None

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	Safe result.

to_dict()

Convert to dictionary format for backward compatibility.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Dictionary representation of the result including optional metadata.

unsafe(reason, filtered_content=None, category=None, policy=None, score=None) classmethod

Create an unsafe result.

Parameters:

Name	Type	Description	Default
reason	str	Reason for rejection.	required
filtered_content	str | None	Filtered content. Defaults to None.	None
category	str | None	Violation category. Defaults to None.	None
policy	str | None	Policy key that triggered the result. Defaults to None.	None
score	float | None	Score or probability for the violation. Defaults to None.	None

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	Unsafe result.

NemoGuardrailEngine(config)

Bases: GuardrailEngine

Default engine that keeps current NeMo Guardrails behavior.

This engine implements the GuardrailEngine protocol to provide NeMo Guardrails functionality for content safety checking.

Attributes:

Name	Type	Description
config	NemoGuardrailEngineConfig	Engine configuration specifying what content types to check.
rails	LLMRails | None	LLMRails instance.

Initializes a new instance of the NemoGuardrailEngine class.

The engine uses structured safety configuration via content_safety_config and topic_safety_config. Either or both can be provided; if only one is given, that check runs and the other is skipped.

LLM Invoker Override: If self.config.nemo_kwargs is provided and contains a model configuration with type: main, that model will be overridden by self.config.lm_invoker to ensure consistency across the SDK.

Example

Activate the new configuration by providing at least one safety config::

from gllm_guardrail.config.safety import ContentSafetyConfigBuilder, TopicSafetyConfigBuilder

# Both enabled (recommended)
config = NemoGuardrailEngineConfig(
    lm_invoker=my_invoker,
content_safety_config=ContentSafetyConfigBuilder(),
topic_safety_config=TopicSafetyConfigBuilder(),
)

# Only content safety enabled, topic safety skipped
config = NemoGuardrailEngineConfig(
    lm_invoker=my_invoker,
    content_safety_config=ContentSafetyConfigBuilder(),
)

engine = NemoGuardrailEngine(config=config)

Parameters:

Name	Type	Description	Default
config	NemoGuardrailEngineConfig	NemoGuardrailEngineConfig specifying both base and NeMo-specific settings.	required

check_input(content, **kwargs) async

Check input content for safety violations.

This method implements NeMo Guardrails checking for user input such as query content, context, or prompts before sending to LLM.

Parameters:

Name	Type	Description	Default
content	str	The input text to evaluate for safety.	required
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe.

Raises:

Type	Description
RuntimeError	If the guardrail engine is not initialized.

check_input_stream(chunks, **kwargs)

Check an input stream with one complete-text NeMo safety call.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterator[str]	Asynchronous iterator of source chunk strings.	required
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Type	Description
AsyncIterator[GuardrailStreamChunk]	AsyncIterator[GuardrailStreamChunk]: Original chunks when safe, or one terminal unsafe marker when NeMo judges the joined text unsafe.

check_output(content, **kwargs) async

Check output content for safety violations.

This method implements NeMo Guardrails checking for system output such as LLM responses, generated text, or static messages before showing to users.

Parameters:

Name	Type	Description	Default
content	str	The output text to evaluate for safety.	required
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Name	Type	Description
GuardrailResult	GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe.

Raises:

Type	Description
RuntimeError	If the guardrail engine is not initialized.

check_output_stream(chunks, **kwargs)

Check an output stream with one complete-text NeMo safety call.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterator[str]	Asynchronous iterator of source chunk strings.	required
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Type	Description
AsyncIterator[GuardrailStreamChunk]	AsyncIterator[GuardrailStreamChunk]: Original chunks when safe, or one terminal unsafe marker when NeMo judges the joined text unsafe.

PhraseMatcherEngine(config=None, banned_phrases=None, use_spacy=None, model_name='en_core_web_sm')

Bases: GuardrailEngine

Engine that uses SpaCy PhraseMatcher or Regex for banned phrase detection.

This engine implements the GuardrailEngine protocol to check content for banned phrases using either SpaCy for optimized matching or regex as a fallback.

Attributes:

Name	Type	Description
config		Engine configuration specifying what content types to check
banned_phrases	list[str]	Phrases that are explicitly banned.
use_spacy	bool	Whether to use SpaCy for phrase matching.
model_name	str	SpaCy model name to load.
banned_phrases_regex	Pattern | None	Compiled regex pattern for fallback.
phrase_matcher	PhraseMatcher | None	SpaCy phrase matcher.
nlp	Language | None	SpaCy language model.

Initialize the PhraseMatcherEngine.

Parameters:

Name	Type	Description	Default
config	BaseGuardrailEngineConfig | None	Engine configuration. Defaults to BaseGuardrailEngineConfig with INPUT_ONLY mode.	None
banned_phrases	list[str] | None	List of banned phrases. Defaults to DEFAULT_BANNED_PHRASES.	None
use_spacy	bool | None	Whether to use SpaCy for phrase matching. Defaults to SPACY_AVAILABLE.	None
model_name	str	SpaCy model name to load. Defaults to "en_core_web_sm".	'en_core_web_sm'

check_input(content, **kwargs) async

Check input content for banned phrases.

Parameters:

Name	Type	Description	Default
content	str	The input text to evaluate for safety	required
**kwargs	Any	Additional engine-specific parameters (optional)	{}

Returns:

Type	Description
GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe

Raises:

Type	Description
Exception	If the safety check fails due to engine errors

check_input_stream(chunks, max_stream_buffer_bytes=DEFAULT_MAX_STREAM_BUFFER_BYTES, max_stream_buffer_chunks=DEFAULT_MAX_STREAM_BUFFER_CHUNKS, **kwargs)

Check an input stream by buffering and evaluating the complete text.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterator[str]	Asynchronous iterator of source chunk strings.	required
max_stream_buffer_bytes	int	Maximum cumulative UTF-8 bytes to buffer. Defaults to DEFAULT_MAX_STREAM_BUFFER_BYTES.	DEFAULT_MAX_STREAM_BUFFER_BYTES
max_stream_buffer_chunks	int	Maximum source chunks to buffer. Defaults to DEFAULT_MAX_STREAM_BUFFER_CHUNKS.	DEFAULT_MAX_STREAM_BUFFER_CHUNKS
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Type	Description
AsyncIterator[GuardrailStreamChunk]	AsyncIterator[GuardrailStreamChunk]: Original chunks when safe, or one terminal unsafe marker when the joined text matches a banned phrase.

check_output(content, **kwargs) async

Check output content for banned phrases.

Parameters:

Name	Type	Description	Default
content	str	The output text to evaluate for safety	required
**kwargs	Any	Additional engine-specific parameters (optional)	{}

Returns:

Type	Description
GuardrailResult	GuardrailResult indicating if content is safe, with reason if unsafe

Raises:

Type	Description
Exception	If the safety check fails due to engine errors

check_output_stream(chunks, max_stream_buffer_bytes=DEFAULT_MAX_STREAM_BUFFER_BYTES, max_stream_buffer_chunks=DEFAULT_MAX_STREAM_BUFFER_CHUNKS, **kwargs)

Check an output stream by buffering and evaluating the complete text.

Parameters:

Name	Type	Description	Default
chunks	AsyncIterator[str]	Asynchronous iterator of source chunk strings.	required
max_stream_buffer_bytes	int	Maximum cumulative UTF-8 bytes to buffer. Defaults to DEFAULT_MAX_STREAM_BUFFER_BYTES.	DEFAULT_MAX_STREAM_BUFFER_BYTES
max_stream_buffer_chunks	int	Maximum source chunks to buffer. Defaults to DEFAULT_MAX_STREAM_BUFFER_CHUNKS.	DEFAULT_MAX_STREAM_BUFFER_CHUNKS
**kwargs	Any	Additional engine-specific parameters.	{}

Returns:

Type	Description
AsyncIterator[GuardrailStreamChunk]	AsyncIterator[GuardrailStreamChunk]: Original chunks when safe, or one terminal unsafe marker when the joined text matches a banned phrase.