""" Agent Researcher - Premier agent du pipeline. Effectue la recherche web sur un sujet donné et retourne des sources pertinentes. """ from typing import List, Dict, Any, Optional import asyncio from datetime import datetime from src.agents.base_agent import BaseAgent from src.models.research_models import ResearchQuery, ResearchOutput, SearchResult from src.models.state_models import AgentType from src.services.search_api import SearchAPIManager, SearchAPIError from src.services.llm_service import LLMService, LLMError from src.core.logging import setup_logger from config.prompts import RESEARCHER_PROMPT, SYSTEM_PROMPTS, KEYWORD_EXTRACTION_PROMPT class ResearcherAgent(BaseAgent[ResearchQuery, ResearchOutput]): """ Agent de recherche web. Responsabilités: - Recevoir une requête de recherche - Effectuer des recherches sur le web via des APIs - Analyser et filtrer les résultats - Retourner une liste de sources pertinentes """ def __init__( self, name: str = "researcher", max_retries: int = 3, timeout: float = 120.0 # 2 minutes pour la recherche ): super().__init__( agent_type=AgentType.RESEARCHER, name=name, max_retries=max_retries, timeout=timeout ) # Initialisation du gestionnaire de recherche try: self.search_manager = SearchAPIManager() self.logger.info(f"APIs disponibles: {self.search_manager.get_available_apis()}") except Exception as e: self.logger.error(f"Impossible d'initialiser le gestionnaire de recherche: {e}") raise # Initialisation du service LLM pour l'extraction de mots-clés try: self.llm_service = LLMService() self.logger.info("Service LLM initialisé pour l'extraction de mots-clés") except Exception as e: self.logger.error(f"Impossible d'initialiser le service LLM: {e}") raise # Configuration de recherche self.default_search_params = { "preferred_api": "tavily", "search_depth": "basic", "include_answer": True } def validate_input(self, input_data: ResearchQuery) -> bool: """ Valide la requête de recherche. Args: input_data: Requête de recherche à valider Returns: True si la requête est valide """ if not input_data.topic or len(input_data.topic.strip()) < 3: self.logger.error("Le sujet de recherche doit contenir au moins 3 caractères") return False if input_data.max_results <= 0 or input_data.max_results > 20: self.logger.error("Le nombre de résultats doit être entre 1 et 20") return False return True async def process(self, input_data: ResearchQuery) -> ResearchOutput: """ Traite la requête de recherche. Args: input_data: Requête de recherche Returns: Résultats de recherche structurés """ start_time = datetime.now() self.logger.info(f"Début de recherche pour: '{input_data.topic}'") # Préparation de la requête search_query = self._prepare_search_query(input_data) self.logger.info(f"Requête préparée: '{search_query}'") # Configuration des paramètres de recherche search_params = { **self.default_search_params, "search_depth": input_data.search_depth, "max_results": input_data.max_results } try: # Recherche principale results = await self.search_manager.search( query=search_query, **search_params ) # Post-traitement des résultats filtered_results = self._filter_and_rank_results( results, input_data.topic, input_data.keywords ) # Limitation au nombre demandé final_results = filtered_results[:input_data.max_results] # Calcul du temps de recherche search_time = (datetime.now() - start_time).total_seconds() # Création de l'output research_output = ResearchOutput( query=input_data, results=final_results, total_found=len(results), search_time=search_time, search_engine=search_params["preferred_api"], timestamp=datetime.now() ) self.logger.info( f"Recherche terminée: {len(final_results)} résultats finaux " f"sur {len(results)} trouvés en {search_time:.2f}s" ) return research_output except SearchAPIError as e: self.logger.error(f"Erreur de recherche: {e}") raise except Exception as e: self.logger.error(f"Erreur inattendue lors de la recherche: {e}") raise def _prepare_search_query(self, query: ResearchQuery) -> str: """ Prépare la requête de recherche en optimisant les mots-clés. Args: query: Requête originale Returns: Requête optimisée pour la recherche """ # Commencer par le sujet principal search_terms = [query.topic] # Ajouter les mots-clés s'ils existent if query.keywords: # Éviter la redondance avec le sujet principal unique_keywords = [ kw for kw in query.keywords if kw.lower() not in query.topic.lower() ] search_terms.extend(unique_keywords) # Joindre avec des espaces search_query = " ".join(search_terms) ##################### A Améliorer selon ce qu'on veut rechercher ################################# # Optimisations spécifiques selon la profondeur ################################################################################################## if query.search_depth == "advanced": # Pour les recherches avancées, ajouter des termes de contexte if "intelligence artificielle" in search_query.lower() or "ia" in search_query.lower(): search_query += " 2024 2025 récent" if "emploi" in search_query.lower() or "travail" in search_query.lower(): search_query += " marché impact" return search_query.strip() def _filter_and_rank_results( self, results: List[SearchResult], topic: str, keywords: List[str] ) -> List[SearchResult]: """ Filtre et classe les résultats par pertinence. Args: results: Résultats bruts de la recherche topic: Sujet de recherche original keywords: Mots-clés de recherche Returns: Résultats filtrés et classés """ if not results: return [] # Mots-clés pour le scoring (topic + keywords) scoring_terms = [topic.lower()] + [kw.lower() for kw in keywords] # Calcul du score de pertinence pour chaque résultat scored_results = [] for result in results: score = self._calculate_relevance_score(result, scoring_terms) # Mise à jour du score dans le résultat result.score = score scored_results.append(result) # Tri par score décroissant scored_results.sort(key=lambda x: x.score or 0, reverse=True) # Filtrage des résultats de faible qualité min_score = 0.1 # Score minimum acceptable filtered_results = [r for r in scored_results if (r.score or 0) >= min_score] self.logger.info(f"Filtrage: {len(filtered_results)} résultats conservés sur {len(results)}") return filtered_results #Améiorer le score selon le site # EX: if result.url.endswith(".edu") or result.url.endswith(".gov"): # score += 0.1 def _calculate_relevance_score( self, result: SearchResult, scoring_terms: List[str] ) -> float: """ Calcule un score de pertinence pour un résultat. Args: result: Résultat à scorer scoring_terms: Termes de référence pour le scoring Returns: Score entre 0 et 1 """ score = 0.0 # Texte à analyser (titre + snippet) text_to_analyze = f"{result.title} {result.snippet}".lower() # Score basé sur la présence des termes de recherche term_matches = 0 for term in scoring_terms: if term in text_to_analyze: term_matches += 1 if scoring_terms: term_score = term_matches / len(scoring_terms) score += term_score * 0.6 # 60% du score # Bonus pour les titres pertinents title_matches = sum(1 for term in scoring_terms if term in result.title.lower()) if scoring_terms: title_score = title_matches / len(scoring_terms) score += title_score * 0.3 # 30% du score # Bonus pour les sources récentes (si date disponible) if result.published_date: days_old = (datetime.now() - result.published_date.replace(tzinfo=None)).days if days_old <= 365: # Moins d'un an recency_score = max(0, 1 - (days_old / 365)) score += recency_score * 0.1 # 10% du score # Score existant de l'API (si disponible) if result.score and result.score > 0: score = (score + result.score) / 2 # Moyenne avec le score API return min(score, 1.0) # Cap à 1.0 async def extract_keywords_with_llm(self, topic: str) -> List[str]: """ Extrait automatiquement des mots-clés pertinents à partir du sujet en utilisant le service LLM. Args: topic: Sujet de recherche Returns: Liste de mots-clés extraits """ try: self.logger.info(f"Extraction de mots-clés pour: '{topic}'") # Préparation du prompt avec le template prompt = KEYWORD_EXTRACTION_PROMPT.format(topic=topic) # Appel au service LLM response = await self.llm_service.generate_completion( prompt=prompt, system_prompt="Tu es un expert en analyse sémantique spécialisé dans l'extraction de mots-clés pour la recherche web.", temperature=0.3, # Faible température pour plus de cohérence max_tokens=150 # Limite pour les mots-clés ) # Parsing de la réponse keywords = self._parse_keywords_response(response) self.logger.info(f"Mots-clés extraits: {keywords}") return keywords except LLMError as e: self.logger.error(f"Erreur LLM lors de l'extraction de mots-clés: {e}") # Fallback: extraction simple basée sur le sujet return self._extract_keywords_fallback(topic) except Exception as e: self.logger.error(f"Erreur inattendue lors de l'extraction de mots-clés: {e}") return self._extract_keywords_fallback(topic) def _parse_keywords_response(self, response: str) -> List[str]: """ Parse la réponse du LLM pour extraire les mots-clés. Args: response: Réponse brute du LLM Returns: Liste de mots-clés nettoyés """ # Nettoyer la réponse response = response.strip() # Supprimer les préfixes potentiels for prefix in ["mots-clés:", "keywords:", "réponse:", "voici:", "liste:"]: if response.lower().startswith(prefix): response = response[len(prefix):].strip() # Séparer par virgules keywords = [kw.strip() for kw in response.split(",")] # Nettoyer et filtrer cleaned_keywords = [] for kw in keywords: # Supprimer les numéros et tirets kw = kw.strip("0123456789.-\t\n ") # Filtrer les mots trop courts ou vides if len(kw) >= 2 and kw.lower() not in ["et", "ou", "le", "la", "les", "de", "du", "des"]: cleaned_keywords.append(kw) # Limiter le nombre de mots-clés return cleaned_keywords[:7] def _extract_keywords_fallback(self, topic: str) -> List[str]: """ Méthode de fallback pour extraire des mots-clés simples. Args: topic: Sujet de recherche Returns: Liste de mots-clés basiques """ self.logger.info("Utilisation du fallback pour l'extraction de mots-clés") # Mots communs à ignorer stop_words = { "le", "la", "les", "de", "du", "des", "et", "ou", "sur", "dans", "avec", "pour", "par", "en", "à", "un", "une", "ce", "cette", "ces" } # Extraction simple basée sur les mots significatifs words = topic.lower().split() keywords = [word for word in words if len(word) >= 3 and word not in stop_words] return keywords[:5] # Limiter à 5 mots-clés max async def search_with_fallback( self, query: str, max_results: int = 5 ) -> List[SearchResult]: """ Méthode utilitaire pour recherche simple avec fallback. Args: query: Requête de recherche simple max_results: Nombre de résultats souhaités Returns: Liste des résultats """ research_query = ResearchQuery( topic=query, max_results=max_results ) output = await self.process(research_query) return output.results def get_search_stats(self) -> Dict[str, Any]: """ Retourne les statistiques de recherche de l'agent. Returns: Dictionnaire avec les statistiques """ base_stats = self.get_status() search_stats = { "available_apis": self.search_manager.get_available_apis(), "search_params": self.default_search_params } return {**base_stats, **search_stats} # Fonctions utilitaires pour la sauvegarde def save_research_output(output: ResearchOutput, filename: str = None) -> str: """ Sauvegarde un ResearchOutput dans un fichier JSON. Args: output: Sortie de recherche à sauvegarder filename: Nom du fichier (optionnel) Returns: Nom du fichier sauvegardé """ import json from datetime import datetime if not filename: # Générer un nom de fichier basé sur le sujet et timestamp clean_topic = "".join(c for c in output.query.topic if c.isalnum() or c in (' ', '-', '_')).rstrip() clean_topic = clean_topic.replace(' ', '_')[:30] timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"research_output_{clean_topic}_{timestamp}.json" try: # Conversion en dictionnaire avec sérialisation des dates output_dict = output.model_dump(mode='json') # Sauvegarde dans le fichier with open(filename, 'w', encoding='utf-8') as f: json.dump(output_dict, f, indent=2, ensure_ascii=False) return filename except Exception as e: raise Exception(f"Erreur lors de la sauvegarde: {e}") def load_research_output(filename: str) -> ResearchOutput: """ Charge un ResearchOutput depuis un fichier JSON. Args: filename: Nom du fichier à charger Returns: ResearchOutput chargé """ import json try: with open(filename, 'r', encoding='utf-8') as f: data = json.load(f) # Reconstruction du ResearchOutput return ResearchOutput(**data) except Exception as e: raise Exception(f"Erreur lors du chargement: {e}") # Ecrire un main pour tester ici la classe if __name__ == "__main__": import asyncio import json from datetime import datetime from src.core.logging import setup_logger logger = setup_logger("researcher_agent_test") async def main(): agent = ResearcherAgent() # Test 1: Extraction automatique de mots-clés avec LLM topic = "impact de l'intelligence artificielle sur le marché de l'emploi" logger.info(f"=== Test d'extraction de mots-clés pour: {topic} ===") try: # Extraction automatique des mots-clés keywords = await agent.extract_keywords_with_llm(topic) logger.info(f"Mots-clés extraits automatiquement: {keywords}") # Création de la requête avec les mots-clés extraits query = ResearchQuery( topic=topic, keywords=keywords, # Utilisation des mots-clés extraits automatiquement max_results=2, search_depth="basic" ) if agent.validate_input(query): logger.info("=== Début de la recherche avec mots-clés automatiques ===") output = await agent.process(query) logger.info(f"Résultats obtenus: {len(output.results)}") # Affichage des résultats for i, res in enumerate(output.results, 1): logger.info(f"{i}. {res.title}") logger.info(f" URL: {res.url}") logger.info(f" Score: {res.score:.3f}") logger.info(f" Snippet: {res.snippet[:100]}...") logger.info("") # === SAUVEGARDE DU RESEARCHOUTPUT === logger.info("=== Sauvegarde du ResearchOutput ===") try: filename = save_research_output(output) logger.info(f"✅ ResearchOutput sauvegardé dans: {filename}") # Affichage du contenu sauvegardé logger.info("📄 Contenu sauvegardé:") logger.info(f" • Sujet: {output.query.topic}") logger.info(f" • Mots-clés: {output.query.keywords}") logger.info(f" • Nombre de résultats: {len(output.results)}") logger.info(f" • Temps de recherche: {output.search_time:.2f}s") logger.info(f" • Moteur utilisé: {output.search_engine}") logger.info(f" • Timestamp: {output.timestamp}") # Test de chargement pour vérifier l'intégrité logger.info("=== Test de chargement ===") loaded_output = load_research_output(filename) logger.info(f"✅ ResearchOutput rechargé avec succès") logger.info(f" • Vérification: {len(loaded_output.results)} résultats chargés") # Comparaison des données if loaded_output.query.topic == output.query.topic: logger.info("✅ Intégrité des données vérifiée") else: logger.error("❌ Erreur d'intégrité des données") # Affichage du format JSON pour référence logger.info("\n📋 EXEMPLE DE FORMAT JSON SAUVEGARDÉ:") logger.info("-" * 50) # Créer un exemple compact pour l'affichage example_output = { "query": { "topic": output.query.topic, "keywords": output.query.keywords[:3], # Limiter pour l'affichage "max_results": output.query.max_results, "search_depth": output.query.search_depth }, "results": [ { "title": res.title, "url": str(res.url), "snippet": res.snippet[:100] + "...", "score": res.score } for res in output.results[:2] # Limiter à 2 résultats pour l'affichage ], "total_found": output.total_found, "search_time": output.search_time, "search_engine": output.search_engine, "timestamp": output.timestamp.isoformat() } print(json.dumps(example_output, indent=2, ensure_ascii=False)) except Exception as save_error: logger.error(f"❌ Erreur lors de la sauvegarde: {save_error}") else: logger.error("Requête invalide.") except Exception as e: logger.error(f"Erreur lors du test: {e}") # Fonction utilitaire pour tester la sauvegarde indépendamment async def test_save_load(): """Test spécifique de sauvegarde/chargement.""" logger.info("=== TEST SAUVEGARDE/CHARGEMENT SEUL ===") # Créer un ResearchOutput factice pour le test from datetime import datetime fake_results = [ SearchResult( title="Test Article 1", url="https://example.com/test1", snippet="Ceci est un test de snippet pour l'article 1", score=0.85 ), SearchResult( title="Test Article 2", url="https://example.com/test2", snippet="Ceci est un test de snippet pour l'article 2", score=0.78 ) ] fake_query = ResearchQuery( topic="test sauvegarde", keywords=["test", "sauvegarde", "json"], max_results=2 ) fake_output = ResearchOutput( query=fake_query, results=fake_results, total_found=2, search_time=1.5, search_engine="test", timestamp=datetime.now() ) try: # Test de sauvegarde filename = save_research_output(fake_output, "test_research_output.json") logger.info(f"✅ Test sauvegarde réussi: {filename}") # Test de chargement loaded = load_research_output(filename) logger.info(f"✅ Test chargement réussi: {len(loaded.results)} résultats") except Exception as e: logger.error(f"❌ Test sauvegarde/chargement échoué: {e}") # Choix du test à exécuter import sys if len(sys.argv) > 1 and sys.argv[1] == "--test-save": asyncio.run(test_save_load()) else: asyncio.run(main())