from .base_agent import BaseAgent from typing import Dict, Any, Optional import logging import json import os import sys from datetime import datetime # Ajout du chemin des utilitaires au PATH pour pouvoir les importer sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from utils.ticket_data_loader import TicketDataLoader logger = logging.getLogger("AgentTicketAnalyser") class AgentTicketAnalyser(BaseAgent): """ Agent pour analyser les tickets (JSON ou Markdown) et en extraire les informations importantes. Remplace l'ancien AgentJsonAnalyser avec des fonctionnalités améliorées. """ def __init__(self, llm): super().__init__("AgentTicketAnalyser", llm) # Configuration locale de l'agent self.temperature = 0.1 # Besoin d'analyse très précise self.top_p = 0.8 self.max_tokens = 1500 # Centralisation des objectifs d'analyse self.objectifs_analyse = """ Ta mission principale: 1. Mettre en perspective le NOM DE LA DEMANDE qui contient souvent le problème soulevé par le client 2. Analyser la DESCRIPTION du problème qui ajoute du contexte 3. Établir une chronologie claire des échanges client/support en identifiant précisément: - Les questions posées par le client - Les réponses fournies par le support - Les informations techniques fournies par chaque partie - Fourinir un tableau clair des questions/Réponses support/client sur deux colonnes """ # Centralisation de la structure de réponse self.structure_reponse = """ Structure ta réponse: 1. Analyse du problème initial (nom de la demande + description) 2. Informations techniques essentielles (logiciels, versions, configurations) 3. Chronologie des échanges client/support avec identification claire des questions/réponses """ # Construction du prompt système self.system_prompt = f"""Tu es un expert en analyse de tickets pour le support informatique de BRG-Lab. Ton rôle est d'extraire et d'analyser les informations importantes des tickets. {self.objectifs_analyse} Sois factuel et reste dans une démarche technique. Ton analyse sera utilisée comme contexte pour l'analyse des images pertinentes. {self.structure_reponse}""" # Initialiser le loader de données self.ticket_loader = TicketDataLoader() # Appliquer la configuration au LLM self._appliquer_config_locale() logger.info("AgentTicketAnalyser initialisé") def _appliquer_config_locale(self) -> None: """ Applique la configuration locale au modèle LLM. """ # Appliquer le prompt système if hasattr(self.llm, "prompt_system"): self.llm.prompt_system = self.system_prompt # Appliquer les paramètres if hasattr(self.llm, "configurer"): params = { "temperature": self.temperature, "top_p": self.top_p, "max_tokens": self.max_tokens } self.llm.configurer(**params) def _generer_prompt_analyse(self, ticket_formate: str, source_format: str) -> str: """ Génère le prompt d'analyse standardisé Args: ticket_formate: Texte du ticket formaté pour l'analyse source_format: Format source du ticket (JSON, Markdown, etc.) Returns: Prompt formaté pour l'analyse du ticket """ return f"""Analyse ce ticket de support technique et fournis une synthèse structurée: {ticket_formate} Concentre-toi sur: 1. L'analyse du problème initial décrit dans le nom de la demande et la description 2. L'extraction des informations techniques importantes 3. L'établissement d'une chronologie claire des échanges client/support en identifiant précisément les questions posées et les réponses fournies Ce ticket provient d'un fichier au format {source_format.upper()}. Réponds de manière factuelle, en te basant uniquement sur les informations fournies.""" def executer(self, ticket_data: Dict[str, Any]) -> str: """ Analyse un ticket pour en extraire les informations pertinentes Args: ticket_data: Dictionnaire contenant les données du ticket à analyser ou chemin vers un fichier de ticket (JSON ou Markdown) Returns: Réponse formatée contenant l'analyse du ticket """ # Détecter si ticket_data est un chemin de fichier ou un dictionnaire if isinstance(ticket_data, str) and os.path.exists(ticket_data): try: ticket_data = self.ticket_loader.charger(ticket_data) logger.info(f"Données chargées depuis le fichier: {ticket_data}") except Exception as e: error_message = f"Erreur lors du chargement du fichier: {str(e)}" logger.error(error_message) return f"ERREUR: {error_message}" # Vérifier que les données sont bien un dictionnaire if not isinstance(ticket_data, dict): error_message = "Les données du ticket doivent être un dictionnaire ou un chemin de fichier valide" logger.error(error_message) return f"ERREUR: {error_message}" ticket_code = ticket_data.get('code', 'Inconnu') logger.info(f"Analyse du ticket: {ticket_code}") print(f"AgentTicketAnalyser: Analyse du ticket {ticket_code}") # Récupérer les métadonnées sur la source des données source_format = "inconnu" source_file = "non spécifié" if "metadata" in ticket_data and isinstance(ticket_data["metadata"], dict): source_format = ticket_data["metadata"].get("format", "inconnu") source_file = ticket_data["metadata"].get("source_file", "non spécifié") logger.info(f"Format source: {source_format}, Fichier source: {source_file}") # Préparer le ticket pour l'analyse ticket_formate = self._formater_ticket_pour_analyse(ticket_data) # Créer le prompt pour l'analyse, adapté au format source prompt = self._generer_prompt_analyse(ticket_formate, source_format) try: logger.info("Interrogation du LLM") response = self.llm.interroger(prompt) logger.info(f"Réponse reçue: {len(response)} caractères") print(f" Analyse terminée: {len(response)} caractères") except Exception as e: error_message = f"Erreur lors de l'analyse du ticket: {str(e)}" logger.error(error_message) response = f"ERREUR: {error_message}" print(f" ERREUR: {error_message}") # Enregistrer l'historique avec le prompt complet pour la traçabilité self.ajouter_historique("analyse_ticket", { "ticket_id": ticket_code, "format_source": source_format, "source_file": source_file, "prompt": prompt, "temperature": self.temperature, "top_p": self.top_p, "max_tokens": self.max_tokens, "timestamp": self._get_timestamp() }, response) return response def _formater_ticket_pour_analyse(self, ticket_data: Dict) -> str: """ Formate les données du ticket pour l'analyse LLM, avec une meilleure gestion des différents formats et structures de données. Args: ticket_data: Les données du ticket Returns: Représentation textuelle formatée du ticket """ # Initialiser avec les informations de base ticket_name = ticket_data.get('name', 'Sans titre') ticket_code = ticket_data.get('code', 'Inconnu') info = f"## TICKET {ticket_code}: {ticket_name}\n\n" info += f"## NOM DE LA DEMANDE (PROBLÈME INITIAL)\n{ticket_name}\n\n" # Ajouter la description description = ticket_data.get('description', '') if description: info += f"## DESCRIPTION DU PROBLÈME\n{description}\n\n" # Ajouter les informations du ticket (exclure certains champs spécifiques) champs_a_exclure = ['code', 'name', 'description', 'messages', 'metadata'] info += "## INFORMATIONS TECHNIQUES DU TICKET\n" for key, value in ticket_data.items(): if key not in champs_a_exclure and value: # Formater les valeurs complexes si nécessaire if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False, indent=2) info += f"- {key}: {value}\n" info += "\n" # Ajouter les messages (conversations) avec un formatage amélioré pour distinguer client/support messages = ticket_data.get('messages', []) if messages: info += "## CHRONOLOGIE DES ÉCHANGES CLIENT/SUPPORT\n" for i, msg in enumerate(messages): # Vérifier que le message est bien un dictionnaire if not isinstance(msg, dict): continue sender = msg.get('from', 'Inconnu') date = msg.get('date', 'Date inconnue') content = msg.get('content', '') # Identifier si c'est client ou support sender_type = "CLIENT" if "client" in sender.lower() else "SUPPORT" if "support" in sender.lower() else "AUTRE" # Formater correctement la date si possible try: if date != 'Date inconnue': # Essayer différents formats de date for date_format in ['%Y-%m-%d %H:%M:%S', '%Y-%m-%d', '%d/%m/%Y']: try: date_obj = datetime.strptime(date, date_format) date = date_obj.strftime('%d/%m/%Y %H:%M') break except ValueError: continue except Exception: pass # Garder la date d'origine en cas d'erreur info += f"### Message {i+1} - [{sender_type}] De: {sender} - Date: {date}\n{content}\n\n" # Ajouter les métadonnées techniques si présentes metadata = ticket_data.get('metadata', {}) # Exclure certaines métadonnées internes for key in ['source_file', 'format']: if key in metadata: metadata.pop(key) if metadata: info += "## MÉTADONNÉES TECHNIQUES\n" for key, value in metadata.items(): if isinstance(value, (dict, list)): value = json.dumps(value, ensure_ascii=False, indent=2) info += f"- {key}: {value}\n" info += "\n" return info def analyser_depuis_fichier(self, chemin_fichier: str) -> str: """ Analyse un ticket à partir d'un fichier (JSON ou Markdown) Args: chemin_fichier: Chemin vers le fichier à analyser Returns: Résultat de l'analyse """ try: ticket_data = self.ticket_loader.charger(chemin_fichier) return self.executer(ticket_data) except Exception as e: error_message = f"Erreur lors de l'analyse du fichier {chemin_fichier}: {str(e)}" logger.error(error_message) return f"ERREUR: {error_message}" def _get_timestamp(self) -> str: """Retourne un timestamp au format YYYYMMDD_HHMMSS""" return datetime.now().strftime("%Y%m%d_%H%M%S")