GPT-5.5, Claude et DeepSeek : Le Guide Complet du Multi-Model Routing pour Réduire les Coûts de vos Agents IA de 85%

Introduction

Comprendre l'Architecture du Multi-Model Routing

Le Principe Fondamental

Architecture Composant par Composant

Implémentation du Router Intelligent

Le Classifieur de Tâches

import hashlib
import json
import time
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Dict, List, Any
from collections import OrderedDict
import asyncio

class TaskComplexity(Enum):
    TRIVIAL = 1      # Tâches simples, réponses courtes
    STANDARD = 2     # Requêtes ordinaires
    COMPLEX = 3      # Raisonnement multi-étapes
    EXPERT = 4       # Analyse approfondie, expertise spécialisée

class ModelType(Enum):
    DEEPSEEK = "deepseek"
    CLAUDE = "claude"
    GPT = "gpt"
    GEMINI = "gemini"

@dataclass
class RoutingDecision:
    model: ModelType
    endpoint: str
    estimated_cost: float  # en dollars par 1M tokens
    estimated_latency: int  # en millisecondes
    confidence: float
    reasoning: str

class LRUCache:
    """Cache LRU thread-safe pour les classifications répétées"""
    
    def __init__(self, capacity: int = 10000):
        self.capacity = capacity
        self.cache: OrderedDict = OrderedDict()
        self.lock = asyncio.Lock()
    
    async def get(self, key: str) -> Optional[TaskComplexity]:
        async with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)
                return self.cache[key]
            return None
    
    async def put(self, key: str, value: TaskComplexity):
        async with self.lock:
            if key in self.cache:
                self.cache.move_to_end(key)
            self.cache[key] = value
            if len(self.cache) > self.capacity:
                self.cache.popitem(last=False)

class MultiModelRouter:
    """
    Router intelligent multi-modèles avec classification automatique.
    VERSION PRODUCTION - 2026
    """
    
    # Tarification 2026 en $/million de tokens (entrée + sortie)
    MODEL_COSTS = {
        ModelType.GPT: {
            "model_id": "gpt-4.1",
            "input_cost": 2.00,
            "output_cost": 8.00,
            "base_url": "https://api.holysheep.ai/v1",
            "max_latency_ms": 8000,
            "strengths": ["format_json_strict", "code_generation", "mathematics"]
        },
        ModelType.CLAUDE: {
            "model_id": "claude-sonnet-4.5",
            "input_cost": 3.75,
            "output_cost": 15.00,
            "base_url": "https://api.holysheep.ai/v1",
            "max_latency_ms": 12000,
            "strengths": ["creative_writing", "nuance_analysis", "long_context"]
        },
        ModelType.DEEPSEEK: {
            "model_id": "deepseek-v3.2",
            "input_cost": 0.14,
            "output_cost": 0.42,
            "base_url": "https://api.holysheep.ai/v1",
            "max_latency_ms": 6000,
            "strengths": ["code_refactoring", "reasoning", "structured_output"]
        },
        ModelType.GEMINI: {
            "model_id": "gemini-2.5-flash",
            "input_cost": 0.35,
            "output_cost": 2.50,
            "base_url": "https://api.holysheep.ai/v1",
            "max_latency_ms": 4000,
            "strengths": ["speed", "multimodal", "batch_processing"]
        }
    }
    
    def __init__(self, api_key: str, budget_multiplier: float = 1.0):
        self.api_key = api_key
        self.budget_multiplier = budget_multiplier
        self.cache = LRUCache(capacity=10000)
        self.request_counts: Dict[ModelType, int] = {m: 0 for m in ModelType}
        self.circuit_breakers: Dict[ModelType, Dict[str, Any]] = {}
        
        # Initialisation des circuit breakers
        for model in ModelType:
            self.circuit_breakers[model] = {
                "failure_count": 0,
                "last_failure": 0,
                "is_open": False,
                "cooldown_until": 0
            }
    
    def _compute_cache_key(self, prompt: str, system_prompt: str = "") -> str:
        """Génère une clé de cache unique pour la requête"""
        content = f"{system_prompt}:{prompt}"
        return hashlib.sha256(content.encode()).hexdigest()[:32]
    
    def _classify_task_complexity(self, prompt: str, system_prompt: str = "") -> TaskComplexity:
        """
        Classification de la complexité de la tâche via analyse sémantique.
        Version simplifiée - en production, utilisez un modèle de classification dédié.
        """
        combined = f"{system_prompt} {prompt}".lower()
        word_count = len(combined.split())
        
        # Indicateurs de complexité
        expert_keywords = [
            "analyse approfondie", "stratégie", "architecturer", "optimiser",
            "expert", "spécialisé", "multi-étapes", "raffiné", "nuancé"
        ]
        complex_keywords = [
            "explique", "compare", "résous", "développe", "justifie",
            "calcule", "implémente", "crée", "génère", "écris"
        ]
        
        expert_count = sum(1 for kw in expert_keywords if kw in combined)
        complex_count = sum(1 for kw in complex_keywords if kw in combined)
        
        # Heuristiques de classification
        if expert_count >= 2 or (word_count > 500 and expert_count >= 1):
            return TaskComplexity.EXPERT
        elif complex_count >= 3 or word_count > 300:
            return TaskComplexity.COMPLEX
        elif word_count > 50 or complex_count >= 1:
            return TaskComplexity.STANDARD
        else:
            return TaskComplexity.TRIVIAL
    
    def _detect_specialized_task(self, prompt: str, system_prompt: str = "") -> Optional[str]:
        """Détecte les tâches spécialisées pour un routing optimal"""
        combined = f"{system_prompt} {prompt}".lower()
        
        # Détection du type de tâche
        if any(kw in combined for kw in ["json", "api", "rest", "graphql"]):
            return "code_generation"
        elif any(kw in combined for kw in ["traduis", "translate", "traduction"]):
            return "translation"
        elif any(kw in combined for kw in ["résume", "summary", "synthétis"]):
            return "summarization"
        elif any(kw in combined for kw in ["écris", "article", "blog", "content"]):
            return "creative_writing"
        elif any(kw in combined for kw in ["code", "fonction", "python", "javascript"]):
            return "code"
        elif any(kw in combined for kw in ["analyse", "évalue", "critique"]):
            return "analysis"
        
        return None
    
    async def route_request(
        self,
        prompt: str,
        system_prompt: str = "",
        max_cost: Optional[float] = None,
        max_latency: Optional[int] = None,
        force_model: Optional[ModelType] = None
    ) -> RoutingDecision:
        """
        Détermine le modèle optimal pour une requête donnée.
        
        Args:
            prompt: La requête utilisateur
            system_prompt: Instructions système optionnelles
            max_cost: Budget maximum en $/M tokens (input + output)
            max_latency: Latence maximale tolérée en ms
            force_model: Forcer un modèle spécifique (contourne le routing)
        
        Returns:
            RoutingDecision avec le modèle recommandé et métadonnées
        """
        start_time = time.time()
        
        # 1. Vérification du cache
        cache_key = self._compute_cache_key(prompt, system_prompt)
        cached_complexity = await self.cache.get(cache_key)
        
        if cached_complexity:
            complexity = cached_complexity
        else:
            # 2. Classification de la tâche
            complexity = self._classify_task_complexity(prompt, system_prompt)
            await self.cache.put(cache_key, complexity)
        
        # 3. Détection des tâches spécialisées
        task_type = self._detect_specialized_task(prompt, system_prompt)
        
        # 4. Si modèle forcé, retourne directement
        if force_model:
            model_config = self.MODEL_COSTS[force_model]
            total_cost = model_config["input_cost"] + model_config["output_cost"]
            return RoutingDecision(
                model=force_model,
                endpoint=f"{model_config['base_url']}/chat/completions",
                estimated_cost=total_cost,
                estimated_latency=model_config["max_latency_ms"],
                confidence=1.0,
                reasoning=f"Modèle forcé par l'appelant"
            )
        
        # 5. Évaluation des modèles disponibles
        candidates = []
        
        for model_type, config in self.MODEL_COSTS.items():
            # Vérification du circuit breaker
            cb = self.circuit_breakers[model_type]
            current_time = time.time()
            
            if cb["is_open"]:
                if current_time < cb["cooldown_until"]:
                    continue  # Circuit ouvert, skip ce modèle
                else:
                    # Fin du cooldown, reset
                    cb["is_open"] = False
                    cb["failure_count"] = 0
            
            # Calcul du score de matching
            score = 0.0
            reasons = []
            
            # Bonus selon la complexité
            if complexity == TaskComplexity.TRIVIAL:
                if model_type == ModelType.DEEPSEEK:
                    score += 3.0
                    reasons.append("DeepSeek optimal pour tâches simples (coût minimal)")
                elif model_type == ModelType.GEMINI:
                    score += 2.5
                    reasons.append("Gemini Flash excellent ratio vitesse/coût")
            
            elif complexity == TaskComplexity.COMPLEX:
                if model_type == ModelType.DEEPSEEK:
                    score += 2.5
                    reasons.append("DeepSeek V3.2 excellent en raisonnement structuré")
                elif model_type == ModelType.CLAUDE:
                    score += 3.0
                    reasons.append("Claude Sonnet 4.5 maîtrise le raisonnement nuancé")
            
            elif complexity == TaskComplexity.EXPERT:
                if model_type == ModelType.CLAUDE:
                    score += 4.0
                    reasons.append("Claude Sonnet 4.5 obligatoire pour tâches expertes")
                elif model_type == ModelType.GPT:
                    score += 3.0
                    reasons.append("GPT-4.1 excellent pour formats stricts")
            
            # Bonus pour tâches spécialisées
            if task_type:
                if task_type in config["strengths"]:
                    score += 2.5
                    reasons.append(f"匹配专业任务: {task_type}")
            
            # Malus pour coût élevé si budget constraint
            total_cost = config["input_cost"] + config["output_cost"]
            if max_cost and total_cost > max_cost * self.budget_multiplier:
                score -= 5.0
                reasons.append(f"Dépasse le budget: {total_cost}$ vs {max_cost}$ max")
            
            # Bonus pour latence si contrainte
            if max_latency and config["max_latency_ms"] > max_latency:
                score -= 3.0
                reasons.append(f"Latence trop élevée: {config['max_latency_ms']}ms")
            
            # Bonus pour modèles sous-utilisés (load balancing)
            usage_ratio = self.request_counts[model_type] / max(self.request_counts.values() or [1])
            if usage_ratio < 0.2:
                score += 1.5
                reasons.append("Load balancing: modèle sous-utilisé")
            
            candidates.append({
                "model": model_type,
                "config": config,
                "score": score,
                "reasons": reasons,
                "cost": total_cost
            })
        
        # 6. Sélection du meilleur candidat
        if not candidates:
            # Fallback sur DeepSeek si aucun candidat disponible
            return RoutingDecision(
                model=ModelType.DEEPSEEK,
                endpoint=self.MODEL_COSTS[ModelType.DEEPSEEK]["base_url"] + "/chat/completions",
                estimated_cost=self.MODEL_COSTS[ModelType.DEEPSEEK]["input_cost"] + self.MODEL_COSTS[ModelType.DEEPSEEK]["output_cost"],
                estimated_latency=4000,
                confidence=0.5,
                reasoning="Fallback vers DeepSeek (aucun candidat disponible)"
            )
        
        best_candidate = max(candidates, key=lambda x: x["score"])
        routing_time_ms = int((time.time() - start_time) * 1000)
        
        return RoutingDecision(
            model=best_candidate["model"],
            endpoint=f"{best_candidate['config']['base_url']}/chat/completions",
            estimated_cost=best_candidate["cost"],
            estimated_latency=best_candidate["config"]["max_latency_ms"],
            confidence=min(best_candidate["score"] / 5.0, 1.0),
            reasoning=" | ".join(best_candidate["reasons"])
        )
    
    def record_success(self, model: ModelType, latency_ms: int):
        """Enregistre un succès pour le suivi des métriques"""
        self.request_counts[model] += 1
        cb = self.circuit_breakers[model]
        cb["failure_count"] = 0
        cb["is_open"] = False
    
    def record_failure(self, model: ModelType):
        """Enregistre un échec et potentiellement ouvre le circuit breaker"""
        cb = self.circuit_breakers[model]
        cb["failure_count"] += 1
        cb["last_failure"] = time.time()
        
        # Ouvre le circuit après 5 échecs consécutifs
        if cb["failure_count"] >= 5:
            cb["is_open"] = True
            cb["cooldown_until"] = time.time() + 30  # 30 secondes de cooldown

import aiohttp
import asyncio
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass, field
import logging
from datetime import datetime, timedelta
import statistics

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

@dataclass
class RequestMetrics:
    """Métriques détaillées d'une requête"""
    model: str
    timestamp: datetime
    latency_ms: int
    input_tokens: int
    output_tokens: int
    total_cost: float
    success: bool
    error_message: Optional[str] = None

@dataclass
class PoolConfig:
    """Configuration d'un pool de connexions"""
    max_connections: int = 50
    max_connections_per_host: int = 10
    timeout_seconds: int = 60
    keepalive_timeout_seconds: int = 300

class ConnectionPool:
    """
    Gestionnaire de pool de connexions avec métriques temps réel.
    Supporte HTTP/2 multiplexing pour optimiser l'utilisation des connexions.
    """
    
    def __init__(
        self,
        router: MultiModelRouter,
        pool_config: Optional[PoolConfig] = None
    ):
        self.router = router
        self.config = pool_config or PoolConfig()
        self.session: Optional[aiohttp.ClientSession] = None
        self.metrics: List[RequestMetrics] = []
        self.metrics_lock = asyncio.Lock()
        
        # Pool par provider pour optimization
        self.pools: Dict[str, aiohttp.TCPConnector] = {}
        
        # Rate limiting par provider
        self.rate_limiters: Dict[str, asyncio.Semaphore] = {}
        self.rate_limit_config = {
            "deepseek": 100,   # req/min
            "claude": 80,
            "gpt": 60,
            "gemini": 120
        }
    
    async def initialize(self):
        """Initialise la session aiohttp avec configuration optimisée"""
        if self.session:
            return
        
        # Configuration du connector avec HTTP/2 si disponible
        connector = aiohttp.TCPConnector(
            limit=self.config.max_connections,
            limit_per_host=self.config.max_connections_per_host,
            keepalive_timeout=self.config.keepalive_timeout_seconds,
            enable_cleanup_closed=True,
            force_close=False
        )
        
        timeout = aiohttp.ClientTimeout(
            total=self.config.timeout_seconds,
            connect=10,
            sock_read=30
        )
        
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={
                "HTTP-Referer": "https://holysheep.ai",
                "X-Title": "MultiModelRouter"
            }
        )
        
        # Initialisation des rate limiters
        for provider, limit in self.rate_limit_config.items():
            self.rate_limiters[provider] = asyncio.Semaphore(limit)
        
        logger.info("Connection pool initialisé avec succès")
    
    async def close(self):
        """Ferme proprement tous les pools et sessions"""
        if self.session:
            await self.session.close()
            # Attente que les connexions se ferment proprement
            await asyncio.sleep(0.25)
        logger.info("Connection pool fermé")
    
    async def _execute_request(
        self,
        endpoint: str,
        headers: Dict[str, str],
        payload: Dict[str, Any],
        model_name: str
    ) -> Tuple[Dict[str, Any], int, bool, Optional[str]]:
        """
        Exécute une requête HTTP vers l'API.
        Retourne (response_data, latency_ms, success, error_message)
        """
        start_time = asyncio.get_event_loop().time()
        
        try:
            async with self.session.request(
                method="POST",
                url=endpoint,
                headers=headers,
                json=payload
            ) as response:
                latency_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
                
                if response.status == 200:
                    data = await response.json()
                    return data, latency_ms, True, None
                elif response.status == 429:
                    # Rate limited - retry avec backoff
                    retry_after = int(response.headers.get("Retry-After", 5))
                    await asyncio.sleep(retry_after)
                    return {}, latency_ms, False, f"Rate limited (retry after {retry_after}s)"
                else:
                    error_text = await response.text()
                    return {}, latency_ms, False, f"HTTP {response.status}: {error_text[:200]}"
        
        except asyncio.TimeoutError:
            latency_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
            return {}, latency_ms, False, "Timeout"
        except aiohttp.ClientError as e:
            latency_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
            return {}, latency_ms, False, f"Client error: {str(e)}"
        except Exception as e:
            latency_ms = int((asyncio.get_event_loop().time() - start_time) * 1000)
            return {}, latency_ms, False, f"Unexpected error: {str(e)}"
    
    async def execute_routed_request(
        self,
        prompt: str,
        system_prompt: str = "",
        temperature: float = 0.7,
        max_tokens: int = 2048,
        **kwargs
    ) -> Dict[str, Any]:
        """
        Exécute une requête avec routing automatique.
        Gère automatiquement le retry et le fallback.
        """
        await self.initialize()
        
        # Obtention de la décision de routing
        decision = await self.router.route_request(
            prompt=prompt,
            system_prompt=system_prompt,
            **kwargs
        )
        
        logger.info(f"Routing vers {decision.model.value}: {decision.reasoning[:100]}")
        
        # Construction du payload
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        
        model_config = MultiModelRouter.MODEL_COSTS[decision.model]
        payload = {
            "model": model_config["model_id"],
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        headers = {
            "Authorization": f"Bearer {self.router.api_key}",
            "Content-Type": "application/json"
        }
        
        # Exécution de la requête
        data, latency_ms, success, error = await self._execute_request(
            endpoint=decision.endpoint + "/chat/completions",
            headers=headers,
            payload=payload,
            model_name=model_config["model_id"]
        )
        
        # Enregistrement des métriques
        metric = RequestMetrics(
            model=model_config["model_id"],
            timestamp=datetime.now(),
            latency_ms=latency_ms,
            input_tokens=data.get("usage", {}).get("prompt_tokens", 0),
            output_tokens=data.get("usage", {}).get("completion_tokens", 0),
            total_cost=0.0,  # Calculé plus tard
            success=success,
            error_message=error
        )
        
        # Calcul du coût
        if success and "usage" in data:
            usage = data["usage"]
            metric.total_cost = (
                (usage.get("prompt_tokens", 0) / 1_000_000) * model_config["input_cost"] +
                (usage.get("completion_tokens", 0) / 1_000_000) * model_config["output_cost"]
            )
            self.router.record_success(decision.model, latency_ms)
        else:
            self.router.record_failure(decision.model)
            # Log pour monitoring
            logger.warning(f"Requête échouée: {error}")
        
        async with self.metrics_lock:
            self.metrics.append(metric)
            # Garde uniquement les 10000 dernières métriques
            if len(self.metrics) > 10000:
                self.metrics = self.metrics[-10000:]
        
        return {
            "success": success,
            "data": data,
            "latency_ms": latency_ms,
            "model_used": model_config["model_id"],
            "routing_confidence": decision.confidence,
            "cost_estimate": metric.total_cost,
            "error": error
        }
    
    async def execute_batch_parallel(
        self,
        requests: List[Dict[str, Any]],
        max_concurrent: int = 20
    ) -> List[Dict[str, Any]]:
        """
        Exécute plusieurs requêtes en parallèle avec contrôle de concurrence.
        Idéal pour les workloads de type agent où plusieurs sous-tâches
        peuvent être traitées simultanément.
        """
        semaphore = asyncio.Semaphore(max_concurrent)
        
        async def execute_single(req: Dict[str, Any]) -> Dict[str, Any]:
            async with semaphore:
                return await self.execute_routed_request(**req)
        
        # Exécution parallèle avec gestion des erreurs
        tasks = [execute_single(req) for req in requests]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                processed_results.append({
                    "success": False,
                    "error": str(result),
                    "original_request": requests[i]
                })
            else:
                processed_results.append(result)
        
        return processed_results
    
    async def get_statistics(self) -> Dict[str, Any]:
        """Retourne les statistiques agrégées du pool de connexions"""
        async with self.metrics_lock:
            if not self.metrics:
                return {}
            
            successful = [m for m in self.metrics if m.success]
            
            if not successful:
                return {"error": "Aucune requête réussie"}
            
            total_cost = sum(m.total_cost for m in successful)
            avg_latency = statistics.mean(m.latency_ms for m in successful)
            p99_latency = sorted([m.latency_ms for m in successful])[int(len(successful) * 0.99)]
            
            # Stats par modèle
            by_model = {}
            for metric in successful:
                if metric.model not in by_model:
                    by_model[metric.model] = {
                        "count": 0,
                        "total_cost": 0.0,
                        "avg_latency": 0,
                        "tokens_in": 0,
                        "tokens_out": 0
                    }
                by_model[metric.model]["count"] += 1
                by_model[metric.model]["total_cost"] += metric.total_cost
                by_model[metric.model]["tokens_in"] += metric.input_tokens
                by_model[metric.model]["tokens_out"] += metric.output_tokens
            
            for model_data in by_model.values():
                if model_data["count"] > 0:
                    model_data["avg_latency"] = sum(
                        m.latency_ms for m in successful if m.model == model_data
                    ) / model_data["count"] if model_data["count"] > 0 else 0
            
            return {
                "total_requests": len(self.metrics),
                "successful_requests": len(successful),
                "success_rate": len(successful) / len(self.metrics) * 100,
                "total_cost_usd": total_cost,
                "avg_latency_ms": avg_latency,
                "p99_latency_ms": p99_latency,
                "by_model": by_model,
                "period_start": self.metrics[0].timestamp.isoformat() if self.metrics else None,
                "period_end": self.metrics[-1].timestamp.isoformat() if self.metrics else None
            }

from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class ConversationTurn:
    """Représente un tour de conversation"""
    role: str  # 'user' ou 'assistant'
    content: str
    timestamp: datetime
    model_used: Optional[str] = None
    cost: float = 0.0
    latency_ms: int = 0

@dataclass
class AgentContext:
    """
    Contexte d'agent avec historique de conversation
    et métadonnées pour le routing intelligent.
    """
    conversation_id: str
    turns: List[ConversationTurn] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_turn(
        self,
        role: str,
        content: str,
        model_used: Optional[str] = None,
        cost: float = 0.0,
        latency_ms: int = 0
    ):
        """Ajoute un tour à l'historique"""
        self.turns.append(ConversationTurn(
            role=role,
            content=content,
            timestamp=datetime.now(),
            model_used=model_used,
            cost=cost,
            latency_ms=latency_ms
        ))
        
        # Garde uniquement les 50 derniers tours
        if len(self.turns) > 50:
            self.turns = self.turns[-50:]
    
    def get_context_window(self, max_turns: int = 10) -> List[Dict[str, str]]:
        """Retourne les derniers tours pour le contexte"""
        recent_turns = self.turns[-max_turns:]
        return [
            {"role": turn.role, "content": turn.content}
            for turn in recent_turns
        ]
    
    def get_total_cost(self) -> float:
        """Calcule le coût total de la conversation"""
        return sum(turn.cost for turn in self.turns)
    
    def estimate_total_tokens(self) -> int:
        """Estime le nombre total de tokens utilisés"""
        total = 0
        for turn in self.turns:
            # Approximation: 4 caractères ~= 1 token en moyenne
            total += len(turn.content) // 4
        return total

class ContextualRouter:
    """
    Router intelligent qui prend en compte le contexte de conversation.
    Optimise le routing selon l'historique et les patterns détectés.
    """
    
    def __init__(self, base_router: MultiModelRouter):
        self.router = base_router
        self.contexts: Dict[str, AgentContext] = {}
        
        # Patterns de comportement appris
        self.user_preferences: Dict[str, Dict[str, Any]] = {}
        
        # Modèle recommandé selon la longueur du contexte
        self.context_length_rules = {
            (0, 500): ModelType.DEEPSEEK,      # Contexte court -> modèle économique
            (500, 2000): ModelType.GPT,         # Contexte moyen -> bon équilibre
            (2000, 8000): ModelType.CLAUDE,     # Contexte long -> excellent contexte
            (8000, 128000): ModelType.CLAUDE    # Contexte très long -> Claude Sonnet 4.5
        }
    
    def get_or_create_context(self, conversation_id: str) -> AgentContext:
        """Récupère ou crée un contexte de conversation"""
        if conversation_id not in self.contexts:
            self.contexts[conversation_id] = AgentContext(
                conversation_id=conversation_id
            )
        return self.contexts[conversation_id]
    
    def detect_conversation_pattern(self, context: AgentContext) -> str:
        """
        Détecte le pattern de conversation pour optimiser le routing.
        Retourne un pattern reconnu ou 'standard'.
        """
        if len(context.turns) < 3:
            return "initial"
        
        recent_content = " ".join([
            turn.content.lower() 
            for turn in context.turns[-5:]
        ])
        
        # Détection de patterns
        if "code" in recent_content and "debug" in recent_content:
            return "code_debugging"
        elif all(kw in recent_content for kw in ["écris", "article", "blog"]):
            return "content_creation"
        elif "analyse" in recent_content and "données" in recent_content:
            return "data_analysis"
        elif len(context.turns) > 10 and context.get_total_cost() > 5.0:
            return "long_running"  # Conversation coûteuse, optimiser
        
        return "standard"
    
    async def route_with_context(
        self,
        conversation_id: str,
        prompt: str,
        system_prompt: str = "",
        force_model: Optional[ModelType] = None,
        **kwargs
    ) -> RoutingDecision:
        """
        Route une requête en tenant compte du contexte de conversation.
        """
        context = self.get_or_create_context(conversation_id)
        
        # Construction du prompt complet avec contexte
        context_window = context.get_context_window(max_turns=8)
        
        # Construction du system prompt enrichi
        pattern = self.detect_conversation_pattern(context)
        enhanced_system = system_prompt
        
        if pattern == "code_debugging":
            enhanced_system += "\n\n[CONTEXTE: Session de debugging detected. "
            enhanced_system += "Fournis des
Ressources connexes
📚 Tutoriels API IA
💰 Voir les tarifs
📖 Documentation
🚀 Inscription gratuite
Articles connexes
Comment utiliser DeepSeek V4 avec un contexte d'un million d
Guide complet : Configuration API Proxy GPT-5.5 et Claude Op
MCP Server en entreprise : Mon retour terrain sur la gateway