import os import time import json from dotenv import load_dotenv from openai import OpenAI from modules.logger import log_event load_dotenv() api_key_openrouter = os.getenv("OPENROUTER_API_KEY") or "sk-or-v1-cle-non-definie" client = OpenAI(base_url="https://openrouter.ai/api/v1", api_key=api_key_openrouter) CATALOGUE_FILE = "/DATA/AppData/MULTI-IA-AETHAS38/catalogue_dynamique.json" def verifier_cle_api(): cle = os.getenv("OPENROUTER_API_KEY") if not cle or cle == "sk-or-v1-cle-non-definie": raise Exception("Clé OPENROUTER_API_KEY absente dans le fichier .env.") return cle def obtenir_modele_dynamique(premium: bool, prompt: str = ""): """Charge le catalogue dynamiquement pour un seul modèle optimal.""" est_education = "RÈGLE ÉDUCATION NATIONALE" in prompt try: if os.path.exists(CATALOGUE_FILE): with open(CATALOGUE_FILE, "r", encoding="utf-8") as f: data = json.load(f) if premium: paid_models = data.get("paid_models", []) if not paid_models: return "anthropic/claude-3.5-sonnet" claude = [m for m in paid_models if "claude-3.5-sonnet" in m.lower()] return claude[0] if claude else paid_models[0] else: free_models = [m for m in data.get("free_models", []) if not m.startswith("huggingface/")] if not free_models: return "google/gemma-2-9b-it:free" if est_education: mistral = [m for m in free_models if "mistral" in m.lower()] return mistral[0] if mistral else free_models[0] priorites = [m for m in free_models if "gemma" in m.lower() or "nemotron" in m.lower()] return priorites[0] if priorites else free_models[0] except Exception: pass return "anthropic/claude-3.5-sonnet" if premium else "google/gemma-2-9b-it:free" def obtenir_modeles_multi_dynamiques(premium: bool, prompt: str) -> list: """Détermine dynamiquement les modèles et leur nombre selon le sujet de la requête.""" try: with open(CATALOGUE_FILE, "r", encoding="utf-8") as f: data = json.load(f) except Exception: data = {"free_models": [], "paid_models": []} modeles_dispos = data.get("paid_models", []) if premium else data.get("free_models", []) modeles_or = [m for m in modeles_dispos if not m.startswith("huggingface/")] if not modeles_or: return ["anthropic/claude-3.5-sonnet"] if premium else ["google/gemma-2-9b-it:free"] # Analyse sémantique du sujet pour définir le nombre d'IA et la priorité prompt_lower = prompt.lower() nb_ia = 3 # Par défaut if any(mot in prompt_lower for mot in ["code", "script", "python", "javascript", "html", "css", "erreur", "bug", "kubejs", "serveur"]): nb_ia = 5 mots_cles = ["coder", "claude", "llama", "qwen", "deepseek"] elif any(mot in prompt_lower for mot in ["loi", "juridique", "droit", "contrat", "urssaf", "légal"]): nb_ia = 4 mots_cles = ["mistral", "claude", "llama", "mixtral"] elif any(mot in prompt_lower for mot in ["traduis", "rédaction", "écris", "lettre", "mail", "éducation"]): nb_ia = 2 mots_cles = ["mistral", "claude", "llama", "gemini", "gemma"] else: mots_cles = ["gemini", "llama", "mistral", "qwen", "gemma"] modeles_selectionnes = [] # Étape 1 : Sélectionner les modèles qui matchent les mots-clés for mc in mots_cles: for m in modeles_or: if mc in m.lower() and m not in modeles_selectionnes: modeles_selectionnes.append(m) if len(modeles_selectionnes) == nb_ia: break if len(modeles_selectionnes) == nb_ia: break # Étape 2 : Compléter avec les autres modèles du catalogue si le nombre n'est pas atteint for m in modeles_or: if len(modeles_selectionnes) >= nb_ia: break if m not in modeles_selectionnes: modeles_selectionnes.append(m) return modeles_selectionnes def interroger_openrouter(prompt: str, type_modele: str = "openrouter_chat", premium: bool = False, max_retries: int = 2, chat_id: str = None): verifier_cle_api() nom_modele_api = obtenir_modele_dynamique(premium, prompt) nom_affichage = f"OpenRouter ({nom_modele_api.split('/')[-1]})" couleur = "#d97757" if premium else "#4a90e2" extra_body_data = {"session_id": chat_id} if chat_id else {} for tentative in range(max_retries): try: response = client.chat.completions.create( model=nom_modele_api, messages=[{"role": "user", "content": prompt}], extra_headers={"HTTP-Referer": "https://aethas38.duckdns.org", "X-Title": "Aethas38 Multi-IA"}, extra_body=extra_body_data, timeout=90.0 ) texte = response.choices[0].message.content usage = response.usage tokens_in = usage.prompt_tokens if usage else 0 tokens_out = usage.completion_tokens if usage else 0 return texte, couleur, "OpenRouter", nom_affichage, tokens_in, tokens_out except Exception as e: log_event("WARNING", "OPENROUTER", f"Tentative {tentative + 1} échouée pour {nom_modele_api} : {e}") if tentative == max_retries - 1: try: from modules.model_updater import trigger_update_now trigger_update_now() except Exception: pass raise e time.sleep(1) return None, None, None, None, 0, 0 def interroger_openrouter_multi(prompt: str, type_modele: str = "openrouter_code", premium: bool = False, chat_id: str = None) -> list: verifier_cle_api() extra_body_data = {"session_id": chat_id} if chat_id else {} # La liste s'adapte dynamiquement au nombre d'IA requis par le sujet et aux disponibilités modeles = obtenir_modeles_multi_dynamiques(premium, prompt) log_event("INFO", "OPENROUTER_MULTI", f"Lancement Multi-IA avec {len(modeles)} modèles adaptatifs.") reponses = [] for modele in modeles: try: res = client.chat.completions.create( model=modele, messages=[{"role": "user", "content": prompt}], extra_headers={"HTTP-Referer": "https://aethas38.duckdns.org", "X-Title": "Aethas38 Multi-IA"}, extra_body=extra_body_data, timeout=90.0 ) contenu = res.choices[0].message.content if contenu: reponses.append({"modele": modele, "texte": contenu}) except Exception as e: log_event("WARNING", "OPENROUTER_MULTI", f"Échec pour {modele} : {e}") continue return reponses