Terminal / models /provider_router.py
Baida-A
Initial clean deploy (Reverse Proxy removed)
28a08e7
Raw
History Blame
2.55 kB
import logging
import asyncio
import os
from typing import List, Dict, Optional, Any, Tuple
from enum import Enum
from .ai_client import AIClient, ProviderConfig
from .role_router import Role
_logger = logging.getLogger("models.provider_router")
class LLMCapability(str, Enum):
FAST = "fast"
REASONING = "reasoning"
CODING = "coding"
VISION = "vision"
RESEARCH = "researcher"
ARCHITECT = "architect"
DEFAULT = "default"
class LLMProviderRouter:
"""
ARCH-I4.4: Provider Router
Astrazione dei provider LLM. Gestisce la selezione del provider
in base alla disponibilità e al tier.
"""
def __init__(self, ai_client: Optional[AIClient] = None):
self.client = ai_client or AIClient()
async def get_best_provider_for_tier(self, tier: int = 0) -> Optional[ProviderConfig]:
"""Ritorna il miglior provider disponibile per il tier richiesto."""
# Filtra i provider per tier e verifica salute (TODO: integrare HealthManager)
candidates = [p for p in self.client.providers if p.tier <= tier]
if not candidates:
return None
return candidates[0] # Per ora il primo è il migliore (ordinati per priorità in AIClient)
class LLMCapabilityRouter:
"""
ARCH-I4.4: Capability Router
Sceglie automaticamente il miglior modello/provider in base alla capacità richiesta.
"""
def __init__(self, provider_router: LLMProviderRouter):
self.provider_router = provider_router
def resolve_capability(self, capability: str) -> Role:
"""Mappa una stringa di capability a un Role noto di RoleRouter."""
mapping = {
"fast": Role.FAST,
"chat": Role.FAST,
"reasoning": Role.REASONER,
"coding": Role.CODER,
"vision": Role.RESEARCHER,
"research": Role.RESEARCHER,
"architect": Role.ARCHITECT,
"context": Role.CONTEXT,
"tester": Role.TESTER,
}
return mapping.get(capability.lower(), Role.DEFAULT)
async def get_client_for_capability(self, capability: str) -> Any:
"""Ritorna un'istanza di AIClient configurata per la capability specifica."""
from .role_router import RoleRouter
role = self.resolve_capability(capability)
_logger.info(f"Risoluzione capability LLM: '{capability}' -> Role: {role}")
return RoleRouter.get_client(role)
# Singleton instances
provider_router = LLMProviderRouter()
capability_router = LLMCapabilityRouter(provider_router)