llm_ticket3/test_ticket_images_analysis_large.py
2025-04-21 16:04:53 +02:00

674 lines
28 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Script pour tester la chaîne d'analyse complète avec priorité au tri d'images:
1. Tri des images
2. Analyse du ticket
3. Analyse des images pertinentes
4. Génération d'un rapport final
"""
import os
import sys
import argparse
import json
import glob
from typing import Dict, Any, Optional, List
from llm_classes.mistral_large import MistralLarge
from llm_classes.pixtral_large import PixtralLarge
from agents.mistral_large.agent_ticket_analyser import AgentTicketAnalyser
from agents.pixtral_large.agent_image_sorter import AgentImageSorter
from agents.pixtral_large.agent_image_analyser import AgentImageAnalyser
from agents.mistral_large.agent_report_generator import AgentReportGenerator
from utils.image_dedup import filtrer_images_uniques
def get_ticket_report_file(ticket_id: str, output_dir: str) -> Optional[str]:
"""
Récupère le fichier de rapport du ticket dans le répertoire codeticket_rapports.
Args:
ticket_id: ID du ticket (ex: T1234)
output_dir: Répertoire de base contenant les tickets
Returns:
Chemin vers le fichier de rapport JSON, ou None si non trouvé
"""
# Construire le chemin vers le répertoire des rapports
ticket_path = os.path.join(output_dir, f"ticket_{ticket_id}")
# Chercher le répertoire d'extraction le plus récent
extractions = [d for d in os.listdir(ticket_path) if os.path.isdir(os.path.join(ticket_path, d)) and d.startswith(ticket_id)]
if not extractions:
return None
# Trier par ordre décroissant pour avoir la plus récente en premier
extractions.sort(reverse=True)
latest_extraction = extractions[0]
# Construire le chemin vers le répertoire des rapports
rapport_dir = os.path.join(ticket_path, latest_extraction, f"{ticket_id}_rapports")
rapport_file = os.path.join(rapport_dir, f"{ticket_id}_rapport.json")
if os.path.exists(rapport_file):
return rapport_file
return None
def get_images_in_extraction(extraction_path: str) -> List[str]:
"""
Récupère toutes les images dans un répertoire d'extraction
Args:
extraction_path: Chemin vers le répertoire d'extraction
Returns:
Liste des chemins d'accès aux images
"""
# Extensions d'images courantes
image_extensions = ['*.jpg', '*.jpeg', '*.png', '*.gif', '*.webp']
# Liste pour stocker les chemins d'images
images = []
# Chercher dans le répertoire principal
for extension in image_extensions:
pattern = os.path.join(extraction_path, extension)
images.extend(glob.glob(pattern))
# Chercher dans le sous-répertoire "attachments"
attachments_path = os.path.join(extraction_path, "attachments")
if os.path.exists(attachments_path) and os.path.isdir(attachments_path):
for extension in image_extensions:
pattern = os.path.join(attachments_path, extension)
images.extend(glob.glob(pattern))
# Chercher dans les sous-répertoires de "attachments" (si des ID sont utilisés)
for subdir in os.listdir(attachments_path):
subdir_path = os.path.join(attachments_path, subdir)
if os.path.isdir(subdir_path):
for extension in image_extensions:
pattern = os.path.join(subdir_path, extension)
images.extend(glob.glob(pattern))
return images
def main():
# Configuration de l'analyseur d'arguments
parser = argparse.ArgumentParser(description="Tester la chaîne d'analyse avec priorité au tri d'images.")
parser.add_argument("ticket_id", help="ID du ticket à analyser (ex: T1234)")
parser.add_argument("--output_dir", default="output", help="Répertoire de sortie contenant les tickets")
parser.add_argument("--ticket_file", help="Chemin spécifique vers un fichier JSON du ticket à analyser")
parser.add_argument("--no-dedup", action="store_true", help="Désactiver le préfiltrage des doublons")
parser.add_argument("--seuil", type=int, default=5, help="Seuil de similarité pour détecter les doublons (0-10, défaut=5)")
parser.add_argument("--save_results", action="store_true", help="Sauvegarder les résultats dans un fichier JSON")
parser.add_argument("--image", help="Chemin spécifique vers une image à analyser")
parser.add_argument("--show-analysis", action="store_true", help="Afficher l'analyse du ticket et des images")
parser.add_argument("--debug", action="store_true", help="Afficher des informations de débogage supplémentaires")
parser.add_argument("--generate-report", action="store_true", help="Générer un rapport final à partir des analyses")
parser.add_argument("--interactive", action="store_true", help="Utiliser le menu interactif")
parser.add_argument("--mode", type=int, choices=[1, 2, 3, 4],
help="Mode d'analyse: 1=Tri uniquement, 2=Tri+Ticket, 3=Tri+Ticket+Images, 4=Analyse complète avec rapport")
# Analyser les arguments
args = parser.parse_args()
# Mode interactif si demandé
if args.interactive:
args = _interactive_menu(args)
# Si un mode est spécifié, configurer les options correspondantes
if args.mode:
if args.mode >= 1: # Tous les modes incluent le tri
pass # Tri toujours activé
if args.mode >= 2: # Modes 2, 3, 4 incluent l'analyse de ticket
pass # Analyse de ticket toujours activée
if args.mode >= 3: # Modes 3, 4 incluent l'analyse d'images
pass # Analyse d'images toujours activée
if args.mode >= 4: # Mode 4 inclut la génération de rapport
args.generate_report = True
# Construire le chemin vers le ticket
ticket_path = os.path.join(args.output_dir, f"ticket_{args.ticket_id}")
# Vérifier que le répertoire du ticket existe
if not os.path.exists(ticket_path):
print(f"ERREUR: Le ticket {args.ticket_id} n'existe pas dans {args.output_dir}")
return 1
# Rechercher la dernière extraction (la plus récente)
extractions = [d for d in os.listdir(ticket_path) if os.path.isdir(os.path.join(ticket_path, d)) and d.startswith(args.ticket_id)]
if not extractions:
print(f"ERREUR: Aucune extraction trouvée pour le ticket {args.ticket_id}")
return 1
# Trier par ordre décroissant pour avoir la plus récente en premier
extractions.sort(reverse=True)
latest_extraction = extractions[0]
extraction_path = os.path.join(ticket_path, latest_extraction)
print(f"Utilisation de l'extraction: {latest_extraction}")
# ============ TRAITEMENT D'UNE SEULE IMAGE ============
if args.image:
# 1. Initialiser uniquement l'agent de tri d'images
try:
print("Initialisation du modèle Pixtral-Large pour le tri...")
model_tri = PixtralLarge()
# Configuration explicite du modèle
model_tri.configurer(temperature=0.2, top_p=0.8, max_tokens=300)
image_sorter = AgentImageSorter(model_tri)
print("Agent de tri d'images initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle de tri: {str(e)}")
return 1
# 2. Trier l'image
image_path = args.image
if not os.path.exists(image_path):
print(f"ERREUR: L'image spécifiée n'existe pas: {image_path}")
return 1
print(f"Analyse de l'image: {os.path.basename(image_path)}")
resultat = image_sorter.executer(image_path)
# Afficher le résultat du tri
print("\nRésultat de l'analyse:")
print(f"Pertinence: {'OUI' if resultat.get('is_relevant', False) else 'NON'}")
print(f"Raison: {resultat.get('reason', 'Non spécifiée')}")
# Si l'image est pertinente, on peut procéder à l'analyse du ticket puis à l'analyse de l'image
if resultat.get('is_relevant', False):
# 3. Charger et analyser le ticket
ticket_data = _charger_ticket_data(args)
if not ticket_data:
return 1
# 4. Initialiser l'agent d'analyse de tickets et exécuter l'analyse
try:
print("Initialisation du modèle Mistral Large...")
text_model = MistralLarge()
ticket_agent = AgentTicketAnalyser(text_model)
print("Agent d'analyse de tickets initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle d'analyse de tickets: {str(e)}")
return 1
# Exécuter l'analyse du ticket
print(f"\nAnalyse du ticket {ticket_data.get('code', args.ticket_id)} en cours...")
ticket_analysis = ticket_agent.executer(ticket_data)
print(f"Analyse du ticket terminée: {len(ticket_analysis)} caractères")
if args.show_analysis:
print("\nRésultat de l'analyse du ticket:")
print(ticket_analysis)
# 5. Initialiser l'agent d'analyse d'images et exécuter l'analyse
try:
print("Initialisation du modèle Pixtral-Large pour l'analyse d'images...")
model_analyse = PixtralLarge()
# Configuration explicite du modèle
model_analyse.configurer(temperature=0.2, top_p=0.9, max_tokens=3000)
image_analyser = AgentImageAnalyser(model_analyse)
print("Agent d'analyse d'images initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle d'analyse d'images: {str(e)}")
return 1
# Analyser l'image avec le contexte du ticket
print(f"\nAnalyse approfondie de l'image: {os.path.basename(image_path)}")
analysis_result = image_analyser.executer(image_path, contexte=ticket_analysis)
if "error" in analysis_result and analysis_result.get("error", False):
print(f" => Erreur: {analysis_result.get('analyse', '')}")
else:
analyse_text = analysis_result.get("analyse", "")
print(f" => Analyse terminée: {len(analyse_text)} caractères")
if args.show_analysis:
print("\nRésultat de l'analyse d'image:")
print(analyse_text)
else:
print(f" => Début de l'analyse: {analyse_text[:100]}...")
# 6. Générer un rapport final si demandé
if args.generate_report:
try:
print("\nInitialisation de l'agent de génération de rapport...")
report_model = MistralLarge()
report_generator = AgentReportGenerator(report_model)
print("Agent de génération de rapport initialisé avec succès")
# Préparer les données pour le rapport
rapport_data = {
"ticket_id": args.ticket_id,
"ticket_data": ticket_data,
"ticket_analyse": ticket_analysis,
"analyse_images": {
image_path: {
"sorting": resultat,
"analysis": analysis_result
}
}
}
# Générer le rapport
print("\nGénération du rapport final...")
rapport_final = report_generator.executer(rapport_data)
print(f"Rapport généré: {len(rapport_final)} caractères")
if args.show_analysis:
print("\nRapport final:")
print(rapport_final)
else:
print(f"Début du rapport: {rapport_final[:200]}...")
except Exception as e:
print(f"ERREUR: Impossible de générer le rapport: {str(e)}")
return 0
# ============ TRAITEMENT COMPLET DU TICKET ============
# ÉTAPE 1: Récupérer et trier les images avec un agent isolé
# Initialiser uniquement l'agent de tri d'images
try:
print("Initialisation du modèle Pixtral-Large pour le tri...")
model_tri = PixtralLarge()
# Configuration explicite du modèle
model_tri.configurer(temperature=0.2, top_p=0.8, max_tokens=300)
image_sorter = AgentImageSorter(model_tri)
print("Agent de tri d'images initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle de tri: {str(e)}")
return 1
print("\nRécupération et tri des images...")
all_images = get_images_in_extraction(extraction_path)
if not all_images:
print(f"Aucune image trouvée dans l'extraction {latest_extraction}")
return 1
print(f"Nombre d'images trouvées: {len(all_images)}")
# Appliquer le préfiltrage de doublons si activé
if not args.no_dedup:
images_avant = len(all_images)
all_images = filtrer_images_uniques(all_images, seuil_hamming=args.seuil, ticket_id=args.ticket_id)
images_apres = len(all_images)
if images_avant > images_apres:
print(f"Préfiltrage des doublons: {images_avant}{images_apres} images ({images_avant - images_apres} doublons supprimés)")
else:
print("Préfiltrage terminé: aucun doublon détecté")
# Trier les images avec l'agent dédié
relevant_images = []
image_sorting_results = {}
# Analyser chaque image - exactement comme dans test_image_sorter
results = []
for img_path in all_images:
img_name = os.path.basename(img_path)
print(f"\nAnalyse de l'image: {img_name}")
if args.debug:
print(f"DEBUG: Chemin complet de l'image: {img_path}")
print(f"DEBUG: L'image existe: {os.path.exists(img_path)}")
print(f"DEBUG: L'image est accessible: {os.access(img_path, os.R_OK)}")
resultat = image_sorter.executer(img_path)
results.append(resultat)
is_relevant = resultat.get("is_relevant", False)
image_sorting_results[img_path] = resultat
# Afficher les résultats comme dans test_image_sorter
print(f"Pertinence: {'OUI' if is_relevant else 'NON'}")
print(f"Raison: {resultat.get('reason', 'Non spécifiée')}")
if is_relevant:
relevant_images.append(img_path)
# Afficher un résumé à la fin comme dans test_image_sorter
pertinentes = sum(1 for r in results if r.get('is_relevant', False))
print(f"\nRésumé: {pertinentes}/{len(results)} images pertinentes")
# Préparer les résultats pour les modes limités
combined_results = {
"ticket_id": args.ticket_id,
"images_total": len(all_images),
"images_relevant": len(relevant_images),
"analyse_images": {}
}
# Ajouter les résultats de tri
for img_path in all_images:
img_name = os.path.basename(img_path)
combined_results["analyse_images"][img_path] = {
"path": img_path,
"sorting": image_sorting_results.get(img_path, {})
}
# Si on est en mode 1 (tri uniquement), s'arrêter ici
if args.mode == 1:
print("\n=== ANALYSE DE TRI TERMINÉE ===")
print(f"Ticket: {args.ticket_id}")
print(f"Images totales: {len(all_images)}")
print(f"Images pertinentes: {len(relevant_images)}")
# Sauvegarder les résultats si demandé
if args.save_results:
save_dir = os.path.join("results", args.ticket_id)
os.makedirs(save_dir, exist_ok=True)
save_file = os.path.join(save_dir, f"tri_images_{args.ticket_id}.json")
with open(save_file, "w", encoding="utf-8") as f:
json.dump(combined_results, f, ensure_ascii=False, indent=2)
print(f"\nRésultats du tri sauvegardés dans: {save_file}")
# Libérer les ressources de l'agent de tri
del image_sorter
del model_tri
return 0
# Libérer les ressources de l'agent de tri
del image_sorter
del model_tri
# ÉTAPE 2: Charger et analyser le ticket avec un agent isolé
# Initialiser l'agent d'analyse de tickets
try:
print("Initialisation du modèle Mistral Large...")
text_model = MistralLarge()
ticket_agent = AgentTicketAnalyser(text_model)
print("Agent d'analyse de tickets initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle d'analyse de tickets: {str(e)}")
return 1
ticket_data = _charger_ticket_data(args)
if not ticket_data:
return 1
# Exécuter l'analyse du ticket - Format comme dans test_agent_ticket_analyser_mistral_large.py
print(f"\nAnalyse du ticket {ticket_data.get('code', args.ticket_id)} en cours...")
ticket_analysis = ticket_agent.executer(ticket_data)
print(f"Analyse du ticket terminée: {len(ticket_analysis)} caractères")
if args.show_analysis:
print("\nRésultat de l'analyse:")
print(ticket_analysis)
# Ajouter l'analyse du ticket aux résultats
combined_results["ticket_data"] = ticket_data
combined_results["ticket_analyse"] = ticket_analysis
# Si on est en mode 2 (tri + ticket), s'arrêter ici
if args.mode == 2:
print("\n=== ANALYSE DE TICKET TERMINÉE ===")
print(f"Ticket: {args.ticket_id}")
print(f"Images totales: {len(all_images)}")
print(f"Images pertinentes: {len(relevant_images)}")
print(f"Analyse du ticket: {len(ticket_analysis)} caractères")
# Sauvegarder les résultats si demandé
if args.save_results:
save_dir = os.path.join("results", args.ticket_id)
os.makedirs(save_dir, exist_ok=True)
save_file = os.path.join(save_dir, f"tri_ticket_{args.ticket_id}.json")
with open(save_file, "w", encoding="utf-8") as f:
json.dump(combined_results, f, ensure_ascii=False, indent=2)
print(f"\nRésultats du tri et de l'analyse de ticket sauvegardés dans: {save_file}")
# Libérer les ressources de l'agent de tickets
del ticket_agent
del text_model
return 0
# Libérer les ressources de l'agent de tickets
del ticket_agent
del text_model
# ÉTAPE 3: Analyser les images pertinentes avec un agent isolé
image_analysis_results = {}
if relevant_images:
# Initialiser l'agent d'analyse d'images
try:
print("Initialisation du modèle Pixtral-Large pour l'analyse d'images...")
model_analyse = PixtralLarge()
# Configuration explicite du modèle
model_analyse.configurer(temperature=0.2, top_p=0.9, max_tokens=3000)
image_analyser = AgentImageAnalyser(model_analyse)
print("Agent d'analyse d'images initialisé avec succès")
except Exception as e:
print(f"ERREUR: Impossible d'initialiser le modèle d'analyse d'images: {str(e)}")
return 1
print("\nAnalyse des images pertinentes avec contexte du ticket...")
for img_path in relevant_images:
img_name = os.path.basename(img_path)
print(f"\nAnalyse de l'image: {img_name}")
try:
analysis_result = image_analyser.executer(img_path, contexte=ticket_analysis)
image_analysis_results[img_path] = analysis_result
if "error" in analysis_result and analysis_result.get("error", False):
print(f" => Erreur: {analysis_result.get('analyse', '')}")
else:
analyse_text = analysis_result.get("analyse", "")
print(f" => Analyse terminée: {len(analyse_text)} caractères")
if args.show_analysis:
print("\nContenu de l'analyse:")
print(analyse_text)
else:
print(f" => Début de l'analyse: {analyse_text[:100]}...")
except Exception as e:
print(f"ERREUR: Impossible d'analyser l'image {img_name}: {str(e)}")
image_analysis_results[img_path] = {"error": True, "analyse": f"ERREUR: {str(e)}"}
else:
print("Aucune image pertinente à analyser")
# Ajout des résultats d'analyse d'images
combined_results["images_analysed"] = len(image_analysis_results)
# Mise à jour des analyses d'images
for img_path in all_images:
if img_path in image_analysis_results:
combined_results["analyse_images"][img_path]["analysis"] = image_analysis_results[img_path]
# Libérer les ressources de l'agent d'analyse d'images
if relevant_images:
del image_analyser
del model_analyse
# Si on est en mode 3 (tri + ticket + images), s'arrêter ici
if args.mode == 3:
print("\n=== ANALYSE D'IMAGES TERMINÉE ===")
print(f"Ticket: {args.ticket_id}")
print(f"Images totales: {len(all_images)}")
print(f"Images pertinentes: {len(relevant_images)}")
print(f"Analyse du ticket: {len(ticket_analysis)} caractères")
print(f"Images analysées: {len(image_analysis_results)}")
# Sauvegarder les résultats si demandé
if args.save_results:
save_dir = os.path.join("results", args.ticket_id)
os.makedirs(save_dir, exist_ok=True)
save_file = os.path.join(save_dir, f"tri_ticket_images_{args.ticket_id}.json")
with open(save_file, "w", encoding="utf-8") as f:
json.dump(combined_results, f, ensure_ascii=False, indent=2)
print(f"\nRésultats complets des analyses sauvegardés dans: {save_file}")
return 0
# ÉTAPE 4: Générer un rapport final si en mode 4 ou avec l'option --generate-report
if args.mode == 4 or args.generate_report:
try:
print("\nInitialisation de l'agent de génération de rapport...")
report_model = MistralLarge()
report_generator = AgentReportGenerator(report_model)
print("Agent de génération de rapport initialisé avec succès")
# Préparer les données pour le rapport
rapport_data = {
"ticket_id": args.ticket_id,
"ticket_data": ticket_data,
"ticket_analyse": ticket_analysis,
"analyse_images": combined_results["analyse_images"]
}
# Générer le rapport
print("\nGénération du rapport final...")
rapport_final = report_generator.executer(rapport_data)
print(f"Rapport généré: {len(rapport_final)} caractères")
# Ajouter le rapport final aux résultats combinés
combined_results["rapport_final"] = rapport_final
if args.show_analysis:
print("\nRapport final:")
print(rapport_final)
else:
print(f"Début du rapport: {rapport_final[:200]}...")
except Exception as e:
print(f"ERREUR: Impossible de générer le rapport: {str(e)}")
# Sauvegarder les résultats si demandé
if args.save_results:
save_dir = os.path.join("results", args.ticket_id)
os.makedirs(save_dir, exist_ok=True)
save_file = os.path.join(save_dir, f"complete_analysis_{args.ticket_id}.json")
with open(save_file, "w", encoding="utf-8") as f:
json.dump(combined_results, f, ensure_ascii=False, indent=2)
print(f"\nRésultats complets sauvegardés dans: {save_file}")
# Afficher un résumé final comme dans le workflow d'analyse d'images
print("\n=== RÉSUMÉ DE L'ANALYSE COMPLÈTE ===")
print(f"Ticket: {args.ticket_id}")
print(f"Images totales: {len(all_images)}")
print(f"Images pertinentes: {len(relevant_images)}")
print(f"Analyse du ticket: {len(ticket_analysis)} caractères")
print(f"Images analysées: {len(image_analysis_results)}")
if args.generate_report or args.mode == 4:
print(f"Rapport final généré: {len(combined_results.get('rapport_final', ''))} caractères")
return 0
def _interactive_menu(args):
"""
Affiche un menu interactif pour choisir le mode d'analyse
"""
print("\n=== MENU D'ANALYSE ===")
print("Ticket à analyser: " + args.ticket_id)
print("1. Tri d'images uniquement")
print("2. Tri d'images + Analyse du ticket")
print("3. Tri d'images + Analyse du ticket + Analyse approfondie des images")
print("4. Analyse complète avec génération de rapport")
print("5. Analyse d'une seule image")
print("0. Quitter")
while True:
try:
choice = int(input("\nVotre choix (0-5): "))
if 0 <= choice <= 5:
break
else:
print("Veuillez entrer un chiffre entre 0 et 5.")
except ValueError:
print("Veuillez entrer un chiffre valide.")
if choice == 0:
print("Au revoir!")
sys.exit(0)
if choice == 5:
# Mode analyse d'une seule image
image_path = input("Chemin de l'image à analyser: ")
if not os.path.exists(image_path):
print(f"ERREUR: L'image spécifiée n'existe pas: {image_path}")
sys.exit(1)
args.image = image_path
else:
# Configurer le mode d'analyse
args.mode = choice
# Paramètres supplémentaires
if input("Afficher les analyses détaillées? (o/n): ").lower().startswith('o'):
args.show_analysis = True
if input("Sauvegarder les résultats? (o/n): ").lower().startswith('o'):
args.save_results = True
if input("Désactiver le préfiltrage des doublons? (o/n): ").lower().startswith('o'):
args.no_dedup = True
if args.no_dedup is False:
try:
seuil = int(input("Seuil de similarité pour détecter les doublons (0-10, défaut=5): "))
if 0 <= seuil <= 10:
args.seuil = seuil
except ValueError:
print("Valeur invalide, utilisation du seuil par défaut: 5")
return args
def _charger_ticket_data(args):
"""
Charge les données du ticket à partir d'un fichier spécifié ou du fichier de rapport par défaut.
"""
ticket_data = {}
if args.ticket_file:
if not os.path.exists(args.ticket_file):
print(f"ERREUR: Le fichier de ticket spécifié n'existe pas: {args.ticket_file}")
return None
try:
with open(args.ticket_file, "r", encoding="utf-8") as f:
ticket_data = json.load(f)
print(f"Fichier de ticket chargé: {args.ticket_file}")
except json.JSONDecodeError:
print(f"ERREUR: Le fichier {args.ticket_file} n'est pas un JSON valide")
return None
else:
# Chercher le fichier de rapport du ticket
rapport_file = get_ticket_report_file(args.ticket_id, args.output_dir)
if not rapport_file:
print(f"ERREUR: Aucun fichier de rapport trouvé pour le ticket {args.ticket_id}")
return None
print(f"Fichier de rapport trouvé: {rapport_file}")
try:
with open(rapport_file, "r", encoding="utf-8") as f:
ticket_data = json.load(f)
print(f"Fichier de rapport chargé: {rapport_file}")
except json.JSONDecodeError:
print(f"ERREUR: Le fichier {rapport_file} n'est pas un JSON valide")
return None
# Vérifier que les données du ticket contiennent un code
if "code" not in ticket_data:
ticket_data["code"] = args.ticket_id
print(f"Code de ticket ajouté: {args.ticket_id}")
return ticket_data
if __name__ == "__main__":
sys.exit(main())