Módulo 7: Production Considerations para RAG

Cápsula 08: Proyecto - Production Readiness Checklist

Descripción de la cápsula

Construirás una herramienta Python ejecutable que valida la configuración de un sistema RAG contra criterios de producción. La herramienta recibe una configuración (dict o archivo JSON/YAML), evalúa cada área (escalado, monitoreo, backups, seguridad, optimización de costos, preparación para migración), y produce un reporte de readiness con pass/fail por criterio y score global. Es una herramienta práctica que puedes usar para auditar tu sistema real, priorizar brechas y planificar mejoras.

Tiempo estimado: 45-60 minutos


Objetivo del proyecto

  • Implementar un validador que evalúe configuración RAG contra 6 áreas de producción.
  • Producir reporte con estado pass/fail por criterio y score global (0-100).
  • Priorizar brechas por impacto y generar plan de cierre recomendado.
  • Código completo, ejecutable y reutilizable en tu proyecto.

Especificaciones del proyecto

Requisitos funcionales

  1. Entrada: Configuración RAG como dict o archivo JSON/YAML con campos por área.
  2. Evaluación: Cada criterio retorna pass, fail o warning con mensaje explicativo.
  3. Score global: Promedio ponderado por criticidad de cada área.
  4. Salida: Reporte legible en consola y opcionalmente en Markdown/JSON.

Áreas a validar

ÁreaCriterios clave
ScalingEstrategia definida, límites documentados, plan de autoscaling
MonitoringDashboards (p95, error rate, throughput), alertas activas, runbooks
BackupsBackups automáticos, RTO/RPO definidos, restore probado recientemente
SecurityAuth, rate limiting, validación de entradas
Cost optimizationCache de embeddings, métricas de costo por query, batch ops
Migration readinessDual-write ready, canary plan, criterios de rollback

Success criteria

  • ✅ Las 6 áreas evaluadas con criterios concretos y verificables.
  • ✅ Pass/fail/warning por criterio con mensajes accionables.
  • ✅ Score global 0-100 con desglose por área.
  • ✅ Top N brechas priorizadas por impacto.
  • ✅ Modo batch para validar múltiples configs (--validate).

Contexto antes de empezar

Este proyecto sintetiza las cápsulas 02-07 del módulo:

  • 02: Estrategias de escalado (vertical, horizontal, sharding).
  • 03: Monitoring (Prometheus, Grafana, alertas, runbooks).
  • 04: Backup y disaster recovery (RTO, RPO, restore).
  • 05: Seguridad (auth, rate limiting, validación).
  • 06: Optimización de costos (cache, batch, métricas).
  • 07: Migración sin downtime (dual-write, canary, rollback).

La herramienta no ejecuta tu sistema RAG: valida la configuración declarativa que describes. Es útil para auditorías pre-release, revisiones periódicas y onboarding de equipos.


Implementación paso a paso

Paso 1: Definir el esquema de configuración

La herramienta espera un diccionario con la siguiente estructura. Cada campo es opcional; los ausentes se consideran "no implementado".

# production_readiness_checklist.py
"""
Production Readiness Checklist para sistemas RAG.
Evalúa configuración contra criterios de producción.
"""

from dataclasses import dataclass
from typing import Literal
import json
from pathlib import Path

# ----- Esquema de configuración RAG -----

DEFAULT_CONFIG = {
    "scaling": {
        "strategy_defined": False,
        "strategy_type": None,  # "vertical" | "horizontal" | "sharding" | None
        "resource_limits_documented": False,
        "autoscaling_plan": False,
    },
    "monitoring": {
        "dashboards_p95": False,
        "dashboards_error_rate": False,
        "dashboards_throughput": False,
        "alerts_active": False,
        "alerts_tested": False,
        "runbooks_available": False,
    },
    "backups": {
        "automatic_backups": False,
        "rto_hours": None,
        "rpo_hours": None,
        "restore_tested_recently": False,
    },
    "security": {
        "auth_enabled": False,
        "rate_limiting_enabled": False,
        "input_validation": False,
    },
    "cost_optimization": {
        "embedding_cache": False,
        "result_cache": False,
        "cost_per_query_metric": False,
        "batch_operations": False,
    },
    "migration_readiness": {
        "dual_write_ready": False,
        "canary_plan": False,
        "rollback_criteria": False,
    },
}

¿Por qué este diseño? La configuración es declarativa: describes qué tienes hoy, no cómo lo implementaste. Así puedes validar configs reales, simuladas o de múltiples entornos.


Paso 2: Definir criterios y pesos por área

Cada área tiene criterios con peso. El score del área es el porcentaje de criterios que pasan, ponderado.

@dataclass
class Criterion:
    key: str
    label: str
    weight: int  # 1-5, mayor = más crítico
    description: str = ""

CRITERIA_BY_AREA = {
    "scaling": [
        Criterion("strategy_defined", "Estrategia de escalado definida", 5, "vertical/horizontal/sharding documentado"),
        Criterion("resource_limits_documented", "Límites de recursos documentados", 3, "CPU, RAM, conexiones definidos"),
        Criterion("autoscaling_plan", "Plan de autoscaling", 4, "Reglas de escala up/down definidas"),
    ],
    "monitoring": [
        Criterion("dashboards_p95", "Dashboard con p95 latencia", 5, "p50/p95/p99 de retrieval visible"),
        Criterion("dashboards_error_rate", "Dashboard con error rate", 5, "Error rate por endpoint"),
        Criterion("dashboards_throughput", "Dashboard con throughput", 3, "Queries/seg, ingestion docs/seg"),
        Criterion("alerts_active", "Alertas activas", 5, "Prometheus/Grafana/Cloud alertando"),
        Criterion("alerts_tested", "Alertas probadas", 3, "Run de prueba de alertas ejecutado"),
        Criterion("runbooks_available", "Runbooks disponibles", 4, "Procedimiento por alerta"),
    ],
    "backups": [
        Criterion("automatic_backups", "Backups automáticos en ejecución", 5, "Cron o job programado"),
        Criterion("rto_hours", "RTO definido (horas)", 4, "Recovery Time Objective documentado"),
        Criterion("rpo_hours", "RPO definido (horas)", 4, "Recovery Point Objective documentado"),
        Criterion("restore_tested_recently", "Restore probado recientemente", 5, "Últimos 30 días"),
    ],
    "security": [
        Criterion("auth_enabled", "Autenticación activa", 5, "API key, JWT u OAuth"),
        Criterion("rate_limiting_enabled", "Rate limiting activo", 4, "Por cliente/tenant"),
        Criterion("input_validation", "Validación de entradas", 5, "Anti prompt-injection, sanitización"),
    ],
    "cost_optimization": [
        Criterion("embedding_cache", "Cache de embeddings", 3, "No re-embedear docs sin cambios"),
        Criterion("result_cache", "Cache de resultados", 4, "Redis o similar para queries repetidas"),
        Criterion("cost_per_query_metric", "Métrica costo por query", 4, "Costo estimado por consulta"),
        Criterion("batch_operations", "Operaciones batch en ingestion", 3, "Batch upsert, no one-by-one"),
    ],
    "migration_readiness": [
        Criterion("dual_write_ready", "Dual-write preparado", 4, "Código listo para escribir a 2 DBs"),
        Criterion("canary_plan", "Plan canary documentado", 4, "Porcentaje tráfico, duración, métricas"),
        Criterion("rollback_criteria", "Criterios de rollback", 5, "Cuándo abortar migración"),
    ],
}

# Peso de cada área en el score global (1-5)
AREA_WEIGHTS = {
    "scaling": 3,
    "monitoring": 5,
    "backups": 5,
    "security": 5,
    "cost_optimization": 2,
    "migration_readiness": 2,
}

Paso 3: Motor de evaluación

Evalúa cada criterio contra la configuración. Los criterios tienen validaciones específicas (p.ej. RTO <= 24 para producción customer-facing).

Status = Literal["pass", "fail", "warning"]

@dataclass
class CriterionResult:
    criterion: Criterion
    status: Status
    message: str

@dataclass
class AreaResult:
    area: str
    score: float
    passed: int
    total: int
    results: list[CriterionResult]

def eval_criterion(area: str, criterion: Criterion, config: dict) -> CriterionResult:
    """Evalúa un criterio contra la configuración."""
    area_config = config.get(area, {})
    value = area_config.get(criterion.key)

    if value is None:
        return CriterionResult(criterion, "fail", f"No configurado: {criterion.label}")
    if isinstance(value, bool):
        if value:
            return CriterionResult(criterion, "pass", f"✓ {criterion.label}")
        return CriterionResult(criterion, "fail", f"✗ {criterion.label} desactivado")

    # Criterios con valores numéricos
    if criterion.key == "rto_hours":
        if value is None or value <= 0:
            return CriterionResult(criterion, "fail", "RTO no definido")
        if value <= 2:
            return CriterionResult(criterion, "pass", f"RTO ≤ 2h ({value}h) - producción alta")
        if value <= 8:
            return CriterionResult(criterion, "warning", f"RTO {value}h - aceptable para media criticidad")
        return CriterionResult(criterion, "fail", f"RTO {value}h - alto para producción")

    if criterion.key == "rpo_hours":
        if value is None or value <= 0:
            return CriterionResult(criterion, "fail", "RPO no definido")
        if value <= 6:
            return CriterionResult(criterion, "pass", f"RPO ≤ 6h ({value}h)")
        if value <= 24:
            return CriterionResult(criterion, "warning", f"RPO {value}h - más pérdida de datos aceptable")
        return CriterionResult(criterion, "fail", f"RPO {value}h - riesgo alto")

    return CriterionResult(criterion, "fail", f"Valor inesperado: {value}")


def eval_area(area: str, config: dict) -> AreaResult:
    """Evalúa un área completa."""
    criteria = CRITERIA_BY_AREA.get(area, [])
    results = [eval_criterion(area, c, config) for c in criteria]
    total_weight = sum(c.weight for c in criteria)
    passed_weight = sum(
        c.weight for r in results
        for c in [r.criterion]
        if r.status == "pass"
    )
    score = (passed_weight / total_weight * 100) if total_weight else 0.0
    passed_count = sum(1 for r in results if r.status == "pass")
    return AreaResult(area, round(score, 1), passed_count, len(results), results)


def eval_all(config: dict) -> dict:
    """Evalúa todas las áreas y calcula score global."""
    area_results = {area: eval_area(area, config) for area in CRITERIA_BY_AREA}
    total_weight = sum(AREA_WEIGHTS.get(a, 1) for a in area_results)
    global_score = sum(
        AREA_WEIGHTS.get(area, 1) * r.score
        for area, r in area_results.items()
    ) / total_weight if total_weight else 0.0
    return {
        "global_score": round(global_score, 1),
        "areas": area_results,
    }

Paso 4: Priorización de brechas

Ordena las brechas (criterios fail o warning) por impacto: peso del criterio × peso del área.

@dataclass
class Gap:
    area: str
    criterion: str
    label: str
    status: Status
    message: str
    impact_score: float


def get_prioritized_gaps(eval_result: dict, top_n: int = 5) -> list[Gap]:
    """Extrae brechas y las ordena por impacto."""
    gaps = []
    for area, area_result in eval_result["areas"].items():
        area_weight = AREA_WEIGHTS.get(area, 1)
        for r in area_result.results:
            if r.status in ("fail", "warning"):
                criterion_weight = next(
                    (c.weight for c in CRITERIA_BY_AREA[area] if c.key == r.criterion.key),
                    1,
                )
                impact = area_weight * criterion_weight
                gaps.append(Gap(
                    area=area,
                    criterion=r.criterion.key,
                    label=r.criterion.label,
                    status=r.status,
                    message=r.message,
                    impact_score=impact,
                ))
    gaps.sort(key=lambda g: g.impact_score, reverse=True)
    return gaps[:top_n]

Paso 5: Generador de reporte

Produce un reporte legible en consola.

def generate_report(eval_result: dict, config_name: str = "RAG Config") -> str:
    """Genera reporte de readiness."""
    lines = []
    lines.append("=" * 70)
    lines.append(f"  PRODUCTION READINESS REPORT — {config_name}")
    lines.append("=" * 70)

    # Score global
    score = eval_result["global_score"]
    status_emoji = "✅" if score >= 80 else ("⚠️" if score >= 60 else "❌")
    lines.append(f"\n  Score global: {score:.1f}/100  {status_emoji}")
    lines.append("")

    # Por área
    lines.append("  ESTADO POR ÁREA")
    lines.append("  " + "-" * 50)
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        emoji = "🟢" if r.score >= 80 else ("🟡" if r.score >= 50 else "🔴")
        lines.append(f"  {emoji} {area_label}: {r.score:.1f}% ({r.passed}/{r.total})")
    lines.append("")

    # Detalle por criterio
    lines.append("  DETALLE POR CRITERIO")
    lines.append("  " + "-" * 50)
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"\n  [{area_label}]")
        for res in r.results:
            sym = "✓" if res.status == "pass" else ("!" if res.status == "warning" else "✗")
            lines.append(f"    {sym} {res.message}")

    # Top brechas
    gaps = get_prioritized_gaps(eval_result, top_n=5)
    if gaps:
        lines.append("\n  TOP 5 BRECHAS PRIORIZADAS")
        lines.append("  " + "-" * 50)
        for i, g in enumerate(gaps, 1):
            lines.append(f"  {i}. [{g.area}] {g.label}{g.message}")

    # Plan sugerido
    lines.append("\n  PLAN SUGERIDO (2 SEMANAS)")
    lines.append("  " + "-" * 50)
    critical_areas = [
        (a, r) for a, r in eval_result["areas"].items()
        if r.score < 60
    ]
    critical_areas.sort(key=lambda x: x[1].score)
    if critical_areas:
        week1 = [a for a, _ in critical_areas[:2]]
        week2 = [a for a, _ in critical_areas[2:4]]
        lines.append(f"  Semana 1: {', '.join(week1)}")
        lines.append(f"  Semana 2: {', '.join(week2) if week2 else 'Refinamiento y documentación'}")
    else:
        lines.append("  Semana 1–2: Refinar áreas en amarillo y documentar runbooks")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)

Paso 6: Carga de configuración desde archivo

Soporta JSON y YAML.

def load_config(path: str) -> dict:
    """Carga configuración desde JSON o YAML."""
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"No existe: {path}")

    raw = p.read_text(encoding="utf-8")
    suffix = p.suffix.lower()

    if suffix == ".json":
        config = json.loads(raw)
    elif suffix in (".yaml", ".yml"):
        try:
            import yaml
            config = yaml.safe_load(raw)
        except ImportError:
            raise ImportError("PyYAML requerido para archivos YAML: pip install pyyaml")
    else:
        raise ValueError("Formato no soportado. Usa .json o .yaml")

    # Merge con defaults para campos faltantes
    return merge_config(DEFAULT_CONFIG, config)


def merge_config(base: dict, override: dict) -> dict:
    """Merge recursivo de configs. Override gana."""
    result = base.copy()
    for k, v in override.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = merge_config(result[k], v)
        else:
            result[k] = v
    return result

Paso 7: Script principal

Dos modos: validar una config o ejecutar escenarios de prueba.

import sys

def main():
    print("=" * 70)
    print("  PRODUCTION READINESS CHECKLIST — Sistemas RAG")
    print("  Módulo 7 — Production Considerations")
    print("=" * 70)

    if len(sys.argv) > 1 and sys.argv[1] == "--validate":
        run_validation_mode()
    elif len(sys.argv) > 1:
        run_config_file_mode(sys.argv[1])
    else:
        run_demo_mode()


def run_demo_mode():
    """Usa configuración de ejemplo y genera reporte."""
    print("\n  Modo demo: config de ejemplo\n")
    config = {
        "scaling": {"strategy_defined": True, "strategy_type": "horizontal", "resource_limits_documented": True, "autoscaling_plan": False},
        "monitoring": {"dashboards_p95": True, "dashboards_error_rate": True, "dashboards_throughput": True, "alerts_active": True, "alerts_tested": False, "runbooks_available": True},
        "backups": {"automatic_backups": True, "rto_hours": 2, "rpo_hours": 6, "restore_tested_recently": False},
        "security": {"auth_enabled": True, "rate_limiting_enabled": True, "input_validation": True},
        "cost_optimization": {"embedding_cache": False, "result_cache": True, "cost_per_query_metric": False, "batch_operations": True},
        "migration_readiness": {"dual_write_ready": False, "canary_plan": False, "rollback_criteria": False},
    }
    result = eval_all(config)
    print(generate_report(result, "Demo RAG System"))


def run_config_file_mode(path: str):
    """Valida configuración desde archivo."""
    print(f"\n  Cargando config desde: {path}\n")
    config = load_config(path)
    result = eval_all(config)
    print(generate_report(result, Path(path).stem))


def run_validation_mode():
    """Ejecuta escenarios de validación predefinidos."""
    print("\n  Modo validación: 3 escenarios\n")
    scenarios = [
        {"name": "MVP / PoC", "config": {"scaling": {"strategy_defined": False}, "monitoring": {"alerts_active": False}, "backups": {"automatic_backups": False}, "security": {"auth_enabled": False}, "cost_optimization": {}, "migration_readiness": {}}},
        {"name": "Producción temprana", "config": {"scaling": {"strategy_defined": True, "strategy_type": "horizontal"}, "monitoring": {"dashboards_p95": True, "alerts_active": True}, "backups": {"automatic_backups": True, "rto_hours": 4, "rpo_hours": 12, "restore_tested_recently": False}, "security": {"auth_enabled": True, "rate_limiting_enabled": True}, "cost_optimization": {"result_cache": True}, "migration_readiness": {"rollback_criteria": True}}},
        {"name": "Producción madura", "config": {"scaling": {"strategy_defined": True, "resource_limits_documented": True, "autoscaling_plan": True}, "monitoring": {"dashboards_p95": True, "dashboards_error_rate": True, "dashboards_throughput": True, "alerts_active": True, "alerts_tested": True, "runbooks_available": True}, "backups": {"automatic_backups": True, "rto_hours": 1, "rpo_hours": 4, "restore_tested_recently": True}, "security": {"auth_enabled": True, "rate_limiting_enabled": True, "input_validation": True}, "cost_optimization": {"embedding_cache": True, "result_cache": True, "cost_per_query_metric": True, "batch_operations": True}, "migration_readiness": {"dual_write_ready": True, "canary_plan": True, "rollback_criteria": True}}},
    ]
    for s in scenarios:
        result = eval_all(merge_config(DEFAULT_CONFIG, s["config"]))
        print(f"  Escenario: {s['name']}")
        print(f"  Score: {result['global_score']:.1f}/100")
        print()
    print("  ✓ Validación completada\n")


if __name__ == "__main__":
    main()

Archivo de configuración de ejemplo

Guarda esto como rag_config_example.json para probar:

{
  "scaling": {
    "strategy_defined": true,
    "strategy_type": "horizontal",
    "resource_limits_documented": true,
    "autoscaling_plan": true
  },
  "monitoring": {
    "dashboards_p95": true,
    "dashboards_error_rate": true,
    "dashboards_throughput": true,
    "alerts_active": true,
    "alerts_tested": true,
    "runbooks_available": true
  },
  "backups": {
    "automatic_backups": true,
    "rto_hours": 2,
    "rpo_hours": 6,
    "restore_tested_recently": true
  },
  "security": {
    "auth_enabled": true,
    "rate_limiting_enabled": true,
    "input_validation": true
  },
  "cost_optimization": {
    "embedding_cache": true,
    "result_cache": true,
    "cost_per_query_metric": true,
    "batch_operations": true
  },
  "migration_readiness": {
    "dual_write_ready": true,
    "canary_plan": true,
    "rollback_criteria": true
  }
}

Output esperado

Ejecutando python production_readiness_checklist.py:

======================================================================
  PRODUCTION READINESS REPORT — Demo RAG System
======================================================================

  Score global: 72.3/100  ⚠️

  ESTADO POR ÁREA
  --------------------------------------------------
  🟢 Scaling: 83.3% (2/3)
  🟢 Monitoring: 88.9% (8/9)
  🟡 Backups: 75.0% (3/4)
  🟢 Security: 100.0% (3/3)
  🟡 Cost Optimization: 50.0% (2/4)
  🔴 Migration Readiness: 0.0% (0/3)

  DETALLE POR CRITERIO
  --------------------------------------------------
  [Scaling]
    ✓ Estrategia de escalado definida
    ✓ Límites de recursos documentados
    ✗ Plan de autoscaling desactivado
  ...

  TOP 5 BRECHAS PRIORIZADAS
  --------------------------------------------------
  1. [migration_readiness] Criterios de rollback — No configurado
  2. [migration_readiness] Dual-write preparado — No configurado
  3. [backups] Restore probado recientemente — ✗ desactivado
  4. [cost_optimization] Cache de embeddings — ✗ desactivado
  5. [cost_optimization] Métrica costo por query — ✗ desactivado

  PLAN SUGERIDO (2 SEMANAS)
  --------------------------------------------------
  Semana 1: migration_readiness, cost_optimization
  Semana 2: Refinamiento y documentación

======================================================================

Entregables del proyecto

Al terminar debes tener:

  1. Script ejecutable production_readiness_checklist.py con modos demo, archivo y validación.
  2. Config de ejemplo en JSON o YAML que represente tu sistema real o un escenario típico.
  3. Reporte generado con score, detalle por área y top 5 brechas.
  4. Plan de 2 semanas para cerrar brechas críticas, basado en la salida del tool.

Formato recomendado de entrega

Incluye una tabla como esta en tu documentación:

ÁreaEstado actualScoreBrecha principalPrioridadDue date
EscaladoVerde83%Falta autoscaling planMediaSem 1
ObservabilidadVerde89%Alertas sin testBajaSem 2
DRAmarillo75%Restore no probadoAltaSem 1
SeguridadVerde100%
CostoAmarillo50%Falta embedding cacheMediaSem 2
MigraciónRojo0%Sin plan canaryAltaSem 1

Criterio de aprobación sugerido

Define mínimo para pasar a "readiness aceptable":

  • 0 brechas críticas sin owner asignado.
  • Alertas básicas activas (p95, error rate).
  • Backup + restore probado al menos una vez.
  • Plan de incidentes (runbook) disponible para alertas críticas.
  • Score global ≥ 60 para staging, ≥ 80 para producción customer-facing.

Preguntas de defensa técnica

Al presentar el checklist, debes poder responder:

  1. ¿Cuál es hoy tu mayor riesgo operativo según el reporte?
  2. ¿Qué acción reduce más riesgo en menos tiempo?
  3. ¿Qué métrica confirma que mejoraste después de cerrar brechas?
  4. ¿Cómo priorizaste entre áreas de igual impacto?
  5. ¿Qué criterio quitarías o añadirías para tu contexto?

Troubleshooting del proyecto

"Todo está en rojo/amarillo y no sabemos por dónde empezar"

Prioriza por impacto a usuario + probabilidad de fallo. Usa el Top 5 brechas del reporte: el orden ya está calculado por impact_score. Empieza por las 2 primeras. Si son de áreas distintas (p.ej. backups y security), reparte: una persona en backups, otra en security.


"Checklist muy grande, equipo saturado"

Divide en fases de 2 semanas con objetivos concretos. Semana 1: solo backups + security (bloqueantes). Semana 2: monitoring + un área más. No intentes cerrar todo a la vez.


"No hay acuerdo sobre prioridad entre equipos"

Usa criterio común: riesgo de caída (¿qué pasa si falla?), impacto en SLA (¿afecta a usuarios?), costo (¿cuánto cuesta no hacerlo?). El tool ya ordena por impacto. Si el desacuerdo persiste, documenta las dos prioridades y el trade-off explícito.


"La config no refleja la realidad de nuestro sistema"

El tool valida configuración declarativa, no el comportamiento en runtime. Si tu config dice auth_enabled: true pero hay un endpoint sin auth, el tool no lo detecta. Complementa con: (1) tests de integración que verifiquen auth en todos los endpoints, (2) auditorías manuales periódicas. La config debe actualizarse cuando cambies el sistema.


"Queremos criterios más estrictos o diferentes"

Los criterios están en CRITERIA_BY_AREA y los pesos en AREA_WEIGHTS. Modifica el código para añadir criterios, cambiar pesos o ajustar la lógica en eval_criterion. Si añades un criterio con validación numérica (como RTO), extiende eval_criterion con un nuevo if criterion.key == "tu_nuevo_criterio".


Ejercicios post-proyecto

Ejercicio 1: Exportar reporte a Markdown

Extiende la herramienta con una función que genere un archivo Markdown con el reporte, incluyendo tablas para el detalle por criterio y el top de brechas.

Pistas:

  • Crea export_to_markdown(eval_result, output_path: str).
  • Usa tablas Markdown (| col1 | col2 |).
  • Incluye fecha de generación y nombre del config.
Ver solución
from datetime import date

def export_to_markdown(eval_result: dict, output_path: str, config_name: str = "RAG Config"):
    """Exporta el reporte a un archivo Markdown."""
    lines = []
    lines.append(f"# Production Readiness Report — {config_name}")
    lines.append(f"\n**Fecha:** {date.today().isoformat()}\n")
    lines.append(f"**Score global:** {eval_result['global_score']:.1f}/100\n")
    lines.append("## Estado por área\n")
    lines.append("| Área | Score | Passed | Total |")
    lines.append("|------|------:|-------:|-----:|")
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"| {area_label} | {r.score:.1f}% | {r.passed} | {r.total} |")
    lines.append("\n## Detalle por criterio\n")
    for area, r in eval_result["areas"].items():
        area_label = area.replace("_", " ").title()
        lines.append(f"### {area_label}\n")
        lines.append("| Criterio | Estado |")
        lines.append("|----------|--------|")
        for res in r.results:
            status = "✓" if res.status == "pass" else ("!" if res.status == "warning" else "✗")
            lines.append(f"| {res.criterion.label} | {status} {res.message} |")
    lines.append("\n## Top 5 brechas\n")
    gaps = get_prioritized_gaps(eval_result, 5)
    for i, g in enumerate(gaps, 1):
        lines.append(f"{i}. **[{g.area}]** {g.label}: {g.message}")
    Path(output_path).write_text("\n".join(lines), encoding="utf-8")
    print(f"Reporte exportado a: {output_path}")

# Uso en main():
# export_to_markdown(result, "readiness_report.md", "Mi RAG")

Ejercicio 2: Añadir criterio de "incidentes documentados"

Agrega un criterio a la área monitoring: que exista un registro de incidentes de los últimos 90 días (aunque sea un campo booleano incidents_logged).

Pistas:

  • Añade el criterio a CRITERIA_BY_AREA["monitoring"].
  • Añade el campo al DEFAULT_CONFIG["monitoring"].
  • En eval_criterion, si criterion.key == "incidents_logged", valida el booleano.
Ver solución
# 1. En CRITERIA_BY_AREA["monitoring"], añadir:
Criterion("incidents_logged", "Incidentes documentados (90 días)", 3, "Registro de incidentes y post-mortems"),

# 2. En DEFAULT_CONFIG["monitoring"], añadir:
"incidents_logged": False,

# 3. En eval_criterion, el caso bool ya cubre este campo.
# Si value es True -> pass, si False -> fail.

Ejercicio 3: Modo comparación entre dos configs

Crea una función que reciba dos paths de config y muestre qué criterios pasan en una y fallan en la otra, útil para comparar staging vs producción.

Pistas:

  • compare_configs(path_a: str, path_b: str) -> str
  • Carga ambas configs, evalúa cada una, compara results por criterio.
  • Genera tabla: Criterio | Config A | Config B | Mejoró?
Ver solución
def compare_configs(path_a: str, path_b: str) -> str:
    """Compara dos configs y muestra diferencias."""
    config_a = load_config(path_a)
    config_b = load_config(path_b)
    eval_a = eval_all(config_a)
    eval_b = eval_all(config_b)
    lines = []
    lines.append("=" * 60)
    lines.append("  COMPARACIÓN DE READINESS")
    lines.append(f"  A: {path_a} ({eval_a['global_score']:.1f})")
    lines.append(f"  B: {path_b} ({eval_b['global_score']:.1f})")
    lines.append("=" * 60)
    lines.append("\n| Área | Criterio | A | B | Cambio |")
    lines.append("|------|----------|---|---|--------|")
    for area in eval_a["areas"]:
        ra = eval_a["areas"][area].results
        rb = eval_b["areas"][area].results
        for i, res_a in enumerate(ra):
            res_b = rb[i]
            a_sym = "✓" if res_a.status == "pass" else "✗"
            b_sym = "✓" if res_b.status == "pass" else "✗"
            change = "→" if a_sym == b_sym else ("↑" if b_sym == "✓" else "↓")
            lines.append(f"| {area} | {res_a.criterion.label[:25]} | {a_sym} | {b_sym} | {change} |")
    return "\n".join(lines)

Checklist de completitud

Estructura del código:

  • DEFAULT_CONFIG cubre las 6 áreas con campos por criterio.
  • CRITERIA_BY_AREA tiene criterios con peso y descripción.
  • AREA_WEIGHTS define peso de cada área en el score global.

Motor de evaluación:

  • eval_criterion() maneja bool, RTO, RPO y valores inesperados.
  • eval_area() calcula score ponderado por peso de criterio.
  • eval_all() retorna score global y resultados por área.

Priorización y reporte:

  • get_prioritized_gaps() ordena brechas por impacto.
  • generate_report() produce salida legible con plan sugerido.

Entrada y modos:

  • load_config() soporta JSON y YAML.
  • Modo demo, modo archivo y modo --validate funcionan.

Validación:

  • Al menos 3 escenarios (MVP, producción temprana, madura) en --validate.
  • Los scores reflejan el nivel de madurez esperado.

Output:

  • Score global 0-100 visible.
  • Estado por área (verde/amarillo/rojo).
  • Top 5 brechas y plan de 2 semanas.

Resumen

  • Construiste una herramienta Python ejecutable que valida configuración RAG contra 6 áreas de producción.
  • Cada área tiene criterios concretos con pass/fail/warning y mensajes accionables.
  • El score global es un promedio ponderado; las áreas de monitoring, backups y security pesan más.
  • El top 5 de brechas está ordenado por impacto (área × criterio) para priorizar trabajo.
  • El plan sugerido asigna semanas según áreas críticas (score < 60).
  • La configuración es declarativa (dict/JSON/YAML): describes qué tienes, no cómo.
  • Puedes extender criterios modificando CRITERIA_BY_AREA y eval_criterion.
  • Este proyecto cierra el Módulo 7 y conecta con el Módulo 8 (proyecto integrador RAG).

Recursos adicionales


Tiempo estimado: 45-60 minutos
Siguiente módulo: ../../module-08-proyecto-final/es/01-introduccion-modulo.md