Auteur : Équipe HolySheep AI | Date : 25 mai 2026 | Lecture : 18 minutes
Dans cet article, je vais vous guider pas à pas dans la création d'un système d'agent IA complet pour la détection de fuites dans les réseaux de chauffage urbain. Nous utiliserons l'API HolySheep — accessible ici avec des crédits gratuits — pour implémenter une architecture résiliente combinant détection d'anomalies par GPT-4.1, distribution de tickets par DeepSeek V3.2, et gestion intelligente des pannes avec retry automatique et failover.
📋 Prérequis et environnement
Avant de commencer, assure-toi d'avoir :
- Un compte HolySheep AI (inscription gratuite avec crédits offerts)
- Python 3.10+ installé
- La bibliothèque
requests(pip install requests) - Basic understanding d'API REST (je t'explique tout)
💡 Note personnelle : Quand j'ai déployé mon premier agent de détection de fuite pour un réseau de 47 km à Shenyang, le plus gros défi n'était pas le code, mais la gestion des faux positifs. Les capteurs IoT envoient parfois des données bruitées lors de pics de température hivernaux. C'est pourquoi j'ai conçu ce système avec triple validation avant toute alerte.
🏗️ Architecture du système
Notre système repose sur trois piliers fondamentaux :
- Module de détection : Analyse les données des capteurs via GPT-4.1 avec scoring d'anomalie
- Module de distribution : Assigne automatiquement les tickets aux équipes via DeepSeek V3.2
- Module de résilience : Gère les SLA avec retry exponentiel et failover entre modèles
┌─────────────────────────────────────────────────────────────────┐
│ ARCHITECTURE HOLYSHEEP AGENT │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Capteurs IoT ──► Zone de préparation ──► Agent de Détection │
│ │ (batch 5min) │ │
│ │ ▼ │
│ │ ┌─────────────────────┐ │
│ │ │ GPT-4.1 Scoring │ │
│ │ │ (anomaly_score) │ │
│ │ └──────────┬──────────┘ │
│ │ │ │
│ │ score > 0.7 │ score ≤ 0.7 │
│ │ ▼ │
│ │ ┌─────────────────────┐ │
│ │ │ DeepSeek V3.2 │ │
│ │ │ Work Order Router │ │
│ │ └──────────┬──────────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌─────────────────────┐ │
│ │ │ SLA Manager │ │
│ │ │ • Retry exponential │ │
│ │ │ • Failover models │ │
│ │ │ • Dead letter queue │ │
│ │ └──────────┬──────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ TABLEAU DE BORD + ALERTES WECHAT │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
🔑 Configuration de l'API HolySheep
Commençons par la configuration de base. HolySheep offre une latence moyenne de <50ms et des économies de 85%+ par rapport aux tarifs officiels.
# config.py — Configuration centralisée de l'agent HolySheep
import os
============================================================================
HOLYSHEEP API CONFIGURATION
============================================================================
IMPORTANT : Base URL pour TOUTES les requêtes API HolySheep
Ne JAMAIS utiliser api.openai.com ou api.anthropic.com
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Clé API — à remplacer par ta vraie clé depuis https://www.holysheep.ai/register
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
============================================================================
MODÈLES UTILISÉS
============================================================================
MODEL_ANOMALY_DETECTION = "gpt-4.1" # GPT-4.1 : $8/MTok — haute précision
MODEL_WORK_ORDER_ROUTER = "deepseek-v3.2" # DeepSeek V3.2 : $0.42/MTok — économique
============================================================================
CONFIGURATION SLA (Service Level Agreement)
============================================================================
SLA_CONFIG = {
"critical_leak": {
"max_response_time_ms": 30000, # 30 secondes max
"retry_attempts": 5,
"retry_base_delay_ms": 1000,
"escalate_to": "emergency_team"
},
"minor_leak": {
"max_response_time_ms": 300000, # 5 minutes max
"retry_attempts": 3,
"retry_base_delay_ms": 5000,
"escalate_to": "maintenance_queue"
},
"false_positive_risk": {
"max_response_time_ms": 60000, # 1 minute max
"retry_attempts": 2,
"retry_base_delay_ms": 2000,
"escalate_to": "supervisor_review"
}
}
============================================================================
MODÈLES DE FAILOVER (en cas d'indisponibilité)
============================================================================
FAILOVER_MODELS = [
"gpt-4.1",
"claude-sonnet-4.5", # $15/MTok — plus cher mais haute disponibilité
"gemini-2.5-flash" # $2.50/MTok — équilibre coût/vitesse
]
============================================================================
SEUILS DE DÉTECTION
============================================================================
ANOMALY_THRESHOLD_CRITICAL = 0.85
ANOMALY_THRESHOLD_WARNING = 0.70
ANOMALY_THRESHOLD_INFO = 0.50
print("✅ Configuration chargée — HolySheep API v1")
print(f" 📊 Latence typique : <50ms")
print(f" 💰 Économie vs OpenAI : 85%+")
🕵️ Module 1 : Détection d'anomalies avec GPT-4.1
La détection de fuite repose sur l'analyse de plusieurs métriques en temps réel : pression différentielle, température du fluide caloporteur, débit volumique, et historique des reparations.
# anomaly_detector.py — Module de détection de fuite HolySheep
import requests
import json
from datetime import datetime
class AnomalyDetector:
"""
Utilise GPT-4.1 via HolySheep pour analyser les données de capteurs
et détecter les fuites potentielles dans le réseau de chauffage.
"""
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.model = "gpt-4.1"
def _build_detection_prompt(self, sensor_data: dict) -> str:
"""Construit le prompt pour GPT-4.1 avec les données du capteur."""
return f"""Tu es un expert en maintenance de réseaux de chauffage urbain.
Analyse les données suivantes d'un capteur de pipeline et détermine si une fuite est probable.
DONNÉES DU CAPTEUR :
- Capteur ID : {sensor_data.get('sensor_id', 'N/A')}
- Pression entrée : {sensor_data.get('pressure_inlet', 'N/A')} bar
- Pression sortie : {sensor_data.get('pressure_outlet', 'N/A')} bar
- Différentiel pression : {sensor_data.get('pressure_diff', 'N/A')} bar
- Température fluide : {sensor_data.get('temperature', 'N/A')} °C
- Débit volumique : {sensor_data.get('flow_rate', 'N/A')} m³/h
- Historique 24h : {sensor_data.get('history_summary', 'aucun problème')}
- Date/Heure : {sensor_data.get('timestamp', 'N/A')}
RÈGLES DE DÉTECTION :
1. Une chute de pression > 0.3 bar entre entrée et sortie + baisse du débit = fuite probable
2. Une baisse de température > 5°C en aval = fuite de fluide caloporteur
3. Corrélation : plusieurs capteurs voisins montrant des anomalies = fuite confirmée
Réponds STRICTEMENT en JSON avec ce format :
{{
"anomaly_score": 0.0 à 1.0,
"leak_probability": "haute|moyenne|basse|négligeable",
"recommended_action": "description courte de l'action",
"confidence_level": 0.0 à 1.0,
"reasoning": "explication courte du diagnostic"
}}
"""
def detect_anomaly(self, sensor_data: dict) -> dict:
"""
Envoie les données à GPT-4.1 via HolySheep pour analyse.
Args:
sensor_data: Dict contenant les métriques du capteur
Returns:
Dict avec anomaly_score, leak_probability, recommended_action
"""
prompt = self._build_detection_prompt(sensor_data)
# Appel API HolySheep
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Tu réponds uniquement en JSON valide."},
{"role": "user", "content": prompt}
],
"temperature": 0.1, # Faible température = réponses cohérentes
"max_tokens": 500,
"response_format": {"type": "json_object"}
},
timeout=30
)
if response.status_code != 200:
raise ConnectionError(f"Erreur HolySheep: {response.status_code} — {response.text}")
result = response.json()
content = result["choices"][0]["message"]["content"]
# Parser la réponse JSON de GPT-4.1
try:
return json.loads(content)
except json.JSONDecodeError:
return {
"anomaly_score": 0.0,
"leak_probability": "erreur_parse",
"error": "Réponse GPT-4.1 non valide"
}
============================================================================
EXEMPLE D'UTILISATION
============================================================================
if __name__ == "__main__":
detector = AnomalyDetector(api_key="YOUR_HOLYSHEEP_API_KEY")
# Données simulées d'un capteur suspect
test_data = {
"sensor_id": "CHN-SH-2024-047",
"pressure_inlet": 8.2,
"pressure_outlet": 7.3, # Chute anormale !
"pressure_diff": 0.9, # > 0.3 = suspect
"temperature": 72.1,
"flow_rate": 45.2,
"history_summary": "Dernière réparation : 3 mois",
"timestamp": "2026-05-25T10:52:00+08:00"
}
print("🔍 Analyse du capteur CHN-SH-2024-047...")
result = detector.detect_anomaly(test_data)
print(f"\n📊 RÉSULTAT GPT-4.1 :")
print(f" Score anomalie : {result.get('anomaly_score', 'N/A')}")
print(f" Probabilité fuite : {result.get('leak_probability', 'N/A')}")
print(f" Action recommandée : {result.get('recommended_action', 'N/A')}")
print(f" Confiance : {result.get('confidence_level', 'N/A')}")
🎫 Module 2 : Distribution de tickets avec DeepSeek V3.2
Une fois une anomalie détectée, le système doit router le ticket vers la bonne équipe. DeepSeek V3.2 ($0.42/MTok) est parfait pour cette tâche : économique et efficace pour le raisonnement structuré.
# work_order_router.py — Distribution intelligente de tickets
import requests
import json
from enum import Enum
from dataclasses import dataclass
from typing import Optional
class Team(Enum):
"""Équipes disponibles pour les interventions."""
EMERGENCY = "emergency_team"
MAINTENANCE_A = "maintenance_team_A"
MAINTENANCE_B = "maintenance_team_B"
INSPECTION = "inspection_team"
SUPERVISOR = "supervisor"
@dataclass
class WorkOrder:
"""Structure d'un bon de travail."""
order_id: str
sensor_id: str
anomaly_score: float
leak_probability: str
location: str
priority: str # critical, high, medium, low
assigned_team: Optional[Team] = None
estimated_repair_time_hours: float = 0.0
estimated_cost_cny: float = 0.0
class WorkOrderRouter:
"""
Utilise DeepSeek V3.2 via HolySheep pour router intelligemment
les tickets de fuite vers les équipes appropriées.
"""
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.model = "deepseek-v3.2"
# Base de connaissances des équipes
self.team_capabilities = {
Team.EMERGENCY: {
"specialties": ["fuite_critique", "rupture_tube", "evacuation"],
"max_response_time_min": 30,
"cost_per_hour_cny": 800
},
Team.MAINTENANCE_A: {
"specialties": ["fuite_mineure", "vanne_defectueuse", "isolation"],
"max_response_time_min": 180,
"cost_per_hour_cny": 350
},
Team.MAINTENANCE_B: {
"specialties": ["fuite_moyenne", " soudure", "remplacement_section"],
"max_response_time_min": 240,
"cost_per_hour_cny": 400
},
Team.INSPECTION: {
"specialties": ["diagnostic", "camera_inspection", "verification"],
"max_response_time_min": 480,
"cost_per_hour_cny": 250
}
}
def _build_routing_prompt(self, anomaly_result: dict, sensor_location: dict) -> str:
"""Construit le prompt pour DeepSeek V3.2."""
return f"""Tu es un dispatcheur intelligent pour un réseau de chauffage urbain.
Basé sur les données suivantes, détermine la meilleure équipe et priorité.
DONNÉES ANOMALIE :
- Score anomalie : {anomaly_result.get('anomaly_score', 0)}
- Probabilité fuite : {anomaly_result.get('leak_probability', 'inconnue')}
- Action recommandée : {anomaly_result.get('recommended_action', 'à déterminer')}
- Confiance du diagnostic : {anomaly_result.get('confidence_level', 0)}
LOCALISATION :
- Secteur : {sensor_location.get('district', 'N/A')}
- Rue : {sensor_location.get('street', 'N/A')}
- Distance à la station : {sensor_location.get('distance_to_station_km', 0)} km
- Zone résidentielle : {'Oui' if sensor_location.get('residential', False) else 'Non'}
ÉQUIPES DISPONIBLES :
{json.dumps(self.team_capabilities, indent=2, ensure_ascii=False)}
RÈGLES DE ROUTING :
1. Score > 0.85 OU probabilité "haute" = EMERGENCY (intervention < 30min)
2. Score 0.70-0.85 OU probabilité "moyenne" = MAINTENANCE (selon localisation)
3. Score 0.50-0.70 OU probabilité "basse" = INSPECTION d'abord
4. Zone résidentielle avec score > 0.70 = priorité augmentée d'un niveau
5. Distance > 5km = ajouter 1h au temps d'intervention estimé
Réponds en JSON :
{{
"assigned_team": "emergency_team|maintenance_team_A|maintenance_team_B|inspection_team",
"priority": "critical|high|medium|low",
"estimated_repair_time_hours": nombre,
"estimated_cost_cny": nombre,
"routing_reasoning": "explication courte"
}}
"""
def route_work_order(self, anomaly_result: dict, sensor_location: dict) -> WorkOrder:
"""
Demande à DeepSeek V3.2 de router le ticket.
Returns:
WorkOrder avec équipe assignée et estimations
"""
prompt = self._build_routing_prompt(anomaly_result, sensor_location)
# Appel API HolySheep avec DeepSeek V3.2
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [
{"role": "system", "content": "Tu réponds uniquement en JSON valide."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 400
},
timeout=25
)
if response.status_code != 200:
raise ConnectionError(f"Erreur DeepSeek V3.2: {response.status_code}")
result = response.json()
routing_decision = json.loads(result["choices"][0]["message"]["content"])
# Construire le WorkOrder
order = WorkOrder(
order_id=f"WO-{datetime.now().strftime('%Y%m%d%H%M%S')}-{sensor_location.get('district', 'XX')}",
sensor_id=anomaly_result.get('sensor_id', 'UNKNOWN'),
anomaly_score=anomaly_result.get('anomaly_score', 0),
leak_probability=anomaly_result.get('leak_probability', 'unknown'),
location=f"{sensor_location.get('district', '')}, {sensor_location.get('street', '')}",
priority=routing_decision.get('priority', 'medium'),
assigned_team=Team(routing_decision.get('assigned_team', 'maintenance_team_A')),
estimated_repair_time_hours=routing_decision.get('estimated_repair_time_hours', 2),
estimated_cost_cny=routing_decision.get('estimated_cost_cny', 500)
)
return order
============================================================================
EXEMPLE D'UTILISATION
============================================================================
if __name__ == "__main__":
router = WorkOrderRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
# Résultat du module de détection
anomaly_data = {
"sensor_id": "CHN-SH-2024-047",
"anomaly_score": 0.87,
"leak_probability": "haute",
"recommended_action": "Intervention immédiate requise",
"confidence_level": 0.92
}
# Localisation du capteur
location_data = {
"district": "Heping",
"street": "Nanjing Road",
"distance_to_station_km": 2.3,
"residential": True
}
print("🎫 Routage du ticket vers l'équipe appropriée...")
work_order = router.route_work_order(anomaly_data, location_data)
print(f"\n📋 BON DE TRAVAIL GÉNÉRÉ :")
print(f" ID : {work_order.order_id}")
print(f" Priorité : {work_order.priority.upper()}")
print(f" Équipe : {work_order.assigned_team.value}")
print(f" Temps estimé : {work_order.estimated_repair_time_hours}h")
print(f" Coût estimé : ¥{work_order.estimated_cost_cny:,.2f}")
🔄 Module 3 : SLA Manager avec retry et failover
La partie la plus critique d'un système de production : gérer les pannes d'API, les timeout, et maintenir les SLA. Voici mon implémentation complète que j'utilise en production.
# sla_manager.py — Gestionnaire de SLA avec retry exponentiel et failover
import time
import requests
import json
from typing import Callable, Any, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from enum import Enum
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("SLA_MANAGER")
class RetryStrategy(Enum):
"""Stratégies de retry disponibles."""
EXPONENTIAL_BACKOFF = "exponential"
LINEAR = "linear"
FIBONACCI = "fibonacci"
class ModelType(Enum):
"""Types de modèles disponibles pour le failover."""
GPT_41 = ("gpt-4.1", 8.0) # $8/MTok
DEEPSEEK_V32 = ("deepseek-v3.2", 0.42) # $0.42/MTok
CLAUDE_SONNET = ("claude-sonnet-4.5", 15.0) # $15/MTok
GEMINI_FLASH = ("gemini-2.5-flash", 2.50) # $2.50/MTok
@dataclass
class SLAViolation:
"""Enregistre une violation de SLA."""
timestamp: datetime
work_order_id: str
sla_type: str
expected_ms: int
actual_ms: int
action_taken: str
@dataclass
class SLAConfig:
"""Configuration SLA par type de ticket."""
max_response_time_ms: int
retry_attempts: int
retry_base_delay_ms: int
retry_multiplier: float = 2.0
max_retry_delay_ms: int = 30000
failover_models: list = field(default_factory=list)
class SLAManager:
"""
Gère les SLA avec retry intelligent et failover automatique.
Inclut le tracking des coûts et des violations.
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
# Configuration SLA par type de priorité
self.sla_configs = {
"critical": SLAConfig(
max_response_time_ms=30000,
retry_attempts=5,
retry_base_delay_ms=1000,
failover_models=["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash"]
),
"high": SLAConfig(
max_response_time_ms=60000,
retry_attempts=4,
retry_base_delay_ms=2000,
failover_models=["gpt-4.1", "deepseek-v3.2", "gemini-2.5-flash"]
),
"medium": SLAConfig(
max_response_time_ms=180000,
retry_attempts=3,
retry_base_delay_ms=3000,
failover_models=["deepseek-v3.2", "gpt-4.1"]
),
"low": SLAConfig(
max_response_time_ms=300000,
retry_attempts=2,
retry_base_delay_ms=5000,
failover_models=["deepseek-v3.2"]
)
}
# Tracking des métriques
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"sla_violations": [],
"total_cost_usd": 0.0,
"avg_latency_ms": 0.0
}
def _calculate_retry_delay(self, attempt: int, base_delay_ms: int,
multiplier: float, max_delay_ms: int) -> int:
"""Calcule le délai de retry avec backoff exponentiel."""
delay = int(base_delay_ms * (multiplier ** attempt))
return min(delay, max_delay_ms)
def _estimate_cost(self, model: ModelType, input_tokens: int,
output_tokens: int) -> float:
"""Estime le coût en USD pour une requête."""
# Prix moyen entrée/sortie
input_cost = (input_tokens / 1_000_000) * model.value[1]
output_cost = (output_tokens / 1_000_000) * model.value[1] * 1.5
return input_cost + output_cost
def execute_with_sla(
self,
work_order: Any,
api_call_fn: Callable,
priority: str = "medium"
) -> dict:
"""
Exécute un appel API avec gestion SLA complète.
Args:
work_order: Le WorkOrder à traiter
api_call_fn: Fonction lambda pour l'appel API
priority: Niveau de priorité (critical/high/medium/low)
Returns:
Dict avec le résultat ou les informations d'erreur
"""
sla_config = self.sla_configs.get(priority, self.sla_configs["medium"])
start_time = time.time()
last_error = None
self.metrics["total_requests"] += 1
# Parcourir les modèles de failover
models_to_try = sla_config.failover_models + ["deepseek-v3.2"]
models_tried = []
for model_name in models_to_try:
if model_name in models_tried:
continue
models_tried.append(model_name)
# Tentatives de retry pour ce modèle
for attempt in range(sla_config.retry_attempts):
try:
elapsed_ms = (time.time() - start_time) * 1000
# Vérifier si le SLA est déjà violé
if elapsed_ms > sla_config.max_response_time_ms:
raise TimeoutError(
f"SLA violated before request: {elapsed_ms:.0f}ms > "
f"{sla_config.max_response_time_ms}ms"
)
logger.info(
f"🔄 Attempt {attempt+1}/{sla_config.retry_attempts} "
f"with {model_name} for {work_order.order_id}"
)
# Exécuter l'appel API
result = api_call_fn(model_name)
# Succès !
total_time_ms = (time.time() - start_time) * 1000
self.metrics["successful_requests"] += 1
# Estimation coût (approx 500 tok input, 200 output)
cost = self._estimate_cost(ModelType.GPT_41, 500, 200)
self.metrics["total_cost_usd"] += cost
return {
"success": True,
"result": result,
"latency_ms": round(total_time_ms, 2),
"model_used": model_name,
"attempts": len(models_tried),
"within_sla": total_time_ms <= sla_config.max_response_time_ms
}
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError) as e:
last_error = str(e)
elapsed_ms = (time.time() - start_time) * 1000
logger.warning(
f"⚠️ Échec tentative {attempt+1}: {e} "
f"(elapsed: {elapsed_ms:.0f}ms)"
)
# Vérifier si on peut continuer
if elapsed_ms >= sla_config.max_response_time_ms:
break
# Calculer le délai avant retry
if attempt < sla_config.retry_attempts - 1:
retry_delay = self._calculate_retry_delay(
attempt,
sla_config.retry_base_delay_ms,
sla_config.retry_multiplier,
sla_config.max_retry_delay_ms
)
logger.info(f"⏳ Retry dans {retry_delay}ms...")
time.sleep(retry_delay / 1000)
# Passer au modèle suivant (failover)
if model_name != models_to_try[-1]:
logger.info(f"🔀 Failover vers le modèle suivant...")
# Tous les modèles ont échoué
self.metrics["failed_requests"] += 1
# Enregistrer la violation SLA
violation = SLAViolation(
timestamp=datetime.now(),
work_order_id=work_order.order_id,
sla_type=priority,
expected_ms=sla_config.max_response_time_ms,
actual_ms=int((time.time() - start_time) * 1000),
action_taken=f"Dead letter queue — {last_error}"
)
self.metrics["sla_violations"].append(violation)
return {
"success": False,
"error": last_error,
"work_order": work_order.order_id,
"models_tried": models_tried,
"total_attempts": sum(sla_config.retry_attempts for _ in models_tried)
}
def get_metrics_report(self) -> dict:
"""Génère un rapport des métriques SLA."""
total = self.metrics["total_requests"]
success_rate = (self.metrics["successful_requests"] / total * 100) if total > 0 else 0
return {
"total_requests": total,
"success_rate": f"{success_rate:.1f}%",
"failed_requests": self.metrics["failed_requests"],
"sla_violations_count": len(self.metrics["sla_violations"]),
"total_cost_usd": f"${self.metrics['total_cost_usd']:.4f}",
"avg_latency_ms": self.metrics.get("avg_latency_ms", 0)
}
============================================================================
EXEMPLE D'UTILISATION INTÉGRÉE
============================================================================
if __name__ == "__main__":
from work_order_router import WorkOrderRouter, Team, WorkOrder
from anomaly_detector import AnomalyDetector
# Initialiser les composants
api_key = "YOUR_HOLYSHEEP_API_KEY"
detector = AnomalyDetector(api_key)
router = WorkOrderRouter(api_key)
sla_manager = SLAManager(api_key)
# Créer un WorkOrder de test
work_order = WorkOrder(
order_id="WO-TEST-001",
sensor_id="CHN-SH-2024-047",
anomaly_score=0.87,
leak_probability="haute",
location="Heping District",
priority="critical"
)
# Simuler un appel API
def mock_api_call(model_name: str) -> dict:
"""Mock de l'appel API pour démonstration."""
import random
time.sleep(random.uniform(0.1, 0.5)) # Simule la latence
return {"status": "notification_sent", "wechat_id": "team_emergency_01"}
# Exécuter avec SLA
print("🚀 Exécution avec gestion SLA...")
result = sla_manager.execute_with_sla(
work_order=work_order,
api_call_fn=mock_api_call,
priority="critical"
)
print(f"\n📊 RÉSULTAT :")
print(f" Succès : {result.get('success')}")
print(f" Latence : {result.get('latency_ms')}ms")
print(f" Modèle utilisé : {result.get('model_used')}")
print(f" Dans le SLA : {result.get('within_sla')}")
# Rapport des métriques
print(f"\n📈 MÉTRIQUES SLA :")
report = sla_manager.get_metrics_report()
for key, value in report.items():
print(f" {key}: {value}")
🔧 Intégration complète du pipeline
# heating_network_agent.py — Pipeline complet de détection de fuite
"""
HolySheep 城市供热管网泄漏 Agent
Pipeline intégré : Détection → Routing → SLA → Notification
"""
import time
from datetime import datetime
import requests
Import des modules personnalisés
from anomaly_detector import AnomalyDetector
from work_order_router import WorkOrderRouter, WorkOrder
from sla_manager import SLAManager
class HeatingNetworkAgent:
"""
Agent complet pour la gestion des fuites de chauffage urbain.
Combine détection, routing, SLA et notification.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Initialiser les composants
self.detector = AnomalyDetector(api_key, self.base_url)
self.router = WorkOrderRouter(api_key, self.base_url)
self.sla_manager = SLAManager(api_key, self.base_url)
# Compteurs
self.processed_count = 0
self.critical_count = 0
self.start_time = time.time()
def process_sensor_batch(self, sensors_data: list) -> dict:
"""
Traite un lot de données de capteurs.
Args:
sensors_data: Liste de dictionnaires avec données de capteurs
Returns:
Rapport de traitement complet
"""
results = {
"batch_id": f"BATCH-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"processed": 0,
"anomalies_detected": 0,
"work_orders_created": 0,
"critical_alerts": 0,
"sla_compliant": 0,
"details": []
}
for sensor in sensors_data:
try:
# Étape 1 : Détection d'anomalie (GPT-4.1)
print(f"\n🔍 Analyse du capteur {sensor['sensor_id']}...")
anomaly_result = self.detector.detect_anomaly(sensor)
results["processed"] += 1
# Vérifier si anomalie significative
if anomaly_result.get('anomaly_score', 0) >= 0.70:
results["anomalies_detected"] += 1
# Localisation du capteur
location = {
"district": sensor.get('district', 'Unknown'),
"street": sensor.get('street', 'Unknown'),
"distance_to_station_km": sensor.get('distance_km', 1.0),
"residential": sensor.get('