Module 8: Final Capstone Project - Complete RAG System

Production Deployment: Docker, FastAPI, Kubernetes

Next →

Description

In this capsule you'll take your RAG system to production with Docker (containerization), FastAPI (REST API), and Kubernetes (scaling). You'll learn the best practices for deployment, monitoring, and health checks.

A production-ready RAG system isn't just code that runs locally—it must be:

  • Containerized (Docker) for consistent deployment
  • API-fied (FastAPI) for consumption by other services
  • Scalable (Kubernetes) to handle variable traffic
  • Observable (Prometheus/Grafana) for troubleshooting
  • Resilient (health checks, graceful shutdown)

By the end you'll have a complete system deployable to any cloud provider (AWS, GCP, Azure).

Estimated duration: 50-60 minutes

Objectives

By completing this capsule, you'll be able to:

  • ✅ Containerize the RAG system with Docker
  • ✅ Create a REST API with FastAPI
  • ✅ Configure docker-compose for local development
  • ✅ Implement health checks and graceful shutdown
  • ✅ Deploy to Kubernetes (local or cloud)
  • ✅ Set up monitoring with Prometheus

Step 1: Docker Setup

1.1: Optimized Dockerfile

Create Dockerfile:

# Multi-stage build for a smaller 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"]

Optimizations:

  • ✅ Multi-stage build (reduces image size ~50%)
  • ✅ Non-root user (security)
  • ✅ Integrated health check
  • ✅ Layer caching (dependencies separated)

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

Step 2: FastAPI REST API

2.1: Complete API with monitoring

Create 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 the RAG components"""
        if self._initialized:
            return
        
        logger.info("Initializing RAG service...")
        
        # Load the pipeline
        self.pipeline = RAGPipeline(chunk_size=500, overlap=50)
        
        # Load the FAISS index
        self.index = FAISSIndex(dim=1536)
        self.index.load("./faiss_index")
        
        # Load the reranker
        self.reranker = Reranker()
        
        self._initialized = True
        logger.info("✅ RAG service initialized")
    
    def health_check(self) -> Dict:
        """Check the 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 the 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


# Logging middleware
@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 the search
        results = rag_service.search(
            query=request.query,
            k=request.k,
            use_reranking=request.use_reranking
        )
        
        # Format the 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)

Step 3: Docker Compose for 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'

Step 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

Deployment commands

Local development:

# Build and 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

Problem 1: Container crashes immediately

Cause: FAISS index not found

Solution:

# Verify that the index exists
ls -la faiss_index/

# Or use a volume mount
docker run -v $(pwd)/faiss_index:/app/faiss_index rag-system

Problem 2: Health check fails

Cause: The service takes time to initialize

Solution:

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

Summary

In this capsule you implemented:

  • ✅ Optimized multi-stage Dockerfile
  • ✅ FastAPI REST API with Prometheus metrics
  • ✅ docker-compose for local development
  • ✅ Health checks and graceful shutdown
  • ✅ Kubernetes deployment with autoscaling
  • ✅ Monitoring with Prometheus + Grafana

Next capsule: Conclusions and Next Steps - A recap of everything you learned.

Additional Resources

  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

Module 8 - Capsule 07

← PreviousQuery Expansion & Reranking: Two-Stage RetrievalNext →Conclusions and Next Steps