En tant qu'architecte système spécialisé dans l'intégration d'API IA, j'ai déployé des systèmes处理 plus de 50 millions de requêtes mensuelles pour des entreprises de toutes tailles. La leçon la plus importante que j'ai apprise ? Sans une stratégie robuste de traffic shaping et de scheduling de priorité, même les modèles les plus puissants peuvent s'effondrer sous la charge. Aujourd'hui, je vais vous partager mon framework complet pour construire une infrastructure résiliente et optimisée en termes de coûts.
Comprendre le Problème : Pourquoi le Traffic Shaping est Essentiel
Lorsque vous envoyez 10 millions de tokens par mois vers différentes API IA, la gestion du flux devient critique. Voici ma comparaison de coûts réelle pour mai 2026 :
- GPT-4.1 (OpenAI compatible via HolySheep) : 8 $/MTok × 10M tokens = 80 $/mois
- Claude Sonnet 4.5 : 15 $/MTok × 10M tokens = 150 $/mois
- Gemini 2.5 Flash : 2,50 $/MTok × 10M tokens = 25 $/mois
- DeepSeek V3.2 : 0,42 $/MTok × 10M tokens = 4,20 $/mois
Vous voyez le problème ? Un rapport de 1 à 36 entre les options les plus chères et les plus économiques. Le traffic shaping vous permet d'acheminer intelligemment les requêtes vers le modèle optimal selon le contexte, réduisant drastiquement vos coûts.
Architecture du Système de Traffic Shaping
Le Pattern Token Bucket Revisité
J'utilise une implémentation Python modifiée du pattern Token Bucket qui prend en compte la priorité des requêtes. Voici ma classe principale que j'utilise en production :
import asyncio
import time
import heapq
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from enum import IntEnum
import aiohttp
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RequestPriority(IntEnum):
"""Niveau de priorité des requêtes (plus élevé = plus prioritaire)"""
BACKGROUND = 0 # Tâches de fond, batch processing
NORMAL = 1 # Requêtes utilisateur standard
HIGH = 2 # Utilisateurs premium
CRITICAL = 3 # Opérations financières, sécurité
EMERGENCY = 4 # Incidents, monitoring
@dataclass(order=True)
class QueuedRequest:
"""Requête en file d'attente avec priorité"""
priority: int = field(compare=True)
timestamp: float = field(compare=True)
request_id: str = field(compare=False, default="")
model: str = field(compare=False, default="")
prompt_tokens: int = field(compare=False, default=0)
callback: Callable = field(compare=False, default=None)
retry_count: int = field(compare=False, default=0)
class TrafficShaper:
"""
Traffic Shaper avancé avec scheduling par priorité.
Supporte plusieurs providers avec rates différents.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
requests_per_minute: int = 60,
burst_size: int = 10
):
self.api_key = api_key
self.base_url = base_url
self.rpm_limit = requests_per_minute
self.burst_size = burst_size
# Token bucket state
self.tokens = float(burst_size)
self.last_refill = time.time()
self.refill_rate = requests_per_minute / 60.0 # tokens/seconde
# Queues par priorité (min-heap pour O(log n))
self.priority_queues: Dict[int, List[QueuedRequest]] = {
priority: [] for priority in RequestPriority
}
# Rate limits par modèle
self.model_rpm_limits: Dict[str, int] = {
"gpt-4.1": 500,
"claude-sonnet-4.5": 300,
"gemini-2.5-flash": 1000,
"deepseek-v3.2": 800
}
# Métriques
self.metrics = {
"total_requests": 0,
"total_tokens": 0,
"by_priority": {p.name: 0 for p in RequestPriority},
"by_model": {},
"avg_latency_ms": 0,
"queue_wait_avg_ms": 0
}
# Verrous pour thread-safety
self._lock = asyncio.Lock()
self._session: Optional[aiohttp.ClientSession] = None
logger.info(f"TrafficShaper initialisé: {rpm_limit} RPM, burst {burst_size}")
async def _refill_tokens(self):
"""Rajoute les tokens selon le refill rate"""
now = time.time()
elapsed = now - self.last_refill
new_tokens = elapsed * self.refill_rate
async with self._lock:
self.tokens = min(self.burst_size, self.tokens + new_tokens)
self.last_refill = now
async def _acquire_token(self, timeout: float = 30.0) -> bool:
"""Acquiert un token, attend si nécessaire"""
start_wait = time.time()
while True:
await self._refill_tokens()
async with self._lock:
if self.tokens >= 1.0:
self.tokens -= 1.0
return True
if time.time() - start_wait > timeout:
return False
await asyncio.sleep(0.05) # 50ms entre chaque tentative
async def enqueue(
self,
prompt: str,
model: str = "deepseek-v3.2",
priority: RequestPriority = RequestPriority.NORMAL,
request_id: Optional[str] = None
) -> str:
"""Ajoute une requête dans la queue avec priorité"""
if not request_id:
request_id = f"req_{int(time.time() * 1000)}"
queued = QueuedRequest(
priority=priority,
timestamp=time.time(),
request_id=request_id,
model=model,
prompt_tokens=len(prompt.split()) * 1.3, # Approximation
callback=None
)
async with self._lock:
heapq.heappush(self.priority_queues[priority], queued)
logger.debug(f"Requête {request_id} enqueued avec priorité {priority.name}")
return request_id
async def process_next(self) -> Optional[Dict]:
"""Traite la prochaine requête de la plus haute priorité"""
# Trouver la queue non-vide avec la priorité la plus haute
async with self._lock:
for priority in reversed(list(RequestPriority)):
if self.priority_queues[priority]:
request = heapq.heappop(self.priority_queues[priority])
break
else:
return None
# Acquérir un token
if not await self._acquire_token():
# Remettre dans la queue si timeout
async with self._lock:
heapq.heappush(self.priority_queues[RequestPriority(priority)], request)
return None
# Exécuter la requête
try:
result = await self._execute_request(request)
self.metrics["total_requests"] += 1
self.metrics["by_priority"][RequestPriority(priority).name] += 1
self.metrics["total_tokens"] += result.get("tokens_used", 0)
return result
except Exception as e:
logger.error(f"Erreur requête {request.request_id}: {e}")
if request.retry_count < 3:
request.retry_count += 1
async with self._lock:
heapq.heappush(
self.priority_queues[RequestPriority(priority)],
request
)
return {"error": str(e), "request_id": request.request_id}
async def _execute_request(self, request: QueuedRequest) -> Dict:
"""Exécute une requête vers l'API"""
if not self._session:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
endpoint = f"{self.base_url}/chat/completions"
payload = {
"model": request.model,
"messages": [{"role": "user", "content": f"Analyze: {request.prompt_tokens} tokens"}],
"max_tokens": 100
}
start_time = time.time()
async with self._session.post(endpoint, json=payload) as resp:
response = await resp.json()
latency_ms = (time.time() - start_time) * 1000
return {
"request_id": request.request_id,
"model": request.model,
"response": response,
"latency_ms": latency_ms,
"tokens_used": request.prompt_tokens
}
async def get_metrics(self) -> Dict:
"""Retourne les métriques du shaper"""
async with self._lock:
return self.metrics.copy()
async def close(self):
"""Ferme la session aiohttp"""
if self._session:
await self._session.close()
Exemple d'utilisation
async def main():
shaper = TrafficShaper(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1",
requests_per_minute=100
)
# Simulation de requêtes avec différentes priorités
tasks = []
for i in range(20):
priority = RequestPriority(i % 5)
model = ["deepseek-v3.2", "gpt-4.1", "gemini-2.5-flash"][i % 3]
await shaper.enqueue(
prompt=f"Analyse requête #{i}",
model=model,
priority=priority
)
# Traiter quelques requêtes
for _ in range(5):
result = await shaper.process_next()
if result:
print(f"Résultat: {result.get('model')}, Latence: {result.get('latency_ms'):.2f}ms")
metrics = await shaper.get_metrics()
print(f"Métriques: {metrics}")
await shaper.close()
if __name__ == "__main__":
asyncio.run(main())Implémentation du Router Intelligent par Coût
La véritable optimisation vient d'un router qui dirige automatiquement les requêtes vers le modèle le plus économique capable de remplir la tâche. Voici mon implémentation de production :
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from enum import Enum
import hashlib
class TaskComplexity(Enum):
"""Classification de la complexité des tâches"""
TRIVIAL = "trivial" # Génération simple, formatting
SIMPLE = "simple" # Classification basique, extraction
MODERATE = "moderate" # Analyse, résumé,问答
COMPLEX = "complex" # Raisonnement multi-étapes
EXPERT = "expert" # Code complexe, mathématiques avancées
@dataclass
class ModelConfig:
"""Configuration d'un modèle avec ses caractéristiques"""
name: str
provider: str
cost_per_1k_input: float
cost_per_1k_output: float
max_tokens: int
strengths: List[str]
weaknesses: List[str]
avg_latency_ms: float
class CostAwareRouter:
"""
Router intelligent qui achemine les requêtes vers
le modèle optimal en termes de coût/performance.
"""
# Configuration des modèles disponibles via HolySheep (2026)
MODELS: Dict[str, ModelConfig] = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai",
cost_per_1k_input=2.00, # 2$/MTok input
cost_per_1k_output=8.00, # 8$/MTok output
max_tokens=128000,
strengths=["raisonnement complexe", "code", "analyse"],
weaknesses=["coût élevé", "latence moyenne"],
avg_latency_ms=850.0
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
cost_per_1k_input=3.00,
cost_per_1k_output=15.00,
max_tokens=200000,
strengths=["contexte long", "écriture créative", "safety"],
weaknesses=["coût très élevé", "rate limiting strict"],
avg_latency_ms=920.0
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="google",
cost_per_1k_input=0.30,
cost_per_1k_output=2.50,
max_tokens=1000000,
strengths=["vitesse", "contexte massif", "prix"],
weaknesses=["qualité variable", "pas toujours précis"],
avg_latency_ms=180.0
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
cost_per_1k_input=0.10,
cost_per_1k_output=0.42,
max_tokens=64000,
strengths=["code", "mathématiques", "prix imbattable"],
weaknesses=["anglais parfois imparfait", "contexte limité"],
avg_latency_ms=320.0
)
}
# Mapping tâche -> modèle optimal (basé sur mon expérience)
TASK_MODEL_MAP: Dict[str, List[str]] = {
"classification": ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"],
"sentiment_analysis": ["deepseek-v3.2", "gpt-4.1"],
"code_generation": ["deepseek-v3.2", "gpt-4.1", "claude-sonnet-4.5"],
"code_review": ["claude-sonnet-4.5", "gpt-4.1", "deepseek-v3.2"],
"translation": ["deepseek-v3.2", "gpt-4.1"],
"summarization": ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1"],
"question_answering": ["gemini-2.5-flash", "deepseek-v3.2"],
"creative_writing": ["claude-sonnet-4.5", "gpt-4.1"],
"data_extraction": ["deepseek-v3.2", "gpt-4.1"],
"math_proof": ["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"]
}
def __init__(self, fallback_enabled: bool = True):
self.fallback_enabled = fallback_enabled
self.routing_cache: Dict[str, str] = {}
self.cost_savings = 0.0
self.total_requests = 0
def estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Estime le coût d'une requête en dollars"""
config = self.MODELS.get(model)
if not config:
return float('inf')
input_cost = (input_tokens / 1000) * config.cost_per_1k_input
output_cost = (output_tokens / 1000) * config.cost_per_1k_output
return input_cost + output_cost
def classify_task(self, prompt: str, context: Optional[Dict] = None) -> TaskComplexity:
"""Classification automatique de la complexité de la tâche"""
prompt_lower = prompt.lower()
# Mots-clés pour classification
expert_keywords = [
"prouve", "démontrer", "théorème", "algorithme complexe",
"architecture système", "optimisation performance critique",
"implémentation cryptographique"
]
complex_keywords = [
"analyse", "compare", "évalue", "recommande", "résous",
"explique en détail", "reasoning", "step by step"
]
moderate_keywords = [
"résume", "traduis", "convertis", "formate", "extrais"
]
# Vérification par mots-clés
if any(kw in prompt_lower for kw in expert_keywords):
return TaskComplexity.EXPERT
if any(kw in prompt_lower for kw in complex_keywords):
# Vérifier la longueur comme indicateur supplémentaire
if len(prompt) > 1000:
return TaskComplexity.COMPLEX
return TaskComplexity.MODERATE
if any(kw in prompt_lower for kw in moderate_keywords):
return TaskComplexity.SIMPLE
return TaskComplexity.TRIVIAL
def detect_task_type(self, prompt: str) -> str:
"""Détection automatique du type de tâche"""
prompt_lower = prompt.lower()
task_keywords = {
"code_generation": ["écris", "génère", "code", "fonction", "class"],
"code_review": ["review", "corrige", "optimise le code", "lint"],
"classification": ["classe", "catégorise", "classe selon"],
"sentiment_analysis": ["sentiment", "positif", "négatif", "émotion"],
"translation": ["traduis", "traduction", "en anglais", "en français"],
"summarization": ["résume", "résumé", "tl;dr", "points clés"],
"question_answering": ["quelle est", "qui est", "quand", "pourquoi"],
"creative_writing": ["écris une histoire", "crée un", "raconte"],
"data_extraction": ["extrais", "identifie les", "trouve tous les"],
"math_proof": ["prouve que", "démontre", "calcule", "résous l'équation"]
}
for task, keywords in task_keywords.items():
if any(kw in prompt_lower for kw in keywords):
return task
return "general_conversation"
def route(
self,
prompt: str,
input_tokens: int,
output_tokens: int = 500,
user_tier: str = "free",
context: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Route la requête vers le modèle optimal.
Args:
prompt: Le prompt de la requête
input_tokens: Nombre estimé de tokens d'entrée
output_tokens: Nombre estimé de tokens de sortie
user_tier: Niveau de l'utilisateur (free, pro, enterprise)
context: Contexte additionnel
Returns:
Dict avec le modèle choisi, le coût estimé, et la raison
"""
self.total_requests += 1
# Vérifier le cache
cache_key = hashlib.md5(prompt[:100].encode()).hexdigest()
if cache_key in self.routing_cache:
cached_model = self.routing_cache[cache_key]
return {
"model": cached_model,
"estimated_cost": self.estimate_cost(cached_model, input_tokens, output_tokens),
"routing_reason": "cache_hit",
"alternative_models": self._get_alternatives(cached_model)
}
# Classifier la tâche
task_type = self.detect_task_type(prompt)
complexity = self.classify_task(prompt, context)
# Obtenir les modèles candidates pour cette tâche
candidates = self.TASK_MODEL_MAP.get(task_type, ["gpt-4.1"])
# Ajuster selon la complexité et le tier
if complexity == TaskComplexity.EXPERT:
candidates = ["gpt-4.1", "claude-sonnet-4.5", "deepseek-v3.2"]
elif complexity == TaskComplexity.TRIVIAL:
candidates = ["deepseek-v3.2", "gemini-2.5-flash"]
# Pour les utilisateurs gratuits, prioriser les modèles économiques
if user_tier == "free":
candidates = [c for c in candidates if c in ["deepseek-v3.2", "gemini-2.5-flash"]]
# Trouver le modèle optimal par coût
best_model = None
best_cost = float('inf')
best_latency = float('inf')
for model_name in candidates:
config = self.MODELS[model_name]
cost = self.estimate_cost(model_name, input_tokens, output_tokens)
# Score composite: 70% coût, 30% latence
normalized_cost = cost / 0.50 # Normaliser par rapport au coût max
normalized_latency = config.avg_latency_ms / 1000 # Normaliser
score = 0.7 * normalized_cost + 0.3 * normalized_latency
if score < (0.7 * best_cost / 0.50 + 0.3 * best_latency / 1000):
best_cost = cost
best_model = model_name
best_latency = config.avg_latency_ms
# Mettre en cache
self.routing_cache[cache_key] = best_model
# Calculer les économies vs utiliser toujours GPT-4.1
gpt41_cost = self.estimate_cost("gpt-4.1", input_tokens, output_tokens)
self.cost_savings += (gpt41_cost - best_cost)
return {
"model": best_model,
"estimated_cost": best_cost,
"gpt41_cost_comparison": gpt41_cost,
"savings_percentage": ((gpt41_cost - best_cost) / gpt41_cost * 100) if gpt41_cost > 0 else 0,
"task_type": task_type,
"complexity": complexity.value,
"routing_reason": self._explain_routing(best_model, task_type, complexity),
"alternative_models": self._get_alternatives(best_model)
}
def _explain_routing(self, model: str, task_type: str, complexity: TaskComplexity) -> str:
"""Génère une explication humaine du routage"""
reasons = {
"deepseek-v3.2": "excellent rapport qualité/prix, particulièrement efficace pour le code",
"gemini-2.5-flash": "vitesse optimale pour les tâches simples, contexte massif",
"gpt-4.1": "meilleur pour les tâches complexes nécessitant un raisonnement approfondi",
"claude-sonnet-4.5": "excellent pour les tâches créatives et le contexte long"
}
return reasons.get(model, "modèle sélectionné par défaut")
def _get_alternatives(self, primary: str) -> List[Dict]:
"""Retourne les modèles alternatifs avec leurs coûts"""
alternatives = []
for name, config in self.MODELS.items():
if name != primary:
alternatives.append({
"model": name,
"cost_factor": self.MODELS[primary].cost_per_1k_output / config.cost_per_1k_output
if config.cost_per_1k_output > 0 else float('inf')
})
return sorted(alternatives, key=lambda x: x["cost_factor"])[:2]
def get_cost_report(self) -> Dict:
"""Génère un rapport d'économie de coûts"""
return {
"total_requests": self.total_requests,
"total_cost_savings": self.cost_savings,
"average_savings_per_request": self.cost_savings / self.total_requests
if self.total_requests > 0 else 0,
"cache_hit_rate": len(self.routing_cache) / self.total_requests
if self.total_requests > 0 else 0
}
Exemple d'utilisation intégrée avec le TrafficShaper
async def smart_request_example():
router = CostAwareRouter()
test_prompts = [
("Écris une fonction Python pour calculer la suite de Fibonacci", "free"),
("Traduis ce texte en anglais: Bonjour le monde", "pro"),
("Analyse le sentiment de: Ce produit est amazing!", "enterprise"),
("Explique la démonstration du dernier théorème de Fermat", "pro"),
]
shaper = TrafficShaper(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
for prompt, tier in test_prompts:
# Router la requête
route = router.route(
prompt=prompt,
input_tokens=len(prompt.split()) * 1.3,
output_tokens=300,
user_tier=tier
)
print(f"\n📝 Prompt: {prompt[:50]}...")
print(f" 👤 Tier: {tier}")
print(f" 🤖 Modèle: {route['model']}")
print(f" 💰 Coût estimé: {route['estimated_cost']:.4f}$")
print(f" 💡 Économie vs GPT-4.1: {route['savings_percentage']:.1f}%")
print(f" 📊 Raison: {route['routing_reason']}")
# Enqueue avec la priorité appropriée
priority = RequestPriority.HIGH if tier == "enterprise" else RequestPriority.NORMAL
await shaper.enqueue(prompt, model=route['model'], priority=priority)
# Rapport final
print("\n" + "="*50)
report = router.get_cost_report()
print(f"📈 Rapport d'économie:")
print(f" Requêtes traitées: {report['total_requests']}")
print(f" Économies totales: {report['total_cost_savings']:.2f}$")
print(f" Moyenne par requête: {report['average_savings_per_request']:.4f}$")
await shaper.close()
if __name__ == "__main__":
asyncio.run(smart_request_example())Monitoring et Ajustement Dynamique
En production, le monitoring en temps réel est crucial. Voici mon dashboard de métriques intégré :
import asyncio
from typing import Dict, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import json
@dataclass
class MonitoringMetrics:
"""Structure des métriques de monitoring"""
timestamp: datetime = field(default_factory=datetime.now)
rpm_current: float = 0.0
rpm_limit: float = 0.0
avg_latency_ms: float = 0.0
error_rate: float = 0.0
queue_depth_by_priority: Dict[str, int] = field(default_factory=dict)
tokens_by_model: Dict[str, int] = field(default_factory=dict)
cost_current_month: float = 0.0
class ProductionMonitor:
"""
Système de monitoring pour la production.
Collecte, analyse et alerte sur les métriques.
"""
def __init__(self, alert_threshold_error_rate: float = 0.05):
self.alert_threshold_error_rate = alert_threshold_error_rate
self.metrics_history: List[MonitoringMetrics] = []
self.alerts: List[Dict] = []
# Configuration des budgets (exemple: 500$/mois)
self.monthly_budget = 500.0
self.daily_budget = 50.0
# HolySheep tarifs 2026 pour calcul précis
self.pricing = {
"gpt-4.1": {"input": 2.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 3.00, "output": 15.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.10, "output": 0.42}
}
def calculate_current_cost(
self,
tokens_by_model: Dict[str, int],
input_output_ratio: float = 0.7
) -> float:
"""Calcule le coût actuel basé sur l'utilisation"""
total_cost = 0.0
for model, tokens in tokens_by_model.items():
if model not in self.pricing:
continue
input_tokens = int(tokens * input_output_ratio)
output_tokens = int(tokens * (1 - input_output_ratio))
pricing = self.pricing[model]
cost = (input_tokens / 1_000_000 * pricing["input"] +
output_tokens / 1_000_000 * pricing["output"])
total_cost += cost
return total_cost
def check_budget_alerts(self, current_cost: float) -> List[Dict]:
"""Vérifie si les seuils de budget sont atteints"""
alerts = []
# Alerte budget quotidien (80% utilisé)
if current_cost > self.daily_budget * 0.8:
alerts.append({
"type": "budget_warning",
"severity": "warning",
"message": f"Budget quotidien à {current_cost/self.daily_budget*100:.1f}%",
"current_cost": current_cost,
"limit": self.daily_budget
})
# Alerte budget mensuel (90% utilisé)
if current_cost > self.monthly_budget * 0.9:
alerts.append({
"type": "budget_critical",
"severity": "critical",
"message": f"Budget mensuel à {current_cost/self.monthly_budget*100:.1f}%",
"current_cost": current_cost,
"limit": self.monthly_budget
})
return alerts
def generate_optimization_suggestions(
self,
metrics: MonitoringMetrics
) -> List[str]:
"""Génère des suggestions d'optimisation basées sur les métriques"""
suggestions = []
# Suggestion basée sur l'utilisation des modèles coûteux
expensive_usage = sum(
metrics.tokens_by_model.get(model, 0)
for model in ["gpt-4.1", "claude-sonnet-4.5"]
)
total_tokens = sum(metrics.tokens_by_model.values())
if total_tokens > 0 and expensive_usage / total_tokens > 0.5:
suggestions.append(
"🚀 50%+ des tokens utilisent des modèles coûteux. "
"Envisagez de router plus de requêtes vers DeepSeek V3.2 ou Gemini Flash."
)
# Suggestion basée sur la latence
if metrics.avg_latency_ms > 2000:
suggestions.append(
"⚠️ Latence moyenne > 2s. Vérifiez le rate limiting "
"et envisagez de prioriser les modèles plus rapides."
)
# Suggestion basée sur le taux d'erreur
if metrics.error_rate > 0.02:
suggestions.append(
"🔴 Taux d'erreur > 2%. Analysez les logs d'erreur "
"et implémentez des retries avec backoff exponentiel."
)
# Suggestion de cache
suggestions.append(
"💡 Activez le caching des prompts similaires pour réduire "
"les coûts de 15-30% sur les requêtes répétitives."
)
return suggestions
async def run_monitoring_loop(self, shaper, interval_seconds: int = 60):
"""Boucle principale de monitoring"""
print("📊 Monitoring activé - Ctrl+C pour arrêter\n")
while True:
try:
# Collecter les métriques
raw_metrics = await shaper.get_metrics()
# Calculer les métriques agrégées
tokens_by_model = raw_metrics.get("by_model", {})
current_cost = self.calculate_current_cost(tokens_by_model)
metrics = MonitoringMetrics(
rpm_current=raw_metrics.get("total_requests", 0) / 60,
rpm_limit=shaper.rpm_limit,
avg_latency_ms=raw_metrics.get("avg_latency_ms", 0),
error_rate=raw_metrics.get("errors", 0) / max(raw_metrics.get("total_requests", 1), 1),
queue_depth_by_priority={
p.name: len(shaper.priority_queues[p])
for p in shaper.priority_queues
},
tokens_by_model=tokens_by_model,
cost_current_month=current_cost
)
self.metrics_history.append(metrics)
# Affichage dashboard
print(f"\n{'='*60}")
print(f"⏰ {datetime.now().strftime('%H:%M:%S')} | "
f"Coût mensuel: {current_cost:.2f}$ / {self.monthly_budget}$")
print(f"{'='*60}")
print(f"📈 RPM: {metrics.rpm_current:.1f} / {metrics.rpm_limit}")
print(f"⏱️ Latence avg: {metrics.avg_latency_ms:.0f}ms")
print(f"❌ Taux erreur: {metrics.error_rate*100:.2f}%")
# Distribution par modèle
print(f"\n📊 Distribution tokens:")
for model, tokens in sorted(tokens_by_model.items(), key=lambda x: -x[1])[:4]:
pct = tokens / max(sum(tokens_by_model.values()), 1) * 100
bar = "█" * int(pct / 5)
print(f" {model:25s} {tokens/1_000_000:6.2f}M ({pct:5.1f}%) {bar}")
# Alertes
alerts = self.check_budget_alerts(current_cost)
if alerts:
print(f"\n🚨 ALERTES:")
for alert in alerts:
print(f" [{alert['severity'].upper()}] {alert['message']}")
# Suggestions
suggestions = self.generate_optimization_suggestions(metrics)
if suggestions:
print(f"\n💡 Suggestions:")
for suggestion in suggestions:
print(f" {suggestion}")
await asyncio.sleep