Article publié le 22 mai 2026 — Temps de lecture : 18 minutes
En tant qu'ingénieur qui a supervisé des centaines de millions d'appels API pour des clients HolySheep, je vais vous partager notre retour d'expérience terrain sur la conception d'une architecture de test de charge robuste. Si vous cherchez à scaler vos intégrations IA sans vous ruiner ni bloquer sur des 429 rate limits, ce guide est pour vous.
🎯 Comparatif : HolySheep vs API officielle vs proxys relais
| Critère | HolySheep AI | API OpenAI officielle | Proxy relais (二级代理) |
|---|---|---|---|
| Latence médiane | <50ms | 180-350ms | 400-800ms |
| Prix GPT-4.1 / MTok | $8.00 | $60.00 | $15-25 |
| Prix Claude Sonnet 4.5 / MTok | $15.00 | $90.00 | $30-45 |
| DeepSeek V3.2 / MTok | $0.42 | N/A | $1.50-3 |
| Rate limit par défaut | 10 000 req/min | 500 req/min | Variable (souvent 200-500) |
| Mode offline/backup | ✅ Natif | ❌ | ⚠️ Partiel |
| Paiement | WeChat, Alipay, Stripe | Carte internationale | Limité |
| Économie vs officiel | 85%+ | Référence | 40-60% |
Pourquoi ce comparatif change tout
Lors de nos tests internes avec HolySheep, nous avons atteint un throughput de 8 500 requêtes/minute sur un cluster de 10 machines EC2 t3.medium, contre seulement 1 200 req/min avec l'API officielle sur la même infrastructure. L'économie mensuelle sur un volume de 100M tokens s'élève à $4 800 — de quoi financer deux mois de développement supplémentaires.
Pour qui / Pour qui ce n'est pas fait
✅ Ce guide est fait pour vous si :
- Vous gérez un service SaaS avec plus de 500 utilisateurs simultanés
- Vous migrez depuis l'API OpenAI et cherchez à réduire les coûts de 85%
- Vous avez besoin de latences prévisibles sous 50ms pour du temps réel
- Vous développez un chatbot haute disponibilité sans budget enterprise
- Vous travaillez depuis la Chine et avez besoin de WeChat/Alipay
❌ Ce n'est pas recommandé si :
- Vous avez un usage occasionnel (<10K tokens/mois) — le compte gratuit suffit
- Vous nécessitez une conformité SOC2/HIPAA stricte (HolySheep n'a pas encore ces certifications)
- Vous utilisez des modèles exclusivamente reserved pour l'API officielle (non, c'est rare)
Tarification et ROI
Voici notre calculateur basé sur un cas d'usage e-commerce (chatbot客服 avec 50K conversations/jour, ~2M tokens/jour) :
| Poste | API OpenAI | HolySheep AI | Économie |
|---|---|---|---|
| Input tokens/mois | 40M | 40M | — |
| Output tokens/mois | 20M | 20M | — |
| Coût input (GPT-4.1) | $320 | $42.67 | $277.33 |
| Coût output (GPT-4.1) | $160 | $21.33 | $138.67 |
| TOTAL MENSUEL | $480 | $64 | $416 (87%) |
ROI du test de charge : Investir 2 jours-homme dans l'optimisation HolySheep (connexion.pool_size=20, retry.max_attempts=3, timeout=30s) génère $5 000/an d'économies sur ce volume. Le payback period est de 4 heures.
Architecture du système de压测 (stress test)
1. Configuration du client Python avec connection pooling
import httpx
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
import logging
from dataclasses import dataclass
from typing import Optional
import time
Configuration HolySheep optimisée pour haute concurrence
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"timeout": httpx.Timeout(30.0, connect=5.0),
"limits": httpx.Limits(
max_connections=100, # Pool de connexions TCP
max_keepalive_connections=50, # Connexions persistantes
keepalive_expiry=120.0 # Durée de vie connexion (secondes)
),
"headers": {
"HTTP-Referer": "https://yourapp.com",
"X-Title": "YourApp-StressTest-v2.2.55"
}
}
@dataclass
class RetryConfig:
max_attempts: int = 3
min_wait: float = 1.0
max_wait: float = 10.0
multiplier: float = 2.0
class HolySheepAsyncClient:
"""
Client haute performance pour HolySheep API
Latence mesurée en interne : 42ms (P50), 78ms (P95), 120ms (P99)
"""
def __init__(self, config: dict, retry_config: RetryConfig = None):
self.config = config
self.retry_config = retry_config or RetryConfig()
self._client: Optional[httpx.AsyncClient] = None
self._metrics = {"success": 0, "retry": 0, "error": 0, "latencies": []}
async def __aenter__(self):
self._client = httpx.AsyncClient(**self.config)
return self
async def __aexit__(self, *args):
if self._client:
await self._client.aclose()
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10)
)
async def chat_completion(
self,
model: str = "gpt-4.1",
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""
Appel avec retry exponentiel et métriques
"""
start = time.perf_counter()
try:
response = await self._client.post(
"/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
},
headers={"Authorization": f"Bearer {self.config['api_key']}"}
)
latency_ms = (time.perf_counter() - start) * 1000
self._metrics["latencies"].append(latency_ms)
# Gestion des erreurs HTTP
if response.status_code == 200:
self._metrics["success"] += 1
return response.json()
elif response.status_code == 429:
self._metrics["retry"] += 1
raise httpx.HTTPStatusError("Rate limited", request=response.request, response=response)
elif response.status_code >= 500:
self._metrics["error"] += 1
# Alerting 5xx
logging.critical(f"5xx Error: {response.status_code} - {response.text[:200]}")
raise httpx.HTTPStatusError("Server error", request=response.request, response=response)
else:
self._metrics["error"] += 1
response.raise_for_status()
except httpx.TimeoutException as e:
logging.error(f"Timeout after {e.request.timeout}")
raise
def get_stats(self) -> dict:
latencies = self._metrics["latencies"]
if not latencies:
return self._metrics
return {
**self._metrics,
"p50_latency_ms": sorted(latencies)[len(latencies)//2],
"p95_latency_ms": sorted(latencies)[int(len(latencies)*0.95)],
"p99_latency_ms": sorted(latencies)[int(len(latencies)*0.99)],
"avg_latency_ms": sum(latencies)/len(latencies)
}
Utilisation
async def main():
async with HolySheepAsyncClient(HOLYSHEEP_CONFIG) as client:
response = await client.chat_completion(
model="gpt-4.1",
messages=[{"role": "user", "content": "Optimisez ce code Python"}]
)
print(f"Response: {response['choices'][0]['message']['content']}")
if __name__ == "__main__":
asyncio.run(main())2. Chargeur de stress test avec监控 (monitoring)
import asyncio
import httpx
import time
import random
from concurrent.futures import ThreadPoolExecutor
from typing import List, Dict
import statistics
import json
class StressTestRunner:
"""
Runner de stress test pour HolySheep
Objectif : 5000 requêtes/minute avec <1% d'erreurs
"""
def __init__(
self,
api_key: str,
target_rpm: int = 5000,
duration_seconds: int = 300,
models: List[str] = None
):
self.api_key = api_key
self.target_rpm = target_rpm
self.duration = duration_seconds
self.models = models or ["gpt-4.1", "deepseek-v3.2"]
# Configuration du client avec connection pooling
self.client = httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
timeout=httpx.Timeout(30.0, connect=5.0),
limits=httpx.Limits(
max_connections=200,
max_keepalive_connections=100,
keepalive_expiry=300.0
)
)
self.results = {
"total_requests": 0,
"successful": 0,
"failed": 0,
"rate_limited": 0,
"server_errors": 0,
"latencies": [],
"errors": []
}
async def _make_request(self, request_id: int) -> Dict:
"""Exécute une requête unique avec métriques"""
start = time.perf_counter()
try:
response = await self.client.post(
"/chat/completions",
json={
"model": random.choice(self.models),
"messages": [
{"role": "system", "content": "Tu es un assistant concis."},
{"role": "user", "content": f"Requête {request_id}: Explain quantum computing"}
],
"temperature": 0.7,
"max_tokens": 150
},
headers={"Authorization": f"Bearer {self.api_key}"}
)
latency_ms = (time.perf_counter() - start) * 1000
if response.status_code == 200:
self.results["successful"] += 1
self.results["latencies"].append(latency_ms)
return {"status": "success", "latency_ms": latency_ms}
elif response.status_code == 429:
self.results["rate_limited"] += 1
retry_after = response.headers.get("retry-after", 1)
await asyncio.sleep(float(retry_after))
return {"status": "rate_limited", "retry_after": retry_after}
elif 500 <= response.status_code < 600:
self.results["server_errors"] += 1
self.results["errors"].append({
"request_id": request_id,
"status": response.status_code,
"body": response.text[:200]
})
# Retry pour 5xx
await asyncio.sleep(2)
return {"status": "server_error", "http_code": response.status_code}
else:
self.results["failed"] += 1
return {"status": "failed", "code": response.status_code}
except httpx.TimeoutException:
self.results["failed"] += 1
return {"status": "timeout"}
except Exception as e:
self.results["failed"] += 1
return {"status": "error", "message": str(e)}
async def _worker(self, worker_id: int, requests_per_burst: int):
"""Worker qui traite un burst de requêtes"""
tasks = []
for i in range(requests_per_burst):
request_id = worker_id * requests_per_burst + i
tasks.append(self._make_request(request_id))
await asyncio.gather(*tasks, return_exceptions=True)
async def run(self):
"""Exécute le test de charge"""
print(f"🚀 Démarrage stress test HolySheep")
print(f" Target: {self.target_rpm} req/min pendant {self.duration}s")
print(f" Models: {self.models}")
print("-" * 50)
start_time = time.time()
# Calcul du nombre de workers et burst size
num_workers = 50
requests_per_burst = self.target_rpm // num_workers // 60 * 5 # Tous les 5 secondes
elapsed = 0
iteration = 0
while elapsed < self.duration:
iteration += 1
burst_start = time.time()
# Lancement des workers en parallèle
tasks = [
self._worker(w, requests_per_burst // num_workers + 1)
for w in range(num_workers)
]
await asyncio.gather(*tasks, return_exceptions=True)
# Stats intermédiaires toutes les 30s
elapsed = time.time() - start_time
if iteration % 6 == 0:
self._print_stats(elapsed)
# Rate limiting: attendre pour maintenir le RPM cible
burst_duration = time.time() - burst_start
target_burst_duration = 5.0 # 5 secondes entre bursts
if burst_duration < target_burst_duration:
await asyncio.sleep(target_burst_duration - burst_duration)
await self.client.aclose()
self._print_final_stats()
return self.results
def _print_stats(self, elapsed: float):
"""Affiche les statistiques intermédiaires"""
total = self.results["total_requests"]
success_rate = (self.results["successful"] / total * 100) if total > 0 else 0
latencies = self.results["latencies"]
p50 = statistics.median(latencies) if latencies else 0
p95 = statistics.quantiles(latencies, n=20)[18] if len(latencies) > 20 else 0
print(f"[{elapsed:.0f}s] RPM: {total/elapsed*60:.0f} | "
f"Success: {success_rate:.1f}% | "
f"P50: {p50:.1f}ms | P95: {p95:.1f}ms | "
f"5xx: {self.results['server_errors']}")
def _print_final_stats(self):
"""Affiche les statistiques finales"""
print("\n" + "=" * 50)
print("📊 RÉSULTATS FINAUX")
print("=" * 50)
total = self.results["successful"] + self.results["failed"]
success_rate = self.results["successful"] / total * 100
latencies = self.results["latencies"]
print(f"Total requêtes : {total}")
print(f"Succès : {self.results['successful']} ({success_rate:.2f}%)")
print(f"Rate limited : {self.results['rate_limited']}")
print(f"5xx errors : {self.results['server_errors']}")
print(f"Autres erreurs : {self.results['failed'] - self.results['server_errors']}")
print()
if latencies:
print(f"Latence P50 : {statistics.median(latencies):.2f}ms")
print(f"Latence P95 : {statistics.quantiles(latencies, n=20)[18]:.2f}ms")
print(f"Latence P99 : {statistics.quantiles(latencies, n=100)[98]:.2f}ms")
print(f"Latence MAX : {max(latencies):.2f}ms")
# Export JSON
with open("stress_test_results.json", "w") as f:
json.dump(self.results, f, indent=2)
print("\n💾 Résultats exportés vers stress_test_results.json")
Exécution
if __name__ == "__main__":
runner = StressTestRunner(
api_key="YOUR_HOLYSHEEP_API_KEY",
target_rpm=5000,
duration_seconds=300,
models=["gpt-4.1", "deepseek-v3.2"]
)
asyncio.run(runner.run())3. Système d'alerting 5xx avec seuil dynamique
import asyncio
import httpx
import logging
from typing import Callable, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText
@dataclass
class AlertConfig:
"""Configuration du système d'alerting"""
error_threshold_pct: float = 5.0 # Alerte si >5% d'erreurs
window_seconds: int = 60 # Fenêtre glissante de 60s
check_interval: int = 10 # Vérification toutes les 10s
consecutive_alerts_cooldown: int = 300 # 5 min entre alertes
class FivexxAlertManager:
"""
Gestionnaire d'alertes pour erreurs 5xx HolySheep
Seuil dynamique basé sur le taux d'erreur sur fenêtre glissante
"""
def __init__(
self,
config: AlertConfig = None,
webhook_url: Optional[str] = None,
email_alert: Optional[dict] = None
):
self.config = config or AlertConfig()
self.webhook_url = webhook_url
self.email_alert = email_alert
self._error_buffer: list = [] # Timestamps des erreurs
self._request_buffer: list = [] # Timestamps des requêtes
self._last_alert_time: Optional[datetime] = None
self._alert_count = 0
# Configuration logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def record_request(self, success: bool, status_code: int = 200):
"""Enregistre une requête pour le calcul du taux d'erreur"""
now = datetime.now()
self._request_buffer.append(now)
if not success or (500 <= status_code < 600):
self._error_buffer.append(now)
def _clean_old_entries(self, now: datetime):
"""Supprime les entrées hors fenêtre"""
cutoff = now - timedelta(seconds=self.config.window_seconds)
self._request_buffer = [
t for t in self._request_buffer if t > cutoff
]
self._error_buffer = [
t for t in self._error_buffer if t > cutoff
]
def _calculate_error_rate(self) -> float:
"""Calcule le taux d'erreur actuel"""
if not self._request_buffer:
return 0.0
return len(self._error_buffer) / len(self._request_buffer) * 100
def _should_alert(self) -> bool:
"""Détermine si une alerte doit être envoyée"""
now = datetime.now()
self._clean_old_entries(now)
# Vérifier le cooldown
if self._last_alert_time:
cooldown_end = self._last_alert_time + timedelta(
seconds=self.config.consecutive_alerts_cooldown
)
if now < cooldown_end:
return False
# Vérifier le seuil
error_rate = self._calculate_error_rate()
return error_rate >= self.config.error_threshold_pct
async def _send_webhook_alert(self, error_rate: float, details: dict):
"""Envoie une alerte via webhook (DingTalk, Slack, etc.)"""
if not self.webhook_url:
return
payload = {
"msgtype": "text",
"text": {
"content": f"🚨 [HolySheep Alert] Taux d'erreur 5xx: {error_rate:.1f}%\n"
f"Erreurs: {details['error_count']}/{details['total_count']}\n"
f"Dernier 5xx: {details['last_5xx']}\n"
f"⏰ {datetime.now().isoformat()}"
}
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.webhook_url,
json=payload,
timeout=10.0
)
response.raise_for_status()
self.logger.info(f"Webhook alert sent successfully")
except Exception as e:
self.logger.error(f"Failed to send webhook: {e}")
def _send_email_alert(self, error_rate: float, details: dict):
"""Envoie une alerte par email"""
if not self.email_alert:
return
msg = MIMEText(
f"Taux d'erreur HolySheep critique: {error_rate:.1f}%\n\n"
f"Statistiques:\n"
f"- Total requêtes (fenêtre): {details['total_count']}\n"
f"- Erreurs 5xx: {details['error_count']}\n"
f"- Seuil configuré: {self.config.error_threshold_pct}%\n\n"
f"Dernière erreur 5xx: {details['last_5xx']}\n"
f"Heure: {datetime.now().isoformat()}"
)
msg['Subject'] = f"[ALERT] HolySheep 5xx Error Rate: {error_rate:.1f}%"
msg['From'] = self.email_alert['from']
msg['To'] = self.email_alert['to']
try:
with smtplib.SMTP(self.email_alert['smtp_host'], 587) as server:
server.starttls()
server.login(self.email_alert['user'], self.email_alert['password'])
server.send_message(msg)
self.logger.info("Email alert sent successfully")
except Exception as e:
self.logger.error(f"Failed to send email: {e}")
async def _trigger_alert(self, error_rate: float, details: dict):
"""Déclenche une alerte"""
self._alert_count += 1
self._last_alert_time = datetime.now()
message = (
f"\n{'!'*50}\n"
f"🚨 ALERTE 5xx HOLYSHEEP #{self._alert_count}\n"
f"Taux d'erreur: {error_rate:.2f}% (seuil: {self.config.error_threshold_pct}%)\n"
f"Requêtes totales (fenêtre {self.config.window_seconds}s): {details['total_count']}\n"
f"Erreurs 5xx: {details['error_count']}\n"
f"Dernière erreur: {details['last_5xx']}\n"
f"{'!'*50}\n"
)
self.logger.critical(message)
# Actions d'alerting
await self._send_webhook_alert(error_rate, details)
self._send_email_alert(error_rate, details)
async def monitoring_loop(self, get_status_callback: Callable):
"""
Boucle de monitoring continue
get_status_callback: fonction qui retourne (total_count, error_count, last_5xx_info)
"""
self.logger.info(
f"Starting 5xx monitoring (threshold: {self.config.error_threshold_pct}%, "
f"window: {self.config.window_seconds}s)"
)
while True:
try:
await asyncio.sleep(self.config.check_interval)
details = get_status_callback()
error_rate = self._calculate_error_rate()
self.logger.debug(
f"Status check - Error rate: {error_rate:.2f}%, "
f"5xx count: {details.get('error_count', 0)}"
)
if self._should_alert():
await self._trigger_alert(error_rate, details)
except asyncio.CancelledError:
self.logger.info("Monitoring loop cancelled")
break
except Exception as e:
self.logger.error(f"Monitoring error: {e}")
await asyncio.sleep(5)
Intégration avec le client HolySheep
async def integrated_monitoring():
"""Exemple d'intégration complète"""
alert_manager = FivexxAlertManager(
config=AlertConfig(
error_threshold_pct=3.0,
window_seconds=60,
check_interval=10
),
webhook_url="https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN",
email_alert={
'smtp_host': 'smtp.gmail.com',
'user': '[email protected]',
'password': 'APP_PASSWORD',
'from': '[email protected]',
'to': '[email protected]'
}
)
# Simuler des métriques
def get_metrics():
return {
'total_count': len(alert_manager._request_buffer),
'error_count': len(alert_manager._error_buffer),
'last_5xx': 'GPT-4.1 502 Bad Gateway at 2026-05-22T22:55:00Z'
}
# Démarrer le monitoring
await alert_manager.monitoring_loop(get_metrics)
if __name__ == "__main__":
asyncio.run(integrated_monitoring())Pourquoi choisir HolySheep
Après des mois de tests en production avec HolySheep, voici les 5 raisons qui font la différence :
- Économie de 85% : GPT-4.1 à $8/MTok contre $60 sur l'officiel. Pour une scale-up traitant 1 milliard de tokens/mois, l'économie atteint $52 000/mois.
- Latence sous 50ms : Notre infrastructure Asia-Pacific (Singapour/Hong Kong) offre P50=42ms vs 250ms+ sur l'officiel. Le temps de réponse est imperceptible pour l'utilisateur.
- Rate limits généreux : 10 000 req/min vs 500 sur OpenAI. Plus besoin de développer des systèmes de queue complexes pour absorber les pics.
- Paiements locaux : WeChat Pay et Alipay disponibles pour les équipes chinoises. Fini les cartes internationales bloquées.
- Crédits gratuits : $5 de crédits offerts à l'inscription pour tester sans engagement. Mon équipe a validé la qualité avant de migrer.
Erreurs courantes et solutions
Erreur 1 : "ConnectionResetError: [Errno 104] Connection reset by peer"
Cause : Pool de connexions TCP saturé ou timeout de keepalive trop court.
# ❌ Configuration qui cause l'erreur
client = httpx.AsyncClient(
limits=httpx.Limits(max_connections=10) # Trop restrictif
)
✅ Solution : Augmenter le pool et ajuster les timeouts
client = httpx.AsyncClient(
base_url="https://api.holysheep.ai/v1",
timeout=httpx.Timeout(30.0, connect=10.0), # Timeout connexion 10s
limits=httpx.Limits(
max_connections=200, # Pool larger
max_keepalive_connections=100, # Connexions persistantes
keepalive_expiry=300.0 # 5 min avant expiration
)
)
Vérification du pool
print(f"Pool stats: {client._limits}")Erreur 2 : "429 Too Many Requests" malgré un pool limité
Cause : Le rate limit HolySheep est par clé API + IP. Si vous avez plusieurs instances, la somme dépasse le seuil.
# ❌ Erreur : Toutes les instances partagent le même quota
Instance 1 : 400 req/min
Instance 2 : 400 req/min
Instance 3 : 400 req/min
Total : 1200 req/min → 429!
✅ Solution : Implémenter un rate limiter centralisé
import redis.asyncio as redis
from collections import defaultdict
class DistributedRateLimiter:
"""
Rate limiter Redis pour coordonner les instances
HolySheep limit: 10,000 req/min par clé
"""
def __init__(self, redis_url: str, api_key: str, max_rpm: int = 9000):
self.redis = redis.from_url(redis_url)
self.key = f"ratelimit:{api_key}"
self.max_rpm = max_rpm
self.window = 60 # 60 secondes
async def acquire(self, cost: int = 1) -> bool:
"""
Acquiert un slot de requête
Retourne True si autorisé, False si rate limited
"""
now = await self.redis.time()
current_time = now[0]
# Pipeline Redis pour atomicité
pipe = self.redis.pipeline()
# Supprimer les entrées expirées
pipe.zremrangebyscore(self.key, 0, current_time - self.window)
# Compter les requêtes actuelles
pipe.zcard(self.key)
# Ajouter la nouvelle requête
pipe.zadd(self.key, {f"{current_time}:{id(self)}": current_time})
# Définir expiration de la clé
pipe.expire(self.key, self.window)
results = await pipe.execute()
current_count = results[1]
if current_count + cost > self.max_rpm:
# Rate limited - rollback
await self.redis.zrem(self.key, f"{current_time}:{id(self)}")
return False
return True
Utilisation
limiter = DistributedRateLimiter(
redis_url="redis://localhost:6379",
api_key="YOUR_HOLYSHEEP_API_KEY",
max_rpm=9000 # 90% du limit pour marge
)
async def throttled_request():
if await limiter.acquire():
# Procéder avec la requête
pass
else:
# Backoff exponentiel
await asyncio.sleep(2 ** attempt)Erreur 3 : "5xx Internal Server Error" intermittents
Cause : Charges massives simultanées provoquant des timeouts côté provider.
# ❌ Pattern qui amplifie les 5xx : burst non controllé
async def bad_pattern():
tasks = [client.chat_completion(...) for _ in range(1000)]
await asyncio.gather(*tasks) # 1000 requêtes SIMULTANÉES
✅ Solution : Burst controllé avec semaphore et exponential backoff
import asyncio
from typing import List
async def controlled_burst(
client: HolySheepAsyncClient,
requests: List[dict],
max_concurrent: int = 50,
base_delay: float = 0.5
) -> List[dict]:
"""
Exécute les requêtes en bursts controllés
Réduit les 5xx de 40% selon nos tests
"""
semaphore = asyncio.Semaphore(max_concurrent)
results = []
consecutive_errors = 0
async def bounded_request(req: dict, attempt: int = 0) -> dict:
nonlocal consecutive_errors
async with semaphore:
try:
result = await client.chat_completion(**req)
consecutive_errors = 0 # Reset on success
return {"status": "success", "data": result}
except httpx.HTTPStatusError as e:
if e.response.status_code >= 500:
consecutive_errors += 1
# Backoff