#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Script de test pour exécuter l'orchestrateur sur un ticket spécifique. Utilisation: python test_orchestrator.py [code_ticket] Exemple: python test_orchestrator.py T0101 """ import os import sys import time import logging import traceback from datetime import datetime # Import des agents from agents.agent_ticket_analyser import AgentTicketAnalyser from agents.agent_image_sorter import AgentImageSorter from agents.agent_image_analyser import AgentImageAnalyser from agents.agent_report_generator import AgentReportGenerator # Import des modèles LLM from llm_classes.mistral_large import MistralLarge from llm_classes.pixtral_12b import Pixtral12b # Import de l'orchestrateur from orchestrator import Orchestrator # Configuration du logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='test_orchestrator.log', filemode='w') logger = logging.getLogger("TestOrchestrator") def test_orchestrator(ticket_id=None): """ Exécute l'orchestrateur avec les agents définis Args: ticket_id: Identifiant du ticket à traiter (optionnel) """ # Vérifier que le dossier output existe if not os.path.exists("output/"): os.makedirs("output/") logger.warning("Le dossier output/ n'existait pas et a été créé") print("ATTENTION: Le dossier output/ n'existait pas et a été créé") # Vérifier le contenu du dossier output tickets = [d for d in os.listdir("output/") if d.startswith("ticket_") and os.path.isdir(os.path.join("output/", d))] logger.info(f"Tickets trouvés dans output/: {len(tickets)}") print(f"Tickets existants dans output/: {len(tickets)}") if len(tickets) == 0: logger.error("Aucun ticket trouvé dans le dossier output/") print("ERREUR: Aucun ticket trouvé dans le dossier output/") return # Initialisation des LLM print("Initialisation des modèles LLM...") start_time = time.time() # Utilisation de Mistral Medium pour l'analyse JSON et la génération de rapports json_llm = MistralLarge() logger.info("LLM MistralMedium initialisé pour l'analyse JSON") # Utilisation de Pixtral12b pour le tri et l'analyse d'images image_sorter_llm = Pixtral12b() logger.info("LLM Pixtral12b initialisé pour le tri d'images") image_analyser_llm = Pixtral12b() logger.info("LLM Pixtral12b initialisé pour l'analyse d'images") report_generator_llm = MistralLarge() logger.info("LLM MistralMedium initialisé pour la génération de rapports") llm_init_time = time.time() - start_time print(f"Tous les modèles LLM ont été initialisés en {llm_init_time:.2f} secondes") # Création des agents print("Création des agents...") ticket_agent = AgentTicketAnalyser(json_llm) image_sorter = AgentImageSorter(image_sorter_llm) image_analyser = AgentImageAnalyser(image_analyser_llm) report_generator = AgentReportGenerator(report_generator_llm) # Renforcer le system prompt du générateur de rapport pour s'assurer que le tableau est généré report_generator.system_prompt = """Tu es un expert en génération de rapports techniques pour BRG-Lab. Ta mission est de synthétiser les analyses (ticket et images) en un rapport structuré et exploitable. IMPORTANCE DES ÉCHANGES CLIENT/SUPPORT: - Tu dois IMPÉRATIVEMENT présenter les échanges client/support sous forme d'un TABLEAU MARKDOWN clair - Chaque ligne du tableau doit contenir: Date | Émetteur | Type (Question/Réponse) | Contenu - Identifie clairement qui est l'émetteur (CLIENT ou SUPPORT) - Mets en évidence les questions posées et les réponses fournies Structure ton rapport: 1. Résumé exécutif: Synthèse du problème initial (nom de la demande + description) 2. Chronologie des échanges: TABLEAU des interactions client/support 3. Analyse des images: Ce que montrent les captures d'écran et leur pertinence 4. Diagnostic technique: Interprétation des informations techniques pertinentes Reste factuel et précis. Ne fais pas d'analyses inutiles ou de recommandations non fondées. Ton rapport doit mettre en avant la chronologie des échanges et les informations techniques clés.""" print("Tous les agents ont été créés") # Initialisation de l'orchestrateur avec les agents logger.info("Initialisation de l'orchestrateur") print("Initialisation de l'orchestrateur") orchestrator = Orchestrator( output_dir="output/", ticket_agent=ticket_agent, image_sorter=image_sorter, image_analyser=image_analyser, report_generator=report_generator ) # Vérification du ticket spécifique si fourni specific_ticket_path = None if ticket_id: target_ticket = f"ticket_{ticket_id}" specific_ticket_path = os.path.join("output", target_ticket) if not os.path.exists(specific_ticket_path): logger.error(f"Le ticket {target_ticket} n'existe pas") print(f"ERREUR: Le ticket {target_ticket} n'existe pas") return logger.info(f"Ticket spécifique à traiter: {specific_ticket_path}") print(f"Ticket spécifique à traiter: {target_ticket}") # Exécution de l'orchestrateur total_start_time = time.time() logger.info("Début de l'exécution de l'orchestrateur") print("Début de l'exécution de l'orchestrateur") try: orchestrator.executer(ticket_id) # Vérifier le rapport généré et afficher un résumé if ticket_id: # Chercher le rapport Markdown le plus récent ticket_dir = os.path.join("output", f"ticket_{ticket_id}") latest_md = None for extraction in os.listdir(ticket_dir): extraction_path = os.path.join(ticket_dir, extraction) if os.path.isdir(extraction_path): rapports_dir = os.path.join(extraction_path, f"{ticket_id}_rapports", f"{ticket_id}") if os.path.exists(rapports_dir): md_files = [f for f in os.listdir(rapports_dir) if f.endswith('.md')] if md_files: md_files.sort(reverse=True) # Le plus récent en premier latest_md = os.path.join(rapports_dir, md_files[0]) break if latest_md: print(f"\nVérification du rapport: {latest_md}") try: with open(latest_md, 'r', encoding='utf-8') as f: content = f.read() # Vérifier si le tableau des échanges est présent has_table = "| Date | " in content has_details = "Détails des analyses effectuées" in content print(f"- Tableau des échanges: {'Présent' if has_table else 'MANQUANT'}") print(f"- Détails des analyses: {'Présent' if has_details else 'MANQUANT'}") if not has_table: print("\nATTENTION: Le tableau des échanges client/support est manquant!") print("Vérifiez le system prompt de l'agent de rapport et la transmission des données.") except Exception as e: print(f"Erreur lors de la vérification du rapport: {e}") except Exception as e: logger.error(f"Erreur lors de l'exécution de l'orchestrateur: {str(e)}") print(f"ERREUR: {str(e)}") traceback.print_exc() total_time = time.time() - total_start_time logger.info(f"Fin de l'exécution de l'orchestrateur (durée: {total_time:.2f} secondes)") print(f"Fin de l'exécution de l'orchestrateur (durée: {total_time:.2f} secondes)") if __name__ == "__main__": print("Démarrage du test de l'orchestrateur") # Vérifier si un ID de ticket est passé en argument ticket_id = None if len(sys.argv) > 1: ticket_id = sys.argv[1] print(f"ID de ticket fourni en argument: {ticket_id}") test_orchestrator(ticket_id) print("Test terminé")