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
- Entrada: Configuración RAG como
dicto archivo JSON/YAML con campos por área. - Evaluación: Cada criterio retorna
pass,failowarningcon mensaje explicativo. - Score global: Promedio ponderado por criticidad de cada área.
- Salida: Reporte legible en consola y opcionalmente en Markdown/JSON.
Áreas a validar
| Área | Criterios clave |
|---|---|
| Scaling | Estrategia definida, límites documentados, plan de autoscaling |
| Monitoring | Dashboards (p95, error rate, throughput), alertas activas, runbooks |
| Backups | Backups automáticos, RTO/RPO definidos, restore probado recientemente |
| Security | Auth, rate limiting, validación de entradas |
| Cost optimization | Cache de embeddings, métricas de costo por query, batch ops |
| Migration readiness | Dual-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:
- Script ejecutable
production_readiness_checklist.pycon modos demo, archivo y validación. - Config de ejemplo en JSON o YAML que represente tu sistema real o un escenario típico.
- Reporte generado con score, detalle por área y top 5 brechas.
- 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:
| Área | Estado actual | Score | Brecha principal | Prioridad | Due date |
|---|---|---|---|---|---|
| Escalado | Verde | 83% | Falta autoscaling plan | Media | Sem 1 |
| Observabilidad | Verde | 89% | Alertas sin test | Baja | Sem 2 |
| DR | Amarillo | 75% | Restore no probado | Alta | Sem 1 |
| Seguridad | Verde | 100% | — | — | — |
| Costo | Amarillo | 50% | Falta embedding cache | Media | Sem 2 |
| Migración | Rojo | 0% | Sin plan canary | Alta | Sem 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:
- ¿Cuál es hoy tu mayor riesgo operativo según el reporte?
- ¿Qué acción reduce más riesgo en menos tiempo?
- ¿Qué métrica confirma que mejoraste después de cerrar brechas?
- ¿Cómo priorizaste entre áreas de igual impacto?
- ¿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, sicriterion.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
resultspor 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_CONFIGcubre las 6 áreas con campos por criterio. -
CRITERIA_BY_AREAtiene criterios con peso y descripción. -
AREA_WEIGHTSdefine 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
--validatefuncionan.
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_AREAyeval_criterion. - Este proyecto cierra el Módulo 7 y conecta con el Módulo 8 (proyecto integrador RAG).
Recursos adicionales
- Google SRE Workbook — Incident Response
- The Four Golden Signals (Google SRE)
- Martin Fowler — Canary Release
- 12-Factor App — Config
- Prometheus — Alerting Rules
- ChromaDB — Deployment and Persistence
- OWASP — API Security Top 10
- Backblaze — Backup Best Practices
Tiempo estimado: 45-60 minutos
Siguiente módulo: ../../module-08-proyecto-final/es/01-introduccion-modulo.md