Simple Guardrail

This guide will walk you through adding a Guardrail component to your existing RAG pipeline that validates inputs and terminates execution when conditions are not met, ensuring your pipeline only processes valid requests.

Guardrail functionality provides input validation and safety checks, preventing errors and protecting your system from malicious or malformed inputs.

This tutorial extends the Your First RAG Pipeline tutorial. Ensure you have followed the instructions to set up your repository.

Prerequisites

This example specifically requires:

1. Completion of the Your First RAG Pipeline tutorial - this builds directly on top of it

2. Completion of all setup steps listed on the Prerequisites page

3. A working OpenAI API key configured in your environment variables

You should be familiar with these concepts and components:

1. Components in Your First RAG Pipeline - Required foundation

2. guardstep

---

View full project code on GitHub

Installation

Linux, macOS, or Windows WSL

# you can use a Conda environment
pip install --extra-index-url "https://oauth2accesstoken:$(gcloud auth print-access-token)@glsdk.gdplabs.id/gen-ai-internal/simple/" gllm-rag gllm-core gllm-generation gllm-inference gllm-pipeline gllm-retrieval gllm-misc gllm-datastore

Windows Powershell

Windows Command Prompt

How to Use this Guide

You can either:

1. Download or copy the complete guide file(s) to get everything ready instantly by heading to 📂 Complete Guide Files section in the end of this page. You can refer to the guide whenever you need explanation or want to clarify how each part works.

2. Follow along with each step to recreate the files yourself while learning about the components and how to integrate them.

Both options will work—choose based on whether you prefer speed or learning by doing!

Project Setup

1

Extend Your RAG Pipeline Project

Start with your completed RAG pipeline project from the Your First RAG Pipeline tutorial. We don't need to add any new file for this tutorial. Therefore, the structure should stay as is:

<project-name>/
├── data/
│   ├── <index>/...                     # preset data index folder
│   ├── chroma.sqlite3                  # preset database file
│   ├── imaginary_animals.csv           # sample data
├── modules/
│   ├── retriever.py
│   └── response_synthesizer.py
├── .env
├── indexer.py                    
└── pipeline.py    # 👈 Will be updated with guardrail functionality

---

1) Build the Guardrail Pipeline

1

Create modules/validators.py with a simple length validation function:

from typing import Any

def validate_message_length(inputs: dict[str, Any]) -> bool:
    user_query = inputs["user_query"]
    max_query_length = inputs["max_query_length"]
    min_query_length = inputs["min_query_length"]
    return len(user_query) <= max_query_length and len(user_query) >= min_query_length

The validator function takes an inputs dictionary and returns True for valid queries, False for invalid ones.

2

Define the extended state

Create a custom state that includes validation parameters:

from gllm_pipeline.pipeline import RAGState

class GuardrailState(RAGState):
    max_query_length: int
    min_query_length: int

3

Create the guardrail step

This is the core guard logic that validates inputs and controls execution:

from gllm_core.constants import EventLevel
from gllm_pipeline.steps import guard

guardrail_step = guard(
    validate_message_length,
    success_branch=retrieve_step,
    failure_branch=log( # for extra logging step
        message="User query length is not valid: '{user_query}'",
        emit_kwargs={"event_level": EventLevel.INFO},
    ),
    input_map={
        "user_query": "user_query",
        "max_query_length": "max_query_length",
        "min_query_length": "min_query_length",
    },
)

How it works: If validation passes, executes retriever step; if it fails, logs error and terminates.

4

Compose the final pipeline

Chain all steps to create the complete guardrail pipeline:

e2e_pipeline = guardrail_step | synthesize_step
e2e_pipeline.state_type = GuardrailState

This creates a pipeline that validates inputs before processing and automatically terminates on invalid requests.

🧠 The guard step acts as a gatekeeper, ensuring only valid requests reach your expensive operations.

2) Run the Pipeline

When running the pipeline, you may encounter an error like this:

[2025-08-26T14:36:10+0700.550 chromadb.telemetry.product.posthog ERROR] Failed to send telemetry event CollectionQueryEvent: capture() takes 1 positional argument but 3 were given

Don't worry about this, since we do not use this Chroma feature. Your Pipeline should still work.

1

Configure the pipeline state for testing

Set up test cases to see guardrail validation in action:

from gllm_core.event import EventEmitter

async def main():
    state = {
        "user_query": "Give me nocturnal creatures from the dataset", # Replace with your actual query
        "max_query_length": 100,
        "min_query_length": 1,
    }

    # Test with invalid query length
    invalid_state = {
        "user_query": "this is a very long message that should be rejected by the guardrail, with length over 100 characters, and once again, it should be rejected",
        "max_query_length": 100,
        "min_query_length": 1,
    }

    event_emitter = EventEmitter.with_console_handler() # for extra logging step
    config = {
        "top_k": 5,
        "event_emitter": event_emitter, # for extra logging step
    }
    result = await e2e_pipeline.invoke(
        # state,
        invalid_state, # to test guardrail
        config,
    )
    print(f"Pipeline result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

When you run the pipeline with valid input, you should see normal pipeline execution with retrieval and response generation. Otherwise, you should see validation failure, error logging, and pipeline termination without expensive operations.

Troubleshooting

1. Validation not working:

   1. Ensure validator function returns boolean (True/False)

   2. Check input_state_map keys match your state variables

   3. Verify validator function signature accepts inputs: dict[str, Any]

   4. Use debug mode to see if guard step is executing

2. Pipeline not terminating on invalid input:

   1. Confirm guard step is properly configured with success/failure branches

   2. Check that the validator function logic is correct

   3. Verify all required state variables are present

   4. Use debug mode to inspect guard step execution

3. State mapping errors:

   1. Ensure all required state variables are initialized in your test

   2. Check that input_state_map keys exist in your pipeline state

   3. Verify validator function accesses inputs through the dictionary

   4. Match state field names exactly in the mapping

---

Congratulations! You've successfully enhanced your RAG pipeline with guardrail functionality. Your pipeline now validates inputs before processing and automatically blocks invalid requests, protecting your system resources and improving reliability.