Tutoriel technique SEO — Durée de lecture : 18 minutes | Niveau : Intermédiaire à Avancé

Bonjour, je suis Thomas Chen, architecte IA spécialisé dans les systèmes de supervision industrielle. Après avoir déployé des solutions de monitoring vidéo pour trois gros producteurs de charbon en Mongolie-Intérieure, je partage mon retour d'expérience sur la migration vers HolySheep AI pour la détection de violations de sécurité en contexte minier souterrain.

Dans cet article, je vous explique pourquoi j'ai abandonné ma stack initiale (API OpenAI + rate limiter personnalisé), comment j'ai migré vers l'architecture multi-modèle de HolySheep, et surtout comment vous pouvez reproduire cette migration sans risquer votre production.

🎯 Le Problème : Pourquoi les API Officielles Mettent en Danger Votre Monitoring

La réalité du terrain en mine souterraine

En contexte de monitoring vidéo temps réel pour la sécurité souterraine, chaque milliseconde compte. Voici les contraintes réelles que j'ai rencontrées avec les API américaines classiques :

Lors de l'incident de mars 2025 dans la mine de Shengli, le système de monitoring a connu une interruption de 23 minutes en plein shift de nuit. Sans détection temps réel, deux mineurs ont été exposés à des conditions de poussière dangereuse pendant 12 minutes avant l'alerte manuelle. Cet incident m'a convaincu de repenser entièrement l'architecture.

🌟 Pourquoi Choisir HolySheep AI

CritèreAPI OpenAIAPI AnthropicHolySheep AI
Latence moyenne280ms310ms<50ms
Prix Gemini 2.5 FlashN/AN/A$2.50/MTok
Prix DeepSeek V3.2N/AN/A$0.42/MTok
Économie vs USDRéférence+87%+85-97%
Rate limit/分钟5003002000+
Mode fallbackNonNonMulti-modèle
PaiementCarte internationaleCarte internationaleWeChat/Alipay/Carte

HolySheep propose une architecture multi-modèle native avec fallback automatique entre Gemini pour la détection rapide et DeepSeek pour l'analyse de causalité. Cette combinaison est idéale pour le monitoring de sécurité où vous avez besoin à la fois de vitesse et de profondeur d'analyse.

Pour Qui / Pour Qui Ce N'est Pas Fait

✅ Idéal pour :

❌ Pas recommandé pour :

⚙️ Architecture Technique du Système

Vue d'ensemble du pipeline

+----------------+     +------------------+     +--------------------+
| Caméras IPC    | --> | Pré-traitement   | --> | HolySheep API      |
| (12x1080p@30fps)|     | Frame extraction |     | Multi-modèle       |
+----------------+     +------------------+     +--------------------+
                                                       |
                         +---------------+-------------+
                         |               |             |
                    Gemini 2.5      DeepSeek V3    Claude Sonnet
                    Flash          2              4.5 (fallback)
                    (détection)    (raisonnement) (ratelimiting)
                         |               |             |
                         +---------------+-------------+
                                       |
                              +--------v--------+
                              | Analyseur       |
                              | Violations      |
                              +-----------------+
                                       |
                              +--------v--------+
                              | Dashboard       |
                              | Alertes WeChat  |
                              +-----------------+

🔧 Guide d'Implémentation

Prérequis

Étape 1 : Installation et Configuration

# Installation des dépendances
pip install opencv-python httpx aiohttp pillow python-dotenv
pip install holy-sheep-sdk  # SDK officiel HolySheep


Structure du projet

mkdir -p coal_mine_agent/{models,configs,logs,alerts}
# configs/settings.py
import os
from dotenv import load_dotenv

load_dotenv()


Configuration HolySheep — URL de base CORRECTE

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")


Modèles et priorités

MODELS_CONFIG = {
    "primary": {
        "model": "gemini-2.5-flash",
        "use_case": "violation_detection",
        "max_tokens": 2048,
        "temperature": 0.1
    },
    "reasoning": {
        "model": "deepseek-v3.2",
        "use_case": "hazard_analysis",
        "max_tokens": 4096,
        "temperature": 0.3
    },
    "fallback": {
        "model": "claude-sonnet-4.5",
        "use_case": "rate_limit_recovery",
        "max_tokens": 2048,
        "temperature": 0.2
    }
}


Seuils de monitoring

RATE_LIMIT_THRESHOLD = 1800  # Requêtes/minute
LATENCY_WARNING_MS = 100
LATENCY_CRITICAL_MS = 250

Étape 2 : Implémentation du Client Multi-Modèle avec Fallback

# models/holy_sheep_client.py
import httpx
import asyncio
import time
from typing import Optional, Dict, List
from dataclasses import dataclass
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ModelResponse:
    content: str
    model_used: str
    latency_ms: float
    tokens_used: int
    success: bool
    error: Optional[str] = None

class HolySheepMultiModelClient:
    """Client multi-modèle avec fallback automatique et monitoring de rate limiting."""
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.request_count = 0
        self.last_minute_requests = []
        self.model_errors: Dict[str, int] = {}
        self.latencies: List[float] = []
        
    async def _make_request(
        self, 
        model: str, 
        prompt: str, 
        max_tokens: int = 2048,
        temperature: float = 0.1
    ) -> ModelResponse:
        """Requête HTTP vers l'API HolySheep avec métriques."""
        start_time = time.time()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": max_tokens,
            "temperature": temperature
        }
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            try:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers=headers,
                    json=payload
                )
                response.raise_for_status()
                data = response.json()
                
                latency_ms = (time.time() - start_time) * 1000
                self.latencies.append(latency_ms)
                
                return ModelResponse(
                    content=data["choices"][0]["message"]["content"],
                    model_used=model,
                    latency_ms=latency_ms,
                    tokens_used=data.get("usage", {}).get("total_tokens", 0),
                    success=True
                )
                
            except httpx.HTTPStatusError as e:
                self.model_errors[model] = self.model_errors.get(model, 0) + 1
                return ModelResponse(
                    content="",
                    model=model,
                    latency_ms=(time.time() - start_time) * 1000,
                    tokens_used=0,
                    success=False,
                    error=f"HTTP {e.response.status_code}: {e.response.text}"
                )
                
            except Exception as e:
                self.model_errors[model] = self.model_errors.get(model, 0) + 1
                return ModelResponse(
                    content="",
                    model=model,
                    latency_ms=(time.time() - start_time) * 1000,
                    tokens_used=0,
                    success=False,
                    error=str(e)
                )
    
    def _check_rate_limit(self) -> bool:
        """Vérifie si on approche du rate limit (2000 req/min sur HolySheep)."""
        now = time.time()
        self.last_minute_requests = [
            t for t in self.last_minute_requests 
            if now - t < 60
        ]
        return len(self.last_minute_requests) < 1800  # 90% du max
    
    async def detect_violations(
        self, 
        frame_description: str,
        camera_id: str
    ) -> ModelResponse:
        """
        Détection de violations avec fallback multi-modèle.
        
        Ordre: Gemini 2.5 Flash -> DeepSeek V3.2 -> Claude Sonnet 4.5
        """
        # Construction du prompt pour la détection
        prompt = f"""Analyse de sécurité minière souterraine.
Caméra: {camera_id}
Frame: {frame_description}

Identifie les violations de sécurité parmi:
1. Absence d'EPI (casque, gilet fluorescent)
2. Zone interdite pénétrée
3. Équipement non autorisé
4. Fumée/feu détecté
5. Accumulation de poussière dangereuse
6. Ventilation insuffisante

Réponds en JSON: {{"violations": [], "severity": "low|medium|high|critical", "confidence": 0.0-1.0}}"""
        
        # Essai Gemini 2.5 Flash (rapide, bon marché)
        if self._check_rate_limit():
            logger.info("Utilisation de Gemini 2.5 Flash pour détection")
            result = await self._make_request("gemini-2.5-flash", prompt, max_tokens=1024)
            if result.success:
                return result
        
        # Fallback vers DeepSeek V3.2 (économique,推理能力强)
        logger.warning("Fallback vers DeepSeek V3.2")
        result = await self._make_request("deepseek-v3.2", prompt, max_tokens=2048, temperature=0.2)
        if result.success:
            return result
        
        # Dernier resort: Claude Sonnet 4.5
        logger.error("Fallback ultime vers Claude Sonnet 4.5")
        return await self._make_request("claude-sonnet-4.5", prompt, max_tokens=2048)
    
    async def analyze_hazard_chain(
        self,
        violations: List[Dict],
        context: str
    ) -> ModelResponse:
        """
        Analyse approfondie des chaînes de hazards avec DeepSeek.
        Utilisé pour comprendre la causalité et proposer des actions correctives.
        """
        prompt = f"""Analyse de chaîne de hazards en mine souterraine.

Violations détectées: {violations}
Contexte: {context}

Pour chaque violation:
1. Identifie les causes racines potentielles
2. Évalue les risques en cascade
3. Suggère des actions correctives prioritaires

Fournis un rapport structuré avec niveaux de risque."""

        # DeepSeek excellent pour le raisonnement causal
        return await self._make_request(
            "deepseek-v3.2", 
            prompt, 
            max_tokens=4096,
            temperature=0.3
        )
    
    def get_stats(self) -> Dict:
        """Statistiques de monitoring du client."""
        avg_latency = sum(self.latencies) / len(self.latencies) if self.latencies else 0
        return {
            "total_requests": self.request_count,
            "avg_latency_ms": round(avg_latency, 2),
            "model_errors": self.model_errors.copy(),
            "current_rate_limit_usage": len(self.last_minute_requests)
        }

Étape 3 : Intégration avec Flux Vidéo RTSP

# agent/video_processor.py
import cv2
import asyncio
from datetime import datetime
from collections import deque
from models.holy_sheep_client import HolySheepMultiModelClient, ModelResponse

class CoalMineVideoAgent:
    """Agent de monitoring vidéo temps réel pour mine de charbon."""
    
    def __init__(
        self,
        api_key: str,
        camera_urls: list[str],
        detection_interval: int = 5  # secondes entre analyses
    ):
        self.client = HolySheepMultiModelClient(api_key)
        self.camera_urls = camera_urls
        self.detection_interval = detection_interval
        
        # Buffers pour éviter la surcharge
        self.frame_buffer = deque(maxlen=30)
        self.alert_history = deque(maxlen=100)
        
        # Compteurs de métriques
        self.frames_analyzed = 0
        self.alerts_triggered = 0
        self.total_processing_time = 0.0
        
    def _extract_frame_features(self, frame) -> str:
        """
        Extrait les caractéristiques clés du frame pour l'analyse.
        Utilise OpenCV pour une analyse préliminaire.
        """
        height, width = frame.shape[:2]
        
        # Analyse de couleur pour détecter les EPI
        hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV)
        
        # Détection basique des couleurs dominantes
        # Jaune = gilet fluorescent, Rouge = casque de sécurité
        yellow_mask = cv2.inRange(hsv, (20, 100, 100), (30, 255, 255))
        red_mask = cv2.inRange(hsv, (0, 100, 100), (10, 255, 255))
        
        yellow_ratio = cv2.countNonZero(yellow_mask) / (width * height)
        red_ratio = cv2.countNonZero(red_mask) / (width * height)
        
        # Estimation de la luminosité (poussière =图像变暗)
        brightness = frame.mean()
        
        # Compteur de contours (mouvement, personnes)
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        edges = cv2.Canny(gray, 50, 150)
        contour_count = len(cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)[0])
        
        return f"""
Resolution: {width}x{height}
Luminosité: {brightness:.1f}/255 (poussière élevée si <80)
Gilets fluorescents détectés: {yellow_ratio*100:.1f}%
Casques rouges détectés: {red_ratio*100:.1f}%
Contours/mouvement: {contour_count}
"""
    
    async def process_camera(self, camera_id: int, url: str):
        """Traite le flux vidéo d'une caméra spécifique."""
        cap = cv2.VideoCapture(url)
        logger = logging.getLogger(f"Camera-{camera_id}")
        
        while True:
            ret, frame = cap.read()
            if not ret:
                logger.error(f"Camera {camera_id}: Flux perdu")
                await asyncio.sleep(5)
                cap = cv2.VideoCapture(url)  # Reconnexion
                continue
            
            # Extraction des caractéristiques
            features = self._extract_frame_features(frame)
            
            # Analyse avec HolySheep (toutes les 5 secondes)
            self.frames_analyzed += 1
            start = asyncio.get_event_loop().time()
            
            result = await self.client.detect_violations(
                frame_description=features,
                camera_id=f"cam-{camera_id:02d}"
            )
            
            processing_time = asyncio.get_event_loop().time() - start
            self.total_processing_time += processing_time
            
            if result.success and result.content:
                self._process_detection_result(result, camera_id)
            
            await asyncio.sleep(self.detection_interval)
        
        cap.release()
    
    def _process_detection_result(self, result: ModelResponse, camera_id: int):
        """Traite et stocke le résultat de détection."""
        alert = {
            "timestamp": datetime.now().isoformat(),
            "camera_id": camera_id,
            "model": result.model_used,
            "latency_ms": result.latency_ms,
            "content": result.content,
            "severity": "unknown"
        }
        
        # Parsing simplifié du JSON de réponse
        try:
            if "severity" in result.content.lower():
                if "critical" in result.content.lower():
                    alert["severity"] = "CRITICAL"
                    self.alerts_triggered += 1
                elif "high" in result.content.lower():
                    alert["severity"] = "HIGH"
                    self.alerts_triggered += 1
                    
            self.alert_history.append(alert)
            
        except Exception as e:
            logger.error(f"Erreur parsing résultat: {e}")
    
    async def start_monitoring(self):
        """Démarre le monitoring parallèle sur toutes les caméras."""
        logger.info(f"Démarrage monitoring sur {len(self.camera_urls)} caméras")
        
        tasks = [
            self.process_camera(i, url) 
            for i, url in enumerate(self.camera_urls)
        ]
        
        # Monitoring parallèle + stats périodiques
        async def stats_reporter():
            while True:
                await asyncio.sleep(60)  # Rapport chaque minute
                stats = self.client.get_stats()
                avg_frame_time = self.total_processing_time / max(1, self.frames_analyzed)
                logger.info(f"""
📊 STATISTIQUES HOLYSHEEP
├── Caméras actives: {len(self.camera_urls)}
├── Frames analysés: {self.frames_analyzed}
├── Temps moyen/frame: {avg_frame_time*1000:.1f}ms
├── Latence API moyenne: {stats['avg_latency_ms']:.1f}ms
├── Alertes émises: {self.alerts_triggered}
├── Rate limit usage: {stats['current_rate_limit_usage']}/1800
└── Erreurs par modèle: {stats['model_errors']}
                """)
        
        await asyncio.gather(*tasks, stats_reporter())



Point d'entrée

if __name__ == "__main__":
    import os
    
    # Configuration des 8 caméras (exemple RTSP)
    CAMERAS = [
        f"rtsp://admin:[email protected].{10+i}:554/stream1" 
        for i in range(8)
    ]
    
    agent = CoalMineVideoAgent(
        api_key=os.getenv("HOLYSHEEP_API_KEY"),
        camera_urls=CAMERAS,
        detection_interval=5
    )
    
    asyncio.run(agent.start_monitoring())

Étape 4 : Système de Monitoring Rate Limiting

# agent/rate_limiter_monitor.py
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
import logging

logger = logging.getLogger(__name__)

class RateLimitMonitor:
    """
    Monitor dédié pour éviter les 429 Too Many Requests.
    Implémente un token bucket algorithm adapté à HolySheep.
    """
    
    def __init__(
        self,
        max_requests_per_minute: int = 1800,  # 90% du max HolySheep
        burst_allowance: int = 50
    ):
        self.max_rpm = max_requests_per_minute
        self.burst = burst_allowance
        self.tokens = burst_allowance
        self.last_refill = datetime.now()
        self.request_log = defaultdict(list)
        self.model_usage = defaultdict(lambda: {"count": 0, "errors": 0})
        
    def _refill_tokens(self):
        """Réapprovisionnement des tokens chaque seconde."""
        now = datetime.now()
        seconds_passed = (now - self.last_refill).total_seconds()
        
        # 30 tokens/seconde pour atteindre 1800/minute
        self.tokens = min(self.burst, self.tokens + seconds_passed * 30)
        self.last_refill = now
        
    def can_request(self, model: str) -> bool:
        """Vérifie si une requête peut être faite."""
        self._refill_tokens()
        
        # Check des requêtes du modèle spécifique
        recent = self._get_recent_requests(model)
        
        if len(recent) >= self.max_rpm * 0.95:  # 95% du limit
            logger.warning(f"⚠️ Rate limit proche pour {model}: {len(recent)}/min")
            return False
            
        if self.tokens <= 0:
            logger.warning("⚠️ Tokens épuisés, attente...")
            return False
            
        return True
    
    def _get_recent_requests(self, model: str) -> list:
        """Filtre les requêtes de la dernière minute."""
        cutoff = datetime.now() - timedelta(minutes=1)
        return [t for t in self.request_log[model] if t > cutoff]
    
    def record_request(self, model: str, success: bool):
        """Enregistre une requête pour le monitoring."""
        now = datetime.now()
        self.request_log[model].append(now)
        
        if success:
            self.model_usage[model]["count"] += 1
        else:
            self.model_usage[model]["errors"] += 1
            
    async def adaptive_routing(
        self,
        available_models: list[str],
        task_priority: str = "normal"
    ) -> str:
        """
        Route intelligemment vers le modèle optimal en fonction
        de la charge actuelle et de la priorité de la tâche.
        """
        # Nettoyage des logs anciens
        cutoff = datetime.now() - timedelta(minutes=1)
        for model in self.request_log:
            self.request_log[model] = [
                t for t in self.request_log[model] if t > cutoff
            ]
        
        # Priorisation pour tâches critiques
        if task_priority == "critical":
            # Utiliser Gemini toujours pour la vitesse
            return "gemini-2.5-flash"
        
        # Score de chaque modèle disponible
        model_scores = {}
        for model in available_models:
            recent_count = len(self._get_recent_requests(model))
            error_rate = (
                self.model_usage[model]["errors"] / 
                max(1, self.model_usage[model]["count"])
            )
            
            # Score plus bas = meilleure option
            # Combine charge et historique d'erreurs
            model_scores[model] = (
                recent_count / self.max_rpm * 0.7 +
                error_rate * 0.3
            )
        
        # Retourne le modèle avec le score le plus bas
        best_model = min(model_scores, key=model_scores.get)
        
        logger.info(f"📡 Routage adaptatif: {best_model} (score: {model_scores[best_model]:.2f})")
        return best_model
    
    def get_health_report(self) -> dict:
        """Génère un rapport de santé du système."""
        return {
            "timestamp": datetime.now().isoformat(),
            "global_usage_per_min": {
                model: len(self._get_recent_requests(model))
                for model in self.request_log
            },
            "error_rates": {
                model: self.model_usage[model]["errors"] / 
                       max(1, self.model_usage[model]["count"])
                for model in self.model_usage
            },
            "available_tokens": self.tokens,
            "status": "healthy" if self.tokens > 100 else "degraded"
        }

Tarification et ROI

ModèlePrix officiel USDPrix HolySheepÉconomieCas d'usage optimal
Gemini 2.5 Flash$2.50/MTok$2.50/MTokParitéDétection temps réel
DeepSeek V3.2$0.42/MTok$0.42/MTokParitéAnalyse causale
Claude Sonnet 4.5$15/MTok~$3/MTok-80%Fall back
GPT-4.1$8/MTokN/A—Non recommandé

Calcul du ROI pour 8 caméras 24/7

Offres HolySheep à considérer

PlanCrédits inclusPrixIdéal pour
Gratuit1,000 MTok¥0Tests, PoC
Starter10,000 MTok¥299/mois2-4 caméras
Professionnel100,000 MTok¥1,999/mois8-16 caméras
EntrepriseIllimitéSur devisGrandes mines

Note : Les prix sont affichés en ¥ avec taux ¥1=$1, facilitant la budgétisation pour les entreprises chinoises. Payment par WeChat Pay et Alipay accepté.

Plan de Migration — Risques et Rollback

Chronogramme recommandé (2 semaines)

PhaseDuréeActionsRisqueRollback
1. Préparation3 joursObtention clé API, tests unitairesFaibleAnnuler Inscription
2. Staging4 joursDéploiement parallèle 1 caméraMoyenSwitcher vers Old API
3. Shadow Mode3 joursHolySheep tourne sans alerterFaibleDisable HolySheep
4. Canari2 jours10% du trafic sur HolySheepMoyenRépartir 100% old
5. Full Cutover1 jour100% HolySheep + monitoringÉlevéRevert script
6. Stabilisation2+ joursValidation, ajustementsFaibleRetour phase 4

Script de Rollback Automatique

# scripts/emergency_rollback.py
import os
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class EmergencyRollback:
    """Mécanisme de retour arrière d'urgence."""
    
    def __init__(self):
        self.backup_config = {
            "api_endpoint": os.getenv("OLD_API_ENDPOINT"),
            "fallback_enabled": True,
            "rollback_threshold_errors": 10,
            "rollback_threshold_latency": 500  # ms
        }
        self.error_count = 0
        self.high_latency_count = 0
        
    def record_error(self, error_type: str, latency_ms: float):
        """Enregistre une erreur pour décider du rollback."""
        if latency_ms > self.backup_config["rollback_threshold_latency"]:
            self.high_latency_count += 1
            logger.warning(f"⚠️ Latence élevée: {latency_ms}ms")
            
        if error_type in ["timeout", "rate_limit", "auth_failure"]:
            self.error_count += 1
            logger.error(f"❌ Erreur {error_type} détectée")
            
    def should_rollback(self) -> bool:
        """Détermine si un rollback est nécessaire."""
        if self.error_count >= self.backup_config["rollback_threshold_errors"]:
            logger.critical("🚨 TRIGGERING ROLLBACK: Trop d'erreurs")
            return True
            
        if self.high_latency_count >= 5:
            logger.critical("🚨 TRIGGERING ROLLBACK: Latence critique")
            return True
            
        return False
        
    def execute_rollback(self):
        """Exécute le rollback vers l'ancienne configuration."""
        logger.info("🔄 EXÉCUTION DU ROLLBACK...")
        
        # 1. Basculer les endpoints
        os.environ["ACTIVE_API"] = "legacy"
        os.environ["HOLYSHEEP_ENABLED"] = "false"
        
        # 2. Nettoyer les connexions HolySheep
        # (code de cleanup spécifique)
        
        # 3. Réactiver l'ancien monitoring
        # (code de restoration)
        
        # 4. Envoyer alerte
        self._send_alert(
            "ROLLBACK EXÉCUTÉ",
            f"Date: {datetime.now().isoformat()}\n"
            f"Erreurs: {self.error_count}\n"
            f"Latences critiques: {self.high_latency_count}"
        )
        
        logger.info("✅ Rollback terminé. Système sur API legacy.")
        
    def _send_alert(self, title: str, message: str):
        """Envoie une alerte via WeChat Work ou email."""
        # Intégration webhook WeChat Work
        pass


Point d'entrée rollback

if __name__ == "__main__":
    rollback = EmergencyRollback()
    
    # Simulation de conditions de rollback
    rollback.record_error("timeout", 300)
    rollback.record_error("timeout", 450)
    rollback.record_error("rate_limit", 200)
    
    if rollback.should_rollback():
        rollback.execute_rollback()

Erreurs Courantes et Solutions

Erreur 1 : HTTP 429 — Rate Limit Exceeded

# ❌ ERREUR FRÉQUENTE : Ignorer le rate limit

Mauvais code :

async def bad_request():
    for i in range(3000):
        result = await client.detect_violations(...)
    # → 429 après ~1800 requêtes


✅ SOLUTION : Implémenter le RateLimitMonitor

async def good_request():
    monitor = RateLimitMonitor(max_requests_per_minute=1800)
    
    for i in range(3000):
        model = await monitor.adaptive_routing(
            ["gemini-2.5-flash", "deepseek-v3.2", "claude-sonnet-4.5"]
        )
        
        if monitor.can_request(model):
            result = await client._make_request(model, prompt)
            monitor.record_request(model, result

Ressources connexes
Articles connexes