Módulo 8: Proyecto Final Integrador - RAG System Completo

Production Deployment: Docker, FastAPI, Kubernetes

Siguiente →

Descripción

En esta cápsula llevarás tu sistema RAG a production con Docker (containerization), FastAPI (REST API), y Kubernetes (scaling). Aprenderás las mejores prácticas para deployment, monitoring, y health checks.

Un sistema RAG production-ready no es solo código que funciona localmente—debe ser:

  • Containerizado (Docker) para deployment consistente
  • API-fied (FastAPI) para consumo de otros servicios
  • Escalable (Kubernetes) para manejar tráfico variable
  • Observable (Prometheus/Grafana) para troubleshooting
  • Resiliente (health checks, graceful shutdown)

Al final tendrás un sistema completo deployable a cualquier cloud provider (AWS, GCP, Azure).

Duración estimada: 50-60 minutos

Objetivos

Al completar esta cápsula, serás capaz de:

  • ✅ Containerizar sistema RAG con Docker
  • ✅ Crear REST API con FastAPI
  • ✅ Configurar docker-compose para local development
  • ✅ Implementar health checks y graceful shutdown
  • ✅ Deployar a Kubernetes (local o cloud)
  • ✅ Setup monitoring con Prometheus

Paso 1: Docker Setup

1.1: Dockerfile optimizado

Crear Dockerfile:

# Multi-stage build para menor image size
FROM python:3.10-slim as builder

# Install build dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Create venv
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# --------------------------------------------------
# Final stage (smaller image)
FROM python:3.10-slim

# Copy venv from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

WORKDIR /app

# Copy source code
COPY src/ ./src/
COPY api.py .
COPY .env.example .env

# Create non-root user
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Expose API port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

Optimizaciones:

  • ✅ Multi-stage build (reduce image size ~50%)
  • ✅ Non-root user (security)
  • ✅ Health check integrado
  • ✅ Layer caching (dependencies separadas)

1.2: requirements.txt

# Core
fastapi==0.109.0
uvicorn[standard]==0.27.0
pydantic==2.6.0

# RAG components
openai==1.12.0
sentence-transformers==2.3.1
faiss-cpu==1.7.4
langchain==0.1.6
tiktoken==0.5.2

# Infrastructure
redis==5.0.1
prometheus-client==0.19.0
python-dotenv==1.0.0

# Utilities
numpy==1.26.3
scipy==1.12.0

1.3: .dockerignore

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/

# Environment
.env
.env.local

# IDE
.vscode/
.idea/
*.swp
*.swo

# Data
data/
*.pkl
*.index
faiss_index/

# Tests
.pytest_cache/
.coverage

# Docs
docs/
*.md
README.md

Paso 2: FastAPI REST API

2.1: API completa con monitoring

Crear api.py:

"""
RAG System REST API
Production-ready FastAPI server
"""

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import List, Dict, Optional
import time
import logging
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from prometheus_client.core import CollectorRegistry
from fastapi.responses import Response

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Prometheus metrics
REQUESTS = Counter('rag_requests_total', 'Total requests', ['endpoint', 'status'])
LATENCY = Histogram('rag_request_duration_seconds', 'Request latency', ['endpoint'])

# Initialize app
app = FastAPI(
    title="RAG System API",
    description="Production-ready RAG system with embeddings and vector search",
    version="1.0.0"
)

# Initialize RAG components (lazy loading)
from src.pipeline.rag_pipeline import RAGPipeline
from src.search.faiss_index import FAISSIndex
from src.retrieval.reranker import Reranker

class RAGService:
    """Singleton RAG service"""
    
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialized = False
        return cls._instance
    
    def initialize(self):
        """Initialize RAG components"""
        if self._initialized:
            return
        
        logger.info("Initializing RAG service...")
        
        # Load pipeline
        self.pipeline = RAGPipeline(chunk_size=500, overlap=50)
        
        # Load FAISS index
        self.index = FAISSIndex(dim=1536)
        self.index.load("./faiss_index")
        
        # Load reranker
        self.reranker = Reranker()
        
        self._initialized = True
        logger.info("✅ RAG service initialized")
    
    def health_check(self) -> Dict:
        """Check service health"""
        return {
            'status': 'healthy' if self._initialized else 'initializing',
            'index_size': self.index.index.ntotal if self._initialized else 0,
            'version': '1.0.0'
        }
    
    def search(self, query: str, k: int = 5, use_reranking: bool = True) -> List[Dict]:
        """Search with optional reranking"""
        if not self._initialized:
            raise RuntimeError("Service not initialized")
        
        # Generate embedding
        query_emb = self.pipeline.embedder.embed(query)
        
        # FAISS search
        if use_reranking:
            candidates = self.index.search(query_emb, k=k*4)
            results = self.reranker.rerank(query, candidates, top_k=k)
        else:
            results = self.index.search(query_emb, k=k)
        
        return results

# Global service instance
rag_service = RAGService()


# Pydantic models
class SearchRequest(BaseModel):
    query: str = Field(..., description="Search query", min_length=1, max_length=1000)
    k: int = Field(5, description="Number of results", ge=1, le=50)
    use_reranking: bool = Field(True, description="Use cross-encoder reranking")

class SearchResult(BaseModel):
    chunk_id: str
    text: str
    score: float
    metadata: Dict

class SearchResponse(BaseModel):
    results: List[SearchResult]
    query: str
    k: int
    latency_ms: float

class HealthResponse(BaseModel):
    status: str
    index_size: int
    version: str


# Middleware para logging
@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log all requests"""
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    
    logger.info(
        f"{request.method} {request.url.path} "
        f"status={response.status_code} duration={duration:.3f}s"
    )
    
    # Prometheus metrics
    REQUESTS.labels(
        endpoint=request.url.path,
        status=response.status_code
    ).inc()
    
    LATENCY.labels(endpoint=request.url.path).observe(duration)
    
    return response


# API Endpoints
@app.on_event("startup")
async def startup_event():
    """Initialize on startup"""
    rag_service.initialize()

@app.get("/", tags=["Health"])
async def root():
    """Root endpoint"""
    return {
        "service": "RAG System API",
        "version": "1.0.0",
        "status": "running"
    }

@app.get("/health", response_model=HealthResponse, tags=["Health"])
async def health():
    """Health check endpoint"""
    try:
        health_status = rag_service.health_check()
        return HealthResponse(**health_status)
    except Exception as e:
        logger.error(f"Health check failed: {e}")
        raise HTTPException(status_code=503, detail="Service unhealthy")

@app.post("/search", response_model=SearchResponse, tags=["Search"])
async def search(request: SearchRequest):
    """
    Search endpoint with vector similarity
    
    - **query**: Search query (1-1000 characters)
    - **k**: Number of results (1-50)
    - **use_reranking**: Use cross-encoder reranking (slower but more accurate)
    """
    start_time = time.time()
    
    try:
        # Execute search
        results = rag_service.search(
            query=request.query,
            k=request.k,
            use_reranking=request.use_reranking
        )
        
        # Format response
        formatted_results = []
        for r in results:
            formatted_results.append(SearchResult(
                chunk_id=r['chunk'].id,
                text=r['chunk'].text,
                score=r.get('cross_encoder_score', r['score']),
                metadata=r['chunk'].metadata
            ))
        
        latency_ms = (time.time() - start_time) * 1000
        
        return SearchResponse(
            results=formatted_results,
            query=request.query,
            k=request.k,
            latency_ms=latency_ms
        )
    
    except Exception as e:
        logger.error(f"Search error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/metrics", tags=["Monitoring"])
async def metrics():
    """Prometheus metrics endpoint"""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.get("/stats", tags=["Monitoring"])
async def stats():
    """System statistics"""
    try:
        return {
            'index': rag_service.index.get_stats(),
            'service': rag_service.health_check()
        }
    except Exception as e:
        logger.error(f"Stats error: {e}")
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Paso 3: Docker Compose para Development

3.1: docker-compose.yml

version: '3.8'

services:
  # RAG API
  rag-api:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - redis
    volumes:
      - ./faiss_index:/app/faiss_index:ro
    networks:
      - rag-network
    restart: unless-stopped
  
  # Redis cache
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    networks:
      - rag-network
    restart: unless-stopped
    command: redis-server --appendonly yes
  
  # Prometheus monitoring
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus_data:/prometheus
    networks:
      - rag-network
    restart: unless-stopped
  
  # Grafana dashboards
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana
    networks:
      - rag-network
    restart: unless-stopped
    depends_on:
      - prometheus

networks:
  rag-network:
    driver: bridge

volumes:
  redis_data:
  prometheus_data:
  grafana_data:

3.2: prometheus.yml

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'rag-api'
    static_configs:
      - targets: ['rag-api:8000']
    metrics_path: '/metrics'

Paso 4: Kubernetes Deployment

4.1: deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-system
  labels:
    app: rag
spec:
  replicas: 3
  selector:
    matchLabels:
      app: rag
  template:
    metadata:
      labels:
        app: rag
    spec:
      containers:
      - name: rag-api
        image: your-registry/rag-system:latest
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-keys
              key: openai-key
        - name: REDIS_HOST
          value: "redis-service"
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        volumeMounts:
        - name: faiss-index
          mountPath: /app/faiss_index
          readOnly: true
      volumes:
      - name: faiss-index
        persistentVolumeClaim:
          claimName: faiss-index-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: rag-service
spec:
  selector:
    app: rag
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

Comandos de Deployment

Local development:

# Build y start
docker-compose up --build

# Test API
curl http://localhost:8000/health

# Search
curl -X POST http://localhost:8000/search \
  -H "Content-Type: application/json" \
  -d '{"query": "What is Python?", "k": 5}'

# Stop
docker-compose down

Kubernetes:

# Create secret
kubectl create secret generic api-keys \
  --from-literal=openai-key=sk-...

# Deploy
kubectl apply -f deployment.yaml

# Check status
kubectl get pods
kubectl get services

# Logs
kubectl logs -f deployment/rag-system

# Scale
kubectl scale deployment rag-system --replicas=5

Troubleshooting

Problema 1: Container crashes inmediatamente

Causa: FAISS index no encontrado

Solución:

# Verificar que index existe
ls -la faiss_index/

# O usar volume mount
docker run -v $(pwd)/faiss_index:/app/faiss_index rag-system

Problema 2: Health check fails

Causa: Service tarda en inicializar

Solución:

# Aumentar initialDelaySeconds en healthcheck
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

Resumen

En esta cápsula implementaste:

  • ✅ Dockerfile multi-stage optimizado
  • ✅ FastAPI REST API con Prometheus metrics
  • ✅ docker-compose para local development
  • ✅ Health checks y graceful shutdown
  • ✅ Kubernetes deployment con autoscaling
  • ✅ Monitoring con Prometheus + Grafana

Próxima cápsula: Conclusiones y Next Steps - Resumen de todo lo aprendido.

Recursos Adicionales

  1. FastAPI Documentation - Official docs
  2. Docker Best Practices - Docker guides
  3. Kubernetes Docs - K8s reference
  4. Prometheus Client - Metrics library
  5. Uvicorn - ASGI server
  6. Docker Compose - Multi-container Docker
  7. Production FastAPI - Deployment guide

Módulo 8 - Cápsula 07

← AnteriorQuery Expansion & Reranking: Two-Stage RetrievalSiguiente →Conclusiones y Next Steps