#!/usr/bin/env python3 """ Barouia-Cortex 3.0 - Version corrigée Système Avancé d'Optimisation Cognitive """ from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect from fastapi.responses import HTMLResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn import logging import json import uuid import time import os import asyncio from typing import Dict, List, Optional, Tuple, Any import numpy as np from datetime import datetime, timedelta import hashlib import zlib import base64 from collections import deque, defaultdict import re # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("barouia_cortex_3") app = FastAPI( title="Barouia-Cortex 3.0", description="Système IA avancé avec optimisation", version="3.0.0" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"] ) # ============================================================================= # SYSTÈME DE COMPRESSION DE TOKENS # ============================================================================= class TokenCompressor: """Système avancé de compression lexicale""" def __init__(self): self.dictionnaire_compresse = {} self.dictionnaire_decompresse = {} self.mots_frequents = self.charger_mots_frequents() self.compression_level = 12 logger.info("🔤 Compresseur Token initialisé") def charger_mots_frequents(self) -> Dict[str, int]: """Charge les mots fréquents du français""" mots_base = { 'le', 'la', 'les', 'de', 'du', 'des', 'un', 'une', 'et', 'ou', 'que', 'qui', 'dans', 'en', 'sur', 'avec', 'pour', 'par', 'est' } return {mot: idx for idx, mot in enumerate(mots_base)} def compresser_texte(self, texte: str) -> Tuple[str, float]: """Compresse un texte avec optimisation token""" try: # Compression zlib avancée texte_bytes = texte.encode('utf-8') texte_compresse = zlib.compress(texte_bytes, level=self.compression_level) texte_final = base64.b85encode(texte_compresse).decode('ascii') # Calcul taux de compression taux_compression = (1 - len(texte_final) / len(texte)) * 100 return texte_final, max(0, taux_compression) except Exception as e: logger.error(f"❌ Erreur compression: {e}") return texte, 0.0 def decompresser_texte(self, texte_compresse: str) -> str: """Décompresse un texte optimisé""" try: texte_bytes = base64.b85decode(texte_compresse.encode('ascii')) texte_decompresse = zlib.decompress(texte_bytes) return texte_decompresse.decode('utf-8') except Exception as e: logger.error(f"❌ Erreur décompression: {e}") return texte_compresse # ============================================================================= # SYSTÈME DE MÉMOIRE OPTIMISÉE # ============================================================================= class MemoireCompressee: """Système de mémoire avec compression""" def __init__(self, capacite_max: int = 100000): self.capacite_max = capacite_max self.memoire_principale = {} self.cache_actif = deque(maxlen=1000) self.compresseur = TokenCompressor() logger.info("💾 Mémoire compressée initialisée") def stocker(self, cle: str, valeur: Any) -> bool: """Stocke une valeur avec compression""" try: if len(self.memoire_principale) >= self.capacite_max: self.liberer_espace() if isinstance(valeur, str): valeur_compressee, _ = self.compresseur.compresser_texte(valeur) else: valeur_compressee = valeur self.memoire_principale[cle] = { 'valeur': valeur_compressee, 'type': 'texte' if isinstance(valeur, str) else 'autre', 'timestamp': time.time(), 'acces': 0 } self.mettre_a_jour_cache(cle) return True except Exception as e: logger.error(f"❌ Erreur stockage: {e}") return False def recuperer(self, cle: str) -> Optional[Any]: """Récupère une valeur avec décompression""" try: if cle not in self.memoire_principale: return None donnees = self.memoire_principale[cle] donnees['acces'] += 1 valeur = donnees['valeur'] if donnees['type'] == 'texte': valeur = self.compresseur.decompresser_texte(valeur) self.mettre_a_jour_cache(cle) return valeur except Exception as e: logger.error(f"❌ Erreur récupération: {e}") return None def liberer_espace(self): """Libère de l'espace""" if not self.memoire_principale: return elements_tries = sorted( self.memoire_principale.items(), key=lambda x: (x[1]['acces'], x[1]['timestamp']) ) a_supprimer = max(1, len(elements_tries) // 10) for i in range(a_supprimer): cle = elements_tries[i][0] del self.memoire_principale[cle] def mettre_a_jour_cache(self, cle: str): """Met à jour le cache LRU""" if cle in self.cache_actif: self.cache_actif.remove(cle) self.cache_actif.append(cle) def get_statistiques(self) -> Dict[str, Any]: """Retourne les statistiques de la mémoire""" return { "utilisation": len(self.memoire_principale), "capacite_max": self.capacite_max, "elements_caches": len(self.cache_actif) } # ============================================================================= # MOTEUR D'ENTRAÎNEMENT ACCÉLÉRÉ # ============================================================================= class MoteurEntrainement: """Système d'entraînement neuronal accéléré""" def __init__(self): self.cycles_entrainement = 0 self.vitesse_entrainement = 1.0 self.performance_history = deque(maxlen=100) logger.info("⚡ Moteur d'entraînement initialisé") async def entrainer_modele(self, donnees: List[Any], epochs: int = 10) -> Dict[str, Any]: """Entraîne un modèle avec optimisation""" debut = time.time() try: for epoch in range(epochs): await asyncio.sleep(0.01) performance = self.calculer_performance(epoch, epochs) self.performance_history.append(performance) duree = time.time() - debut self.cycles_entrainement += 1 self.ajuster_vitesse(duree) return { "success": True, "cycles_completes": epochs, "duree_totale": duree, "vitesse_finale": self.vitesse_entrainement } except Exception as e: logger.error(f"❌ Erreur entraînement: {e}") return {"success": False, "error": str(e)} def calculer_performance(self, epoch: int, total_epochs: int) -> Dict[str, float]: """Calcule les métriques de performance""" progression = epoch / total_epochs accuracy = 0.5 + 0.5 * progression return { "epoch": epoch, "accuracy": accuracy, "loss": 1.0 - progression, "timestamp": time.time() } def ajuster_vitesse(self, duree_entrainement: float): """Ajuste automatiquement la vitesse d'entraînement""" if duree_entrainement < 1.0: self.vitesse_entrainement *= 1.1 elif duree_entrainement > 5.0: self.vitesse_entrainement *= 0.9 self.vitesse_entrainement = max(0.1, min(5.0, self.vitesse_entrainement)) def get_statistiques(self) -> Dict[str, Any]: """Retourne les statistiques d'entraînement""" return { "cycles_entrainement": self.cycles_entrainement, "vitesse_actuelle": self.vitesse_entrainement, "performance_history": list(self.performance_history) } # ============================================================================= # FENÊTRE DE CONTEXTE ÉTENDUE # ============================================================================= class FenetreContexteEtendue: """Système de fenêtre de contexte étendue""" def __init__(self, taille_max: int = 10000): self.taille_max = taille_max self.contenu = deque(maxlen=taille_max) self.compresseur = TokenCompressor() logger.info("🪟 Fenêtre de contexte étendue initialisée") def ajouter_contexte(self, texte: str, importance: float = 1.0): """Ajoute du contexte avec gestion d'importance""" texte_compresse, _ = self.compresseur.compresser_texte(texte) self.contenu.append(texte_compresse) def obtenir_contexte(self, limite_tokens: int = 2000) -> str: """Récupère le contexte le plus pertinent""" contexte_pertinent = [] for element in list(self.contenu)[-10:]: # 10 derniers éléments element_decompresse = self.compresseur.decompresser_texte(element) contexte_pertinent.append(element_decompresse) return " ".join(contexte_pertinent) def get_statistiques(self) -> Dict[str, Any]: """Retourne les statistiques de la fenêtre de contexte""" return { "taille_actuelle": len(self.contenu), "taille_max": self.taille_max } # ============================================================================= # CORTEX OPTIMISÉ 3.0 # ============================================================================= class CortexOptimise: """Cortex 3.0 avec toutes les optimisations intégrées""" def __init__(self): self.compresseur = TokenCompressor() self.memoire = MemoireCompressee() self.moteur_entrainement = MoteurEntrainement() self.fenetre_contexte = FenetreContexteEtendue() self.etat = { "performance_globale": { "temps_reponse": 0.05, "precision": 95.0, "efficacite_memoire": 85.0, "vitesse_traitement": 1.0 } } self.historique_performances = deque(maxlen=100) logger.info("🧠 Cortex Optimisé 3.0 initialisé") async def traiter_requete_optimisee(self, requete: str) -> Dict[str, Any]: """Traite une requête avec toutes les optimisations""" debut = time.time() try: requete_compressee, taux_compression = self.compresseur.compresser_texte(requete) self.fenetre_contexte.ajouter_contexte(requete) contexte = self.fenetre_contexte.obtenir_contexte() reponse = await self.generer_reponse_optimisee(requete, contexte) reponse_compressee, _ = self.compresseur.compresser_texte(reponse) duree = time.time() - debut self.mettre_a_jour_statistiques(duree, taux_compression) return { "success": True, "reponse_originale": reponse, "reponse_compressee": reponse_compressee, "taux_compression": taux_compression, "temps_traitement": duree } except Exception as e: logger.error(f"❌ Erreur traitement optimisé: {e}") return {"success": False, "error": str(e)} async def generer_reponse_optimisee(self, requete: str, contexte: str) -> str: """Génère une réponse optimisée""" await asyncio.sleep(0.02) reponses_avancees = [ f"🧠 [Système Optimisé] J'ai analysé votre requête avec compression token avancée.", f"⚡ [Traitement Accéléré] Votre demande a été processée en mode haute performance.", f"💾 [Mémoire Optimisée] Système de compression activé avec succès.", ] reponse_base = np.random.choice(reponses_avancees) if "état" in requete.lower() and "cortex" in requete.lower(): stats = self.get_statistiques_completes() return f"{reponse_base} État du système: Performance {stats['etat_cortex']['performance_globale']['precision']}%" elif "performance" in requete.lower(): perf = self.moteur_entrainement.get_statistiques() return f"{reponse_base} Vitesse d'entraînement: {perf['vitesse_actuelle']:.2f}x" elif "mémoire" in requete.lower(): mem = self.memoire.get_statistiques() return f"{reponse_base} Mémoire: {mem['utilisation']}/{mem['capacite_max']} éléments" else: return f"{reponse_base} J'ai bien compris: '{requete}'. Comment puis-je vous aider ?" def mettre_a_jour_statistiques(self, duree: float, taux_compression: float): """Met à jour les statistiques avancées""" self.etat["performance_globale"]["temps_reponse"] = ( self.etat["performance_globale"]["temps_reponse"] * 0.9 + duree * 0.1 ) self.historique_performances.append({ "timestamp": time.time(), "temps_reponse": duree, "taux_compression": taux_compression }) def get_statistiques_completes(self) -> Dict[str, Any]: """Retourne toutes les statistiques du système""" return { "etat_cortex": self.etat, "performance_entrainement": self.moteur_entrainement.get_statistiques(), "statistiques_memoire": self.memoire.get_statistiques(), "fenetre_contexte": self.fenetre_contexte.get_statistiques(), "timestamp": datetime.now().isoformat() } # ============================================================================= # SYSTÈME D'INTÉGRATION AVANCÉE DES APIS IA # ============================================================================= import aiohttp class IntegrationAPIs: """Système unifié de connexion aux APIs IA externes""" def __init__(self): self.apis_config = { "openai": { "url": "https://api.openai.com/v1/chat/completions", "api_key": os.getenv("OPENAI_API_KEY", ""), "modeles": ["gpt-4", "gpt-3.5-turbo"] }, "anthropic": { "url": "https://api.anthropic.com/v1/messages", "api_key": os.getenv("ANTHROPIC_API_KEY", ""), "modeles": ["claude-3-opus-20240229", "claude-3-sonnet-20240229"] }, "huggingface": { "url": "https://api-inference.huggingface.co/models", "api_key": os.getenv("HUGGINGFACE_API_KEY", ""), "modeles": ["mistralai/Mistral-7B-Instruct-v0.2", "google/flan-t5-xxl"] } } self.session = None self.modele_actif = "openai/gpt-4" self.historique_appels = deque(maxlen=100) logger.info("🌐 Système d'intégration APIs IA initialisé") async def get_session(self): """Crée une session HTTP partagée""" if self.session is None: self.session = aiohttp.ClientSession() return self.session async def communiquer_avec_gpt4(self, message: str, contexte: str = "") -> Dict[str, Any]: """Communication avancée avec GPT-4""" try: if not self.apis_config["openai"]["api_key"]: return self._reponse_simulation("GPT-4", message) session = await self.get_session() headers = { "Authorization": f"Bearer {self.apis_config['openai']['api_key']}", "Content-Type": "application/json" } payload = { "model": "gpt-4", "messages": [ { "role": "system", "content": f"Tu es Barouia-Cortex, un système IA avancé. Contexte: {contexte}" }, { "role": "user", "content": message } ], "temperature": 0.7, "max_tokens": 1000 } async with session.post( self.apis_config["openai"]["url"], headers=headers, json=payload, timeout=30 ) as response: if response.status == 200: result = await response.json() reponse = result["choices"][0]["message"]["content"] # Enregistrer l'appel self._enregistrer_appel("openai/gpt-4", message, reponse) return { "success": True, "reponse": reponse, "modele": "gpt-4", "tokens_utilises": result.get("usage", {}).get("total_tokens", 0) } else: return { "success": False, "error": f"Erreur API: {response.status}", "reponse_fallback": self._generer_reponse_fallback(message) } except Exception as e: logger.error(f"❌ Erreur GPT-4: {e}") return { "success": False, "error": str(e), "reponse_fallback": self._generer_reponse_fallback(message) } async def communiquer_avec_claude(self, message: str, contexte: str = "") -> Dict[str, Any]: """Communication avec Claude (Anthropic)""" try: if not self.apis_config["anthropic"]["api_key"]: return self._reponse_simulation("Claude", message) session = await self.get_session() headers = { "x-api-key": self.apis_config["anthropic"]["api_key"], "Content-Type": "application/json", "anthropic-version": "2023-06-01" } payload = { "model": "claude-3-sonnet-20240229", "max_tokens": 1000, "messages": [ { "role": "user", "content": f"Contexte: {contexte}\n\nMessage: {message}" } ] } async with session.post( self.apis_config["anthropic"]["url"], headers=headers, json=payload, timeout=30 ) as response: if response.status == 200: result = await response.json() reponse = result["content"][0]["text"] self._enregistrer_appel("anthropic/claude-3", message, reponse) return { "success": True, "reponse": reponse, "modele": "claude-3-sonnet", "tokens_utilises": result.get("usage", {}).get("input_tokens", 0) } else: return { "success": False, "error": f"Erreur API: {response.status}", "reponse_fallback": self._generer_reponse_fallback(message) } except Exception as e: logger.error(f"❌ Erreur Claude: {e}") return { "success": False, "error": str(e), "reponse_fallback": self._generer_reponse_fallback(message) } async def communiquer_avec_mistral(self, message: str) -> Dict[str, Any]: """Communication avec Mistral via Hugging Face""" try: if not self.apis_config["huggingface"]["api_key"]: return self._reponse_simulation("Mistral", message) session = await self.get_session() headers = { "Authorization": f"Bearer {self.apis_config['huggingface']['api_key']}", "Content-Type": "application/json" } payload = { "inputs": message, "parameters": { "max_new_tokens": 500, "temperature": 0.7, "top_p": 0.9 } } async with session.post( f"{self.apis_config['huggingface']['url']}/mistralai/Mistral-7B-Instruct-v0.2", headers=headers, json=payload, timeout=30 ) as response: if response.status == 200: result = await response.json() reponse = result[0]["generated_text"] if isinstance(result, list) else "Réponse reçue" self._enregistrer_appel("huggingface/mistral", message, reponse) return { "success": True, "reponse": reponse, "modele": "mistral-7b", "tokens_utilises": len(message.split()) } else: return { "success": False, "error": f"Erreur API: {response.status}", "reponse_fallback": self._generer_reponse_fallback(message) } except Exception as e: logger.error(f"❌ Erreur Mistral: {e}") return { "success": False, "error": str(e), "reponse_fallback": self._generer_reponse_fallback(message) } async def analyser_et_choisir_modele(self, message: str) -> str: """Analyse le message et choisit le modèle IA le plus approprié""" message_lower = message.lower() # Logique de sélection intelligente if any(mot in message_lower for mot in ['complexe', 'créatif', 'analyse', 'stratégie']): return "openai/gpt-4" elif any(mot in message_lower for mot in ['raisonnement', 'logique', 'philosophie', 'éthique']): return "anthropic/claude-3" elif any(mot in message_lower for mot in ['rapide', 'simple', 'direct']): return "huggingface/mistral" else: return self.modele_actif async def communiquer_intelligemment(self, message: str, contexte: str = "") -> Dict[str, Any]: """Communication intelligente avec sélection automatique du modèle""" modele_choisi = await self.analyser_et_choisir_modele(message) logger.info(f"🎯 Modèle choisi: {modele_choisi}") if modele_choisi == "openai/gpt-4": return await self.communiquer_avec_gpt4(message, contexte) elif modele_choisi == "anthropic/claude-3": return await self.communiquer_avec_claude(message, contexte) elif modele_choisi == "huggingface/mistral": return await self.communiquer_avec_mistral(message) else: return await self.communiquer_avec_gpt4(message, contexte) def _enregistrer_appel(self, modele: str, entree: str, sortie: str): """Enregistre les appels API pour l'analyse""" appel = { "timestamp": time.time(), "modele": modele, "entree": entree[:100] + "..." if len(entree) > 100 else entree, "sortie": sortie[:100] + "..." if len(sortie) > 100 else sortie, "taille_entree": len(entree), "taille_sortie": len(sortie) } self.historique_appels.append(appel) def _reponse_simulation(self, modele: str, message: str) -> Dict[str, Any]: """Simule une réponse quand l'API n'est pas configurée""" reponses_simulees = { "GPT-4": f"🧠 [Simulation GPT-4] J'analyserais: '{message}'. Configurez OPENAI_API_KEY pour une vraie réponse.", "Claude": f"🤖 [Simulation Claude] Réflexion sur: '{message}'. Configurez ANTHROPIC_API_KEY.", "Mistral": f"⚡ [Simulation Mistral] Traitement de: '{message}'. Configurez HUGGINGFACE_API_KEY." } return { "success": True, "reponse": reponses_simulees.get(modele, "Réponse simulée"), "modele": f"simulation-{modele.lower()}", "tokens_utilises": 0, "simulation": True } def _generer_reponse_fallback(self, message: str) -> str: """Génère une réponse de fallback intelligente""" return f"🔧 [Système Barouia] En mode autonome. J'ai compris: '{message}'. Mes capacités natives sont activées." def get_statistiques(self) -> Dict[str, Any]: """Retourne les statistiques d'utilisation des APIs""" total_appels = len(self.historique_appels) appels_par_modele = {} for appel in self.historique_appels: modele = appel["modele"] appels_par_modele[modele] = appels_par_modele.get(modele, 0) + 1 return { "total_appels": total_appels, "appels_par_modele": appels_par_modele, "modele_actif": self.modele_actif, "apis_configures": { "openai": bool(self.apis_config["openai"]["api_key"]), "anthropic": bool(self.apis_config["anthropic"]["api_key"]), "huggingface": bool(self.apis_config["huggingface"]["api_key"]) } } # ============================================================================= # SYSTÈME AVANCÉ DE RECOMMANDATIONS PERSONNALISÉES # ============================================================================= class SystemeRecommandations: """Système intelligent de recommandations basé sur le comportement utilisateur""" def __init__(self): self.historique_interactions = deque(maxlen=1000) self.profil_utilisateur = { "themes_frequents": defaultdict(int), "horaires_activite": defaultdict(int), "types_requetes": defaultdict(int), "fonctionnalites_preferees": defaultdict(int), "niveau_technique": "intermediaire" # debutant, intermediaire, expert } self.recommandations_actives = [] self.derniere_analyse = time.time() self.score_personnalisation = 0.0 logger.info("🎯 Système de recommandations personnalisées initialisé") def enregistrer_interaction(self, interaction: Dict[str, Any]): """Enregistre une interaction utilisateur pour l'analyse""" self.historique_interactions.append(interaction) # Mettre à jour le profil en temps réel self._mettre_a_jour_profil(interaction) # Générer de nouvelles recommandations si nécessaire if time.time() - self.derniere_analyse > 300: # Toutes les 5 minutes self.analyser_comportement() self.derniere_analyse = time.time() def analyser_comportement(self) -> Dict[str, Any]: """Analyse approfondie du comportement utilisateur""" logger.info("🔍 Analyse du comportement utilisateur...") if not self.historique_interactions: return {"statut": "donnees_insuffisantes"} # Analyse des thèmes fréquents themes = self._analyser_themes_frequents() # Analyse des patterns temporels patterns_temps = self._analyser_patterns_temporels() # Analyse du niveau technique niveau_tech = self._evaluer_niveau_technique() # Détection des besoins implicites besoins = self._detecter_besoins_implicites() # Générer les recommandations self._generer_recommandations(themes, patterns_temps, niveau_tech, besoins) return { "themes_principaux": dict(sorted(themes.items(), key=lambda x: x[1], reverse=True)[:5]), "patterns_temporels": patterns_temps, "niveau_technique": niveau_tech, "besoins_detectes": besoins, "score_personnalisation": self.score_personnalisation, "recommandations_actives": len(self.recommandations_actives) } def _analyser_themes_frequents(self) -> Dict[str, int]: """Analyse les thèmes les plus fréquents dans les interactions""" themes = defaultdict(int) for interaction in self.historique_interactions: message = interaction.get("message", "").lower() # Détection de thèmes if any(mot in message for mot in ['cortex', 'mémoire', 'optimisation', 'performance']): themes["optimisation_technique"] += 1 if any(mot in message for mot in ['bonjour', 'salut', 'hello', 'merci']): themes["communication"] += 1 if any(mot in message for mot in ['état', 'statut', 'santé', 'santé']): themes["monitoring_systeme"] += 1 if any(mot in message for mot in ['aide', 'comment', 'quoi', 'pourquoi']): themes["recherche_aide"] += 1 if any(mot in message for mot in ['ia', 'intelligence', 'apprentissage']): themes["ia_avancee"] += 1 if any(mot in message for mot in ['projet', 'developpement', 'code', 'programmation']): themes["developpement"] += 1 # Mettre à jour le profil for theme, count in themes.items(): self.profil_utilisateur["themes_frequents"][theme] += count return themes def _analyser_patterns_temporels(self) -> Dict[str, Any]: """Analyse les patterns d'utilisation temporels""" heures_activite = defaultdict(int) jours_activite = defaultdict(int) for interaction in self.historique_interactions: timestamp = interaction.get("timestamp", time.time()) heure = datetime.fromtimestamp(timestamp).hour jour = datetime.fromtimestamp(timestamp).strftime("%A") heures_activite[heure] += 1 jours_activite[jour] += 1 # Déterminer les pics d'activité heure_pic = max(heures_activite.items(), key=lambda x: x[1])[0] if heures_activite else 12 jour_pic = max(jours_activite.items(), key=lambda x: x[1])[0] if jours_activite else "Lundi" return { "heures_actives": dict(heures_activite), "jours_actifs": dict(jours_activite), "pic_activite": f"{heure_pic}h", "jour_favori": jour_pic } def _evaluer_niveau_technique(self) -> str: """Évalue le niveau technique de l'utilisateur""" termes_techniques = 0 termes_simples = 0 for interaction in self.historique_interactions: message = interaction.get("message", "").lower() # Termes techniques if any(mot in message for mot in ['api', 'endpoint', 'docker', 'uvicorn', 'fastapi', 'websocket']): termes_techniques += 1 # Termes simples if any(mot in message for mot in ['comment', 'quoi', 'aide', 'simple']): termes_simples += 1 total_interactions = len(self.historique_interactions) if total_interactions == 0: return "debutant" ratio_technique = termes_techniques / total_interactions if ratio_technique > 0.3: return "expert" elif ratio_technique > 0.1: return "intermediaire" else: return "debutant" def _detecter_besoins_implicites(self) -> List[str]: """Détecte les besoins non exprimés directement""" besoins = [] # Analyser les patterns pour détecter les besoins themes = self.profil_utilisateur["themes_frequents"] if themes.get("optimisation_technique", 0) > 5: besoins.append("optimisation_avancee") if themes.get("monitoring_systeme", 0) > 3: besoins.append("dashboard_ameliore") if themes.get("recherche_aide", 0) > 5: besoins.append("assistant_intelligent") if themes.get("ia_avancee", 0) > 2: besoins.append("integrations_ia") if themes.get("developpement", 0) > 3: besoins.append("outils_developement") return besoins def _generer_recommandations(self, themes: Dict, patterns: Dict, niveau: str, besoins: List[str]): """Génère des recommandations personnalisées""" self.recommandations_actives = [] # Recommandations basées sur les thèmes for theme, count in themes.items(): if count > 3: # Seuil d'activation if theme == "optimisation_technique": self.recommandations_actives.append({ "type": "fonctionnalite", "titre": "🔧 Mode Optimisation Avancée", "description": "Activez le mode haute performance pour l'optimisation mémoire", "priorite": "haute", "action": "/api/optimisation/activer", "score_confiance": min(1.0, count / 10) }) elif theme == "monitoring_systeme": self.recommandations_actives.append({ "type": "interface", "titre": "📊 Dashboard Détaillé", "description": "Accédez à des métriques système avancées", "priorite": "moyenne", "action": "/dashboard/avance", "score_confiance": min(1.0, count / 8) }) # Recommandations basées sur le niveau technique if niveau == "debutant": self.recommandations_actives.append({ "type": "guide", "titre": "🎓 Guide de Démarrage", "description": "Découvrez les fonctionnalités principales de Barouia-Cortex", "priorite": "haute", "action": "/guide/demarrage", "score_confiance": 0.9 }) elif niveau == "expert": self.recommandations_actives.append({ "type": "fonctionnalite", "titre": "⚙️ API Avancée", "description": "Accédez aux endpoints experts et à la documentation technique", "priorite": "moyenne", "action": "/api/docs", "score_confiance": 0.8 }) # Recommandations basées sur les besoins détectés for besoin in besoins: if besoin == "integrations_ia": self.recommandations_actives.append({ "type": "integration", "titre": "🤖 Intégrations IA", "description": "Connectez Barouia-Cortex à GPT-4, Claude et autres modèles", "priorite": "haute", "action": "/api/ia/configurer", "score_confiance": 0.85 }) elif besoin == "assistant_intelligent": self.recommandations_actives.append({ "type": "fonctionnalite", "titre": "💡 Assistant Contextuel", "description": "Activez l'assistant qui comprend le contexte de vos projets", "priorite": "moyenne", "action": "/api/assistant/activer", "score_confiance": 0.75 }) # Calcul du score de personnalisation self.score_personnalisation = min(1.0, len(self.historique_interactions) / 50) # Trier par priorité et score de confiance self.recommandations_actives.sort( key=lambda x: (1 if x["priorite"] == "haute" else 0.5, x["score_confiance"]), reverse=True ) def suggerer_fonctionnalites(self, limite: int = 5) -> List[Dict[str, Any]]: """Suggère des fonctionnalités personnalisées""" if not self.recommandations_actives: self.analyser_comportement() # Retourner les recommandations les plus pertinentes return self.recommandations_actives[:limite] def suggerer_themes(self, limite: int = 3) -> List[str]: """Suggère des thèmes de discussion basés sur les intérêts""" themes_tries = sorted( self.profil_utilisateur["themes_frequents"].items(), key=lambda x: x[1], reverse=True ) themes_suggerees = [] correspondance_themes = { "optimisation_technique": "Optimisation du système et performances", "communication": "Communication et interaction naturelle", "monitoring_systeme": "Surveillance et métriques avancées", "ia_avancee": "Intelligence artificielle et apprentissage", "developpement": "Développement et programmation" } for theme, count in themes_tries[:limite]: if count > 2: # Seuil minimum themes_suggerees.append(correspondance_themes.get(theme, theme)) return themes_suggerees def get_statistiques(self) -> Dict[str, Any]: """Retourne les statistiques du système de recommandations""" return { "total_interactions": len(self.historique_interactions), "score_personnalisation": self.score_personnalisation, "recommandations_actives": len(self.recommandations_actives), "profil_utilisateur": { "niveau_technique": self.profil_utilisateur["niveau_technique"], "themes_principaux": dict( sorted( self.profil_utilisateur["themes_frequents"].items(), key=lambda x: x[1], reverse=True )[:5] ) }, "derniere_analyse": self.derniere_analyse } # ============================================================================= # INITIALISATION DES SYSTÈMES # ============================================================================= # Initialisation du système optimisé cortex_optimise = CortexOptimise() integration_apis = IntegrationAPIs() systeme_recommandations = SystemeRecommandations() # ============================================================================= # INTÉGRATION AVEC LE SYSTÈME EXISTANT # ============================================================================= async def traiter_requete_avec_recommandations(self, requete: str) -> Dict[str, Any]: """Version améliorée qui enregistre les interactions pour les recommandations""" resultat = await self.traiter_requete_optimisee(requete) # Enregistrer l'interaction pour les recommandations interaction = { "timestamp": time.time(), "message": requete, "type": "chat", "longueur": len(requete), "themes_detectes": [] # Serait rempli par analyse NLP } systeme_recommandations.enregistrer_interaction(interaction) return resultat async def generer_reponse_optimisee_avancee(self, requete: str, contexte: str) -> str: """Nouvelle version utilisant les APIs IA externes""" try: # Utiliser l'API IA intelligente resultat = await integration_apis.communiquer_intelligemment(requete, contexte) if resultat["success"]: return f"🧠 [IA Externe - {resultat['modele']}] {resultat['reponse']}" else: # Fallback vers le système original return await self.generer_reponse_optimisee(requete, contexte) except Exception as e: logger.error(f"❌ Erreur API IA: {e}") return await self.generer_reponse_optimisee(requete, contexte) # Remplacez les méthodes dans CortexOptimise CortexOptimise.traiter_requete_optimisee = traiter_requete_avec_recommandations CortexOptimise.generer_reponse_optimisee = generer_reponse_optimisee_avancee # ============================================================================= # MODÈLES PYDANTIC # ============================================================================= class ChatRequest(BaseModel): message: str conversation_id: Optional[str] = None class EntrainementRequest(BaseModel): donnees: List[str] epochs: int = 5 # ============================================================================= # ROUTES API # ============================================================================= @app.get("/") async def root(): return { "message": "Barouia-Cortex 3.0 - Système Optimisé", "status": "active", "version": "3.0.0" } @app.get("/health") async def health_check(): return {"status": "ok", "timestamp": time.time()} @app.get("/api/statistiques/avancees") async def get_statistiques_avancees(): return cortex_optimise.get_statistiques_completes() @app.post("/api/chat/optimise") async def chat_optimise(request: ChatRequest): """Endpoint de chat avec optimisations""" try: resultat = await cortex_optimise.traiter_requete_optimisee(request.message) return { "success": True, "reponse": resultat.get("reponse_originale", "Erreur"), "statistiques": resultat, "cortex_state": cortex_optimise.get_statistiques_completes() } except Exception as e: return {"success": False, "error": str(e)} @app.post("/api/entrainement/demarrer") async def demarrer_entrainement(request: EntrainementRequest): """Démarre un cycle d'entraînement optimisé""" try: resultat = await cortex_optimise.moteur_entrainement.entrainer_modele( request.donnees, request.epochs ) return { "success": resultat["success"], "resultat_entrainement": resultat } except Exception as e: return {"success": False, "error": str(e)} @app.websocket("/ws/optimise") async def websocket_optimise(websocket: WebSocket): """WebSocket pour communications optimisées""" await websocket.accept() try: while True: data = await websocket.receive_text() message_data = json.loads(data) resultat = await cortex_optimise.traiter_requete_optimisee( message_data.get("message", "") ) await websocket.send_json({ "type": "reponse_optimisee", "reponse": resultat.get("reponse_originale", ""), "statistiques": resultat }) except WebSocketDisconnect: logger.info("WebSocket optimisé déconnecté") # ============================================================================= # ROUTES POUR L'INTÉGRATION IA # ============================================================================= @app.post("/api/ia/gpt4") async def api_gpt4(request: dict): """Endpoint dédié pour GPT-4""" message = request.get("message", "") contexte = request.get("contexte", "") resultat = await integration_apis.communiquer_avec_gpt4(message, contexte) return { "success": resultat["success"], "reponse": resultat.get("reponse", ""), "modele": resultat.get("modele", ""), "tokens_utilises": resultat.get("tokens_utilises", 0), "simulation": resultat.get("simulation", False) } @app.post("/api/ia/claude") async def api_claude(request: dict): """Endpoint dédié pour Claude""" message = request.get("message", "") contexte = request.get("contexte", "") resultat = await integration_apis.communiquer_avec_claude(message, contexte) return { "success": resultat["success"], "reponse": resultat.get("reponse", ""), "modele": resultat.get("modele", ""), "tokens_utilises": resultat.get("tokens_utilises", 0), "simulation": resultat.get("simulation", False) } @app.post("/api/ia/intelligente") async def api_ia_intelligente(request: dict): """Endpoint intelligent avec sélection automatique du modèle""" message = request.get("message", "") contexte = request.get("contexte", "") resultat = await integration_apis.communiquer_intelligemment(message, contexte) return { "success": resultat["success"], "reponse": resultat.get("reponse", ""), "modele": resultat.get("modele", ""), "tokens_utilises": resultat.get("tokens_utilises", 0), "simulation": resultat.get("simulation", False) } @app.get("/api/ia/statistiques") async def api_ia_statistiques(): """Statistiques d'utilisation des APIs IA""" return integration_apis.get_statistiques() @app.post("/api/ia/configurer") async def api_ia_configurer(request: dict): """Configure les clés API (pour interface admin)""" try: if request.get("openai_api_key"): integration_apis.apis_config["openai"]["api_key"] = request["openai_api_key"] if request.get("anthropic_api_key"): integration_apis.apis_config["anthropic"]["api_key"] = request["anthropic_api_key"] if request.get("huggingface_api_key"): integration_apis.apis_config["huggingface"]["api_key"] = request["huggingface_api_key"] return { "success": True, "message": "Configuration mise à jour", "apis_configures": integration_apis.get_statistiques()["apis_configures"] } except Exception as e: return {"success": False, "error": str(e)} # ============================================================================= # ROUTES POUR LES RECOMMANDATIONS # ============================================================================= @app.get("/api/recommandations/obtenir") async def obtenir_recommandations(limite: int = 5): """Endpoint pour obtenir les recommandations personnalisées""" recommandations = systeme_recommandations.suggerer_fonctionnalites(limite) return { "success": True, "recommandations": recommandations, "statistiques": systeme_recommandations.get_statistiques(), "themes_suggerees": systeme_recommandations.suggerer_themes(3) } @app.post("/api/recommandations/analyser") async def analyser_comportement(): """Déclenche une analyse manuelle du comportement""" analyse = systeme_recommandations.analyser_comportement() return { "success": True, "analyse": analyse, "message": "Analyse du comportement terminée" } @app.get("/api/recommandations/statistiques") async def get_recommandations_statistiques(): """Statistiques détaillées du système de recommandations""" return { "success": True, "statistiques": systeme_recommandations.get_statistiques() } @app.post("/api/recommandations/feedback") async def donner_feedback_recommandation(request: dict): """Donne un feedback sur une recommandation (utile pour l'apprentissage)""" recommandation_id = request.get("recommandation_id") note = request.get("note") # 1-5 utile = request.get("utile", False) # Ici on pourrait enregistrer le feedback pour améliorer le système logger.info(f"📝 Feedback reçu: {recommandation_id} - Note: {note} - Utile: {utile}") return { "success": True, "message": "Feedback enregistré pour amélioration du système" } # ============================================================================= # INTERFACES WEB # ============================================================================= @app.get("/dashboard", response_class=HTMLResponse) async def interface_revolutionnaire(): # Le contenu HTML complet reste le même que dans votre code original # Pour des raisons de longueur, je ne le répète pas ici html_content = """ Barouia-Cortex 3.0 • Intelligence Évolutive """ return HTMLResponse(content=html_content) @app.get("/recommandations", response_class=HTMLResponse) async def interface_recommandations(): """Interface dédiée aux recommandations personnalisées""" return """ Recommandations Personnalisées - Barouia-Cortex

🎯 Recommandations Personnalisées

Découvrez les fonctionnalités adaptées à votre usage

Chargement des recommandations...

""" # ============================================================================= # LANCEMENT DE L'APPLICATION # ============================================================================= if __name__ == "__main__": port = int(os.getenv("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)