Update agent.py

agent.py

@@ -11,20 +11,21 @@ from smolagents import CodeAgent, MultiStepAgent, AgentError, PythonInterpreterT
 from models import ModelManager
 from tools import search_web, scrape_website, read_file
 
-
+# --- CLASSE ORCHESTRATEUR COMPLÈTE ---
 class OrchestratorAgent(MultiStepAgent):
     """
-    Agent orchestrateur qui hérite de MultiStepAgent et implémente 
+    Agent orchestrateur qui hérite de MultiStepAgent et implémente
     les méthodes requises pour le prompt système et la réponse finale.
     """
     def initialize_system_prompt(self) -> str:
         """Définit le prompt système pour l'agent orchestrateur."""
         return (
-            "You are an expert orchestrator agent. Your primary role is to understand a user's question, "
-            "devise a multi-step plan, and use the available tools to find the answer. "
-            "Think step-by-step. Analyze the results of each tool call. "
-            "When you have gathered all the necessary information, formulate a clear and comprehensive final answer. "
-            "Cite your sources if you used the search or scrape tools."
+            "You are a world-class autonomous agent. Your goal is to fully answer the user's question. "
+            "To do so, you have access to a set of tools. "
+            "First, think step-by-step and lay out a plan to solve the problem. "
+            "Then, execute the plan by calling the tools in the required sequence. "
+            "Analyze the results of each tool call. If the plan is not working, reassess and create a new plan. "
+            "When you have the final answer, present it clearly to the user."
         )
 
     def render_final_answer(self, final_context: dict, final_response: str) -> str:
@@ -34,7 +35,6 @@ class OrchestratorAgent(MultiStepAgent):
         # Pour l'instant, nous retournons simplement la réponse brute du modèle.
         return final_response
 
-
 class UltraAgent:
     """
     Agent ultra-puissant avec orchestration multi-modèles.
@@ -46,93 +46,48 @@ class UltraAgent:
         print("🚀 Initialisation de l'UltraAgent...")
         
         self.hf_token = hf_token or os.getenv("HF_TOKEN")
-        
-        # Gestionnaire de modèles
         self.model_manager = ModelManager(self.hf_token)
-        
-        # Outils disponibles
-        self.tools = [
-            search_web,
-            scrape_website,
-            read_file,
-            PythonInterpreterTool()
-        ]
-        
-        # Agents spécialisés
+        self.tools = [search_web, scrape_website, read_file, PythonInterpreterTool()]
         self._init_specialized_agents()
-        
-        # Historique des interactions
         self.conversation_history = []
-        
         print("✅ UltraAgent initialisé avec succès!")
 
     def _init_specialized_agents(self):
-        """Initialise les agents spécialisés comme l'orchestrateur et l'agent de code."""
+        """Initialise les agents spécialisés."""
         try:
-
-            # Utiliser la nouvelle classe OrchestratorAgent au lieu de MultiStepAgent
             self.orchestrator = OrchestratorAgent(
                 model=self.model_manager.get_orchestrator(),
                 tools=self.tools,
                 max_steps=15,
                 planning_interval=3
             )
-            
-            
-            # Agent de code
             self.code_agent = CodeAgent(
                 model=self.model_manager.get_code_agent(),
                 tools=[PythonInterpreterTool()],
                 additional_authorized_imports=["requests", "pandas", "numpy", "matplotlib", "seaborn", "scipy", "sklearn"]
             )
-            
             print("✅ Agents spécialisés initialisés")
-            
         except Exception as e:
             print(f"❌ Erreur lors de l'initialisation des agents: {e}")
             raise
 
     def __call__(self, question: str) -> str:
-        """
-        Point d'entrée principal de l'agent.
-        Analyse la question, détermine la stratégie et exécute la tâche.
-        """
+        """Point d'entrée principal de l'agent."""
         print("\n" + "="*80)
-        print("🧠 ULTRA-AGENT - Nouvelle question reçue")
-        print(f"Question: {question[:200]}{'...' if len(question) > 200 else ''}")
+        print(f"🧠 ULTRA-AGENT - Nouvelle question reçue\nQuestion: {question[:200]}{'...' if len(question) > 200 else ''}")
         print("="*80)
         
         try:
-            # Ajoute à l'historique
-            self.conversation_history.append({
-                "role": "user",
-                "content": question,
-                "timestamp": time.time()
-            })
-            
-            # Analyse de la question et choix de la stratégie
+            self.conversation_history.append({"role": "user", "content": question, "timestamp": time.time()})
             strategy = self._analyze_question(question)
             print(f"📋 Stratégie sélectionnée: {strategy['type']}")
-            
-            # Exécution selon la stratégie
             response = self._execute_strategy(question, strategy)
-            
-            # Ajoute la réponse à l'historique
-            self.conversation_history.append({
-                "role": "assistant", 
-                "content": response,
-                "strategy": strategy,
-                "timestamp": time.time()
-            })
-            
+            self.conversation_history.append({"role": "assistant", "content": response, "strategy": strategy, "timestamp": time.time()})
             print(f"✅ Réponse générée avec succès ({len(response)} caractères)")
             return response
-            
         except Exception as e:
             error_msg = f"❌ Erreur critique dans l'UltraAgent: {e}"
             print(error_msg)
-            
-            # Tentative de récupération avec le modèle de raisonnement
             try:
                 print("🔄 Tentative de récupération avec le modèle de raisonnement...")
                 fallback_response = self._fallback_reasoning(question, str(e))
@@ -143,211 +98,61 @@ class UltraAgent:
                 return final_error
 
     def _analyze_question(self, question: str) -> Dict[str, Any]:
-        """
-        Analyse la question pour déterminer la meilleure stratégie à adopter.
-        """
+        """Analyse la question pour déterminer la meilleure stratégie."""
+        # Cette fonction de scoring simple reste la même
         question_lower = question.lower()
-        
-        # Détection de mots-clés pour différents types de tâches
         vision_keywords = ['image', 'photo', 'picture', 'visual', 'voir', 'regarder', 'analyser l\'image', 'screenshot']
         code_keywords = ['code', 'program', 'script', 'algorithm', 'python', 'javascript', 'sql', 'debug', 'programmer']
-        search_keywords = ['search', 'find', 'look up', 'recherche', 'chercher', 'trouver', 'internet', 'web', 'actualité']
-        file_keywords = ['file', 'document', 'pdf', 'excel', 'csv', 'json', 'read', 'fichier', 'lire', 'ouvrir']
-        reasoning_keywords = ['complex', 'analyze', 'think', 'reason', 'solve', 'problem', 'complexe', 'analyser', 'résoudre', 'problème']
-        
-        # Calcul des scores
-        scores = {
-            "vision": sum(1 for kw in vision_keywords if kw in question_lower),
-            "code": sum(1 for kw in code_keywords if kw in question_lower),
-            "search_web": sum(1 for kw in search_keywords if kw in question_lower),
-            "file_processing": sum(1 for kw in file_keywords if kw in question_lower),
-            "reasoning": sum(1 for kw in reasoning_keywords if kw in question_lower),
-        }
-
-        # Détection d'URL pour forcer la recherche web
-        if any(domain in question_lower for domain in ['http', 'www', '.com', '.fr', '.org']):
-            scores["search_web"] += 2
-        
-        # Analyse de la complexité
-        complexity = len(question.split()) + len(re.findall(r'[.!?]', question))
-
-        # Détermination de la stratégie
-        if scores["vision"] > 0:
-            return {"type": "vision", "priority": scores["vision"], "complexity": complexity}
-        if scores["code"] > 1:
-            return {"type": "code", "priority": scores["code"], "complexity": complexity}
-        if scores["search_web"] > 0:
-            return {"type": "search_web", "priority": scores["search_web"], "complexity": complexity}
-        if scores["file_processing"] > 0:
-            return {"type": "file_processing", "priority": scores["file_processing"], "complexity": complexity}
-        if complexity > 50 or scores["reasoning"] > 0:
-            return {"type": "reasoning", "priority": scores["reasoning"], "complexity": complexity}
-        
-        return {"type": "general", "priority": 0, "complexity": complexity}
+        if any(kw in question_lower for kw in vision_keywords):
+            return {"type": "vision"}
+        if any(kw in question_lower for kw in code_keywords):
+            return {"type": "code"}
+        return {"type": "general"} # Simplification de la stratégie pour le débogage
 
     def _execute_strategy(self, question: str, strategy: Dict[str, Any]) -> str:
-        """Exécute la stratégie déterminée pour répondre à la question."""
+        """Exécute la stratégie déterminée."""
         strategy_type = strategy["type"]
-        
         try:
             if strategy_type == "vision":
                 return self._handle_vision_task(question)
             elif strategy_type == "code":
                 return self._handle_code_task(question)
-            elif strategy_type == "search_web":
-                return self._handle_web_search_task(question)
-            elif strategy_type == "file_processing":
-                return self._handle_file_task(question)
-            elif strategy_type == "reasoning":
-                return self._handle_reasoning_task(question)
-            else: # general
+            else: # general, search_web, file_processing, reasoning
                 return self._handle_general_task(question)
-                
         except Exception as e:
             print(f"❌ Erreur dans l'exécution de la stratégie {strategy_type}: {e}")
-            # Fallback vers l'orchestrateur général en cas d'échec
-            return self._handle_general_task(question)
+            raise e # Fait remonter l'erreur pour la gestion centrale
 
     def _handle_vision_task(self, question: str) -> str:
-        """Gère les tâches impliquant la vision."""
+        """Gère les tâches de vision."""
         print("👁️ Traitement avec le modèle de vision...")
-        
-        try:
-            vision_prompt = f"Analyse visuelle experte requise pour la question suivante : {question}. Si une image est mentionnée ou nécessaire, décris précisément ce que tu dois analyser."
-            response = self.model_manager.get_vision_model()(vision_prompt)
-            return f"🔍 Analyse visuelle:\n{response}"
-        except Exception as e:
-            print(f"❌ Erreur vision: {e}")
-            return self._handle_general_task(question)
+        response = self.model_manager.get_vision_model()(question)
+        return f"🔍 Analyse visuelle:\n{response}"
 
     def _handle_code_task(self, question: str) -> str:
-        """Gère les tâches de programmation avec l'agent de code."""
+        """Gère les tâches de code."""
         print("💻 Traitement avec l'agent de code...")
-        try:
-            response = self.code_agent.run(question)
-            return f"💻 Solution de code:\n{response}"
-        except Exception as e:
-            print(f"❌ Erreur code agent: {e}")
-            return self._handle_general_task(question)
-
-    def _handle_web_search_task(self, question: str) -> str:
-        """Gère les tâches nécessitant une recherche web."""
-        print("🌐 Traitement avec recherche web...")
-        try:
-            search_prompt = f"Répondez à cette question en utilisant une recherche web approfondie. Utilisez les outils de recherche et de scraping, et citez vos sources. Question : {question}"
-            response = self.orchestrator.run(search_prompt)
-            return response
-        except Exception as e:
-            print(f"❌ Erreur web search: {e}")
-            return self._handle_general_task(question)
-
-    def _handle_file_task(self, question: str) -> str:
-        """Gère les tâches impliquant la lecture de fichiers."""
-        print("📁 Traitement de fichiers...")
-        try:
-            file_prompt = f"Traitez les fichiers nécessaires pour répondre à cette question. Utilisez l'outil `read_file` pour analyser les fichiers mentionnés et extraire les informations pertinentes. Question : {question}"
-            response = self.orchestrator.run(file_prompt)
-            return response
-        except Exception as e:
-            print(f"❌ Erreur file processing: {e}")
-            return self._handle_general_task(question)
-
-    def _handle_reasoning_task(self, question: str) -> str:
-        """Gère les tâches complexes nécessitant un raisonnement approfondi."""
-        print("🧠 Traitement avec le modèle de raisonnement...")
-        try:
-            reasoning_prompt = f"En tant qu'expert en résolution de problèmes complexes, analysez et répondez de manière structurée à la question suivante : {question}. Décomposez le problème, analysez chaque partie et proposez une solution étape par étape."
-            response = self.model_manager.get_reasoning_model()(reasoning_prompt)
-            return f"🧠 Analyse approfondie:\n{response}"
-        except Exception as e:
-            print(f"❌ Erreur reasoning: {e}")
-            return self._handle_general_task(question)
+        return self.code_agent.run(question)
 
     def _handle_general_task(self, question: str) -> str:
-        """Gère les tâches générales avec l'orchestrateur principal."""
+        """Gère toutes les autres tâches avec l'orchestrateur."""
         print("🎯 Traitement général avec l'orchestrateur...")
-        try:
-            context = self._get_conversation_context()
-            enhanced_prompt = f"{context}\nQuestion actuelle: {question}\nInstructions: Analysez et répondez de manière complète, en utilisant les outils si nécessaire."
-            response = self.orchestrator.run(enhanced_prompt)
-            return response
-        except Exception as e:
-            print(f"❌ Erreur general task: {e}")
-            try:
-                # Dernière tentative simple
-                simple_response = self.model_manager.get_orchestrator()(question)
-                return simple_response
-            except Exception as final_e:
-                raise final_e # Propage l'erreur au handler principal
+        context = self._get_conversation_context()
+        enhanced_prompt = f"{context}\nQuestion actuelle: {question}"
+        return self.orchestrator.run(enhanced_prompt)
 
     def _fallback_reasoning(self, question: str, error: str) -> str:
-        """Utilise le modèle de raisonnement en cas d'erreur pour fournir une réponse de secours."""
-        try:
-            fallback_prompt = f"Une erreur s'est produite en traitant la question. Question: '{question}'. Erreur: '{error}'. Fournissez la meilleure réponse possible en expliquant les limitations dues à l'erreur."
-            response = self.model_manager.get_reasoning_model()(fallback_prompt)
-            return f"⚠️ Réponse de récupération:\n{response}"
-        except Exception as e:
-            raise e
+        """Fallback avec le modèle de raisonnement."""
+        fallback_prompt = f"An error occurred while processing the question. Question: '{question}'. Error: '{error}'. Provide the best possible answer, explaining the limitations due to the error."
+        return self.model_manager.get_reasoning_model()(fallback_prompt)
 
     def _get_conversation_context(self, max_exchanges: int = 3) -> str:
-        """Récupère le contexte des échanges récents pour l'injecter dans le prompt."""
-        if not self.conversation_history:
-            return ""
-        
+        """Récupère le contexte des échanges récents."""
+        if not self.conversation_history: return ""
         recent_history = self.conversation_history[-max_exchanges*2:]
         context_parts = ["Contexte de la conversation récente:"]
         for entry in recent_history:
             role = "Utilisateur" if entry["role"] == "user" else "Assistant"
-            content = str(entry["content"])[:200] + "..." if len(str(entry["content"])) > 200 else str(entry["content"])
+            content = str(entry["content"])[:200]
             context_parts.append(f"{role}: {content}")
-        
-        return "\n".join(context_parts)
-
-    def get_status(self) -> Dict[str, Any]:
-        """Retourne le statut actuel de l'agent."""
-        return {
-            "status": "active",
-            "conversation_length": len(self.conversation_history),
-            "available_tools": [tool.__name__ for tool in self.tools],
-            "models_status": self.model_manager.test_models(),
-        }
-
-    def reset_conversation(self):
-        """Réinitialise l'historique de la conversation."""
-        self.conversation_history = []
-        print("🔄 Historique de conversation remis à zéro.")
-
-
-class BasicAgent:
-    """
-    Classe de compatibilité pour remplacer l'ancien BasicAgent dans app.py.
-    Elle utilise l'UltraAgent en arrière-plan pour toute la logique.
-    """
-    
-    def __init__(self):
-        """Initialise le BasicAgent en créant une instance de l'UltraAgent."""
-        print("🚀 Initialisation du BasicAgent (powered by UltraAgent)...")
-        try:
-            # Assurez-vous que le token HF est disponible (via les secrets de l'espace HF)
-            if not os.getenv("HF_TOKEN"):
-                print("⚠️  Attention: Le token Hugging Face (HF_TOKEN) n'est pas défini. L'initialisation pourrait échouer.")
-            
-            self.ultra_agent = UltraAgent()
-            print("✅ BasicAgent initialisé avec succès!")
-        except Exception as e:
-            print(f"❌ Erreur critique lors de l'initialisation de l'UltraAgent: {e}")
-            self.ultra_agent = None
-    
-    def __call__(self, question: str) -> str:
-        """
-        Point d'entrée compatible avec l'interface de app.py.
-        Délègue l'appel à l'UltraAgent.
-        """
-        if self.ultra_agent is None:
-            return "Erreur: L'agent n'a pas pu être initialisé correctement. Veuillez vérifier les logs et la configuration (notamment le HF_TOKEN)."
-        
-        try:
-            return self.ultra_agent(question)
-        except Exception as e:
-            print(f"❌ Erreur lors du traitement par l'UltraAgent: {e}")
-            return f"Une erreur s'est produite lors du traitement de votre demande: {str(e)[:200]}"
+        return "\n".join(context_parts)