mirror of
https://github.com/Ladebeze66/llm_ticket3.git
synced 2025-12-15 20:06:51 +01:00
368 lines
17 KiB
Python
368 lines
17 KiB
Python
import json
|
|
import os
|
|
from .base_agent import BaseAgent
|
|
from datetime import datetime
|
|
from typing import Dict, Any, Tuple, Optional, List
|
|
import logging
|
|
import traceback
|
|
import re
|
|
import sys
|
|
from .utils.report_utils import extraire_et_traiter_json
|
|
from .utils.report_formatter import extraire_sections_texte, generer_rapport_markdown, construire_rapport_json
|
|
from .utils.agent_info_collector import collecter_info_agents, collecter_prompts_agents
|
|
|
|
logger = logging.getLogger("AgentReportGenerator")
|
|
|
|
class AgentReportGenerator(BaseAgent):
|
|
"""
|
|
Agent pour générer un rapport synthétique à partir des analyses de ticket et d'images.
|
|
"""
|
|
def __init__(self, llm):
|
|
super().__init__("AgentReportGenerator", llm)
|
|
|
|
# Configuration locale de l'agent
|
|
self.temperature = 0.2
|
|
self.top_p = 0.9
|
|
self.max_tokens = 10000
|
|
|
|
# Prompt système principal
|
|
self.system_prompt = """Tu es un expert en génération de rapports techniques pour BRG-Lab pour la société CBAO.
|
|
Ta mission est de synthétiser les analyses (ticket et images) en un rapport structuré.
|
|
|
|
EXIGENCE ABSOLUE - Ton rapport DOIT inclure dans l'ordre:
|
|
1. Un résumé du problème initial (nom de la demande + description)
|
|
2. Une analyse détaillée des images pertinentes en lien avec le problème
|
|
3. Une synthèse globale des analyses d'images
|
|
4. Une reconstitution du fil de discussion client/support
|
|
5. Un tableau JSON de chronologie des échanges avec cette structure:
|
|
```json
|
|
{
|
|
"chronologie_echanges": [
|
|
{"date": "date exacte", "emetteur": "CLIENT ou SUPPORT", "type": "Question ou Réponse", "contenu": "contenu synthétisé"}
|
|
]
|
|
}
|
|
```
|
|
6. Un diagnostic technique des causes probables
|
|
|
|
MÉTHODE D'ANALYSE (ÉTAPES OBLIGATOIRES):
|
|
1. ANALYSE TOUTES les images AVANT de créer le tableau des échanges
|
|
2. Concentre-toi sur les éléments mis en évidence (encadrés/surlignés) dans chaque image
|
|
3. Réalise une SYNTHÈSE TRANSVERSALE en expliquant comment les images se complètent
|
|
4. Remets les images en ordre chronologique selon le fil de discussion
|
|
5. CONSERVE TOUS les liens documentaires, FAQ et références techniques
|
|
6. Ajoute une entrée "Complément visuel" dans le tableau des échanges"""
|
|
|
|
# Version du prompt pour la traçabilité
|
|
self.prompt_version = "v3.2"
|
|
|
|
# Appliquer la configuration au LLM
|
|
self._appliquer_config_locale()
|
|
|
|
logger.info("AgentReportGenerator 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)
|
|
logger.info(f"Configuration appliquée au modèle: {str(params)}")
|
|
|
|
def _formater_prompt_pour_rapport(self, ticket_analyse: str, images_analyses: List[Dict]) -> str:
|
|
"""
|
|
Formate le prompt pour la génération du rapport
|
|
"""
|
|
num_images = len(images_analyses)
|
|
logger.info(f"Formatage du prompt avec {num_images} analyses d'images")
|
|
|
|
# Construire la section d'analyse du ticket
|
|
prompt = f"""Génère un rapport technique complet, en te basant sur les analyses suivantes.
|
|
|
|
## ANALYSE DU TICKET
|
|
{ticket_analyse}
|
|
"""
|
|
|
|
# Ajouter la section d'analyse des images si présente
|
|
if num_images > 0:
|
|
prompt += f"\n## ANALYSES DES IMAGES ({num_images} images)\n"
|
|
for i, img_analyse in enumerate(images_analyses, 1):
|
|
image_name = img_analyse.get("image_name", f"Image {i}")
|
|
analyse = img_analyse.get("analyse", "Analyse non disponible")
|
|
prompt += f"\n### IMAGE {i}: {image_name}\n{analyse}\n"
|
|
else:
|
|
prompt += "\n## ANALYSES DES IMAGES\nAucune image n'a été fournie pour ce ticket.\n"
|
|
|
|
# Instructions pour le rapport
|
|
prompt += """
|
|
## INSTRUCTIONS POUR LE RAPPORT
|
|
|
|
STRUCTURE OBLIGATOIRE ET ORDRE À SUIVRE:
|
|
1. Titre principal (# Rapport d'analyse: Nom du ticket)
|
|
2. Résumé du problème (## Résumé du problème)
|
|
3. Analyse des images (## Analyse des images) - CRUCIAL: FAIRE CETTE SECTION AVANT LE TABLEAU
|
|
4. Synthèse globale des analyses d'images (## 3.1 Synthèse globale des analyses d'images)
|
|
5. Fil de discussion (## Fil de discussion)
|
|
6. Tableau questions/réponses (## Tableau questions/réponses)
|
|
7. Diagnostic technique (## Diagnostic technique)
|
|
|
|
MÉTHODE POUR ANALYSER LES IMAGES:
|
|
- Pour chaque image, concentre-toi prioritairement sur:
|
|
* Les éléments mis en évidence (zones encadrées, surlignées)
|
|
* La relation avec le problème décrit
|
|
* Le lien avec le fil de discussion
|
|
|
|
SYNTHÈSE GLOBALE DES IMAGES (SECTION CRUCIALE):
|
|
- Titre à utiliser OBLIGATOIREMENT: ## 3.1 Synthèse globale des analyses d'images
|
|
- Premier sous-titre à utiliser OBLIGATOIREMENT: _Analyse transversale des captures d'écran_
|
|
- Structure cette section avec les sous-parties:
|
|
* Points communs et complémentaires entre les images
|
|
* Corrélation entre les éléments et le problème global
|
|
* Confirmation visuelle des informations du support
|
|
- Montre comment les images se complètent pour illustrer le processus complet
|
|
- Cette synthèse transversale servira de base pour le "Complément visuel"
|
|
|
|
POUR LE TABLEAU QUESTIONS/RÉPONSES:
|
|
- Tu DOIS créer et inclure un tableau JSON structuré comme ceci:
|
|
```json
|
|
{
|
|
"chronologie_echanges": [
|
|
{"date": "date demande", "emetteur": "CLIENT", "type": "Question", "contenu": "Texte exact du problème initial extrait du ticket"},
|
|
{"date": "date exacte", "emetteur": "SUPPORT", "type": "Réponse", "contenu": "réponse avec TOUS les liens documentaires"},
|
|
{"date": "date analyse", "emetteur": "SUPPORT", "type": "Complément visuel", "contenu": "synthèse unifiée de TOUTES les images"}
|
|
]
|
|
}
|
|
```
|
|
|
|
DIRECTIVES ESSENTIELLES:
|
|
- COMMENCE ABSOLUMENT par une entrée CLIENT avec les questions du NOM et de la DESCRIPTION du ticket
|
|
- Si le premier message chronologique est une réponse du SUPPORT qui cite la question, extrais la question citée pour l'ajouter comme première entrée CLIENT
|
|
- CONSERVE ABSOLUMENT TOUS les liens vers la documentation, FAQ, manuels et références techniques
|
|
- Ajoute UNE SEULE entrée "Complément visuel" qui synthétise l'apport global des images
|
|
- Cette entrée doit montrer comment les images confirment/illustrent le processus complet
|
|
- Formulation recommandée: "L'analyse des captures d'écran confirme visuellement le processus: (1)..., (2)..., (3)... Ces interfaces complémentaires illustrent..."
|
|
- Évite de traiter les images séparément dans le tableau; présente une vision unifiée
|
|
- Identifie clairement chaque intervenant (CLIENT ou SUPPORT)
|
|
"""
|
|
|
|
return prompt
|
|
|
|
def executer(self, rapport_data: Dict, rapport_dir: str) -> Tuple[Optional[str], Optional[str]]:
|
|
"""
|
|
Génère un rapport à partir des analyses effectuées
|
|
"""
|
|
try:
|
|
# 1. PRÉPARATION
|
|
ticket_id = self._extraire_ticket_id(rapport_data, rapport_dir)
|
|
logger.info(f"Génération du rapport pour le ticket: {ticket_id}")
|
|
print(f"AgentReportGenerator: Génération du rapport pour {ticket_id}")
|
|
|
|
# Créer le répertoire de sortie si nécessaire
|
|
os.makedirs(rapport_dir, exist_ok=True)
|
|
|
|
# 2. EXTRACTION DES DONNÉES
|
|
ticket_analyse = self._extraire_analyse_ticket(rapport_data)
|
|
images_analyses = self._extraire_analyses_images(rapport_data)
|
|
|
|
# 3. COLLECTE DES INFORMATIONS SUR LES AGENTS (via le nouveau module)
|
|
agent_info = {
|
|
"model": getattr(self.llm, "modele", str(type(self.llm))),
|
|
"temperature": self.temperature,
|
|
"top_p": self.top_p,
|
|
"max_tokens": self.max_tokens,
|
|
"prompt_version": self.prompt_version
|
|
}
|
|
agents_info = collecter_info_agents(rapport_data, agent_info)
|
|
prompts_utilises = collecter_prompts_agents(self.system_prompt)
|
|
|
|
# 4. GÉNÉRATION DU RAPPORT
|
|
prompt = self._formater_prompt_pour_rapport(ticket_analyse, images_analyses)
|
|
|
|
logger.info("Génération du rapport avec le LLM")
|
|
print(f" Génération du rapport avec le LLM...")
|
|
|
|
# Mesurer le temps d'exécution
|
|
start_time = datetime.now()
|
|
rapport_genere = self.llm.interroger(prompt)
|
|
generation_time = (datetime.now() - start_time).total_seconds()
|
|
|
|
logger.info(f"Rapport généré: {len(rapport_genere)} caractères")
|
|
print(f" Rapport généré: {len(rapport_genere)} caractères")
|
|
|
|
# 5. EXTRACTION DES DONNÉES DU RAPPORT
|
|
# Utiliser l'utilitaire de report_utils.py pour extraire les données JSON
|
|
rapport_traite, echanges_json, _ = extraire_et_traiter_json(rapport_genere)
|
|
|
|
# Vérifier que echanges_json n'est pas None pour éviter l'erreur de type
|
|
if echanges_json is None:
|
|
echanges_json = {"chronologie_echanges": []}
|
|
logger.warning("Aucun échange JSON extrait du rapport, création d'une structure vide")
|
|
|
|
# Extraire les sections textuelles (résumé, diagnostic)
|
|
resume, analyse_images, diagnostic = extraire_sections_texte(rapport_genere)
|
|
|
|
# 6. CRÉATION DU RAPPORT JSON
|
|
# Préparer les métadonnées de l'agent
|
|
agent_metadata = {
|
|
"model": getattr(self.llm, "modele", str(type(self.llm))),
|
|
"model_version": getattr(self.llm, "version", "non spécifiée"),
|
|
"temperature": self.temperature,
|
|
"top_p": self.top_p,
|
|
"max_tokens": self.max_tokens,
|
|
"generation_time": generation_time,
|
|
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"agents": agents_info
|
|
}
|
|
|
|
# Construire le rapport JSON
|
|
rapport_json = construire_rapport_json(
|
|
rapport_genere=rapport_genere,
|
|
rapport_data=rapport_data,
|
|
ticket_id=ticket_id,
|
|
ticket_analyse=ticket_analyse,
|
|
images_analyses=images_analyses,
|
|
generation_time=generation_time,
|
|
resume=resume,
|
|
analyse_images=analyse_images,
|
|
diagnostic=diagnostic,
|
|
echanges_json=echanges_json,
|
|
agent_metadata=agent_metadata,
|
|
prompts_utilises=prompts_utilises
|
|
)
|
|
|
|
# 7. SAUVEGARDE DU RAPPORT JSON
|
|
json_path = os.path.join(rapport_dir, f"{ticket_id}_rapport_final.json")
|
|
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(rapport_json, f, ensure_ascii=False, indent=2)
|
|
|
|
logger.info(f"Rapport JSON sauvegardé: {json_path}")
|
|
print(f" Rapport JSON sauvegardé: {json_path}")
|
|
|
|
# 8. GÉNÉRATION DU RAPPORT MARKDOWN
|
|
md_path = generer_rapport_markdown(json_path)
|
|
|
|
if md_path:
|
|
logger.info(f"Rapport Markdown généré: {md_path}")
|
|
print(f" Rapport Markdown généré: {md_path}")
|
|
else:
|
|
logger.error("Échec de la génération du rapport Markdown")
|
|
print(f" ERREUR: Échec de la génération du rapport Markdown")
|
|
|
|
return json_path, md_path
|
|
|
|
except Exception as e:
|
|
error_message = f"Erreur lors de la génération du rapport: {str(e)}"
|
|
logger.error(error_message)
|
|
logger.error(traceback.format_exc())
|
|
print(f" ERREUR: {error_message}")
|
|
return None, None
|
|
|
|
def _extraire_ticket_id(self, rapport_data: Dict, rapport_dir: str) -> str:
|
|
"""Extrait l'ID du ticket des données ou du chemin"""
|
|
# Essayer d'extraire depuis les données du rapport
|
|
ticket_id = rapport_data.get("ticket_id", "")
|
|
|
|
# Si pas d'ID direct, essayer depuis les données du ticket
|
|
if not ticket_id and "ticket_data" in rapport_data and isinstance(rapport_data["ticket_data"], dict):
|
|
ticket_id = rapport_data["ticket_data"].get("code", "")
|
|
|
|
# En dernier recours, extraire depuis le chemin
|
|
if not ticket_id:
|
|
# Essayer d'extraire un ID de ticket (format Txxxx) du chemin
|
|
match = re.search(r'T\d+', rapport_dir)
|
|
if match:
|
|
ticket_id = match.group(0)
|
|
else:
|
|
# Sinon, utiliser le dernier segment du chemin
|
|
ticket_id = os.path.basename(rapport_dir)
|
|
|
|
return ticket_id
|
|
|
|
def _extraire_analyse_ticket(self, rapport_data: Dict) -> str:
|
|
"""Extrait l'analyse du ticket des données"""
|
|
# Essayer les différentes clés possibles
|
|
for key in ["ticket_analyse", "analyse_json", "analyse_ticket"]:
|
|
if key in rapport_data and rapport_data[key]:
|
|
logger.info(f"Utilisation de {key}")
|
|
return rapport_data[key]
|
|
|
|
# Créer une analyse par défaut si aucune n'est disponible
|
|
logger.warning("Aucune analyse de ticket disponible, création d'un message par défaut")
|
|
ticket_data = rapport_data.get("ticket_data", {})
|
|
ticket_name = ticket_data.get("name", "Sans titre")
|
|
ticket_desc = ticket_data.get("description", "Pas de description disponible")
|
|
return f"Analyse par défaut du ticket:\nNom: {ticket_name}\nDescription: {ticket_desc}\n(Aucune analyse détaillée n'a été fournie)"
|
|
|
|
def _extraire_analyses_images(self, rapport_data: Dict) -> List[Dict]:
|
|
"""
|
|
Extrait et formate les analyses d'images pertinentes
|
|
"""
|
|
images_analyses = []
|
|
analyse_images_data = rapport_data.get("analyse_images", {})
|
|
|
|
# Parcourir toutes les images
|
|
for image_path, analyse_data in analyse_images_data.items():
|
|
# Vérifier si l'image est pertinente
|
|
is_relevant = False
|
|
if "sorting" in analyse_data and isinstance(analyse_data["sorting"], dict):
|
|
is_relevant = analyse_data["sorting"].get("is_relevant", False)
|
|
|
|
# Si l'image est pertinente, extraire son analyse
|
|
if is_relevant:
|
|
image_name = os.path.basename(image_path)
|
|
analyse = self._extraire_analyse_image(analyse_data)
|
|
|
|
if analyse:
|
|
images_analyses.append({
|
|
"image_name": image_name,
|
|
"image_path": image_path,
|
|
"analyse": analyse,
|
|
"sorting_info": analyse_data.get("sorting", {}),
|
|
"metadata": analyse_data.get("analysis", {}).get("metadata", {})
|
|
})
|
|
logger.info(f"Analyse de l'image {image_name} ajoutée")
|
|
|
|
return images_analyses
|
|
|
|
def _extraire_analyse_image(self, analyse_data: Dict) -> Optional[str]:
|
|
"""
|
|
Extrait l'analyse d'une image depuis les données
|
|
"""
|
|
# Si pas de données d'analyse, retourner None
|
|
if not "analysis" in analyse_data or not analyse_data["analysis"]:
|
|
if "sorting" in analyse_data and isinstance(analyse_data["sorting"], dict):
|
|
reason = analyse_data["sorting"].get("reason", "Non spécifiée")
|
|
return f"Image marquée comme pertinente. Raison: {reason}"
|
|
return None
|
|
|
|
# Extraire l'analyse selon le format des données
|
|
analysis = analyse_data["analysis"]
|
|
|
|
# Structure type 1: {"analyse": "texte"}
|
|
if isinstance(analysis, dict) and "analyse" in analysis:
|
|
return analysis["analyse"]
|
|
|
|
# Structure type 2: {"error": false, ...} - contient d'autres données utiles
|
|
if isinstance(analysis, dict) and "error" in analysis and not analysis.get("error", True):
|
|
return str(analysis)
|
|
|
|
# Structure type 3: texte d'analyse direct
|
|
if isinstance(analysis, str):
|
|
return analysis
|
|
|
|
# Structure type 4: autre format de dictionnaire - convertir en JSON
|
|
if isinstance(analysis, dict):
|
|
return json.dumps(analysis, ensure_ascii=False, indent=2)
|
|
|
|
# Aucun format reconnu
|
|
return None |