Update agent.py

agent.py

@@ -11,30 +11,6 @@ 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
-    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 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:
-        """
-        Formate et retourne la réponse finale de l'agent.
-        """
-        # 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,48 +22,91 @@ 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)
-        self.tools = [search_web, scrape_website, read_file, PythonInterpreterTool()]
+        
+        # Outils disponibles
+        self.tools = [
+            search_web,
+            scrape_website,
+            read_file,
+            PythonInterpreterTool()
+        ]
+        
+        # Agents spécialisés
         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."""
+        """Initialise les agents spécialisés comme l'orchestrateur et l'agent de code."""
         try:
-            self.orchestrator = OrchestratorAgent(
+            # Agent principal (orchestrateur)
+            self.orchestrator = MultiStepAgent(
                 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."""
+        """
+        Point d'entrée principal de l'agent.
+        Analyse la question, détermine la stratégie et exécute la tâche.
+        """
         print("\n" + "="*80)
-        print(f"🧠 ULTRA-AGENT - Nouvelle question reçue\nQuestion: {question[:200]}{'...' if len(question) > 200 else ''}")
+        print("🧠 ULTRA-AGENT - Nouvelle question reçue")
+        print(f"Question: {question[:200]}{'...' if len(question) > 200 else ''}")
         print("="*80)
         
         try:
-            self.conversation_history.append({"role": "user", "content": question, "timestamp": time.time()})
+            # Ajoute à l'historique
+            self.conversation_history.append({
+                "role": "user",
+                "content": question,
+                "timestamp": time.time()
+            })
+            
+            # Analyse de la question et choix de la stratégie
             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)
-            self.conversation_history.append({"role": "assistant", "content": response, "strategy": strategy, "timestamp": time.time()})
+            
+            # Ajoute la réponse à l'historique
+            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))
@@ -98,65 +117,189 @@ class UltraAgent:
                 return final_error
 
     def _analyze_question(self, question: str) -> Dict[str, Any]:
-        """Analyse la question pour déterminer la meilleure stratégie."""
-        # Cette fonction de scoring simple reste la même
+        """
+        Analyse la question pour déterminer la meilleure stratégie à adopter.
+        """
         question_lower = question.lower()
-        vision_keywords = ['image', 'photo', 'picture', 'visual', 'voir', 'regarder', 'analyser l\'image', 'screenshot']
+        
+        # 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', 'chess position', 'position']
         code_keywords = ['code', 'program', 'script', 'algorithm', 'python', 'javascript', 'sql', 'debug', 'programmer']
-        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
+        search_keywords = ['search', 'find', 'look up', 'recherche', 'chercher', 'trouver', 'internet', 'web', 'actualité', 'wikipedia', 'youtube', 'studio albums']
+        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', 'youtube.com', 'wikipedia']):
+            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}
 
     def _execute_strategy(self, question: str, strategy: Dict[str, Any]) -> str:
-        """Exécute la stratégie déterminée."""
+        """Exécute la stratégie déterminée pour répondre à la question."""
         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)
-            else: # general, search_web, file_processing, reasoning
+            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
                 return self._handle_general_task(question)
+                
         except Exception as e:
             print(f"❌ Erreur dans l'exécution de la stratégie {strategy_type}: {e}")
-            raise e # Fait remonter l'erreur pour la gestion centrale
+            # Fallback vers l'orchestrateur général en cas d'échec
+            return self._handle_general_task(question)
 
     def _handle_vision_task(self, question: str) -> str:
-        """Gère les tâches de vision."""
+        """Gère les tâches impliquant la vision."""
         print("👁️ Traitement avec le modèle de vision...")
-        response = self.model_manager.get_vision_model()(question)
-        return f"🔍 Analyse visuelle:\n{response}"
+        
+        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."
+            
+            # CORRECTION: Appel direct du modèle avec le prompt seulement
+            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)
 
     def _handle_code_task(self, question: str) -> str:
-        """Gère les tâches de code."""
+        """Gère les tâches de programmation avec l'agent de code."""
         print("💻 Traitement avec l'agent de code...")
-        return self.code_agent.run(question)
+        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."
+            
+            # CORRECTION: Appel direct du modèle avec le prompt seulement
+            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)
 
     def _handle_general_task(self, question: str) -> str:
-        """Gère toutes les autres tâches avec l'orchestrateur."""
+        """Gère les tâches générales avec l'orchestrateur principal."""
         print("🎯 Traitement général avec l'orchestrateur...")
-        context = self._get_conversation_context()
-        enhanced_prompt = f"{context}\nQuestion actuelle: {question}"
-        return self.orchestrator.run(enhanced_prompt)
+        try:
+            # CORRECTION: Utilisation directe du modèle orchestrateur au lieu de l'agent MultiStep
+            context = self._get_conversation_context()
+            enhanced_prompt = f"{context}\nQuestion actuelle: {question}\nInstructions: Analysez et répondez de manière complète à cette question."
+            
+            # Appel direct du modèle au lieu de l'orchestrateur MultiStep
+            response = self.model_manager.get_orchestrator()(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
 
     def _fallback_reasoning(self, question: str, error: str) -> str:
-        """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)
+        """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."
+            
+            # CORRECTION: Appel direct du modèle avec le prompt seulement
+            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
 
     def _get_conversation_context(self, max_exchanges: int = 3) -> str:
-        """Récupère le contexte des échanges récents."""
-        if not self.conversation_history: return ""
+        """Récupère le contexte des échanges récents pour l'injecter dans le prompt."""
+        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]
+            content = str(entry["content"])[:200] + "..." if len(str(entry["content"])) > 200 else str(entry["content"])
             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 if hasattr(tool, '__name__')],
+            "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:
     """