import os import json import logging import time import traceback from typing import List, Dict, Any, Optional, Union from agents.base_agent import BaseAgent from utils.ticket_data_loader import TicketDataLoader # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='orchestrator.log', filemode='w') logger = logging.getLogger("Orchestrator") class Orchestrator: def __init__(self, output_dir: str = "output/", ticket_agent: Optional[BaseAgent] = None, image_sorter: Optional[BaseAgent] = None, image_analyser: Optional[BaseAgent] = None, report_generator: Optional[BaseAgent] = None): self.output_dir = output_dir # Assignation directe des agents self.ticket_agent = ticket_agent self.image_sorter = image_sorter self.image_analyser = image_analyser self.report_generator = report_generator # Initialisation du loader de données de ticket self.ticket_loader = TicketDataLoader() # Collecter et enregistrer les informations détaillées sur les agents agents_info = self._collecter_info_agents() logger.info(f"Orchestrator initialisé avec output_dir: {output_dir}") logger.info(f"Agents disponibles: TicketAgent={ticket_agent is not None}, ImageSorter={image_sorter is not None}, ImageAnalyser={image_analyser is not None}, ReportGenerator={report_generator is not None}") logger.info(f"Configuration des agents: {json.dumps(agents_info, indent=2)}") def _collecter_info_agents(self) -> Dict[str, Dict[str, Any]]: """ Collecte des informations détaillées sur les agents configurés """ agents_info = {} # Information sur l'agent Ticket if self.ticket_agent: agents_info["ticket_agent"] = self._get_agent_info(self.ticket_agent) # Information sur l'agent Image Sorter if self.image_sorter: agents_info["image_sorter"] = self._get_agent_info(self.image_sorter) # Information sur l'agent Image Analyser if self.image_analyser: agents_info["image_analyser"] = self._get_agent_info(self.image_analyser) # Information sur l'agent Report Generator if self.report_generator: agents_info["report_generator"] = self._get_agent_info(self.report_generator) return agents_info def detecter_tickets(self) -> List[str]: """Détecte tous les tickets disponibles dans le répertoire de sortie""" logger.info(f"Recherche de tickets dans: {self.output_dir}") tickets = [] if not os.path.exists(self.output_dir): logger.warning(f"Le répertoire de sortie {self.output_dir} n'existe pas") print(f"ERREUR: Le répertoire {self.output_dir} n'existe pas") return tickets for ticket_dir in os.listdir(self.output_dir): ticket_path = os.path.join(self.output_dir, ticket_dir) if os.path.isdir(ticket_path) and ticket_dir.startswith("ticket_"): tickets.append(ticket_path) logger.info(f"Tickets trouvés: {len(tickets)}") print(f"Tickets détectés: {len(tickets)}") return tickets def lister_tickets(self) -> Dict[int, str]: """Liste les tickets disponibles et retourne un dictionnaire {index: chemin}""" tickets = self.detecter_tickets() ticket_dict = {} print("\nTickets disponibles:") for i, ticket_path in enumerate(tickets, 1): ticket_id = os.path.basename(ticket_path) ticket_dict[i] = ticket_path print(f"{i}. {ticket_id}") return ticket_dict def trouver_rapport(self, extraction_path: str, ticket_id: str) -> Dict[str, Optional[str]]: """ Cherche le rapport du ticket dans différents emplacements possibles (JSON ou MD) Args: extraction_path: Chemin de l'extraction ticket_id: ID du ticket (ex: T0101) Returns: Un dictionnaire avec les chemins des fichiers JSON et MD s'ils sont trouvés """ # Utilise la nouvelle méthode de TicketDataLoader resultats = self.ticket_loader.trouver_ticket(extraction_path, ticket_id) if resultats is None: return {"json": None, "markdown": None} return resultats def traiter_ticket(self, ticket_path: str) -> bool: """Traite un ticket spécifique et retourne True si le traitement a réussi""" logger.info(f"Début du traitement du ticket: {ticket_path}") print(f"\nTraitement du ticket: {os.path.basename(ticket_path)}") success = False extractions_trouvees = False if not os.path.exists(ticket_path): logger.error(f"Le chemin du ticket n'existe pas: {ticket_path}") print(f"ERREUR: Le chemin du ticket n'existe pas: {ticket_path}") return False ticket_id = os.path.basename(ticket_path).replace("ticket_", "") for extraction in os.listdir(ticket_path): extraction_path = os.path.join(ticket_path, extraction) if os.path.isdir(extraction_path): extractions_trouvees = True logger.info(f"Traitement de l'extraction: {extraction}") print(f" Traitement de l'extraction: {extraction}") # Recherche des rapports (JSON et MD) dans différents emplacements rapports = self.trouver_rapport(extraction_path, ticket_id) # Dossier des pièces jointes attachments_dir = os.path.join(extraction_path, "attachments") # Dossier pour les rapports générés rapports_dir = os.path.join(extraction_path, f"{ticket_id}_rapports") os.makedirs(rapports_dir, exist_ok=True) # Préparer les données du ticket à partir des rapports trouvés ticket_data = self._preparer_donnees_ticket(rapports, ticket_id) if ticket_data: success = True logger.info(f"Données du ticket chargées avec succès") print(f" Données du ticket chargées") # Traitement avec l'agent Ticket if self.ticket_agent: logger.info("Exécution de l'agent Ticket") print(" Analyse du ticket en cours...") # Log détaillé sur l'agent Ticket agent_info = self._get_agent_info(self.ticket_agent) logger.info(f"Agent Ticket: {json.dumps(agent_info, indent=2)}") ticket_analysis = self.ticket_agent.executer(ticket_data) logger.info("Analyse du ticket terminée") print(f" Analyse du ticket terminée: {len(ticket_analysis) if ticket_analysis else 0} caractères") else: logger.warning("Agent Ticket non disponible") ticket_analysis = None print(" Agent Ticket non disponible, analyse ignorée") # Traitement des images relevant_images = [] images_analyses = {} images_count = 0 if os.path.exists(attachments_dir): logger.info(f"Vérification des pièces jointes dans: {attachments_dir}") print(f" Vérification des pièces jointes...") # Log détaillé sur l'agent Image Sorter if self.image_sorter: agent_info = self._get_agent_info(self.image_sorter) logger.info(f"Agent Image Sorter: {json.dumps(agent_info, indent=2)}") # Compter le nombre d'images images = [f for f in os.listdir(attachments_dir) if f.lower().endswith(('.png', '.jpg', '.jpeg', '.bmp', '.gif'))] images_count = len(images) # Tri des images for attachment in images: attachment_path = os.path.join(attachments_dir, attachment) if self.image_sorter: logger.info(f"Évaluation de la pertinence de l'image: {attachment}") print(f" Évaluation de l'image: {attachment}") sorting_result = self.image_sorter.executer(attachment_path) is_relevant = sorting_result.get("is_relevant", False) reason = sorting_result.get("reason", "") # Log détaillé du résultat if is_relevant: logger.info(f"Image {attachment} considérée comme pertinente") else: logger.info(f"Image {attachment} considérée comme non pertinente") # Ajouter les métadonnées de tri à la liste des analyses images_analyses[attachment_path] = { "sorting": sorting_result, "analysis": None # Sera rempli plus tard si pertinent } if is_relevant: logger.info(f"Image pertinente identifiée: {attachment} ({reason})") print(f" => Pertinente: {reason}") relevant_images.append(attachment_path) else: logger.info(f"Image non pertinente: {attachment} ({reason})") print(f" => Non pertinente: {reason}") else: logger.warning("Image Sorter non disponible") # Si pas de tri, considérer toutes les images comme pertinentes relevant_images.append(attachment_path) images_analyses[attachment_path] = { "sorting": {"is_relevant": True, "reason": "Auto-sélectionné (pas de tri)"}, "analysis": None } print(f" => Auto-sélectionné (pas de tri)") logger.info(f"Images analysées: {images_count}, Images pertinentes: {len(relevant_images)}") print(f" Images analysées: {images_count}, Images pertinentes: {len(relevant_images)}") else: logger.warning(f"Répertoire des pièces jointes non trouvé: {attachments_dir}") print(f" Répertoire des pièces jointes non trouvé") # Analyse approfondie des images pertinentes if relevant_images and self.image_analyser: agent_info = self._get_agent_info(self.image_analyser) logger.info(f"Agent Image Analyser: {json.dumps(agent_info, indent=2)}") # S'assurer que l'analyse du ticket est disponible comme contexte contexte_ticket = ticket_analysis if ticket_analysis else "Aucune analyse de ticket disponible" # Analyse de chaque image pertinente for image_path in relevant_images: image_name = os.path.basename(image_path) logger.info(f"Analyse approfondie de l'image: {image_name}") print(f" Analyse approfondie de l'image: {image_name}") # Appeler l'analyseur d'images avec le contexte du ticket analysis_result = self.image_analyser.executer(image_path, contexte=contexte_ticket) # Vérifier si l'analyse a réussi if "error" in analysis_result and analysis_result["error"]: logger.warning(f"Erreur lors de l'analyse de l'image {image_name}: {analysis_result.get('analyse', 'Erreur inconnue')}") print(f" => ERREUR: {analysis_result.get('analyse', 'Erreur inconnue')}") else: logger.info(f"Analyse réussie pour l'image {image_name}") print(f" => Analyse réussie: {len(analysis_result.get('analyse', '')) if 'analyse' in analysis_result else 0} caractères") # Ajouter l'analyse au dictionnaire des analyses d'images if image_path in images_analyses: images_analyses[image_path]["analysis"] = analysis_result else: images_analyses[image_path] = { "sorting": {"is_relevant": True, "reason": "Auto-sélectionné"}, "analysis": analysis_result } logger.info(f"Analyse complétée pour {image_name}") # Préparer les données pour le rapport final rapport_data = { "ticket_data": ticket_data, "ticket_id": ticket_id, "ticket_analyse": ticket_analysis, # Utiliser ticket_analyse au lieu de analyse_ticket pour cohérence "analyse_images": images_analyses, "metadata": { "timestamp_debut": self._get_timestamp(), "ticket_id": ticket_id, "images_analysees": images_count, "images_pertinentes": len(relevant_images) } } # Ajout de la clé alternative pour compatibilité rapport_data["analyse_json"] = ticket_analysis if self.report_generator: logger.info("Génération du rapport final") print(" Génération du rapport final") # Log détaillé sur l'agent Report Generator agent_info = self._get_agent_info(self.report_generator) logger.info(f"Agent Report Generator: {json.dumps(agent_info, indent=2)}") # Créer le répertoire pour le rapport si nécessaire rapport_path = os.path.join(rapports_dir, ticket_id) os.makedirs(rapport_path, exist_ok=True) # Générer le rapport json_path, md_path = self.report_generator.executer(rapport_data, rapport_path) if json_path and md_path: logger.info(f"Rapport généré à: {rapport_path}") print(f" Rapport généré avec succès") print(f" - JSON: {os.path.basename(json_path)}") print(f" - Markdown: {os.path.basename(md_path)}") else: logger.warning("Erreur lors de la génération du rapport") print(f" ERREUR: Problème lors de la génération du rapport") else: logger.warning("Report Generator non disponible") print(" Report Generator non disponible, génération de rapport ignorée") print(f"Traitement du ticket {os.path.basename(ticket_path)} terminé avec succès.\n") logger.info(f"Traitement du ticket {ticket_path} terminé avec succès.") else: logger.warning(f"Aucune donnée de ticket trouvée pour: {ticket_id}") print(f" ERREUR: Aucune donnée de ticket trouvée pour {ticket_id}") if not extractions_trouvees: logger.warning(f"Aucune extraction trouvée dans le ticket: {ticket_path}") print(f" ERREUR: Aucune extraction trouvée dans le ticket") return success def _preparer_donnees_ticket(self, rapports: Dict[str, Optional[str]], ticket_id: str) -> Optional[Dict]: """ Prépare les données du ticket à partir des rapports trouvés (JSON et/ou MD) Args: rapports: Dictionnaire avec les chemins des rapports JSON et MD ticket_id: ID du ticket Returns: Dictionnaire avec les données du ticket, ou None si aucun rapport n'est trouvé """ ticket_data = None # Si aucun rapport n'est trouvé if not rapports or (not rapports.get("json") and not rapports.get("markdown")): logger.warning(f"Aucun rapport trouvé pour le ticket {ticket_id}") return None # Essayer d'abord le fichier JSON if rapports.get("json") and rapports["json"] is not None: try: ticket_data = self.ticket_loader.charger(rapports["json"]) logger.info(f"Données JSON chargées depuis: {rapports['json']}") print(f" Rapport JSON chargé: {os.path.basename(rapports['json'])}") except Exception as e: logger.error(f"Erreur lors du chargement du JSON: {e}") print(f" ERREUR: Impossible de charger le fichier JSON: {e}") # Si pas de JSON ou erreur, essayer le Markdown if not ticket_data and rapports.get("markdown") and rapports["markdown"] is not None: try: # Utiliser le loader pour charger les données depuis le Markdown ticket_data = self.ticket_loader.charger(rapports["markdown"]) logger.info(f"Données Markdown chargées depuis: {rapports['markdown']}") print(f" Rapport Markdown chargé: {os.path.basename(rapports['markdown'])}") except Exception as e: logger.error(f"Erreur lors du chargement du Markdown: {e}") print(f" ERREUR: Impossible de charger le fichier Markdown: {e}") # Assurer que l'ID du ticket est correct if ticket_data: ticket_data["code"] = ticket_id return ticket_data def executer(self, ticket_specifique: Optional[str] = None): """ Exécute l'orchestrateur soit sur un ticket spécifique, soit permet de choisir Args: ticket_specifique: Chemin du ticket spécifique à traiter (optionnel) """ start_time = time.time() # Stocker le ticket spécifique self.ticket_specifique = ticket_specifique # Obtenir la liste des tickets if ticket_specifique: # Utiliser juste le ticket spécifique ticket_dirs = self.detecter_tickets() ticket_dirs = [t for t in ticket_dirs if t.endswith(ticket_specifique)] logger.info(f"Ticket spécifique à traiter: {ticket_specifique}") else: # Lister tous les tickets ticket_dirs = self.detecter_tickets() logger.info(f"Tickets à traiter: {len(ticket_dirs)}") if not ticket_dirs: logger.warning("Aucun ticket trouvé dans le répertoire de sortie") return # Un seul log de début d'exécution logger.info("Début de l'exécution de l'orchestrateur") print("Début de l'exécution de l'orchestrateur") # Traitement des tickets for ticket_dir in ticket_dirs: if ticket_specifique and not ticket_dir.endswith(ticket_specifique): continue try: self.traiter_ticket(ticket_dir) except Exception as e: logger.error(f"Erreur lors du traitement du ticket {ticket_dir}: {str(e)}") print(f"Erreur lors du traitement du ticket {ticket_dir}: {str(e)}") traceback.print_exc() # Calcul de la durée d'exécution duration = time.time() - start_time logger.info(f"Fin de l'exécution de l'orchestrateur (durée: {duration:.2f} secondes)") print(f"Fin de l'exécution de l'orchestrateur (durée: {duration:.2f} secondes)") def _get_timestamp(self) -> str: """Retourne un timestamp au format YYYYMMDD_HHMMSS""" from datetime import datetime return datetime.now().strftime("%Y%m%d_%H%M%S") def _get_agent_info(self, agent: Optional[BaseAgent]) -> Dict: """ Récupère les informations détaillées sur un agent. """ if not agent: return {"status": "non configuré"} # Récupérer les informations du modèle model_info = { "nom": agent.nom, "model": getattr(agent.llm, "modele", str(type(agent.llm))), } # Ajouter les paramètres de configuration s'ils sont disponibles directement dans l'agent # Utiliser getattr avec une valeur par défaut pour éviter les erreurs model_info["temperature"] = getattr(agent, "temperature", None) model_info["top_p"] = getattr(agent, "top_p", None) model_info["max_tokens"] = getattr(agent, "max_tokens", None) # Ajouter le prompt système s'il est disponible if hasattr(agent, "system_prompt"): prompt_preview = getattr(agent, "system_prompt", "") # Tronquer le prompt s'il est trop long if prompt_preview and len(prompt_preview) > 200: prompt_preview = prompt_preview[:200] + "..." model_info["system_prompt_preview"] = prompt_preview # Supprimer les valeurs None model_info = {k: v for k, v in model_info.items() if v is not None} return model_info