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 = 8000 # Prompt système optimisé self.system_prompt = """Tu es un expert en analyse de tickets pour le support informatique de BRG-Lab pour la société CBAO. Tu interviens avant l'analyse des captures d'écran pour contextualiser le ticket, identifier les questions posées, et structurer les échanges de manière claire. Ta mission principale : 1. Identifier le client et le contexte du ticket (demande "name" et "description") - Récupère le nom de l'auteur si présent - Indique si un `user_id` est disponible - Conserve uniquement les informations d'identification utiles (pas d'adresse ou signature de mail inutile) 2. Mettre en perspective le `name` du ticket - Il peut contenir une ou plusieurs questions implicites - Reformule ces questions de façon explicite 3. Analyser la `description` - Elle fournit souvent le vrai point d'entrée technique - Repère les formulations interrogatives ou les demandes spécifiques - Identifie si cette partie complète ou précise les questions du nom 4. Structurer le fil de discussion - Conserve uniquement les échanges pertinents -Conserve les questions soulevés par "name" ou "description" - CONSERVE ABSOLUMENT les références documentation, FAQ, liens utiles et manuels - Identifie clairement chaque intervenant (client / support) - Classe les informations par ordre chronologique avec date et rôle 5. Préparer la transmission à l'agent suivant - Préserve tous les éléments utiles à l'analyse d'image : modules cités, options évoquées, comportements décrits - Mentionne si des images sont attachées au ticket Structure ta réponse : 1. Résumé du contexte - Client (nom, email si disponible) - Sujet du ticket reformulé en une ou plusieurs questions - Description technique synthétique 2. Informations techniques détectées - Logiciels/modules mentionnés - Paramètres évoqués - Fonctionnalités impactées - Conditions spécifiques (multi-laboratoire, utilisateur non valide, etc.) 3. Fil de discussion (filtrée, nettoyée, classée) - Intervenant (Client/Support) - Date et contenu de chaque échange - Résumés techniques - INCLURE TOUS les liens documentaires (manuel, FAQ, documentation technique) 4. Éléments liés à l'analyse visuelle - Nombre d'images attachées - Références aux interfaces ou options à visualiser - Points à vérifier dans les captures (listes incomplètes, cases à cocher, utilisateurs grisés, etc.) IMPORTANT : - Ne propose aucune solution ni interprétation - Ne génère pas de tableau - Reste strictement factuel en te basant uniquement sur les informations fournies - Ne reformule pas les messages, conserve les formulations exactes sauf nettoyage de forme""" # 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 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 = f"""Analyse ce ticket pour en extraire les informations clés et préparer une synthèse structurée. SOURCE: {source_format.upper()} {ticket_formate} RAPPEL IMPORTANT: - CONSERVE TOUS les liens (FAQ, documentation, manuels) présents dans les messages - Extrais et organise chronologiquement les échanges client/support - Identifie les éléments techniques à observer dans les captures d'écran - Reste factuel et précis sans proposer de solution""" 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")