Déploiement d'ingénierie IA

Vous avez exécuté un modèle sur votre ordinateur portable, et il a généré des réponses rapidement et avec précision. Mais lorsque vous le placez sur un serveur et laissez une centaine d'utilisateurs l'appeler simultanément, la situation change :Certains utilisateurs attendent dix secondes pour une réponse, certaines requêtes expirent et signalent des erreurs, la mémoire GPU se remplit en quelques secondes, et la facture grimpe à un niveau douloureux.

C'est le problème que le déploiement de l'IA doit résoudre :Transformer un modèle exécutable en un service utilisable。

Le goulot d'étranglement des services web traditionnels est généralement le CPU et la base de données.

Le goulot d'étranglement des services d'IA est principalement le GPU — si la mémoire est suffisante, si le calcul est rapide, et comment les requêtes concurrentes sont mises en file d'attente.

Les principaux défis du déploiement de l'IA peuvent se résumer en trois mots : latence, débit et coût.

Métrique	Signification	Expérience utilisateur
Latence	Temps entre l'envoi d'une requête par un utilisateur et la réception du premier caractère	Rapide ou non
Débit	Combien de requêtes peuvent être traitées par seconde	S'il peut servir de nombreuses personnes en même temps
Coût	Coût du GPU/serveur pour exécuter le service	Cher ou non

Un bon déploiement d'IA consiste à trouver un équilibre entre ces trois éléments : une latence suffisamment faible, un débit suffisamment élevé et un coût maîtrisable.

Framework de service de modèles

Pour encapsuler un modèle entraîné sous forme de service API, vous avez besoin d'un framework dédié. Cette section présente trois solutions courantes : vLLM, TGI et Ollama.

vLLM : Inférence haute performance propulsée par PagedAttention

vLLM est un moteur d'inférence développé par l'équipe de l'UC Berkeley, et sa principale caractéristique est la vitesse.

L'innovation clé de vLLM estPagedAttention— une technique de gestion efficace de la mémoire GPU, inspirée par la pagination de la mémoire virtuelle des systèmes d'exploitation.

Dans les frameworks d'inférence traditionnels, le KV Cache (cache clé-valeur) de chaque requête doit occuper un espace mémoire GPU contigu.

Lorsque les requêtes ont des longueurs variables, la mémoire GPU devient fragmentée, ce qui entraîne une faible utilisation.

PagedAttention divise la mémoire GPU en pages de taille fixe. Le KV Cache de chaque requête peut être dispersé sur différentes pages, avec une table de pages qui enregistre leurs emplacements.

Cela améliore considérablement l'utilisation de la mémoire GPU et permet de servir plus de requêtes simultanément.

Exemple

# Installer vLLM
pip install vllm

# Démarrer un service API compatible OpenAI avec vLLM
# --model spécifie le modèle, --host et --port spécifient l'adresse d'écoute
# --tensor-parallel-size : nombre de GPU pour l'inférence parallèle
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192

Une fois le service démarré, vous pouvez l'appeler directement à l'aide du SDK OpenAI :

Exemple

# Chemin du fichier : test_vllm_client.py
from openai import OpenAI

# Se connecter au service vLLM local
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="example-demo-key" # vLLM ne nécessite pas de véritable clé API par défaut
)

# Appeler la complétion de chat
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "Vous êtes un assistant IA utile."},
{"role": "user", "content": "Présentez le tutoriel Python en une phrase."}
],
temperature=0.7,
max_tokens=500,
stream=True # Sortie en streaming
)

print("Réponse de l'IA :", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

vLLM permet aussi d'écrire des services personnalisés directement avec du code Python :

Exemple

# Chemin du fichier : custom_vllm_server.py
from vllm import LLM, SamplingParams

# Initialiser le modèle
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
gpu_memory_utilization=0.9, # Taux d'utilisation de la mémoire GPU
tensor_parallel_size=1, # Parallélisme tensoriel (nombre de GPU)
max_model_len=8192, # Longueur maximale du contexte
)

# Configurer les paramètres d'échantillonnage
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=500,
stop=["</s>"],
)

# Inférence par lots
prompts = [
"Présentez Python",
"Qu'est-ce que l'IA ?",
"Comment apprendre la programmation ?",
]

# Générer la réponse
outputs = llm.generate(prompts, sampling_params)

# Afficher les résultats
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Prompt : {prompt}")
print(f"Généré : {generated_text}")
print("-" * 50)

TGI : Moteur d'inférence de Hugging Face

TGI (Text Generation Inference) est un framework de service d'inférence lancé par Hugging Face.

Il dispose d'un écosystème complet, d'une intégration profonde avec Hugging Face Hub, et prend en charge des optimisations telles que Flash Attention et le batching dynamique.

Exemple

# Exécuter TGI avec Docker (la manière la plus simple)
# Remarque : Vous devez d'abord installer Docker et configurer la prise en charge du GPU
docker run -d \
--gpus all \
-p 8080:80 \
-v $PWD/data:/data \
--name tgi-server \
ghcr.io/huggingface/text-generation-inference:latest \
--model-id Qwen/Qwen2.5-7B-Instruct \
--max-input-length 4096 \
--max-total-tokens 8192 \
--max-batch-prefill-tokens 4096

# Ou installez directement avec Python (environnement de développement)
# pip install text-generation

Appel du service TGI :

Exemple

# Chemin du fichier : test_tgi_client.py
from text_generation import Client

# Se connecter au service TGI
client = Client("http://localhost:8080", timeout=60)

# Appel sans streaming
response = client.generate(
"Présentez le tutoriel Python",
max_new_tokens=500,
temperature=0.7,
top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# Appel en streaming
print("Sortie en streaming :")
for chunk in client.generate_stream(
"Écrivez un Hello World en Python",
max_new_tokens=200,
):
if not chunk.token.special:
print(chunk.token.text, end="", flush=True)
print()

Ollama : de la démo locale au déploiement en production

Ollama est connu pour « exécution locale de modèles en un clic », et beaucoup de personnes l'utilisent pour le développement et les démos.

Mais en fait, Ollama convient aussi à un déploiement en production à petite échelle—il est simple, stable et a une utilisation des ressources contrôlable.

Exemple

# Installer Ollama (serveur Linux)
curl -fsSL https://ollama.com/install.sh | sh

# Configurer Ollama pour écouter sur toutes les interfaces réseau (faites attention au pare-feu en production)
# Modifier le fichier de service systemd
sudo systemctl edit ollama.service

# Ajoutez le contenu suivant :
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# Redémarrer le service
sudo systemctl restart ollama

# Récupérer et exécuter le modèle
ollama pull qwen2.5:7b

Appeler Ollama via l'API :

Exemple

# Chemin du fichier : test_ollama_client.py
import requests
import json

# Adresse de l'API Ollama
base_url = "http://localhost:11434/api"

# Interface de chat (streaming)
def chat_stream(model, messages):
url = f"{base_url}/chat"
data = {
"model": model,
"messages": messages,
"stream": True,
"options": {
"temperature": 0.7,
"num_ctx": 8192,
}
}

response = requests.post(url, json=data, stream=True)
full_response = ""

for line in response.iter_lines():
if line:
chunk = json.loads(line)
if "message" in chunk:
content = chunk["message"].get("content", "")
full_response += content
print(content, end="", flush=True)
if chunk.get("done", False):
break

print()
return full_response

# Exemple d'appel
messages = [
{"role": "system", "content": "Vous êtes un assistant IA utile."},
{"role": "user", "content": "Présentez le tutoriel Python"}
]

print("Réponse de l'IA :", end="")
result = chat_stream("qwen2.5:7b", messages)

Comparaison des frameworks

Les trois frameworks ont chacun leurs propres caractéristiques. Comment choisir ? Cela dépend de vos besoins.

Framework	Avantages	Inconvénients	Scénarios applicables
vLLM	Performances les plus élevées, débit maximal	Configuration relativement complexe	Environnements de production à forte concurrence, à la recherche de performance
TGI	Écosystème Hugging Face, déploiement simple	Performances légèrement inférieures à vLLM	Utilise déjà la chaîne d'outils Hugging Face
Ollama	Déploiement minimal, faible coût de maintenance	Ne convient pas à une concurrence extrêmement élevée	Services de petite et moyenne envergure, lancement rapide

Phase initiale : commencez avec Ollama, simple et fiable. Lorsque le trafic augmente : migrez vers vLLM, en privilégiant les performances.

Techniques d'optimisation de l'inférence

Le même modèle, exécuté de différentes manières, peut varier de plusieurs fois en vitesse et en coût.

Cette section présente les techniques d'optimisation de l'inférence couramment utilisées dans les environnements de production.

Traitement par lots continu

Le traitement par lots traditionnel est statique : rassembler suffisamment de requêtes, exécuter l'inférence ensemble, et ne traiter le lot suivant qu'une fois toutes terminées.

Le problème est le suivant : certaines requêtes sont courtes (par ex., ne génèrent que 10 caractères), et d'autres sont longues (génèrent 500 caractères).

Les requêtes courtes doivent attendre que les requêtes longues se terminent avant que les ressources puissent être libérées ensemble, de sorte que l'utilisation du GPU ne peut pas augmenter.

Traitement par lots continuIl est dynamique : lorsqu'une requête se termine, de nouvelles requêtes de la file d'attente sont immédiatement ajoutées, sans attendre que l'ensemble du lot soit terminé.

De cette façon, le GPU est presque toujours à pleine charge, ce qui augmente considérablement le débit.

vLLM et TGI disposent tous deux d'un traitement par lots continu intégré. Vous n'avez pas besoin de l'implémenter manuellement ; il suffit de l'activer.

Décodage spéculatif

Générer un token avec un grand modèle nécessite une passe avant complète, ce qui est lent.

L'idée du décodage spéculatif est :Utiliser un petit modèle pour « deviner » les quelques tokens suivants, puis demander au grand modèle de les vérifier tous en une seule fois.。

Le petit modèle devine rapidement, et bien qu'il puisse ne pas être entièrement correct, la partie correctement devinée peut être produite par lots.

Les suppositions incorrectes sont écartées et devinées à nouveau.

Cela peut améliorer la vitesse globale de 2-3x tout en conservant une qualité de sortie inchangée.

Exemple

# Activer le décodage spéculatif dans vLLM
# --speculative-model spécifie le petit modèle (modèle brouillon)
# --num-speculative-tokens spécifie le nombre de tokens à deviner à chaque fois
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--speculative-model Qwen/Qwen2.5-0.5B-Instruct \
--num-speculative-tokens 5 \
--host 0.0.0.0 \
--port 8000

Inférence quantifiée : AWQ et GPTQ

La quantification de modèle compresse les poids du modèle de nombres à virgule flottante 16 bits (FP16) vers des entiers de 8 bits ou 4 bits.

L'utilisation de la mémoire est réduite de moitié ou même moins, et l'inférence peut être plus rapide.

Les deux méthodes de quantification les plus couramment utilisées sont AWQ et GPTQ.

Méthode de quantification	Fonctionnalité	Perte de précision	Vitesse d'inférence
AWQ	Quantification des poids tenant compte des activations	Minimal	Rapide
GPTQ	Quantification optimisée couche par couche	petit	Rapide
GGUF	Adapté au déploiement local	Acceptable	Rapide

Exemple

# Exécuter un modèle quantifié AWQ directement avec vLLM
# De nombreux modèles quantifiés peuvent être téléchargés directement depuis Hugging Face
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Qwen2.5-7B-Instruct-AWQ \
--quantization awq \
--host 0.0.0.0 \
--port 8000

# Ou exécuter des modèles quantifiés GGUF avec Ollama
ollama run qwen2.5:7b-instruct-q4_0

Charger un modèle quantifié AWQ en Python :

Exemple

# Chemin du fichier : awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# Charger le modèle quantifié AWQ
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# Remarque : la bibliothèque autoawq doit être installée pour exécuter ceci
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)

# Inférence
messages = [
{"role": "system", "content": "Vous êtes un assistant IA utile."},
{"role": "user", "content": "Présentez Python"}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.7,
)

response = tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
print(response)

Distillation de modèle

La distillation utilise un grand modèle (enseignant) pour entraîner un petit modèle (élève), permettant au petit modèle d'acquérir les capacités du grand modèle.

Le petit modèle est plus rapide et moins coûteux, et bien que ses performances soient légèrement inférieures, il est suffisant dans de nombreux scénarios.

Par exemple, utiliser Qwen2.5-72B comme enseignant pour entraîner un élève Qwen2.5-7B peut donner des résultats proches du modèle 72B, mais avec une vitesse et un coût au niveau du 7B.

Les avantages de la distillation sont généralement les suivants :Taille du modèle réduite à 1/10, vitesse d'inférence améliorée de 5-10x, avec plus de 90% des performances conservées.。

Conception de service API

Envelopper le modèle n'est que la première étape ; l'API de production a aussi besoin de nombreuses fonctionnalités : authentification, limitation de débit, réponses en streaming et gestion des erreurs.

FastAPI est un excellent choix pour cela.

FastAPI : encapsulation des services LLM

Écrivons un service API complet avec authentification, limitation de débit et réponses en streaming.

Exemple

# Chemin du fichier : llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. Initialiser l'application et la configuration
# ============================================

app = FastAPI(title="EXAMPLE LLM API", version="1.0.0")

# Authentification simple par clé API (une solution plus robuste est recommandée pour la production)
VALID_API_KEYS = {
"sk-example-123456": {"user": "demo_user", "quota": 1000},
"sk-example-789012": {"user": "pro_user", "quota": 10000},
}

# Limiteur de débit simple (implémentation en mémoire ; Redis est recommandé pour la production)
class RateLimiter:
def __init__(self):
self.requests = defaultdict(list) # clé -> [horodatage, ...]

async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
"""Vérifie si la limite de débit est dépassée"""
now = time.time()
# Nettoyer les enregistrements expirés
self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
# Vérifier si la limite est dépassée
if len(self.requests[key]) >= max_requests:
return False
# Enregistrer cette requête
self.requests[key].append(now)
return True

rate_limiter = RateLimiter()

# ============================================
# 2. Modèles de données
# ============================================

class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"

class Message(BaseModel):
role: Role = Field(..., description="Rôle du message")
content: str = Field(..., description="Contenu du message")

class ChatCompletionRequest(BaseModel):
model: str = Field(..., description="Nom du modèle")
messages: List[Message] = Field(..., description="Liste des messages de la conversation")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="Température d'échantillonnage")
top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Échantillonnage Top-P")
max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="Nombre maximal de tokens générés")
stream: Optional[bool] = Field(default=False, description="Indique si la sortie doit être diffusée en flux")

class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[dict]
usage: dict

# ============================================
# 3. Dépendance d'authentification
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Valider la clé API et renvoyer les informations de l'utilisateur"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Clé API invalide"
)
return VALID_API_KEYS[api_key]

# ============================================
# 4. Simuler la génération du LLM (remplacer par de véritables appels au modèle dans des projets réels)
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
"""Simuler le processus de génération du LLM"""
# Ceci n'est qu'une simulation ; dans un projet réel, appelez le modèle réel
response_text = (
"Bonjour ! Je suis l'assistant IA d'EXAMPLE.\n\n"
"EXAMPLE propose une multitude de tutoriels de programmation, notamment Python, Java, C++, le développement front-end, et plus encore.\n"
"Vous pouvez visiter https://www.example.com pour en savoir plus !\n\n"
"Comment puis-je vous aider ?"
)
# Simuler un délai de génération
await asyncio.sleep(0.5)
return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
"""Simuler la génération en streaming du LLM"""
response_text = await mock_llm_generate(messages, max_tokens)
# Afficher caractère par caractère pour simuler un effet de streaming
for char in response_text:
yield char
await asyncio.sleep(0.03) # Délai de 30ms par caractère

# ============================================
# 5. Points de terminaison de l'API
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
request: ChatCompletionRequest,
user: dict = Depends(get_current_user)
):
"""Point de terminaison de complétion de chat, compatible avec le format OpenAI"""

# Vérification de la limitation de débit : chaque utilisateur peut effectuer au maximum 60 requêtes par minute
user_key = user["user"]
if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Limite de débit dépassée"
)

# Générer l'ID de requête
request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(datetime.now().timestamp())

if request.stream:
# Réponse en streaming
async def stream_generator():
async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
# Construire un bloc au format SSE
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": created,
"model": request.model,
"choices": [{
"index": 0,
"delta": {"content": token},
"finish_reason": None
}]
}
yield f"data: {__import__('json').dumps(chunk)}\n\n"
# Envoyer le marqueur de fin
yield "data: [DONE]\n\n"

return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)

else:
# Réponse sans streaming
response_text = await mock_llm_generate(request.messages, request.max_tokens)

return ChatCompletionResponse(
id=request_id,
created=created,
model=request.model,
choices=[{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}],
usage={
"prompt_tokens": sum(len(m.content) for m in request.messages),
"completion_tokens": len(response_text),
"total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
}
)

@app.get("/health")
async def health_check():
"""Point de terminaison de vérification d'état"""
return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
"""Lister les modèles disponibles"""
return {
"object": "list",
"data": [
{
"id": "qwen2.5-7b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "exemple"
},
{
"id": "llama-3.1-8b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "exemple"
}
]
}

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

Démarrer le service :

Exemple

# Installer les dépendances
pip install fastapi uvicorn python-multipart pydantic

# Démarrer le service
python llm_api_server.py

# Tester dans un autre terminal
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-example-123456" \
-d '{
"model": "qwen2.5-7b-instruct",
"messages": [
{"role": "user", "content": "Bonjour"}
],
"stream": false
}'

Appel avec le client Python :

Exemple

# Chemin du fichier : client_example.py
from openai import OpenAI

# Se connecter à notre propre service API
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="sk-example-123456"
)

# Tester le mode non-streaming
print("Appel non-streaming :")
response = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Présentez Python"}],
stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# Test du streaming
print("Appel streaming :")
stream = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Présentez Python"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

Avantages des interfaces compatibles avec OpenAI

Rendre votre API compatible avec OpenAI présente plusieurs avantages évidents :

Les utilisateurs n'ont pas besoin de modifier le code, il suffit de changer base_url et api_key.

Écosystème riche — tout SDK, outil ou application qui prend en charge OpenAI peut directement utiliser votre service.

Pas besoin de concevoir votre propre format d'API ; la conception d'OpenAI est déjà bien pensée.

Authentification et limitation de débit dans les environnements de production

Les exemples précédents utilisaient une limitation de débit simple en mémoire ; pour la production, il est recommandé d'utiliser Redis pour une limitation de débit distribuée.

Exemple

# Chemin du fichier : redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
"""Limiteur de débit distribué basé sur Redis"""

def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)

async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int
) -> bool:
"""
Algorithme de limitation de débit à fenêtre glissante
key : clé de limitation de débit (par ex., user_id ou api_key)
max_requests: nombre maximal de requêtes dans la fenêtre
window_seconds: taille de la fenêtre (secondes)
"""
now = time.time()
window_start = now - window_seconds

# Clé Redis
redis_key = f"rate_limit:{key}"

# Utiliser les opérations atomiques du pipeline
async with self.redis.pipeline() as pipe:
# 1. Supprimer les enregistrements en dehors de la fenêtre
pipe.zremrangebyscore(redis_key, 0, window_start)
# 2. Compter les requêtes dans la fenêtre actuelle
pipe.zcard(redis_key)
# 3. Ajouter la requête actuelle
pipe.zadd(redis_key, {str(now): now})
# 4. Définir le temps d'expiration
pipe.expire(redis_key, window_seconds)
# Exécuter
_, current_count, _, _ = await pipe.execute()

return current_count < max_requests

async def close(self):
await self.redis.close()

# Exemple d'utilisation
async def main():
limiter = RedisRateLimiter("redis://localhost:6379")

# Test : même utilisateur au maximum 10 requêtes par minute
user_id = "user_123"
for i in range(15):
allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
print(f"Request {i+1}: {'allowed' if allowed else 'denied'}")

await limiter.close()

if __name__ == "__main__":
import asyncio
asyncio.run(main())

L'authentification et la limitation de débit constituent l'infrastructure d'une API ; mieux vaut les rendre un peu plus complexes que de le regretter après avoir été submergé.

Équilibrage de charge et mise à l'échelle

Lorsqu'il y a de plus en plus d'utilisateurs et qu'un seul serveur ne peut pas suivre, vous devez ajouter des machines et répartir le trafic sur plusieurs machines.

C'est le problème que l'équilibrage de charge et la mise à l'échelle automatique doivent résoudre.

Conception de service sans état

Tout d'abord, rappelez-vous :les services d'IA doivent être conçus pour être sans état。

Sans état signifie : les requêtes ne partagent pas de mémoire, et n'importe quel serveur peut traiter n'importe quelle requête.

Cela simplifie la mise à l'échelle : ajoutez simplement des machines, pas besoin de se soucier de la synchronisation des données.

L'état de session (comme l'historique de conversation) peut être stocké côté client (demandez à l'utilisateur de l'envoyer à chaque fois), ou dans Redis.

Proxy inverse Nginx

Nginx est le répartiteur de charge le plus couramment utilisé, distribuant le trafic vers plusieurs serveurs backend.

Exemple

# Chemin du fichier : nginx.conf
# Configuration Nginx : équilibrage de charge pour plusieurs services LLM

upstream llm_backends {
# Algorithme des connexions les moins nombreuses : envoyer les requêtes au serveur ayant le moins de connexions en cours
least_conn;

# Liste des serveurs backend
server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

# Si vous avez plusieurs modèles de GPU, vous pouvez allouer différentes proportions de trafic selon le poids
# weight=2 signifie que ce serveur reçoit deux fois plus de trafic que les autres
}

server {
listen 80;
server_name api.example.com;

# Transfert des requêtes API
location /v1/ {
# Redirection vers le backend
proxy_pass http://llm_backends;

# Paramètres de délai d'attente (la génération IA peut être lente)
proxy_connect_timeout 10s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;

# Transmettre l'adresse IP réelle du client
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# Paramètres requis pour les réponses en streaming
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
}

# Point de terminaison de vérification de l'état
location /health {
proxy_pass http://llm_backends/health;
proxy_connect_timeout 5s;
proxy_read_timeout 5s;
}

# Limitation de débit : max 10 requêtes par seconde par IP
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
}

Mise à l'échelle automatique HPA Kubernetes

Si vous déployez avec Kubernetes, vous pouvez utiliser HPA (Horizontal Pod Autoscaler) pour mettre à l'échelle automatiquement à la hausse et à la baisse en fonction de la charge.

Exemple

# Chemin du fichier : llm-deployment.yaml
# Configuration de déploiement Kubernetes pour le service LLM

# 1. Déploiement : déployer le pod LLM
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-server
namespace: example
spec:
replicas: 3 # Nombre initial de réplicas
selector:
matchLabels:
app: llm-server
template:
metadata:
labels:
app: llm-server
spec:
containers:
- name: llm-server
image: example/llm-server:latest
ports:
- containerPort: 8000
ressources:
requests:
cpu: "4"
memory: "16Gi"
nvidia.com/gpu: 1 # Requête 1 GPU
limits:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: 1 # Limite 1 GPU
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300 # Le chargement du modèle prend du temps
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300
periodSeconds: 10
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"

---
# 2. Service : équilibrage de charge Service
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: example
spec:
selector:
app: llm-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

---
# 3. HPA : mise à l'échelle automatique
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-hpa
namespace: example
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-server
minReplicas: 3 # Minimum 3 Pods
maxReplicas: 20 # Maximum 20 Pods
metrics:
# Mettre à l'échelle en fonction de l'utilisation du CPU
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# Mettre à l'échelle en fonction de l'utilisation du GPU (nécessite des métriques personnalisées)
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: 80
# Mettre à l'échelle en fonction de la longueur de la file d'attente (métrique personnalisée)
- type: Pods
pods:
metric:
name: pending_requests
target:
type: AverageValue
averageValue: 10
behavior:
scaleUp:
stabilizationWindowSeconds: 60 # Observer pendant 60 secondes avant d'augmenter l'échelle
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # Observer pendant 5 minutes avant de réduire l'échelle
policies:
- type: Percent
value: 20
periodSeconds: 120

Gestion du pool de ressources GPU

Les GPU sont coûteux, il faut donc maximiser leur utilisation.

Une stratégie courante consiste à regrouper par type de modèle, en plaçant les Pods du même modèle ensemble.

Par exemple :

Pool de ressources	Modèle GPU	Modèle en cours d'exécution	Objectif
small-pool	RTX 3090 / L4	Modèle 7B	Trafic quotidien
medium-pool	A10 / L40	Modèle 13B-70B	Tâches complexes
large-pool	A100 / H100	Modèle 70B+	Utilisateurs à forte valeur
spot-pool	Divers types de machines	Traitement par lots hors ligne	Utiliser des instances Spot à bas prix

Test A/B de modèles d'IA

Vous avez publié une nouvelle version du modèle. Comment savoir si elle est meilleure que l'ancienne ? Utilisez le test A/B.

Le cœur du test A/B est :Répartir le trafic en groupes, utiliser différents modèles pour différents groupes et comparer les métriques。

Stratégie de répartition du trafic

Il existe plusieurs méthodes courantes de répartition du trafic :

Exemple

# Chemin du fichier : ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
"""Une variante du test A/B"""
name: str # Nom de la variante
model_name: str # Nom du modèle
weight: int # Poids du trafic
config: dict # Autre configuration

class ABTestRouter:
"""Routeur de trafic de test A/B"""

def __init__(self, test_name: str, variants: List[Variante]):
self.test_name = test_name
self.variants = variants
# Calculer le poids total
total_weight = sum(v.weight for v in variants)
# Construire les plages de segments
self.ranges: List[Tuple[int, int, Variante]] = []
current = 0
for v in variants:
self.ranges.append((current, current + v.weight, v))
current += v.weight
self.total_weight = total_weight

def get_variant_by_user(self, user_id: str) -> Variant:
"""
Répartition stable du trafic basée sur l'ID utilisateur
Le même utilisateur est toujours affecté au même groupe pour éviter une expérience incohérente
"""
# Calculer le score de l'utilisateur à l'aide d'un hachage
hash_input = f"{self.test_name}-{user_id}".encode()
hash_value = hashlib.md5(hash_input).hexdigest()
# Convertir en un nombre de 0-999
score = int(hash_value[:4], 16) % 1000
# Trouver la variante correspondante
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

def get_variant_random(self) -> Variant:
"""Répartition aléatoire (adaptée aux scénarios qui ne nécessitent pas de persistance)"""
score = random.randint(0, self.total_weight - 1)
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

# Exemple d'utilisation
def main():
# Définir trois variantes
variants = [
Variante(
name="control",
model_name="Qwen2.5-7B-Instruct",
weight=50, # 50% du trafic
config={"temperature": 0.7}
),
Variante(
name="treatment-v1",
model_name="Qwen2.5-14B-Instruct",
weight=30, # 30% du trafic
config={"temperature": 0.7}
),
Variante(
name="treatment-v2",
model_name="Llama-3.1-8B-Instruct",
weight=20, # 20% du trafic
config={"temperature": 0.8}
)
]

router = ABTestRouter("model-comparison-v1", variants)

# Test de répartition des utilisateurs
test_users = [f"user_{i}" for i in range(10)]
for user_id in test_users:
variant = router.get_variant_by_user(user_id)
print(f"Utilisateur {user_id} → Groupe {variant.name} → Modèle {variant.model_name}")

# Statistiques de distribution
count: Dict[str, int] = {}
for i in range(10000):
variant = router.get_variant_random()
count[variant.name] = count.get(variant.name, 0) + 1
print("\nStatistiques de répartition aléatoire (10000 fois):")
for name, cnt in count.items():
print(f" {name}: {cnt} fois ({cnt/100}%)")

if __name__ == "__main__":
main()

Définition et collecte des métriques

Les tests A/B ne peuvent pas simplement regarder « quel modèle répond plus intelligemment » ; il faut définir des métriques quantifiables.

Types de métriques	Métriques spécifiques	Description
Métriques système	Latence, débit, taux d'erreur	Si le modèle est rapide et stable
Comportement des utilisateurs	Satisfaction des utilisateurs, nombre de tours de dialogue, taux de renouvellement	Si les utilisateurs l'aiment
Indicateurs métier	Taux de conversion, taux de rétention, revenus	Si cela aide l'entreprise
Indicateurs de qualité	Évaluations humaines, taux de nocivité, exactitude factuelle	Si la qualité de sortie est bonne

Exemple

# Chemin du fichier : metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
"""Métriques complètes pour les requêtes de chat"""
# Informations de base
request_id: str
user_id: str
variant_name: str # Nom du groupe de test A/B
model_name: str

# Métriques de temps
timestamp: str
ttft: float # Temps jusqu'au premier token
total_time: float # Temps total écoulé

# Métriques de génération
prompt_tokens: int
completion_tokens: int
total_tokens: int

# Métriques métier
user_rating: Optional[int] = None # Note de l'utilisateur (1-5)
is_error: bool = False
error_message: Optional[str] = None

class MetricsCollector:
"""Collecteur de métriques"""

def __init__(self):
# Dans un projet réel, cela devrait être écrit dans Kafka ou une base de données
self.buffer = []

def collect(self, metrics: ChatMetrics):
"""Collecter une métrique"""
self.buffer.append(asdict(metrics))
# Sortie simple ; dans un projet réel, écrire dans le stockage
print(f"[Métriques] {json.dumps(asdict(metrics), ensure_ascii=False)}")

# Lorsque le tampon est plein, écrire par lots
if len(self.buffer) >= 100:
self.flush()

def flush(self):
"""Écriture par lots dans le stockage"""
if self.buffer:
# Dans un projet réel : écrire dans ClickHouse, BigQuery, etc.
print(f"{len(self.buffer)} métriques vidées")
self.buffer = []

# Exemple d'utilisation
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
"""Traiter une requête de chat et enregistrer les métriques"""
request_id = str(uuid.uuid4())

# Répartir le trafic
variant = router.get_variant_by_user(user_id)

# Enregistrer l'heure de début
start_time = time.time()
ttft = 0.0
prompt_tokens = len(prompt)
completion_tokens = 0

try:
# Remplacer ceci par un véritable appel au modèle
# response = call_model(variant.model_name, prompt)
response = "Ceci est la réponse de l'IA"
completion_tokens = len(response)

# Simuler le TTFT (supposer que le premier token est reçu après 0.2 seconds)
ttft = 0.2

# Enregistrer les métriques de succès
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
is_error=False
)
collector.collect(metrics)
return response, request_id

except Exception as e:
# Enregistrer les métriques d'erreur
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=0,
total_tokens=prompt_tokens,
is_error=True,
error_message=str(e)
)
collector.collect(metrics)
raise

# Retour utilisateur
def submit_user_feedback(request_id: str, rating: int):
"""L'utilisateur soumet un retour d'évaluation"""
# Dans un projet réel : mettre à jour l'enregistrement correspondant dans la base de données
print(f"La requête {request_id} a reçu la note {rating}")

Évaluation de la signification statistique

Après avoir obtenu les données, vous ne pouvez pas simplement regarder 52% pour le groupe A et 55% pour le groupe B et conclure que le groupe B est meilleur.

Un test de signification statistique doit être effectué pour confirmer que la différence n'est pas due à une fluctuation aléatoire.

Exemple

# Chemin du fichier : ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
control_conv: float,
treatment_conv: float,
control_total: int,
treatment_total: int
) -> Tuple[float, float]:
"""
Calculer la valeur p du test Z
control_conv: taux de conversion du groupe contrôle
treatment_conv: taux de conversion du groupe de traitement
control_total: taille de l'échantillon du groupe contrôle
treatment_total: taille de l'échantillon du groupe de traitement
"""
# taux de conversion combiné
p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
# erreur standard
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
# statistique Z
z_score = (treatment_conv - control_conv) / se_pool
# calculer la valeur p (test bilatéral)
p_value = 2 * (1 - normal_cdf(abs(z_score)))
return z_score, p_value

def normal_cdf(x: float) -> float:
"""Fonction de répartition de la loi normale standard (approximative)"""
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
control_conversions: int,
control_total: int,
treatment_conversions: int,
treatment_total: int,
alpha: float = 0.05
):
"""Analyser les résultats du test A/B"""
control_conv = control_conversions / control_total
treatment_conv = treatment_conversions / treatment_total

z_score, p_value = calculate_z_test(
control_conv, treatment_conv,
control_total, treatment_total
)

relative_lift = (treatment_conv - control_conv) / control_conv

⟦⟧"Analyse des résultats du test A/B"("A/B test results analysis")
print("=" * 50)
print(f"Groupe de contrôle : {control_conversions}/{control_total} ({control_conv:.2%})")
print(f"Groupe de traitement : {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
print(f"Lift relatif : {relative_lift:+.2%}")
print("-" * 50)
print(f"Statistique Z : {z_score:.4f}")
print(f"valeur p : {p_value:.4f}")
print("-" * 50)

if p_value < alpha:
print(f"✅ Le résultat est statistiquement significatif (p < {alpha})")
if relative_lift > 0:
print(" Le groupe de traitement obtient de meilleurs résultats !")
else:
print(" Le groupe de contrôle obtient de meilleurs résultats !")
else:
print(f"❌ Le résultat n'est pas significatif (p >= {alpha})")
print(" Impossible de déterminer quelle version est meilleure ; davantage de données sont nécessaires")

# Exemple : en supposant les résultats d'un test A/B
if __name__ == "__main__":
# Scénario 1 : différence significative
print("Scénario 1 : différence significative")
analyze_ab_test(
control_conversions=450, # Groupe de contrôle : 450 satisfaits
control_total=1000, # Total du groupe témoin : 1000 personnes
treatment_conversions=520, # Groupe de traitement : 520 satisfaits
treatment_total=1000 # Total du groupe de traitement : 1000 personnes
)

print("\n" + "="*50 + "\n")

# Scénario 2 : aucune différence significative
print("Scénario 2 : aucune différence significative")
analyze_ab_test(
control_conversions=48,
control_total=100,
treatment_conversions=52,
treatment_total=100
)

La signification statistique est importante. Une "amélioration" non testée pourrait n'être qu'une fluctuation aléatoire.

Suivi des expériences MLflow

Vous avez entraîné des dizaines de versions de modèle. Laquelle est la plus performante ? Quelle version est déployée ?

MLflow vous aide à gérer les expériences, suivre les modèles et enregistrer les métriques.

Gestion des expériences

Exemple

# Chemin de fichier : mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# Définir l'URI de suivi MLflow (serveur local ou distant)
mlflow.set_tracking_uri("http://localhost:5000")

# Créer ou définir une expérience
experiment_name = "example-llm-finetuning"
try:
experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# Simuler une expérience de fine-tuning d'un LLM
def run_finetuning_experiment(
model_name: str,
learning_rate: float,
batch_size: int,
num_epochs: int,
lora_rank: int,
dataset_name: str = "example-demo"
):
"""Exécuter une expérience de fine-tuning et la consigner dans MLflow"""

with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
run_id = run.info.run_id

# Enregistrer les paramètres
mlflow.log_param("model_name", model_name)
mlflow.log_param("learning_rate", learning_rate)
mlflow.log_param("batch_size", batch_size)
mlflow.log_param("num_epochs", num_epochs)
mlflow.log_param("lora_rank", lora_rank)
mlflow.log_param("dataset_name", dataset_name)

# Simuler le processus d'entraînement et enregistrer les métriques
print(f"Démarrage de l'expérience d'entraînement {run_id}...")

for epoch in range(num_epochs):
# Simuler la perte d'entraînement (diminue avec les époques)
train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

# Enregistrer les métriques
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("val_loss", val_loss, step=epoch)
mlflow.log_metric("epoch", epoch, step=epoch)

print(f"Époque {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
time.sleep(0.5)

# Métriques finales d'évaluation
final_val_loss = 1.2 + np.random.normal(0, 0.1)
final_accuracy = 0.85 + np.random.normal(0, 0.03)
final_f1 = 0.83 + np.random.normal(0, 0.04)

mlflow.log_metric("final_val_loss", final_val_loss)
mlflow.log_metric("final_accuracy", final_accuracy)
mlflow.log_metric("final_f1", final_f1)

# Enregistrer les fichiers (par ex., journaux d'entraînement, exemples de sorties)
with open("training_log.txt", "w") as f:
f.write(f"Expérience run_id : {run_id}\n")
f.write(f"Modèle : {model_name}\n")
f.write(f"Exactitude finale : {final_accuracy:.4f}\n")

mlflow.log_artifact("training_log.txt")

# Enregistrer le modèle (simulé avec sklearn ici ; dans de vrais projets, enregistrez le LLM réel)
# Dans de vrais projets, vous pouvez utiliser mlflow.transformers.log_model
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

mlflow.sklearn.log_model(
model,
"model",
registered_model_name=f"{model_name.replace('/', '-')}-demo"
)

print(f"Expérience terminée ! run_id : {run_id}")
print(f"Exactitude finale : {final_accuracy:.4f}")

return run_id, final_accuracy

# Exécuter plusieurs expériences
if __name__ == "__main__":
print("Démarrer l'interface MLflow (exécuter dans un autre terminal) :")
print(" mlflow ui --port 5000")
print("\nLancement des expériences...\n")

# Expérience 1 : Configuration de base
run_id1, acc1 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()

# Expérience 2 : Rang LoRA plus grand
run_id2, acc2 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=16
)

print()

# Expérience 3 : Taux d'apprentissage plus élevé
run_id3, acc3 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=5e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()
print("Toutes les expériences sont terminées !")
print(f"Expérience 1 (run_id={run_id1}): {acc1:.4f}")
print(f"Expérience 2 (run_id={run_id2}): {acc2:.4f}")
print(f"Expérience 3 (run_id={run_id3}): {acc3:.4f}")
print("\n"Voir les résultats : http://localhost:5000")

Registre de modèles

MLflow Model Registry gère le cycle de vie du modèle : de Staging (test) à Production.

Exemple

# Chemin du fichier : mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
model_name: str,
version: int,
description: str = ""
):
"""Promouvoir le modèle vers l'étape Production"""
# Mettre à jour la description de la version
client.update_model_version(
name=model_name,
version=version,
description=description
)

# Définir cette version comme Production
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
archive_existing_versions=True # Archiver les anciennes versions
)
print(f"Le modèle {model_name} version {version} a été promu en Production")

def get_production_model(model_name: str):
"""Obtenir le modèle Production actuel"""
versions = client.get_latest_versions(model_name, stages=["Production"])
if versions:
return versions[0]
return None

def list_all_models():
"""Lister tous les modèles enregistrés"""
models = client.search_registered_models()
print("Modèles enregistrés :")
for model in models:
print(f" - {model.name}")
# Lister toutes les versions de ce modèle
versions = client.search_model_versions(f"name='{model.name}'")
for v in versions:
print(f" Version {v.version}: {v.current_stage}")

# Exemple d'utilisation
if __name__ == "__main__":
model_name = "Qwen2.5-7B-Instruct-demo"

print("Lister tous les modèles :")
list_all_models()

print("\n" + "="*50)
# Supposons que nous voulions promouvoir la version 1 en production
# promote_model_to_production(
# model_name=model_name,
# version=1,
# description="Modèle d'environnement de production : version optimisée 2024"
# )

print("\nObtenir le modèle de production actuel : ")
prod_model = get_production_model(model_name)
if prod_model:
print(f" Nom du modèle : {prod_model.name}")
print(f" Version : {prod_model.version}")
print(f" Étape : {prod_model.current_stage}")
print(f" ID d'exécution : {prod_model.run_id}")
else:
print(" Aucun modèle de production pour l'instant")

Surveillance et alertes

Une fois le service en production, comment savez-vous qu'il fonctionne correctement ? Les utilisateurs rencontreront-ils des délais d'attente ? Le modèle deviendra-t-il de plus en plus imprécis ?

Vous avez besoin d'un système de surveillance complet.

Indicateurs clés : TTFT, TPS, taux d'erreur

Les services d'IA ont plusieurs indicateurs de surveillance essentiels :

Métrique	Nom complet	Signification	Exemple de seuil d'alerte
TTFT	Temps jusqu'au premier token	Temps entre la requête de l'utilisateur et la réception du premier token	Alerte > 3 secondes
TPOT	Temps par token de sortie	Temps moyen par token de sortie	Alerte > 100ms
TPS	Tokens par seconde	Nombre de tokens générés par seconde	Alerte < 100
Taux d'erreur	Taux d'erreur	Proportion des requêtes ayant échoué	Alerte > 5%
Utilisation du GPU	Utilisation du GPU	Taux d'utilisation du GPU	Alerte < 30% ou > 95%
Utilisation de la VRAM	Mémoire GPU	Taux d'occupation de la VRAM du GPU	Alerte > 90%

Prometheus + Grafana

Prometheus collecte les métriques, Grafana affiche les tableaux de bord - la combinaison gagnante pour la supervision.

Exemple

# Chemin du fichier : metrics_prometheus.py
from prometheus_client import (
Counter, Histogram, Gauge,
start_http_server, generate_latest
)
import time
import random

# ============================================
# Définir les métriques
# ============================================

# Counter : un compteur qui ne fait qu'augmenter
REQUEST_COUNT = Counter(
"llm_requests_total",
"Nombre total de requêtes LLM",
["model", "variant", "status"] # Étiquettes
)

# Histogramme : distribution statistique (par ex., latence)
REQUEST_LATENCY = Histogramme(
"llm_request_latency_seconds",
"Latence des requêtes LLM",
["model", "variant"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60] # Compartiments
)

TTFT_HISTOGRAM = Histogramme(
"llm_ttft_seconds",
"Temps jusqu'au premier token",
["model", "variant"],
buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Jauge : une jauge qui peut monter ou descendre
ACTIVE_CONNECTIONS = Jauge(
"llm_active_connections",
"Nombre de connexions actives",
["model"]
)

GPU_UTILIZATION = Gauge(
"llm_gpu_utilization_percent",
"Utilisation du GPU",
["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
"llm_gpu_memory_used_bytes",
"Mémoire GPU utilisée",
["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
"llm_tokens_per_second",
"Tokens générés par seconde",
["model"]
)

# ============================================
# Simuler le service
# ============================================

def process_request(model: str, variant: str):
"""Simuler le traitement d'une requête et enregistrer les métriques"""
start_time = time.time()

# Connexions actives +1
ACTIVE_CONNECTIONS.labels(model=model).inc()

try:
# Simuler le TTFT (temps jusqu'au premier token)
ttft = random.uniform(0.05, 0.3)
TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

# Simuler le temps de traitement
processing_time = random.uniform(0.5, 3.0)
time.sleep(processing_time)

# Simuler le succès
REQUEST_COUNT.labels(
model=model, variant=variant, status="success"
).inc()

# Enregistrer la latence
latency = time.time() - start_time
REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

# Simuler le TPS
tps = random.uniform(50, 200)
TOKENS_PER_SECOND.labels(model=model).set(tps)

except Exception:
# Simuler une erreur
REQUEST_COUNT.labels(
model=model, variant=variant, status="error"
).inc()
raise
finally:
# Connexions actives -1
ACTIVE_CONNECTIONS.labels(model=model).dec()

def update_gpu_metrics():
"""Simuler la mise à jour des métriques GPU"""
# Simuler les métriques pour le GPU 0
gpu0_util = random.uniform(40, 90)
gpu0_mem = random.uniform(10, 20) * 1024**3 # 10-20 GB
GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)

# ============================================
# Démarrer le service
# ============================================

if __name__ == "__main__":
# Démarrer le point de terminaison des métriques Prometheus
start_http_server(8081)
print("Serveur de métriques Prometheus démarré sur le port 8081")
print("Métriques disponibles à l'adresse http://localhost:8081/metrics")

# Simuler le trafic
models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
variants = ["control", "treatment-v1"]

while True:
model = random.choice(models)
variant = random.choice(variants)
process_request(model, variant)
update_gpu_metrics()
time.sleep(random.uniform(0.1, 0.5))

Fichier de configuration Prometheus :

Exemple

# Chemin du fichier : prometheus.yml
global:
scrape_interval: 15s

scrape_configs:
# Collecter les métriques du service LLM
- job_name: "llm-service"
static_configs:
- targets: ["localhost:8081"]
labels:
service: "llm-api"
environment: "production"

# Récupérer les métriques des nœuds (GPU, CPU, mémoire)
- job_name: "node"
static_configs:
- targets: ["localhost:9100"]

alerting:
alertmanagers:
- static_configs:
- targets: ["localhost:9093"]

# Règles d'alerte
rule_files:
- "alerts.yml"

Règles d'alerte :

Exemple

# Chemin du fichier : alerts.yml
groups:
- name: llm_alerts
interval: 30s
rules:
# Alerte de taux d'erreur élevé
- alert: HighErrorRate
expr: |
sum(rate(llm_requests_total{status="error"}[5m]))
/
sum(rate(llm_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Alerte de taux d'erreur élevé"
description: "Le taux d'erreur dépasse 5%, valeur actuelle : {{ $value | humanizePercentage }}"

# Alerte TTFT trop élevé
- alert: SlowTTFT
expr: |
histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "TTFT trop élevé"
description: "Le 95e percentile de TTFT du modèle {{ $labels.model }} dépasse 2 secondes"

# Utilisation du GPU trop faible (gaspillage de ressources)
- alert: LowGPUUtilization
expr: llm_gpu_utilization_percent < 30
for: 30m
labels:
severity: info
annotations:
summary: "Utilisation du GPU trop faible"
description: "Utilisation du GPU {{ $labels.gpu_id }} {{ $value }}%, gaspillage possible de ressources"

# Utilisation du GPU trop élevée (risque de goulot d'étranglement)
- alert: HighGPUUtilization
expr: llm_gpu_utilization_percent > 95
for: 5m
labels:
severity: warning
annotations:
summary: "Utilisation du GPU trop élevée"
description: "Utilisation du GPU {{ $labels.gpu_id }} {{ $value }}%, peut devenir un goulot d'étranglement"

Surveillance spécifique à l'IA : Détection de dérive

Les modèles d'IA ne "cassent" pas seulement, ils "dérivent" aussi : au fil du temps, la distribution des données change et les performances du modèle se dégradent.

Exemple

# Chemin de fichier : drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
"""Statistiques de distribution"""
mean: float
std: float
min: float
max: float
percentiles: dict

class DataDriftDetector:
"""Détecteur de dérive des données"""

def __init__(self, baseline_data: np.ndarray):
"""Initialiser avec les données de référence"""
self.baseline_stats = self._compute_stats(baseline_data)
self.baseline_data = baseline_data

def _compute_stats(self, data: np.ndarray) -> DistributionStats:
"""Calculer les statistiques pour les données"""
return DistributionStats(
mean=float(np.mean(data)),
std=float(np.std(data)),
min=float(np.min(data)),
max=float(np.max(data)),
percentiles={
"p10": float(np.percentile(data, 10)),
"p25": float(np.percentile(data, 25)),
"p50": float(np.percentile(data, 50)),
"p75": float(np.percentile(data, 75)),
"p90": float(np.percentile(data, 90))
}
)

def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""
Détecter la dérive
Renvoie : présence d'une dérive, valeur p, statistiques
"""
# Test KS : comparer si deux distributions sont identiques
ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

# Calculer les statistiques pour les nouvelles données
new_stats = self._compute_stats(new_data)

# Calculer la différence des moyennes (pourcentage)
mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

# Déterminer si une dérive s'est produite
is_drift = ks_pvalue < threshold

return {
"is_drift": is_drift,
"ks_statistic": float(ks_statistic),
"ks_pvalue": float(ks_pvalue),
"mean_diff_percent": float(mean_diff_pct),
"baseline_stats": self.baseline_stats,
"new_stats": new_stats,
"threshold": threshold
}

# Simulation : détection de dérive pour la longueur de saisie utilisateur
def simulate_input_length_drift():
# Données de référence : longueur moyenne de saisie utilisateur 50
np.random.seed(42)
baseline = np.random.normal(loc=50, scale=15, size=1000)
baseline = np.clip(baseline, 5, 200)

detector = DataDriftDetector(baseline)

print("Exemple de détection de dérive des données")
print("="*50)
print(fRéférence : moyenne={detector.baseline_stats.mean:.2f}, écart-type={detector.baseline_stats.std:.2f})
print()

# Scénario 1 : Données normales (aucune dérive)
print(Scénario 1 : Données normales)
normal_data = np.random.normal(loc=50, scale=15, size=500)
normal_data = np.clip(normal_data, 5, 200)
result = detector.detect(normal_data)
print(fMoyenne des nouvelles données : {result['new_stats'].mean:.2f})
print(f" p-valeur KS : {result['ks_pvalue']:.4f}")
print(fDérive ? {'Oui' if result['is_drift'] else 'Non'})
print()

# Scénario 2 : Dérive (l'entrée s'est allongée)
print(Scénario 2 : L'entrée s'est allongée)
drift_data = np.random.normal(loc=100, scale=20, size=500)
drift_data = np.clip(drift_data, 5, 300)
result = detector.detect(drift_data)
print(fMoyenne des nouvelles données : {result['new_stats'].mean:.2f})
print(f" p-valeur KS : {result['ks_pvalue']:.4f}")
print(fDérive ? {'Oui' if result['is_drift'] else 'Non'})
print(fDifférence moyenne : {result['mean_diff_percent']:.1f}%)

if __name__ == "__main__":
simulate_input_length_drift()

Stratégies d'optimisation des coûts

Les GPU sont chers. Optimiser les coûts, ce n'est pas "être radin" — c'est de la "compétence d'ingénierie".

Mise en cache des requêtes : mise en cache sémantique

La mise en cache traditionnelle est une correspondance exacte : pour exactement la même question, renvoyer directement la réponse précédente.

La mise en cache sémantique va plus loin : des questions similaires peuvent également réutiliser des réponses précédentes ou des résultats intermédiaires.

Exemple

# Chemin du fichier : semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
"""Implémentation simple d'un cache sémantique"""

def __init__(self, threshold: float = 0.9):
self.threshold = threshold
self.queries = [] # Stocker les vecteurs des questions
self.answers = [] # Stocker les réponses correspondantes
self.timestamps = [] # Stocker les horodatages

def _simulate_embedding(self, text: str) -> np.ndarray:
"""
Embedding simulé ; dans de vrais projets, utilisez OpenAI Embeddings, Sentence-Transformers, etc.
"""
# Simulation de hachage simple ; dans de vrais projets, utilisez un vrai modèle d'embedding
np.random.seed(hash(text) % (2**32))
return np.random.randn(128) # vecteur à 128 dimensions

def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Calculer la similarité cosinus"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get(self, query: str) -> Optional[str]:
"""Cache de requêtes : si une question similaire est trouvée, renvoyer la réponse"""
query_vec = self._simulate_embedding(query)

for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
# Nettoyage des expirations (plus de 1 heure)
if time.time() - ts > 3600:
continue

similarity = self._cosine_similarity(query_vec, q_vec)
if similarity >= self.threshold:
return answer

return None

def put(self, query: str, answer: str):
"""Stocker dans le cache"""
query_vec = self._simulate_embedding(query)
self.queries.append(query_vec)
self.answers.append(answer)
self.timestamps.append(time.time())

# Contrôle simple de capacité : stocker au maximum 1000 entrées
if len(self.queries) > 1000:
self.queries = self.queries[-1000:]
self.answers = self.answers[-1000:]
self.timestamps = self.timestamps[-1000:]

# Exemple d'utilisation
def main():
cache = SemanticCache(threshold=0.9)

# Première requête
query1 = Comment lire un fichier CSV en Python ?
print(fRequête 1 : {query1})
answer1 = cache.get(query1)
if answer1:
print(fRésultat trouvé dans le cache : {answer1})
else:
print(Absent du cache, appel du modèle...)
answer1 = Vous pouvez utiliser pandas.read_csv('file.csv') pour lire des fichiers CSV.
cache.put(query1, answer1)
print(fStocker dans le cache : {answer1})

print()

# Deuxième fois : question similaire
query2 = Comment lire un CSV en Python ?
print(fRequête 2 : {query2})
answer2 = cache.get(query2)
if answer2:
print(fCorrespondance dans le cache : {answer2})
else:
print(Absent du cache, appel du modèle...)

print()

# Troisième fois : question complètement différente
query3 = Qu'est-ce que l'apprentissage automatique ?
print(fRequête 3 : {query3})
answer3 = cache.get(query3)
if answer3:
print(fCorrespondance dans le cache : {answer3})
else:
print(Absent du cache, appel du modèle...)

if __name__ == "__main__":
main()

Routage de modèle : petit modèle comme solution de repli

Toutes les requêtes n'ont pas besoin d'un grand modèle. Utilisez un petit modèle pour les questions simples et un grand modèle pour les questions complexes.

Exemple

# Chemin du fichier : model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float
max_tokens: int
capabilities: List[str] # Liste des capacités

class SmartModelRouter:
"""Routeur de modèle intelligent"""

def __init__(self):
self.models = {
"small": ModelConfig(
name="Qwen2.5-0.5B-Instruct",
cost_per_1k_tokens=0.001,
max_tokens=4096,
capabilities=["chat", "simple_qa", "summarization"]
),
"medium": ModelConfig(
name="Qwen2.5-7B-Instruct",
cost_per_1k_tokens=0.01,
max_tokens=8192,
capabilities=["chat", "qa", "coding", "reasoning"]
),
"large": ModelConfig(
name="Qwen2.5-72B-Instruct",
cost_per_1k_tokens=0.1,
max_tokens=32768,
capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
)
}

def _classify_query(self, query: str) -> str:
"""
Classification de questions simples (vous pouvez utiliser un classificateur dans de vrais projets)
Retour : petit/moyen/grand
"""
query_lower = query.lower()

# Mots-clés pour les questions simples
simple_keywords = [
"Bonjour", "bonjour", "Merci", "salut", "Au revoir",
"Météo d'aujourd'hui", "Quelle heure", "Date",
"Aide à traduire", "traduire",
"Brève introduction", "Qu'est-ce que"
]

# Mots-clés pour les questions complexes
complex_keywords = [
"Code", "Écrire un", "Implémenter", "code", "python",
"Analyser", "Expliquer pourquoi", "Dériver", "Prouver",
"Comparer", "Opposer", "Différence",
"Détaillé", "Approfondi", "Complexe"
]

# Vérifier s'il s'agit d'une question simple
for keyword in simple_keywords:
if keyword in query_lower:
return "petit"

# Vérifier s'il s'agit d'une question complexe
for keyword in complex_keywords:
if keyword in query_lower:
return "grand"

# Utiliser le modèle moyen par défaut
return "medium"

def route(self, query: str, user_tier: str = "free") -> str:
"""
Décision de routage
"""
# Classer d'abord la question
category = self._classify_query(query)

# Les utilisateurs payants peuvent utiliser un meilleur modèle
if user_tier == "pro":
upgrade_map = {
"small": "small",
"medium": "medium",
"large": "large"
}
category = upgrade_map[category]
elif user_tier == "enterprise":
upgrade_map = {
"petit": "moyen",
"moyen": "grand",
"large": "large"
}
category = upgrade_map.get(category, category)

# Retourne le nom du modèle
return self.models[category].name

def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Estimer le coût"""
for model in self.models.values():
if model.name == model_name:
return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
return 0.0

# Exemple d'utilisation
def main():
router = SmartModelRouter()

test_queries = [
"Bonjour, j'aimerais en savoir plus",
"Qu'est-ce que Python ?",
"Aide-moi à écrire un tri rapide en Python",
Fournissez une analyse détaillée de l'architecture Transformer.
]

print(Exemple de routage intelligent)
print("="*50)

for query in test_queries:
model = router.route(query, user_tier="free")
cost = router.get_cost_estimate(model, 100, 200)
print(fRequête : {query})
print(fRoute vers : {model})
print(fCoût estimé : ${cost:.4f})
print()

# Différents niveaux d'utilisateur
print(Comparaison de différents niveaux d'utilisateur)
print("-"*50)
requête = Aide-moi à écrire un quicksort en Python
for tier in ["free", "pro", "enterprise"]:
model = router.route(query, user_tier=tier)
print(f{tier} user: {model})

if __name__ == "__main__":
main()

Instances Spot et traitement par lots

Utilisez des instances à la demande pour les requêtes en temps réel (stables mais coûteuses), et des instances Spot pour le traitement par lots hors ligne (économiques mais pouvant être préemptées).

Résumé de plusieurs méthodes d'optimisation des coûts :

Stratégie	Scénario	Économies de coûts
Cache sémantique	De nombreuses requêtes répétées	30-70%
Routage de modèle	Forte proportion de requêtes simples	50-80%
Quantification de modèle	Tous les scénarios	30-50%
Instances Spot	Traitement par lots hors ligne	60-90%
Mise à l'échelle automatique	Fortes fluctuations de trafic	30-60%

Gestion des pannes et dégradation

Dans les environnements de production, le « zéro panne » n'existe pas — seulement « comment gérer les pannes avec élégance ».

Stratégie de dégradation

Exemple

# Chemin du fichier : fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"

@dataclass
class FallbackResponse:
content: str
is_fallback: bool
fallback_reason: Optional[str] = None

class FallbackHandler:
"""Gestionnaire de repli"""

def __init__(self):
self.error_count = 0
self.last_error_time = 0
self.circuit_open = False
self.circuit_open_time = 0

# Configuration du disjoncteur
self.error_threshold = 5 # Déclenchement après 5 erreurs consécutives
self.circuit_timeout = 30 # Tentative de rétablissement après 30 secondes de circuit ouvert

def _call_primary_service(self, query: str) -> str:
"""Simuler l'appel au service principal"""
# Remplacer ceci par un véritable appel au modèle
return "Voici la réponse du service principal"

def _call_fallback_service(self, query: str) -> str:
"""Simuler l'appel au service de secours (plus rapide mais de moindre qualité)"""
# Peut être un modèle plus petit, une réponse mise en cache, ou même un gabarit prédéfini
return "Désolé, le service est temporairement occupé. Ceci est une réponse simplifiée."

def _call_static_fallback(self, query: str) -> str:
"""Réponse de secours statique (dernière ligne de défense)"""
return (
"Nous sommes vraiment désolés, notre service rencontre actuellement quelques problèmes.\n"
"Veuillez réessayer plus tard. Pour les problèmes urgents, veuillez contacter [email protected]."
)

def process_request(self, query: str) -> FallbackResponse:
Gérer les requêtes avec une logique de repli complète

# 1. Vérifier si le disjoncteur est ouvert
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_timeout:
# Délai dépassé, essayer l'état semi-ouvert, autoriser un petit nombre de requêtes à passer
print("Disjoncteur semi-ouvert, tentative de récupération...")
self.circuit_open = False
else:
# Basculer directement en repli
print("Le disjoncteur est ouvert, basculement direct en repli")
return FallbackResponse(
content=self._call_fallback_service(query),
is_fallback=True,
fallback_reason="circuit_open"
)

# 2. Essayer le service principal
try:
result = self._call_primary_service(query)
# Succès : réinitialiser le compteur d'erreurs
self.error_count = 0
return FallbackResponse(
content=result,
is_fallback=False
)

except Exception as e:
print(f"L'appel au service principal a échoué : {e}")
self.error_count += 1
self.last_error_time = time.time()

# 3. Vérifier si le déclenchement du disjoncteur est nécessaire
if self.error_count >= self.error_threshold:
print("Trop d'erreurs, ouverture du disjoncteur")
self.circuit_open = True
self.circuit_open_time = time.time()

# 4. Essayer le service de repli
try:
print("Essayer d'appeler le service de repli")
result = self._call_fallback_service(query)
return FallbackResponse(
content=result,
is_fallback=True,
fallback_reason="primary_failed"
)
except Exception as e2:
print(f"Le service de repli a également échoué : {e2}")

# 5. Réponse statique finale
return FallbackResponse(
content=self._call_static_fallback(query),
is_fallback=True,
fallback_reason="all_failed"
)

# Exemple d'utilisation
def main():
handler = FallbackHandler()

print("Exemple de gestion de repli")
print("="*50)

# Requête normale
print("Requête normale :")
resp = handler.process_request("Hello")
print(f" Réponse : {resp.content}")
print(f" Est-ce un repli ? {resp.is_fallback}")
print()

# Simuler une erreur (dans les projets réels, les exceptions sont réellement levées)
# Ici, nous démontrons simplement la logique
print("Résumé de la stratégie de repli :")
print(" 1. Préférer le service principal")
print(" 2. Le service principal échoue → utiliser un petit modèle/cache")
print(" 3. Défaillances consécutives → disjonction pour éviter l'avalanche")
print(" 4. Pendant la disjonction → renvoyer rapidement une réponse de repli")
print(" 5. Dernière ligne de défense → réponse statique")

Déploiement canari et retour arrière

Ne déployez pas la nouvelle version auprès de tous les utilisateurs en même temps. Commencez par l'envoyer à 1% du trafic, et s'il n'y a aucun problème, alors à 5%, 10%, 100%.

Si des problèmes surviennent, vous devez pouvoir revenir en arrière rapidement. N'attendez pas que les utilisateurs se soient plaints longtemps avant de modifier la configuration dans la panique.

Autres extensions