5. Testing Parsers y Output Processors

Descripción

Los parsers y output processors son la lógica más fácil y valiosa de testear en una app AI: son 100% determinísticos, no necesitan mocks, y representan la mayor parte del código que puede fallar cuando el LLM produce variaciones. Esta cápsula cubre cómo testear funciones que transforman el raw output del LLM en datos estructurados. Es el "testing aburrido" que la mayoría ignora — y donde más ROI hay.

¿Por qué parsers son tan importantes?

El LLM puede producir output "correcto" semánticamente pero con variaciones de formato que rompen tu app:

# El LLM debe retornar JSON pero a veces produce:
'{"sentiment": "positivo"}'                           # Formato ideal
'```json\n{"sentiment": "positivo"}\n```'             # Con markdown
'El resultado es: {"sentiment": "positivo"}'           # Con texto previo
'{"sentiment": "positivo",}'                           # Trailing comma (JSON inválido)
'{"sentiment":"positivo"}'                             # Sin espacios
'{"Sentiment": "Positivo"}'                            # Capitalización diferente
'\n\n{"sentiment": "positivo"}\n\n'                    # Con whitespace extra

Sin un parser robusto, cualquiera de estos formatos puede romper tu app. Con tests de parser, puedes verificar que manejas todos correctamente.

Qué es un parser en contexto AI

Un parser toma el output crudo (string) del LLM y lo convierte en datos estructurados:

# Input: string crudo del LLM (puede tener markdown, texto extra, variaciones)
# Output: estructura de datos (dict, list, Pydantic model)

def parse_sentiment_response(raw: str) -> dict:
    """
    Extrae y parsea el JSON de la respuesta del LLM.
    
    Maneja múltiples formatos que el LLM puede producir:
    - JSON directo: '{"sentiment": "positivo"}'
    - JSON en markdown: '```json\\n{...}\\n```'
    - JSON con texto previo: 'El análisis: {...}'
    """
    if not raw or not raw.strip():
        raise ValueError("La respuesta del LLM está vacía")
    
    # Intentar parsear directamente primero
    try:
        return json.loads(raw.strip())
    except json.JSONDecodeError:
        pass
    
    # Buscar JSON en markdown code blocks
    markdown_pattern = r'```(?:json)?\s*\n?(.*?)\n?```'
    match = re.search(markdown_pattern, raw, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1).strip())
        except json.JSONDecodeError:
            pass
    
    # Buscar cualquier objeto JSON en el texto
    json_pattern = r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}'
    matches = re.findall(json_pattern, raw, re.DOTALL)
    for match in matches:
        try:
            return json.loads(match)
        except json.JSONDecodeError:
            continue
    
    raise ValueError(f"No se encontró JSON válido en la respuesta: {raw[:100]!r}")

Estructura de tests para parsers

La estructura óptima es parametrize con múltiples formatos de input:

import pytest
import json

# Casos de test para el parser
VALID_PARSER_CASES = [
    pytest.param(
        '{"sentiment": "positivo", "score": 0.9}',
        {"sentiment": "positivo", "score": 0.9},
        id="json_directo"
    ),
    pytest.param(
        '```json\n{"sentiment": "positivo", "score": 0.9}\n```',
        {"sentiment": "positivo", "score": 0.9},
        id="json_en_markdown"
    ),
    pytest.param(
        '```\n{"sentiment": "positivo", "score": 0.9}\n```',
        {"sentiment": "positivo", "score": 0.9},
        id="json_en_code_block_sin_lenguaje"
    ),
    pytest.param(
        'El análisis de sentimiento es: {"sentiment": "positivo", "score": 0.9}',
        {"sentiment": "positivo", "score": 0.9},
        id="json_con_texto_previo"
    ),
    pytest.param(
        '\n\n{"sentiment": "positivo", "score": 0.9}\n\n',
        {"sentiment": "positivo", "score": 0.9},
        id="json_con_whitespace"
    ),
    pytest.param(
        '{"sentiment": "positivo", "score": 0.9, "extra_field": "ignored"}',
        {"sentiment": "positivo", "score": 0.9, "extra_field": "ignored"},
        id="json_con_campos_extra"
    ),
    pytest.param(
        '{"sentiment": "positivo", "score": 0.9, "keywords": ["bien", "excelente"]}',
        {"sentiment": "positivo", "score": 0.9, "keywords": ["bien", "excelente"]},
        id="json_con_array"
    ),
]

@pytest.mark.parametrize("raw_input,expected", VALID_PARSER_CASES)
def test_parse_sentiment_valid_formats(raw_input, expected):
    """El parser maneja todos los formatos válidos que el LLM puede producir."""
    result = parse_sentiment_response(raw_input)
    assert result == expected

Tests de edge cases y errores

Los edge cases son donde más fallan los parsers en producción:

ERROR_CASES = [
    pytest.param("", ValueError, id="input_vacio"),
    pytest.param("   ", ValueError, id="input_solo_espacios"),
    pytest.param("Texto sin JSON", ValueError, id="sin_json"),
    pytest.param("No hay estructura", ValueError, id="texto_libre"),
]

@pytest.mark.parametrize("raw_input,expected_exception", ERROR_CASES)
def test_parse_sentiment_invalid_inputs(raw_input, expected_exception):
    """El parser lanza excepciones apropiadas para inputs inválidos."""
    with pytest.raises(expected_exception):
        parse_sentiment_response(raw_input)

def test_parse_malformed_json_raises():
    """JSON malformado lanza ValueError."""
    with pytest.raises((json.JSONDecodeError, ValueError)):
        parse_sentiment_response('{"sentiment": "positivo", "score": 0.9')  # Sin cerrar

def test_parse_truncated_response():
    """Respuesta truncada (LLM cortado por token limit) lanza error apropiado."""
    truncated = '{"sentiment": "positivo", "score": 0.9, "explanation": "El texto es muy'
    with pytest.raises((json.JSONDecodeError, ValueError)):
        parse_sentiment_response(truncated)

def test_parse_handles_unicode():
    """El parser maneja caracteres Unicode correctamente."""
    raw = '{"sentiment": "positivo", "keywords": ["excelente", "fantástico", "müde"]}'
    result = parse_sentiment_response(raw)
    assert "fantástico" in result["keywords"]
    assert "müde" in result["keywords"]

def test_parse_handles_nested_quotes():
    """El parser maneja comillas dentro del JSON."""
    raw = '{"sentiment": "positivo", "explanation": "El producto es \\"increíble\\""}'
    result = parse_sentiment_response(raw)
    assert result["explanation"] == 'El producto es "increíble"'

Output processors: post-procesamiento

Un output processor toma el dict del parser y lo normaliza, valida y estructura:

# app/processors.py

def process_sentiment_output(raw_dict: dict) -> dict:
    """
    Normaliza y valida el output parseado del LLM.
    
    Garantías de output:
    - sentiment: siempre uno de ["positivo", "negativo", "neutral"]
    - score: siempre float entre 0.0 y 1.0
    - keywords: siempre lista (puede estar vacía)
    - explanation: siempre string (puede estar vacío)
    """
    VALID_SENTIMENTS = {"positivo", "negativo", "neutral"}
    
    # Normalizar sentiment: case-insensitive, con fallback
    raw_sentiment = raw_dict.get("sentiment", "").strip().lower()
    sentiment = raw_sentiment if raw_sentiment in VALID_SENTIMENTS else "neutral"
    
    # Clamp score al rango [0, 1]
    try:
        score = float(raw_dict.get("score", 0.5))
        score = max(0.0, min(1.0, score))
    except (ValueError, TypeError):
        score = 0.5
    
    # Keywords: convertir a lista si es string, filtrar vacíos
    raw_keywords = raw_dict.get("keywords", [])
    if isinstance(raw_keywords, str):
        keywords = [k.strip() for k in raw_keywords.split(",") if k.strip()]
    elif isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    else:
        keywords = []
    
    # Explanation: string con strip y truncación
    explanation = str(raw_dict.get("explanation", "")).strip()[:500]
    
    return {
        "sentiment": sentiment,
        "score": score,
        "keywords": keywords,
        "explanation": explanation
    }

# Tests del output processor:

def test_process_normalizes_sentiment():
    """El processor normaliza sentiment a lowercase."""
    assert process_sentiment_output({"sentiment": "POSITIVO", "score": 0.9})["sentiment"] == "positivo"
    assert process_sentiment_output({"sentiment": "Negativo", "score": 0.1})["sentiment"] == "negativo"
    assert process_sentiment_output({"sentiment": "NEUTRAL", "score": 0.5})["sentiment"] == "neutral"

def test_process_invalid_sentiment_defaults_to_neutral():
    """Sentimientos no reconocidos se mapean a neutral."""
    assert process_sentiment_output({"sentiment": "muy_positivo", "score": 0.9})["sentiment"] == "neutral"
    assert process_sentiment_output({"sentiment": "", "score": 0.5})["sentiment"] == "neutral"
    assert process_sentiment_output({})["sentiment"] == "neutral"

def test_process_clamps_score():
    """Score fuera de rango se clampea a [0, 1]."""
    assert process_sentiment_output({"sentiment": "positivo", "score": 1.5})["score"] == 1.0
    assert process_sentiment_output({"sentiment": "negativo", "score": -0.1})["score"] == 0.0
    assert process_sentiment_output({"sentiment": "neutral", "score": 999})["score"] == 1.0

def test_process_handles_missing_score():
    """Score faltante usa valor default (0.5)."""
    result = process_sentiment_output({"sentiment": "positivo"})
    assert result["score"] == 0.5

def test_process_handles_invalid_score_type():
    """Score con tipo inválido usa valor default."""
    assert process_sentiment_output({"sentiment": "positivo", "score": "no-es-float"})["score"] == 0.5
    assert process_sentiment_output({"sentiment": "positivo", "score": None})["score"] == 0.5

def test_process_normalizes_keywords():
    """Keywords se normalizan a lista limpia."""
    # Lista normal
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9, "keywords": ["bien", "  excelente  "]})
    assert result["keywords"] == ["bien", "excelente"]  # Whitespace removido
    
    # String comma-separated (a veces el LLM hace esto)
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9, "keywords": "bien, excelente, increíble"})
    assert result["keywords"] == ["bien", "excelente", "increíble"]
    
    # Lista vacía
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9, "keywords": []})
    assert result["keywords"] == []
    
    # Faltante
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9})
    assert result["keywords"] == []

def test_process_truncates_long_explanation():
    """Explanation muy larga se trunca a 500 caracteres."""
    long_explanation = "texto " * 200  # 1200 caracteres
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9, "explanation": long_explanation})
    assert len(result["explanation"]) <= 500

Testear el pipeline completo: parser + processor

El pipeline completo es: raw string → parser → processor → output final. Puedes testear cada paso aislado Y el pipeline completo:

def process_llm_response(raw_string: str) -> dict:
    """Pipeline completo: parse + process."""
    parsed = parse_sentiment_response(raw_string)
    return process_sentiment_output(parsed)

# Test del pipeline completo (sin mock necesario):
@pytest.mark.parametrize("raw_input,expected_sentiment,expected_score_range", [
    ('{"sentiment": "POSITIVO", "score": 0.9}', "positivo", (0.8, 1.0)),
    ('```json\n{"sentiment": "negativo", "score": 0.2}\n```', "negativo", (0.0, 0.3)),
    ('{"sentiment": "invalid", "score": 5.0}', "neutral", (1.0, 1.0)),  # Score clampeado
    ('El análisis: {"sentiment": "neutral", "score": 0.5}', "neutral", (0.4, 0.6)),
])
def test_full_pipeline(raw_input, expected_sentiment, expected_score_range):
    """Pipeline completo: parser + processor para múltiples formatos."""
    result = process_llm_response(raw_input)
    
    assert result["sentiment"] == expected_sentiment
    assert expected_score_range[0] <= result["score"] <= expected_score_range[1]

Usar outputs reales del LLM como test cases

La forma más efectiva de mejorar tus tests de parser: capturar outputs reales del LLM y usarlos como casos:

# En desarrollo: loguea el raw output antes de parsear
import logging

logger = logging.getLogger(__name__)

def analyze_sentiment(text: str, client) -> dict:
    response = client.chat.completions.create(...)
    raw = response.choices[0].message.content
    
    # En desarrollo: loguea el raw output
    logger.debug(f"Raw LLM output: {raw!r}")
    
    return parse_and_process(raw)

# Después de capturar outputs reales, úsalos como test cases:

# tests/data/real_llm_outputs.json
[
    {
        "id": "real_001",
        "input": "Este producto es fantástico",
        "raw_output": "{\n    \"sentiment\": \"positivo\",\n    \"score\": 0.95,\n    \"explanation\": \"El texto usa adjetivos claramente positivos.\",\n    \"keywords\": [\n        \"fantástico\"\n    ]\n}",
        "expected_sentiment": "positivo"
    },
    {
        "id": "real_002",
        "input": "El servicio fue horrible",
        "raw_output": "```json\n{\"sentiment\": \"negativo\", \"score\": 0.05, \"explanation\": \"Adjetivo muy negativo.\", \"keywords\": [\"horrible\"]}\n```",
        "expected_sentiment": "negativo"
    }
]

# Test usando los outputs reales capturados:
import json
from pathlib import Path

def load_real_outputs():
    path = Path("tests/data/real_llm_outputs.json")
    with open(path) as f:
        return json.load(f)

@pytest.mark.parametrize("case", load_real_outputs(), ids=lambda c: c["id"])
def test_parser_handles_real_outputs(case):
    """El parser maneja correctamente los outputs reales capturados del LLM."""
    result = process_llm_response(case["raw_output"])
    assert result["sentiment"] == case["expected_sentiment"]

Coverage del parser: alcanzar 100%

Los parsers son excelentes candidatos para cobertura del 100%:

# Ejecutar con coverage para el parser:
pytest tests/test_parsers.py --cov=app.parsers --cov-report=term-missing -v

# Output esperado:
# Name              Stmts   Miss  Cover   Missing
# -----------------------------------------------
# app/parsers.py       45      2    96%   38-39

# Si hay líneas no cubiertas, añade el test que las ejecuta:
# Línea 38-39: el branch de "JSON anidado con subobjetos"

def test_parse_nested_json():
    """Cubre el branch de JSON con objetos anidados."""
    raw = '{"sentiment": "positivo", "metadata": {"source": "twitter", "length": 50}}'
    result = parse_sentiment_response(raw)
    assert result["metadata"]["source"] == "twitter"

Anti-patrones en testing de parsers

# ❌ Anti-patrón 1: Testear solo el happy path
def test_parser_bad():
    result = parse_sentiment_response('{"sentiment": "positivo"}')
    assert result["sentiment"] == "positivo"
    # Solo prueba el formato más simple — no el markdown, no los errores

# ✅ Correcto: probar múltiples formatos
@pytest.mark.parametrize("raw,expected", VALID_PARSER_CASES)
def test_parser_good(raw, expected):
    assert parse_sentiment_response(raw) == expected

# ❌ Anti-patrón 2: No testear errores
def test_processor_bad():
    result = process_sentiment_output({"sentiment": "positivo", "score": 0.9})
    assert result["sentiment"] == "positivo"
    # No prueba score fuera de rango, sentiment inválido, fields faltantes

# ✅ Correcto: testear todos los edge cases
def test_processor_good():
    assert process_sentiment_output({"sentiment": "invalid"})["sentiment"] == "neutral"
    assert process_sentiment_output({"score": 1.5})["score"] == 1.0
    assert process_sentiment_output({})["keywords"] == []

# ❌ Anti-patrón 3: Testear parser + LLM juntos
def test_full_integration_bad():
    # Llama al LLM real para testear el parser
    real_client = openai.OpenAI()
    result = analyze_sentiment("texto", client=real_client)
    assert "sentiment" in result
    # Mezcla responsabilidades: el test puede fallar por el LLM, no por el parser

# ✅ Correcto: testear el parser con inputs directos
def test_parser_isolated():
    raw_from_llm = '{"sentiment": "positivo", "score": 0.9}'
    result = parse_sentiment_response(raw_from_llm)
    assert result["sentiment"] == "positivo"
    # Testea solo el parser — sin dependencia del LLM

Comparación: Parser vs Chain completa

Conclusión: Testear parser y processor aislados es más rápido, más barato, y cubre más casos que testear la chain completa.

Ejercicios

Ejercicio 1: Parser de entidades con regex

El LLM retorna texto con este formato: "Entidades: Juan García (PERSONA), 15 de enero (FECHA), Madrid (LUGAR)". Escribe el parser y 5 tests.

import re
from typing import Optional

def parse_entities_response(raw: str) -> list[dict]:
    """
    Parsea respuestas del LLM que listan entidades.
    
    Formato esperado: "Nombre (TIPO), Nombre2 (TIPO2)"
    Retorna: [{"entity": "Nombre", "type": "TIPO"}, ...]
    """
    if not raw or not raw.strip():
        return []
    
    # Remover prefijos como "Entidades: ", "Las entidades son: ", etc.
    text = re.sub(r'^[^:]+:\s*', '', raw.strip())
    
    # Patrón: texto (TIPO)
    pattern = r'([^,(]+?)\s*\(([^)]+)\)'
    matches = re.findall(pattern, text)
    
    return [
        {"entity": entity.strip(), "type": entity_type.strip().upper()}
        for entity, entity_type in matches
        if entity.strip() and entity_type.strip()
    ]

# Tests:
@pytest.mark.parametrize("raw,expected", [
    (
        "Juan García (PERSONA), 15 de enero (FECHA)",
        [{"entity": "Juan García", "type": "PERSONA"}, {"entity": "15 de enero", "type": "FECHA"}]
    ),
    (
        "Entidades: Madrid (LUGAR)",
        [{"entity": "Madrid", "type": "LUGAR"}]
    ),
    (
        "",
        []
    ),
    (
        "Texto sin entidades",
        []
    ),
    (
        "persona (tipo1), otra entidad (tipo2), tercera (tipo3)",
        [
            {"entity": "persona", "type": "TIPO1"},
            {"entity": "otra entidad", "type": "TIPO2"},
            {"entity": "tercera", "type": "TIPO3"}
        ]
    ),
])
def test_parse_entities(raw, expected):
    assert parse_entities_response(raw) == expected

Ejercicio 2: Processor con truncado graceful

Escribe el output processor para análisis de tickets de soporte. Debe manejar:

• priority debe ser uno de ["high", "medium", "low"], default "medium"
• title máximo 100 caracteres, truncado con "..." si es más largo
• tags lista de strings, máximo 5 elementos, vacía por default

def process_ticket_analysis(raw: dict) -> dict:
    VALID_PRIORITIES = {"high", "medium", "low"}
    
    # Priority con fallback
    priority = str(raw.get("priority", "")).strip().lower()
    priority = priority if priority in VALID_PRIORITIES else "medium"
    
    # Title con truncación graceful
    title = str(raw.get("title", "")).strip()
    if len(title) > 100:
        title = title[:97] + "..."
    
    # Tags: lista de strings, máximo 5
    raw_tags = raw.get("tags", [])
    if isinstance(raw_tags, list):
        tags = [str(t).strip() for t in raw_tags if t and str(t).strip()][:5]
    else:
        tags = []
    
    return {"priority": priority, "title": title, "tags": tags}

# Tests:
def test_process_ticket_valid():
    result = process_ticket_analysis({"priority": "high", "title": "Server down", "tags": ["prod", "critical"]})
    assert result == {"priority": "high", "title": "Server down", "tags": ["prod", "critical"]}

def test_process_ticket_invalid_priority():
    assert process_ticket_analysis({"priority": "urgent"})["priority"] == "medium"

def test_process_ticket_long_title():
    long_title = "A" * 150
    result = process_ticket_analysis({"title": long_title})
    assert len(result["title"]) == 100
    assert result["title"].endswith("...")

def test_process_ticket_too_many_tags():
    result = process_ticket_analysis({"tags": ["t1", "t2", "t3", "t4", "t5", "t6"]})
    assert len(result["tags"]) == 5

def test_process_ticket_empty():
    result = process_ticket_analysis({})
    assert result == {"priority": "medium", "title": "", "tags": []}

Ejercicio 3: Test con outputs reales capturados

Imagina que capturaste estos 3 outputs reales del LLM. Escribe los tests parametrizados:

REAL_OUTPUTS = [
    {
        "raw": "{\n  \"sentiment\": \"positivo\",\n  \"score\": 0.92,\n  \"explanation\": \"Usa adjetivos positivos.\"\n}",
        "expected_sentiment": "positivo",
        "expected_score_range": (0.8, 1.0)
    },
    {
        "raw": "```json\n{\"sentiment\": \"NEGATIVO\", \"score\": 0.08}\n```",
        "expected_sentiment": "negativo",
        "expected_score_range": (0.0, 0.2)
    },
    {
        "raw": "El análisis: {\"sentiment\": \"neutral\", \"score\": 0.5, \"extra\": \"ignored\"}",
        "expected_sentiment": "neutral",
        "expected_score_range": (0.4, 0.6)
    }
]

@pytest.mark.parametrize("case", REAL_OUTPUTS, ids=[f"real_{i}" for i in range(len(REAL_OUTPUTS))])
def test_parser_with_real_captured_outputs(case):
    """
    Tests parametrizados con outputs reales del LLM.
    Estos casos fueron capturados en desarrollo para asegurar
    que el parser maneja los formatos reales correctamente.
    """
    result = process_llm_response(case["raw"])
    
    assert result["sentiment"] == case["expected_sentiment"]
    
    min_score, max_score = case["expected_score_range"]
    assert min_score <= result["score"] <= max_score, \
        f"Score {result['score']} fuera de rango [{min_score}, {max_score}]"

Ejercicio 4: Coverage

Ejecuta coverage en el parser y explica cómo llegarías al 100%:

pytest tests/test_parsers.py --cov=app.parsers --cov-report=term-missing -v

Ejercicio 5: Testear el pipeline completo

Escribe un test que prueba el pipeline completo parse + process para un output de clasificación de tickets:

El raw output puede venir en cualquiera de estos formatos:

1. '{"priority": "HIGH", "title": "Server down", "tags": ["prod"]}'
2. '```json\n{"priority": "medium", "title": "Login issue"}\n```'

@pytest.mark.parametrize("raw,expected_priority,expected_title_contains", [
    (
        '{"priority": "HIGH", "title": "Server down", "tags": ["prod"]}',
        "high",
        "Server down"
    ),
    (
        '```json\n{"priority": "medium", "title": "Login issue"}\n```',
        "medium",
        "Login issue"
    ),
])
def test_ticket_pipeline_complete(raw, expected_priority, expected_title_contains):
    """Pipeline completo: parse JSON de ticket + process."""
    # Paso 1: Parser
    parsed = parse_ticket_response(raw)
    assert isinstance(parsed, dict)
    
    # Paso 2: Processor
    result = process_ticket_analysis(parsed)
    
    # Assert: pipeline completo
    assert result["priority"] == expected_priority
    assert expected_title_contains in result["title"]
    assert isinstance(result["tags"], list)

Resumen

• Parsers y processors son 100% determinísticos — testéalos sin mocks, sin costo de API
• parametrize para múltiples formatos — el LLM produce variaciones reales de formato
• Edge cases críticos: vacío, malformado, truncado, con markdown, con texto previo
• Output processors: normalización, clamping de rangos, defaults para campos faltantes
• Coverage al 100% es alcanzable para parsers — es el componente ideal para esto
• Captura outputs reales del LLM y úsalos como test cases para coverage realista

Recursos adicionales

1. pytest.mark.parametrize — Para múltiples casos de input
2. pytest.raises — Para testear excepciones esperadas
3. Python json module — Reference del módulo JSON estándar
4. Python re module — Regex para parsing de texto
5. Pydantic v2 — Validators — Para validación estructurada
6. pytest-cov — Para medir cobertura de los parsers
7. Hypothesis — Property-based testing para parsers (se ve en Módulo 3)