Spaces:
Running
Running
| 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) | |