En tant qu'ingénieur DevOps avec 12 ans d'expérience dans l'automatisation industrielle, j'ai déployé des dizaines de systèmes de contrôle qualité. Ce qui me frustrait le plus ? La fragmentation des outils IA et les coûts explosifs lors des pics de production. En mars 2026, j'ai migré notre ligne de装配 électronique de Hanwang vers HolySheep AI. Aujourd'hui, je vous montre exactement comment construire une passerelle de质检 visuelle industrielle robuste avec leur API, réduisant nos coûts de 85% tout en maintenant un temps de réponse sous 50ms.
🎯 Cas d'Utilisation Réel : E-commerce de Pièces Automobiles avec Pic Saisonnier
Imaginons DrAutoParts GmbH, un fournisseur de pièces automobiles en ligne. Chaque novembre, pendant les promotions du Black Friday, leur volume de commandes quadruple. Leur système de contrôle qualité doit inspecter :
- 100 000 images/jour en période normale
- 400 000+ images/jour pendant les pics
- Détection de 23 types de défauts (rayures, déformations, mauvaise couleur)
- Retour client sous 200ms maximum
- Disponibilité SLA 99.9%
Avec l'architecture traditionnelle (API OpenAI + serveur on-premise), ce pic nécessitait 47 serveurs Kubernetes coûtant 12 000€/mois. Avec HolySheep AI et leur passerelle de limitation intelligente, 3 instances suffisent pour 2 800€/mois — une économie de 77%.
Architecture de la Passerelle API HolySheep pour le Contrôle Qualité
Schéma Global du Système
┌─────────────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE HOLYSHEEP AI GATEWAY │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌─────────────────┐ ┌──────────────────────────┐ │
│ │ Cameras 4K │───▶│ Pre-processing │───▶│ HolySheep API Gateway │ │
│ │ (25 cams) │ │ (resize/format) │ │ base_url: api.holysheep │ │
│ └──────────────┘ └─────────────────┘ │ .ai/v1 │ │
│ │ │ │
│ │ ┌────────────────────┐ │ │
│ │ │ Rate Limiter │ │ │
│ │ │ • Token Bucket │ │ │
│ │ │ • Exponential Back │ │ │
│ │ │ • Circuit Breaker │ │ │
│ │ └────────────────────┘ │ │
│ │ │ │
│ │ ┌────────────────────┐ │ │
│ │ │ Gemini Multimodal │ │ │
│ │ │ Review (v2.5) │ │ │
│ │ │ 128K context │ │ │
│ │ └────────────────────┘ │ │
│ └──────────┬───────────────┘ │
│ │ │
│ ┌──────────────┐ ┌─────────────────┐ │ │
│ │ Dashboard │◀───│ Prometheus + │◀───────────────┘ │
│ │ Grafana SLA │ │ AlertManager │ │
│ └──────────────┘ └─────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
Implémentation Complète : Gateway Python avec HolySheep
# holy_sheep_qa_gateway.py
Passerelle de contrôle qualité visuel industriel
Auteur: Équipe HolySheep AI - Version 2.2.53
import asyncio
import aiohttp
import base64
import hashlib
import time
import json
import logging
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
from collections import defaultdict
import redis.asyncio as redis
Configuration HolySheep - BASE_URL OBLIGATOIRE
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé
=== CONFIGURATION ÉTENDUE ===
@dataclass
class RateLimitConfig:
"""Configuration des limites de taux HolySheep"""
requests_per_minute: int = 500 # Limite par défaut HolySheep
tokens_per_minute: int = 150_000 # Pour images: basé sur les tokens visuels
burst_size: int = 50 # Taille du burst autorisé
retry_base_delay: float = 1.0 # Délai initial entre réessais (secondes)
retry_max_delay: float = 60.0 # Délai maximum entre réessais
max_retries: int = 5 # Nombre maximum de réessais
@dataclass
class SLAConfig:
"""Configuration SLA pour le monitoring"""
target_availability: float = 99.9 # Pourcentage
max_latency_p95: float = 150.0 # Latence P95 en ms
max_error_rate: float = 0.01 # Taux d'erreur maximum (1%)
alert_cooldown: int = 300 # Secondes entre alertes
@dataclass
class QualityCheckResult:
"""Résultat d'une inspection de qualité"""
image_id: str
is_accepted: bool
confidence: float
defect_types: List[str] = field(default_factory=list)
processing_time_ms: float = 0.0
tokens_used: int = 0
cost_usd: float = 0.0
cached: bool = False
retry_count: int = 0
class CircuitState(Enum):
"""États du disjoncteur"""
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Bloquant les requêtes
HALF_OPEN = "half_open" # Test de reprise
class HolySheepQAGateway:
"""
Passerelle de contrôle qualité visuel industriel
Basée sur HolySheep AI API avec limitation intelligente et retry
"""
def __init__(
self,
api_key: str,
rate_limit: RateLimitConfig = None,
sla_config: SLAConfig = None,
redis_url: str = "redis://localhost:6379"
):
self.api_key = api_key
self.base_url = BASE_URL
self.rate_limit = rate_limit or RateLimitConfig()
self.sla_config = sla_config or SLAConfig()
# Compteurs de taux
self.request_counts = defaultdict(lambda: {"count": 0, "reset_at": 0})
self.token_counts = defaultdict(lambda: {"count": 0, "reset_at": 0})
# Disjoncteur
self.circuit_state = CircuitState.CLOSED
self.circuit_failure_count = 0
self.circuit_open_time = 0
self.circuit_recovery_timeout = 30 # Secondes avant test
# Cache de résultats
self._cache = {}
self._redis = None
self._redis_url = redis_url
# Métriques SLA
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"retried_requests": 0,
"cached_results": 0,
"latencies": [],
"errors_by_type": defaultdict(int),
"circuit_trips": 0
}
self.logger = logging.getLogger("HolySheepQAGateway")
async def initialize(self):
"""Initialisation asynchrone"""
try:
self._redis = await redis.from_url(self._redis_url)
self.logger.info("✓ Connexion Redis établie")
except Exception as e:
self.logger.warning(f"Redis non disponible, mode fallback: {e}")
async def close(self):
"""Fermeture propre des connexions"""
if self._redis:
await self._redis.close()
# === GESTION DU TAUX ET RETRY ===
async def _check_rate_limit(self, client_id: str) -> bool:
"""Vérifie et applique la limitation de taux"""
current_time = time.time()
window_start = int(current_time // 60) * 60 # Fenêtre de 1 minute
# Vérification requests/minute
req_counter = self.request_counts[client_id]
if req_counter["reset_at"] != window_start:
req_counter["count"] = 0
req_counter["reset_at"] = window_start
if req_counter["count"] >= self.rate_limit.requests_per_minute:
wait_time = 60 - (current_time - window_start)
self.logger.warning(f"Rate limit atteint pour {client_id}, attente {wait_time:.1f}s")
await asyncio.sleep(max(0.1, wait_time))
return False
req_counter["count"] += 1
return True
def _get_cache_key(self, image_data: bytes, model: str) -> str:
"""Génère une clé de cache pour les images"""
content_hash = hashlib.sha256(image_data).hexdigest()[:16]
return f"qa:result:{model}:{content_hash}"
async def _check_cache(self, cache_key: str) -> Optional[Dict]:
"""Vérifie le cache pour des résultats précédents"""
if self._redis:
cached = await self._redis.get(cache_key)
if cached:
self.metrics["cached_results"] += 1
return json.loads(cached)
elif cache_key in self._cache:
self.metrics["cached_results"] += 1
return self._cache[cache_key]
return None
async def _save_to_cache(self, cache_key: str, result: Dict, ttl: int = 3600):
"""Sauvegarde un résultat en cache"""
try:
if self._redis:
await self._redis.setex(cache_key, ttl, json.dumps(result))
else:
self._cache[cache_key] = result
except Exception as e:
self.logger.warning(f"Échec sauvegarde cache: {e}")
# === CIRCUIT BREAKER ===
def _should_allow_request(self) -> bool:
"""Détermine si une requête doit être autorisée selon le disjoncteur"""
if self.circuit_state == CircuitState.CLOSED:
return True
if self.circuit_state == CircuitState.OPEN:
if time.time() - self.circuit_open_time > self.circuit_recovery_timeout:
self.circuit_state = CircuitState.HALF_OPEN
self.logger.info("Circuit breaker: passage en HALF_OPEN")
return True
return False
# HALF_OPEN: une seule requête de test
return True
def _record_success(self):
"""Enregistre un succès pour le disjoncteur"""
if self.circuit_state == CircuitState.HALF_OPEN:
self.circuit_state = CircuitState.CLOSED
self.circuit_failure_count = 0
self.logger.info("Circuit breaker: retour à CLOSED")
def _record_failure(self):
"""Enregistre un échec pour le disjoncteur"""
self.circuit_failure_count += 1
if self.circuit_state == CircuitState.HALF_OPEN:
self.circuit_state = CircuitState.OPEN
self.circuit_open_time = time.time()
self.metrics["circuit_trips"] += 1
self.logger.error("Circuit breaker: OPEN après échec HALF_OPEN")
elif self.circuit_failure_count >= 5:
self.circuit_state = CircuitState.OPEN
self.circuit_open_time = time.time()
self.metrics["circuit_trips"] += 1
self.logger.error(f"Circuit breaker: OPEN après {self.circuit_failure_count} échecs")
# === APPEL API HOLYSHEEP ===
async def _call_holy_sheep_api(
self,
image_data: bytes,
model: str = "gemini-2.5-flash",
defect_categories: List[str] = None
) -> Dict[str, Any]:
"""
Appel à l'API HolySheep pour analyse visuelle
Utilise le endpoint multimodal pour inspection industrielle
"""
# Encodage de l'image en base64
image_base64 = base64.b64encode(image_data).decode('utf-8')
# Construction du prompt industriel
defect_list = defect_categories or [
"rayure", "déformation", "tache", "mauvais_alignement",
"défaut_couleur", "fissure", "texture_anormale", "contamination"
]
prompt = f"""Analyse de contrôle qualité industriel.
Inspecter l'image pour détecter les défauts suivants:
{', '.join(defect_list)}
Répondre au format JSON:
{{
"is_accepted": boolean,
"confidence": float (0-1),
"defects": [
{{
"type": string,
"severity": "minor|moderate|critical",
"location": "approx zone dans l'image",
"confidence": float
}}
],
"summary": "résumé en français"
}}"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": prompt
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"temperature": 0.1, # Faible température pour cohérence
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 429:
raise RateLimitError("Limitation de débit HolySheep atteinte")
elif response.status == 401:
raise AuthenticationError("Clé API invalide")
elif response.status != 200:
text = await response.text()
raise APIError(f"Erreur API: {response.status} - {text}")
result = await response.json()
return result
# === FONCTION PRINCIPALE DE QUALITY CHECK ===
async def inspect_image(
self,
image_data: bytes,
image_id: str = None,
client_id: str = "default",
defect_categories: List[str] = None,
use_cache: bool = True
) -> QualityCheckResult:
"""
Effectue une inspection de qualité sur une image
Args:
image_data: Données binaires de l'image
image_id: Identifiant unique de l'image
client_id: Identifiant du client pour le rate limiting
defect_categories: Catégories de défauts à détecter
use_cache: Utiliser le cache pour les images similaires
Returns:
QualityCheckResult avec verdict et métadonnées
"""
start_time = time.time()
image_id = image_id or hashlib.md5(image_data).hexdigest()[:12]
# Vérification rate limit
await self._check_rate_limit(client_id)
# Vérification disjoncteur
if not self._should_allow_request():
raise CircuitOpenError("Circuit breaker ouvert")
# Vérification cache
cache_key = self._get_cache_key(image_data, "gemini-2.5-flash")
if use_cache:
cached = await self._check_cache(cache_key)
if cached:
return QualityCheckResult(
image_id=image_id,
is_accepted=cached["is_accepted"],
confidence=cached["confidence"],
defect_types=[d["type"] for d in cached.get("defects", [])],
processing_time_ms=(time.time() - start_time) * 1000,
cached=True
)
# Retry avec backoff exponentiel
last_error = None
for attempt in range(self.rate_limit.max_retries):
try:
self.metrics["total_requests"] += 1
result = await self._call_holy_sheep_api(
image_data,
"gemini-2.5-flash",
defect_categories
)
# Extraction du contenu
content = result["choices"][0]["message"]["content"]
usage = result.get("usage", {})
# Parsing du JSON dans la réponse
try:
import re
json_match = re.search(r'\{.*\}', content, re.DOTALL)
parsed = json.loads(json_match.group()) if json_match else {}
except:
parsed = {"is_accepted": True, "confidence": 0.95}
# Calcul du coût (tarifs HolySheep 2026)
tokens_used = usage.get("total_tokens", 0)
cost_usd = tokens_used * (2.50 / 1_000_000) # Gemini 2.5 Flash: $2.50/M tokens
processing_time = (time.time() - start_time) * 1000
quality_result = QualityCheckResult(
image_id=image_id,
is_accepted=parsed.get("is_accepted", True),
confidence=parsed.get("confidence", 0.95),
defect_types=[d["type"] for d in parsed.get("defects", [])],
processing_time_ms=processing_time,
tokens_used=tokens_used,
cost_usd=cost_usd,
retry_count=attempt
)
# Mise à jour disjoncteur
self._record_success()
# Sauvegarde cache
await self._save_to_cache(cache_key, parsed)
self.metrics["successful_requests"] += 1
self.metrics["latencies"].append(processing_time)
return quality_result
except RateLimitError as e:
self.logger.warning(f"Tentative {attempt + 1}: {e}")
self.metrics["retried_requests"] += 1
# Backoff exponentiel avec jitter
delay = min(
self.rate_limit.retry_base_delay * (2 ** attempt),
self.rate_limit.retry_max_delay
)
delay *= (0.5 + asyncio.random() * 0.5) # Jitter
await asyncio.sleep(delay)
last_error = e
except (APIError, aiohttp.ClientError) as e:
self.logger.error(f"Tentative {attempt + 1} échouée: {e}")
self._record_failure()
self.metrics["failed_requests"] += 1
self.metrics["errors_by_type"][type(e).__name__] += 1
last_error = e
if attempt < self.rate_limit.max_retries - 1:
await asyncio.sleep(self.rate_limit.retry_base_delay * (2 ** attempt))
# Toutes les tentatives ont échoué
raise InspectionError(f"Échec après {self.rate_limit.max_retries} tentatives: {last_error}")
# === SURVEILLANCE SLA ===
def get_sla_metrics(self) -> Dict[str, Any]:
"""Calcule et retourne les métriques SLA"""
total = self.metrics["total_requests"]
if total == 0:
return {"status": "no_data"}
success_rate = self.metrics["successful_requests"] / total
error_rate = self.metrics["failed_requests"] / total
cache_rate = self.metrics["cached_results"] / total
latencies = sorted(self.metrics["latencies"])
p50 = latencies[len(latencies) // 2] if latencies else 0
p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
p99 = latencies[int(len(latencies) * 0.99)] if latencies else 0
avg_latency = sum(latencies) / len(latencies) if latencies else 0
# Calcul disponibilité
sla_violation = (
error_rate > self.sla_config.max_error_rate or
p95 > self.sla_config.max_latency_p95
)
return {
"availability": f"{(1 - error_rate) * 100:.3f}%",
"sla_status": "✓ OK" if not sla_violation else "✗ VIOLATION",
"success_rate": f"{success_rate * 100:.2f}%",
"error_rate": f"{error_rate * 100:.3f}%",
"cache_hit_rate": f"{cache_rate * 100:.2f}%",
"latency": {
"avg_ms": f"{avg_latency:.1f}",
"p50_ms": f"{p50:.1f}",
"p95_ms": f"{p95:.1f}",
"p99_ms": f"{p99:.1f}"
},
"circuit_breaker_state": self.circuit_state.value,
"circuit_trips": self.metrics["circuit_trips"],
"total_requests": total,
"sla_target": f"{self.sla_config.target_availability}%",
"latency_target": f"{self.sla_config.max_latency_p95}ms (P95)"
}
def reset_metrics(self):
"""Réinitialise les compteurs de métriques"""
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"retried_requests": 0,
"cached_results": 0,
"latencies": [],
"errors_by_type": defaultdict(int),
"circuit_trips": 0
}
=== CLASSES D'ERREURS PERSONNALISÉES ===
class RateLimitError(Exception):
"""Erreur de limitation de débit"""
pass
class AuthenticationError(Exception):
"""Erreur d'authentification"""
pass
class APIError(Exception):
"""Erreur générale de l'API"""
pass
class CircuitOpenError(Exception):
"""Disjoncteur ouvert - service temporairement indisponible"""
pass
class InspectionError(Exception):
"""Erreur lors de l'inspection"""
pass
=== FONCTION D'UTILISATION PRINCIPALE ===
async def main():
"""Exemple d'utilisation de la passerelle"""
gateway = HolySheepQAGateway(
api_key="YOUR_HOLYSHEEP_API_KEY",
rate_limit=RateLimitConfig(
requests_per_minute=500,
max_retries=5
)
)
await gateway.initialize()
# Simulation d'images de contrôle qualité
sample_image = b"..." # Données image réelles
try:
result = await gateway.inspect_image(
image_data=sample_image,
image_id="PART-12345-2026-05-21",
client_id="drautoparts-prod"
)
print(f"Résultat inspection:")
print(f" - Accepté: {result.is_accepted}")
print(f" - Confiance: {result.confidence:.2%}")
print(f" - Défauts: {result.defect_types}")
print(f" - Latence: {result.processing_time_ms:.1f}ms")
print(f" - Coût: ${result.cost_usd:.6f}")
except CircuitOpenError:
print("⚠ Service temporairement indisponible - réessayez plus tard")
except InspectionError as e:
print(f"✗ Erreur inspection: {e}")
# Affichage métriques SLA
print("\n📊 Métriques SLA:")
for key, value in gateway.get_sla_metrics().items():
print(f" {key}: {value}")
await gateway.close()
if __name__ == "__main__":
asyncio.run(main())
Comparatif : HolySheep AI vs Solutions Concurrentes (Mai 2026)
| Critère | HolySheep AI | OpenAI GPT-4.1 | Anthropic Claude Sonnet 4.5 | Google Gemini 2.5 Flash (direct) |
|---|---|---|---|---|
| Prix par Million de Tokens | $2.50 (DeepSeek V3.2: $0.42) | $8.00 | $15.00 | $2.50 |
| Latence Moyenne (P95) | <50ms | 850ms | 1200ms | 200ms |
| Multi-modalité Native | ✓ Oui | ✓ Oui | ✓ Oui | ✓ Oui |
| Gateway Rate Limiting | ✓ Intelligent & configurable | ✗ Basique | ✗ Basique | ✗ Limité |
| Circuit Breaker | ✓ Intégré | ✗ Non | ✗ Non | ✗ Non |
| Monitoring SLA | ✓ Dashboard intégré | ✗ Externe requis | ✗ Externe requis | ✗ Externe requis |
| Mode Sans Facture (Rate Limits) | ✓ 500 req/min | 3 req/min | 5 req/min | 15 req/min |
| Paiements Locaux | ✓ WeChat/Alipay | ✗ Carte internationale | ✗ Carte internationale | ✗ Carte internationale |
| Crédits Gratuits | ✓ Inclus | $5 | $5 | $300 (Google Cloud) |
| Économie vs OpenAI | 85%+ | - | -47% | +69% |
Pour Qui / Pour Qui Ce N'est Pas Fait
✓ HolySheep AI est parfait pour :
- PME industrielles souhaitant automatiser le contrôle qualité sans équipe IA dédiée
- Startups e-commerce avec pics de volume prévisibles (soldes, Black Friday)
- Développeurs indépendants construisant des applications de vision par ordinateur
- Entreprises chinoises nécessitant WeChat Pay ou Alipay pour les paiements
- Équipes avec budget limité mais exigeant une latence <50ms
- Projets RAG combinant analyse d'images et génération de texte
✗ HolySheep AI n'est pas idéal pour :
- Grandes entreprises avec SLA contractuels stricts (préférer Azure AI ou AWS Rekognition)
- Cas d'usage nécessitant une résidence des données en Europe (GDPR strict)
- Applications temps réel critiques (<10ms - nécessiteraitedge computing)
- Organisations、政府客户 avec restrictions sur les API externes
Implémentation Avancée : Batch Processing et监控 Dashboard
# holy_sheep_batch_processor.py
Traitement par lots pour inspection industrielle de masse
import asyncio
import aiohttp
import json
import logging
from typing import List, Dict, Optional, Callable
from dataclasses import dataclass
from datetime import datetime
import hashlib
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class BatchConfig:
"""Configuration du traitement par lots"""
batch_size: int = 10 # Images par lot
max_concurrent_batches: int = 5 # Lots parallèles max
max_retries: int = 3
retry_delay: float = 2.0
timeout_per_image: float = 8.0 # Secondes
@dataclass
class BatchResult:
"""Résultat d'un lot"""
batch_id: str
total_images: int
accepted: int
rejected: int
processing_time_seconds: float
total_cost_usd: float
failed_images: List[str]
errors: List[str]
class HolySheepBatchProcessor:
"""
Processeur de lots pour inspection visuelle industrielle
Optimisé pour les pics de production
"""
def __init__(self, api_key: str, config: BatchConfig = None):
self.api_key = api_key
self.base_url = BASE_URL
self.config = config or BatchConfig()
self.logger = logging.getLogger("BatchProcessor")
self._semaphore = asyncio.Semaphore(self.config.max_concurrent_batches)
async def _process_single_image(
self,
session: aiohttp.ClientSession,
image_data: bytes,
image_id: str,
defect_categories: List[str],
semaphore: asyncio.Semaphore
) -> Dict:
"""Traite une image unique avec retry"""
for attempt in range(self.config.max_retries):
try:
async with semaphore:
result = await self._call_api(
session, image_data, image_id, defect_categories
)
return {
"image_id": image_id,
"status": "success",
"result": result
}
except Exception as e:
if attempt == self.config.max_retries - 1:
return {
"image_id": image_id,
"status": "failed",
"error": str(e),
"attempts": attempt + 1
}
await asyncio.sleep(self.config.retry_delay * (2 ** attempt))
return {"image_id": image_id, "status": "failed", "error": "Max retries"}
async def _call_api(
self,
session: aiohttp.ClientSession,
image_data: bytes,
image_id: str,
defect_categories: List[str]
) -> Dict:
"""Appel API pour une image"""
import base64
image_base64 = base64.b64encode(image_data).decode('utf-8')
prompt = f"""Contrôle qualité industriel - Inspection {image_id}
Catégories de défauts à détecter: {', '.join(defect_categories)}
Format JSON obligatoire:
{{
"image_id": "{image_id}",
"is_accepted": boolean,
"confidence": float,
"defects": [{{"type": string, "severity": string}}],
"notes": string
}}"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.5-flash",
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
]
}],
"temperature": 0.1,
"max_tokens": 1024
}
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=self.config.timeout_per_image)
) as response:
if response.status == 429:
raise aiohttp.ClientResponseError(
request_info=None,
history=None,
status=429,
message="Rate limited"
)
result = await response.json()
content = result["choices"][0]["message"]["content"]
# Parsing JSON robuste
import re
json_match = re.search(r'\{.*\}', content, re.DOTALL)
if json_match:
parsed = json.loads(json_match.group())
else:
parsed = {"is_accepted": True, "confidence": 0.95, "defects": []}
return parsed
async def process_batch(
self,
images: List[Dict], # [{"id": str, "data": bytes}]
defect_categories: List[str],
progress_callback: Optional[Callable] = None
) -> BatchResult:
"""
Traite un lot d'images en parallèle
Args:
images: Liste de dictionnaires avec 'id' et 'data'
defect_categories: Catégories de défauts
progress_callback: Fonction appelée avec (current, total)
"""
batch_id = hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]
start_time = asyncio.get_event_loop().time()
self.logger.info(f"Début traitement lot {batch_id}: {len(images)} images")
connector = aiohttp.TCPConnector(limit=self.config.max_concurrent_batches * 2)
async with aiohttp.ClientSession(connector=connector) as session:
# Création des tâches
semaphore = asyncio.Semaphore(self.config.batch_size)
tasks = [
self._process_single_image(
session,
img["data"],
img["id"],
defect_categories,
semaphore
)
for img in images
]
# Exécution avec suivi du progrès
results = []
for i, coro in enumerate(asyncio.as_completed(tasks)):
result = await coro
results.append(result)
if progress_callback and (i + 1) % 10 == 0:
progress_callback(i + 1, len(images))
# Calcul des statistiques
accepted = sum(1 for r in results if r["status"] == "success" and r["result"].get("is_accepted"))
rejected = sum(1 for r in results if r["status"]