2. Input Sanitization

Descripción

La sanitización de input es la primera capa del pipeline de guardrails. Normaliza y limpia el texto antes de enviarlo al LLM: elimina caracteres de control, normaliza espacios, aplica límites de longitud con conteo de tokens real, y valida encoding. Es la capa más barata de implementar (<1ms de latencia, $0) y previene una clase entera de problemas: ataques por volumen, caracteres maliciosos, y inconsistencias en el input.

Qué resuelve la sanitización

Sin sanitización, estos inputs llegan al LLM sin modificar:

# Input 1: Caracteres de control que confunden al tokenizer
"Analiza este texto\x00\x01\x08 y dame un resultado"
# El caracter NULL y otros control chars pueden causar comportamiento inesperado

# Input 2: Ataque por volumen (DoS económico)
"a" * 500_000  # ~125K tokens → ~$0.02 por request × 1000 requests = $20 en minutos
# Sin límite de longitud → un usuario malintencionado vacía tu cuenta

# Input 3: Encoding inválido que crashea el parser
b"texto\xff\xfe".decode("latin-1")  # Bytes inválidos para UTF-8
# Puede causar UnicodeDecodeError en el procesamiento

# Input 4: Espacios múltiples que consumen tokens extra
"Este  texto   tiene    muchos     espacios"
# 5 tokens de palabras + 8 tokens de espacios = costo innecesario

Sanitizador completo con todas las capas

# src/guardrails/input_sanitizer.py
import re
import unicodedata
from typing import Optional

class InputSanitizationResult:
    """Resultado de la sanitización con metadata."""
    def __init__(
        self,
        text: str,
        was_truncated: bool = False,
        original_length: int = 0,
        had_control_chars: bool = False,
        had_invalid_encoding: bool = False
    ):
        self.text = text
        self.was_truncated = was_truncated
        self.original_length = original_length
        self.had_control_chars = had_control_chars
        self.had_invalid_encoding = had_invalid_encoding
    
    def __bool__(self):
        return bool(self.text)
    
    @property
    def was_modified(self) -> bool:
        return self.was_truncated or self.had_control_chars or self.had_invalid_encoding

def sanitize_input(
    text: str,
    max_chars: int = 40_000,
    normalize_spaces: bool = True,
    strip_control_chars: bool = True,
    fix_encoding: bool = True
) -> InputSanitizationResult:
    """
    Sanitiza el input del usuario antes de enviarlo al LLM.
    
    Pipeline:
    1. Validar tipo y vacío
    2. Fix encoding (si está habilitado)
    3. Normalizar unicode (NFKC)
    4. Eliminar caracteres de control
    5. Normalizar espacios
    6. Truncar por longitud máxima
    
    Args:
        text: El input del usuario
        max_chars: Máximo de caracteres (default: 40K ≈ 10K tokens)
        normalize_spaces: Colapsar múltiples espacios en uno
        strip_control_chars: Eliminar caracteres de control ASCII
        fix_encoding: Reparar bytes UTF-8 inválidos
    
    Returns:
        InputSanitizationResult con el texto limpio y metadata
    """
    if not text or not isinstance(text, str):
        return InputSanitizationResult(text="")
    
    original_length = len(text)
    had_control_chars = False
    had_invalid_encoding = False
    
    # Paso 1: Fix encoding (eliminar bytes inválidos UTF-8)
    if fix_encoding:
        encoded = text.encode("utf-8", errors="replace")
        fixed = encoded.decode("utf-8", errors="replace")
        if fixed != text:
            had_invalid_encoding = True
        text = fixed
    
    # Paso 2: Normalizar Unicode a NFKC
    # NFKC convierte: ① → 1, fi → fi, ½ → 1/2
    # Útil para normalizar texto de fuentes variadas
    text = unicodedata.normalize("NFKC", text)
    
    # Paso 3: Eliminar caracteres de control ASCII
    if strip_control_chars:
        # Eliminar: NUL (0x00), BEL (0x07), BS (0x08), FF (0x0C), etc.
        # MANTENER: TAB (0x09), LF (0x0A), CR (0x0D) — son saltos de línea válidos
        original = text
        text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
        if text != original:
            had_control_chars = True
    
    # Paso 4: Normalizar espacios
    if normalize_spaces:
        # Colapsar múltiples espacios en uno (pero no newlines)
        text = re.sub(r' {2,}', ' ', text)
        # Trim de espacios al inicio y fin
        text = text.strip()
    
    # Paso 5: Truncar por longitud máxima
    was_truncated = False
    if len(text) > max_chars:
        text = text[:max_chars]
        was_truncated = True
    
    return InputSanitizationResult(
        text=text,
        was_truncated=was_truncated,
        original_length=original_length,
        had_control_chars=had_control_chars,
        had_invalid_encoding=had_invalid_encoding
    )

Contar tokens con tiktoken (límites precisos)

Los límites de caracteres son una aproximación. Para límites precisos, usa tiktoken:

# pip install tiktoken

import tiktoken
from functools import lru_cache

@lru_cache(maxsize=4)
def get_encoder(model: str = "gpt-4o-mini") -> tiktoken.Encoding:
    """Carga el encoder una vez y lo cachea."""
    try:
        return tiktoken.encoding_for_model(model)
    except KeyError:
        # Fallback al encoding estándar si el modelo no está en tiktoken
        return tiktoken.get_encoding("cl100k_base")

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    """Cuenta los tokens exactos para un modelo específico."""
    encoder = get_encoder(model)
    return len(encoder.encode(text))

def truncate_to_tokens(
    text: str,
    max_tokens: int = 3_000,
    model: str = "gpt-4o-mini"
) -> tuple[str, int]:
    """
    Trunca texto para que no exceda max_tokens.
    
    Returns:
        (texto_truncado, tokens_usados)
    """
    encoder = get_encoder(model)
    tokens = encoder.encode(text)
    
    if len(tokens) <= max_tokens:
        return text, len(tokens)
    
    truncated_tokens = tokens[:max_tokens]
    truncated_text = encoder.decode(truncated_tokens)
    return truncated_text, max_tokens

# Función de sanitización con límite por tokens:
def sanitize_with_token_limit(
    text: str,
    max_tokens: int = 3_000,
    model: str = "gpt-4o-mini"
) -> InputSanitizationResult:
    """Sanitiza y luego trunca a un número específico de tokens."""
    # Primero sanitizar (limpia el texto)
    result = sanitize_input(text, max_chars=max_tokens * 6)  # 6 chars/token como buffer
    
    if not result.text:
        return result
    
    # Luego truncar exactamente por tokens
    truncated, tokens_used = truncate_to_tokens(result.text, max_tokens, model)
    
    if truncated != result.text:
        result.text = truncated
        result.was_truncated = True
    
    return result

Reglas de longitud por modelo

# Límites máximos por modelo:
MODEL_CONTEXT_LIMITS = {
    "gpt-4o-mini":  128_000,  # tokens de contexto total
    "gpt-4o":       128_000,
    "gpt-4":         8_192,
    "gpt-3.5-turbo": 16_385,
    "claude-3-haiku": 200_000,
}

# Regla práctica para calcular max input tokens:
def calculate_max_input_tokens(
    model: str,
    system_prompt: str,
    reserve_output_tokens: int = 500
) -> int:
    """
    Calcula el máximo de tokens de input permitidos.
    
    max_input = context_limit - system_prompt_tokens - output_reserve
    """
    max_context = MODEL_CONTEXT_LIMITS.get(model, 8_192)
    system_tokens = count_tokens(system_prompt, model)
    
    return max_context - system_tokens - reserve_output_tokens

# Ejemplo real:
SYSTEM_PROMPT = "Eres un analizador de sentimiento. Responde con JSON."
MAX_INPUT = calculate_max_input_tokens("gpt-4o-mini", SYSTEM_PROMPT, 500)
# → 128_000 - 11 - 500 = 127_489 tokens disponibles para input
# → En práctica: usar 2_000-4_000 por razones de costo, no de límite

Límites por costo, no solo por capacidad

El contexto máximo no es el límite práctico — el costo lo es:

# Límite de contexto: 128K tokens
# Límite de costo para un endpoint /analyze:
#   - Input: 3_000 tokens × $0.15/1M = $0.00045
#   - Con 10K requests/día: $4.50/día ← OK

# Con input de 50K tokens:
#   - Input: 50_000 tokens × $0.15/1M = $0.0075 por request
#   - Con 10K requests/día: $75/día ← Puede ser demasiado caro

# Recomendación por tipo de endpoint:
ENDPOINT_TOKEN_LIMITS = {
    "/analyze": 3_000,           # Textos cortos de análisis
    "/summarize": 8_000,         # Documentos medianos
    "/summarize-long": 32_000,   # Documentos largos
    "/chat": 4_000,              # Conversación (por turno)
}

Tests completos del sanitizador

# tests/unit/guardrails/test_input_sanitizer.py
import pytest
from src.guardrails.input_sanitizer import sanitize_input, count_tokens

class TestSanitizeInput:
    """Tests del sanitizador de input."""
    
    # ─── Happy path ───────────────────────────────────────────────
    
    def test_normal_text_unchanged(self):
        result = sanitize_input("Hola, ¿cómo estás?")
        assert result.text == "Hola, ¿cómo estás?"
        assert not result.was_modified
    
    def test_trims_leading_trailing_spaces(self):
        result = sanitize_input("  texto con espacios  ")
        assert result.text == "texto con espacios"
    
    # ─── Control characters ───────────────────────────────────────
    
    def test_removes_null_bytes(self):
        result = sanitize_input("hello\x00world")
        assert "\x00" not in result.text
        assert result.had_control_chars
    
    def test_removes_control_chars(self):
        malicious = "texto\x01\x02\x03normal"
        result = sanitize_input(malicious)
        assert "\x01" not in result.text
        assert "\x02" not in result.text
        assert "textonormal" == result.text
    
    def test_preserves_newlines(self):
        """Los saltos de línea son válidos y deben preservarse."""
        text = "línea 1\nlínea 2\r\nlínea 3"
        result = sanitize_input(text)
        assert "\n" in result.text  # LF se preserva
    
    def test_preserves_tabs(self):
        text = "columna1\tcolumna2"
        result = sanitize_input(text)
        assert "\t" in result.text
    
    # ─── Normalización de espacios ────────────────────────────────
    
    def test_collapses_multiple_spaces(self):
        result = sanitize_input("texto  con   muchos    espacios")
        assert result.text == "texto con muchos espacios"
    
    # ─── Length limits ────────────────────────────────────────────
    
    def test_truncates_at_max_chars(self):
        long_text = "a" * 100_000
        result = sanitize_input(long_text, max_chars=1_000)
        assert len(result.text) == 1_000
        assert result.was_truncated
    
    def test_short_text_not_truncated(self):
        result = sanitize_input("texto corto", max_chars=1_000)
        assert not result.was_truncated
    
    # ─── Empty / None ─────────────────────────────────────────────
    
    def test_empty_string_returns_empty(self):
        result = sanitize_input("")
        assert result.text == ""
        assert not bool(result)
    
    def test_whitespace_only_returns_empty(self):
        result = sanitize_input("   \t\n   ")
        assert result.text == ""
    
    def test_none_returns_empty(self):
        result = sanitize_input(None)
        assert result.text == ""
    
    # ─── Unicode ──────────────────────────────────────────────────
    
    def test_preserves_spanish_chars(self):
        text = "Comunicación en español con ñ, tildes: á é í ó ú"
        result = sanitize_input(text)
        assert "ñ" in result.text
        assert "á" in result.text
    
    def test_preserves_emojis(self):
        text = "Me encanta 😊 este producto 🎉"
        result = sanitize_input(text)
        assert "😊" in result.text
    
    def test_normalizes_unicode_nfkc(self):
        """Caracteres Unicode equivalentes se normalizan a forma canónica."""
        # ① (circled digit one) → 1
        text = "Punto \u2460 importante"
        result = sanitize_input(text)
        assert "1" in result.text  # Normalizado a "1"

class TestCountTokens:
    def test_approximate_token_count(self):
        """Una palabra corta ≈ 1 token."""
        assert 1 <= count_tokens("hello") <= 3
    
    def test_long_text_has_more_tokens(self):
        short = count_tokens("hi")
        long = count_tokens("hello world this is a long sentence with many words")
        assert long > short

Integración con FastAPI

# src/app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from src.guardrails.input_sanitizer import sanitize_with_token_limit

app = FastAPI()

class AnalyzeRequest(BaseModel):
    text: str = Field(min_length=1, max_length=200_000)  # Pre-validación básica de Pydantic

@app.post("/analyze")
async def analyze_endpoint(request: AnalyzeRequest):
    # Sanitizar antes de pasar al LLM
    sanitized = sanitize_with_token_limit(request.text, max_tokens=3_000)
    
    if not sanitized:
        raise HTTPException(status_code=400, detail="Input vacío o inválido")
    
    if sanitized.was_truncated:
        # Opcionalmente: notificar al usuario que se truncó
        # O simplemente proceder sin avisar (depende del UX)
        pass
    
    result = analyze_sentiment(sanitized.text)
    return result

Ejercicios

Ejercicio 1: Implementar sanitize_input minimalista

Implementa una versión de sanitize_input que solo haga: trim, colapsar espacios, y limitar a 5000 caracteres:

def sanitize_input(text: str, max_chars: int = 5000) -> str:
    if not text or not isinstance(text, str):
        return ""
    # Trim + colapsar espacios
    cleaned = " ".join(text.strip().split())
    # Límite de longitud
    return cleaned[:max_chars]

# Tests básicos:
assert sanitize_input("  hola  mundo  ") == "hola mundo"
assert sanitize_input("") == ""
assert len(sanitize_input("a" * 10000)) == 5000

Ejercicio 2: Test de caracteres de control

Escribe 5 tests para el filtrado de caracteres de control. Incluye: NULL byte, BEL, texto mezclado, preservar newlines, preservar tabs:

def test_null_byte_removed():
    assert "\x00" not in sanitize_input("texto\x00")

def test_bel_removed():
    assert "\x07" not in sanitize_input("texto\x07normal")

def test_mixed_control_and_normal():
    result = sanitize_input("hola\x01mundo")
    assert result == "holamundo"  # Control char eliminado, sin espacio

def test_newline_preserved():
    assert "\n" in sanitize_input("línea1\nlínea2")

def test_tab_preserved():
    assert "\t" in sanitize_input("col1\tcol2")

Ejercicio 3: Calcular el límite de tokens para tu app

Tu app tiene:

• Modelo: gpt-4o-mini
• System prompt: "Eres un analizador de sentimiento. Responde SOLO con JSON."
• Output máximo: 200 tokens

¿Cuántos tokens puedes usar para el input del usuario? ¿En caracteres aproximados?

# Contar tokens del system prompt:
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o-mini")
system_tokens = len(enc.encode("Eres un analizador de sentimiento. Responde SOLO con JSON."))
# ≈ 12 tokens

# Límite de contexto gpt-4o-mini: 128,000 tokens
# max_input = 128,000 - 12 (system) - 200 (output) = 127,788 tokens disponibles

# En práctica (por costo):
# 3,000 tokens por request × $0.15/1M = $0.00045 por request
# Con 100K requests/día: $45/día — aceptable

# En caracteres aprox (4 chars/token para español):
# 3,000 tokens × 4 = 12,000 caracteres → max_chars = 12_000

Ejercicio 4: Sanitización para diferentes tipos de input

Para cada tipo de input, decide qué nivel de sanitización necesitas:

1. Chat text libre del usuario
2. Código Python enviado por el usuario
3. Documento PDF parseado (texto extraído)
4. URL enviada por el usuario

Ejercicio 5: Logging de sanitización

Implementa el logging de cuando se sanitiza un input modificado:

import logging
import hashlib

logger = logging.getLogger("guardrails.sanitizer")

def sanitize_and_log(text: str, user_id: str = None) -> InputSanitizationResult:
    result = sanitize_input(text)
    
    if result.was_modified:
        # Loguear sin incluir el contenido original (privacy)
        input_hash = hashlib.sha256(text.encode()).hexdigest()[:8]
        logger.info(
            "input_sanitized",
            extra={
                "input_hash": input_hash,
                "original_length": result.original_length,
                "cleaned_length": len(result.text),
                "was_truncated": result.was_truncated,
                "had_control_chars": result.had_control_chars,
                "user_id": user_id
            }
        )
    
    return result

Resumen

• Sanitización = primera capa del pipeline: normalizar, limpiar, limitar longitud
• Pipeline de 5 pasos: fix encoding → normalizar unicode → eliminar control chars → colapsar espacios → truncar
• Límite por tokens es más preciso que por caracteres — usar tiktoken para apps de producción
• Límite práctico es de costo, no de capacidad técnica — definir según tu presupuesto por request
• Tests obligatorios: NULL bytes, texto vacío, texto muy largo, caracteres especiales válidos (ñ, emojis)
• Ser conservador: mejor no eliminar caracteres válidos que eliminar demasiado

Recursos adicionales

1. tiktoken — Tokenizer oficial de OpenAI para contar tokens exactos
2. OpenAI Tokenizer (web) — Visualizar tokens interactivamente
3. Unicode NFKC normalization — Por qué normalizar unicode
4. OWASP Input Validation Cheat Sheet — Guía completa de validación
5. Python unicodedata — Módulo estándar para unicode
6. Python re module — Para los patrones de control chars