""" Workflow d'analyse d'images avec contexte. Ce module implémente le workflow suivant: 1. Analyse du ticket 2. Tri des images 3. Analyse des images pertinentes avec le contexte du ticket """ import logging import time import sys from typing import Dict, Any, Optional, List, Tuple # Import des factories from tests.common.agent_factory import ( create_ticket_analyser, create_image_sorter, create_image_analyser ) from tests.common.llm_factory import create_llm from tests.common.ticket_utils import get_ticket_json, get_ticket_images logger = logging.getLogger("ImageAnalyserWorkflow") def execute_workflow(ticket_id: str, output_dir: str = "output", text_model: str = "mistral_large", vision_model: str = "pixtral_large"): """ Exécute le workflow d'analyse d'image avec contexte. Args: ticket_id: Identifiant du ticket output_dir: Répertoire contenant les tickets text_model: Nom du modèle à utiliser pour l'analyse de ticket vision_model: Nom du modèle à utiliser pour l'analyse d'image Returns: Dict avec les résultats du workflow """ logger.info(f"Démarrage du workflow pour le ticket {ticket_id}") print(f"Démarrage du workflow pour le ticket {ticket_id}") workflow_start = time.time() # ÉTAPE 1: Initialiser les modèles LLM try: print("Initialisation des modèles LLM...") text_llm = create_llm(text_model) vision_llm = create_llm(vision_model) print(f"Modèles initialisés: {text_model}, {vision_model}") except Exception as e: logger.error(f"Erreur lors de l'initialisation des modèles: {e}") print(f"ERREUR: Impossible d'initialiser les modèles: {e}") return {"error": str(e), "stage": "init_llm"} # ÉTAPE 2: Initialiser les agents try: print("Création des agents...") ticket_agent = create_ticket_analyser(text_llm) image_sorter = create_image_sorter(vision_llm) image_analyser = create_image_analyser(vision_llm) print("Agents créés avec succès") except Exception as e: logger.error(f"Erreur lors de la création des agents: {e}") print(f"ERREUR: Impossible de créer les agents: {e}") return {"error": str(e), "stage": "init_agents"} # ÉTAPE 3: Charger les données du ticket try: print("Chargement des données du ticket...") json_file, ticket_data = get_ticket_json(ticket_id, output_dir) if not json_file or not ticket_data: error_msg = f"Impossible de charger les données du ticket {ticket_id}" logger.error(error_msg) print(f"ERREUR: {error_msg}") return {"error": error_msg, "stage": "load_ticket"} print(f"Données du ticket chargées: {json_file}") except Exception as e: logger.error(f"Erreur lors du chargement du ticket: {e}") print(f"ERREUR: Impossible de charger le ticket: {e}") return {"error": str(e), "stage": "load_ticket"} # ÉTAPE 4: Analyser le ticket try: print("Analyse du ticket en cours...") ticket_analysis = ticket_agent.executer(ticket_data) print(f"Analyse du ticket terminée: {len(ticket_analysis) if ticket_analysis else 0} caractères") except Exception as e: logger.error(f"Erreur lors de l'analyse du ticket: {e}") print(f"ERREUR: Impossible d'analyser le ticket: {e}") ticket_analysis = f"Erreur d'analyse: {e}" # ÉTAPE 5: Récupérer et trier les images try: print("Récupération et tri des images...") all_images = get_ticket_images(ticket_id, output_dir) if not all_images: print("Aucune image trouvée pour ce ticket") return { "ticket_id": ticket_id, "ticket_analysis": ticket_analysis, "images_count": 0, "relevant_images": [], "analysis_results": {} } print(f"Images trouvées: {len(all_images)}") # Trier les images relevant_images = [] for img_path in all_images: print(f"Évaluation de l'image: {img_path}") try: sorting_result = image_sorter.executer(img_path) is_relevant = sorting_result.get("is_relevant", False) if is_relevant: print(f" => Pertinente") relevant_images.append(img_path) else: print(f" => Non pertinente") except Exception as e: logger.error(f"Erreur lors du tri de l'image {img_path}: {e}") print(f"ERREUR: Impossible de trier l'image {img_path}: {e}") print(f"Images pertinentes: {len(relevant_images)}/{len(all_images)}") except Exception as e: logger.error(f"Erreur lors de la récupération/tri des images: {e}") print(f"ERREUR: Impossible de récupérer/trier les images: {e}") return {"error": str(e), "stage": "sort_images", "ticket_analysis": ticket_analysis} # ÉTAPE 6: Analyser les images pertinentes avec contexte analysis_results = {} if relevant_images: print("Analyse des images pertinentes avec contexte...") for img_path in relevant_images: print(f"Analyse de l'image: {img_path}") try: analysis_result = image_analyser.executer(img_path, contexte=ticket_analysis) analysis_results[img_path] = analysis_result print(f" => Analyse terminée: {len(analysis_result.get('analyse', '')) if analysis_result else 0} caractères") except Exception as e: logger.error(f"Erreur lors de l'analyse de l'image {img_path}: {e}") print(f"ERREUR: Impossible d'analyser l'image {img_path}: {e}") analysis_results[img_path] = {"error": str(e)} else: print("Aucune image pertinente à analyser") # Calcul du temps total d'exécution workflow_time = time.time() - workflow_start print(f"Workflow terminé en {workflow_time:.2f} secondes") # Retourner les résultats return { "ticket_id": ticket_id, "ticket_analysis": ticket_analysis, "images_count": len(all_images), "relevant_images": relevant_images, "analysis_results": analysis_results, "execution_time": workflow_time } def main(): # Vérifier les arguments if len(sys.argv) < 2: print("ERREUR: Vous devez fournir un ID de ticket") print("Usage: python -m tests.workflows.image_analyser_workflow [output_dir] [text_model] [vision_model]") return 1 # Récupérer les arguments ticket_id = sys.argv[1] output_dir = sys.argv[2] if len(sys.argv) > 2 else "output" text_model = sys.argv[3] if len(sys.argv) > 3 else "mistral_large" vision_model = sys.argv[4] if len(sys.argv) > 4 else "pixtral_large" # Configurer le logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', handlers=[ logging.FileHandler(f"image_analyser_workflow_{ticket_id}.log"), logging.StreamHandler() ] ) # Exécuter le workflow results = execute_workflow(ticket_id, output_dir, text_model, vision_model) # Afficher un résumé des résultats if "error" in results: print(f"\nErreur lors de l'exécution du workflow: {results['error']}") print(f"Étape en échec: {results.get('stage', 'inconnue')}") return 1 print("\nRésumé du workflow:") print(f"Ticket: {results['ticket_id']}") print(f"Analyse du ticket: {len(results['ticket_analysis']) if results['ticket_analysis'] else 0} caractères") print(f"Images totales: {results['images_count']}") print(f"Images pertinentes: {len(results['relevant_images'])}") print(f"Images analysées: {len(results['analysis_results'])}") print(f"Temps d'exécution: {results['execution_time']:.2f} secondes") return 0 if __name__ == "__main__": sys.exit(main())