7. Production Checklist y Deployment

Descripción

El checklist completo que debes completar antes de deployar un sistema LLM a producción: evaluation, tests, cost estimation, rollback plan, monitoring, security. Estrategias de deployment (canary, blue-green, feature flags). Proceso post-deploy.

Por Qué Necesitas un Checklist

En el entusiasmo de tener algo funcionando, es fácil omitir pasos críticos:

Deployes sin checklist:
✗ "Parecía funcionar en mis tests"
✗ Sin estimación de costo — sorpresa al final del mes
✗ Sin rollback — incidente de 4 horas para revertir
✗ Sin monitoring — enterado del problema 3 días después
✗ API key en el código → leak en GitHub

Deployes con checklist:
✓ Evaluation pasada con métricas documentadas
✓ Costo estimado: $X/mes a Y requests/día
✓ Rollback documentado y probado: 2 comandos
✓ Alertas configuradas — 5 minutos para detectar problema
✓ Secrets en variables de entorno

El Checklist Pre-Deploy Completo

Sección 1: Evaluación y Calidad

## Checklist de Evaluación

### Golden Set
- [ ] Golden set existe con mínimo 100 ejemplos
- [ ] Distribución correcta: happy path (60%) + edge cases (30%) + adversarial (10%)
- [ ] Revisión humana del 20% del golden set completada
- [ ] Golden set versionado en Git o dataset storage

### Métricas de Calidad
- [ ] Accuracy >= umbral definido (ej. 85%)
- [ ] Faithfulness >= 0.80 (para sistemas de resumen/RAG)
- [ ] Format compliance >= 0.95 (para structured output)
- [ ] LLM-as-judge ejecutado en sample de 50+ ejemplos

### Tests de Regresión
- [ ] Baseline guardado en baseline.json
- [ ] Regression tests pasan (sin métricas por debajo del baseline - tolerancia)
- [ ] Test suite ejecutado en el código actual (no en cache)
- [ ] Tests de edge cases ejecutados
- [ ] Tests adversariales ejecutados

Sección 2: Performance y Costo

## Checklist de Performance

### Latencia
- [ ] Latencia p95 medida: ___ ms (¿cumple SLA?)
- [ ] Latencia p99 medida: ___ ms
- [ ] Timeout configurado (ej. 30s max)
- [ ] Retry logic implementado (3 intentos con exponential backoff)

### Costo
- [ ] Tokens promedio por request medidos: ___ tokens
- [ ] Costo estimado por request: $___
- [ ] Costo estimado mensual a proyección de tráfico: $___ /mes
- [ ] Costo mensual dentro del budget aprobado: YES/NO
- [ ] Budget alerts configuradas (80% y 100% del budget)
- [ ] max_tokens configurado (no dejar sin límite)
- [ ] Caching configurado si aplica (hit rate esperado: ___%)

### Scaling
- [ ] Rate limits de la API considerados
- [ ] Rate limiting del propio servicio configurado
- [ ] Queue o async si hay picos de tráfico esperados

Sección 3: Reliability

## Checklist de Reliability

### Error Handling
- [ ] Try/except en todas las llamadas a la API
- [ ] Retry con exponential backoff implementado
- [ ] Fallback definido para cuando la API falla
- [ ] Timeout configurado
- [ ] Parsing de output con manejo de errores (no asumir formato correcto)
- [ ] Output validation implementado

### Availability
- [ ] Dependency de la API documentada (qué pasa si OpenAI cae?)
- [ ] Fallback a modelo alternativo si aplica
- [ ] Circuit breaker si aplica (detener llamadas si error rate > X%)

Sección 4: Versioning y Rollback

## Checklist de Versioning

### Versionado
- [ ] Prompt registrado con versión semántica (vX.Y.Z)
- [ ] Changelog actualizado con descripción del cambio
- [ ] Versión anterior documentada (para rollback)
- [ ] Registry actualizado con nueva versión como "activa"

### Rollback Plan
- [ ] Versión estable anterior identificada: ___
- [ ] Comando/proceso de rollback documentado
- [ ] Rollback probado en staging (no solo documentado)
- [ ] Tiempo estimado de rollback: ___ minutos
- [ ] Responsable de ejecutar rollback: ___

Sección 5: Monitoring y Alertas

## Checklist de Monitoring

### Métricas
- [ ] Latencia (p50, p95) monitoreada
- [ ] Error rate monitoreado
- [ ] Costo por request monitoreado
- [ ] Tokens por request monitoreado
- [ ] Quality score monitoreado (sampling)

### Alertas
- [ ] Alerta de latencia alta configurada (p95 > SLA)
- [ ] Alerta de error rate alta configurada (> 1-5%)
- [ ] Alerta de costo alto configurada (> budget)
- [ ] Alerta de quality degradation configurada
- [ ] Destinatarios de alertas configurados (Slack/email)

### Dashboards
- [ ] Dashboard básico de métricas operacionales disponible
- [ ] Acceso al dashboard compartido con el equipo

Sección 6: Seguridad

## Checklist de Seguridad

### Secrets
- [ ] API key en variables de entorno (NO en código ni logs)
- [ ] .env en .gitignore
- [ ] Secrets rotados si fueron expuestos alguna vez
- [ ] Acceso a API key restringido (no compartido)

### Input/Output
- [ ] Input validation implementado (limitar longitud, caracteres)
- [ ] Sanitización de inputs antes de insertar en prompts
- [ ] Output filtering si hay riesgo de PII o contenido sensible
- [ ] Prompt injection mitigado (instrucciones de sistema vs usuario separadas)

### Logging
- [ ] PII no logueado (nombres, emails, datos sensibles)
- [ ] Logs tienen retención apropiada
- [ ] Audit trail para cambios de prompts

Implementación del Checklist en Código

from dataclasses import dataclass, field
from typing import Optional
import json


@dataclass
class ChecklistItem:
    """Un item del checklist."""
    id: str
    seccion: str
    descripcion: str
    requerido: bool = True
    completado: bool = False
    valor: Optional[str] = None
    notas: str = ""


class ProductionChecklist:
    """
    Checklist ejecutable para deploys de sistemas LLM.
    
    Genera un documento de evidencia del proceso de deploy.
    """
    
    def __init__(self, prompt_name: str, version: str, responsable: str):
        self.prompt_name = prompt_name
        self.version = version
        self.responsable = responsable
        self.fecha_inicio = None
        self.fecha_completado = None
        self.items: list[ChecklistItem] = self._init_items()
    
    def _init_items(self) -> list[ChecklistItem]:
        """Inicializa todos los items del checklist."""
        return [
            # EVALUACIÓN
            ChecklistItem("eval_1", "Evaluación", "Golden set con 100+ ejemplos existe"),
            ChecklistItem("eval_2", "Evaluación", "Accuracy >= umbral definido", requerido=True),
            ChecklistItem("eval_3", "Evaluación", "Regression tests pasan (sin regresión)"),
            ChecklistItem("eval_4", "Evaluación", "Format compliance >= 0.95 (si structured output)"),
            
            # PERFORMANCE
            ChecklistItem("perf_1", "Performance", "Latencia p95 medida y dentro del SLA"),
            ChecklistItem("perf_2", "Performance", "Costo estimado mensual calculado"),
            ChecklistItem("perf_3", "Performance", "max_tokens configurado"),
            ChecklistItem("perf_4", "Performance", "Retry logic con exponential backoff implementado"),
            
            # RELIABILITY
            ChecklistItem("rel_1", "Reliability", "Error handling en todas las llamadas API"),
            ChecklistItem("rel_2", "Reliability", "Timeout configurado"),
            ChecklistItem("rel_3", "Reliability", "Output validation implementado"),
            ChecklistItem("rel_4", "Reliability", "Fallback definido", requerido=False),
            
            # VERSIONING
            ChecklistItem("ver_1", "Versioning", "Prompt registrado con semver"),
            ChecklistItem("ver_2", "Versioning", "Changelog actualizado"),
            ChecklistItem("ver_3", "Versioning", "Rollback plan documentado"),
            
            # MONITORING
            ChecklistItem("mon_1", "Monitoring", "Latencia y error rate monitoreados"),
            ChecklistItem("mon_2", "Monitoring", "Alertas de costo configuradas"),
            ChecklistItem("mon_3", "Monitoring", "Alertas de quality configuradas"),
            
            # SEGURIDAD
            ChecklistItem("sec_1", "Seguridad", "API keys en variables de entorno (no en código)"),
            ChecklistItem("sec_2", "Seguridad", "Input validation implementado"),
            ChecklistItem("sec_3", "Seguridad", "PII no logueado"),
        ]
    
    def completar(self, item_id: str, valor: str = "", notas: str = "") -> None:
        """Marca un item como completado."""
        for item in self.items:
            if item.id == item_id:
                item.completado = True
                item.valor = valor
                item.notas = notas
                print(f"✅ [{item_id}] {item.descripcion}")
                return
        raise KeyError(f"Item '{item_id}' no encontrado")
    
    def completar_seccion(self, seccion: str, valores: dict = None) -> None:
        """Marca todos los items de una sección como completados."""
        for item in self.items:
            if item.seccion == seccion:
                valor = (valores or {}).get(item.id, "")
                item.completado = True
                item.valor = valor
    
    def puede_deployar(self) -> tuple[bool, list[str]]:
        """
        Verifica si se puede deployar.
        
        Returns: (puede_deployar, lista_de_bloqueadores)
        """
        bloqueadores = [
            item.descripcion
            for item in self.items
            if item.requerido and not item.completado
        ]
        
        return len(bloqueadores) == 0, bloqueadores
    
    def resumen(self) -> dict:
        """Resumen del estado del checklist."""
        requeridos_total = sum(1 for i in self.items if i.requerido)
        requeridos_completados = sum(1 for i in self.items if i.requerido and i.completado)
        opcionales_completados = sum(1 for i in self.items if not i.requerido and i.completado)
        
        return {
            "prompt_name": self.prompt_name,
            "version": self.version,
            "responsable": self.responsable,
            "requeridos": f"{requeridos_completados}/{requeridos_total}",
            "opcionales": f"{opcionales_completados}/{len(self.items) - requeridos_total}",
            "puede_deployar": requeridos_completados == requeridos_total,
            "items_pendientes": [i.descripcion for i in self.items if not i.completado]
        }
    
    def generar_reporte(self) -> str:
        """Genera un reporte del checklist en markdown."""
        puede, bloqueadores = self.puede_deployar()
        
        from datetime import datetime
        lineas = [
            f"# Production Deploy Checklist",
            f"**Prompt:** {self.prompt_name} {self.version}  ",
            f"**Responsable:** {self.responsable}  ",
            f"**Fecha:** {datetime.now().strftime('%Y-%m-%d %H:%M')}  ",
            f"**Status:** {'✅ APROBADO para deploy' if puede else '❌ BLOQUEADO'}  ",
            "",
        ]
        
        if bloqueadores:
            lineas.extend([
                "## ❌ Bloqueadores (completar antes de deploy)",
                "",
            ])
            for b in bloqueadores:
                lineas.append(f"- {b}")
            lineas.append("")
        
        # Secciones
        secciones = {}
        for item in self.items:
            if item.seccion not in secciones:
                secciones[item.seccion] = []
            secciones[item.seccion].append(item)
        
        for seccion, items in secciones.items():
            completados = sum(1 for i in items if i.completado)
            total = len(items)
            estado_seccion = "✅" if completados == total else "⚠️" if completados > 0 else "❌"
            
            lineas.append(f"## {estado_seccion} {seccion} ({completados}/{total})")
            lineas.append("")
            
            for item in items:
                estado = "✅" if item.completado else ("⚡" if not item.requerido else "❌")
                valor_str = f" — {item.valor}" if item.valor else ""
                notas_str = f" *(Nota: {item.notas})*" if item.notas else ""
                lineas.append(f"- {estado} {item.descripcion}{valor_str}{notas_str}")
            
            lineas.append("")
        
        return "\n".join(lineas)
    
    def guardar(self, path: str = None) -> str:
        """Guarda el reporte a disco."""
        from pathlib import Path
        from datetime import datetime
        
        if not path:
            ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            path = f"checklists/{self.prompt_name}_{self.version}_{ts}.md"
        
        Path(path).parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w") as f:
            f.write(self.generar_reporte())
        
        return path

Estrategias de Deployment

Comparación de Estrategias

Estrategia Recomendada: Feature Flag + Canary

from enum import Enum
import hashlib

class DeployStrategy(Enum):
    BIG_BANG = "big_bang"
    CANARY = "canary"
    FEATURE_FLAG = "feature_flag"


class GradualDeployer:
    """
    Implementa deployment gradual con feature flags y canary releases.
    """
    
    def __init__(self, registry, alertas_manager):
        self.registry = registry
        self.alertas = alertas_manager
        self._canary_configs: dict = {}
    
    def iniciar_canary(
        self,
        prompt_name: str,
        nueva_version: str,
        porcentaje_inicial: float = 0.05,
        metricas_objetivo: dict = None
    ) -> dict:
        """
        Inicia un canary deployment.
        
        El tráfico se divide: porcentaje_inicial → nueva versión
                              resto → versión actual (estable)
        """
        version_estable = self.registry.version_activa(prompt_name)
        
        self._canary_configs[prompt_name] = {
            "version_nueva": nueva_version,
            "version_estable": version_estable,
            "porcentaje": porcentaje_inicial,
            "metricas_objetivo": metricas_objetivo or {"error_rate": 0.01, "latencia_p95": 3000},
            "inicio": time.time(),
            "estado": "ACTIVO"
        }
        
        print(f"🚀 Canary iniciado: {prompt_name}")
        print(f"   Nueva versión: {nueva_version} ({porcentaje_inicial:.0%} tráfico)")
        print(f"   Versión estable: {version_estable} ({1-porcentaje_inicial:.0%} tráfico)")
        
        return self._canary_configs[prompt_name]
    
    def obtener_version(self, prompt_name: str, request_id: str) -> str:
        """
        Determina qué versión usar para este request.
        
        Usa hashing del request_id para consistencia
        (mismo request siempre va al mismo bucket).
        """
        if prompt_name not in self._canary_configs:
            return self.registry.version_activa(prompt_name)
        
        config = self._canary_configs[prompt_name]
        
        if config["estado"] != "ACTIVO":
            return self.registry.version_activa(prompt_name)
        
        # Hash determinista para consistencia
        hash_val = int(hashlib.md5(f"{prompt_name}:{request_id}".encode()).hexdigest(), 16)
        use_nueva = (hash_val % 100) < (config["porcentaje"] * 100)
        
        return config["version_nueva"] if use_nueva else config["version_estable"]
    
    def promover_canary(
        self,
        prompt_name: str,
        nuevo_porcentaje: float
    ) -> None:
        """Aumenta el porcentaje de tráfico al canary."""
        if prompt_name not in self._canary_configs:
            raise ValueError(f"No hay canary activo para '{prompt_name}'")
        
        config = self._canary_configs[prompt_name]
        anterior = config["porcentaje"]
        config["porcentaje"] = nuevo_porcentaje
        
        print(f"📈 Canary promovido: {prompt_name} {anterior:.0%} → {nuevo_porcentaje:.0%}")
    
    def completar_canary(self, prompt_name: str) -> None:
        """Finaliza el canary: 100% tráfico a nueva versión."""
        if prompt_name not in self._canary_configs:
            raise ValueError(f"No hay canary activo para '{prompt_name}'")
        
        version_nueva = self._canary_configs[prompt_name]["version_nueva"]
        
        # Activar nueva versión en el registry
        self.registry.activar(prompt_name, version_nueva)
        
        # Limpiar config de canary
        del self._canary_configs[prompt_name]
        
        print(f"✅ Canary completado: {prompt_name} ahora es {version_nueva}")
    
    def cancelar_canary(self, prompt_name: str, razon: str = "manual") -> None:
        """
        Cancela el canary (rollback automático).
        100% del tráfico vuelve a la versión estable.
        """
        if prompt_name not in self._canary_configs:
            return
        
        config = self._canary_configs[prompt_name]
        version_estable = config["version_estable"]
        
        # Asegurar que la versión estable está activa
        self.registry.activar(prompt_name, version_estable)
        config["estado"] = "CANCELADO"
        
        print(f"🔄 Canary cancelado: {prompt_name} → {version_estable} (razón: {razon})")
        del self._canary_configs[prompt_name]


# Ejemplo de rollout progresivo:
"""
Día 1, Hora 0:  iniciar_canary(5%)   — monitorear
Día 1, Hora 4:  promover_canary(10%)  — si métricas OK
Día 2:          promover_canary(25%)  — si métricas OK
Día 3:          promover_canary(50%)  — si métricas OK
Día 4:          promover_canary(100%) — si métricas OK
Día 5:          completar_canary()    — confirmar en registry
"""

Proceso de Deploy Paso a Paso

import asyncio
from openai import OpenAI

client = OpenAI()

async def proceso_deploy_completo(
    prompt_name: str,
    nueva_version: str,
    nuevo_prompt: str,
    golden_set: list[dict],
    responsable: str,
    usar_canary: bool = True
) -> dict:
    """
    Proceso completo de deploy con checklist, evaluación y deployment gradual.
    
    Returns: dict con resultado del proceso.
    """
    print(f"\n{'='*60}")
    print(f"🚀 INICIANDO PROCESO DE DEPLOY")
    print(f"   {prompt_name} → {nueva_version}")
    print(f"   Responsable: {responsable}")
    print(f"{'='*60}\n")
    
    # 1. INICIALIZAR CHECKLIST
    checklist = ProductionChecklist(prompt_name, nueva_version, responsable)
    
    # 2. EVALUACIÓN (automática)
    print("📊 Paso 1: Evaluación de calidad...")
    
    correctos = 0
    for ej in golden_set:
        r = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": nuevo_prompt.format(input=ej["input"])}],
            temperature=0
        )
        if r.choices[0].message.content.strip().lower() == str(ej["expected_output"]).lower():
            correctos += 1
    
    accuracy = correctos / len(golden_set)
    print(f"   Accuracy: {accuracy:.2%}")
    
    if accuracy >= 0.85:
        checklist.completar("eval_1", f"100 ejemplos en golden set")
        checklist.completar("eval_2", f"Accuracy: {accuracy:.2%}")
        checklist.completar("eval_3", "Tests regression: PASS")
    else:
        print(f"❌ BLOQUEADO: Accuracy {accuracy:.2%} < 85%")
        return {"exito": False, "razon": "accuracy_insuficiente", "accuracy": accuracy}
    
    # 3. VERIFICAR PUEDE DEPLOYAR
    puede, bloqueadores = checklist.puede_deployar()
    
    if not puede:
        print(f"❌ BLOQUEADO por {len(bloqueadores)} items:")
        for b in bloqueadores:
            print(f"   - {b}")
        
        reporte = checklist.guardar()
        return {"exito": False, "razon": "checklist_incompleto", "bloqueadores": bloqueadores}
    
    # 4. DEPLOY
    print("\n🚢 Paso 2: Deploy...")
    
    if usar_canary:
        # Registrar en el registry
        # registry.registrar(prompt_name, nueva_version, nuevo_prompt)
        
        # Iniciar canary al 5%
        # deployer.iniciar_canary(prompt_name, nueva_version, 0.05)
        
        print(f"   Canary iniciado al 5% → monitorear por 24h")
        print(f"   Comando para promover: deployer.promover_canary('{prompt_name}', 0.25)")
        print(f"   Comando para cancelar: deployer.cancelar_canary('{prompt_name}')")
    else:
        # Big bang deploy
        # registry.activar(prompt_name, nueva_version)
        print(f"   Deploy directo: {prompt_name} → {nueva_version}")
    
    # 5. POST-DEPLOY
    print("\n📋 Post-deploy tasks:")
    print("   [ ] Monitorear métricas primeras 2 horas")
    print("   [ ] Verificar alertas están activas")
    print("   [ ] Documentar en changelog")
    
    # 6. GUARDAR REPORTE
    reporte_path = checklist.guardar()
    print(f"\n📄 Reporte guardado: {reporte_path}")
    
    return {
        "exito": True,
        "accuracy": accuracy,
        "estrategia": "canary" if usar_canary else "big_bang",
        "reporte": reporte_path
    }

Rollback de Emergencia

def rollback_emergencia(
    prompt_name: str,
    registry,
    alertas_manager,
    razon: str = "manual"
) -> None:
    """
    Ejecuta rollback de emergencia.
    
    Diseñado para ser rápido y simple — en una emergencia no hay tiempo para complejidad.
    """
    print(f"\n🚨 EJECUTANDO ROLLBACK DE EMERGENCIA")
    print(f"   Prompt: {prompt_name}")
    print(f"   Razón: {razon}")
    print(f"   Iniciado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # Paso 1: Obtener versión actual
    version_actual = registry.version_activa(prompt_name)
    print(f"   Versión actual: {version_actual}")
    
    # Paso 2: Rollback
    try:
        registry.rollback(prompt_name)
        version_nueva = registry.version_activa(prompt_name)
        print(f"\n✅ Rollback exitoso: {version_actual} → {version_nueva}")
        print(f"   Tiempo: {(time.time() - time.time()):.1f}s")  # Casi instantáneo
    except Exception as e:
        print(f"❌ Error en rollback: {e}")
        print("   ACCIÓN MANUAL REQUERIDA")
        return
    
    # Paso 3: Verificar que está funcionando
    print("\n🔍 Verificando health post-rollback...")
    
    test_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "test"}],
        temperature=0
    )
    
    if test_response:
        print("   ✅ Sistema respondiendo correctamente")
    
    # Paso 4: Notificar
    print("\n📢 Notificaciones enviadas al equipo")
    
    # Post-rollback checklist
    print("\n📋 Post-rollback tasks:")
    print("   [ ] Investigar causa del rollback")
    print("   [ ] Documentar incidente")
    print("   [ ] Analizar logs del período afectado")
    print("   [ ] Corregir el problema antes del próximo intento de deploy")

Post-Deploy: Las Primeras 24 Horas

class PostDeployMonitor:
    """
    Monitoreo intensivo durante las primeras 24 horas post-deploy.
    
    Más frecuente que el monitoring normal para detectar problemas temprano.
    """
    
    def __init__(
        self,
        metrics_collector,
        alertas_manager,
        baseline_metricas: dict,
        intervalo_verificacion: int = 300  # 5 minutos
    ):
        self.metrics = metrics_collector
        self.alertas = alertas_manager
        self.baseline = baseline_metricas
        self.intervalo = intervalo_verificacion
        self._inicio = time.time()
        self._horas_monitoreo = 24
    
    def verificar(self, prompt_name: str) -> dict:
        """Verificación post-deploy."""
        metricas_actuales = self.metrics.metricas_actuales(prompt_name)
        
        issues = []
        
        # Comparar con baseline
        for metrica in ["error_rate", "latencia_p95"]:
            valor_actual = metricas_actuales.get(metrica, 0)
            valor_baseline = self.baseline.get(metrica, 0)
            
            if valor_baseline == 0:
                continue
            
            degradacion = (valor_actual - valor_baseline) / valor_baseline
            
            if degradacion > 0.20:  # 20% de degradación
                issues.append({
                    "metrica": metrica,
                    "baseline": valor_baseline,
                    "actual": valor_actual,
                    "degradacion": f"{degradacion:.0%}"
                })
        
        horas_transcurridas = (time.time() - self._inicio) / 3600
        
        return {
            "horas_post_deploy": horas_transcurridas,
            "status": "PROBLEMAS_DETECTADOS" if issues else "OK",
            "issues": issues,
            "metricas": metricas_actuales,
            "recomendacion": (
                "Considerar rollback si los problemas persisten"
                if issues else
                "Sistema estable"
            )
        }
    
    def reporte_24h(self, prompt_name: str) -> str:
        """Reporte final de las primeras 24 horas."""
        estado = self.verificar(prompt_name)
        
        return f"""
## Reporte Post-Deploy 24h: {prompt_name}

**Status:** {estado['status']}

### Métricas Finales vs Baseline
| Métrica | Baseline | Actual | Delta |
|---------|---------|--------|-------|
| Error rate | {self.baseline.get('error_rate', 0):.2%} | {estado['metricas'].get('error_rate', 0):.2%} | ... |
| Latencia p95 | {self.baseline.get('latencia_p95', 0):.0f}ms | {estado['metricas'].get('latencia_p95', 0):.0f}ms | ... |

### Conclusión
{'✅ Deploy exitoso — promover a 100%' if estado['status'] == 'OK' else '⚠️ Problemas detectados — revisar antes de promover'}
"""

Troubleshooting

Problema 1: Deploy rompe producción inesperadamente

Síntoma: Poco después del deploy, error rate sube al 20%.

Acción inmediata:

# 1. ROLLBACK INMEDIATO (no debuggear en producción)
registry.rollback("clasificador")  # 1 línea

# 2. VERIFICAR que rollback fue exitoso
print(f"Versión activa ahora: {registry.version_activa('clasificador')}")

# 3. INVESTIGAR en staging (no en producción)
# Revisar logs del período de problema
# Identificar qué cambio causó el problema

Regla de oro: Si algo falla en producción, haz rollback primero. Debuggea después.

Problema 2: Checklist se omite "porque hay prisa"

Síntoma: Deploy rápido que resulta en incidente.

Solución estructural:

# Automatizar el checklist para que sea más rápido que omitirlo
def deploy_gate(prompt_name: str, version: str) -> bool:
    """
    Gate automático que bloquea el deploy si el checklist no está completo.
    Se ejecuta en CI/CD.
    """
    # Verificar métricas del último test
    with db_conn() as conn:
        test = conn.execute(
            "SELECT accuracy FROM test_results WHERE prompt_nombre = ? AND prompt_version = ?",
            (prompt_name, version)
        ).fetchone()
    
    if not test:
        print(f"❌ DEPLOY BLOQUEADO: No hay tests para {prompt_name} {version}")
        return False
    
    if test["accuracy"] < 0.85:
        print(f"❌ DEPLOY BLOQUEADO: Accuracy {test['accuracy']:.2%} < 85%")
        return False
    
    print(f"✅ Deploy gate pasado para {prompt_name} {version}")
    return True

# En CI/CD pipeline:
# if not deploy_gate(PROMPT_NAME, VERSION):
#     sys.exit(1)  # Bloquea el pipeline

Ejercicios

Ejercicio 1: Completar el checklist para un deploy real

Toma un prompt que tengas en producción o en desarrollo y completa el checklist:

# Inicializar checklist
checklist = ProductionChecklist(
    prompt_name="mi_clasificador",
    version="v1.2.0",
    responsable="tu_nombre"
)

# Completar items que ya tienes
checklist.completar("eval_1", "150 ejemplos en golden_set.json")
checklist.completar("eval_2", "Accuracy: 91.3%")
checklist.completar("sec_1", "API key en .env, .env en .gitignore")

# Ver qué falta
resumen = checklist.resumen()
print(f"Items pendientes: {resumen['items_pendientes']}")

# Generar reporte
reporte = checklist.generar_reporte()
print(reporte)

Ejercicio 2: Implementar un canary deployment simple

Implementa un router que envíe el 10% del tráfico a una nueva versión basándose en el user_id:

import hashlib
from openai import OpenAI

client = OpenAI()

PROMPT_V1 = "Clasifica como POSITIVO, NEGATIVO o NEUTRO: {input}. Solo la categoría."
PROMPT_V2 = """Clasifica el sentimiento. Considera sarcasmo e ironía.
Responde SOLO con: POSITIVO, NEGATIVO, o NEUTRO

Texto: {input}
Categoría:"""

def canary_router(user_id: str, pct_nueva: float = 0.10) -> tuple[str, str]:
    """Retorna (prompt, version) para un user_id dado."""
    hash_val = int(hashlib.md5(f"clasificador:{user_id}".encode()).hexdigest(), 16)
    use_new = (hash_val % 100) < (pct_nueva * 100)
    
    if use_new:
        return PROMPT_V2, "v2"
    return PROMPT_V1, "v1"

# Simular routing
distribucion = {"v1": 0, "v2": 0}
for user_num in range(200):
    user_id = f"user_{user_num:04d}"
    _, version = canary_router(user_id, pct_nueva=0.10)
    distribucion[version] += 1

print(f"v1: {distribucion['v1']} usuarios ({distribucion['v1']/200:.0%})")
print(f"v2: {distribucion['v2']} usuarios ({distribucion['v2']/200:.0%})")
# v1: ~180 usuarios (90%), v2: ~20 usuarios (10%) ✓

Resumen

• Checklist: 6 secciones — Evaluación, Performance, Reliability, Versioning, Monitoring, Seguridad
• Gate de deploy: Automatizar el checklist como parte del pipeline de CI/CD
• Big bang: Simple pero riesgoso — solo para hotfixes urgentes
• Canary: 5% → 25% → 50% → 100% — con ventanas de monitoring en cada etapa
• Feature flags: Flexibles, permiten cancelar sin rollback de código
• Rollback: Debe ser automático y < 5 minutos — si es más complejo, algo está mal
• Post-deploy: Monitoreo intensivo las primeras 24 horas — detectar problemas antes de que escalen

Recursos adicionales

1. Google SRE Book — Reliability engineering best practices
2. Feature Flags Guide — LaunchDarkly feature flags
3. Martin Fowler: Canary Release — Pattern de canary
4. OpenAI Production Best Practices — Guía oficial
5. Deployment Checklist Template — Template de referencia