import requests import time import argparse import json import sys import uuid from pathlib import Path from typing import List, Dict, Any, Optional, Tuple from enum import Enum # Configuration de l'API PrivateGPT PGPT_URL = "http://127.0.0.1:8001" API_URL = f"{PGPT_URL}/v1" DEFAULT_INPUT_PATH = "rapport_final.md" DEFAULT_OUTPUT_PATH = "rapport_analyse_genere.md" MAX_CONTEXT_LENGTH = 3000 # Taille maximale du contexte en caractères TEMP_DIR = Path("temp_sections") # Répertoire pour les fichiers intermédiaires # Stratégies de gestion du contexte class ContextStrategy(str, Enum): NONE = "none" # Pas de contexte TRUNCATE = "truncate" # Tronquer le contexte SUMMARIZE = "summarize" # Résumer le contexte LATEST = "latest" # Uniquement la dernière section FILE = "file" # Utiliser des fichiers ingérés (meilleure option) # Les 5 étapes de génération du rapport steps = [ { "title": "Analyse Principale", "prompt": """Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels). Objectif : Analyser les vulnérabilités systémiques dans une chaîne de valeur numérique décrite dans le fichier rapport_final.md en vous appuyant exclusivement sur ce rapport structuré fourni. Votre tâche est de développer les sections suivantes en respectant toutes les contraintes décrites : 1. Analyse détaillée (vulnérabilités critiques → modérées) 2. Interdépendances (effets en cascade) 3. Points de vigilance (indicateurs à surveiller) Interprétation des indices : • IHH (Herfindahl-Hirschmann) : - Échelle : 0-100 - Seuils : <25 = Vert (faible), 25-50 = Orange (modéré), >50 = Rouge (élevée) • ICS (Substituabilité) : - Échelle : 0-1 - Interprétation : 0 = facilement substituable, 1 = impossible • ISG (Stabilité Géopolitique) : - Échelle : 0-100 - Seuils : <40 = Vert, 40-60 = Orange, >60 = Rouge • IVC (Concurrence) : - Échelle : 0-150 - Seuils : <10 = Faible, 10-50 = Moyenne, >50 = Forte Contraintes strictes : 1. N'utiliser aucune connaissance externe. Seul le contenu fourni compte. 2. Ne pas recommander de diversification minière, industrielle ou politique. 3. Concentrez-vous sur les impacts sur les produits numériques. Votre mission : - Interpréter les indices (IHH, ICS, ISG, IVC) - Croiser et hiérarchiser les vulnérabilités - Distinguer les horizons temporels - Identifier les chaînes de vulnérabilités interdépendantes - Identifier des indicateurs d'alerte pertinents Style attendu : - Narratif, fluide, impactant, structuré - Paragraphes complets + encadrés synthétiques avec données chiffrées Commencez par lire attentivement les indices et les fiches dans la section "Éléments factuels", puis produisez vos sections d'analyse.""" }, { "title": "Synthèse et Conclusion", "prompt": """Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels). Sur la base de l'analyse précédente, rédigez les deux sections finales du rapport : 1. Synthèse (narration + hiérarchie vulnérabilités) 2. Conclusion (scénarios d'impact) Pour la Synthèse : - Résumez de façon percutante les vulnérabilités identifiées dans la chaîne de valeur numérique - Hiérarchisez les risques en fonction du croisement des indices (IHH, ISG, ICS, IVC) - Orientez cette synthèse vers la prise de décision pour différents acteurs économiques - Utilisez un ton direct, factuel et orienté décision Pour la Conclusion : - Proposez 3-4 scénarios d'impact précis avec déclencheur, horizon temporel, gravité et conséquences - Basez chaque scénario sur les données factuelles de l'analyse - Montrez comment les différents indices interagissent dans chaque scénario - Concentrez-vous particulièrement sur les applications numériques identifiées Interprétation des indices (rappel) : • IHH (Herfindahl-Hirschmann) : <25 = Vert, 25-50 = Orange, >50 = Rouge • ICS (Substituabilité) : 0 = facilement substituable, 1 = impossible • ISG (Stabilité Géopolitique) : <40 = Vert, 40-60 = Orange, >60 = Rouge • IVC (Concurrence) : <10 = Faible, 10-50 = Moyenne, >50 = Forte Style attendu : - Narratif, percutant, orienté décideur - Encadrés synthétiques avec données chiffrées clés - Focus sur les implications concrètes pour les entreprises""" } ] def check_api_availability() -> bool: """Vérifie si l'API PrivateGPT est disponible""" try: response = requests.get(f"{PGPT_URL}/health") if response.status_code == 200: print("✅ API PrivateGPT disponible") return True else: print(f"❌ L'API PrivateGPT a retourné le code d'état {response.status_code}") return False except requests.RequestException as e: print(f"❌ Erreur de connexion à l'API PrivateGPT: {e}") return False def ingest_document(file_path: Path, session_id: str = "") -> bool: """Ingère un document dans PrivateGPT""" try: with open(file_path, "rb") as f: # Si un session_id est fourni, l'ajouter au nom du fichier pour le tracking if session_id: file_name = f"input_{session_id}_{file_path.name}" else: file_name = file_path.name files = {"file": (file_name, f, "text/markdown")} # Ajouter des métadonnées pour identifier facilement ce fichier d'entrée metadata = { "type": "input_file", "session_id": session_id, "document_type": "rapport_analyse_input" } response = requests.post( f"{API_URL}/ingest/file", files=files, data={"metadata": json.dumps(metadata)} if "metadata" in requests.get(f"{API_URL}/ingest/file").text else None ) response.raise_for_status() print(f"✅ Document '{file_path}' ingéré avec succès sous le nom '{file_name}'") return True except FileNotFoundError: print(f"❌ Fichier '{file_path}' introuvable") return False except requests.RequestException as e: print(f"❌ Erreur lors de l'ingestion du document: {e}") return False def setup_temp_directory() -> None: """Crée le répertoire temporaire pour les fichiers intermédiaires""" if not TEMP_DIR.exists(): TEMP_DIR.mkdir(parents=True) print(f"📁 Répertoire temporaire '{TEMP_DIR}' créé") def save_section_to_file(section: Dict[str, str], index: int, session_uuid: str) -> Path: """Sauvegarde une section dans un fichier temporaire et retourne le chemin""" setup_temp_directory() section_file = TEMP_DIR / f"temp_section_{session_uuid}_{index+1}_{section['title'].lower().replace(' ', '_')}.md" # Contenu du fichier avec métadonnées et commentaire explicite content = ( f"# SECTION TEMPORAIRE GÉNÉRÉE - {section['title']}\n\n" f"Note: Ce document est une section temporaire du rapport d'analyse en cours de génération.\n" f"UUID de session: {session_uuid}\n\n" f"{section['output']}" ) # Écrire dans le fichier section_file.write_text(content, encoding="utf-8") return section_file def ingest_section_files(section_files: List[Path]) -> List[str]: """Ingère les fichiers de section et retourne leurs noms de fichiers""" ingested_file_names = [] for file_path in section_files: try: with open(file_path, "rb") as f: files = {"file": (file_path.name, f, "text/markdown")} # Ajouter des métadonnées pour identifier facilement nos fichiers temporaires metadata = { "type": "temp_section", "document_type": "rapport_analyse_section" } response = requests.post( f"{API_URL}/ingest/file", files=files, data={"metadata": json.dumps(metadata)} if "metadata" in requests.get(f"{API_URL}/ingest/file").text else None ) response.raise_for_status() # Ajouter le nom du fichier à la liste pour pouvoir le retrouver et le supprimer plus tard ingested_file_names.append(file_path.name) print(f"✅ Section '{file_path.name}' ingérée") except Exception as e: print(f"⚠️ Erreur lors de l'ingestion de '{file_path.name}': {e}") return ingested_file_names def get_context(sections: List[Dict[str, str]], strategy: ContextStrategy, max_length: int) -> str: """Génère le contexte selon la stratégie choisie""" if not sections or strategy == ContextStrategy.NONE: return "" # Stratégie: utiliser des fichiers ingérés if strategy == ContextStrategy.FILE: # Cette stratégie n'utilise pas de contexte dans le prompt # Elle repose sur les fichiers ingérés et le paramètre use_context=True section_names = [s["title"] for s in sections] context_note = f"NOTE IMPORTANTE: Les sections précédentes ({', '.join(section_names)}) " + \ f"ont été ingérées sous forme de fichiers temporaires avec l'identifiant unique '{session_uuid}'. " + \ f"Utilisez UNIQUEMENT le document '{input_file.name}' et ces sections temporaires pour votre analyse. " + \ f"IGNOREZ tous les autres documents qui pourraient être présents dans la base de connaissances. " + \ f"Assurez la cohérence avec les sections déjà générées pour maintenir la continuité du rapport." print(f"📄 Utilisation de {len(sections)} sections ingérées comme contexte") return context_note # Stratégie: uniquement la dernière section if strategy == ContextStrategy.LATEST: latest = sections[-1] context = f"# {latest['title']}\n{latest['output']}" if len(context) > max_length: context = context[:max_length] + "..." print(f"📄 Contexte basé sur la dernière section: {latest['title']} ({len(context)} caractères)") return context # Stratégie: tronquer toutes les sections if strategy == ContextStrategy.TRUNCATE: full_context = "\n\n".join([f"# {s['title']}\n{s['output']}" for s in sections]) if len(full_context) > max_length: truncated = full_context[:max_length] + "..." print(f"✂️ Contexte tronqué à {max_length} caractères") return truncated return full_context # Stratégie: résumer intelligemment if strategy == ContextStrategy.SUMMARIZE: try: # Construire un prompt pour demander un résumé des sections précédentes sections_text = "\n\n".join([f"# {s['title']}\n{s['output']}" for s in sections]) # Limiter la taille du texte à résumer si nécessaire if len(sections_text) > max_length * 2: sections_text = sections_text[:max_length * 2] + "..." summary_prompt = { "messages": [ {"role": "system", "content": "Vous êtes un assistant spécialisé dans le résumé concis. Créez un résumé très court mais informatif qui conserve les points clés."}, {"role": "user", "content": f"Résumez les sections suivantes en {max_length // 10} mots maximum, en conservant les idées principales et les points essentiels:\n\n{sections_text}"} ], "temperature": 0.1, "stream": False } # Envoyer la requête pour obtenir un résumé response = requests.post( f"{API_URL}/chat/completions", json=summary_prompt, headers={"accept": "application/json"} ) response.raise_for_status() # Extraire et retourner le résumé result = response.json() summary = result["choices"][0]["message"]["content"] print(f"✅ Résumé de contexte généré ({len(summary)} caractères)") return summary except Exception as e: print(f"⚠️ Impossible de générer un résumé du contexte: {e}") # En cas d'échec, revenir à la stratégie de troncature return get_context(sections, ContextStrategy.TRUNCATE, max_length) def generate_text(prompt: str, previous_context: str = "", use_context: bool = True, retry_on_error: bool = True) -> Optional[str]: """Génère du texte avec l'API PrivateGPT""" try: # Préparer le prompt avec le contexte précédent si disponible et demandé full_prompt = prompt if previous_context and use_context: full_prompt += "\n\nContexte précédent (informations des sections déjà générées) :\n" + previous_context # Définir les paramètres de la requête system_message = "Vous êtes un assistant stratégique expert chargé de produire des rapports destinés à des décideurs de haut niveau (COMEX, directions risques, stratèges politiques ou industriels). " + \ f"Objectif : Analyser les vulnérabilités systémiques dans une chaîne de valeur numérique selon les données du fichier {input_file.name}. " + \ "Interprétation des indices : " + \ "• IHH (Herfindahl-Hirschmann) : Échelle 0-100, Seuils : <25 = Vert (faible), 25-50 = Orange (modéré), >50 = Rouge (élevée) " + \ "• ISG (Stabilité Géopolitique) : Échelle 0-100, Seuils : <40 = Vert (stable), 40-60 = Orange, >60 = Rouge (instable) " + \ "• ICS (Substituabilité) : Échelle 0-1, 0 = facilement substituable, 1 = impossible " + \ "• IVC (Concurrence) : Échelle 0-150, Seuils : <10 = Faible, 10-50 = Moyenne, >50 = Forte " + \ "Style attendu : Narratif, fluide, impactant, structuré avec paragraphes complets et données chiffrées. " + \ f"Concentrez-vous UNIQUEMENT sur le document principal '{input_file.name}' et sur les sections temporaires préfixées par 'temp_section_{session_uuid}_'. " + \ "IGNOREZ absolument tous les autres documents qui pourraient être présents dans la base de connaissances. " + \ "Les sections précédentes ont été générées et ingérées comme documents temporaires - " + \ "assurez la cohérence avec leur contenu pour maintenir la continuité du rapport." # Définir les paramètres de la requête payload = { "messages": [ {"role": "system", "content": system_message}, {"role": "user", "content": full_prompt} ], "use_context": True, # Active la recherche RAG dans les documents ingérés "temperature": 0.2, # Température réduite pour plus de cohérence "stream": False } # Tenter d'ajouter un filtre de contexte (fonctionnalité expérimentale qui peut ne pas être supportée) try: # Vérifier si le filtre de contexte est supporté sans faire de requête supplémentaire filter_metadata = { "document_name": [input_file.name] + [f.name for f in section_files] } payload["filter_metadata"] = filter_metadata except Exception as e: print(f"ℹ️ Remarque: Impossible d'appliquer le filtre de contexte: {e}") # Envoyer la requête response = requests.post( f"{API_URL}/chat/completions", json=payload, headers={"accept": "application/json"} ) response.raise_for_status() # Extraire la réponse générée result = response.json() if "choices" in result and len(result["choices"]) > 0: return result["choices"][0]["message"]["content"] else: print("⚠️ Format de réponse inattendu:", json.dumps(result, indent=2)) return None except requests.RequestException as e: print(f"❌ Erreur lors de la génération de texte: {e}") if hasattr(e, 'response') and e.response is not None: print(f"Détails: {e.response.text}") return None def cleanup_temp_files(temp_file_names: List[str] = None, remove_directory: bool = False, session_id: str = "") -> None: """Nettoie les fichiers temporaires et les documents ingérés""" try: # Supprimer les fichiers du répertoire temporaire if TEMP_DIR.exists(): for temp_file in TEMP_DIR.glob("*.md"): temp_file.unlink() print(f"🗑️ Fichier temporaire supprimé : {temp_file.name}") # Supprimer le répertoire s'il est vide et si demandé if remove_directory and not any(TEMP_DIR.iterdir()): TEMP_DIR.rmdir() print(f"🗑️ Répertoire temporaire '{TEMP_DIR}' supprimé") # Supprimer les documents ingérés via l'API de liste et suppression try: # Lister tous les documents ingérés list_response = requests.get(f"{API_URL}/ingest/list") if list_response.status_code == 200: documents_data = list_response.json() # Format de réponse OpenAI if "data" in documents_data: documents = documents_data.get("data", []) # Format alternatif else: documents = documents_data.get("documents", []) deleted_count = 0 # Parcourir les documents et supprimer ceux qui correspondent à nos fichiers temporaires for doc in documents: doc_metadata = doc.get("doc_metadata", {}) file_name = doc_metadata.get("file_name", "") or doc_metadata.get("filename", "") # Vérifier si c'est un de nos fichiers temporaires ou le fichier d'entrée is_our_file = False if temp_file_names and file_name in temp_file_names: is_our_file = True elif f"temp_section_{session_uuid}_" in file_name: is_our_file = True elif session_id and f"input_{session_id}_" in file_name: is_our_file = True if is_our_file: doc_id = doc.get("doc_id") or doc.get("id") if doc_id: delete_response = requests.delete(f"{API_URL}/ingest/{doc_id}") if delete_response.status_code == 200: deleted_count += 1 if deleted_count > 0: print(f"🗑️ {deleted_count} documents supprimés de PrivateGPT") except Exception as e: print(f"⚠️ Erreur lors de la suppression des documents ingérés: {e}") except Exception as e: print(f"⚠️ Erreur lors du nettoyage des fichiers temporaires: {e}") def clean_ai_thoughts(text: str) -> str: """Nettoie le texte généré en supprimant les 'pensées' de l'IA entre balises """ import re # Supprimer tout le contenu entre les balises et , y compris les balises text = re.sub(r'.*?', '', text, flags=re.DOTALL) # Supprimer les lignes vides multiples text = re.sub(r'\n{3,}', '\n\n', text) return text def main(input_path: str, output_path: str, context_strategy: ContextStrategy = ContextStrategy.FILE, context_length: int = MAX_CONTEXT_LENGTH): """Fonction principale qui exécute le processus complet""" global input_file, section_files, session_uuid # Variables globales pour le filtre de contexte et l'UUID # Générer un UUID unique pour cette session session_uuid = str(uuid.uuid4())[:8] # Utiliser les 8 premiers caractères pour plus de concision print(f"🔑 UUID de session généré: {session_uuid}") # Vérifier la disponibilité de l'API if not check_api_availability(): sys.exit(1) # Convertir les chemins en objets Path (accessibles globalement) input_file = Path(input_path) output_file = Path(output_path) # Ingérer le document principal avec l'UUID de session if not ingest_document(input_file, session_uuid): sys.exit(1) # Récupérer la valeur du délai depuis args delay = args.delay if 'args' in globals() else 5 # Attendre que l'ingestion soit complètement traitée print(f"⏳ Attente du traitement de l'ingestion pendant {delay} secondes...") time.sleep(delay) print(f"🔧 Stratégie de contexte initiale: {context_strategy.value}, taille max: {context_length} caractères") # Préparer le répertoire pour les fichiers temporaires setup_temp_directory() # Générer chaque section du rapport step_outputs = [] section_files = [] # Chemins des fichiers temporaires ingested_file_names = [] # Noms des fichiers ingérés # Liste des stratégies de fallback dans l'ordre fallback_strategies = [ ContextStrategy.FILE, ContextStrategy.TRUNCATE, ContextStrategy.SUMMARIZE, ContextStrategy.LATEST, ContextStrategy.NONE ] for i, step in enumerate(steps): print(f"\n🚧 Étape {i+1}/{len(steps)}: {step['title']}") # Si nous avons des sections précédentes et utilisons la stratégie FILE if step_outputs and context_strategy == ContextStrategy.FILE: # Sauvegarder et ingérer toutes les sections précédentes qui ne l'ont pas encore été for j, section in enumerate(step_outputs): if j >= len(section_files): # Cette section n'a pas encore été sauvegardée section_file = save_section_to_file(section, j, session_uuid) section_files.append(section_file) # Ingérer le fichier si nous utilisons la stratégie FILE new_ids = ingest_section_files([section_file]) ingested_section_ids.extend(new_ids) # Attendre que l'ingestion soit traitée print(f"⏳ Attente du traitement de l'ingestion des sections précédentes...") time.sleep(2) # Essayer chaque stratégie jusqu'à ce qu'une réussisse output = None current_strategy_index = fallback_strategies.index(context_strategy) while output is None and current_strategy_index < len(fallback_strategies): current_strategy = fallback_strategies[current_strategy_index] print(f"🔄 Tentative avec la stratégie: {current_strategy.value}") # Préparer le contexte selon la stratégie actuelle previous = get_context(step_outputs, current_strategy, context_length) if step_outputs else "" # Générer le texte try: output = generate_text(step["prompt"], previous, current_strategy != ContextStrategy.NONE) if output is None: # Passer à la stratégie suivante current_strategy_index += 1 print(f"⚠️ Échec avec la stratégie {current_strategy.value}, passage à la suivante") time.sleep(1) # Pause avant nouvelle tentative except Exception as e: print(f"⚠️ Erreur lors de la génération avec {current_strategy.value}: {e}") current_strategy_index += 1 time.sleep(1) # Pause avant nouvelle tentative if output: step_outputs.append({"title": step["title"], "output": output}) print(f"✅ Section '{step['title']}' générée ({len(output)} caractères)") # Si nous sommes en mode FILE, sauvegarder immédiatement cette section if context_strategy == ContextStrategy.FILE: section_file = save_section_to_file(step_outputs[-1], len(step_outputs)-1, session_uuid) section_files.append(section_file) # Ingérer le fichier new_file_names = ingest_section_files([section_file]) ingested_file_names.extend(new_file_names) # Petite pause pour permettre l'indexation time.sleep(1) else: print(f"❌ Échec de la génération pour '{step['title']}' après toutes les stratégies") # Vérifier si nous avons généré toutes les sections if len(step_outputs) != len(steps): print(f"⚠️ Attention: Seules {len(step_outputs)}/{len(steps)} sections ont été générées") # Écrire le rapport final dans l'ordre souhaité (Synthèse en premier) # Réordonner les sections pour que la synthèse soit en premier ordered_outputs = sorted(step_outputs, key=lambda x: 0 if x["title"] == "Synthèse" else 1 if x["title"] == "Analyse détaillée" else 2 if x["title"] == "Interdépendances" else 3 if x["title"] == "Points de vigilance" else 4) # Assembler le rapport report_sections = [] for s in ordered_outputs: # Nettoyer les "pensées" de l'IA dans chaque section cleaned_output = clean_ai_thoughts(s['output']) report_sections.append(f"## {s['title']}\n\n{cleaned_output}") report_text = "# Analyse des vulnérabilités de la chaine de fabrication du numériquee\n\n".join(report_sections) # Écrire le rapport dans un fichier try: output_file.write_text(report_text, encoding="utf-8") print(f"\n📄 Rapport final généré dans '{output_file}'") except IOError as e: print(f"❌ Erreur lors de l'écriture du fichier de sortie: {e}") # Nettoyer les fichiers temporaires si demandé if args.clean_temp: print("\n🧹 Nettoyage des fichiers temporaires et documents ingérés...") cleanup_temp_files(ingested_file_names, args.remove_temp_dir, session_uuid) if __name__ == "__main__": # Définir les arguments en ligne de commande parser = argparse.ArgumentParser(description="Générer un rapport d'analyse structuré avec PrivateGPT") parser.add_argument("-i", "--input", type=str, default=DEFAULT_INPUT_PATH, help=f"Chemin du fichier d'entrée (défaut: {DEFAULT_INPUT_PATH})") parser.add_argument("-o", "--output", type=str, default=DEFAULT_OUTPUT_PATH, help=f"Chemin du fichier de sortie (défaut: {DEFAULT_OUTPUT_PATH})") parser.add_argument("--context", type=str, choices=[s.value for s in ContextStrategy], default=ContextStrategy.FILE.value, help=f"Stratégie de gestion du contexte (défaut: {ContextStrategy.FILE.value})") parser.add_argument("--context-length", type=int, default=MAX_CONTEXT_LENGTH, help=f"Taille maximale du contexte en caractères (défaut: {MAX_CONTEXT_LENGTH})") parser.add_argument("--delay", type=int, default=5, help="Délai d'attente en secondes après l'ingestion (défaut: 5)") parser.add_argument("--clean-temp", action="store_true", default=True, help="Nettoyer les fichiers temporaires et documents ingérés par le script (défaut: True)") parser.add_argument("--keep-temp", action="store_true", help="Conserver les fichiers temporaires (remplace --clean-temp)") parser.add_argument("--remove-temp-dir", action="store_true", help="Supprimer le répertoire temporaire après exécution (défaut: False)") args = parser.parse_args() # Gérer les options contradictoires if args.keep_temp: args.clean_temp = False # Exécuter le programme principal avec les options configurées context_strategy = ContextStrategy(args.context) main(args.input, args.output, context_strategy, args.context_length)