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 loaders.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 = 2500 # 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 - IMPORTANT: Le NOM DE LA DEMANDE peut contenir une ou plusieurs questions à identifier clairement 2. Analyser la DESCRIPTION du problème qui ajoute du contexte - IMPORTANT: La DESCRIPTION peut également contenir une ou plusieurs questions à identifier clairement 3. Établir une chronologie claire des échanges client/support en identifiant précisément: - Il peut y avoir une discussion dans le même message - Ne tient pas compte des bas de page, des adresses ou des coordonnées des personnes si ce n'est pas pertinent - Les questions posées par le client (y compris celles dans le NOM ou la DESCRIPTION) - Les réponses fournies par le support - Les informations techniques fournies par chaque partie - Si une référence à une norme ou autre élément technique est faite, note-la clairement car c'est un élément essentiel - Assure-toi d'identifier clairement les intervenants (client/support) pour maintenir un fil de discussion clair """ # Centralisation de la structure de réponse self.structure_reponse = """ Structure ta réponse: 1. Résumé du contexte basé sur le nom de la demande et la description (si présente) 2. Informations techniques essentielles (logiciels, versions, configurations) 3. Chronologie des échanges client/support avec identification claire des intervenants 4. IMPORTANT: Ne génère PAS de tableau question/réponse """ # Construction du prompt système self.system_prompt = f"""Tu es un expert en analyse de tickets pour le support informatique de BRG-Lab pour la société CBAO. 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 - IMPORTANT: Identifie toute question explicite ou implicite dans le NOM et la DESCRIPTION - Ces questions sont cruciales car elles définissent souvent l'objectif principal du ticket 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 par le client (y compris celles dans le nom et la description) - Les réponses fournies par le support - Les zones d'incertitude ou questions sans réponse 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")