""" ai_client.py — Real Provider Fleet (LLM API Ensemble) Legge la flotta di provider LLM da Supabase (tabella `ai_providers`, se disponibile) con fallback su variabili d'ambiente per gli 8 provider realmente configurati nel progetto (Groq, OpenRouter, Cerebras, SambaNova, Gemini, NVIDIA NIM, OpenAI, HF Router). STORIA: la versione precedente ("Dynamic SQL Fleet Edition") assumeva una flotta di 10 ZeroGPU Spaces (HF_SPACE_A1..E2) mai realmente esistita — nessuna di quelle env var è mai stata configurata né documentata altrove nel repo, quindi self.providers era sempre vuoto in produzione (ogni chiamata falliva silenziosamente con "tutti i nodi A-E sono offline"). Sostituito con i provider realmente attivi (vedi ENV_AUDIT_P21.md §6 e artifacts/agente-ai/src/config/env.ts). """ from __future__ import annotations import asyncio import os import time as _time_mod from dataclasses import dataclass from typing import AsyncIterator, Optional, List, Tuple from openai import OpenAI from api.semantic_cache import get_cached_response, set_cached_response import logging _logger = logging.getLogger("agente_ai") @dataclass(frozen=True) class ProviderConfig: id: int = 0 name: str = "" api_key: str = "" base_url: str = "" default_model: str = "" tier: int = 1 purpose: str = "reasoning" profile: str = "general" # Definizione statica dei provider LLM realmente attivi nel progetto. # base_url punta sempre all'endpoint OpenAI-compatible ufficiale del provider # (nessun proxy CF Worker qui: questo client gira lato backend Python, non browser). _PROVIDER_DEFS = [ # tier 0 — free tier veloce e affidabile {"name": "groq", "env_key": "GROQ_API_KEY", "base_url": "https://api.groq.com/openai/v1", "model_env": "GROQ_MODEL", "default_model": "llama-3.3-70b-versatile", "tier": 0, "purpose": "reasoning"}, {"name": "cerebras", "env_key": "CEREBRAS_API_KEY", "base_url": "https://api.cerebras.ai/v1", "model_env": "CEREBRAS_MODEL", "default_model": "llama-4-scout", "tier": 0, "purpose": "reasoning"}, {"name": "sambanova", "env_key": "SAMBANOVA_API_KEY", "base_url": "https://api.sambanova.ai/v1", "model_env": "SAMBANOVA_MODEL", "default_model": "DeepSeek-V3.2", "tier": 0, "purpose": "reasoning"}, # tier 1 — free tier con rate limit più stretti {"name": "openrouter", "env_key": "OPENROUTER_API_KEY", "base_url": "https://openrouter.ai/api/v1", "model_env": "OPENROUTER_MODEL","default_model": "meta-llama/llama-4-scout:free", "tier": 1, "purpose": "coding"}, {"name": "hf_router", "env_key": "HF_TOKEN", "base_url": "https://router.huggingface.co/v1", "model_env": "HF_MODEL", "default_model": "Qwen/Qwen2.5-Coder-32B-Instruct", "tier": 1, "purpose": "coding"}, {"name": "gemini", "env_key": "GEMINI_API_KEY", "base_url": "https://generativelanguage.googleapis.com/v1beta/openai/", "model_env": "GEMINI_MODEL", "default_model": "gemini-2.0-flash-exp", "tier": 1, "purpose": "memory"}, # tier 2 — fallback opzionale (spesso a pagamento o quota limitata) {"name": "nvidia", "env_key": "NVIDIA_API_KEY", "base_url": "https://integrate.api.nvidia.com/v1", "model_env": "NVIDIA_MODEL", "default_model": "nvidia/nemotron-3-ultra-550b-a55b", "tier": 2, "purpose": "audit"}, ] class AIClient: def __init__(self) -> None: self.providers = self._load_providers() self._client_cache: dict[str, OpenAI] = {} # S-DUAL-10: Indice per round-robin tra provider dello stesso purpose self._rr_indices: dict[str, int] = {} def _load_providers(self) -> list[ProviderConfig]: """Carica la flotta: prova Supabase (tabella `ai_providers`, source of truth dichiarata in supabase/migrations/20260711_ai_providers_fleet.sql), fallback sui provider reali via env se Supabase non è raggiungibile/vuoto (es. progetto sospeso per fatturazione, tabella non ancora popolata).""" providers = self._try_load_from_supabase() return providers if providers else self._discover_providers_from_env() def _try_load_from_supabase(self) -> list[ProviderConfig]: url = os.getenv("SUPABASE_URL", "") key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "") if not url or not key: return [] try: from supabase import create_client sb = create_client(url, key) res = ( sb.table("ai_providers") .select("id,name,api_key,base_url,default_model,tier,purpose") .eq("is_active", True) .order("tier", desc=False) .order("success_count", desc=True) .execute() ) rows = res.data or [] return [ ProviderConfig( id=row["id"], name=row["name"], api_key=row["api_key"], base_url=row["base_url"], default_model=row["default_model"], tier=row["tier"], purpose=row["purpose"], profile="general", ) for row in rows ] except Exception as e: _logger.warning(f"AIClient: Supabase ai_providers non disponibile, uso fallback env: {e}") return [] def _discover_providers_from_env(self) -> list[ProviderConfig]: """Discovery dei provider LLM realmente configurati via env var (vedi ENV_AUDIT_P21.md §6). Un provider viene incluso solo se la sua API key è impostata — nessun placeholder, nessun nodo fantasma.""" providers = [] for i, d in enumerate(_PROVIDER_DEFS): api_key = os.getenv(d["env_key"], "") if not api_key: continue providers.append(ProviderConfig( id=i, name=d["name"], api_key=api_key, base_url=d["base_url"], default_model=os.getenv(d["model_env"], d["default_model"]), tier=d["tier"], purpose=d["purpose"], profile="general", )) if not providers: _logger.error("AIClient: nessuna API key provider configurata (Groq/OpenRouter/Cerebras/SambaNova/Gemini/NVIDIA/HF_TOKEN tutte assenti)") return providers def _client_for(self, provider: ProviderConfig) -> OpenAI: if provider.name not in self._client_cache: self._client_cache[provider.name] = OpenAI( api_key=provider.api_key, base_url=provider.base_url, timeout=20, max_retries=0 ) return self._client_cache[provider.name] async def _fetch_one(self, provider: ProviderConfig, messages: list, temperature: float, max_tokens: int) -> Tuple[ProviderConfig, str, float]: client = self._client_for(provider) start = _time_mod.monotonic() try: response = await asyncio.wait_for( asyncio.to_thread( client.chat.completions.create, model=provider.default_model, messages=messages, temperature=temperature, max_tokens=max_tokens ), timeout=15 ) return provider, response.choices[0].message.content or "", _time_mod.monotonic() - start except Exception as e: _logger.warning(f"Provider {provider.name} fallito: {e}") return provider, f"ERROR: {str(e)}", 0.0 def _get_round_robin_provider(self, purpose: str) -> Optional[ProviderConfig]: """Ritorna uno dei provider con lo stesso purpose usando round-robin.""" purpose_providers = [p for p in self.providers if p.purpose == purpose] if not purpose_providers: return None idx = self._rr_indices.get(purpose, 0) provider = purpose_providers[idx % len(purpose_providers)] self._rr_indices[purpose] = idx + 1 return provider async def chat(self, messages: list, model: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 4096) -> str: if not self.providers: self.providers = self._load_providers() # S-CACHE-1: Lookup semantico preventivo cached = await get_cached_response(messages) if cached: _logger.info("[fleet] Semantic Cache HIT — risparmio quota provider") return cached # 1. Determina l'intento per il routing iniziale (specializzazione per purpose, # non per nodo hardware: i provider reali non sono dedicati per profilo). prompt = str(messages[-1].get("content", "")).lower() if any(k in prompt for k in ["code", "python", "fix", "script"]): primary_purpose = "coding" elif any(k in prompt for k in ["cerca", "ricorda", "database", "memory"]): primary_purpose = "memory" elif any(k in prompt for k in ["sicuro", "audit", "verifica", "check"]): primary_purpose = "audit" else: primary_purpose = "reasoning" # 2. Costruisce il pool di esecuzione (provider del purpose target + fallback globale) pool = [p for p in self.providers if p.purpose == primary_purpose] # Se il pool è vuoto, prendiamo i primi 4 provider disponibili (per tier) if not pool: pool = self.providers[:4] if not pool: return "🔴 Nessun provider LLM configurato (verifica le API key nei secrets del backend)." # 3. Esecuzione parallela (Ensemble Intelligence) tasks = [self._fetch_one(p, messages, temperature, max_tokens) for p in pool] results = await asyncio.gather(*tasks) best_r = self._judge_best_response(results, primary_purpose) # S-CACHE-1: Popolamento cache asincrono if not best_r.startswith("🔴"): asyncio.create_task(set_cached_response(messages, best_r)) return best_r def _judge_best_response(self, results: List[Tuple[ProviderConfig, str, float]], target_purpose: str) -> str: valid = [(p, r, t) for p, r, t in results if not r.startswith("ERROR:") and len(r) > 10] if not valid: return "🔴 Tutti i provider configurati hanno fallito o sono saturi. Riprovo con provider esterni..." def score(item): p, r, t = item s = len(r) # Bonus purpose target (Specializzazione) if p.purpose == target_purpose: s += 2000 # Bonus latenza (Performance mobile-first) if t < 2.0: s += 500 if "```" in r: s += 300 return s _, best_r, _ = max(valid, key=score) return best_r async def stream_chat(self, messages: list, model: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 4096) -> AsyncIterator[str]: """Streaming con failover immediato tra i provider configurati.""" if not self.providers: self.providers = self._load_providers() # Nessun provider configurato: feedback immediato all'utente invece di # cadere silenziosamente nel loop vuoto e dare un messaggio generico. if not self.providers: yield ( "⚠️ Nessun provider LLM configurato. " "Imposta almeno una delle seguenti variabili d'ambiente: " "GROQ_API_KEY, CEREBRAS_API_KEY, SAMBANOVA_API_KEY, " "OPENROUTER_API_KEY, HF_TOKEN, GEMINI_API_KEY." ) return # Nello streaming proviamo i provider in ordine di tier e performance for provider in self.providers: client = self._client_for(provider) try: stream = await asyncio.to_thread( client.chat.completions.create, model=provider.default_model, messages=messages, temperature=temperature, max_tokens=max_tokens, stream=True, ) iterator = iter(stream) while True: chunk = await asyncio.to_thread(next, iterator, None) if chunk is None: break if chunk.choices and chunk.choices[0].delta.content: yield chunk.choices[0].delta.content return except Exception as e: _logger.warning(f"Streaming fallito su {provider.name}: {e}") continue yield "🔴 Errore critico: tutti i provider configurati sono falliti o non disponibili."