import json import os from .base_agent import BaseAgent from datetime import datetime from typing import Dict, Any, Tuple import logging logger = logging.getLogger("AgentReportGenerator") class AgentReportGenerator(BaseAgent): """ Agent pour générer un rapport à partir des informations collectées. """ def __init__(self, llm): super().__init__("AgentReportGenerator", llm, "report_generator") logger.info("AgentReportGenerator initialisé") def executer(self, rapport_data: Dict, filename: str) -> Tuple[str, str]: logger.info(f"Génération du rapport pour le fichier: {filename}") timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Ajouter les métadonnées des LLM utilisés if "metadata" not in rapport_data: rapport_data["metadata"] = {} rapport_data["metadata"]["report_generator"] = { "model": getattr(self.llm, "modele", str(type(self.llm))), "configuration": self.config.to_dict() } # Créer les dossiers si nécessaire reports_dir = "../reports" json_dir = os.path.join(reports_dir, "json_reports") md_dir = os.path.join(reports_dir, "markdown_reports") os.makedirs(json_dir, exist_ok=True) os.makedirs(md_dir, exist_ok=True) # Sauvegarde JSON json_path = f"{json_dir}/{filename}_{timestamp}.json" with open(json_path, "w", encoding="utf-8") as f_json: json.dump(rapport_data, f_json, ensure_ascii=False, indent=4) logger.info(f"Rapport JSON sauvegardé à: {json_path}") # Sauvegarde Markdown md_path = f"{md_dir}/{filename}_{timestamp}.md" with open(md_path, "w", encoding="utf-8") as f_md: # En-tête du rapport ticket_id = rapport_data.get("ticket_id", filename) f_md.write(f"# Rapport d'Analyse du Ticket {ticket_id}\n\n") f_md.write(f"*Généré le: {datetime.now().strftime('%d/%m/%Y à %H:%M:%S')}*\n\n") # Résumé metadata = rapport_data.get("metadata", {}) etapes = metadata.get("etapes", []) f_md.write("## Résumé\n\n") f_md.write(f"- **ID Ticket**: {ticket_id}\n") f_md.write(f"- **Date d'analyse**: {metadata.get('timestamp_debut', timestamp)}\n") f_md.write(f"- **Nombre d'étapes**: {len(etapes)}\n\n") # Vue d'ensemble des agents et modèles utilisés f_md.write("## Modèles et Paramètres Utilisés\n\n") f_md.write("### Vue d'ensemble\n\n") f_md.write("| Agent | Modèle | Température | Top-P | Max Tokens |\n") f_md.write("|-------|--------|-------------|-------|------------|\n") for agent_name in ["json_agent", "image_sorter", "image_analyser", "report_generator"]: agent_info = metadata.get(agent_name, {}) if agent_info.get("status") == "non configuré": continue model = agent_info.get("model", "N/A") config = agent_info.get("configuration", {}).get("config", {}) temp = config.get("temperature", "N/A") top_p = config.get("top_p", "N/A") max_tokens = config.get("max_tokens", "N/A") f_md.write(f"| {agent_name} | {model} | {temp} | {top_p} | {max_tokens} |\n") f_md.write("\n") # Détails des paramètres par agent f_md.write("### Détails des Paramètres\n\n") f_md.write("```json\n") agents_config = {} for agent_name in ["json_agent", "image_sorter", "image_analyser", "report_generator"]: if agent_name in metadata and "configuration" in metadata[agent_name]: agents_config[agent_name] = metadata[agent_name]["configuration"] f_md.write(json.dumps(agents_config, indent=2)) f_md.write("\n```\n\n") # Détails des prompts système f_md.write("### Prompts Système\n\n") for agent_name in ["json_agent", "image_sorter", "image_analyser", "report_generator"]: agent_info = metadata.get(agent_name, {}) if agent_info.get("status") == "non configuré": continue config = agent_info.get("configuration", {}).get("config", {}) prompt = config.get("system_prompt", "") if prompt: f_md.write(f"**{agent_name}**:\n") f_md.write("```\n") f_md.write(prompt) f_md.write("\n```\n\n") # Étapes d'analyse f_md.write("## Étapes d'Analyse\n\n") for i, etape in enumerate(etapes, 1): agent = etape.get("agent", "") action = etape.get("action", "") timestamp = etape.get("timestamp", "") image = etape.get("image", "") title = f"### {i}. {agent.replace('_', ' ').title()}" if image: title += f" - Image: {image}" f_md.write(f"{title}\n\n") f_md.write(f"- **Action**: {action}\n") f_md.write(f"- **Timestamp**: {timestamp}\n") # Métadonnées du modèle pour cette étape etape_metadata = etape.get("metadata", {}) model = etape_metadata.get("model", "") config = etape_metadata.get("configuration", {}).get("config", {}) duree = etape_metadata.get("duree_traitement", "") f_md.write(f"- **Modèle**: {model}\n") if duree: f_md.write(f"- **Durée de traitement**: {duree}\n") # Paramètres spécifiques pour cette exécution if config: f_md.write("- **Paramètres**:\n") f_md.write("```json\n") params = {k: v for k, v in config.items() if k != "system_prompt"} f_md.write(json.dumps(params, indent=2)) f_md.write("\n```\n") # Input/Output (limités en taille) input_data = etape.get("input", "") output_data = etape.get("output", "") if input_data: f_md.write("- **Entrée**:\n") f_md.write("```\n") f_md.write(str(input_data)[:300] + ("..." if len(str(input_data)) > 300 else "")) f_md.write("\n```\n") if output_data: f_md.write("- **Sortie**:\n") f_md.write("```\n") f_md.write(str(output_data)[:300] + ("..." if len(str(output_data)) > 300 else "")) f_md.write("\n```\n") f_md.write("\n") # Résultats des analyses if "analyse_json" in rapport_data: f_md.write("## Résultat de l'Analyse JSON\n\n") f_md.write("```\n") f_md.write(str(rapport_data["analyse_json"])) f_md.write("\n```\n\n") # Analyses des images if "analyse_images" in rapport_data and rapport_data["analyse_images"]: f_md.write("## Analyses des Images\n\n") for image_path, analysis in rapport_data["analyse_images"].items(): image_name = os.path.basename(image_path) f_md.write(f"### Image: {image_name}\n\n") f_md.write("```\n") f_md.write(str(analysis)) f_md.write("\n```\n\n") # Informations supplémentaires f_md.write("## Informations Supplémentaires\n\n") f_md.write(f"Pour plus de détails, consultez le fichier JSON complet: `{os.path.basename(json_path)}`\n\n") logger.info(f"Rapport généré: {md_path}") self.ajouter_historique("generation_rapport", filename, f"Rapport généré: {md_path}") return json_path, md_path