Déploiement production des agents IA

┌─────────────────────────────────────────────────┐
│              UTILISATEURS                       │
└────────────────────┬────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────┐
│          LOAD BALANCER (Nginx/AWS ALB)          │
└────────────────────┬────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────┐
│        API GATEWAY (Kong/AWS API Gateway)       │
│  - Rate limiting                                │
│  - Authentication                               │
│  - Request validation                           │
└────────────────────┬────────────────────────────┘
                     │
                     ↓
┌─────────────────────────────────────────────────┐
│           AGENT SERVICE (FastAPI/Flask)         │
│  - Business logic                               │
│  - Agent orchestration                          │
│  - Prompt management                            │
└────────────────────┬────────────────────────────┘
                     │
         ┌───────────┼───────────┐
         ↓           ↓           ↓
    ┌────────┐ ┌─────────┐ ┌─────────┐
    │ Redis  │ │Vector DB│ │Database │
    │ Cache  │ │(Pinecone│ │(Postgres│
    └────────┘ └─────────┘ └─────────┘
         │
         ↓
┌─────────────────────────────────────────────────┐
│          LLM GATEWAY (LiteLLM/Portkey)          │
│  - Multi-provider routing                       │
│  - Fallbacks                                    │
│  - Cost tracking                                │
└────────────────────┬────────────────────────────┘
                     │
         ┌───────────┼───────────┐
         ↓           ↓           ↓
    ┌────────┐ ┌─────────┐ ┌─────────┐
    │OpenAI  │ │Anthropic│ │Google   │
    │API     │ │API      │ │API      │
    └────────┘ └─────────┘ └─────────┘

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional, List
import redis
import json
import time
from datetime import datetime
import openai

# ========== CONFIGURATION ==========

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

# CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # À restreindre en production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

# Redis pour caching
redis_client = redis.Redis(
    host="localhost",
    port=6379,
    decode_responses=True
)

# ========== MODELS ==========

class AgentRequest(BaseModel):
    """Requête agent"""
    query: str = Field(..., min_length=1, max_length=1000)
    user_id: Optional[str] = None
    context: Optional[dict] = None
    max_tokens: Optional[int] = Field(default=1000, le=4000)

class AgentResponse(BaseModel):
    """Réponse agent"""
    answer: str
    execution_time_ms: float
    tokens_used: int
    cost_usd: float
    cached: bool
    request_id: str

class HealthResponse(BaseModel):
    """Health check"""
    status: str
    timestamp: str
    version: str

# ========== DEPENDENCY INJECTION ==========

async def get_current_user(api_key: str = Depends(get_api_key)):
    """Authentification"""
    # Vérifier l'API key
    user = authenticate_api_key(api_key)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return user

async def rate_limit_check(user: dict = Depends(get_current_user)):
    """Rate limiting"""
    user_id = user["id"]
    key = f"rate_limit:{user_id}"

    # Compteur de requêtes
    count = redis_client.incr(key)
    if count == 1:
        redis_client.expire(key, 60)  # 1 minute

    # Vérifier la limite (ex: 60 req/min)
    if count > 60:
        raise HTTPException(
            status_code=429,
            detail="Rate limit exceeded. Max 60 requests per minute."
        )

    return True

# ========== ROUTES ==========

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0"
    }

@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(
    request: AgentRequest,
    background_tasks: BackgroundTasks,
    user: dict = Depends(get_current_user),
    rate_limited: bool = Depends(rate_limit_check)
):
    """Exécute l'agent"""

    start_time = time.time()
    request_id = generate_request_id()

    # 1. Vérifier le cache
    cache_key = f"agent_cache:{hash(request.query)}"
    cached_response = redis_client.get(cache_key)

    if cached_response:
        response = json.loads(cached_response)
        response["cached"] = True
        response["request_id"] = request_id
        return response

    # 2. Exécuter l'agent
    try:
        agent = create_agent(user)
        answer = agent.run(request.query)

        # 3. Calculer les métriques
        execution_time = (time.time() - start_time) * 1000
        tokens_used = estimate_tokens(request.query + answer)
        cost = calculate_cost(tokens_used)

        # 4. Construire la réponse
        response = AgentResponse(
            answer=answer,
            execution_time_ms=execution_time,
            tokens_used=tokens_used,
            cost_usd=cost,
            cached=False,
            request_id=request_id
        )

        # 5. Cacher la réponse (5 minutes)
        redis_client.setex(
            cache_key,
            300,
            json.dumps(response.dict())
        )

        # 6. Logger en arrière-plan
        background_tasks.add_task(
            log_request,
            request_id=request_id,
            user_id=user["id"],
            query=request.query,
            response=answer,
            metrics={
                "execution_time_ms": execution_time,
                "tokens_used": tokens_used,
                "cost_usd": cost
            }
        )

        return response

    except Exception as e:
        # Logger l'erreur
        background_tasks.add_task(
            log_error,
            request_id=request_id,
            error=str(e)
        )

        raise HTTPException(
            status_code=500,
            detail=f"Agent execution failed: {str(e)}"
        )

@app.post("/agent/stream")
async def stream_agent(request: AgentRequest):
    """Version streaming de l'agent"""

    from fastapi.responses import StreamingResponse

    async def generate():
        agent = create_streaming_agent()
        async for chunk in agent.run_stream(request.query):
            yield f"data: {json.dumps(chunk)}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

# ========== HELPER FUNCTIONS ==========

def create_agent(user: dict):
    """Crée un agent configuré pour l'utilisateur"""
    from agents import MyAgent

    return MyAgent(
        user_id=user["id"],
        model=user.get("preferred_model", "gpt-4"),
        max_iterations=10
    )

def generate_request_id() -> str:
    """Génère un ID de requête unique"""
    import uuid
    return str(uuid.uuid4())

def estimate_tokens(text: str) -> int:
    """Estime le nombre de tokens"""
    # Approximation : ~0.75 tokens par mot
    return int(len(text.split()) * 0.75)

def calculate_cost(tokens: int, model: str = "gpt-4") -> float:
    """Calcule le coût"""
    # GPT-4 : ~$0.03/1k input tokens, $0.06/1k output
    return (tokens / 1000) * 0.045  # Moyenne

def log_request(request_id: str, user_id: str, query: str, response: str, metrics: dict):
    """Log une requête (async)"""
    # Insérer dans la base de données
    pass

def log_error(request_id: str, error: str):
    """Log une erreur (async)"""
    # Insérer dans la base de données + alerting
    pass

# ========== STARTUP/SHUTDOWN ==========

@app.on_event("startup")
async def startup_event():
    """Actions au démarrage"""
    print("🚀 Agent API starting...")

    # Vérifier les connexions
    assert redis_client.ping(), "Redis not available"

    # Charger les modèles si nécessaire
    # warm_up_models()

    print("✅ Agent API ready")

@app.on_event("shutdown")
async def shutdown_event():
    """Actions à l'arrêt"""
    print("⏹️ Agent API shutting down...")

    # Fermer les connexions
    redis_client.close()

    print("✅ Agent API stopped")

# ========== LANCEMENT ==========

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True,  # False en production
        workers=4      # Ajuster selon le trafic
    )

# .github/workflows/deploy.yml

name: Deploy Agent API

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  PYTHON_VERSION: "3.11"
  DOCKER_IMAGE: mycompany/agent-api

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Cache dependencies
        uses: actions/cache@v3
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}

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

      - name: Run tests
        run: |
          pytest tests/ -v --cov=agents --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml

  lint:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install linters
        run: |
          pip install black flake8 mypy

      - name: Run black
        run: black --check .

      - name: Run flake8
        run: flake8 agents/ tests/

      - name: Run mypy
        run: mypy agents/

  build:
    needs: [test, lint]
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2

      - name: Login to DockerHub
        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}

      - name: Build and push
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: ${{ env.DOCKER_IMAGE }}:${{ github.sha }},${{ env.DOCKER_IMAGE }}:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max

  deploy:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - name: Deploy to production
        uses: appleboy/ssh-action@master
        with:
          host: ${{ secrets.PROD_HOST }}
          username: ${{ secrets.PROD_USER }}
          key: ${{ secrets.SSH_PRIVATE_KEY }}
          script: |
            cd /app/agent-api
            docker-compose pull
            docker-compose up -d --no-deps --build api
            docker-compose exec -T api python -m pytest tests/smoke_tests.py

# Dockerfile

FROM python:3.11-slim

WORKDIR /app

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

# Copier les requirements
COPY requirements.txt .

# Installer les dépendances Python
RUN pip install --no-cache-dir -r requirements.txt

# Copier le code
COPY . .

# Créer un utilisateur non-root
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# Exposer le port
EXPOSE 8000

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

# Lancer l'application
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_HOST=redis
      - POSTGRES_HOST=postgres
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - redis
      - postgres
    restart: unless-stopped
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '2.0'
          memory: 4G

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

  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: agent_db
      POSTGRES_USER: agent_user
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
    volumes:
      - grafana_data:/var/lib/grafana
    restart: unless-stopped

volumes:
  redis_data:
  postgres_data:
  prometheus_data:
  grafana_data:

# monitoring.py

from prometheus_client import Counter, Histogram, Gauge, generate_latest
import time

# Métriques
request_counter = Counter(
    "agent_requests_total",
    "Total agent requests",
    ["user_id", "status"]
)

latency_histogram = Histogram(
    "agent_latency_seconds",
    "Agent response latency",
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

token_counter = Counter(
    "agent_tokens_total",
    "Total tokens used",
    ["model"]
)

cost_counter = Counter(
    "agent_cost_usd_total",
    "Total cost in USD",
    ["model"]
)

active_requests = Gauge(
    "agent_active_requests",
    "Number of active requests"
)

# Décorateur pour tracking
def track_metrics(func):
    """Décorateur pour tracker les métriques"""
    async def wrapper(*args, **kwargs):
        active_requests.inc()
        start_time = time.time()

        try:
            result = await func(*args, **kwargs)

            # Succès
            request_counter.labels(
                user_id=kwargs.get("user_id", "unknown"),
                status="success"
            ).inc()

            return result

        except Exception as e:
            # Erreur
            request_counter.labels(
                user_id=kwargs.get("user_id", "unknown"),
                status="error"
            ).inc()

            raise

        finally:
            # Latence
            latency = time.time() - start_time
            latency_histogram.observe(latency)

            active_requests.dec()

    return wrapper

# Endpoint Prometheus
@app.get("/metrics")
async def metrics():
    """Expose Prometheus metrics"""
    return Response(
        generate_latest(),
        media_type="text/plain"
    )

{
  "dashboard": {
    "title": "Agent API Monitoring",
    "panels": [
      {
        "title": "Requests per Second",
        "targets": [
          {
            "expr": "rate(agent_requests_total[5m])"
          }
        ]
      },
      {
        "title": "P95 Latency",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(agent_latency_seconds_bucket[5m]))"
          }
        ]
      },
      {
        "title": "Error Rate",
        "targets": [
          {
            "expr": "rate(agent_requests_total{status=\"error\"}[5m]) / rate(agent_requests_total[5m])"
          }
        ]
      },
      {
        "title": "Cost per Hour",
        "targets": [
          {
            "expr": "rate(agent_cost_usd_total[1h]) * 3600"
          }
        ]
      }
    ]
  }
}

import hashlib

class PromptCache:
    """Cache intelligent pour prompts"""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1 heure

    def get_cached_response(self, prompt: str, model: str) -> Optional[str]:
        """Récupère une réponse en cache"""
        cache_key = self._generate_key(prompt, model)
        return self.redis.get(cache_key)

    def cache_response(self, prompt: str, model: str, response: str):
        """Cache une réponse"""
        cache_key = self._generate_key(prompt, model)
        self.redis.setex(cache_key, self.ttl, response)

    def _generate_key(self, prompt: str, model: str) -> str:
        """Génère une clé de cache"""
        content = f"{model}:{prompt}"
        return f"prompt_cache:{hashlib.md5(content.encode()).hexdigest()}"

# Utilisation
cache = PromptCache(redis_client)

# Vérifier le cache
cached = cache.get_cached_response(prompt, model)
if cached:
    return cached

# Sinon, appeler le LLM et cacher
response = openai.chat.completions.create(...)
cache.cache_response(prompt, model, response)

class IntelligentRouter:
    """Route les requêtes vers le modèle optimal"""

    def __init__(self):
        self.complexity_threshold = 0.5

    def route(self, query: str) -> str:
        """Détermine le modèle optimal"""

        complexity = self._estimate_complexity(query)

        if complexity < 0.3:
            return "gpt-3.5-turbo"  # Requêtes simples
        elif complexity < 0.7:
            return "gpt-4"  # Requêtes moyennes
        else:
            return "gpt-4-turbo"  # Requêtes complexes

    def _estimate_complexity(self, query: str) -> float:
        """Estime la complexité (0-1)"""

        score = 0.0

        # Longueur
        if len(query) > 200:
            score += 0.2

        # Mots-clés complexes
        complex_keywords = ["analyze", "compare", "explain", "detailed"]
        if any(kw in query.lower() for kw in complex_keywords):
            score += 0.3

        # Questions multiples
        if query.count("?") > 1:
            score += 0.2

        return min(score, 1.0)

# Économie estimée : -40% de coûts

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"  # Normal
    OPEN = "open"      # Erreurs détectées, bloque les requêtes
    HALF_OPEN = "half_open"  # Test de récupération

class CircuitBreaker:
    """Circuit breaker pour LLM APIs"""

    def __init__(
        self,
        failure_threshold: int = 5,
        timeout: int = 60,
        recovery_timeout: int = 30
    ):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.recovery_timeout = recovery_timeout

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        """Exécute une fonction avec circuit breaker"""

        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN")

        try:
            result = func(*args, **kwargs)

            # Succès
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0

            return result

        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN

            raise

# Utilisation
breaker = CircuitBreaker()

try:
    response = breaker.call(openai.chat.completions.create, **params)
except Exception as e:
    # Fallback ou erreur
    pass

from pydantic import BaseModel, Field, validator

class SecureAgentRequest(BaseModel):
    """Requête avec validation stricte"""

    query: str = Field(..., min_length=1, max_length=1000)

    @validator("query")
    def validate_query(cls, v):
        """Validation du query"""

        # Bloquer les injections
        dangerous_patterns = [
            "ignore previous",
            "disregard",
            "system:",
            "<script>",
            "DROP TABLE"
        ]

        for pattern in dangerous_patterns:
            if pattern.lower() in v.lower():
                raise ValueError(f"Potential injection detected: {pattern}")

        return v

from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential

class SecretsManager:
    """Gestion sécurisée des secrets"""

    def __init__(self, vault_url: str):
        credential = DefaultAzureCredential()
        self.client = SecretClient(vault_url=vault_url, credential=credential)

    def get_secret(self, name: str) -> str:
        """Récupère un secret"""
        return self.client.get_secret(name).value

# Utilisation
secrets = SecretsManager("https://my-vault.vault.azure.net/")
openai.api_key = secrets.get_secret("openai-api-key")

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/agent/run")
@limiter.limit("60/minute")  # 60 requêtes/minute
async def run_agent(request: Request, ...):
    pass

import asyncio

async def run_with_timeout(func, timeout=30):
    """Exécute avec timeout"""
    try:
        return await asyncio.wait_for(func(), timeout=timeout)
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="Request timeout")

import signal

def handle_sigterm(*args):
    """Handle SIGTERM pour graceful shutdown"""
    print("SIGTERM received, shutting down gracefully...")
    # Terminer les requêtes en cours
    # Fermer les connexions
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_sigterm)

INFRASTRUCTURE
├─ Compute (3× API instances @ 4 vCPUs)      : $400/mois
├─ Redis (managed, 8GB)                      : $100/mois
├─ PostgreSQL (managed, 50GB)                : $150/mois
├─ Load Balancer                             : $30/mois
└─ Bandwidth (200GB sortant)                 : $20/mois
Total Infrastructure                         : $700/mois

LLM COSTS (10k req/jour × 30 jours = 300k req/mois)
├─ Sans caching
│   ├─ GPT-4 (avg 1500 tokens @ $0.045/1k)  : $20,250/mois 💸
│   └─ GPT-3.5 (avg 800 tokens @ $0.002/1k) : $480/mois
│
├─ Avec caching 60% + routing intelligent
│   ├─ 40% GPT-4 (120k req)                 : $8,100/mois
│   ├─ 60% GPT-3.5 (180k req)               : $288/mois
│   └─ Cache hits (60%)                     : $0/mois ✅
│   Total LLM optimisé                      : $8,388/mois

MONITORING & OUTILS
├─ Prometheus + Grafana (self-hosted)        : $0 (inclus compute)
├─ Logging (CloudWatch/Datadog)              : $100/mois
├─ Secrets Manager                           : $10/mois
└─ CI/CD (GitHub Actions)                    : $0 (gratuit <2000 min)
Total Monitoring                             : $110/mois

TOTAL MENSUEL : $9,198/mois
Budget par requête : $0.031/req (3.1 cents)

# Sans caching
300,000 req × $0.0675 (avg GPT-4) = $20,250/mois

# Avec caching 60%
120,000 req × $0.0675 (GPT-4)     = $8,100/mois
180,000 req × $0.0016 (GPT-3.5)   = $288/mois
Total                              = $8,388/mois

Économie                           = $11,862/mois (58.6%)
ROI Redis ($100)                   = 11,862% 🚀

# Vérifier latence réseau vers LLM
curl -w "@curl-format.txt" -o /dev/null -s https://api.openai.com/v1/models

# Analyser logs Prometheus
rate(agent_latency_seconds_bucket[5m])

# Identifier les slow queries
grep "execution_time_ms" logs/*.json | awk '{if($NF>5000) print}'

# Analyser distribution tokens
SELECT
    DATE(timestamp),
    AVG(tokens_used),
    MAX(tokens_used),
    COUNT(*) as requests
FROM agent_logs
GROUP BY DATE(timestamp)
ORDER BY DATE(timestamp) DESC;

# Identifier utilisateurs coûteux
SELECT
    user_id,
    SUM(cost_usd) as total_cost,
    COUNT(*) as requests
FROM agent_logs
WHERE timestamp > NOW() - INTERVAL 24 HOUR
GROUP BY user_id
ORDER BY total_cost DESC
LIMIT 10;

# Top 10 erreurs
grep "ERROR" logs/*.json | cut -d'"' -f4 | sort | uniq -c | sort -rn | head -10

# Vérifier health endpoints
curl http://localhost:8000/health
curl http://localhost:6379/ping  # Redis
curl https://api.openai.com/v1/models  # OpenAI

# Monitorer mémoire container
docker stats agent-api

# Analyser memory leaks potentiels
import tracemalloc
tracemalloc.start()
# ... code ...
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')