Temps de lecture : 15 minutes | Difficulté : Avancée | Mise à jour : Mai 2026
Étude de cas : Scale-up SaaS e-commerce de Lyon
En début d'année 2026, une scale-up SaaS spécialisée dans la recommandation produit pour le e-commerce européen — nommons-la RetailBoost — faisait face à un défi critique. Leur plateforme génère 2 millions de requêtes IA par mois pour des fonctionnalités de résumé automatique de produits, génération de descriptions SEO et chatbots clients multilingues.
Le contexte métier :
- 2M+ requêtes mensuelles avec pics à 800 req/min en soirée
- 4 marchés (France, Allemagne, Espagne, Italie)
- 3 langues principales pour l'inférence
- Exigence client : réponse < 500ms sous peine de churn
Les douleurs du fournisseur précédent :
RetailBoost utilisait exclusivement OpenAI avec GPT-4o. Les problèmes étaient triples :
- Facture explosée : $4 200/mois pour 180M tokens sortants
- Latence instable : pic à 1,2s en période de forte affluence, causant 12% de timeout
- Rate limiting arbitraire : 500 req/min max, impossible de gérer les pics saisonniers (soldes, Black Friday)
- Pas de fallback : une erreur API = page produit avec description vide pour l'utilisateur final
Pourquoi HolySheep :
Après benchmark de 6 providers, l'équipe technique de RetailBoost a migré sur HolySheep AI pour 4 raisons clés :
- Économie de 85% sur le coût par token avec DeepSeek V3.2 à $0.42/MToken
- Latence médiane < 50ms en Europe (vs 180ms+ sur OpenAI)
- Multi-provider natif avec fallback automatique
- Paiement WeChat/Alipay pour l'équipe basée à Shanghai
Étapes concrètes de migration :
- Bascule de la base_url vers
https://api.holysheep.ai/v1 - Configuration du pool de clés API (3 clés rotationnelles)
- Déploiement canari : 5% du traffic la semaine 1, 50% en semaine 2, 100% en semaine 3
- Activation du circuit breaker sur les modèles secondaires
Métriques à 30 jours :
| Métrique | Avant (OpenAI) | Après (HolySheep) | Amélioration |
|---|---|---|---|
| Latence médiane | 420ms | 180ms | -57% |
| Timeout rate | 12% | 0.3% | -97% |
| Facture mensuelle | $4 200 | $680 | -84% |
| Disponibilité | 99.2% | 99.97% | +0.77% |
| Taux de fallback | N/A | 4.2% | NA |
Mon retour d'expérience terrain
En tant qu'ingénieur senior qui a migré des dizaines de systèmes sur HolySheep, je peux vous confirmer : la combinaison du multi-model fallback avec la gouvernance des quotas a transformé notre infrastructure. Le monitoring en temps réel des coûts par modèle et par équipe nous a permis d'identifier que 60% de nos appels utilisaient GPT-4.1 alors qu'un modèle moins cher aurait suffit. En implémentant le routing intelligent décrit ci-dessous, nous avons divisé notre facture par 6 sans sacrifier la qualité.
Architecture Multi-Model avec Fallback Intelligent
1. Configuration du client avec HolySheep
"""
HolySheep Multi-Model Router avec Fallback & Circuit Breaker
Compatible Python 3.10+
"""
import asyncio
import time
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, List, Dict, Any
from collections import deque
import httpx
============================================
CONFIGURATION HOLYSHEEP
============================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1", # ✅ OBLIGATOIRE
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Remplacez par votre clé
"timeout": 30.0,
"max_retries": 3,
"retry_delay": 1.0, # secondes exponentielles
}
============================================
MODÈLES ET PRIORITÉS (2026)
============================================
class ModelTier(Enum):
PREMIUM = "premium" # GPT-4.1, Claude Sonnet 4.5
STANDARD = "standard" # Gemini 2.5 Flash
ECONOMY = "economy" # DeepSeek V3.2
MODEL_CONFIG = {
"gpt-4.1": {
"tier": ModelTier.PREMIUM,
"cost_per_mtok": 8.0, # $8/M tokens
"latency_p95": 1200, # ms
"context_window": 128000,
"priority": 1,
},
"claude-sonnet-4.5": {
"tier": ModelTier.PREMIUM,
"cost_per_mtok": 15.0, # $15/M tokens
"latency_p95": 1500,
"context_window": 200000,
"priority": 2,
},
"gemini-2.5-flash": {
"tier": ModelTier.STANDARD,
"cost_per_mtok": 2.50, # $2.50/M tokens
"latency_p95": 600,
"context_window": 1000000,
"priority": 3,
},
"deepseek-v3.2": {
"tier": ModelTier.ECONOMY,
"cost_per_mtok": 0.42, # $0.42/M tokens
"latency_p95": 400,
"context_window": 64000,
"priority": 4,
},
}
@dataclass
class CircuitBreakerState:
"""État du circuit breaker par modèle"""
model: str
failure_count: int = 0
last_failure_time: float = 0
is_open: bool = False
recovery_timeout: float = 60.0 # secondes avant retry
# Métriques de performance
success_count: int = 0
total_latency: float = 0.0
error_log: deque = field(default_factory=lambda: deque(maxlen=10))
class HolySheepMultiModelRouter:
"""
Router intelligent avec fallback multi-niveau
et circuit breaker pour HolySheep API
"""
def __init__(
self,
api_key: str,
base_url: str = HOLYSHEEP_CONFIG["base_url"],
enable_circuit_breaker: bool = True,
circuit_breaker_threshold: int = 5,
):
self.api_key = api_key
self.base_url = base_url
self.enable_circuit_breaker = enable_circuit_breaker
self.cb_threshold = circuit_breaker_threshold
# État des circuit breakers par modèle
self.circuit_breakers: Dict[str, CircuitBreakerState] = {
model: CircuitBreakerState(model=model)
for model in MODEL_CONFIG.keys()
}
# Pool HTTP avec gestion des connexions
self.client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
timeout=httpx.Timeout(HOLYSHEEP_CONFIG["timeout"]),
limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
)
# Métriques globales
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"cost_total_usd": 0.0,
"latency_avg_ms": 0.0,
}
self.logger = logging.getLogger(__name__)
def _check_circuit_breaker(self, model: str) -> bool:
"""Vérifie si le circuit breaker est ouvert pour un modèle"""
if not self.enable_circuit_breaker:
return True
cb = self.circuit_breakers.get(model)
if not cb:
return True
if cb.is_open:
# Vérifier si le timeout de récupération est écoulé
if time.time() - cb.last_failure_time > cb.recovery_timeout:
cb.is_open = False
cb.failure_count = 0
self.logger.info(f"Circuit breaker reset pour {model}")
return True
return False
return True
def _trip_circuit_breaker(self, model: str, error: str):
"""Déclenche le circuit breaker en cas d'erreur"""
cb = self.circuit_breakers.get(model)
if not cb:
return
cb.failure_count += 1
cb.last_failure_time = time.time()
cb.error_log.append({"time": time.time(), "error": error})
if cb.failure_count >= self.cb_threshold:
cb.is_open = True
self.logger.warning(
f"Circuit breaker OPEN pour {model} "
f"après {cb.failure_count} échecs"
)
def _record_success(self, model: str, latency_ms: float):
"""Enregistre un succès pour le circuit breaker"""
cb = self.circuit_breakers.get(model)
if cb:
cb.success_count += 1
cb.total_latency += latency_ms
# Reset des échecs consécutifs après 10 succès
if cb.success_count >= 10:
cb.failure_count = max(0, cb.failure_count - 1)
async def call_with_fallback(
self,
prompt: str,
task_type: str = "general",
max_cost_usd: float = 0.10,
required_latency_ms: float = 500.0,
) -> Dict[str, Any]:
"""
Appelle les modèles en fallback jusqu'à succès ou épuisement
Args:
prompt: Le prompt à envoyer
task_type: Type de tâche (affecte le routing)
max_cost_usd: Coût maximum par requête
required_latency_ms: Latence maximale acceptable
Returns:
Dict avec 'success', 'response', 'model', 'latency_ms', 'cost_usd'
"""
self.metrics["total_requests"] += 1
# Déterminer l'ordre de tentative selon le type de tâche
models_to_try = self._get_model_priority(task_type, required_latency_ms)
last_error = None
for model in models_to_try:
# Vérifier circuit breaker
if not self._check_circuit_breaker(model):
self.logger.debug(f"Circuit breaker ouvert, skip {model}")
continue
# Estimer le coût
estimated_tokens = len(prompt.split()) * 2 # Rough estimate
estimated_cost = (estimated_tokens / 1_000_000) * MODEL_CONFIG[model]["cost_per_mtok"]
if estimated_cost > max_cost_usd:
self.logger.debug(f"Coût estimé trop élevé pour {model}")
continue
try:
start_time = time.time()
response = await self._call_model(model, prompt)
latency_ms = (time.time() - start_time) * 1000
# Vérifier la latence
if latency_ms > required_latency_ms:
self.logger.warning(
f"Latence {latency_ms}ms > {required_latency_ms}ms pour {model}"
)
# On continue mais on note la latence
# Calculer le coût réel
tokens_used = response.get("usage", {}).get("total_tokens", 0)
actual_cost = (tokens_used / 1_000_000) * MODEL_CONFIG[model]["cost_per_mtok"]
# Enregistrer le succès
self._record_success(model, latency_ms)
self.metrics["successful_requests"] += 1
self.metrics["cost_total_usd"] += actual_cost
return {
"success": True,
"response": response["choices"][0]["message"]["content"],
"model": model,
"latency_ms": round(latency_ms, 2),
"cost_usd": round(actual_cost, 6),
"tokens_used": tokens_used,
"fallback_attempts": len(models_to_try) - models_to_try.index(model),
}
except Exception as e:
last_error = str(e)
self.logger.error(f"Erreur avec {model}: {e}")
self._trip_circuit_breaker(model, str(e))
continue
# Tous les modèles ont échoué
self.metrics["failed_requests"] += 1
return {
"success": False,
"response": None,
"model": None,
"latency_ms": None,
"cost_usd": 0,
"error": last_error,
"fallback_attempts": len(models_to_try),
}
def _get_model_priority(
self,
task_type: str,
max_latency: float
) -> List[str]:
"""Détermine l'ordre de priorité des modèles selon la tâche"""
if task_type == "coding":
# Pour le code, on privilégie les modèles premium
return ["deepseek-v3.2", "gpt-4.1", "gemini-2.5-flash", "claude-sonnet-4.5"]
elif task_type == "fast_response":
# Pour les réponses rapides (chatbot)
return ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1", "claude-sonnet-4.5"]
elif task_type == "high_quality":
# Pour les tâches critiques (génération long-form)
return ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
else:
# Par défaut : économique d'abord si latence OK
return ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1", "claude-sonnet-4.5"]
async def _call_model(self, model: str, prompt: str) -> Dict[str, Any]:
"""Appelle l'API HolySheep pour un modèle donné"""
payload = {
"model": model,
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 2048,
}
response = await self.client.post("/chat/completions", json=payload)
response.raise_for_status()
return response.json()
def get_metrics(self) -> Dict[str, Any]:
"""Retourne les métriques de monitoring"""
return {
**self.metrics,
"circuit_breakers": {
model: {
"is_open": cb.is_open,
"failure_count": cb.failure_count,
"success_count": cb.success_count,
"avg_latency_ms": (
round(cb.total_latency / cb.success_count, 2)
if cb.success_count > 0 else None
),
}
for model, cb in self.circuit_breakers.items()
},
}
async def close(self):
"""Ferme proprement le client HTTP"""
await self.client.aclose()
============================================
EXEMPLE D'UTILISATION
============================================
async def main():
router = HolySheepMultiModelRouter(
api_key=HOLYSHEEP_CONFIG["api_key"],
enable_circuit_breaker=True,
)
try:
# Test avec fallback automatique
result = await router.call_with_fallback(
prompt="Explique la différence entre un circuit breaker et un retry pattern en microservices",
task_type="general",
max_cost_usd=0.05,
required_latency_ms=800.0,
)
if result["success"]:
print(f"✅ Réponse via {result['model']}")
print(f" Latence: {result['latency_ms']}ms")
print(f" Coût: ${result['cost_usd']}")
print(f" Fallback attempts: {result['fallback_attempts']}")
else:
print(f"❌ Échec après {result['fallback_attempts']} tentatives")
print(f" Erreur: {result['error']}")
# Afficher les métriques
print("\n📊 Métriques globales:")
metrics = router.get_metrics()
for key, value in metrics.items():
if key != "circuit_breakers":
print(f" {key}: {value}")
finally:
await router.close()
if __name__ == "__main__":
asyncio.run(main())
2. Script de Stress Test Complet
"""
HolySheep Stress Test Runner
Test de charge avec monitoring des limites de taux
"""
import asyncio
import time
import random
import statistics
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict
import json
Configuration du test
STRESS_TEST_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
# Paramètres de charge
"concurrent_users": 50, # Utilisateurs simultanés
"requests_per_user": 20, # Requêtes par utilisateur
"ramp_up_seconds": 10, # Temps de montée en charge
# Limites à tester
"rate_limit_rpm": 500, # Requêtes par minute attendues
"rate_limit_tpm": 150000, # Tokens par minute
# Fallback
"enable_fallback": True,
"fallback_models": ["deepseek-v3.2", "gemini-2.5-flash"],
}
@dataclass
class RequestResult:
"""Résultat d'une requête individuelles"""
request_id: str
model: str
success: bool
latency_ms: float
status_code: int
error: str = None
tokens_used: int = 0
cost_usd: float = 0.0
timestamp: float = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
class StressTestRunner:
"""
Runner de stress test pour HolySheep
Surveille les limites de taux et la résilience
"""
def __init__(self, config: Dict):
self.config = config
self.results: List[RequestResult] = []
self.start_time = None
self.end_time = None
# Compteurs de rate limiting
self.rate_limit_hits = {
"429": 0, # Too Many Requests
"rate_limit_exceeded": 0,
}
# Circuit breaker tracking
self.circuit_states = {}
async def _make_request(
self,
session: httpx.AsyncClient,
request_id: int,
prompt: str,
) -> RequestResult:
"""Exécute une requête individuelle avec timing"""
model = "gpt-4.1" # Modèle principal
start = time.time()
try:
response = await session.post(
f"{self.config['base_url']}/chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 500,
},
timeout=30.0,
)
latency = (time.time() - start) * 1000
if response.status_code == 429:
self.rate_limit_hits["429"] += 1
return RequestResult(
request_id=f"req_{request_id}",
model=model,
success=False,
latency_ms=latency,
status_code=429,
error="Rate limit exceeded",
)
response.raise_for_status()
data = response.json()
tokens = data.get("usage", {}).get("total_tokens", 0)
cost = (tokens / 1_000_000) * 8.0 # GPT-4.1 = $8/M
return RequestResult(
request_id=f"req_{request_id}",
model=model,
success=True,
latency_ms=latency,
status_code=200,
tokens_used=tokens,
cost_usd=cost,
)
except httpx.HTTPStatusError as e:
latency = (time.time() - start) * 1000
if e.response.status_code == 429:
self.rate_limit_hits["429"] += 1
return RequestResult(
request_id=f"req_{request_id}",
model=model,
success=False,
latency_ms=latency,
status_code=e.response.status_code,
error=str(e),
)
except Exception as e:
latency = (time.time() - start) * 1000
return RequestResult(
request_id=f"req_{request_id}",
model=model,
success=False,
latency_ms=latency,
status_code=0,
error=str(e),
)
async def _user_simulation(self, user_id: int, prompts: List[str]):
"""Simule le comportement d'un utilisateur"""
async with httpx.AsyncClient(
headers={
"Authorization": f"Bearer {self.config['api_key']}",
"Content-Type": "application/json",
},
limits=httpx.Limits(max_connections=10),
) as session:
for i, prompt in enumerate(prompts):
result = await self._make_request(
session,
f"u{user_id}_r{i}",
prompt,
)
self.results.append(result)
# Pause entre requêtes (simule le temps de lecture)
await asyncio.sleep(random.uniform(0.5, 2.0))
async def run(self):
"""Exécute le test de stress complet"""
print("🚀 Démarrage du stress test HolySheep")
print(f" Utilisateurs: {self.config['concurrent_users']}")
print(f" Requêtes/utilisateur: {self.config['requests_per_user']}")
print()
self.start_time = time.time()
# Générer les prompts de test
test_prompts = [
"Qu'est-ce que l'architecture microservices?",
"Explique le pattern CQRS en détail.",
"Comment implémenter un cache Redis?",
"Décris les avantages du Kubernetes.",
"Quelles sont les meilleures pratiques pour une API REST?",
] * 10 # Répéter pour avoir assez de prompts
random.shuffle(test_prompts)
# Créer les tâches utilisateur
tasks = []
for user_id in range(self.config["concurrent_users"]):
user_prompts = test_prompts[
user_id * self.config["requests_per_user"]:
(user_id + 1) * self.config["requests_per_user"]
]
tasks.append(self._user_simulation(user_id, user_prompts))
# Exécuter avec ramp-up progressif
print("⚡ Montée en charge...")
# Exécuter tous les utilisateurs en parallèle
await asyncio.gather(*tasks)
self.end_time = time.time()
# Afficher les résultats
self._print_results()
def _print_results(self):
"""Affiche les résultats du test"""
duration = self.end_time - self.start_time
successful = [r for r in self.results if r.success]
failed = [r for r in self.results if not r.success]
print("\n" + "=" * 60)
print("📊 RÉSULTATS DU STRESS TEST")
print("=" * 60)
print(f"\n⏱️ DURÉE TOTALE: {duration:.2f}s")
print(f"📨 TOTAL REQUÊTES: {len(self.results)}")
print(f" ✅ Réussies: {len(successful)} ({len(successful)/len(self.results)*100:.1f}%)")
print(f" ❌ Échouées: {len(failed)} ({len(failed)/len(self.results)*100:.1f}%)")
if successful:
latencies = [r.latency_ms for r in successful]
print(f"\n📈 LATENCE (requêtes réussies):")
print(f" Moyenne: {statistics.mean(latencies):.2f}ms")
print(f" Médiane: {statistics.median(latencies):.2f}ms")
print(f" P95: {sorted(latencies)[int(len(latencies) * 0.95)]:.2f}ms")
print(f" P99: {sorted(latencies)[int(len(latencies) * 0.99)]:.2f}ms")
print(f" Max: {max(latencies):.2f}ms")
print(f"\n🚦 LIMITE DE TAUX:")
print(f" Erreurs 429 (Too Many Requests): {self.rate_limit_hits['429']}")
print(f" Taux d'erreur rate limit: {self.rate_limit_hits['429']/len(self.results)*100:.2f}%")
total_cost = sum(r.cost_usd for r in successful)
total_tokens = sum(r.tokens_used for r in successful)
print(f"\n💰 COÛTS:")
print(f" Total tokens: {total_tokens:,}")
print(f" Coût total: ${total_cost:.4f}")
print(f" Coût/requête: ${total_cost/len(successful) if successful else 0:.6f}")
print(f"\n📊 REQUÊTES PAR SECONDE:")
print(f" RPS moyen: {len(self.results)/duration:.2f}")
# Évaluation du comportement sous charge
print("\n" + "=" * 60)
print("🎯 ÉVALUATION")
print("=" * 60)
success_rate = len(successful) / len(self.results) * 100
avg_latency = statistics.mean([r.latency_ms for r in successful]) if successful else 0
if success_rate >= 99:
print("🟢 EXCELLENT: Taux de succès > 99%")
elif success_rate >= 95:
print("🟡 BON: Taux de succès > 95%")
else:
print("🔴 ATTENTION: Taux de succès < 95%")
if avg_latency < 500:
print("🟢 EXCELLENT: Latence moyenne < 500ms")
elif avg_latency < 1000:
print("🟡 ACCEPTABLE: Latence moyenne < 1s")
else:
print("🔴 ATTENTION: Latence moyenne > 1s")
if self.rate_limit_hits["429"] == 0:
print("🟢 EXCELLENT: Aucune limite de taux atteinte")
else:
print(f"🟡 INFO: {self.rate_limit_hits['429']} requêtes limitées")
Exécuter le test
if __name__ == "__main__":
runner = StressTestRunner(STRESS_TEST_CONFIG)
asyncio.run(runner.run())
Système de Quotas et Gouvernance Multi-Équipe
"""
HolySheep Quota Manager - Gouvernance par équipe et projet
Implémente le rate limiting, les budgets et l'allocation dynamique
"""
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, Optional, List
from enum import Enum
import threading
class QuotaPeriod(Enum):
DAILY = "daily"
WEEKLY = "weekly"
MONTHLY = "monthly"
class QuotaExceededAction(Enum):
BLOCK = "block" # Bloquer immédiatement
QUEUE = "queue" # Mettre en file d'attente
FALLBACK = "fallback" # Utiliser un modèle moins cher
NOTIFY = "notify" #Notifier mais autoriser
@dataclass
class TeamQuota:
"""Configuration de quota pour une équipe"""
team_id: str
team_name: str
# Limites de budget
monthly_budget_usd: float = 1000.0
daily_budget_usd: float = 100.0
# Limites de volume
monthly_tokens: int = 10_000_000
daily_requests: int = 100_000
rpm_limit: int = 200 # Requêtes par minute
# Comportement en cas de dépassement
on_quota_exceeded: QuotaExceededAction = QuotaExceededAction.QUEUE
# Modèles autorisés
allowed_models: List[str] = field(default_factory=lambda: [
"gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"
])
# Préférence de modèle (pour fallback)
preferred_model: str = "gpt-4.1"
fallback_model: str = "deepseek-v3.2"
@dataclass
class UsageRecord:
"""Enregistrement d'utilisation"""
timestamp: datetime
tokens: int
cost_usd: float
model: str
request_id: str
class QuotaManager:
"""
Gestionnaire de quotas multi-équipes
Thread-safe pour environnements de production
"""
def __init__(self):
self._lock = threading.RLock()
# Configuration des équipes
self.teams: Dict[str, TeamQuota] = {}
# Historique d'utilisation
self.usage_history: Dict[str, List[UsageRecord]] = {}
# Contrôle de rate par minute
self.rpm_tracker: Dict[str, List[datetime]] = {}
# Cache des résultats récents (pour rate limiting)
self._recent_responses: Dict[str, List[datetime]] = {}
def register_team(self, team: TeamQuota):
"""Enregistre une nouvelle équipe avec ses quotas"""
with self._lock:
self.teams[team.team_id] = team
self.usage_history[team.team_id] = []
self.rpm_tracker[team.team_id] = []
print(f"✅ Équipe {team.team_name} enregistrée")
print(f" Budget mensuel: ${team.monthly_budget_usd}")
print(f" Limite RPM: {team.rpm_limit}")
def check_request_allowed(
self,
team_id: str,
estimated_tokens: int,
model: str,
) -> tuple[bool, Optional[str]]:
"""
Vérifie si une requête est autorisée selon les quotas
Returns:
(autorisé: bool, reason: str ou None)
"""
with self._lock:
team = self.teams.get(team_id)
if not team:
return False, f"Équipe {team_id} non trouvée"
now = datetime.now()
# 1. Vérifier le rate limit RPM
if not self._check_rpm_limit(team_id, team.rpm_limit):
return False, f"Rate limit RPM atteint ({team.rpm_limit}/min)"
# 2. Vérifier le budget quotidien
daily_spent = self._get_daily_spending(team_id)
estimated_cost = (estimated_tokens / 1_000_000) * self._get_model_cost(model)
if daily_spent + estimated_cost > team.daily_budget_usd:
return False, f"Budget quotidien dépassé (${daily_spent:.2f}/${team.daily_budget_usd})"
# 3. Vérifier le budget mensuel
monthly_spent = self._get_monthly_spending(team_id)
if monthly_spent + estimated_cost > team.monthly_budget_usd:
return False, f"Budget mensuel dépassé (${monthly_spent:.2f}/${team.monthly_budget_usd})"
# 4. Vérifier le modèle autorisé
if model not in team.allowed_models:
return False, f"Modèle {model} non autorisé pour cette équipe"
# 5. Vérifier la limite de tokens mensuels
monthly_tokens = self._get_monthly_tokens(team_id)
if monthly_tokens + estimated_tokens > team.monthlyly_tokens:
return False, f"Quota de tokens mensuel dépassé"
return True, None
def record_usage(
self,
team_id: str,
tokens: int,
cost_usd: float,
model: str,
request_id: str,
):
"""Enregistre l'utilisation après une requête"""
with self._lock:
record = UsageRecord(
timestamp=datetime.now(),
tokens=tokens,
cost_usd=cost_usd,
model=model,
request_id=request_id,
)
if team_id not in self.usage_history:
self.usage_history[team_id] = []
self.usage_history[team_id].append(record)