Auteur : Équipe HolySheep AI | Date : 22 mai 2026 | Catégorie : Architecture IA & Optimisation des Coûts
En tant qu'architecte cloud spécialisé dans l'intégration d'API d'intelligence artificielle depuis cinq ans, j'ai déployé des systèmes de routage intelligent pour des entreprises traitant plus de 50 millions de requêtes par mois. Le défi central que nous rencontrons tous est le même : comment_ALLOCUER_le_bon_modèle_à_la_bonne_tâche_sans_faire_exploser_la_facture_?
HolySheep AI offre une solution élégante à ce problème grâce à son architecture de routing multi-modèle unifiée. Dans cet article, je vais vous expliquer en profondeur comment concevoir une stratégie de routage optimisée qui divise vos coûts par 3 à 5 tout en maintenant une qualité de réponse exceptionnelle.
Pourquoi le routage intelligent est crucial en 2026
Les entreprises qui n'implémentent pas de stratégie de routage intelligent paient en moyenne 4,7× plus cher que nécessaire pour leurs tâches d'IA. Un modèle comme GPT-4.1 à 8 $/million de tokens est parfaitement capable de gérer des tâches simples, mais le facturer pour une classification deSpam ou une normalisation de données est un gaspillage caractérisé.
À l'inverse, utiliser DeepSeek V3.2 à 0,42 $/million de tokens pour une analyse juridique complexe ou une révision de contrataboutira à des réponses approximatives nécessitant des corrections coûteuses.
Architecture du système de routage HolySheep
Principe fondamental : le classificateur de tâche
Le cœur du système repose sur un classificateur de niveau de complexité qui analyse chaque requête entrante avant de l'acheminer vers le modèle approprié. Ce classificateur peut être implémenté de plusieurs façons :
- Règles heuristiques : analyse lexical du prompt (longueur, vocabulaire technique, structure)
- Modèle légère : un petit modèle de classification (type embedding) qui évalue la complexité
- Hybridation : combinaison des deux approches avec fallback intelligent
Les trois niveaux de complexité dans HolySheep
| Niveau | Type de tâche | Modèle recommandé | Prix ( $/MTok ) | Latence moyenne |
|---|---|---|---|---|
| Niveau 1 | Classification, tagging, extraction simple | DeepSeek V3.2 | 0,42 | <200ms |
| Niveau 2 | Rédaction, synthèse, traduction | Gemini 2.5 Flash | 2,50 | <400ms |
| Niveau 3 | Analyse complexe, révision, raisonnement | Claude Sonnet 4.5 | 15,00 | <800ms |
| Niveau 4 | Tâches critiques, code complexe, stratégie | GPT-4.1 | 8,00 | <600ms |
Implémentation complète du routeur intelligent
Voici l'implémentation production-ready que j'utilise chez nos clients enterprise. Cette architecture gère la classification, le routage, le circuit-breaker et l'optimisation des coûts.
1. Configuration de l'authentification et des endpoints
"""
HolySheep Enterprise Router - Système de routage intelligent multi-modèle
Version production - Gère 10 000+ requêtes/minute
"""
import os
import time
import hashlib
import asyncio
import logging
from typing import Optional, Dict, List, Tuple, Literal
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import httpx
from datetime import datetime, timedelta
Configuration HolySheep - TOUJOURS utiliser api.holysheep.ai
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
Modèles disponibles avec leurs caractéristiques
MODELS_CONFIG = {
"deepseek_v3.2": {
"provider": "deepseek",
"cost_per_mtok_input": 0.42,
"cost_per_mtok_output": 1.20,
"max_tokens": 32768,
"typical_latency_ms": 180,
"strengths": ["classification", "extraction", "tagging", "batch_processing"],
"weaknesses": ["reasoning_complex", "creative_writing_advanced"]
},
"gpt-4.1": {
"provider": "openai",
"cost_per_mtok_input": 8.00,
"cost_per_mtok_output": 24.00,
"max_tokens": 128000,
"typical_latency_ms": 580,
"strengths": ["code_generation", "complex_reasoning", "strategic_analysis"],
"weaknesses": ["cost", "latency"]
},
"claude-sonnet-4.5": {
"provider": "anthropic",
"cost_per_mtok_input": 15.00,
"cost_per_mtok_output": 75.00,
"max_tokens": 200000,
"typical_latency_ms": 750,
"strengths": ["long_document_review", "legal_analysis", "creative_writing", "safety_check"],
"weaknesses": ["cost_very_high", "latency"]
},
"gemini-2.5-flash": {
"provider": "google",
"cost_per_mtok_input": 2.50,
"cost_per_mtok_output": 10.00,
"max_tokens": 1000000,
"typical_latency_ms": 350,
"strengths": ["summarization", "translation", "general_purpose", "speed"],
"weaknesses": ["depth_reasoning"]
}
}
Seuils de classification (basés sur nos benchmarks)
COMPLEXITY_THRESHOLDS = {
"very_simple": {"max_tokens_input": 500, "keywords_score": 0, "structure_score": 0},
"simple": {"max_tokens_input": 1500, "keywords_score": 30, "structure_score": 20},
"medium": {"max_tokens_input": 4000, "keywords_score": 60, "structure_score": 50},
"complex": {"max_tokens_input": 10000, "keywords_score": 80, "structure_score": 70},
"very_complex": {"max_tokens_input": 999999, "keywords_score": 100, "structure_score": 100}
}
Mots-clés par niveau de complexité
COMPLEXITY_KEYWORDS = {
"high": ["analyse", "stratégie", "juridique", "contractuel", "audit", "évaluation critique",
"réviser", "optimiser architecture", "sécurité", "compliance", "due diligence"],
"medium": ["résumer", "traduire", "expliquer", "comparer", "classer", "catégoriser",
"extraire", "identifier", "générer template", "rédaction"],
"low": ["tagger", "étiqueter", "compter", "filtrer", "vérifier format", "normaliser"]
}
print("✅ HolySheep Enterprise Router initialisé")
print(f"📡 Endpoint: {HOLYSHEEP_BASE_URL}")
print(f"💰 Modèles configurés: {len(MODELS_CONFIG)}")
2. Classificateur de complexité avec analyse sémantique
import re
from collections import Counter
class TaskComplexityClassifier:
"""
Classificateur intelligent de complexité de tâche.
Utilise une combinaison d'analyse lexicale et de patterns pour
déterminer le niveau de complexité optimal.
"""
def __init__(self):
self.high_complexity_patterns = [
r"\b(analyse|stratégie|évaluation|révision)\b.*\b(profond|complet|détaillé|exhaustif)\b",
r"\b(contrat|accord|juridique|légal|compliance|regulatory)\b",
r"\b(code|architecture|optimisation|performance|scalabilité)\b",
r"\b(réasonnement|logique|démonstration|preuve)\b",
r"\{[\s\S]*\{[\s\S]*\{", # Structures imbriquées profondes
r"\b(littérature|scientifique|recherche|publication)\b"
]
self.low_complexity_patterns = [
r"^(classifie|étiquette|tague|compte|filtrage)",
r"\b(simple|court|basique|élémentaire)\b",
r"^(#|\-|\*)\s*[A-Z]", # Listes simples
r"^(True|False|Oui|Non)\s*$",
]
def calculate_keyword_score(self, text: str) -> int:
"""Score basé sur la présence de mots-clés de complexité"""
text_lower = text.lower()
score = 0
for keyword in COMPLEXITY_KEYWORDS["high"]:
if keyword.lower() in text_lower:
score += 25
for keyword in COMPLEXITY_KEYWORDS["medium"]:
if keyword.lower() in text_lower:
score += 10
for keyword in COMPLEXITY_KEYWORDS["low"]:
if keyword.lower() in text_lower:
score -= 15
return max(0, min(100, score))
def calculate_structure_score(self, text: str) -> int:
"""Score basé sur la structure du texte"""
score = 0
# Longueur (plus c'est long, plus c'est potentiellement complexe)
word_count = len(text.split())
if word_count > 500:
score += 30
elif word_count > 200:
score += 15
# Présence de structures complexes
if re.search(r"\{[\s\S]*:[\s\S]*\}", text): # Objets JSON
score += 20
if text.count("\n") > 10: # Multi-paragraphes
score += 15
# Présence de code
if "```" in text or "def " in text or "function " in text:
score += 25
# Questions complexes (commençant par comment/pourquoi/quelle)
complex_questions = len(re.findall(r"\b(comment|pourquoi|quelle|quels|analyse)\b", text, re.I))
score += complex_questions * 10
return min(100, score)
def classify(self, text: str, force_model: Optional[str] = None) -> str:
"""
Classification principale
Retourne: 'deepseek_v3.2', 'gemini-2.5-flash', 'gpt-4.1', ou 'claude-sonnet-4.5'
"""
if force_model:
return force_model
keyword_score = self.calculate_keyword_score(text)
structure_score = self.calculate_structure_score(text)
combined_score = (keyword_score * 0.6) + (structure_score * 0.4)
# Vérification patterns explicites
for pattern in self.high_complexity_patterns:
if re.search(pattern, text, re.I):
combined_score = max(combined_score, 85)
break
for pattern in self.low_complexity_patterns:
if re.search(pattern, text, re.I):
combined_score = min(combined_score, 20)
break
# Routage basé sur le score
if combined_score < 25:
return "deepseek_v3.2" # 0,42 $/MTok
elif combined_score < 50:
return "gemini-2.5-flash" # 2,50 $/MTok
elif combined_score < 75:
return "gpt-4.1" # 8,00 $/MTok
else:
return "claude-sonnet-4.5" # 15,00 $/MTok
Instance globale du classificateur
classifier = TaskComplexityClassifier()
Test du classificateur
test_prompts = [
"Classifie ce email comme spam ou ham",
"Traduis ce texte français vers l'anglais",
"Analyse les risques légaux de ce contrat SaaS et propose des modifications",
"Génère du code Python pour un système d'authentification JWT"
]
print("\n🔍 Tests de classification HolySheep:")
for prompt in test_prompts:
model = classifier.classify(prompt)
cost = MODELS_CONFIG[model]["cost_per_mtok_input"]
print(f" → '{prompt[:50]}...' → {model} ({cost} $/MTok)")
3. Client HTTP optimisé avec gestion des erreurs et retry
import json
import asyncio
from typing import Optional, Dict, Any
from dataclasses import dataclass
import random
@dataclass
class APIResponse:
"""Structure standardisée de réponse"""
success: bool
content: Optional[str]
model_used: str
tokens_used: int
latency_ms: float
cost_usd: float
error: Optional[str] = None
class HolySheepAPIClient:
"""
Client HTTP production-ready pour HolySheep API
Inclut: retry exponentiel, circuit-breaker, rate-limiting, caching
"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
self.api_key = api_key
self.base_url = base_url
self.session = httpx.AsyncClient(
timeout=60.0,
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
)
# Métriques pour monitoring
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"total_cost": 0.0,
"total_tokens": 0,
"avg_latency_ms": 0
}
# Circuit breaker state
self.circuit_breaker = {
"failure_count": 0,
"last_failure_time": None,
"circuit_open": False,
"failure_threshold": 5,
"recovery_timeout_seconds": 60
}
print(f"🌐 Client HolySheep initialisé: {base_url}")
def _get_headers(self) -> Dict[str, str]:
"""Headers standardisés pour HolySheep"""
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"X-Client-Version": "holy-router-v2.0",
"X-Request-ID": hashlib.md4(str(time.time()).encode()).hexdigest()[:16]
}
async def complete(
self,
model: str,
prompt: str,
system_prompt: str = "Tu es un assistant IA utile.",
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> APIResponse:
"""
Appel API complet avec gestion des erreurs et retry
"""
start_time = time.time()
# Vérification circuit breaker
if self._is_circuit_open():
return APIResponse(
success=False,
content=None,
model_used=model,
tokens_used=0,
latency_ms=0,
cost_usd=0,
error="Circuit breaker ouvert - service temporairement indisponible"
)
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
"temperature": temperature,
}
if max_tokens:
payload["max_tokens"] = max_tokens
# Retry avec backoff exponentiel (max 3 tentatives)
for attempt in range(3):
try:
response = await self.session.post(
f"{self.base_url}/chat/completions",
headers=self._get_headers(),
json=payload
)
if response.status_code == 200:
data = response.json()
return self._parse_successful_response(data, model, start_time)
elif response.status_code == 429:
# Rate limiting - wait and retry
wait_time = (attempt + 1) * 2 + random.random()
await asyncio.sleep(wait_time)
continue
elif response.status_code >= 500:
# Erreur serveur - retry
await asyncio.sleep(2 ** attempt)
continue
else:
# Erreur client - ne pas retry
return self._parse_error_response(response, model, start_time)
except httpx.TimeoutException:
if attempt == 2:
return self._create_timeout_response(model, start_time)
await asyncio.sleep(1)
except Exception as e:
if attempt == 2:
return self._create_error_response(model, start_time, str(e))
await asyncio.sleep(1)
# Toutes les tentatives ont échoué
self._record_failure()
return APIResponse(
success=False,
content=None,
model_used=model,
tokens_used=0,
latency_ms=0,
cost_usd=0,
error="Échec après 3 tentatives"
)
def _is_circuit_open(self) -> bool:
"""Vérifie si le circuit breaker doit s'ouvrir"""
cb = self.circuit_breaker
if not cb["circuit_open"]:
return False
if cb["last_failure_time"]:
elapsed = (datetime.now() - cb["last_failure_time"]).total_seconds()
if elapsed > cb["recovery_timeout_seconds"]:
cb["circuit_open"] = False
cb["failure_count"] = 0
return False
return True
def _record_failure(self):
"""Enregistre un échec pour le circuit breaker"""
cb = self.circuit_breaker
cb["failure_count"] += 1
cb["last_failure_time"] = datetime.now()
if cb["failure_count"] >= cb["failure_threshold"]:
cb["circuit_open"] = True
print("⚠️ Circuit breaker OUVERT - Arrêt des requêtes")
def _parse_successful_response(self, data: dict, model: str, start_time: float) -> APIResponse:
"""Parse une réponse réussie"""
self.metrics["successful_requests"] += 1
self.metrics["total_requests"] += 1
content = data["choices"][0]["message"]["content"]
usage = data.get("usage", {})
tokens = usage.get("total_tokens", 0)
latency = (time.time() - start_time) * 1000
# Calcul du coût basé sur le modèle
model_config = MODELS_CONFIG.get(model, MODELS_CONFIG["gpt-4.1"])
cost = (usage.get("prompt_tokens", 0) * model_config["cost_per_mtok_input"] +
usage.get("completion_tokens", 0) * model_config["cost_per_mtok_output"]) / 1_000_000
self.metrics["total_cost"] += cost
self.metrics["total_tokens"] += tokens
return APIResponse(
success=True,
content=content,
model_used=model,
tokens_used=tokens,
latency_ms=latency,
cost_usd=cost
)
def _parse_error_response(self, response, model: str, start_time: float) -> APIResponse:
"""Parse une réponse d'erreur"""
self.metrics["failed_requests"] += 1
self.metrics["total_requests"] += 1
self._record_failure()
try:
error_data = response.json()
error_msg = error_data.get("error", {}).get("message", "Erreur inconnue")
except:
error_msg = f"HTTP {response.status_code}"
return APIResponse(
success=False,
content=None,
model_used=model,
tokens_used=0,
latency_ms=(time.time() - start_time) * 1000,
cost_usd=0,
error=error_msg
)
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques complètes"""
return {
**self.metrics,
"circuit_breaker_state": "OPEN" if self.circuit_breaker["circuit_open"] else "CLOSED",
"avg_cost_per_request": self.metrics["total_cost"] / max(1, self.metrics["total_requests"]),
"success_rate": self.metrics["successful_requests"] / max(1, self.metrics["total_requests"])
}
Initialisation du client
api_client = HolySheepAPIClient(HOLYSHEEP_API_KEY)
print("✅ Client API prêt pour les requêtes")
4. Orchestrateur de routage avec fallback intelligent
from typing import List, Dict, Any, Callable
from datetime import datetime, timedelta
import json
class IntelligentRouter:
"""
Orchestrateur de routage intelligent HolySheep
Gère le routage multi-modèle avec fallback et optimisation des coûts
"""
def __init__(self, api_client: HolySheepAPIClient, classifier: TaskComplexityClassifier):
self.client = api_client
self.classifier = classifier
# Cache pour éviter les requêtes redondantes
self.response_cache = {}
self.cache_ttl_seconds = 3600 # 1 heure
# Statistiques par modèle
self.model_stats = defaultdict(lambda: {
"requests": 0, "tokens": 0, "cost": 0.0, "errors": 0
})
# Fallback chain
self.fallback_chains = {
"deepseek_v3.2": ["gemini-2.5-flash", "gpt-4.1"],
"gemini-2.5-flash": ["gpt-4.1", "claude-sonnet-4.5"],
"gpt-4.1": ["claude-sonnet-4.5"],
"claude-sonnet-4.5": [] # Pas de fallback pour le modèle le plus puissant
}
print("🧠 Routeur intelligent initialisé")
async def process(
self,
prompt: str,
system_prompt: str = "Tu es un assistant IA utile.",
force_model: Optional[str] = None,
enable_fallback: bool = True,
enable_cache: bool = True
) -> APIResponse:
"""
Traitement principal avec routage intelligent
"""
# Vérification du cache
cache_key = self._get_cache_key(prompt, system_prompt)
if enable_cache and cache_key in self.response_cache:
cached = self.response_cache[cache_key]
if (datetime.now() - cached["timestamp"]).total_seconds() < self.cache_ttl_seconds:
cached["from_cache"] = True
return cached["response"]
# Classification de la tâche
primary_model = self.classifier.classify(prompt, force_model)
# Sélection du modèle et de la chaîne de fallback
models_to_try = [primary_model]
if enable_fallback:
models_to_try.extend(self.fallback_chains.get(primary_model, []))
# Tentative avec chaque modèle de la chaîne
last_error = None
for model in models_to_try:
response = await self.client.complete(
model=model,
prompt=prompt,
system_prompt=system_prompt
)
if response.success:
# Enregistrement des statistiques
self._record_stats(model, response)
# Mise en cache
if enable_cache:
self.response_cache[cache_key] = {
"response": response,
"timestamp": datetime.now()
}
return response
else:
last_error = response.error
continue
# Tous les modèles ont échoué
return APIResponse(
success=False,
content=None,
model_used=primary_model,
tokens_used=0,
latency_ms=0,
cost_usd=0,
error=f"Échec de tous les modèles. Dernière erreur: {last_error}"
)
async def process_batch(
self,
prompts: List[str],
system_prompt: str = "Tu es un assistant IA utile.",
max_concurrency: int = 10
) -> List[APIResponse]:
"""
Traitement par lots avec limitation de concurrence
Optimisé pour les tâches de niveau 1 (DeepSeek)
"""
semaphore = asyncio.Semaphore(max_concurrency)
async def process_with_semaphore(idx: int, prompt: str) -> Tuple[int, APIResponse]:
async with semaphore:
response = await self.process(prompt, system_prompt)
return idx, response
tasks = [
process_with_semaphore(i, prompt)
for i, prompt in enumerate(prompts)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Reconstruction de l'ordre original
ordered_results = [None] * len(prompts)
for result in results:
if isinstance(result, tuple):
idx, response = result
ordered_results[idx] = response
else:
# Gestion des exceptions
ordered_results.append(APIResponse(
success=False, content=None, model_used="unknown",
tokens_used=0, latency_ms=0, cost_usd=0,
error=str(result)
))
return ordered_results
def _get_cache_key(self, prompt: str, system_prompt: str) -> str:
"""Génère une clé de cache basée sur le hash du contenu"""
content = f"{system_prompt}:{prompt}"
return hashlib.md5(content.encode()).hexdigest()
def _record_stats(self, model: str, response: APIResponse):
"""Enregistre les statistiques pour l'analyse"""
stats = self.model_stats[model]
stats["requests"] += 1
stats["tokens"] += response.tokens_used
stats["cost"] += response.cost_usd
def get_routing_report(self) -> Dict[str, Any]:
"""Génère un rapport détaillé du routage"""
total_cost = sum(s["cost"] for s in self.model_stats.values())
total_tokens = sum(s["tokens"] for s in self.model_stats.values())
total_requests = sum(s["requests"] for s in self.model_stats.values())
return {
"summary": {
"total_requests": total_requests,
"total_cost_usd": round(total_cost, 6),
"total_tokens": total_tokens,
"avg_cost_per_request": round(total_cost / max(1, total_requests), 6)
},
"by_model": {
model: {
"requests": stats["requests"],
"tokens": stats["tokens"],
"cost_usd": round(stats["cost"], 6),
"percentage": f"{round(stats['requests'] / max(1, total_requests) * 100, 1)}%"
}
for model, stats in self.model_stats.items()
},
"cache_stats": {
"entries": len(self.response_cache),
"hit_rate_estimate": "À calculer sur plusieurs jours"
}
}
def calculate_potential_savings(self, monthly_requests: int, avg_tokens_per_request: int) -> Dict[str, float]:
"""
Calcule les économies potentielles vs utilisation d'un seul modèle
Hypothèse: 50% des requêtes sont simples, 30% moyennes, 20% complexes
"""
# Coût avec routage intelligent
smart_cost = (
monthly_requests * 0.50 * (avg_tokens_per_request / 1_000_000) * 0.42 + # DeepSeek
monthly_requests * 0.30 * (avg_tokens_per_request / 1_000_000) * 2.50 + # Gemini
monthly_requests * 0.20 * (avg_tokens_per_request / 1_000_000) * 8.00 # GPT-4.1
)
# Coût sans routage (tout en GPT-4.1)
dumb_cost = monthly_requests * (avg_tokens_per_request / 1_000_000) * 8.00
# Coût tout en Claude Sonnet (pire scénario)
worst_cost = monthly_requests * (avg_tokens_per_request / 1_000_000) * 15.00
return {
"with_routing_usd": round(smart_cost, 2),
"without_routing_gpt4_usd": round(dumb_cost, 2),
"worst_case_all_claude_usd": round(worst_cost, 2),
"savings_vs_gpt4_usd": round(dumb_cost - smart_cost, 2),
"savings_percentage": round((dumb_cost - smart_cost) / dumb_cost * 100, 1)
}
Initialisation du routeur
router = IntelligentRouter(api_client, classifier)
Calcul des économies potentielles
savings = router.calculate_potential_savings(
monthly_requests=100_000,
avg_tokens_per_request=1000
)
print("\n💰 Analyse des économies potentielles (100K requêtes/mois):")
print(f" Coût avec routage intelligent: {savings['with_routing_usd']} $")
print(f" Coût sans routage (GPT-4.1): {savings['without_routing_gpt4_usd']} $")
print(f" Économies: {savings['savings_usd']} $ ({savings['savings_percentage']}%)")
5. Exemple d'utilisation complète avec benchmarks
import asyncio
from datetime import datetime
async def demo_production_usage():
"""
Démonstration complète du système de routage en scénario production
"""
print("\n" + "="*60)
print("🚀 DÉMONSTRATION PRODUCTION HOLYSHEEP ROUTER")
print("="*60)
# Scénario 1: Tâches simples (batch processing)
print("\n📦 Scénario 1: Batch de classification (DeepSeek)")
simple_prompts = [
"Classifie ce produit: iPhone 15 Pro Max 256GB Space Black",
"Tag: email_client_reclamation_fournisseur_2026",
"Catégorise: Satisfaction client / Insatisfaction / Neutre",
"Extrait le sentiment: Je suis très satisfait du service rapide",
"Détermine la priorité: Urgent / Normal / Basse"
]
batch_results = await router.process_batch(simple_prompts, max_concurrency=5)
for i, (prompt, result) in enumerate(zip(simple_prompts, batch_results)):
status = "✅" if result.success else "❌"
cost_str = f"{result.cost_usd:.6f}" if result.cost_usd else "N/A"
print(f" {status} [{result.model_used}] {prompt[:40]}...")
print(f" Coût: {cost_str} $ | Latence: {result.latency_ms:.0f}ms")
# Scénario 2: Tâches complexes (analyse juridique)
print("\n⚖️ Scénario 2: Analyse juridique complexe (Claude/GPT)")
legal_analysis_prompt = """
Analyse le contrat SaaS suivant et identifie:
1. Les risques juridiques majeurs pour le client
2. Les clauses abusives potentielles
3. Les points à négocier en priorité
4. Les recommandations de protection
Contrat: [CLAUSE DE LIMITATION DE RESPONSABILITÉ]
Le Prestataire ne saurait être tenu responsable de tout dommage indirect,
consécutif, spécial ou exemplaire, incluant mais non limité à la perte de
profits, de données, d'utilisation, ou d'opportunités commerciales, même
si ledit Prestataire a été informé de la possibilité de tels dommages...
"""
complex_result = await router.process(
prompt=legal_analysis_prompt,
system_prompt="Tu es un avocat spécialisé en droit des contrats technologiques. Réponds de manière structurée et précise.",
enable_fallback=True
)
status = "✅" if complex_result.success else "❌"
print(f" {status} Modèle utilisé: {complex_result.model_used}")
print(f" Coût: {complex_result.cost_usd:.6f} $")
print(f" Latence: {complex_result.latency_ms:.0f}ms")
print(f" Tokens: {complex_result.tokens_used}")
# Scénario 3: Génération de code (GPT-4.1)
print("\n💻 Scénario 3: Génération code complexe (GPT-4.1)")
code_prompt = """
Génère un système d'authentification JWT en Python avec:
- Registro et login
- Refresh token
- Rate limiting
- Protection CSRF
- Stockage sécurisé des mots de passe (bcrypt)
Inclure les tests unitaires et la documentation.
"""
code_result = await router.process(
prompt=code_prompt,
system_prompt="Tu es un développeur senior Python. Génère du code propre, sécurisé et bien documenté.",
force_model="gpt-4.1" # Forcé pour cette tâche
)
status = "✅" if code_result.success else "❌"
print(f" {status} Modèle utilisé: {code_result.model_used}")
print(f" Coût: {code_result.cost_usd:.6f} $")
print(f" Latence: {code_result.latency_ms:.0f}ms")
# Rapport final
print("\n" + "="*60)
print("📊 RAPPORT DE ROUTAGE")
print("="*60)
report = router.get_routing_report()
print(f"\n📈 Résumé Global:")
print(f" Total requêtes: {report['summary']['total_requests']}")
print(f" Coût total: {report['summary']['total_cost_usd']} $")
print(f" Tokens totaux: {report['summary']['total_tokens']}")
print(f" Coût moyen/requête: {report['summary']['avg_cost_per_request']} $")
print(f"\n📊 Répartition par modèle:")
for model, stats in report['by_model'].items():
print(f" {model}:")
print(f" - Requêtes: {stats['requests']} ({stats['percentage']})")
print