5. Deep Guardrails

Overview

In the previous capsules you built three individual pieces: Input Sanitizer (capsule 02), Output Validator (capsule 03), and Content Filter (capsule 04). Each one solves a specific problem — input normalization, structure validation, content filtering. But in production, you need something more: an orchestration layer that coordinates these pieces, defines the execution order, handles failures between layers, and lets you add/remove validations without rewriting the pipeline.

That orchestration layer is the guardrails. The concept comes from the metal guardrails on a mountain road: they don't control the car's direction, but they keep it from falling off the cliff. In AI, guardrails don't control what the LLM generates, but they prevent dangerous outputs from reaching the user.

Production Best Practices (#13) introduced guardrails as a concept. This module goes deeper in three directions: first, existing guardrail frameworks (guardrails-ai, NeMo Guardrails) that speed up implementation; second, custom guardrails you can build for your business's specific needs; third, the performance impact and the strategies to mitigate it.

Guardrails vs Validation vs Filtering: clarification

Before going deeper, let's clarify the difference between the three concepts we've handled:

from dataclasses import dataclass


@dataclass
class ConceptComparison:
    concept: str
    question_it_answers: str
    operates_on: str
    example: str
    capsule: str


comparison = [
    ConceptComparison(
        concept="Validation (Pydantic)",
        question_it_answers="Does the output have the correct structure?",
        operates_on="Format: types, fields, ranges",
        example='{"price": "abc"} → fails: price must be a float',
        capsule="Capsule 03",
    ),
    ConceptComparison(
        concept="Filtering (Content)",
        question_it_answers="Is the content safe and appropriate?",
        operates_on="Semantics: toxicity, PII, off-topic",
        example='"You are an idiot" → blocked: toxic content',
        capsule="Capsule 04",
    ),
    ConceptComparison(
        concept="Guardrails",
        question_it_answers="Does the output comply with business rules?",
        operates_on="Policies: combination of everything + business logic",
        example="Valid + safe output but violates the no-competitor-recommendation policy",
        capsule="Capsule 05 (this one)",
    ),
]

for c in comparison:
    print(f"{c.concept}")
    print(f"  Question: {c.question_it_answers}")
    print(f"  Operates on: {c.operates_on}")
    print(f"  Example: {c.example}")
    print(f"  Capsule: {c.capsule}")
    print()

# Expected output:
# Validation (Pydantic)
#   Question: Does the output have the correct structure?
#   Operates on: Format: types, fields, ranges
#   ...
# Filtering (Content)
#   Question: Is the content safe and appropriate?
#   ...
# Guardrails
#   Question: Does the output comply with business rules?
#   ...

Guardrails are the broadest layer: they can include validation AND filtering as sub-components, but they also add business logic, orchestration, and decisions that aren't purely technical.

Guardrails AI: a practical framework

Guardrails AI is an open-source framework that simplifies the creation and orchestration of validations for LLM outputs.

Core concept: Guard

A Guard is a wrapper around an LLM call that applies validations before and after:

# Conceptual example — the real guardrails-ai API may vary
# pip install guardrails-ai

from pydantic import BaseModel, Field


class ProductRecommendation(BaseModel):
    product_name: str = Field(description="Name of the recommended product")
    reason: str = Field(description="Reason for the recommendation", min_length=10)
    price_range: str = Field(description="Price range")
    confidence: float = Field(ge=0.0, le=1.0)


# Conceptually, guardrails-ai works like this:
# 1. Define the schema (Pydantic model)
# 2. Define additional validators
# 3. The Guard wraps the LLM call
# 4. If the output doesn't comply, it retries automatically

def demonstrate_guard_concept():
    """Demonstrates the Guard concept without a direct dependency."""

    class Guard:
        def __init__(self, schema, validators=None, max_retries=3):
            self.schema = schema
            self.validators = validators or []
            self.max_retries = max_retries
            self.history = []

        def validate(self, raw_output: str) -> dict:
            import json
            try:
                data = json.loads(raw_output)
                validated = self.schema(**data)

                for validator_fn in self.validators:
                    result = validator_fn(validated)
                    if not result["passed"]:
                        return {
                            "valid": False,
                            "error": result["reason"],
                            "data": None,
                        }

                return {
                    "valid": True,
                    "error": None,
                    "data": validated.model_dump(),
                }
            except Exception as e:
                return {"valid": False, "error": str(e), "data": None}

    def no_competitor_names(output):
        competitors = ["samsung", "google pixel", "huawei"]
        text = f"{output.product_name} {output.reason}".lower()
        for comp in competitors:
            if comp in text:
                return {"passed": False, "reason": f"Mentions competitor: {comp}"}
        return {"passed": True, "reason": ""}

    def reasonable_confidence(output):
        if output.confidence > 0.95:
            return {"passed": False, "reason": "Confidence suspiciously high"}
        return {"passed": True, "reason": ""}

    guard = Guard(
        schema=ProductRecommendation,
        validators=[no_competitor_names, reasonable_confidence],
    )

    test_outputs = [
        '{"product_name": "iPhone 15", "reason": "Excellent camera and performance for the price", "price_range": "$799-$999", "confidence": 0.85}',
        '{"product_name": "Samsung Galaxy", "reason": "Better than our product in camera and price", "price_range": "$699-$899", "confidence": 0.9}',
        '{"product_name": "iPhone 15", "reason": "The best on the market without a doubt", "price_range": "$799-$999", "confidence": 0.99}',
    ]

    for output in test_outputs:
        result = guard.validate(output)
        status = "PASS" if result["valid"] else "FAIL"
        print(f"[{status}] {output[:60]}...")
        if not result["valid"]:
            print(f"  Error: {result['error']}")
        print()

demonstrate_guard_concept()

# Expected output:
# [PASS] {"product_name": "iPhone 15", "reason": "Excellent camera an...
#
# [FAIL] {"product_name": "Samsung Galaxy", "reason": "Better than ou...
#   Error: Mentions competitor: samsung
#
# [FAIL] {"product_name": "iPhone 15", "reason": "The best on the mar...
#   Error: Confidence suspiciously high

Hub validators

Guardrails AI has a hub of pre-built validators that cover common cases:

GUARDRAILS_HUB_VALIDATORS = {
    "toxic_language": {
        "description": "Detects toxic, offensive, or hate speech language",
        "use_case": "Chatbots, content generation",
    },
    "detect_pii": {
        "description": "Detects PII in text (emails, SSN, phones)",
        "use_case": "Any system that handles user data",
    },
    "valid_url": {
        "description": "Verifies that generated URLs are valid and safe",
        "use_case": "Systems that generate links",
    },
    "provenance_llm": {
        "description": "Verifies that the response is based on provided sources",
        "use_case": "RAG systems to detect hallucinations",
    },
    "competitor_check": {
        "description": "Detects mentions of competitors",
        "use_case": "Sales/support chatbots",
    },
    "reading_level": {
        "description": "Verifies that the text is at the appropriate reading level",
        "use_case": "Educational content, public communications",
    },
    "nsfw_text": {
        "description": "Detects NSFW/adult content",
        "use_case": "Platforms with minor users",
    },
    "sql_column_presence": {
        "description": "Verifies that generated SQL only uses valid columns",
        "use_case": "NL2SQL, text-to-SQL systems",
    },
}

print("Guardrails Hub — Available validators:")
for name, info in GUARDRAILS_HUB_VALIDATORS.items():
    print(f"  {name}: {info['description']}")

# Expected output:
# Guardrails Hub — Available validators:
#   toxic_language: Detects toxic, offensive, or hate speech language
#   detect_pii: Detects PII in text (emails, SSN, phones)
#   ... (etc)

NeMo Guardrails: concepts

NeMo Guardrails from NVIDIA is a framework for adding guardrails to LLM applications. Its approach is different from guardrails-ai: it uses a declarative configuration language (Colang) to define conversation rules.

Key concepts

NEMO_CONCEPTS = {
    "Rails": {
        "description": "Rules that control the system's behavior",
        "types": {
            "Input rails": "Process the user's input before the LLM",
            "Output rails": "Process the LLM's output before the user",
            "Dialog rails": "Control the flow of the conversation",
            "Retrieval rails": "Filter documents retrieved in RAG",
        },
    },
    "Colang": {
        "description": "Declarative language for defining conversation rules",
        "example": """
define user ask about competitors
    "What do you think of Samsung?"
    "Is Amazon better?"
    "Compare with Google"

define bot refuse competitor comparison
    "I focus on our products. Can I help you with something specific?"

define flow
    user ask about competitors
    bot refuse competitor comparison
        """,
    },
    "Actions": {
        "description": "Python functions that implement custom logic",
        "example": "check_toxicity(), verify_facts(), detect_pii()",
    },
}

for concept, info in NEMO_CONCEPTS.items():
    print(f"\n{concept}: {info['description']}")
    if "types" in info:
        for t, desc in info["types"].items():
            print(f"  - {t}: {desc}")
    if "example" in info:
        print(f"  Example:\n{info['example'][:200]}")

# Expected output:
# Rails: Rules that control the system's behavior
#   - Input rails: Process the user's input before the LLM
#   - Output rails: Process the LLM's output before the user
#   - Dialog rails: Control the flow of the conversation
#   - Retrieval rails: Filter documents retrieved in RAG
# ...

When to use NeMo vs guardrails-ai vs custom

Building Custom Guardrails

For most cases, custom guardrails give you the control and performance you need. Here we build a modular guardrail system:

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional, Any
from enum import Enum
import time


class GuardrailAction(Enum):
    PASS = "pass"
    WARN = "warn"
    BLOCK = "block"
    MODIFY = "modify"


@dataclass
class GuardrailResult:
    name: str
    action: GuardrailAction
    message: str = ""
    modified_text: Optional[str] = None
    metadata: dict = field(default_factory=dict)
    execution_time_ms: float = 0.0


class Guardrail(ABC):
    """Base class for all guardrails."""

    def __init__(self, name: str, enabled: bool = True):
        self.name = name
        self.enabled = enabled

    @abstractmethod
    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        pass

    def execute(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        if not self.enabled:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.PASS,
                message="Guardrail disabled",
            )
        start = time.perf_counter()
        result = self.check(text, context)
        result.execution_time_ms = (time.perf_counter() - start) * 1000
        return result


class LengthGuardrail(Guardrail):
    def __init__(self, max_length: int = 2000, name: str = "length_check"):
        super().__init__(name)
        self.max_length = max_length

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        if len(text) > self.max_length:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.MODIFY,
                message=f"Output truncated from {len(text)} to {self.max_length}",
                modified_text=text[:self.max_length] + "...",
            )
        return GuardrailResult(
            name=self.name,
            action=GuardrailAction.PASS,
        )


class LanguageConsistencyGuardrail(Guardrail):
    """Verifies that the output is in the same language as the input."""

    def __init__(self, name: str = "language_consistency"):
        super().__init__(name)

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        import re
        if not context or "input_language" not in context:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.PASS,
                message="No input language in context",
            )

        input_lang = context["input_language"]
        es_words = len(re.findall(r"\b(el|la|los|las|es|son|está|para|con|que|por|un|una)\b", text.lower()))
        en_words = len(re.findall(r"\b(the|is|are|was|for|with|that|this|from|and|or)\b", text.lower()))

        total = es_words + en_words
        if total == 0:
            return GuardrailResult(name=self.name, action=GuardrailAction.PASS)

        detected = "es" if es_words > en_words else "en"

        if detected != input_lang:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.WARN,
                message=f"Output in '{detected}' but input was '{input_lang}'",
                metadata={"detected": detected, "expected": input_lang},
            )

        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


class TopicBoundaryGuardrail(Guardrail):
    """Verifies that the output stays within the allowed domain."""

    def __init__(
        self,
        allowed_topics: list[str],
        blocked_topics: list[str],
        name: str = "topic_boundary",
    ):
        super().__init__(name)
        self.allowed_topics = [t.lower() for t in allowed_topics]
        self.blocked_topics = [t.lower() for t in blocked_topics]

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        import re
        text_lower = text.lower()

        for topic in self.blocked_topics:
            if re.search(rf"\b{re.escape(topic)}\b", text_lower):
                return GuardrailResult(
                    name=self.name,
                    action=GuardrailAction.BLOCK,
                    message=f"Output discusses blocked topic: {topic}",
                )

        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


class ConfidenceCalibrationGuardrail(Guardrail):
    """Adds disclaimers when the model generates low-confidence content."""

    def __init__(self, disclaimer: str = "", name: str = "confidence_calibration"):
        super().__init__(name)
        self.disclaimer = disclaimer or (
            "\n\n⚠️ This response may not be completely accurate. "
            "Verify the information with official sources."
        )

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        confidence = (context or {}).get("confidence", 1.0)

        if confidence < 0.5:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.MODIFY,
                message=f"Low confidence ({confidence}), adding disclaimer",
                modified_text=text + self.disclaimer,
            )

        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)

Chaining Guardrails: GuardrailChain

The key piece: chaining guardrails into an ordered pipeline with failure handling:

from dataclasses import dataclass, field
from typing import Optional


@dataclass
class ChainResult:
    passed: bool
    final_text: Optional[str]
    results: list[GuardrailResult] = field(default_factory=list)
    blocked_by: Optional[str] = None
    total_time_ms: float = 0.0

    @property
    def warnings(self) -> list[str]:
        return [
            r.message for r in self.results
            if r.action == GuardrailAction.WARN
        ]


class GuardrailChain:
    def __init__(
        self,
        guardrails: list[Guardrail],
        fail_fast: bool = True,
    ):
        self.guardrails = guardrails
        self.fail_fast = fail_fast

    def run(self, text: str, context: Optional[dict] = None) -> ChainResult:
        results = []
        current_text = text
        total_start = time.perf_counter()

        for guardrail in self.guardrails:
            result = guardrail.execute(current_text, context)
            results.append(result)

            if result.action == GuardrailAction.BLOCK:
                return ChainResult(
                    passed=False,
                    final_text=None,
                    results=results,
                    blocked_by=guardrail.name,
                    total_time_ms=(time.perf_counter() - total_start) * 1000,
                )

            if result.action == GuardrailAction.MODIFY and result.modified_text:
                current_text = result.modified_text

        return ChainResult(
            passed=True,
            final_text=current_text,
            results=results,
            total_time_ms=(time.perf_counter() - total_start) * 1000,
        )


# --- Demonstration ---
# The topic guardrail uses Spanish blocked topics and the language guardrail is
# paired with an "es" input context, so the demo inputs stay in Spanish.

chain = GuardrailChain(
    guardrails=[
        LengthGuardrail(max_length=200),
        TopicBoundaryGuardrail(
            allowed_topics=["electronics", "software"],
            blocked_topics=["política", "religión"],
        ),
        LanguageConsistencyGuardrail(),
        ConfidenceCalibrationGuardrail(),
    ],
    fail_fast=True,
)

test_cases = [
    {
        "text": "El iPhone 15 tiene una cámara de 48MP y cuesta $799.",
        "context": {"input_language": "es", "confidence": 0.9},
    },
    {
        "text": "La política del gobierno sobre tecnología es...",
        "context": {"input_language": "es", "confidence": 0.8},
    },
    {
        "text": "El producto está disponible.",
        "context": {"input_language": "es", "confidence": 0.3},
    },
    {
        "text": "x" * 500,
        "context": {"input_language": "es", "confidence": 0.9},
    },
]

for tc in test_cases:
    result = chain.run(tc["text"], tc["context"])
    status = "PASS" if result.passed else f"BLOCKED by {result.blocked_by}"
    print(f"[{status}] Input: {tc['text'][:50]}...")
    print(f"  Time: {result.total_time_ms:.1f}ms")
    if result.warnings:
        print(f"  Warnings: {result.warnings}")
    if result.final_text and result.final_text != tc["text"]:
        print(f"  Modified: {result.final_text[:80]}...")
    print()

# Expected output (times vary by hardware; the disclaimer text appended by the
# confidence guardrail starts with a blank line):
# [PASS] Input: El iPhone 15 tiene una cámara de 48MP y cuesta $79...
#   Time: 0.1ms
#
# [BLOCKED by topic_boundary] Input: La política del gobierno sobre tecnología es......
#   Time: 0.0ms
#
# [PASS] Input: El producto está disponible....
#   Time: 0.0ms
#   Modified: El producto está disponible.
#
# ⚠️ This response may not be completely accurate. V...
#
# [PASS] Input: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...
#   Time: 0.0ms
#   Modified: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...

Performance Impact

Guardrails add latency. Measuring and optimizing is essential:

import time
from dataclasses import dataclass


@dataclass
class PerformanceMetrics:
    guardrail_name: str
    avg_ms: float
    p95_ms: float
    p99_ms: float
    calls: int


def benchmark_guardrail(guardrail: Guardrail, text: str, iterations: int = 100) -> PerformanceMetrics:
    times = []
    for _ in range(iterations):
        start = time.perf_counter()
        guardrail.execute(text)
        elapsed = (time.perf_counter() - start) * 1000
        times.append(elapsed)

    times.sort()
    return PerformanceMetrics(
        guardrail_name=guardrail.name,
        avg_ms=sum(times) / len(times),
        p95_ms=times[int(len(times) * 0.95)],
        p99_ms=times[int(len(times) * 0.99)],
        calls=iterations,
    )


guardrails_to_benchmark = [
    LengthGuardrail(max_length=2000),
    TopicBoundaryGuardrail(
        allowed_topics=["tech"],
        blocked_topics=["política", "religión", "drogas"],
    ),
    LanguageConsistencyGuardrail(),
]

test_text = "El iPhone 15 tiene características impresionantes. " * 20

print("Performance Benchmarks:")
print(f"{'Guardrail':<30} {'Avg (ms)':<12} {'P95 (ms)':<12} {'P99 (ms)':<12}")
print("-" * 66)

for gr in guardrails_to_benchmark:
    metrics = benchmark_guardrail(gr, test_text)
    print(f"{metrics.guardrail_name:<30} {metrics.avg_ms:<12.3f} {metrics.p95_ms:<12.3f} {metrics.p99_ms:<12.3f}")

# Expected output (varies by hardware):
# Performance Benchmarks:
# Guardrail                      Avg (ms)     P95 (ms)     P99 (ms)
# ------------------------------------------------------------------
# length_check                   0.005        0.010        0.015
# topic_boundary                 0.020        0.035        0.050
# language_consistency           0.015        0.025        0.040

Optimization strategies

OPTIMIZATION_STRATEGIES = {
    "1. Precompile regex": {
        "impact": "10-50x faster",
        "how": "Compile patterns in __init__, not in each check()",
    },
    "2. Short-circuit": {
        "impact": "Variable",
        "how": "Stop at the first BLOCK (fail_fast=True)",
    },
    "3. Order by cost": {
        "impact": "20-80% faster average",
        "how": "Cheap guardrails first (regex), expensive ones later (API)",
    },
    "4. Cache results": {
        "impact": "90%+ for repeated inputs",
        "how": "Hash of the input → cached result with TTL",
    },
    "5. Async execution": {
        "impact": "30-60% faster for independent guardrails",
        "how": "Run guardrails that don't depend on each other in parallel",
    },
}

for strategy, info in OPTIMIZATION_STRATEGIES.items():
    print(f"{strategy}: {info['impact']}")
    print(f"  → {info['how']}")

Guardrail Configuration: per-context profiles

from dataclasses import dataclass


@dataclass
class GuardrailProfile:
    name: str
    guardrails: list[Guardrail]
    description: str


def create_profile(profile_name: str) -> GuardrailProfile:
    profiles = {
        "strict": GuardrailProfile(
            name="strict",
            description="For public chatbots: maximum restrictions",
            guardrails=[
                LengthGuardrail(max_length=1000),
                TopicBoundaryGuardrail(
                    allowed_topics=["products", "support"],
                    blocked_topics=["política", "religión", "competidores", "precios internos"],
                ),
                LanguageConsistencyGuardrail(),
                ConfidenceCalibrationGuardrail(),
            ],
        ),
        "moderate": GuardrailProfile(
            name="moderate",
            description="For internal tools: moderate restrictions",
            guardrails=[
                LengthGuardrail(max_length=5000),
                TopicBoundaryGuardrail(
                    allowed_topics=[],
                    blocked_topics=["contenido adulto"],
                ),
            ],
        ),
        "minimal": GuardrailProfile(
            name="minimal",
            description="For batch pipelines: only critical checks",
            guardrails=[
                LengthGuardrail(max_length=10000),
            ],
        ),
    }
    return profiles.get(profile_name, profiles["moderate"])


for profile_name in ["strict", "moderate", "minimal"]:
    profile = create_profile(profile_name)
    print(f"{profile.name}: {profile.description}")
    print(f"  Guardrails: {[g.name for g in profile.guardrails]}")

# Expected output:
# strict: For public chatbots: maximum restrictions
#   Guardrails: ['length_check', 'topic_boundary', 'language_consistency', 'confidence_calibration']
# moderate: For internal tools: moderate restrictions
#   Guardrails: ['length_check', 'topic_boundary']
# minimal: For batch pipelines: only critical checks
#   Guardrails: ['length_check']

When to use each approach

DECISION_MATRIX = """
┌─────────────────────────────────────────────────────────────────────────┐
│                    Which approach should you use?                       │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  Need to validate JSON structure/types?                                 │
│  └── YES → Pydantic Validation (Cap 03)                                │
│                                                                         │
│  Need to detect toxic/PII/off-topic content?                            │
│  └── YES → Content Filtering (Cap 04)                                  │
│                                                                         │
│  Need to enforce complex business rules?                                │
│  └── YES → Custom Guardrails (Cap 05)                                  │
│                                                                         │
│  Need to control conversational flows?                                  │
│  └── YES → NeMo Guardrails (Colang)                                    │
│                                                                         │
│  Need pre-built validators + hub?                                       │
│  └── YES → guardrails-ai                                               │
│                                                                         │
│  Need maximum control and minimum latency?                              │
│  └── YES → Custom guardrails with GuardrailChain                       │
│                                                                         │
│  In practice, most systems use a combination:                          │
│  Pydantic + Content Filter + 2-3 Custom Guardrails                     │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
"""
print(DECISION_MATRIX)

Troubleshooting

Problem 1: "The guardrails block too many valid responses"

The false positive rate is high — legitimate responses get blocked by rules that are too broad.

Solution: Implement a "shadow" mode where the guardrails log but don't block. Collect data for 1-2 weeks. Analyze the false positives. Adjust thresholds and patterns before enabling blocking.

class ShadowModeGuardrail(Guardrail):
    def __init__(self, wrapped: Guardrail, shadow: bool = True):
        super().__init__(f"shadow_{wrapped.name}")
        self.wrapped = wrapped
        self.shadow = shadow

    def check(self, text, context=None):
        result = self.wrapped.check(text, context)
        if self.shadow and result.action == GuardrailAction.BLOCK:
            result.action = GuardrailAction.WARN
            result.message = f"[SHADOW] Would block: {result.message}"
        return result

Problem 2: "Chained guardrails are hard to debug"

When a chain of 5 guardrails modifies the text, it's hard to know which one caused a problem.

Solution: Each GuardrailResult includes the guardrail's name and message. Log the full chain including the text before/after each guardrail. Add a trace_id to the context to correlate logs.

Problem 3: "guardrails-ai adds heavy dependencies"

The framework brings dependencies that can conflict with yours.

Solution: If you only need 2-3 guardrails, build custom ones with this capsule's Guardrail base. Only adopt frameworks when you need >10 different guardrails or the hub of pre-built validators.

Problem 4: "Custom guardrails don't scale with the team"

Each developer adds guardrails differently, without a standard.

Solution: Define the Guardrail interface (like this capsule's) as the team's contract. Each new guardrail implements check() and is registered in the GuardrailChain. Document each guardrail with a name, purpose, and configurable threshold.

Exercises

Exercise 1: Factual consistency guardrail

Create a guardrail that detects when the output contradicts information provided in the context.

class FactConsistencyGuardrail(Guardrail):
    def __init__(self, name: str = "fact_consistency"):
        super().__init__(name)

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        facts = (context or {}).get("known_facts", {})
        if not facts:
            return GuardrailResult(name=self.name, action=GuardrailAction.PASS)

        text_lower = text.lower()
        contradictions = []
        for fact_key, fact_value in facts.items():
            fact_str = str(fact_value).lower()
            if fact_key.lower() in text_lower and fact_str not in text_lower:
                contradictions.append(f"Mentions '{fact_key}' but doesn't match known value '{fact_value}'")

        if contradictions:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.WARN,
                message=f"Possible contradictions: {contradictions}",
            )
        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


gr = FactConsistencyGuardrail()
result = gr.execute(
    "The iPhone 15 costs $999.",
    context={"known_facts": {"iPhone 15": "$799"}},
)
print(f"Action: {result.action.value}, Message: {result.message}")

# Expected output:
# Action: warn, Message: Possible contradictions: ["Mentions 'iPhone 15' but doesn't match known value '$799'"]

Exercise 2: Guardrail with per-user rate limiting

Create a guardrail that limits how many outputs per minute a specific user can generate.

import time
from collections import defaultdict

class RateLimitGuardrail(Guardrail):
    def __init__(self, max_per_minute: int = 10, name: str = "rate_limit"):
        super().__init__(name)
        self.max_per_minute = max_per_minute
        self.user_timestamps: dict[str, list[float]] = defaultdict(list)

    def check(self, text: str, context: Optional[dict] = None) -> GuardrailResult:
        user_id = (context or {}).get("user_id", "anonymous")
        now = time.time()
        window = now - 60

        self.user_timestamps[user_id] = [
            t for t in self.user_timestamps[user_id] if t > window
        ]
        self.user_timestamps[user_id].append(now)

        count = len(self.user_timestamps[user_id])
        if count > self.max_per_minute:
            return GuardrailResult(
                name=self.name,
                action=GuardrailAction.BLOCK,
                message=f"Rate limit exceeded: {count}/{self.max_per_minute} per minute",
                metadata={"user_id": user_id, "count": count},
            )

        return GuardrailResult(
            name=self.name,
            action=GuardrailAction.PASS,
            metadata={"user_id": user_id, "count": count},
        )


rl = RateLimitGuardrail(max_per_minute=3)
for i in range(5):
    result = rl.execute("test", context={"user_id": "user-123"})
    print(f"Request {i+1}: {result.action.value} (count: {result.metadata.get('count', 0)})")

# Expected output:
# Request 1: pass (count: 1)
# Request 2: pass (count: 2)
# Request 3: pass (count: 3)
# Request 4: block (count: 4)
# Request 5: block (count: 5)

Exercise 3: Composable guardrail with AND/OR logic

Implement guardrails that can be combined with AND logic (all must pass) and OR logic (at least one must pass).

class AndGuardrail(Guardrail):
    def __init__(self, guards: list[Guardrail], name: str = "and_group"):
        super().__init__(name)
        self.guards = guards

    def check(self, text: str, context=None) -> GuardrailResult:
        for guard in self.guards:
            result = guard.execute(text, context)
            if result.action == GuardrailAction.BLOCK:
                return GuardrailResult(
                    name=self.name,
                    action=GuardrailAction.BLOCK,
                    message=f"AND failed at {guard.name}: {result.message}",
                )
        return GuardrailResult(name=self.name, action=GuardrailAction.PASS)


class OrGuardrail(Guardrail):
    def __init__(self, guards: list[Guardrail], name: str = "or_group"):
        super().__init__(name)
        self.guards = guards

    def check(self, text: str, context=None) -> GuardrailResult:
        for guard in self.guards:
            result = guard.execute(text, context)
            if result.action == GuardrailAction.PASS:
                return GuardrailResult(
                    name=self.name,
                    action=GuardrailAction.PASS,
                    message=f"OR passed at {guard.name}",
                )
        return GuardrailResult(
            name=self.name,
            action=GuardrailAction.BLOCK,
            message="OR: no guardrail passed",
        )


combined = AndGuardrail([
    LengthGuardrail(max_length=1000),
    TopicBoundaryGuardrail(allowed_topics=[], blocked_topics=["política"]),
])

print(combined.execute("Short safe text").action.value)
print(combined.execute("x" * 2000).action.value)

# Expected output (note: LengthGuardrail returns MODIFY, not BLOCK, so the AND
# group does not fail on length — both cases pass):
# pass
# pass

Exercise 4: Guardrail metrics dashboard

Create a system that collects metrics from the GuardrailChain and presents them as a dashboard.

from collections import Counter, defaultdict


class GuardrailMetrics:
    def __init__(self):
        self.total_runs = 0
        self.action_counts = Counter()
        self.guardrail_times: dict[str, list[float]] = defaultdict(list)
        self.block_reasons = Counter()

    def record(self, chain_result: ChainResult):
        self.total_runs += 1
        for result in chain_result.results:
            self.action_counts[result.action.value] += 1
            self.guardrail_times[result.name].append(result.execution_time_ms)
            if result.action == GuardrailAction.BLOCK:
                self.block_reasons[result.name] += 1

    def dashboard(self) -> str:
        lines = ["=== Guardrail Dashboard ==="]
        lines.append(f"Total runs: {self.total_runs}")
        lines.append(f"Actions: {dict(self.action_counts)}")
        if self.block_reasons:
            lines.append(f"Block reasons: {dict(self.block_reasons)}")
        for name, times in self.guardrail_times.items():
            avg = sum(times) / len(times)
            lines.append(f"  {name}: avg {avg:.3f}ms ({len(times)} calls)")
        return "\n".join(lines)


metrics = GuardrailMetrics()
chain = GuardrailChain([LengthGuardrail(100), LanguageConsistencyGuardrail()])

for text in ["short", "x" * 200, "hello world", "hola mundo"]:
    result = chain.run(text, {"input_language": "es"})
    metrics.record(result)

print(metrics.dashboard())

# Expected output (the demo inputs don't trigger the Spanish/English word lists,
# so no warnings appear; times vary by hardware):
# === Guardrail Dashboard ===
# Total runs: 4
# Actions: {'pass': 7, 'modify': 1}
#   length_check: avg 0.001ms (4 calls)
#   language_consistency: avg 0.002ms (4 calls)

Summary

• 🔑 Guardrails are the orchestration layer that coordinates validation, filtering, and business rules — broader than Pydantic or content filtering on their own
• 🔑 guardrails-ai is a framework with a hub of pre-built validators; NVIDIA's NeMo Guardrails uses Colang for conversational flows; custom guardrails give maximum control and minimum latency
• 🔑 The GuardrailChain runs guardrails in order with fail-fast: the first BLOCK stops the chain, MODIFY transforms the text for the next guardrail
• 🔑 The performance impact of local guardrails (regex) is minimal (<1ms); guardrails with API calls (Moderation, LLM-based) add 100-500ms
• 🔑 Shadow mode lets you evaluate new guardrails in production without blocking: they log but don't act, giving you data to calibrate before enabling
• 🔑 Guardrail profiles (strict/moderate/minimal) let you adjust the level of protection per endpoint or context
• 🔑 AND/OR composition lets you build complex rules from simple guardrails
• 🔑 Guardrail metrics are essential for operations: false positive rate, latency per guardrail, block reasons

Additional resources

1. Guardrails AI Documentation — Open-source framework for LLM output validation with a hub of validators
2. NeMo Guardrails (NVIDIA) — Conversational guardrails framework with Colang
3. Guardrails AI Hub — Hub of pre-built validators for guardrails-ai
4. OWASP LLM05: Improper Output Handling — Vulnerability that guardrails mitigate
5. LLM Guard — Open-source alternative for guardrails with a security focus
6. Rebuff — Prompt Injection Detector — Tool specifically for prompt injection detection as a guardrail
7. Langchain Safety — Langchain's safety guide with guardrail patterns
8. Building LLM Guardrails (Anthropic) — Anthropic's patterns for constraining LLM outputs

Created: March 2026 Version: 1.0