Déployer des Applications IA en Production : Guide Complet Docker, Kubernetes, Monitoring et Optimisations

┌─────────────────────────────────────────────────────┐
│                    UTILISATEURS                      │
└────────────────────┬────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────┐
│            Load Balancer (NGINX/Traefik)            │
│          + Rate Limiting + SSL/TLS                  │
└────────────────────┬────────────────────────────────┘
                     │
        ┌────────────┴────────────┐
        │                         │
        ▼                         ▼
┌──────────────┐          ┌──────────────┐
│   API Pod 1  │          │   API Pod 2  │
│  (FastAPI)   │   ...    │  (FastAPI)   │
└──────┬───────┘          └──────┬───────┘
       │                         │
       └────────────┬────────────┘
                    │
        ┌───────────┴────────────┐
        │                        │
        ▼                        ▼
┌─────────────┐          ┌─────────────┐
│   Redis     │          │  PostgreSQL │
│  (Cache)    │          │  (Metadata) │
└─────────────┘          └─────────────┘
        │
        ▼
┌─────────────────────────────────────┐
│      LLM Provider (OpenAI/etc)      │
└─────────────────────────────────────┘
        │
        ▼
┌─────────────────────────────────────┐
│   Monitoring (Prometheus/Grafana)   │
│   Logging (ELK/Loki)                │
│   Tracing (LangSmith)               │
└─────────────────────────────────────┘

# main.py
from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from pydantic import BaseModel
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
import logging

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

# Rate limiting
limiter = Limiter(key_func=get_remote_address)

# FastAPI app
app = FastAPI(
    title="AI Assistant API",
    version="1.0.0",
    description="Production-ready AI API"
)

# Rate limit handler
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

# CORS (ajuster selon vos besoins)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Chain LangChain
prompt = ChatPromptTemplate.from_template("Réponds à : {question}")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm

# Ajouter la route LangServe
add_routes(
    app,
    chain,
    path="/chat",
    enable_feedback_endpoint=True,  # Pour le monitoring
)

# Health check
@app.get("/health")
async def health_check():
    return {"status": "healthy", "version": "1.0.0"}

# Endpoint custom avec rate limiting
@app.post("/api/v1/query")
@limiter.limit("10/minute")  # 10 requêtes par minute par IP
async def query(request: Request, query: QueryRequest):
    try:
        logger.info(f"Query received: {query.question[:50]}...")

        # Validation
        if len(query.question) > 1000:
            raise HTTPException(400, "Question trop longue (max 1000 chars)")

        # Exécution
        response = chain.invoke({"question": query.question})

        logger.info(f"Query processed successfully")
        return {"answer": response.content}

    except Exception as e:
        logger.error(f"Error processing query: {str(e)}")
        raise HTTPException(500, f"Internal error: {str(e)}")

# Models
class QueryRequest(BaseModel):
    question: str

# Startup event
@app.on_event("startup")
async def startup_event():
    logger.info("API starting up...")
    # Initialiser Redis, etc.

# Shutdown event
@app.on_event("shutdown")
async def shutdown_event():
    logger.info("API shutting down...")
    # Cleanup

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

# test_api.py
import pytest
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

def test_query_success():
    response = client.post(
        "/api/v1/query",
        json={"question": "Qu'est-ce que le RAG ?"}
    )
    assert response.status_code == 200
    assert "answer" in response.json()

def test_query_rate_limit():
    # Envoyer 11 requêtes (limit = 10/min)
    for i in range(11):
        response = client.post(
            "/api/v1/query",
            json={"question": f"Test {i}"}
        )

    # La 11e doit être rate limitée
    assert response.status_code == 429

def test_query_validation():
    # Question trop longue
    response = client.post(
        "/api/v1/query",
        json={"question": "a" * 1001}
    )
    assert response.status_code == 400

# Dockerfile
FROM python:3.11-slim as builder

# Variables d'environnement
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=1

# Dépendances système
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Workdir
WORKDIR /app

# Installer les dépendances Python
COPY requirements.txt .
RUN pip install --upgrade pip && \
    pip install -r requirements.txt

# Stage final (image minimale)
FROM python:3.11-slim

WORKDIR /app

# Copier les dépendances du builder
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin

# Copier le code
COPY . .

# User non-root pour sécurité
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

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

# Exposer le port
EXPOSE 8000

# Commande de démarrage
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

fastapi==0.109.0
uvicorn[standard]==0.27.0
langchain==0.1.0
langchain-openai==0.0.5
langserve==0.0.43
pydantic==2.5.0
python-dotenv==1.0.0
redis==5.0.1
slowapi==0.1.9
prometheus-client==0.19.0

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    volumes:
      - ./logs:/app/logs

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana

volumes:
  redis_data:
  grafana_data:

# Build
docker build -t ai-api:1.0.0 .

# Test local
docker run -p 8000:8000 -e OPENAI_API_KEY=$OPENAI_API_KEY ai-api:1.0.0

# Test avec compose
docker-compose up -d

# Vérifier les logs
docker-compose logs -f api

# Test de l'API
curl http://localhost:8000/health

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-api
  labels:
    app: ai-api
spec:
  replicas: 3  # 3 instances pour HA
  selector:
    matchLabels:
      app: ai-api
  template:
    metadata:
      labels:
        app: ai-api
    spec:
      containers:
      - name: api
        image: your-registry/ai-api:1.0.0
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: ai-secrets
              key: openai-api-key
        - name: REDIS_URL
          value: "redis://redis-service:6379"
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          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
---
# service.yaml
apiVersion: v1
kind: Service
metadata:
  name: ai-api-service
spec:
  selector:
    app: ai-api
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer
---
# hpa.yaml (Horizontal Pod Autoscaler)
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-api
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
---
# ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: ai-api-ingress
  annotations:
    cert-manager.io/cluster-issuer: "letsencrypt-prod"
    nginx.ingress.kubernetes.io/rate-limit: "100"
spec:
  tls:
  - hosts:
    - api.yourdomain.com
    secretName: ai-api-tls
  rules:
  - host: api.yourdomain.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: ai-api-service
            port:
              number: 80

# Créer un secret pour l'API key
kubectl create secret generic ai-secrets \
  --from-literal=openai-api-key=$OPENAI_API_KEY

# Ou via fichier YAML (encodé en base64)
echo -n "sk-..." | base64  # Copier le résultat

# secrets.yaml
apiVersion: v1
kind: Secret
metadata:
  name: ai-secrets
type: Opaque
data:
  openai-api-key: <base64-encoded-key>

# Appliquer les manifestes
kubectl apply -f deployment.yaml
kubectl apply -f service.yaml
kubectl apply -f hpa.yaml
kubectl apply -f ingress.yaml

# Vérifier le déploiement
kubectl get pods -l app=ai-api
kubectl get svc ai-api-service
kubectl describe hpa ai-api-hpa

# Logs
kubectl logs -f deployment/ai-api

# Scaling manuel
kubectl scale deployment ai-api --replicas=5

# main.py (ajout métriques Prometheus)
from prometheus_client import Counter, Histogram, generate_latest
from prometheus_client import CONTENT_TYPE_LATEST
import time

# Métriques
request_count = Counter(
    'api_requests_total',
    'Total API requests',
    ['method', 'endpoint', 'status']
)

request_duration = Histogram(
    'api_request_duration_seconds',
    'API request duration',
    ['method', 'endpoint']
)

llm_tokens = Counter(
    'llm_tokens_total',
    'Total LLM tokens used',
    ['model']
)

# Middleware pour tracking
@app.middleware("http")
async def track_requests(request: Request, call_next):
    start_time = time.time()

    response = await call_next(request)

    duration = time.time() - start_time
    request_count.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()

    request_duration.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)

    return response

# Endpoint métriques
@app.get("/metrics")
async def metrics():
    return Response(
        generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

# Configuration LangSmith
import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-..."
os.environ["LANGCHAIN_PROJECT"] = "production-api"

# Toutes les requêtes sont automatiquement tracées

# cache.py
import redis
import hashlib
import json
from functools import wraps

redis_client = redis.Redis(
    host='redis',
    port=6379,
    decode_responses=True
)

def cache_llm_response(ttl=3600):
    """Cache les réponses LLM pendant ttl secondes"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Créer une clé de cache
            cache_key = hashlib.md5(
                json.dumps(kwargs, sort_keys=True).encode()
            ).hexdigest()

            # Vérifier le cache
            cached = redis_client.get(cache_key)
            if cached:
                logger.info(f"Cache HIT: {cache_key}")
                return json.loads(cached)

            # Exécuter la fonction
            result = await func(*args, **kwargs)

            # Mettre en cache
            redis_client.setex(
                cache_key,
                ttl,
                json.dumps(result)
            )

            logger.info(f"Cache MISS: {cache_key}")
            return result

        return wrapper
    return decorator

# Utilisation
@cache_llm_response(ttl=3600)
async def get_llm_response(question: str):
    response = chain.invoke({"question": question})
    return {"answer": response.content}

# Batch processing pour embeddings
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(embed_batch_size=100)  # Batch de 100

# Streaming pour réponses longues
async def stream_response(question: str):
    async for chunk in chain.astream({"question": question}):
        yield chunk.content

# Connection pooling pour bases de données
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "postgresql://...",
    poolclass=QueuePool,
    pool_size=20,
    max_overflow=10
)

from langchain.callbacks import get_openai_callback

async def track_cost(question: str):
    with get_openai_callback() as cb:
        response = chain.invoke({"question": question})

        # Logger les coûts
        logger.info(f"Cost: ${cb.total_cost:.4f}")
        logger.info(f"Tokens: {cb.total_tokens}")

        # Envoyer à un système de billing
        await save_cost_metrics({
            "tokens": cb.total_tokens,
            "cost": cb.total_cost,
            "model": "gpt-4o-mini"
        })

    return response

def choose_model(question: str):
    """Choisit le modèle selon la complexité"""
    if len(question) < 50:
        return "gpt-4o-mini"  # Simple
    elif "complex" in question.lower():
        return "gpt-4"  # Complexe
    else:
        return "gpt-4o-mini"  # Par défaut

# .github/workflows/deploy.yml
name: Deploy to Production

on:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest

      - name: Run tests
        run: pytest tests/

  build:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Build Docker image
        run: |
          docker build -t ${{ secrets.REGISTRY }}/ai-api:${{ github.sha }} .

      - name: Push to registry
        run: |
          echo ${{ secrets.DOCKER_PASSWORD }} | docker login -u ${{ secrets.DOCKER_USERNAME }} --password-stdin
          docker push ${{ secrets.REGISTRY }}/ai-api:${{ github.sha }}

  deploy:
    needs: build
    runs-on: ubuntu-latest
    steps:
      - name: Deploy to Kubernetes
        run: |
          kubectl set image deployment/ai-api \
            api=${{ secrets.REGISTRY }}/ai-api:${{ github.sha }}

          kubectl rollout status deployment/ai-api