Práctica: Despliegue y Monitoreo del Sistema RAG

Concepto clave

El despliegue y monitoreo de un sistema RAG (Retrieval Augmented Generation) es el proceso de llevar tu aplicación desde un entorno de desarrollo a producción, asegurando que funcione de manera confiable, escalable y observable. Piensa en esto como lanzar un nuevo restaurante: no basta con tener una gran cocina (desarrollo), necesitas un local abierto al público (despliegue), personal capacitado (infraestructura) y sistemas para medir la satisfacción de los clientes (monitoreo).

En el contexto de sistemas RAG con bases de datos vectoriales como Chroma, Pinecone y pgvector, el despliegue implica configurar servidores, gestionar conexiones a las bases de datos, y asegurar que el retrieval (recuperación de vectores) y la generación (respuestas del LLM) funcionen en tiempo real. El monitoreo, por otro lado, te permite detectar problemas como latencia alta, fallos en la recuperación, o respuestas de baja calidad, similar a cómo un chef supervisa los pedidos y la calidad de los platos.

Cómo funciona en la práctica

Imagina que has desarrollado un sistema RAG para responder preguntas sobre documentación técnica usando Chroma como base de datos vectorial. Aquí está el proceso paso a paso para desplegarlo:

1. Preparación del entorno: Empaqueta tu aplicación en un contenedor Docker con todas las dependencias, incluyendo la conexión a Chroma (almacenada localmente o en la nube).
2. Configuración de la infraestructura: Despliega el contenedor en un servicio como AWS ECS o Kubernetes, asegurando recursos suficientes para manejar picos de tráfico.
3. Integración de monitoreo: Configura herramientas como Prometheus para métricas (ej., tiempo de respuesta del retrieval) y Grafana para dashboards visuales.
4. Pruebas en producción: Lanza una versión inicial a un pequeño grupo de usuarios, monitoreando errores y rendimiento.

Por ejemplo, en código, podrías agregar un endpoint de salud en tu API para verificar que Chroma esté accesible:

from fastapi import FastAPI
import chromadb

app = FastAPI()
chroma_client = chromadb.Client()

@app.get("/health")
def health_check():
    try:
        # Verificar conexión a Chroma
        chroma_client.heartbeat()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

Caso de estudio

Una empresa de e-learning despliega un sistema RAG usando Pinecone para proporcionar respuestas personalizadas a preguntas de estudiantes sobre cursos en línea. El sistema se integra con su plataforma web existente.

Después del despliegue, el equipo notó un pico en la latencia durante horas pico. Al revisar las métricas, descubrieron que las consultas a Pinecone estaban tardando 150ms. Solucionaron esto ajustando la configuración del índice para usar una región más cercana y aumentando la capacidad, reduciendo la latencia a 80ms.

Lección clave: Monitorear métricas específicas como la latencia de retrieval es crucial para mantener la experiencia del usuario.

Errores comunes

• No planificar la escalabilidad: Desplegar sin considerar picos de tráfico puede saturar la base de datos vectorial. Solución: Usa autoescalado en servicios como Pinecone o configura réplicas en pgvector.
• Ignorar el monitoreo de la calidad de respuestas: Solo medir disponibilidad sin evaluar la precisión del retrieval. Solución: Implementa checks periódicos con un conjunto de preguntas de prueba y compara las respuestas generadas.
• Falta de manejo de errores en producción: Asumir que las conexiones a Chroma o Pinecone siempre funcionan. Solución: Agrega reintentos y circuit breakers en tu código para fallos temporales.
• No versionar los datos vectoriales: Actualizar embeddings sin un sistema de control puede romper consultas existentes. Solución: Usa etiquetas o índices separados para diferentes versiones de datos.
• Desplegar sin pruebas de carga: No simular tráfico real antes del lanzamiento. Solución: Realiza pruebas con herramientas como Locust para identificar cuellos de botella.

Checklist de dominio

• He empaquetado mi aplicación RAG en un contenedor Docker con todas las dependencias.
• He configurado conexiones seguras a mi base de datos vectorial (Chroma, Pinecone o pgvector) en producción.
• He implementado métricas de monitoreo para latencia, errores y uso de recursos.
• He establecido alertas para detectar fallos en el retrieval o generación.
• He realizado pruebas de carga para asegurar que el sistema maneje el tráfico esperado.
• He documentado los procedimientos de rollback en caso de problemas.
• He asegurado que los datos vectoriales estén respaldados y versionados.

Despliega y monitorea un sistema RAG básico con Chroma

En este ejercicio, desplegarás una aplicación RAG simple en un entorno local simulado de producción y configurarás monitoreo básico. Sigue estos pasos:

1. Prepara la aplicación: Usa el siguiente código base que incluye un servidor FastAPI, Chroma para almacenamiento vectorial, y un LLM simulado. Guarda esto como app.py.

   from fastapi import FastAPI
   import chromadb
   import time
   import random
   
   app = FastAPI()
   chroma_client = chromadb.Client()
   collection = chroma_client.create_collection(name="test-rag")
   
   # Agregar algunos embeddings de ejemplo
   collection.add(
       embeddings=[[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]],
       documents=["Documento A", "Documento B"],
       ids=["id1", "id2"]
   )
   
   @app.get("/query")
   def query_rag(q: str):
       start_time = time.time()
       try:
           results = collection.query(query_embeddings=[[0.15, 0.25, 0.35]], n_results=1)
           latency = time.time() - start_time
           # Simular generación de respuesta
           response = f"Respuesta basada en: {results['documents'][0][0]}"
           return {"query": q, "response": response, "latency_seconds": round(latency, 3)}
       except Exception as e:
           return {"error": str(e), "latency_seconds": round(time.time() - start_time, 3)}
   
   @app.get("/metrics")
   def get_metrics():
       # Métrica simulada: latencia promedio (en un caso real, esto vendría de un almacén)
       return {"avg_latency": 0.1, "total_queries": 100}

2. Despliega localmente: Crea un Dockerfile para contenerizar la aplicación y ejecútala con Docker Compose, incluyendo Chroma como servicio.
3. Configura monitoreo: Agrega un endpoint /health que verifique la conexión a Chroma y devuelva el estado. Luego, usa una herramienta como curl o un script Python para hacer solicitudes periódicas a /query y registrar la latencia en un archivo de log.
4. Prueba y ajusta: Simula tráfico con 10 solicitudes consecutivas, analiza los logs para identificar cualquier latencia alta o errores, y ajusta la configuración si es necesario (ej., aumentar recursos).

• Usa 'chromadb' en modo persistente para simular producción, configurando un volumen en Docker.
• Para monitoreo simple, puedes escribir las métricas de latencia en un archivo CSV y visualizarlas con una hoja de cálculo.
• Asegúrate de manejar excepciones en el endpoint /query para evitar caídas del servidor.

Evalua tu comprension

Completa el quiz interactivo de arriba para ganar XP.