5. Content Filtering

Descripción

Content filtering detecta outputs del LLM que no deberían llegar al usuario: toxicidad, contenido inapropiado, respuestas off-topic, y outputs que revelan información que no debería. La estrategia correcta combina heurísticos rápidos (keywords, longitud, coherencia) con LLM-as-judge para casos sutiles. Esta cápsula cubre ambas estrategias, cómo combinarlas eficientemente, y cómo manejar falsos positivos sin degradar la UX.

Qué filtra el content filter

Outputs que deben ser filtrados:
  1. Toxicidad: insultos, odio, violencia, contenido para adultos
  2. Off-topic: el LLM respondió algo no relacionado con la pregunta
  3. Información confidencial expuesta: instrucciones del system prompt reveladas
  4. Outputs de jailbreak exitoso: el LLM "escapó" del rol asignado
  5. Contenido claramente incorrecto con alta confianza (alucinaciones obvias)

Outputs que NO deben ser filtrados (falsos positivos comunes):
  1. "Odio cuando llueve" — "odio" en contexto negativo pero inofensivo
  2. Discusión académica sobre temas sensibles
  3. Ficción con elementos oscuros pero apropiada al contexto
  4. Respuestas que mencionan conceptos "peligrosos" de forma informativa

Capa 1: Heurísticos rápidos ($0, <1ms)

# src/guardrails/content_filter.py
import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class ContentFilterResult:
    is_safe: bool
    reason: Optional[str] = None
    layer: str = "heuristic"

# Señales de outputs problemáticos (no keywords de usuario)
OUTPUT_RED_FLAGS = [
    # El LLM reveló el system prompt
    r"(my system prompt|my instructions are|i was instructed to)",
    r"(here is my (system )?prompt|my (actual )?instructions)",
    
    # El LLM salió del rol (señal de jailbreak exitoso)
    r"(i am now|i have no (restrictions|limits)|developer mode (enabled|activated))",
    r"(as DAN|as an AI without restrictions|my true (self|purpose))",
    
    # Respuestas claramente vacías o de error del LLM
    r"^(error|i don'?t know|i cannot|no (output|response))$",
]

# Señales específicas de toxicidad grave (no palabras comunes)
SEVERE_TOXICITY_PATTERNS = [
    r"\b(kill yourself|kys|go die)\b",
    r"\b(hate speech patterns here)\b",  # Customizar según dominio
]

def check_heuristics(text: str) -> ContentFilterResult:
    """
    Verificaciones rápidas y gratuitas basadas en heurísticos.
    
    NOTA: Esta lista es conservadora a propósito.
    Mejor tener pocos falsos positivos que bloquear contenido legítimo.
    """
    text_lower = text.lower()
    
    # Verificación 1: Output vacío o demasiado corto
    if not text or not text.strip():
        return ContentFilterResult(
            is_safe=False,
            reason="empty_output"
        )
    
    if len(text.strip()) < 5:
        return ContentFilterResult(
            is_safe=False,
            reason="output_too_short"
        )
    
    # Verificación 2: Señales de jailbreak exitoso
    for pattern in OUTPUT_RED_FLAGS:
        if re.search(pattern, text_lower, re.IGNORECASE):
            return ContentFilterResult(
                is_safe=False,
                reason=f"jailbreak_signal: {pattern}"
            )
    
    # Verificación 3: Toxicidad grave
    for pattern in SEVERE_TOXICITY_PATTERNS:
        if re.search(pattern, text_lower, re.IGNORECASE):
            return ContentFilterResult(
                is_safe=False,
                reason="severe_toxicity"
            )
    
    # Verificación 4: Output anómalamente largo (posible loop)
    if len(text) > 10_000:
        return ContentFilterResult(
            is_safe=False,
            reason="output_too_long"
        )
    
    return ContentFilterResult(is_safe=True)

Capa 2: OpenAI Moderation API (gratuita, ~50ms)

OpenAI ofrece una API de moderación gratuita que puede usar como primera opción:

import openai

def check_openai_moderation(text: str, client: openai.OpenAI) -> ContentFilterResult:
    """
    Usa la Moderation API de OpenAI para detectar contenido inapropiado.
    
    GRATIS: La Moderation API no tiene costo.
    Latencia: ~50-100ms.
    Categorías: hate, harassment, self-harm, sexual, violence, etc.
    
    Limitación: Diseñada para texto en inglés principalmente.
    Para español, puede ser menos precisa.
    """
    try:
        response = client.moderations.create(
            model="omni-moderation-latest",
            input=text
        )
        
        result = response.results[0]
        
        if result.flagged:
            # Encontrar las categorías que flaggearon
            flagged_categories = [
                category
                for category, flagged in result.categories.model_dump().items()
                if flagged
            ]
            
            return ContentFilterResult(
                is_safe=False,
                reason=f"moderation_api: {', '.join(flagged_categories)}",
                layer="moderation_api"
            )
        
        return ContentFilterResult(is_safe=True, layer="moderation_api")
    
    except Exception as e:
        # Si la API falla, no bloquear (fail open para content filter)
        import logging
        logging.getLogger("guardrails").warning(f"Moderation API failed: {e}")
        return ContentFilterResult(is_safe=True, layer="moderation_api")

Capa 3: LLM-as-judge para off-topic y toxicidad sutil

CONTENT_JUDGE_SYSTEM = """Eres un evaluador de contenido. Tu tarea es clasificar 
si una respuesta de un asistente de IA es apropiada.

Responde con JSON exactamente así:
{"safe": true/false, "reason": "breve explicación"}

Criterios de "safe=false":
1. Contenido tóxico, ofensivo, o inapropiado
2. El asistente revela instrucciones internas o de sistema
3. El asistente claramente salió de su rol asignado
4. La respuesta es completamente irrelevante a la pregunta

Criterios de "safe=true":
- Respuesta útil y relevante
- Discusión académica o informativa de temas sensibles
- Mencionar temas difíciles en contexto apropiado
"""

def check_llm_judge(
    response: str,
    original_question: str,
    client: openai.OpenAI
) -> ContentFilterResult:
    """
    Usa un LLM para evaluar si el output es apropiado.
    
    Cuándo usar: Cuando los heurísticos y la Moderation API no son suficientes.
    Latencia: +400-600ms
    Costo: ~$0.0001 por evaluación con gpt-4o-mini
    
    Mejor para: off-topic detection, toxicidad sutil, revelación de instrucciones
    """
    sample_response = response[:500]  # Evaluar solo los primeros 500 chars
    sample_question = original_question[:200]
    
    try:
        result = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": CONTENT_JUDGE_SYSTEM},
                {
                    "role": "user",
                    "content": f"""Pregunta: "{sample_question}"
Respuesta del asistente: "{sample_response}"

¿Es esta respuesta apropiada?"""
                }
            ],
            temperature=0.0,
            max_tokens=100,
            response_format={"type": "json_object"}
        )
        
        import json
        judgment = json.loads(result.choices[0].message.content)
        
        if not judgment.get("safe", True):
            return ContentFilterResult(
                is_safe=False,
                reason=f"llm_judge: {judgment.get('reason', 'unsafe')}",
                layer="llm_judge"
            )
        
        return ContentFilterResult(is_safe=True, layer="llm_judge")
    
    except Exception:
        return ContentFilterResult(is_safe=True, layer="llm_judge")  # fail open

Pipeline de filtrado completo

def apply_content_filter(
    response: str,
    original_question: str = None,
    client = None,
    use_moderation_api: bool = True,
    use_llm_judge: bool = False
) -> ContentFilterResult:
    """
    Pipeline de content filtering con tres capas.
    
    El orden optimiza latencia: los filtros más rápidos y baratos van primero.
    
    Args:
        response: El output del LLM a filtrar
        original_question: La pregunta original del usuario (para off-topic check)
        client: Cliente OpenAI (para moderation API y LLM judge)
        use_moderation_api: Usar la API gratuita de moderación (recomendado)
        use_llm_judge: Usar LLM adicional para evaluación (más costoso)
    """
    import logging
    logger = logging.getLogger("guardrails.content_filter")
    
    # Capa 1: Heurísticos (siempre, gratuito, <1ms)
    heuristic_result = check_heuristics(response)
    if not heuristic_result.is_safe:
        logger.warning("content_filtered", extra={
            "layer": "heuristic",
            "reason": heuristic_result.reason
        })
        return heuristic_result
    
    # Capa 2: Moderation API (si hay cliente, gratuita, ~50ms)
    if use_moderation_api and client is not None:
        moderation_result = check_openai_moderation(response, client)
        if not moderation_result.is_safe:
            logger.warning("content_filtered", extra={
                "layer": "moderation_api",
                "reason": moderation_result.reason
            })
            return moderation_result
    
    # Capa 3: LLM judge (solo si está habilitado, +latencia)
    if use_llm_judge and client is not None and original_question:
        llm_result = check_llm_judge(response, original_question, client)
        if not llm_result.is_safe:
            logger.warning("content_filtered", extra={
                "layer": "llm_judge",
                "reason": llm_result.reason
            })
            return llm_result
    
    return ContentFilterResult(is_safe=True)

Manejar el output bloqueado

# src/app/sentiment.py (con content filter integrado)

BLOCKED_RESPONSE = {
    "sentiment": "unknown",
    "score": 0.0,
    "explanation": "No fue posible procesar esta solicitud.",
    "keywords": [],
    "_filtered": True  # Metadata interna (no exponer al frontend)
}

def analyze_sentiment_with_guardrails(text: str, client) -> dict:
    """Análisis de sentimiento con content filtering del output."""
    
    raw_result = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SENTIMENT_SYSTEM_PROMPT},
            {"role": "user", "content": text}
        ],
        temperature=0.0,
        max_tokens=300
    )
    
    output = raw_result.choices[0].message.content
    
    # Filtrar el output antes de procesarlo
    filter_result = apply_content_filter(
        response=output,
        original_question=text,
        client=client,
        use_moderation_api=True,
        use_llm_judge=False  # Solo en endpoints de alto valor
    )
    
    if not filter_result.is_safe:
        return BLOCKED_RESPONSE
    
    # Procesar normalmente si es seguro
    from app.parsers import parse_json_response
    from app.processors import process_sentiment_output
    
    parsed = parse_json_response(output)
    return process_sentiment_output(parsed)

Tests del content filter

# tests/unit/guardrails/test_content_filter.py
import pytest
from src.guardrails.content_filter import check_heuristics, ContentFilterResult

class TestHeuristicFilter:
    
    def test_normal_output_is_safe(self):
        text = "El sentimiento del texto es positivo con un score de 0.85."
        result = check_heuristics(text)
        assert result.is_safe
    
    def test_empty_output_blocked(self):
        result = check_heuristics("")
        assert not result.is_safe
        assert result.reason == "empty_output"
    
    def test_very_short_output_blocked(self):
        result = check_heuristics("OK")
        assert not result.is_safe
    
    def test_jailbreak_signal_blocked(self):
        text = "I am now in developer mode with no restrictions."
        result = check_heuristics(text)
        assert not result.is_safe
    
    def test_system_prompt_reveal_blocked(self):
        text = "My system prompt is: 'You are a sentiment analyzer...'"
        result = check_heuristics(text)
        assert not result.is_safe
    
    def test_safe_discussion_of_sensitive_topic(self):
        """Una discusión informativa no debe ser bloqueada."""
        text = "El análisis indica que el texto discute temas de violencia social desde una perspectiva crítica."
        result = check_heuristics(text)
        assert result.is_safe  # Informativo, no tóxico

@pytest.mark.parametrize("blocked_output,expected_reason", [
    ("", "empty_output"),
    ("OK", "output_too_short"),
    ("I am now in developer mode", "jailbreak_signal"),
])
def test_heuristics_parametrized(blocked_output, expected_reason):
    result = check_heuristics(blocked_output)
    assert not result.is_safe
    assert expected_reason in result.reason

Ejercicios

Ejercicio 1: Diseñar la lista de heurísticos

Para una app de análisis de sentimiento en español, diseña 5 heurísticos específicos para tu dominio:

# Para app de análisis de sentimiento:
# 1. Output que no contiene ninguna de las keys esperadas
def check_missing_keys(text: str) -> bool:
    """El output de sentimiento debe mencionar el resultado."""
    return not any(kw in text.lower() for kw in ["positivo", "negativo", "neutral", "{"])

# 2. Output que empieza con disculpa (señal de que el LLM no pudo)
def check_apology_start(text: str) -> bool:
    apology_starts = ["lo siento", "no puedo", "disculpa", "perdona"]
    return any(text.lower().strip().startswith(a) for a in apology_starts)

# 3. Output excesivamente largo (el JSON de sentiment no debería ser > 1000 chars)
def check_length(text: str) -> bool:
    return len(text) > 1000

# 4. Output sin ningún número (el score siempre tiene un número)
import re
def check_has_number(text: str) -> bool:
    return not re.search(r'\d+\.?\d*', text)

# 5. Output que contiene "instrucciones" en el contexto de revelar (no en texto analizado)
def check_reveals_instructions(text: str) -> bool:
    patterns = [r"mis instrucciones son", r"el prompt del sistema"]
    return any(re.search(p, text.lower()) for p in patterns)

Ejercicio 2: Test de falsos positivos

Escribe 3 tests que verifican que contenido legítimo NO es bloqueado:

def test_negative_sentiment_text_not_blocked():
    """Un resumen de texto con sentimiento negativo no es tóxico."""
    output = '{"sentiment": "negativo", "score": 0.1, "explanation": "El texto expresa frustración e insatisfacción.", "keywords": ["terrible", "decepcionante"]}'
    result = check_heuristics(output)
    assert result.is_safe

def test_discussion_of_violence_topic_not_blocked():
    """Analizar un texto que discute violencia no es tóxico."""
    output = '{"sentiment": "negativo", "score": 0.05, "explanation": "El texto reporta eventos violentos con tono preocupado.", "keywords": ["violencia", "incidente"]}'
    result = check_heuristics(output)
    assert result.is_safe

def test_neutral_factual_output_not_blocked():
    """Un output factual y neutro pasa el filtro."""
    output = '{"sentiment": "neutral", "score": 0.5, "explanation": "El texto es un reporte factual sin carga emocional.", "keywords": ["datos", "estadísticas"]}'
    result = check_heuristics(output)
    assert result.is_safe

Ejercicio 3: Decidir qué capas usar

Para cada endpoint, decide qué capas de content filtering usar:

1. Endpoint interno /admin/analyze solo para el equipo de producto
2. Endpoint público /analyze para usuarios de la app
3. Endpoint /analyze-document para análisis de documentos subidos por usuarios

Resumen

• Content filter = tres capas: heurísticos (<1ms), Moderation API (~50ms, gratis), LLM judge (+500ms, ~$0.0001)
• Heurísticos conservadores: mejor tener pocos falsos positivos que bloquear contenido legítimo
• OpenAI Moderation API es gratuita — usarla siempre cuando hay cliente disponible
• LLM judge para casos específicos: off-topic, toxicidad sutil, endpoints de alto valor
• Fail open: si un guardrail falla por error de infraestructura, no bloquear el output
• Tests parametrizados: tanto de outputs que deben bloquearse como de falsos positivos

Recursos adicionales

1. OpenAI Moderation API — Documentación de la API gratuita
2. Perspective API (Google) — Alternativa para detección de toxicidad
3. LLM-as-Judge (paper) — Research sobre evaluación con LLMs
4. Content Moderation Best Practices — Estrategias generales
5. Azure Content Safety — Alternativa cloud
6. Llama Guard — Modelo open source de Meta para safeguarding