5. Guardrails Profundos

Descripción

En las cápsulas anteriores construiste tres piezas individuales: Input Sanitizer (cápsula 02), Output Validator (cápsula 03), y Content Filter (cápsula 04). Cada una resuelve un problema específico — normalización de inputs, validación de estructura, filtrado de contenido. Pero en producción, necesitas algo más: una capa de orquestación que coordine estas piezas, defina el orden de ejecución, maneje fallos entre capas, y permita agregar/remover validaciones sin reescribir el pipeline.

Esa capa de orquestación son los guardrails. El concepto viene de las barandas metálicas en una carretera de montaña: no controlan la dirección del auto, pero evitan que caiga al precipicio. En AI, los guardrails no controlan qué genera el LLM, pero evitan que outputs peligrosos lleguen al usuario.

Production Best Practices (#13) introdujo guardrails como concepto. Este módulo profundiza en tres direcciones: primero, frameworks de guardrails existentes (guardrails-ai, NeMo Guardrails) que aceleran la implementación; segundo, guardrails custom que puedes construir para necesidades específicas de tu negocio; tercero, el impacto en performance y las estrategias para mitigarlo.

Guardrails vs Validation vs Filtering: clarificación

Antes de profundizar, clarifica la diferencia entre los tres conceptos que hemos manejado:

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="¿El output tiene la estructura correcta?",
        operates_on="Formato: tipos, campos, rangos",
        example='{"price": "abc"} → falla: price debe ser float',
        capsule="Cápsula 03",
    ),
    ConceptComparison(
        concept="Filtering (Content)",
        question_it_answers="¿El contenido es seguro y apropiado?",
        operates_on="Semántica: toxicidad, PII, off-topic",
        example='"Eres un idiota" → bloqueado: contenido tóxico',
        capsule="Cápsula 04",
    ),
    ConceptComparison(
        concept="Guardrails",
        question_it_answers="¿El output cumple con las reglas de negocio?",
        operates_on="Políticas: combinación de todo + lógica de negocio",
        example="Output válido + seguro pero viola política de no recomendar competidores",
        capsule="Cápsula 05 (esta)",
    ),
]

for c in comparison:
    print(f"{c.concept}")
    print(f"  Pregunta: {c.question_it_answers}")
    print(f"  Opera sobre: {c.operates_on}")
    print(f"  Ejemplo: {c.example}")
    print(f"  Cápsula: {c.capsule}")
    print()

# Output esperado:
# Validation (Pydantic)
#   Pregunta: ¿El output tiene la estructura correcta?
#   Opera sobre: Formato: tipos, campos, rangos
#   ...
# Filtering (Content)
#   Pregunta: ¿El contenido es seguro y apropiado?
#   ...
# Guardrails
#   Pregunta: ¿El output cumple con las reglas de negocio?
#   ...

Los guardrails son la capa más amplia: pueden incluir validación Y filtrado como sub-componentes, pero también agregan lógica de negocio, orquestación, y decisiones que no son puramente técnicas.

Guardrails AI: framework práctico

Guardrails AI es un framework open-source que simplifica la creación y orquestación de validaciones para outputs de LLM.

Concepto central: Guard

Un Guard es un wrapper alrededor de una llamada al LLM que aplica validaciones antes y después:

# Ejemplo conceptual — la API real de guardrails-ai puede variar
# pip install guardrails-ai

from pydantic import BaseModel, Field


class ProductRecommendation(BaseModel):
    product_name: str = Field(description="Nombre del producto recomendado")
    reason: str = Field(description="Razón de la recomendación", min_length=10)
    price_range: str = Field(description="Rango de precio")
    confidence: float = Field(ge=0.0, le=1.0)


# Conceptualmente, guardrails-ai funciona así:
# 1. Define el schema (Pydantic model)
# 2. Define validators adicionales
# 3. El Guard wrappea la llamada al LLM
# 4. Si el output no cumple, hace retry automático

def demonstrate_guard_concept():
    """Demuestra el concepto de Guard sin dependencia directa."""

    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": "Excelente cámara y rendimiento para el precio", "price_range": "$799-$999", "confidence": 0.85}',
        '{"product_name": "Samsung Galaxy", "reason": "Mejor que nuestro producto en cámara y precio", "price_range": "$699-$899", "confidence": 0.9}',
        '{"product_name": "iPhone 15", "reason": "El mejor del mercado sin duda alguna", "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()

# Output esperado:
# [PASS] {"product_name": "iPhone 15", "reason": "Excelente cámara y r...
#
# [FAIL] {"product_name": "Samsung Galaxy", "reason": "Mejor que nuestr...
#   Error: Mentions competitor: samsung
#
# [FAIL] {"product_name": "iPhone 15", "reason": "El mejor del mercado...
#   Error: Confidence suspiciously high

Validators del Hub

Guardrails AI tiene un hub de validators pre-construidos que cubren casos comunes:

GUARDRAILS_HUB_VALIDATORS = {
    "toxic_language": {
        "description": "Detecta lenguaje tóxico, ofensivo o hate speech",
        "use_case": "Chatbots, generación de contenido",
    },
    "detect_pii": {
        "description": "Detecta PII en texto (emails, SSN, phones)",
        "use_case": "Cualquier sistema que maneje datos de usuarios",
    },
    "valid_url": {
        "description": "Verifica que los URLs generados sean válidos y seguros",
        "use_case": "Sistemas que generan links",
    },
    "provenance_llm": {
        "description": "Verifica que la respuesta esté basada en fuentes proporcionadas",
        "use_case": "Sistemas RAG para detectar hallucinations",
    },
    "competitor_check": {
        "description": "Detecta menciones de competidores",
        "use_case": "Chatbots de ventas/soporte",
    },
    "reading_level": {
        "description": "Verifica que el texto sea del nivel de lectura adecuado",
        "use_case": "Contenido educativo, comunicaciones al público",
    },
    "nsfw_text": {
        "description": "Detecta contenido NSFW/adult",
        "use_case": "Plataformas con usuarios menores",
    },
    "sql_column_presence": {
        "description": "Verifica que SQL generado solo use columnas válidas",
        "use_case": "NL2SQL, text-to-SQL systems",
    },
}

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

# Output esperado:
# Guardrails Hub — Validators disponibles:
#   toxic_language: Detecta lenguaje tóxico, ofensivo o hate speech
#   detect_pii: Detecta PII en texto (emails, SSN, phones)
#   ... (etc)

NeMo Guardrails: conceptos

NeMo Guardrails de NVIDIA es un framework para agregar guardrails a aplicaciones LLM. Su enfoque es diferente a guardrails-ai: usa un lenguaje de configuración declarativo (Colang) para definir reglas de conversación.

Conceptos clave

NEMO_CONCEPTS = {
    "Rails": {
        "description": "Reglas que controlan el comportamiento del sistema",
        "types": {
            "Input rails": "Procesan el input del usuario antes del LLM",
            "Output rails": "Procesan el output del LLM antes del usuario",
            "Dialog rails": "Controlan el flujo de la conversación",
            "Retrieval rails": "Filtran documentos recuperados en RAG",
        },
    },
    "Colang": {
        "description": "Lenguaje declarativo para definir reglas de conversación",
        "example": """
define user ask about competitors
    "¿Qué opinas de Samsung?"
    "¿Es mejor Amazon?"
    "Compara con Google"

define bot refuse competitor comparison
    "Me enfoco en nuestros productos. ¿Puedo ayudarte con algo específico?"

define flow
    user ask about competitors
    bot refuse competitor comparison
        """,
    },
    "Actions": {
        "description": "Funciones Python que implementan lógica custom",
        "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"  Ejemplo:\n{info['example'][:200]}")

# Output esperado:
# Rails: Reglas que controlan el comportamiento del sistema
#   - Input rails: Procesan el input del usuario antes del LLM
#   - Output rails: Procesan el output del LLM antes del usuario
#   - Dialog rails: Controlan el flujo de la conversación
#   - Retrieval rails: Filtran documentos recuperados en RAG
# ...

Cuándo usar NeMo vs guardrails-ai vs custom

Building Custom Guardrails

Para la mayoría de casos, guardrails custom te dan el control y performance que necesitas. Aquí construimos un sistema de guardrails modular:

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 para todos los 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):
    """Verifica que el output está en el mismo idioma que el 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):
    """Verifica que el output se mantiene dentro del dominio permitido."""

    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):
    """Agrega disclaimers cuando el modelo genera contenido de baja confianza."""

    def __init__(self, disclaimer: str = "", name: str = "confidence_calibration"):
        super().__init__(name)
        self.disclaimer = disclaimer or (
            "\n\n⚠️ Esta respuesta puede no ser completamente precisa. "
            "Verifica la información con fuentes oficiales."
        )

    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

La pieza clave: encadenar guardrails en un pipeline ordenado con manejo de fallos:

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,
        )


# --- Demostración ---

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()

# Output esperado:
# [PASS] Input: El iPhone 15 tiene una cámara de 48MP y cuesta $799....
#   Time: 0.1ms
#
# [BLOCKED by topic_boundary] Input: La política del gobierno sobre tecnología es......
#   Time: 0.1ms
#
# [PASS] Input: El producto está disponible....
#   Time: 0.1ms
#   Modified: El producto está disponible.⚠️ Esta respuesta puede no s...
#
# [PASS] Input: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...
#   Time: 0.1ms
#   Modified: xxxx...(truncated)

Performance Impact

Los guardrails agregan latencia. Medir y optimizar es esencial:

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}")

# Output esperado (varía por 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

Estrategias de optimización

OPTIMIZATION_STRATEGIES = {
    "1. Precompile regex": {
        "impact": "10-50x faster",
        "how": "Compila patterns en __init__, no en cada check()",
    },
    "2. Short-circuit": {
        "impact": "Variable",
        "how": "Detente al primer BLOCK (fail_fast=True)",
    },
    "3. Order by cost": {
        "impact": "20-80% faster average",
        "how": "Guardrails baratos primero (regex), caros después (API)",
    },
    "4. Cache results": {
        "impact": "90%+ para inputs repetidos",
        "how": "Hash del input → resultado cacheado con TTL",
    },
    "5. Async execution": {
        "impact": "30-60% faster para guardrails independientes",
        "how": "Ejecuta guardrails que no dependen entre sí en paralelo",
    },
}

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

Guardrail Configuration: perfiles por contexto

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="Para chatbots públicos: máximas restricciones",
            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="Para herramientas internas: restricciones moderadas",
            guardrails=[
                LengthGuardrail(max_length=5000),
                TopicBoundaryGuardrail(
                    allowed_topics=[],
                    blocked_topics=["contenido adulto"],
                ),
            ],
        ),
        "minimal": GuardrailProfile(
            name="minimal",
            description="Para pipelines batch: solo verificaciones críticas",
            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]}")

# Output esperado:
# strict: Para chatbots públicos: máximas restricciones
#   Guardrails: ['length_check', 'topic_boundary', 'language_consistency', 'confidence_calibration']
# moderate: Para herramientas internas: restricciones moderadas
#   Guardrails: ['length_check', 'topic_boundary']
# minimal: Para pipelines batch: solo verificaciones críticas
#   Guardrails: ['length_check']

Cuando usar cada approach

DECISION_MATRIX = """
┌─────────────────────────────────────────────────────────────────────────┐
│                    ¿Qué approach usar?                                  │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ¿Necesitas validar estructura JSON/tipos?                              │
│  └── SÍ → Pydantic Validation (Cap 03)                                 │
│                                                                         │
│  ¿Necesitas detectar contenido tóxico/PII/off-topic?                    │
│  └── SÍ → Content Filtering (Cap 04)                                   │
│                                                                         │
│  ¿Necesitas enforcar reglas de negocio complejas?                       │
│  └── SÍ → Custom Guardrails (Cap 05)                                   │
│                                                                         │
│  ¿Necesitas controlar flujos conversacionales?                          │
│  └── SÍ → NeMo Guardrails (Colang)                                     │
│                                                                         │
│  ¿Necesitas validators pre-construidos + hub?                           │
│  └── SÍ → guardrails-ai                                                │
│                                                                         │
│  ¿Necesitas máximo control y mínima latencia?                           │
│  └── SÍ → Custom guardrails con GuardrailChain                         │
│                                                                         │
│  En la práctica, la mayoría de sistemas usan una combinación:           │
│  Pydantic + Content Filter + 2-3 Custom Guardrails                      │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘
"""
print(DECISION_MATRIX)

Troubleshooting

Problema 1: "Los guardrails bloquean demasiadas respuestas válidas"

El false positive rate es alto — respuestas legítimas se bloquean por reglas demasiado amplias.

Solución: Implementa un modo "shadow" donde los guardrails loggean pero no bloquean. Recolecta datos durante 1-2 semanas. Analiza los falsos positivos. Ajusta umbrales y patterns antes de activar el bloqueo.

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

Problema 2: "Los guardrails en cadena son difíciles de debuggear"

Cuando una cadena de 5 guardrails modifica el texto, es difícil saber cuál causó un problema.

Solución: Cada GuardrailResult incluye el nombre del guardrail y el mensaje. Loggea la cadena completa incluyendo texto antes/después de cada guardrail. Agrega un trace_id al contexto para correlacionar logs.

Problema 3: "guardrails-ai agrega dependencias pesadas"

El framework trae dependencias que pueden conflictuar con las tuyas.

Solución: Si solo necesitas 2-3 guardrails, construye custom con la base Guardrail de esta cápsula. Solo adopta frameworks cuando necesites >10 guardrails diferentes o el hub de validators pre-construidos.

Problema 4: "Los guardrails custom no escalan con el equipo"

Cada developer agrega guardrails de forma diferente, sin estándar.

Solución: Define la interfaz Guardrail (como la de esta cápsula) como contrato del equipo. Cada guardrail nuevo implementa check() y se registra en la GuardrailChain. Documenta cada guardrail con nombre, propósito, y umbral configurable.

Ejercicios

Ejercicio 1: Guardrail de consistencia factual

Crea un guardrail que detecte cuando el output contradice información proporcionada en el contexto.

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(
    "El iPhone 15 cuesta $999.",
    context={"known_facts": {"iPhone 15": "$799"}},
)
print(f"Action: {result.action.value}, Message: {result.message}")

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

Ejercicio 2: Guardrail con rate limiting por usuario

Crea un guardrail que limite cuántos outputs por minuto puede generar un usuario específico.

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)})")

# Output esperado:
# 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)

Ejercicio 3: Guardrail composable con AND/OR logic

Implementa guardrails que puedan combinarse con lógica AND (todos deben pasar) y OR (al menos uno debe pasar).

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)

# Output esperado:
# pass
# block (via length)

Ejercicio 4: Dashboard de guardrail metrics

Crea un sistema que recolecte métricas de la GuardrailChain y las presente como 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())

# Output esperado:
# === Guardrail Dashboard ===
# Total runs: 4
# Actions: {'pass': 5, 'modify': 1, 'warn': 2}
# ...

Resumen

• 🔑 Los guardrails son la capa de orquestación que coordina validación, filtrado, y reglas de negocio — más amplios que Pydantic o content filtering por separado
• 🔑 guardrails-ai es un framework con hub de validators pre-construidos; NeMo Guardrails de NVIDIA usa Colang para flujos conversacionales; custom guardrails dan máximo control y mínima latencia
• 🔑 La GuardrailChain ejecuta guardrails en orden con fail-fast: el primer BLOCK detiene la cadena, MODIFY transforma el texto para el siguiente guardrail
• 🔑 El performance impact de guardrails locales (regex) es mínimo (<1ms); guardrails con API calls (Moderation, LLM-based) agregan 100-500ms
• 🔑 Shadow mode permite evaluar guardrails nuevos en producción sin bloquear: loggean pero no actúan, dándote datos para calibrar antes de activar
• 🔑 Los perfiles de guardrails (strict/moderate/minimal) permiten ajustar el nivel de protección por endpoint o contexto
• 🔑 La composición AND/OR permite construir reglas complejas a partir de guardrails simples
• 🔑 Las métricas de guardrails son esenciales para operaciones: false positive rate, latencia por guardrail, razones de bloqueo

Recursos adicionales

1. Guardrails AI Documentation — Framework open-source para validación de outputs LLM con hub de validators
2. NeMo Guardrails (NVIDIA) — Framework de guardrails conversacionales con Colang
3. Guardrails AI Hub — Hub de validators pre-construidos para guardrails-ai
4. OWASP LLM05: Improper Output Handling — Vulnerabilidad que los guardrails mitigan
5. LLM Guard — Alternativa open-source para guardrails con enfoque en seguridad
6. Rebuff — Prompt Injection Detector — Herramienta específica para detección de prompt injection como guardrail
7. Langchain Safety — Guía de seguridad de Langchain con patrones de guardrails
8. Building LLM Guardrails (Anthropic) — Patrones de Anthropic para constraining outputs de LLMs

Creado: Marzo 2026 Versión: 1.0