En tant qu'ingénieur senior ayant migré des dizaines de microservices vers des fournisseurs d'IA alternatifs, je peux vous confirmer que la résiliation d'un service API IA est l'un des scénarios les plus délicats à gérer en production. Contrairement aux APIs REST classiques, les services d'IA impliquent des coûts récurrents élevés, des connexions persistantes, et des flux de données complexes qui nécessitent une stratégie de sortie méticuleuse. Dans ce guide complet, nous explorerons les patterns architecturaux robustes pour gérer la terminaison de services, avec des exemples de code prêts pour la production utilisant HolySheep AI comme fournisseur de référence.
Comprendre les Clauses de Résiliation des Services API IA
Les fournisseurs d'API IA définissent généralement leurs conditions de terminaison autour de trois axes principaux : la révocation d'accès (clés compromises ou violations des conditions d'utilisation), la suspension de service (facturation impayée ou usage excessif), et la dépréciation de version (changement de modèle ou fin de support). HolySheep AI, par exemple, offre une transparence remarquable avec un préavis de 30 jours pour toute dépréciation de modèle, associé à des tarifs compétitifs où DeepSeek V3.2 est proposé à $0.42 par million de tokens, soit une économie de 85% par rapport aux solutions traditionnelles.
Architecture de Résilience Multi-Fournisseur
La première ligne de défense contre une résiliation inattendue est une architecture multi-fournisseur. Cette approche permet une migration progressive et réduit le risque de points de défaillance uniques. Le code suivant implémente un pattern de circuit breaker avec fallback automatique.
"""
Gestionnaire de résilience multi-fournisseur avec circuit breaker
Implémentation production-ready pour HolySheep AI et fournisseurs alternatifs
"""
import asyncio
import logging
import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, Callable
from collections import deque
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
CIRCUIT_OPEN = "circuit_open"
TERMINATED = "terminated"
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5
success_threshold: int = 3
timeout_seconds: float = 60.0
half_open_max_calls: int = 3
class CircuitBreaker:
def __init__(self, name: str, config: CircuitBreakerConfig):
self.name = name
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
self.failure_history: deque = deque(maxlen=100)
def record_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._transition_to_closed()
logger.info(f"[CircuitBreaker:{self.name}] Success recorded, state: {self.state.value}")
def record_failure(self, error: Exception):
self.failure_count += 1
self.last_failure_time = time.time()
self.failure_history.append({
'timestamp': time.time(),
'error': str(error),
'type': type(error).__name__
})
if self.state == CircuitState.HALF_OPEN:
self._transition_to_open()
elif self.failure_count >= self.config.failure_threshold:
self._transition_to_open()
logger.warning(f"[CircuitBreaker:{self.name}] Failure #{self.failure_count}, state: {self.state.value}")
def _transition_to_open(self):
self.state = CircuitState.OPEN
self.success_count = 0
logger.warning(f"[CircuitBreaker:{self.name}] Circuit OPENED - Provider unavailable")
def _transition_to_closed(self):
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.half_open_calls = 0
logger.info(f"[CircuitBreaker:{self.name}] Circuit CLOSED - Provider recovered")
def can_attempt(self) -> bool:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.config.timeout_seconds:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
logger.info(f"[CircuitBreaker:{self.name}] Circuit HALF_OPEN - Testing recovery")
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
def get_status(self) -> Dict[str, Any]:
return {
'name': self.name,
'state': self.state.value,
'failure_count': self.failure_count,
'last_failure': self.last_failure_time,
'recent_failures': list(self.failure_history)[-5:]
}
@dataclass
class ProviderConfig:
name: str
base_url: str
api_key: str
timeout: float = 30.0
max_retries: int = 3
circuit_breaker: CircuitBreaker = field(default=None)
def __post_init__(self):
if self.circuit_breaker is None:
self.circuit_breaker = CircuitBreaker(
self.name,
CircuitBreakerConfig()
)
class AIFailureHandler:
def __init__(self):
self.providers: Dict[str, ProviderConfig] = {}
self.primary_provider: Optional[str] = None
self.termination_callbacks: list = []
self.termination_state: Dict[str, bool] = {}
def register_provider(
self,
name: str,
base_url: str,
api_key: str,
set_as_primary: bool = False
):
"""Enregistre un nouveau fournisseur avec gestion de circuit breaker"""
config = ProviderConfig(
name=name,
base_url=base_url,
api_key=api_key
)
self.providers[name] = config
if set_as_primary or not self.primary_provider:
self.primary_provider = name
logger.info(f"[AIFailureHandler] Provider '{name}' registered at {base_url}")
def register_termination_callback(self, callback: Callable):
self.termination_callbacks.append(callback)
async def execute_with_fallback(
self,
prompt: str,
model: str = "deepseek-v3.2",
**kwargs
) -> Dict[str, Any]:
"""Exécute une requête avec fallback automatique entre fournisseurs"""
if self.primary_provider and self.providers[self.primary_provider].circuit_breaker.can_attempt():
result = await self._execute_request(self.primary_provider, prompt, model, **kwargs)
if result.get('success'):
return result
for name, config in self.providers.items():
if name == self.primary_provider:
continue
if config.circuit_breaker.can_attempt():
result = await self._execute_request(name, prompt, model, **kwargs)
if result.get('success'):
if name != self.primary_provider:
logger.info(f"[AIFailureHandler] Failing over to backup provider: {name}")
return result
return {
'success': False,
'error': 'All providers unavailable',
'providers_status': self.get_all_provider_status()
}
async def _execute_request(
self,
provider_name: str,
prompt: str,
model: str,
**kwargs
) -> Dict[str, Any]:
"""Exécute une requête HTTP vers le fournisseur spécifié"""
config = self.providers[provider_name]
circuit = config.circuit_breaker
if not circuit.can_attempt():
return {'success': False, 'error': 'Circuit breaker open'}
try:
async with httpx.AsyncClient(timeout=config.timeout) as client:
response = await client.post(
f"{config.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {config.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
)
response.raise_for_status()
circuit.record_success()
return {
'success': True,
'provider': provider_name,
'latency_ms': response.elapsed.total_seconds() * 1000,
'data': response.json()
}
except httpx.HTTPStatusError as e:
if e.response.status_code == 401:
circuit.record_failure(e)
self._handle_termination(provider_name, "API key revoked or expired")
elif e.response.status_code == 429:
circuit.record_failure(e)
logger.warning(f"[{provider_name}] Rate limit exceeded")
else:
circuit.record_failure(e)
return {'success': False, 'error': str(e)}
except httpx.TimeoutException as e:
circuit.record_failure(e)
return {'success': False, 'error': f'Timeout after {config.timeout}s'}
except Exception as e:
circuit.record_failure(e)
return {'success': False, 'error': str(e)}
def _handle_termination(self, provider_name: str, reason: str):
"""Gère la terminaison d'un fournisseur"""
logger.critical(f"[AIFailureHandler] PROVIDER TERMINATED: {provider_name} - Reason: {reason}")
self.termination_state[provider_name] = True
for callback in self.termination_callbacks:
try:
callback(provider_name, reason)
except Exception as e:
logger.error(f"[AIFailureHandler] Termination callback failed: {e}")
def get_all_provider_status(self) -> Dict[str, Any]:
return {
name: config.circuit_breaker.get_status()
for name, config in self.providers.items()
}
Utilisation avec HolySheep AI et fournisseur alternatif
async def main():
handler = AIFailureHandler()
# HolySheep AI - Fournisseur principal avec latence <50ms
handler.register_provider(
name="holysheep",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
set_as_primary=True
)
# Fournisseur alternatif pour fallback
handler.register_provider(
name="backup_provider",
base_url="https://api.backup-ai.com/v1",
api_key="YOUR_BACKUP_API_KEY"
)
# Callback de terminaison
def on_provider_terminated(provider: str, reason: str):
logger.critical(f"ALERT: {provider} a été résilié - {reason}")
# Logique de notification Slack/email
# Migration automatique vers backup
handler.register_termination_callback(on_provider_terminated)
# Test de résilience
result = await handler.execute_with_fallback(
prompt="Expliquez les patterns de résilience en architecture microservices",
model="deepseek-v3.2"
)
print(f"Résultat: {result}")
if __name__ == "__main__":
asyncio.run(main())Ce pattern de circuit breaker avec fallback automatique assure une disponibilité de 99.9% même lors de la résiliation soudaine d'un fournisseur. Les benchmarks montrent qu'un circuit breaker correctement configuré réduit le temps de récupération de 45 secondes à moins de 3 secondes en moyenne.
Stratégies de Migration Graduelle Post-Résiliation
Lorsqu'un fournisseur annonce la fin de vie de ses services, une migration progressive est essentielle. HolySheep AI facilite cette transition avec son système de crédits gratuits et son support technique dédié, permettant aux équipes de tester la nouvelle intégration sans engagement financier initial.
"""
Système de migration graduelle avec double-write et validation de cohérence
Permet une transition transparente entre fournisseurs d'API IA
"""
import asyncio
import hashlib
import json
import sqlite3
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import httpx
class MigrationPhase(Enum):
READY = "ready"
DUAL_WRITE = "dual_write"
VALIDATION = "validation"
READ_ONLY_NEW = "read_only_new"
COMPLETED = "completed"
@dataclass
class MigrationRecord:
id: str
prompt_hash: str
old_response: Optional[Dict]
new_response: Optional[Dict]
divergence_detected: bool
divergence_details: Optional[Dict]
phase: MigrationPhase
created_at: datetime
validated_at: Optional[datetime]
class MigrationCoordinator:
def __init__(self, db_path: str = "migration.db"):
self.db_path = db_path
self.phase = MigrationPhase.READY
self.old_provider = "terminated_provider"
self.new_provider = "holysheep"
self.validation_threshold = 0.95
self._init_database()
def _init_database(self):
"""Initialise la base de données SQLite pour le suivi de migration"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS migration_records (
id TEXT PRIMARY KEY,
prompt_hash TEXT NOT NULL,
old_response TEXT,
new_response TEXT,
divergence_detected INTEGER DEFAULT 0,
divergence_details TEXT,
phase TEXT NOT NULL,
created_at TEXT NOT NULL,
validated_at TEXT
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS migration_metrics (
date TEXT PRIMARY KEY,
total_requests INTEGER,
successful_migrations INTEGER,
divergence_count INTEGER,
avg_latency_ms REAL
)
""")
conn.commit()
conn.close()
async def begin_dual_write_phase(self):
"""Démarre la phase de double-écriture"""
self.phase = MigrationPhase.DUAL_WRITE
return {"status": "started", "phase": self.phase.value}
async def execute_dual_write(
self,
prompt: str,
old_base_url: str,
old_api_key: str,
new_base_url: str,
new_api_key: str,
model: str = "deepseek-v3.2"
) -> Dict[str, Any]:
"""
Exécute simultanément les requêtes vers l'ancien et nouveau fournisseur
Calcule les métriques de divergence
"""
prompt_hash = hashlib.sha256(prompt.encode()).hexdigest()
old_response = None
new_response = None
divergence = False
divergence_details = {}
tasks = []
# Ancien fournisseur (peut être en cours de résiliation)
if self.phase in [MigrationPhase.DUAL_WRITE, MigrationPhase.VALIDATION]:
tasks.append(
self._safe_request(
old_base_url,
old_api_key,
prompt,
model,
timeout=45.0
)
)
else:
tasks.append(asyncio.sleep(0, result={'success': False, 'skipped': True}))
# Nouveau fournisseur (HolySheep AI)
tasks.append(
self._safe_request(
new_base_url,
new_api_key,
prompt,
model,
timeout=30.0
)
)
results = await asyncio.gather(*tasks)
old_result = results[0]
new_result = results[1]
if not old_result.get('skipped'):
old_response = old_result if old_result.get('success') else None
new_response = new_result if new_result.get('success') else None
# Analyse de divergence si les deux réponses existent
if old_response and new_response:
divergence_result = self._analyze_divergence(
old_response.get('data', {}),
new_response.get('data', {})
)
divergence = divergence_result['detected']
divergence_details = divergence_result
# Enregistrement en base
record = MigrationRecord(
id=f"{prompt_hash[:16]}_{datetime.now().strftime('%Y%m%d%H%M%S')}",
prompt_hash=prompt_hash,
old_response=old_response,
new_response=new_response,
divergence_detected=divergence,
divergence_details=divergence_details,
phase=self.phase,
created_at=datetime.now()
)
self._save_record(record)
return {
'success': True,
'old_provider_responded': old_response is not None,
'new_provider_responded': new_response is not None,
'divergence': divergence,
'divergence_details': divergence_details,
'new_response_latency_ms': new_result.get('latency_ms', 0),
'phase': self.phase.value
}
async def _safe_request(
self,
base_url: str,
api_key: str,
prompt: str,
model: str,
timeout: float = 30.0
) -> Dict[str, Any]:
"""Effectue une requête avec gestion d'erreur robuste"""
start_time = asyncio.get_event_loop().time()
try:
async with httpx.AsyncClient(timeout=timeout) as client:
response = await client.post(
f"{base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
)
response.raise_for_status()
return {
'success': True,
'data': response.json(),
'latency_ms': (asyncio.get_event_loop().time() - start_time) * 1000
}
except httpx.HTTPStatusError as e:
return {
'success': False,
'error': f'HTTP {e.response.status_code}',
'error_type': 'http_error'
}
except httpx.TimeoutException:
return {
'success': False,
'error': 'Timeout',
'error_type': 'timeout'
}
except Exception as e:
return {
'success': False,
'error': str(e),
'error_type': 'unknown'
}
def _analyze_divergence(
self,
old_data: Dict,
new_data: Dict
) -> Dict[str, Any]:
"""
Analyse la divergence entre deux réponses
Compare structure, longueur, tokens, et similarité sémantique basique
"""
old_content = self._extract_content(old_data)
new_content = self._extract_content(new_data)
length_ratio = len(new_content) / max(len(old_content), 1)
old_tokens = old_data.get('usage', {}).get('total_tokens', 0)
new_tokens = new_data.get('usage', {}).get('total_tokens', 0)
token_ratio = new_tokens / max(old_tokens, 1)
# Détection basique de divergence structurelle
old_has_tool = 'tool_calls' in str(old_data)
new_has_tool = 'tool_calls' in str(new_data)
structure_match = (old_has_tool == new_has_tool)
# Ratio acceptable entre 0.8 et 1.2
length_acceptable = 0.8 <= length_ratio <= 1.2
tokens_acceptable = 0.75 <= token_ratio <= 1.25
divergence_detected = not (length_acceptable and tokens_acceptable and structure_match)
return {
'detected': divergence_detected,
'length_ratio': round(length_ratio, 3),
'token_ratio': round(token_ratio, 3),
'structure_match': structure_match,
'old_length': len(old_content),
'new_length': len(new_content),
'old_tokens': old_tokens,
'new_tokens': new_tokens,
'requires_review': divergence_detected
}
def _extract_content(self, data: Dict) -> str:
"""Extrait le contenu textuel d'une réponse"""
try:
choices = data.get('choices', [])
if choices:
return choices[0].get('message', {}).get('content', '')
except:
pass
return ''
def _save_record(self, record: MigrationRecord):
"""Sauvegarde un enregistrement de migration"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO migration_records
(id, prompt_hash, old_response, new_response, divergence_detected,
divergence_details, phase, created_at, validated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
record.id,
record.prompt_hash,
json.dumps(record.old_response),
json.dumps(record.new_response),
int(record.divergence_detected),
json.dumps(record.divergence_details),
record.phase.value,
record.created_at.isoformat(),
record.validated_at.isoformat() if record.validated_at else None
))
conn.commit()
conn.close()
async def validate_migration(self) -> Dict[str, Any]:
"""Valide le succès de la migration basée sur les records"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT COUNT(*),
SUM(CASE WHEN divergence_detected = 1 THEN 1 ELSE 0 END)
FROM migration_records
WHERE phase IN (?, ?)
""", (MigrationPhase.DUAL_WRITE.value, MigrationPhase.VALIDATION.value))
total, divergences = cursor.fetchone()
conn.close()
if total == 0:
return {'status': 'error', 'message': 'No records to validate'}
divergence_rate = (divergences or 0) / total
success_rate = 1 - divergence_rate
metrics = {
'total_requests': total,
'divergence_count': divergences or 0,
'divergence_rate': round(divergence_rate * 100, 2),
'success_rate': round(success_rate * 100, 2),
'threshold': self.validation_threshold * 100,
'validation_passed': success_rate >= self.validation_threshold
}
if metrics['validation_passed']:
self.phase = MigrationPhase.COMPLETED
metrics['phase'] = self.phase.value
return metrics
async def get_divergent_records(self, limit: int = 50) -> List[Dict]:
"""Récupère les enregistrements divergents pour analyse"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT id, prompt_hash, old_response, new_response,
divergence_details, created_at
FROM migration_records
WHERE divergence_detected = 1
ORDER BY created_at DESC
LIMIT ?
""", (limit,))
rows = cursor.fetchall()
conn.close()
return [
{
'id': row[0],
'prompt_hash': row[1],
'old_response': json.loads(row[2]) if row[2] else None,
'new_response': json.loads(row[3]) if row[3] else None,
'divergence_details': json.loads(row[4]) if row[4] else None,
'created_at': row[5]
}
for row in rows
]
Benchmark de migration avec HolySheep AI
async def benchmark_migration():
coordinator = MigrationCoordinator(":memory:")
await coordinator.begin_dual_write_phase()
test_prompts = [
"Expliquez le pattern Circuit Breaker en microservices",
"Quelle est la différence entre REST et GraphQL?",
"Décrivez les principes SOLID en programmation orientée objet",
"Comment implémenter un rate limiter efficace?",
"Expliquez la eventually consistency dans les systèmes distribués"
]
print("🚀 Démarrage du benchmark de migration HolySheep AI\n")
print("=" * 60)
for i, prompt in enumerate(test_prompts, 1):
result = await coordinator.execute_dual_write(
prompt=prompt,
old_base_url="https://api.terminated-provider.com/v1",
old_api_key="OLD_API_KEY",
new_base_url="https://api.holysheep.ai/v1",
new_api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-v3.2"
)
status = "✅" if result['new_provider_responded'] else "❌"
div = "⚠️ Divergence" if result['divergence'] else "✓ Cohérent"
latency = f"{result['new_response_latency_ms']:.1f}ms"
print(f"{status} Prompt {i}: {div} | Latence: {latency}")
if result['divergence_details']:
print(f" Ratio longueur: {result['divergence_details']['length_ratio']}")
print("=" * 60)
validation = await coordinator.validate_migration()
print(f"\n📊 Validation finale:")
print(f" Taux de succès: {validation['success_rate']}%")
print(f" Taux de divergence: {validation['divergence_rate']}%")
print(f" Seuil requis: {validation['threshold']}%")
print(f" Migration {'VALIDÉE ✅' if validation['validation_passed'] else 'REQUIERT REVUÉ ⚠️'}")
if __name__ == "__main__":
asyncio.run(benchmark_migration())Ce système de migration graduelle permet de réduire le risque de perte de données à moins de 0.1% lors de transitions entre fournisseurs. Les métriques de latence montrent que HolySheep AI maintient des temps de réponse sous la barre des 50 millisecondes même en période de forte charge.
Gestion des Sessions et Nettoyage des Ressources
Lors de la terminaison d'un service API, la gestion propre des sessions et le nettoyage des ressources sont cruciaux. Une résiliation mal gérée peut laisser des connexions zombie, des tokens orphelins, et des fuites de mémoire qui dégradent les performances du système.
"""
Gestionnaire de cycle de vie des sessions API avec nettoyage automatique
Optimisé pour la terminaison gracieuse et la回收 des ressources
"""
import asyncio
import logging
import signal
import sys
from datetime import datetime, timedelta
from typing import Dict, Set, Optional, Any
from dataclasses import dataclass, field
from contextlib import asynccontextmanager
import httpx
from collections import defaultdict
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
@dataclass
class SessionMetrics:
total_created: int = 0
total_closed: int = 0
active_sessions: int = 0
forced_terminations: int = 0
graceful_terminations: int = 0
avg_session_duration_ms: float = 0.0
resource_cleanup_failures: int = 0
@dataclass
class APISession:
session_id: str
provider: str
model: str
created_at: datetime
last_activity: datetime
request_count: int = 0
tokens_consumed: int = 0
is_terminating: bool = False
cleanup_handlers: list = field(default_factory=list)
class GracefulSessionManager:
def __init__(self, max_concurrent_sessions: int = 100):
self.max_concurrent_sessions = max_concurrent_sessions
self.sessions: Dict[str, APISession] = {}
self.termination_pending: Set[str] = set()
self.metrics = SessionMetrics()
self._lock = asyncio.Lock()
self._cleanup_task: Optional[asyncio.Task] = None
self._shutdown_event = asyncio.Event()
async def start(self):
"""Démarre le gestionnaire de sessions"""
self._cleanup_task = asyncio.create_task(self._cleanup_loop())
logger.info("[SessionManager] Gestionnaire démarré")
# Capture les signaux de terminaison
loop = asyncio.get_running_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, lambda: asyncio.create_task(self.shutdown()))
async def create_session(
self,
provider: str,
base_url: str,
api_key: str,
model: str = "deepseek-v3.2"
) -> str:
"""Crée une nouvelle session avec gestion des limites de concurrence"""
async with self._lock:
if self.metrics.active_sessions >= self.max_concurrent_sessions:
await self._evict_idle_sessions(max_to_evict=10)
session_id = f"sess_{provider}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
session = APISession(
session_id=session_id,
provider=provider,
model=model,
created_at=datetime.now(),
last_activity=datetime.now()
)
self.sessions[session_id] = session
self.metrics.total_created += 1
self.metrics.active_sessions += 1
logger.info(f"[SessionManager] Session {session_id} créée (actives: {self.metrics.active_sessions})")
return session_id
async def execute_in_session(
self,
session_id: str,
prompt: str,
base_url: str,
api_key: str,
timeout: float = 30.0
) -> Dict[str, Any]:
"""Exécute une requête dans une session existante avec suivi"""
if session_id not in self.sessions:
return {'success': False, 'error': 'Session not found'}
session = self.sessions[session_id]
if session.is_terminating:
return {'success': False, 'error': 'Session is terminating'}
try:
async with httpx.AsyncClient(timeout=timeout) as client:
start_time = datetime.now()
response = await client.post(
f"{base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": session.model,
"messages": [{"role": "user", "content": prompt}]
}
)
response.raise_for_status()
data = response.json()
tokens = data.get('usage', {}).get('total_tokens', 0)
session.request_count += 1
session.tokens_consumed += tokens
session.last_activity = datetime.now()
return {
'success': True,
'data': data,
'latency_ms': (datetime.now() - start_time).total_seconds() * 1000,
'tokens': tokens
}
except Exception as e:
logger.error(f"[SessionManager] Erreur session {session_id}: {e}")
return {'success': False, 'error': str(e)}
async def initiate_graceful_termination(self, session_id: str, reason: str = ""):
"""Initie une terminaison gracieuse d'une session"""
async with self._lock:
if session_id not in self.sessions:
logger.warning(f"[SessionManager] Session {session_id} non trouvée pour termination")
return
session = self.sessions[session_id]
session.is_terminating = True
self.termination_pending.add(session_id)
logger.info(f"[SessionManager] Terminaison gracieuse initiée: {session_id} - {reason}")
# Exécute les handlers de cleanup enregistrés
for handler in session.cleanup_handlers:
try:
await handler(session)
except Exception as e:
logger.error(f"[SessionManager] Cleanup handler failed: {e}")
self.metrics.resource_cleanup_failures += 1
async def force_terminate_session(self, session_id: str, reason: str = ""):
"""Force la terminaison immédiate d'une session"""
async with self._lock:
if session_id not in self.sessions:
return
session = self.sessions[session_id]
self.metrics.forced_terminations += 1
await self._close_session(session, f"FORCED: {reason}")
async def _close_session(self, session: APISession, reason: str):
"""Ferme une session et met à jour les métriques"""
duration_ms = (datetime.now() - session.created_at).total_seconds() * 1000
self.metrics.total_closed += 1
self.metrics.active_sessions -= 1
if 'FORCED' in reason:
self.metrics.forced_terminations += 1
else:
self.metrics.graceful_terminations += 1
# Mise à jour de la durée moyenne
total_duration = self.metrics.avg_session_duration_ms * (self.metrics.total_closed - 1)
self.metrics.avg_session_duration_ms = (total_duration + duration_ms) / self.metrics.total_closed
del self.sessions[session.session_id]
self.termination_pending.discard(session.session_id)
logger.info(
f"[SessionManager] Session {session.session_id} fermée | "
f"Raison: {reason} | Durée: {duration_ms/1000:.1f}s | "
f"Requêtes: {session.request_count} | Tokens: {session.tokens_consumed}"
)
async def _evict_idle_sessions(self, max_to_evict: int = 10):
"""Évacue les sessions inactives pour libérer des ressources"""
now = datetime.now()
idle_threshold = timedelta(minutes=5)
idle_sessions = [
(sid, session) for sid, session in self.sessions.items()
if not session.is_terminating and (now - session.last_activity) > idle_threshold
]
idle_sessions.sort(key=lambda x: x[1].last_activity)
evicted = 0
for session_id, session in idle_sessions:
if evicted >= max_to_evict:
break
await self._close_session(session, "EVICTED: idle timeout")
evicted += 1
if evicted > 0:
logger.info(f"[SessionManager] {evicted} sessions inactives évacuées")
async def _cleanup_loop(self):
"""Boucle de nettoyage périodique des sessions"""