Módulo 8: Prompt Engineering en Producción
8. Proyecto Final: Production Prompt System
Descripción
El proyecto final de la guía: construyes un sistema LLM production-ready que integra todas las técnicas aprendidas en los módulos anteriores. El resultado es un sistema funcional con prompt registry, evaluation pipeline, cost tracking, caching, model routing y monitoring.
Este no es un ejercicio académico. Es el sistema que usarías para deployar un clasificador, un generador de respuestas, o cualquier aplicación LLM a producción real.
Por Qué Este Proyecto
Los módulos anteriores cubrieron cada técnica de forma aislada:
- Módulo 2: Few-shot prompting
- Módulo 3: Structured output
- Módulo 4: Chain-of-thought
- Módulo 5: Model routing
- Módulo 7: Evaluation
- Módulo 8: Versioning, caching, monitoring
El problema de aprender técnicas aisladas: En producción, todo interactúa. El cache afecta los costos. El routing afecta la calidad. El versioning afecta la evaluación. Este proyecto integra todo en un sistema cohesivo.
El Sistema que Vas a Construir
Un Sistema de Clasificación de Tickets de Soporte con las siguientes capacidades:
| Capacidad | Módulos Relacionados | Complejidad |
|---|---|---|
| Clasificar tickets (Urgente/Normal/Bajo) | M2, M3, M4 | Base |
| Versionado de prompts | M8 | Intermedia |
| Evaluación automática | M7 | Intermedia |
| Cache de queries repetidas | M8 | Intermedia |
| Routing por complejidad | M5, M8 | Intermedia |
| Cost tracking | M8 | Básica |
| Monitoring y alertas | M8 | Avanzada |
| Deploy checklist | M8 | Proceso |
Arquitectura del Sistema
production-prompt-system/
├── prompts/
│ ├── registry.py # Gestión de versiones de prompts
│ ├── templates.py # Templates con Jinja2
│ └── techniques.py # Few-shot bank, CoT
├── evaluation/
│ ├── metrics.py # Accuracy, faithfulness, format
│ ├── judge.py # LLM-as-judge
│ └── pipeline.py # Pipeline de evaluación completo
├── production/
│ ├── cache.py # InMemory + Semantic cache
│ ├── cost_tracker.py # Token counting y cost tracking
│ ├── router.py # Model routing por complejidad
│ └── monitor.py # Metrics collection y alertas
├── datasets/
│ └── golden_set.json # 100 ejemplos de tickets anotados
├── config/
│ └── settings.py # Configuración centralizada
├── api.py # FastAPI endpoints
├── cli.py # CLI para operaciones
└── main.py # Entry point y demo
Paso 1: Configuración Centralizada
# config/settings.py
from dataclasses import dataclass, field
from typing import Optional
import os
@dataclass
class Settings:
"""Configuración centralizada del sistema."""
# OpenAI
openai_api_key: str = field(default_factory=lambda: os.getenv("OPENAI_API_KEY", ""))
model_economico: str = "gpt-4o-mini"
model_premium: str = "gpt-4o"
# Prompt Registry
registry_path: str = "data/prompt_registry.json"
# Evaluation
golden_set_path: str = "datasets/golden_set.json"
accuracy_minima: float = 0.85
faithfulness_minima: float = 0.80
# Cache
cache_max_size: int = 1000
cache_ttl_segundos: int = 3600 # 1 hora
cache_semantic_threshold: float = 0.92
# Cost
budget_diario_usd: float = 10.0
budget_mensual_usd: float = 200.0
# Monitoring
latencia_max_ms: int = 5000
error_rate_max: float = 0.05
# Canary
canary_porcentaje_inicial: float = 0.05
@property
def tokens_por_dolar_mini(self) -> float:
return 1_000_000 / 0.15 # gpt-4o-mini: $0.15 / 1M tokens input
@property
def tokens_por_dolar_premium(self) -> float:
return 1_000_000 / 2.50 # gpt-4o: $2.50 / 1M tokens input
settings = Settings()Paso 2: El Prompt Registry
# prompts/registry.py
import json
from pathlib import Path
from datetime import datetime
from typing import Optional
class PromptRegistry:
"""Registry de versiones de prompts con rollback."""
def __init__(self, path: str = "data/prompt_registry.json"):
self.path = Path(path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self._data = self._cargar()
def _cargar(self) -> dict:
if self.path.exists():
with open(self.path) as f:
return json.load(f)
return {}
def _guardar(self):
with open(self.path, "w") as f:
json.dump(self._data, f, indent=2, ensure_ascii=False)
def registrar(self, nombre: str, version: str, template: str, metadata: dict = None) -> dict:
"""Registra una nueva versión de un prompt."""
if nombre not in self._data:
self._data[nombre] = {"activa": None, "versiones": {}}
entrada = {
"template": template,
"metadata": metadata or {},
"creado_en": datetime.now().isoformat(),
"estado": "inactivo"
}
self._data[nombre]["versiones"][version] = entrada
self._guardar()
print(f"📝 Registrado: {nombre} {version}")
return entrada
def activar(self, nombre: str, version: str) -> None:
"""Activa una versión específica."""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no existe")
versiones = self._data[nombre]["versiones"]
if version not in versiones:
raise KeyError(f"Versión '{version}' no existe en '{nombre}'")
# Desactivar versión anterior
version_anterior = self._data[nombre]["activa"]
if version_anterior and version_anterior in versiones:
versiones[version_anterior]["estado"] = "inactivo"
# Activar nueva versión
versiones[version]["estado"] = "activo"
self._data[nombre]["activa"] = version
self._guardar()
print(f"✅ Activado: {nombre} → {version}")
def obtener(self, nombre: str, version: str = None) -> str:
"""Obtiene el template de un prompt."""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no existe")
v = version or self._data[nombre]["activa"]
if not v:
raise ValueError(f"'{nombre}' no tiene versión activa")
return self._data[nombre]["versiones"][v]["template"]
def rollback(self, nombre: str) -> str:
"""Hace rollback a la versión estable anterior."""
versiones = self._data.get(nombre, {}).get("versiones", {})
activa = self._data.get(nombre, {}).get("activa")
# Buscar versión anterior por fecha de creación
candidatos = [
(v, info["creado_en"])
for v, info in versiones.items()
if v != activa
]
if not candidatos:
raise ValueError(f"No hay versión anterior para '{nombre}'")
candidatos.sort(key=lambda x: x[1], reverse=True)
version_rollback = candidatos[0][0]
self.activar(nombre, version_rollback)
return version_rollback
def listar_versiones(self, nombre: str) -> list[dict]:
"""Lista todas las versiones de un prompt."""
versiones = self._data.get(nombre, {}).get("versiones", {})
activa = self._data.get(nombre, {}).get("activa")
return [
{
"version": v,
"estado": info["estado"],
"es_activa": v == activa,
"creado_en": info["creado_en"],
"metadata": info.get("metadata", {})
}
for v, info in versiones.items()
]Paso 3: Templates con Few-Shot y CoT
# prompts/templates.py
from jinja2 import Template
from openai import OpenAI
client = OpenAI()
# Few-shot examples para clasificación de tickets
FEW_SHOT_TICKETS = [
{
"ticket": "La aplicación se cayó y perdí 3 horas de trabajo no guardado",
"urgencia": "URGENTE",
"razon": "Pérdida de trabajo del usuario"
},
{
"ticket": "¿Cómo puedo cambiar mi contraseña?",
"urgencia": "BAJO",
"razon": "Consulta de información estándar"
},
{
"ticket": "No puedo procesar pagos desde esta mañana, afecta a todo mi equipo de ventas",
"urgencia": "URGENTE",
"razon": "Impacto en negocio y múltiples usuarios"
},
{
"ticket": "El botón de exportar PDF no funciona bien en Firefox",
"urgencia": "NORMAL",
"razon": "Bug en funcionalidad específica, tiene workaround"
},
{
"ticket": "Me gustaría sugerir una nueva función",
"urgencia": "BAJO",
"razon": "Feature request, no problema actual"
}
]
TEMPLATE_CLASIFICACION = Template("""Clasifica el nivel de urgencia de este ticket de soporte.
ESCALA:
- URGENTE: Sistema caído, pérdida de datos, impacto en múltiples usuarios o negocio crítico
- NORMAL: Bug que afecta funcionalidad importante pero tiene workaround
- BAJO: Consultas, sugerencias, bugs menores
{% if few_shot %}EJEMPLOS DE REFERENCIA:
{% for ej in ejemplos %}
Ticket: "{{ ej.ticket }}"
Urgencia: {{ ej.urgencia }}
Razón: {{ ej.razon }}
{% endfor %}
{% endif %}
{% if usar_cot %}INSTRUCCIONES:
1. Identifica el impacto en el usuario
2. Evalúa si afecta negocio o datos
3. Determina si hay workaround disponible
4. Asigna la urgencia basándote en los criterios
Análisis paso a paso:
{% endif %}
TICKET: {{ ticket }}
Responde en JSON: {"urgencia": "URGENTE|NORMAL|BAJO", "confianza": 0.0-1.0, "razon": "..."}""")
def render_prompt(ticket: str, few_shot: bool = True, usar_cot: bool = False) -> str:
"""Renderiza el prompt con opciones configurables."""
return TEMPLATE_CLASIFICACION.render(
ticket=ticket,
few_shot=few_shot,
ejemplos=FEW_SHOT_TICKETS[:3] if few_shot else [],
usar_cot=usar_cot
)Paso 4: Sistema de Evaluación
# evaluation/metrics.py
import json
from openai import OpenAI
client = OpenAI()
def evaluar_accuracy(predicciones: list[str], ground_truth: list[str]) -> float:
"""Accuracy normalizada para clasificación."""
if not predicciones:
return 0.0
def normalizar(s: str) -> str:
return s.strip().upper()
correctos = sum(
1 for p, g in zip(predicciones, ground_truth)
if normalizar(p) == normalizar(g)
)
return correctos / len(predicciones)
def evaluar_format_compliance(outputs: list[str]) -> float:
"""Verifica que los outputs sean JSON válido con los campos requeridos."""
validos = 0
campos_requeridos = {"urgencia", "confianza", "razon"}
for output in outputs:
try:
data = json.loads(output)
if campos_requeridos.issubset(set(data.keys())):
if data["urgencia"] in ["URGENTE", "NORMAL", "BAJO"]:
validos += 1
except json.JSONDecodeError:
pass
return validos / len(outputs) if outputs else 0.0
def evaluar_confianza_promedio(outputs: list[str]) -> float:
"""Promedio de scores de confianza reportados."""
confianzas = []
for output in outputs:
try:
data = json.loads(output)
confianzas.append(float(data.get("confianza", 0)))
except (json.JSONDecodeError, ValueError):
confianzas.append(0.0)
return sum(confianzas) / len(confianzas) if confianzas else 0.0
def llm_judge_batch(
tickets: list[str],
clasificaciones: list[str],
ground_truth: list[str],
sample_size: int = 20
) -> float:
"""
LLM-as-judge para evaluar calidad de razonamiento.
Solo evalúa una muestra para reducir costos.
"""
import random
indices = random.sample(range(len(tickets)), min(sample_size, len(tickets)))
scores = []
for i in indices:
try:
data = json.loads(clasificaciones[i])
razon = data.get("razon", "")
except (json.JSONDecodeError, KeyError):
razon = clasificaciones[i]
prompt = f"""Evalúa la calidad del razonamiento para esta clasificación de ticket.
Ticket: {tickets[i]}
Clasificación: {ground_truth[i]}
Razonamiento dado: {razon}
Puntúa del 1-5 SOLO el razonamiento (no si la clasificación es correcta):
5 = Razonamiento claro, específico y justificado
3 = Razonamiento válido pero genérico
1 = Sin razonamiento o incorrecto
Responde SOLO con un número del 1 al 5."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=10
)
try:
score = float(response.choices[0].message.content.strip()) / 5.0
scores.append(score)
except ValueError:
scores.append(0.6) # Default si no parsea
return sum(scores) / len(scores) if scores else 0.0# evaluation/pipeline.py
import time
from openai import OpenAI
from evaluation.metrics import (
evaluar_accuracy, evaluar_format_compliance,
evaluar_confianza_promedio, llm_judge_batch
)
client = OpenAI()
class EvaluationPipeline:
"""Pipeline completo de evaluación para el sistema de clasificación."""
def __init__(self, settings):
self.settings = settings
def ejecutar(
self,
prompt_template: str,
golden_set: list[dict],
nombre: str = "evaluacion",
verbose: bool = True
) -> dict:
"""
Ejecuta la evaluación completa.
Returns: dict con todas las métricas y report.
"""
inicio = time.time()
tickets = [ej["input"] for ej in golden_set]
ground_truth = [ej["expected_output"] for ej in golden_set]
if verbose:
print(f"🔍 Evaluando '{nombre}' con {len(golden_set)} ejemplos...")
# 1. Ejecutar el prompt en todos los ejemplos
outputs = []
for ticket in tickets:
try:
response = client.chat.completions.create(
model=self.settings.model_economico,
messages=[{
"role": "user",
"content": prompt_template.replace("{ticket}", ticket)
}],
temperature=0,
max_tokens=200,
response_format={"type": "json_object"}
)
outputs.append(response.choices[0].message.content)
except Exception as e:
outputs.append(f'{{"urgencia": "BAJO", "confianza": 0, "razon": "ERROR: {str(e)}"}}')
# 2. Extraer predicciones del JSON
import json
predicciones = []
for output in outputs:
try:
data = json.loads(output)
predicciones.append(data.get("urgencia", "DESCONOCIDO"))
except json.JSONDecodeError:
predicciones.append("PARSE_ERROR")
# 3. Calcular métricas
accuracy = evaluar_accuracy(predicciones, ground_truth)
format_compliance = evaluar_format_compliance(outputs)
confianza_prom = evaluar_confianza_promedio(outputs)
# LLM judge en muestra (más costoso)
reasoning_quality = llm_judge_batch(tickets, outputs, ground_truth, sample_size=10)
# 4. Identificar fallos
fallos = [
{
"ticket": tickets[i],
"prediccion": predicciones[i],
"ground_truth": ground_truth[i]
}
for i in range(len(tickets))
if predicciones[i] != ground_truth[i]
]
duracion = time.time() - inicio
return {
"nombre": nombre,
"n_ejemplos": len(golden_set),
"duracion_s": round(duracion, 2),
"metricas": {
"accuracy": round(accuracy, 4),
"format_compliance": round(format_compliance, 4),
"confianza_promedio": round(confianza_prom, 4),
"reasoning_quality": round(reasoning_quality, 4)
},
"fallos": fallos[:5], # Top 5 fallos para diagnóstico
"aprobado": (
accuracy >= self.settings.accuracy_minima and
format_compliance >= 0.95
)
}
def comparar_versiones(
self,
prompt_a: str,
prompt_b: str,
golden_set: list[dict]
) -> dict:
"""Compara dos versiones de un prompt."""
print("⚔️ Comparando versiones A vs B...")
resultado_a = self.ejecutar(prompt_a, golden_set, "version_a", verbose=False)
resultado_b = self.ejecutar(prompt_b, golden_set, "version_b", verbose=False)
metrics_a = resultado_a["metricas"]
metrics_b = resultado_b["metricas"]
deltas = {
metrica: round(metrics_b[metrica] - metrics_a[metrica], 4)
for metrica in metrics_a
}
ganador = "B" if deltas["accuracy"] > 0 else ("A" if deltas["accuracy"] < 0 else "EMPATE")
return {
"version_a": metrics_a,
"version_b": metrics_b,
"deltas": deltas,
"ganador": ganador,
"mejora_accuracy": f"{deltas['accuracy']:+.2%}"
}Paso 5: Producción - Cache, Router y Cost Tracker
# production/cache.py
import hashlib
import time
from typing import Optional
class InMemoryCache:
"""Cache en memoria con TTL para queries LLM."""
def __init__(self, max_size: int = 1000, ttl_segundos: int = 3600):
self.max_size = max_size
self.ttl = ttl_segundos
self._cache: dict[str, dict] = {}
self._hits = 0
self._misses = 0
def _key(self, prompt: str, model: str) -> str:
contenido = f"{model}:{prompt}"
return hashlib.sha256(contenido.encode()).hexdigest()[:16]
def get(self, prompt: str, model: str) -> Optional[str]:
key = self._key(prompt, model)
entry = self._cache.get(key)
if not entry:
self._misses += 1
return None
# Verificar TTL
if time.time() - entry["timestamp"] > self.ttl:
del self._cache[key]
self._misses += 1
return None
self._hits += 1
return entry["value"]
def set(self, prompt: str, model: str, value: str) -> None:
# Eviction simple si está lleno
if len(self._cache) >= self.max_size:
oldest_key = min(self._cache.items(), key=lambda x: x[1]["timestamp"])[0]
del self._cache[oldest_key]
key = self._key(prompt, model)
self._cache[key] = {"value": value, "timestamp": time.time()}
@property
def hit_rate(self) -> float:
total = self._hits + self._misses
return self._hits / total if total > 0 else 0.0
def stats(self) -> dict:
return {
"size": len(self._cache),
"max_size": self.max_size,
"hits": self._hits,
"misses": self._misses,
"hit_rate": round(self.hit_rate, 3)
}# production/router.py
import tiktoken
from openai import OpenAI
client = OpenAI()
class ModelRouter:
"""
Enruta requests al modelo apropiado según complejidad.
Simple/rápido → gpt-4o-mini
Complejo/ambiguo → gpt-4o
"""
TICKET_SIMPLES = [
"contraseña", "login", "acceso", "cómo", "dónde", "cuándo",
"tutorial", "guía", "sugerencia", "pregunta"
]
TICKET_COMPLEJOS = [
"perdí", "caído", "crítico", "urgente", "emergencia", "datos",
"no funciona", "error", "falla", "bug", "producción"
]
def __init__(self, settings):
self.settings = settings
self._modelo_decisiones = []
def seleccionar_modelo(self, ticket: str) -> tuple[str, str]:
"""
Selecciona el modelo y devuelve (modelo, razon).
"""
ticket_lower = ticket.lower()
# Heurística rápida por keywords
tiene_simples = any(k in ticket_lower for k in self.TICKET_SIMPLES)
tiene_complejos = any(k in ticket_lower for k in self.TICKET_COMPLEJOS)
if tiene_complejos:
modelo = self.settings.model_premium
razon = "keywords de alta urgencia detectados"
elif tiene_simples and not tiene_complejos:
modelo = self.settings.model_economico
razon = "ticket de baja complejidad"
else:
# Tokens como señal de complejidad
enc = tiktoken.encoding_for_model("gpt-4o")
n_tokens = len(enc.encode(ticket))
if n_tokens > 100:
modelo = self.settings.model_premium
razon = f"ticket largo ({n_tokens} tokens)"
else:
modelo = self.settings.model_economico
razon = "ticket corto y directo"
self._modelo_decisiones.append({
"ticket_preview": ticket[:50],
"modelo": modelo,
"razon": razon
})
return modelo, razon
def estadisticas(self) -> dict:
"""Estadísticas de routing."""
total = len(self._modelo_decisiones)
if not total:
return {}
eco = sum(1 for d in self._modelo_decisiones if d["modelo"] == self.settings.model_economico)
return {
"total_requests": total,
"economico": eco,
"premium": total - eco,
"pct_economico": round(eco / total, 3)
}# production/cost_tracker.py
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class RequestCost:
"""Costo de un request individual."""
timestamp: str
prompt_tokens: int
completion_tokens: int
model: str
costo_usd: float
cached: bool = False
class CostTracker:
"""Tracking de costos en tiempo real."""
PRECIOS = {
"gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
"gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000}
}
def __init__(self, settings):
self.settings = settings
self._requests: list[RequestCost] = []
def registrar(self, response, model: str, cached: bool = False) -> float:
"""Registra el costo de un response de la API."""
usage = response.usage
precios = self.PRECIOS.get(model, self.PRECIOS["gpt-4o-mini"])
costo = (
usage.prompt_tokens * precios["input"] +
usage.completion_tokens * precios["output"]
)
if cached:
costo = 0.0
self._requests.append(RequestCost(
timestamp=datetime.now().isoformat(),
prompt_tokens=usage.prompt_tokens,
completion_tokens=usage.completion_tokens,
model=model,
costo_usd=costo,
cached=cached
))
return costo
def costo_total(self) -> float:
return sum(r.costo_usd for r in self._requests)
def costo_por_modelo(self) -> dict[str, float]:
costos = {}
for r in self._requests:
costos[r.model] = costos.get(r.model, 0.0) + r.costo_usd
return costos
def reporte(self) -> str:
total = self.costo_total()
por_modelo = self.costo_por_modelo()
n_cached = sum(1 for r in self._requests if r.cached)
n_total = len(self._requests)
lines = [
"## Cost Report",
f"Total requests: {n_total}",
f"Cached (sin costo): {n_cached} ({n_cached/n_total:.0%} si n_total>0)",
f"",
"Costo por modelo:",
]
for modelo, costo in por_modelo.items():
lines.append(f" {modelo}: ${costo:.4f}")
lines.append(f"\nTotal: ${total:.4f}")
# Proyección mensual
if n_total > 0:
costo_por_request = total / n_total
proyeccion_1000_req = costo_por_request * 1000
lines.append(f"\nProyección por 1,000 requests: ${proyeccion_1000_req:.2f}")
return "\n".join(lines)Paso 6: Monitoring
# production/monitor.py
import time
import threading
from collections import deque
from datetime import datetime
class ProductionMonitor:
"""
Monitor de producción con ventana deslizante y alertas.
"""
def __init__(self, settings, ventana_segundos: int = 300):
self.settings = settings
self.ventana = ventana_segundos
self._lock = threading.Lock()
self._latencias = deque()
self._errores = deque()
self._requests = deque()
self._alertas_enviadas: set = set()
def registrar(self, latencia_ms: float, error: bool = False) -> None:
"""Registra un request."""
ahora = time.time()
with self._lock:
self._latencias.append((ahora, latencia_ms))
self._requests.append(ahora)
if error:
self._errores.append(ahora)
# Limpiar fuera de la ventana
cutoff = ahora - self.ventana
while self._latencias and self._latencias[0][0] < cutoff:
self._latencias.popleft()
while self._requests and self._requests[0] < cutoff:
self._requests.popleft()
while self._errores and self._errores[0] < cutoff:
self._errores.popleft()
def metricas_actuales(self) -> dict:
"""Calcula métricas actuales de la ventana."""
with self._lock:
latencias = [l for _, l in self._latencias]
n_requests = len(self._requests)
n_errores = len(self._errores)
if not latencias:
return {"status": "sin_datos"}
latencias_sorted = sorted(latencias)
n = len(latencias_sorted)
return {
"n_requests": n_requests,
"error_rate": n_errores / n_requests if n_requests > 0 else 0,
"latencia_p50": latencias_sorted[int(n * 0.50)],
"latencia_p95": latencias_sorted[int(n * 0.95)],
"latencia_p99": latencias_sorted[int(n * 0.99)] if n >= 100 else latencias_sorted[-1],
"ventana_segundos": self.ventana
}
def verificar_alertas(self) -> list[dict]:
"""Verifica si hay condiciones de alerta."""
metricas = self.metricas_actuales()
alertas = []
if metricas.get("status") == "sin_datos":
return alertas
# Alerta de latencia alta
if metricas["latencia_p95"] > self.settings.latencia_max_ms:
alerta_id = "latencia_alta"
if alerta_id not in self._alertas_enviadas:
alertas.append({
"tipo": alerta_id,
"mensaje": f"Latencia p95 alta: {metricas['latencia_p95']:.0f}ms (max: {self.settings.latencia_max_ms}ms)",
"severidad": "WARNING"
})
self._alertas_enviadas.add(alerta_id)
else:
self._alertas_enviadas.discard("latencia_alta")
# Alerta de error rate alta
if metricas["error_rate"] > self.settings.error_rate_max:
alerta_id = "error_rate_alta"
if alerta_id not in self._alertas_enviadas:
alertas.append({
"tipo": alerta_id,
"mensaje": f"Error rate alta: {metricas['error_rate']:.1%} (max: {self.settings.error_rate_max:.1%})",
"severidad": "CRITICAL"
})
self._alertas_enviadas.add(alerta_id)
else:
self._alertas_enviadas.discard("error_rate_alta")
return alertasPaso 7: El Sistema Integrado
# main.py — Sistema completo integrado
import json
import time
from pathlib import Path
from openai import OpenAI
from config.settings import Settings
from prompts.registry import PromptRegistry
from prompts.templates import render_prompt
from evaluation.pipeline import EvaluationPipeline
from production.cache import InMemoryCache
from production.router import ModelRouter
from production.cost_tracker import CostTracker
from production.monitor import ProductionMonitor
settings = Settings()
client = OpenAI()
# Inicializar componentes
registry = PromptRegistry(settings.registry_path)
eval_pipeline = EvaluationPipeline(settings)
cache = InMemoryCache(settings.cache_max_size, settings.cache_ttl_segundos)
router = ModelRouter(settings)
cost_tracker = CostTracker(settings)
monitor = ProductionMonitor(settings)
def clasificar_ticket(ticket: str, request_id: str = None) -> dict:
"""
Clasifica un ticket integrando todos los componentes del sistema.
"""
inicio = time.time()
cached = False
# 1. Obtener prompt activo del registry
prompt_template = registry.obtener("clasificador_tickets")
prompt = prompt_template.replace("{ticket}", ticket)
# 2. Verificar cache
modelo_seleccionado, razon_routing = router.seleccionar_modelo(ticket)
cached_result = cache.get(prompt, modelo_seleccionado)
if cached_result:
resultado = json.loads(cached_result)
latencia_ms = (time.time() - inicio) * 1000
monitor.registrar(latencia_ms, error=False)
return {
**resultado,
"cached": True,
"model": modelo_seleccionado,
"latencia_ms": round(latencia_ms, 1),
"costo_usd": 0.0
}
# 3. Llamar a la API
try:
response = client.chat.completions.create(
model=modelo_seleccionado,
messages=[{"role": "user", "content": prompt}],
temperature=0,
max_tokens=200,
response_format={"type": "json_object"}
)
output = response.choices[0].message.content
resultado = json.loads(output)
# 4. Registrar costo
costo = cost_tracker.registrar(response, modelo_seleccionado, cached=False)
# 5. Guardar en cache
cache.set(prompt, modelo_seleccionado, output)
latencia_ms = (time.time() - inicio) * 1000
monitor.registrar(latencia_ms, error=False)
return {
**resultado,
"cached": False,
"model": modelo_seleccionado,
"routing_razon": razon_routing,
"latencia_ms": round(latencia_ms, 1),
"costo_usd": round(costo, 6)
}
except Exception as e:
latencia_ms = (time.time() - inicio) * 1000
monitor.registrar(latencia_ms, error=True)
return {
"urgencia": "NORMAL",
"confianza": 0.0,
"razon": f"Error al clasificar: {str(e)}",
"error": True,
"latencia_ms": round(latencia_ms, 1)
}
def demo_sistema():
"""Demo del sistema completo."""
# Registrar versión inicial del prompt
registry.registrar(
nombre="clasificador_tickets",
version="v1.0.0",
template=render_prompt("{ticket}", few_shot=True, usar_cot=False),
metadata={"descripcion": "Clasificador con few-shot"}
)
registry.activar("clasificador_tickets", "v1.0.0")
# Tickets de ejemplo
tickets = [
"La aplicación no carga y tengo una demo importante en 30 minutos",
"¿Cómo exporto mis datos a Excel?",
"El sistema de pagos falló y estamos perdiendo ventas ahora mismo",
"¿Tienen opción de dark mode?",
"Encontré un bug pequeño en el hover de un botón",
]
print("🚀 DEMO: Sistema de Clasificación de Tickets")
print("=" * 60)
for ticket in tickets:
resultado = clasificar_ticket(ticket)
emoji = {"URGENTE": "🔴", "NORMAL": "🟡", "BAJO": "🟢"}.get(resultado["urgencia"], "⚪")
cached_str = " [CACHE]" if resultado.get("cached") else ""
print(f"\n{emoji} {resultado['urgencia']}{cached_str}")
print(f" Ticket: {ticket[:60]}...")
print(f" Razón: {resultado.get('razon', '')[:80]}")
print(f" Confianza: {resultado.get('confianza', 0):.0%}")
print(f" Modelo: {resultado.get('model', '')} | {resultado['latencia_ms']:.0f}ms | ${resultado.get('costo_usd', 0):.5f}")
print("\n" + "=" * 60)
print("📊 ESTADÍSTICAS DEL SISTEMA")
print("\n🗄️ Cache:")
stats = cache.stats()
print(f" Hit rate: {stats['hit_rate']:.0%} | Size: {stats['size']}/{stats['max_size']}")
print("\n🔀 Model Routing:")
routing_stats = router.estadisticas()
print(f" Económico: {routing_stats.get('pct_economico', 0):.0%}")
print(f" Premium: {1 - routing_stats.get('pct_economico', 0):.0%}")
print("\n💰 Costos:")
print(cost_tracker.reporte())
print("\n📡 Monitoring:")
metricas = monitor.metricas_actuales()
if metricas.get("status") != "sin_datos":
print(f" Latencia p95: {metricas.get('latencia_p95', 0):.0f}ms")
print(f" Error rate: {metricas.get('error_rate', 0):.1%}")
alertas = monitor.verificar_alertas()
if alertas:
print(f"\n⚠️ {len(alertas)} alerta(s) activa(s)")
else:
print("\n✅ Sin alertas activas")
if __name__ == "__main__":
demo_sistema()Criterios de Éxito
## Checklist del Proyecto Final
### Funcionalidad Core
- [ ] Prompt registry funcional (registrar, activar, rollback)
- [ ] Clasificación de tickets funciona con accuracy ≥ 85%
- [ ] Few-shot examples integrados en el template
- [ ] Output en JSON válido (format_compliance ≥ 95%)
### Evaluación
- [ ] Golden set con ≥ 50 ejemplos en golden_set.json
- [ ] Evaluation pipeline ejecuta y genera reporte
- [ ] Métricas incluyen: accuracy, format_compliance, reasoning_quality
- [ ] Comparison A/B entre dos versiones del prompt
### Producción
- [ ] Cache implementado (hit rate ≥ 20% con queries repetidas)
- [ ] Model routing funcional (≥ 60% requests van a modelo económico)
- [ ] Cost tracker activo (reporte de costo por modelo)
- [ ] Monitor activo (latencia p95 y error rate)
### Deploy
- [ ] ProductionChecklist completada
- [ ] Rollback documentado y probado
- [ ] Alertas configuradas
### Calidad de Código
- [ ] Configuración centralizada en settings.py
- [ ] Sin API keys hardcodeadas
- [ ] Error handling en todas las llamadas API
- [ ] Código ejecutable (no solo demo pseudocódigo)Extensiones Opcionales
Una vez completado el proyecto base, considera estas extensiones para profundizar:
| Extensión | Técnica | Dificultad |
|---|---|---|
| API REST con FastAPI | FastAPI + Pydantic | Intermedia |
| Semantic cache con embeddings | Embeddings + cosine similarity | Intermedia |
| Dashboard en Streamlit | Streamlit | Baja |
| Tracing con LangSmith | LangSmith SDK | Baja |
| A/B testing automatizado | Stats + pytest | Alta |
| Prometheus + Grafana | prometheus_client | Alta |
# api.py — Extensión opcional: FastAPI endpoint
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="Ticket Classifier API", version="1.0.0")
class TicketRequest(BaseModel):
ticket: str
request_id: str = None
class TicketResponse(BaseModel):
urgencia: str
confianza: float
razon: str
model: str
latencia_ms: float
cached: bool
costo_usd: float
@app.post("/clasificar", response_model=TicketResponse)
async def clasificar(request: TicketRequest):
resultado = clasificar_ticket(request.ticket, request.request_id)
if resultado.get("error"):
raise HTTPException(status_code=500, detail=resultado["razon"])
return resultado
@app.get("/health")
async def health():
metricas = monitor.metricas_actuales()
return {"status": "ok", "metricas": metricas}
@app.get("/stats")
async def stats():
return {
"cache": cache.stats(),
"routing": router.estadisticas(),
"costos": cost_tracker.costo_por_modelo()
}
# Para ejecutar: uvicorn api:app --reloadTroubleshooting
Problema 1: Accuracy baja (< 85%)
# Diagnosticar qué ejemplos fallan y por qué
resultado_eval = eval_pipeline.ejecutar(
prompt_template=registry.obtener("clasificador_tickets"),
golden_set=golden_set,
nombre="diagnostico"
)
print("Fallos más comunes:")
for fallo in resultado_eval["fallos"]:
print(f" Ticket: {fallo['ticket'][:60]}")
print(f" Predicción: {fallo['prediccion']} | Ground truth: {fallo['ground_truth']}")
print()
# Soluciones:
# 1. Revisar si los fallos tienen un patrón (todos son NORMAL→URGENTE?)
# 2. Agregar ejemplos del tipo que falla al few-shot bank
# 3. Ajustar las instrucciones del prompt para ese caso específicoProblema 2: Cache hit rate bajo
# Analizar por qué el cache no tiene hits
stats = cache.stats()
print(f"Hit rate: {stats['hit_rate']:.0%}")
# Si hit rate < 5% con queries similares → revisar si el prompt exacto varía
# Solución: normalizar el ticket antes de generar el cache key
def normalizar_ticket(ticket: str) -> str:
"""Normaliza para mejorar hit rate."""
import re
# Lowercase, eliminar puntuación doble, trim espacios
ticket = ticket.lower().strip()
ticket = re.sub(r'\s+', ' ', ticket)
ticket = re.sub(r'[!?]{2,}', '!', ticket)
return ticketResumen
- Arquitectura modular: cada componente es independiente y testeable
- Prompt registry: versionado semántico con rollback en 1 línea
- Evaluation pipeline: accuracy + format_compliance + LLM judge automático
- Cache: reduce costos hasta 40-60% en queries repetidas
- Model routing: 60-80% del tráfico al modelo económico
- Cost tracking: visibilidad total de gasto por modelo y por período
- Monitoring: latencia p95, error rate, alertas automáticas
- Deploy checklist: proceso completo antes de ir a producción
Recursos adicionales
- OpenAI Production Best Practices — Guía oficial
- LangSmith Docs — Observabilidad para LLMs
- FastAPI Tutorial — Construir APIs con FastAPI
- Jinja2 Template Designer — Templates avanzados
- Prometheus Python Client — Métricas de producción
- The Twelve-Factor App — Principios para apps cloud-native