LangChain Integration

NeMo Guardrails integrates seamlessly with LangChain through the RunnableRails class, which implements the LangChain Runnable protocol. This allows you to add guardrails to any LangChain component.

Installation

Install with LangChain support:

pip install nemoguardrails langchain langchain-openai

RunnableRails

The RunnableRails class wraps a guardrails configuration and provides LangChain-compatible interfaces.

Basic Usage

from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Load guardrails configuration
config = RailsConfig.from_path("./config")

# Create a runnable with guardrails
guardrails = RunnableRails(config)

# Use it like any LangChain runnable
result = guardrails.invoke({"input": "Hello!"})
print(result["output"])

Constructor Parameters

config

RailsConfig

required

The rails configuration to use.

llm

BaseLanguageModel

default:"None"

Optional LLM to use with the rails. If not provided, uses the LLM from the config.

tools

List[Tool]

default:"None"

Optional list of LangChain tools to register with the rails.

passthrough

bool

default:"True"

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

runnable

Runnable

default:"None"

Optional runnable to wrap with the rails.

input_key

str

default:"input"

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

output_key

str

default:"output"

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

verbose

bool

default:"False"

Whether to print verbose logs.

Integration Patterns

Wrapping an LLM

Add guardrails around a language model:

from langchain_openai import ChatOpenAI
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Create LLM
llm = ChatOpenAI(model="gpt-3.5-turbo")

# Load guardrails config
config = RailsConfig.from_path("./config")

# Wrap LLM with guardrails
guarded_llm = RunnableRails(config, llm=llm)

# Use it
response = guarded_llm.invoke("Tell me about AI safety")
print(response)

Wrapping a Chain

Add guardrails to an entire chain:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Create a chain
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm | StrOutputParser()

# Load guardrails
config = RailsConfig.from_path("./config")

# Wrap the entire chain
guarded_chain = RunnableRails(config, runnable=chain)

# Use it
result = guarded_chain.invoke({"topic": "quantum computing"})
print(result)

With LangChain Tools

from langchain.agents import load_tools
from langchain.tools import Tool
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Define custom tools
def search_database(query: str) -> str:
    """Search the company database."""
    return f"Results for: {query}"

def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Weather in {location}: Sunny, 72°F"

tools = [
    Tool(
        name="search_database",
        func=search_database,
        description="Search the company database"
    ),
    Tool(
        name="get_weather",
        func=get_weather,
        description="Get current weather"
    )
]

# Create guardrails with tools
config = RailsConfig.from_path("./config")
guardrails = RunnableRails(config, tools=tools)

# Tools are now available in Colang flows
result = guardrails.invoke("What's the weather in Seattle?")

Chaining Multiple Guardrails

Create pipelines with multiple guardrail layers:

from langchain_openai import ChatOpenAI
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Load different guardrail configs
input_rails_config = RailsConfig.from_path("./input_rails")
output_rails_config = RailsConfig.from_path("./output_rails")

llm = ChatOpenAI()

# Chain: input rails -> LLM -> output rails
chain = (
    RunnableRails(input_rails_config) 
    | llm 
    | RunnableRails(output_rails_config)
)

result = chain.invoke("User input")

Async Support

RunnableRails fully supports async operations:

import asyncio
from langchain_openai import ChatOpenAI
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

async def main():
    config = RailsConfig.from_path("./config")
    llm = ChatOpenAI()
    guardrails = RunnableRails(config, llm=llm)
    
    # Async invoke
    result = await guardrails.ainvoke("Tell me about AI")
    print(result)
    
    # Async batch
    results = await guardrails.abatch([
        "What is machine learning?",
        "Explain neural networks",
        "What is deep learning?"
    ])
    for r in results:
        print(r)

asyncio.run(main())

Streaming

Stream responses token-by-token:

import asyncio
from langchain_openai import ChatOpenAI
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

async def stream_example():
    config = RailsConfig.from_path("./config")
    llm = ChatOpenAI(streaming=True)
    guardrails = RunnableRails(config, llm=llm)
    
    async for chunk in guardrails.astream("Tell me a story"):
        print(chunk.content, end="", flush=True)
    print()

asyncio.run(stream_example())

Streaming with output rails requires enabling streaming support:

config.yml

rails:
  output:
    streaming:
      enabled: true

Batch Processing

Process multiple inputs efficiently:

from langchain_openai import ChatOpenAI
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

config = RailsConfig.from_path("./config")
llm = ChatOpenAI()
guardrails = RunnableRails(config, llm=llm)

inputs = [
    "What is AI?",
    "Explain machine learning",
    "What is deep learning?"
]

# Synchronous batch
results = guardrails.batch(inputs)
for result in results:
    print(result.content)

# Async batch with concurrency control
import asyncio

async def batch_example():
    results = await guardrails.abatch(
        inputs,
        config={"max_concurrency": 3}
    )
    for result in results:
        print(result.content)

asyncio.run(batch_example())

Input/Output Formats

RunnableRails supports various input and output formats:

String Input/Output

config = RailsConfig.from_path("./config")
guardrails = RunnableRails(config)

# String in, string out
result = guardrails.invoke("Hello!")
print(result)  # String response

Dict Input/Output

config = RailsConfig.from_path("./config")
guardrails = RunnableRails(
    config,
    input_key="question",
    output_key="answer"
)

result = guardrails.invoke({"question": "What is AI?"})
print(result["answer"])

Message Objects

from langchain.schema import HumanMessage, AIMessage

config = RailsConfig.from_path("./config")
guardrails = RunnableRails(config)

# Message input
message = HumanMessage(content="Hello!")
result = guardrails.invoke(message)
print(result.content)  # AIMessage response

# List of messages
messages = [
    HumanMessage(content="Hi"),
    AIMessage(content="Hello! How can I help?"),
    HumanMessage(content="What is AI?")
]
result = guardrails.invoke(messages)

Advanced Examples

RAG Chain with Guardrails

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["AI is transforming industries", "Machine learning uses data"],
    embeddings
)
retriever = vectorstore.as_retriever()

# Create RAG chain
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template(
    "Answer based on context: {context}\n\nQuestion: {question}"
)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Add guardrails
config = RailsConfig.from_path("./config")
guarded_rag = RunnableRails(config, runnable=rag_chain)

result = guarded_rag.invoke("What is AI?")
print(result)

Agent with Guardrails

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import Tool
from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails

# Define tools
tools = [
    Tool(
        name="calculator",
        func=lambda x: eval(x),
        description="Calculate mathematical expressions"
    )
]

# Create agent
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant"),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# Add guardrails to agent
config = RailsConfig.from_path("./config")
guarded_agent = RunnableRails(config, runnable=agent_executor)

result = guarded_agent.invoke({"input": "What is 25 * 4?"})
print(result["output"])

Context Variables

Pass context variables through the chain:

from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails
from langchain_openai import ChatOpenAI

config = RailsConfig.from_path("./config")
llm = ChatOpenAI()
guardrails = RunnableRails(config, llm=llm)

# Include context in the input
result = guardrails.invoke({
    "input": "What's my account status?",
    "context": {
        "user_id": "12345",
        "account_type": "premium",
        "user_name": "Alice"
    }
})

print(result["output"])

Error Handling

from nemoguardrails import RailsConfig
from nemoguardrails.integrations.langchain import RunnableRails
from nemoguardrails.exceptions import InvalidRailsConfigurationError

try:
    config = RailsConfig.from_path("./config")
    guardrails = RunnableRails(config)
    
    result = guardrails.invoke("Hello!")
    print(result)
    
except InvalidRailsConfigurationError as e:
    print(f"Configuration error: {e}")
except ValueError as e:
    print(f"Input error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Best Practices

Use async for production

Async methods provide better performance and scalability:

result = await guardrails.ainvoke(input)

Enable streaming when needed

For better user experience with long responses:

async for chunk in guardrails.astream(input):
    # Process chunk

Configure concurrency for batch

Control resource usage in batch operations:

results = await guardrails.abatch(
    inputs,
    config={"max_concurrency": 5}
)

Use appropriate input/output keys

Match your chain’s expected format:

guardrails = RunnableRails(
    config,
    input_key="question",
    output_key="answer"
)

Troubleshooting

Input Format Errors

If you get input format errors, verify the expected format:

# For dict input, ensure the key matches
guardrails = RunnableRails(config, input_key="query")
result = guardrails.invoke({"query": "text"})  # Must use 'query'

Streaming Not Working

Enable streaming in both the LLM and config:

llm = ChatOpenAI(streaming=True)
guardrails = RunnableRails(config, llm=llm)

config.yml

rails:
  output:
    streaming:
      enabled: true

Tool Registration Issues

Ensure tools are registered before use:

# Register tools during initialization
guardrails = RunnableRails(config, tools=tools)

# Or register manually
for tool in tools:
    guardrails.rails.register_action(tool, tool.name)

Next Steps

Python API

Core Python API reference

Server Guide

Deploy as a REST API server

Configuration

Configure your guardrails

Examples

More integration examples

Get Started

Core Concepts

Configuration

Guardrails Library

Built-in Guardrails

Usage

Deployment

Evaluation

Installation

RunnableRails

Basic Usage

Constructor Parameters

Integration Patterns

Wrapping an LLM

Wrapping a Chain

With LangChain Tools

Chaining Multiple Guardrails

Async Support

Streaming

Batch Processing

Input/Output Formats

String Input/Output

Dict Input/Output

Message Objects

Advanced Examples

RAG Chain with Guardrails

Agent with Guardrails

Context Variables

Error Handling

Best Practices

Troubleshooting

Input Format Errors

Streaming Not Working

Tool Registration Issues

Next Steps

Python API

Server Guide

Configuration

Examples

Get Started

Core Concepts

Configuration

Guardrails Library

Built-in Guardrails

Usage

Deployment

Evaluation

​Installation

​RunnableRails

​Basic Usage

​Constructor Parameters

​Integration Patterns

​Wrapping an LLM

​Wrapping a Chain

​With LangChain Tools

​Chaining Multiple Guardrails

​Async Support

​Streaming

​Batch Processing

​Input/Output Formats

​String Input/Output

​Dict Input/Output

​Message Objects

​Advanced Examples

​RAG Chain with Guardrails

​Agent with Guardrails

​Context Variables

​Error Handling

​Best Practices

​Troubleshooting

​Input Format Errors

​Streaming Not Working

​Tool Registration Issues

​Next Steps

Python API

Server Guide

Configuration

Examples

Installation

RunnableRails

Basic Usage

Constructor Parameters

Integration Patterns

Wrapping an LLM

Wrapping a Chain

With LangChain Tools

Chaining Multiple Guardrails

Async Support

Streaming

Batch Processing

Input/Output Formats

String Input/Output

Dict Input/Output

Message Objects

Advanced Examples

RAG Chain with Guardrails

Agent with Guardrails

Context Variables

Error Handling

Best Practices

Troubleshooting

Input Format Errors

Streaming Not Working

Tool Registration Issues

Next Steps