nemoguardrails.integrations.langchain.runnable_rails | NVIDIA NeMo Guardrails Library Developer Guide

Module Contents

Classes

Name	Description
`RunnableRails`	A runnable that wraps a rails configuration.

Data

logger

API

class nemoguardrails.integrations.langchain.runnable_rails.RunnableRails(
    config: nemoguardrails.RailsConfig,
    llm: typing.Optional[langchain_core.language_models.BaseLanguageModel] = None,
    tools: typing.Optional[typing.List[langchain_core.tools.Tool]] = None,
    passthrough: bool = True,
    runnable: typing.Optional[langchain_core.runnables.Runnable] = None,
    input_key: str = 'input',
    output_key: str = 'output',
    verbose: bool = False,
    input_blocked_message: str = 'I cannot process this requ...,
    output_blocked_message: str = 'I cannot provide this resp...
)

Bases: Runnable[Input, Output]

A runnable that wraps a rails configuration.

This class implements the LangChain Runnable protocol to provide a way to add guardrails to LangChain components. It can wrap LLM models or entire chains and add input/output rails and dialog rails.

Parameters:

config

RailsConfig

The rails configuration to use.

llm

Optional[BaseLanguageModel]Defaults to None

Optional LLM to use with the rails.

tools

Optional[List[Tool]]Defaults to None

Optional list of tools to register with the rails.

passthrough

boolDefaults to True

Whether to pass through the original prompt or let rails modify it. Defaults to True.

runnable

Optional[Runnable]Defaults to None

Optional runnable to wrap with the rails.

input_key

strDefaults to 'input'

The key to use for the input when dealing with dict input.

output_key

strDefaults to 'output'

The key to use for the output when dealing with dict output.

verbose

boolDefaults to False

Whether to print verbose logs.

input_blocked_message

strDefaults to 'I cannot process this request.'

Message to return when input is blocked by rails.

output_blocked_message

strDefaults to 'I cannot provide this response.'

Message to return when output is blocked by rails.

OutputType

Any

The type of the output of this runnable as a type annotation.

config

Optional[RunnableConfig] = None

kwargs

Dict[str, Any] = {}

rails

= LLMRails(config=config, llm=llm, verbose=verbose)

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.__or__(
    other: typing.Union[langchain_core.language_models.BaseLanguageModel, langchain_core.runnables.Runnable[typing.Any, typing.Any]]
) -> typing.Union['RunnableRails', langchain_core.runnables.Runnable[typing.Any, typing.Any]]

Chain this runnable with another, returning a new runnable.

This method handles two different cases:

If other is a BaseLanguageModel, set it as the LLM for this RunnableRails
If other is a Runnable, either: a. Set it as the passthrough_runnable if this RunnableRails has no passthrough_runnable yet b. Otherwise, delegate to the standard Runnable.or to create a proper chain

This ensures associativity in complex chains.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._convert_messages_to_rails_format(
    messages
) -> typing.List[dict]

Convert LangChain messages to rails message format.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._create_passthrough_messages(
    _input
) -> typing.List[typing.Dict[str, typing.Any]]

Create messages for passthrough mode.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._extract_content_from_result(
    result: typing.Any
) -> str

Extract text content from result, handling both dict and direct formats.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._extract_output_content(
    output: langchain_core.runnables.utils.Output
) -> str

Extract content from output for rails checking.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._extract_text_from_input(
    _input
) -> str

Extract text content from various input types for passthrough mode.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._extract_user_input_from_dict(
    _input: dict
)

Extract user input from dictionary, checking configured key first.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_chat_prompt_output(
    result: typing.Any,
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> AIMessage

Format output for ChatPromptValue input.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_dict_output(
    input_dict: dict,
    result: typing.Any,
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> typing.Dict[str, typing.Any]

Format output for dictionary input.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_dict_output_for_base_message(
    result: typing.Any,
    output_key: str,
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> typing.Dict[str, typing.Any]

Format dict output when user input was a BaseMessage.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_dict_output_for_base_message_list(
    result: typing.Any,
    output_key: str,
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> typing.Dict[str, typing.Any]

Format dict output when user input was a list of BaseMessage objects.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_dict_output_for_dict_message_list(
    result: typing.Any,
    output_key: str
) -> typing.Dict[str, typing.Any]

Format dict output when user input was a list of dict messages.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_dict_output_for_string_input(
    result: typing.Any,
    output_key: str
) -> typing.Dict[str, typing.Any]

Format dict output when the user input was a string.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_message_output(
    result: typing.Any,
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> AIMessage

Format output for BaseMessage input types.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_output(
    input: typing.Any,
    result: typing.Any,
    context: typing.Dict[str, typing.Any],
    tool_calls: typing.Optional[list] = None,
    metadata: typing.Optional[dict] = None
) -> typing.Any

Format the output based on the input type and rails result.

Parameters:

input

Any

The original input.

result

Any

The result from the rails.

context

Dict[str, Any]

The context returned by the rails.

Returns: Any

The formatted output.

Raises:

ValueError: If the input type cannot be handled.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_passthrough_output(
    result: typing.Any,
    context: typing.Dict[str, typing.Any]
) -> typing.Any

Format output for passthrough mode.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_streaming_chunk(
    input: typing.Any,
    chunk
) -> typing.Any

Format a streaming chunk based on the input type.

Parameters:

input

Any

The original input

chunk

The current chunk (string or dict with text and metadata)

Returns: Any

The formatted streaming chunk (using AIMessageChunk for LangChain compatibility)

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._format_string_prompt_output(
    result: typing.Any
) -> str

Format output for StringPromptValue input.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._full_rails_ainvoke(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

async

Full rails mode async: existing LLMRails processing.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._full_rails_invoke(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

Full rails mode: existing LLMRails processing.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._get_bot_message(
    result: typing.Any,
    context: typing.Dict[str, typing.Any]
) -> str

Extract the bot message from context or result.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._init_passthrough_fn()

Initialize the passthrough function for the LLM rails instance.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._input_to_rails_messages(
    input: langchain_core.runnables.utils.Input
) -> typing.List[dict]

Convert various input formats to rails message format.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._transform_chat_prompt_value(
    _input: langchain_core.prompt_values.ChatPromptValue
) -> typing.List[typing.Dict[str, typing.Any]]

Transform ChatPromptValue to messages list.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._transform_dict_input(
    _input: dict
) -> typing.List[typing.Dict[str, typing.Any]]

Transform dictionary input to messages list.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._transform_dict_message_list(
    user_input: list
) -> typing.List[typing.Dict[str, typing.Any]]

Transform list from dictionary input to messages.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._transform_dict_user_input(
    user_input
) -> typing.List[typing.Dict[str, typing.Any]]

Transform user input value from dictionary.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails._transform_input_to_rails_format(
    _input
) -> typing.List[typing.Dict[str, typing.Any]]

Transform input to the format expected by the rails.

Parameters:

_input

The input to transform.

Returns: List[Dict[str, Any]]

A list of messages in the format expected by the rails.

Raises:

ValueError: If the input format cannot be handled.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.abatch(
    inputs: typing.List[langchain_core.runnables.utils.Input],
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> typing.List[langchain_core.runnables.utils.Output]

async

Batch inputs and process them asynchronously.

Concurrency is controlled via config[‘max_concurrency’] following LangChain best practices.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.ainvoke(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

async

Invoke this runnable asynchronously.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.astream(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> typing.AsyncIterator[langchain_core.runnables.utils.Output]

async

Stream the output of this runnable asynchronously.

Provides token-by-token streaming of the LLM response with guardrails applied. Uses LLMRails.stream_async() directly for efficient streaming.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.atransform(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

async

Transform the input asynchronously.

This is just an alias for ainvoke.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.batch(
    inputs: typing.List[langchain_core.runnables.utils.Input],
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> typing.List[langchain_core.runnables.utils.Output]

Batch inputs and process them synchronously.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.get_name(
    suffix: str = ''
) -> str

Get the name of this runnable.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.invoke(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

Invoke this runnable synchronously.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.stream(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> typing.Iterator[langchain_core.runnables.utils.Output]

Stream the output of this runnable synchronously.

Provides token-by-token streaming of the LLM response with guardrails applied. Handles async context properly by running astream in a separate event loop.

nemoguardrails.integrations.langchain.runnable_rails.RunnableRails.transform(
    input: langchain_core.runnables.utils.Input,
    config: typing.Optional[langchain_core.runnables.RunnableConfig] = None,
    kwargs: typing.Optional[typing.Any] = {}
) -> langchain_core.runnables.utils.Output

Transform the input.

This is just an alias for invoke.

nemoguardrails.integrations.langchain.runnable_rails.logger = logging.getLogger(__name__)