Introduction
En tant qu'architecte infrastructure senior ayant déployé des systèmes d'API IA pour des entreprises traitant plusieurs milliards de tokens par mois, je peux vous affirmer sans détour : la protection contre les attaques DDoS et la gestion du rate limiting constituent les fondations absolues de toute infrastructure d'API performante. J'ai récemment migré notre architecture critique vers HolySheep AI et les résultats en termes de latence et de sécurité m'ont impressionné.
Comprendre les Coûts Réels des API d'IA en 2026
Avant d'aborder l'architecture technique, il est crucial de comprendre l'écosystème pricing actuel. Voici les tarifs vérifiés au premier trimestre 2026 pour les modèles de sortie (output) :
- GPT-4.1 : 8,00 $/MTok
- Claude Sonnet 4.5 : 15,00 $/MTok
- Gemini 2.5 Flash : 2,50 $/MTok
- DeepSeek V3.2 : 0,42 $/MTok
Comparaison de Coûts pour 10 Millions de Tokens/Mois
| Modèle | Coût Mensuel | Coût Annuel |
|---|---|---|
| GPT-4.1 | 80 000 $ | 960 000 $ |
| Claude Sonnet 4.5 | 150 000 $ | 1 800 000 $ |
| Gemini 2.5 Flash | 25 000 $ | 300 000 $ |
| DeepSeek V3.2 | 4 200 $ | 50 400 $ |
Ces chiffres illustrent pourquoi une architecture de rate limiting robuste n'est pas optionnelle : une attaque DDoS ciblant votre endpoint d'API pourrait vous coûter des dizaines de milliers de dollars en quelques minutes seulement.
Architecture de Rate Limiting Multi-Niveaux
Principe du Token Bucket
J'utilise personnellement l'algorithme du Token Bucket pour sa flexibilité. Chaque client reçoit un seau de tokens qui se remplit à un débit constant. Les requêtes consomment des tokens, et les requêtes sans tokens disponibles sont rejetées ou mises en file d'attente.
import time
import threading
from collections import defaultdict
class TokenBucket:
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.refill_rate = refill_rate # tokens par seconde
self._buckets = defaultdict(lambda: {"tokens": capacity, "last_refill": time.time()})
self._lock = threading.Lock()
def consume(self, client_id: str, tokens: int = 1) -> bool:
with self._lock:
bucket = self._buckets[client_id]
now = time.time()
# Calcul du refill automatique
elapsed = now - bucket["last_refill"]
bucket["tokens"] = min(
self.capacity,
bucket["tokens"] + elapsed * self.refill_rate
)
bucket["last_refill"] = now
# Vérification et consommation
if bucket["tokens"] >= tokens:
bucket["tokens"] -= tokens
return True
return False
def get_remaining(self, client_id: str) -> float:
with self._lock:
bucket = self._buckets[client_id]
elapsed = time.time() - bucket["last_refill"]
return min(self.capacity, bucket["tokens"] + elapsed * self.refill_rate)
Configuration : 1000 tokens/minute par client
rate_limiter = TokenBucket(capacity=1000, refill_rate=1000/60)
Test du rate limiter
for i in range(5):
client = f"client_{i % 3}"
result = rate_limiter.consume(client, tokens=100)
remaining = rate_limiter.get_remaining(client)
print(f"{client}: {'✓ Autorisé' if result else '✗ Rejeté'} | Restant: {remaining:.1f} tokens")
Middleware de Protection DDoS avec FastAPI
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
import redis.asyncio as redis
from typing import Optional
import hashlib
app = FastAPI()
Rate limiter distribué avec Redis
class DistributedRateLimiter:
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url, decode_responses=True)
async def check_rate_limit(
self,
key: str,
max_requests: int,
window_seconds: int
) -> tuple[bool, int]:
"""
Retourne (autorisé, requêtes restantes)
"""
now = await self.redis.time()
current_window = int(now[0])
pipe = self.redis.pipeline()
window_key = f"rate:{key}:{current_window // window_seconds}"
pipe.incr(window_key)
pipe.expire(window_key, window_seconds * 2)
results = await pipe.execute()
current_count = results[0]
remaining = max(0, max_requests - current_count)
allowed = current_count <= max_requests
return allowed, remaining
async def check_ddos_pattern(
self,
client_ip: str,
threshold: int = 100,
window: int = 10
) -> bool:
"""
Détecte les patterns DDoS : >threshold requêtes en secondes
"""
now = await self.redis.time()
key = f"ddos:{client_ip}"
pipe = self.redis.pipeline()
pipe.lpush(key, now[0])
pipe.ltrim(key, 0, threshold * 2)
pipe.expire(key, window * 2)
await pipe.execute()
recent_requests = await self.redis.lrange(key, 0, -1)
window_start = now[0] - window
requests_in_window = sum(1 for r in recent_requests if int(float(r)) > window_start)
return requests_in_window > threshold
rate_limiter = DistributedRateLimiter("redis://localhost:6379")
@app.middleware("http")
async def security_middleware(request: Request, call_next):
client_ip = get_remote_address(request)
# Vérification DDoS
is_ddos = await rate_limiter.check_ddos_pattern(client_ip)
if is_ddos:
return JSONResponse(
status_code=429,
content={
"error": "Trop de requêtes",
"retry_after": 60,
"code": "DDOS_PROTECTION_ACTIVE"
}
)
# Vérification rate limit par endpoint
endpoint = request.url.path
rate_key = f"{client_ip}:{endpoint}"
allowed, remaining = await rate_limiter.check_rate_limit(
rate_key,
max_requests=100,
window_seconds=60
)
if not allowed:
return JSONResponse(
status_code=429,
content={
"error": "Rate limit atteint",
"remaining": 0,
"retry_after": 60
},
headers={"X-RateLimit-Remaining": "0"}
)