mirror of
https://github.com/Ladebeze66/llm_ticket3.git
synced 2025-12-16 03:57:47 +01:00
358 lines
14 KiB
Python
358 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Utilitaire pour l'analyse de tickets de support utilisant les agents spécialisés.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Dict, List, Any, Optional
|
|
|
|
from agents import AgentFiltreImages, AgentAnalyseImage, AgentQuestionReponse
|
|
from post_process import normaliser_accents, corriger_markdown_accents
|
|
|
|
class TicketAnalyzer:
|
|
"""
|
|
Classe utilitaire pour analyser des tickets de support.
|
|
Coordonne le travail des différents agents spécialisés.
|
|
"""
|
|
|
|
def __init__(self, api_key: Optional[str] = None, llm_params: Optional[Dict[str, Any]] = None):
|
|
"""
|
|
Initialise l'analyseur de tickets.
|
|
|
|
Args:
|
|
api_key: Clé API pour les modèles LLM
|
|
llm_params: Paramètres globaux pour les LLM
|
|
"""
|
|
self.api_key = api_key
|
|
self.llm_params = llm_params or {}
|
|
|
|
# Initialisation des agents
|
|
self.agent_filtre = AgentFiltreImages(api_key=api_key)
|
|
self.agent_analyse = AgentAnalyseImage(api_key=api_key)
|
|
self.agent_qr = AgentQuestionReponse(api_key=api_key)
|
|
|
|
# Appliquer les paramètres globaux
|
|
self._appliquer_parametres_globaux()
|
|
|
|
# Journal d'analyse
|
|
self.entries = []
|
|
|
|
def _appliquer_parametres_globaux(self) -> None:
|
|
"""
|
|
Applique les paramètres globaux à tous les agents.
|
|
"""
|
|
if not self.llm_params:
|
|
return
|
|
|
|
print(f"Application des paramètres globaux LLM: {self.llm_params}")
|
|
self.agent_filtre.appliquer_parametres_globaux(self.llm_params)
|
|
self.agent_analyse.appliquer_parametres_globaux(self.llm_params)
|
|
self.agent_qr.appliquer_parametres_globaux(self.llm_params)
|
|
|
|
def filtrer_images(self, images_paths: List[str]) -> List[str]:
|
|
"""
|
|
Filtre les images pour ne conserver que celles pertinentes.
|
|
|
|
Args:
|
|
images_paths: Liste des chemins d'images à filtrer
|
|
|
|
Returns:
|
|
Liste des chemins d'images pertinentes
|
|
"""
|
|
images_pertinentes = []
|
|
|
|
for image_path in images_paths:
|
|
# Enregistrer le début de l'action
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
entry = {
|
|
"timestamp": timestamp,
|
|
"action": "filter_image",
|
|
"agent": self.agent_filtre.nom,
|
|
"llm": {"model": self.agent_filtre.llm.model},
|
|
"parametres_llm": self.agent_filtre.obtenir_parametres_llm(),
|
|
"image_path": image_path
|
|
}
|
|
|
|
# Exécuter le filtrage d'image
|
|
resultat = self.agent_filtre.executer(image_path)
|
|
|
|
# Ajouter le résultat à l'entrée
|
|
entry["response"] = resultat
|
|
|
|
# Ajouter au journal
|
|
self.entries.append(entry)
|
|
|
|
# Conserver l'image si pertinente
|
|
if resultat.get("pertinente", False):
|
|
images_pertinentes.append(image_path)
|
|
|
|
return images_pertinentes
|
|
|
|
def analyser_images(self, images_paths: List[str], contexte_ticket: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
"""
|
|
Analyse les images pertinentes en détail.
|
|
|
|
Args:
|
|
images_paths: Liste des chemins d'images à analyser
|
|
contexte_ticket: Contexte du ticket pour une analyse plus pertinente
|
|
|
|
Returns:
|
|
Liste des résultats d'analyse
|
|
"""
|
|
resultats_analyses = []
|
|
|
|
for image_path in images_paths:
|
|
# Enregistrer le début de l'action
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
entry = {
|
|
"timestamp": timestamp,
|
|
"action": "analyze_image",
|
|
"agent": self.agent_analyse.nom,
|
|
"llm": {"model": self.agent_analyse.llm.model},
|
|
"parametres_llm": self.agent_analyse.obtenir_parametres_llm(),
|
|
"image_path": image_path
|
|
}
|
|
|
|
# Exécuter l'analyse d'image
|
|
resultat = self.agent_analyse.executer(image_path, contexte_ticket)
|
|
|
|
# Ajouter le résultat à l'entrée
|
|
entry["response"] = resultat.get("content", "")
|
|
if "usage" in resultat:
|
|
entry["tokens"] = resultat["usage"]
|
|
|
|
# Ajouter au journal
|
|
self.entries.append(entry)
|
|
|
|
# Ajouter aux résultats
|
|
if "error" not in resultat:
|
|
resultats_analyses.append({
|
|
"image_path": image_path,
|
|
"analyse": resultat.get("content", ""),
|
|
"usage": resultat.get("usage", {}),
|
|
"parametres_llm": self.agent_analyse.generer_rapport_parametres()
|
|
})
|
|
|
|
return resultats_analyses
|
|
|
|
def extraire_questions_reponses(self, messages: List[Dict[str, Any]], output_path: Optional[str] = None) -> Dict[str, Any]:
|
|
"""
|
|
Extrait les questions et réponses des messages du ticket.
|
|
|
|
Args:
|
|
messages: Liste des messages du ticket
|
|
output_path: Chemin où sauvegarder le tableau Markdown (optionnel)
|
|
|
|
Returns:
|
|
Résultat de l'extraction avec tableau Markdown
|
|
"""
|
|
# Enregistrer le début de l'action
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
entry = {
|
|
"timestamp": timestamp,
|
|
"action": "extract_questions_reponses",
|
|
"agent": self.agent_qr.nom,
|
|
"llm": {"model": self.agent_qr.llm.model},
|
|
"parametres_llm": self.agent_qr.obtenir_parametres_llm()
|
|
}
|
|
|
|
# Exécuter l'extraction des questions/réponses
|
|
resultat = self.agent_qr.executer(messages, output_path)
|
|
|
|
# Ajouter le résultat à l'entrée
|
|
entry["response"] = resultat
|
|
|
|
# Ajouter au journal
|
|
self.entries.append(entry)
|
|
|
|
return resultat
|
|
|
|
def generer_rapport(self, output_path: str) -> Dict[str, str]:
|
|
"""
|
|
Génère un rapport complet d'analyse au format JSON et Markdown.
|
|
|
|
Args:
|
|
output_path: Chemin où sauvegarder le rapport
|
|
|
|
Returns:
|
|
Dictionnaire avec les chemins des fichiers générés
|
|
"""
|
|
# Créer le répertoire de sortie si nécessaire
|
|
os.makedirs(output_path, exist_ok=True)
|
|
|
|
# Chemins des fichiers de sortie
|
|
json_path = os.path.join(output_path, "ticket_analysis.json")
|
|
md_path = os.path.join(output_path, "ticket_analysis.md")
|
|
|
|
# Sauvegarder au format JSON
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump({"entries": self.entries}, f, indent=2, ensure_ascii=False)
|
|
|
|
# Générer le contenu Markdown
|
|
md_content = self._generer_markdown()
|
|
|
|
# Normaliser les accents dans le contenu Markdown avant de l'écrire
|
|
md_content = normaliser_accents(md_content)
|
|
|
|
# Sauvegarder au format Markdown
|
|
with open(md_path, "w", encoding="utf-8") as f:
|
|
f.write(md_content)
|
|
|
|
return {
|
|
"json": json_path,
|
|
"markdown": md_path
|
|
}
|
|
|
|
def _generer_markdown(self) -> str:
|
|
"""
|
|
Génère le contenu Markdown à partir des entrées du journal.
|
|
|
|
Returns:
|
|
Contenu Markdown formaté
|
|
"""
|
|
contenu = ["# Analyse de ticket de support\n"]
|
|
|
|
# Statistiques
|
|
stats = self._calculer_statistiques()
|
|
contenu.append("## Statistiques\n")
|
|
contenu.append(f"- Images analysées: {stats['images_total']} ({stats['images_pertinentes']} pertinentes)")
|
|
contenu.append(f"- Questions identifiées: {stats['questions']}")
|
|
contenu.append(f"- Réponses identifiées: {stats['reponses']}")
|
|
contenu.append("\n")
|
|
|
|
# Paramètres LLM globaux
|
|
if self.llm_params:
|
|
contenu.append("## Paramètres LLM globaux\n")
|
|
for param, valeur in self.llm_params.items():
|
|
contenu.append(f"- **{param}**: {valeur}")
|
|
contenu.append("\n")
|
|
|
|
# Paramètres LLM utilisés
|
|
contenu.append("## Paramètres LLM par agent\n")
|
|
|
|
# Filtre d'images
|
|
agent_filtre_params = self.agent_filtre.obtenir_parametres_llm()
|
|
contenu.append("### Agent de filtrage d'images\n")
|
|
contenu.append(f"- **Type de LLM**: {self.agent_filtre.llm.__class__.__name__}")
|
|
contenu.append(f"- **Modèle**: {agent_filtre_params.get('model', 'Non spécifié')}")
|
|
contenu.append(f"- **Température**: {agent_filtre_params.get('temperature', 'Non spécifiée')}")
|
|
contenu.append(f"- **Tokens max**: {agent_filtre_params.get('max_tokens', 'Non spécifié')}")
|
|
|
|
# Analyse d'images
|
|
agent_analyse_params = self.agent_analyse.obtenir_parametres_llm()
|
|
contenu.append("\n### Agent d'analyse d'images\n")
|
|
contenu.append(f"- **Type de LLM**: {self.agent_analyse.llm.__class__.__name__}")
|
|
contenu.append(f"- **Modèle**: {agent_analyse_params.get('model', 'Non spécifié')}")
|
|
contenu.append(f"- **Température**: {agent_analyse_params.get('temperature', 'Non spécifiée')}")
|
|
contenu.append(f"- **Tokens max**: {agent_analyse_params.get('max_tokens', 'Non spécifié')}")
|
|
|
|
# Questions-réponses
|
|
agent_qr_params = self.agent_qr.obtenir_parametres_llm()
|
|
contenu.append("\n### Agent d'extraction questions-réponses\n")
|
|
contenu.append(f"- **Type de LLM**: {self.agent_qr.llm.__class__.__name__}")
|
|
contenu.append(f"- **Modèle**: {agent_qr_params.get('model', 'Non spécifié')}")
|
|
contenu.append(f"- **Température**: {agent_qr_params.get('temperature', 'Non spécifiée')}")
|
|
contenu.append(f"- **Tokens max**: {agent_qr_params.get('max_tokens', 'Non spécifié')}")
|
|
contenu.append("\n")
|
|
|
|
# Actions chronologiques
|
|
contenu.append("## Journal d'actions\n")
|
|
|
|
for entry in self.entries:
|
|
timestamp = entry.get("timestamp", "")
|
|
action = entry.get("action", "")
|
|
agent = entry.get("agent", "")
|
|
model = entry.get("llm", {}).get("model", "")
|
|
|
|
# En-tête de l'action
|
|
contenu.append(f"### {timestamp} - {agent} ({model})")
|
|
contenu.append(f"**Action**: {action}")
|
|
|
|
# Détails spécifiques selon le type d'action
|
|
if action == "filter_image":
|
|
image_path = entry.get("image_path", "")
|
|
response = entry.get("response", {})
|
|
pertinente = response.get("pertinente", False)
|
|
type_image = response.get("type_image", "inconnue")
|
|
description = response.get("description", "")
|
|
|
|
contenu.append(f"**Image**: {os.path.basename(image_path)}")
|
|
contenu.append(f"**Résultat**: {'Pertinente' if pertinente else 'Non pertinente'}")
|
|
contenu.append(f"**Type**: {type_image}")
|
|
contenu.append(f"**Description**: {description}")
|
|
|
|
# Paramètres LLM utilisés
|
|
params_llm = response.get("parametres_llm", {})
|
|
if params_llm:
|
|
contenu.append("\n**Paramètres LLM utilisés:**")
|
|
temp = params_llm.get("parametres", {}).get("temperature", "N/A")
|
|
contenu.append(f"- Température: {temp}")
|
|
|
|
elif action == "analyze_image":
|
|
image_path = entry.get("image_path", "")
|
|
response = entry.get("response", "")
|
|
|
|
contenu.append(f"**Image analysée**: {os.path.basename(image_path)}")
|
|
contenu.append("\n**Analyse**:")
|
|
contenu.append(f"```\n{response}\n```")
|
|
|
|
elif action == "extract_questions_reponses":
|
|
response = entry.get("response", {})
|
|
tableau_md = response.get("tableau_md", "")
|
|
|
|
contenu.append(f"**Questions**: {response.get('nb_questions', 0)}")
|
|
contenu.append(f"**Réponses**: {response.get('nb_reponses', 0)}")
|
|
contenu.append("\n")
|
|
contenu.append(tableau_md)
|
|
|
|
# Paramètres LLM spécifiques à cette action
|
|
params_llm = entry.get("parametres_llm", {})
|
|
if params_llm and action != "filter_image": # Pour éviter la duplication avec le filtre d'image
|
|
contenu.append("\n**Paramètres LLM utilisés:**")
|
|
for key, value in params_llm.items():
|
|
if key != "system_prompt": # Éviter d'afficher le prompt système ici
|
|
contenu.append(f"- **{key}**: {value}")
|
|
|
|
# Tokens utilisés
|
|
if "tokens" in entry:
|
|
tokens = entry["tokens"]
|
|
contenu.append("\n**Tokens utilisés**:")
|
|
contenu.append(f"- Prompt: {tokens.get('prompt_tokens', 0)}")
|
|
contenu.append(f"- Completion: {tokens.get('completion_tokens', 0)}")
|
|
contenu.append(f"- Total: {tokens.get('total_tokens', 0)}")
|
|
|
|
contenu.append("\n---\n")
|
|
|
|
return "\n".join(contenu)
|
|
|
|
def _calculer_statistiques(self) -> Dict[str, int]:
|
|
"""
|
|
Calcule les statistiques à partir des entrées du journal.
|
|
|
|
Returns:
|
|
Dictionnaire de statistiques
|
|
"""
|
|
stats = {
|
|
"images_total": 0,
|
|
"images_pertinentes": 0,
|
|
"questions": 0,
|
|
"reponses": 0
|
|
}
|
|
|
|
for entry in self.entries:
|
|
action = entry.get("action", "")
|
|
|
|
if action == "filter_image":
|
|
stats["images_total"] += 1
|
|
if entry.get("response", {}).get("pertinente", False):
|
|
stats["images_pertinentes"] += 1
|
|
|
|
elif action == "extract_questions_reponses":
|
|
stats["questions"] = entry.get("response", {}).get("nb_questions", 0)
|
|
stats["reponses"] = entry.get("response", {}).get("nb_reponses", 0)
|
|
|
|
return stats |