Guardrail Manager

Manage content guardrails using pluggable guardrail engines.

GuardrailManager(engine, 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]	Required pluggable engines for guardrail operations. 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]	Required pluggable engine for guardrail operations. 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.	required
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.

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.