import requests import time import argparse import json import sys from pathlib import Path from typing import List, Dict, Optional 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 # 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 # Les 5 étapes de génération du rapport steps = [ { "title": "Analyse détaillée", "prompt": "Vous rédigez la section 'Analyse détaillée' d'un rapport structuré en cinq parties :\n" "1. Synthèse\n2. Analyse détaillée\n3. Interdépendances\n4. Points de vigilance\n5. Conclusion\n\n" "Votre objectif est d'analyser les vulnérabilités critiques, élevées et modérées présentes dans le rapport ingéré.\n" "Structurez votre réponse par niveau de criticité, et intégrez les indices IHH, ICS, ISG, IVC de façon fluide dans le texte.\n" "N'incluez pas de synthèse ni de conclusion." }, { "title": "Interdépendances", "prompt": "Vous rédigez la section 'Interdépendances'. Identifiez les chaînes de vulnérabilités croisées,\n" "les effets en cascade, et les convergences vers des produits numériques communs.\n" "N'introduisez pas de recommandations ou d'évaluations globales." }, { "title": "Points de vigilance", "prompt": "Vous rédigez la section 'Points de vigilance'. Sur la base des analyses précédentes,\n" "proposez une liste d'indicateurs à surveiller, avec explication courte pour chacun." }, { "title": "Synthèse", "prompt": "Vous rédigez la 'Synthèse'. Résumez les vulnérabilités majeures, les effets en cascade et les indicateurs critiques.\n" "Ton narratif, percutant, orienté décideur." }, { "title": "Conclusion", "prompt": "Vous rédigez la 'Conclusion'. Proposez des scénarios d'impact avec : déclencheur, horizon, gravité, conséquences." } ] 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) -> bool: """Ingère un document dans PrivateGPT""" try: with open(file_path, "rb") as f: files = {"file": (file_path.name, f, "text/markdown")} response = requests.post(f"{API_URL}/ingest/file", files=files) response.raise_for_status() print(f"Document '{file_path}' ingéré avec succès") 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 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: 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 payload = { "messages": [ {"role": "system", "content": """ 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, et produire un rapport narratif percutant orienté décision. """}, {"role": "user", "content": full_prompt} ], "use_context": True, "temperature": 0.2, # Température réduite pour plus de cohérence "stream": False } # 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 main(input_path: str, output_path: str, context_strategy: ContextStrategy = ContextStrategy.SUMMARIZE, context_length: int = MAX_CONTEXT_LENGTH): """Fonction principale qui exécute le processus complet""" # Vérifier la disponibilité de l'API if not check_api_availability(): sys.exit(1) # Convertir les chemins en objets Path input_file = Path(input_path) output_file = Path(output_path) # Ingérer le document if not ingest_document(input_file): 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: {context_strategy.value}, taille max: {context_length} caractères") # Générer chaque section du rapport step_outputs = [] for i, step in enumerate(steps): print(f"\nÉtape {i+1}/{len(steps)}: {step['title']}") # Préparer le contexte selon la stratégie choisie previous = get_context(step_outputs, context_strategy, context_length) if step_outputs else "" # Générer le texte pour cette étape max_retries = 3 if context_strategy != ContextStrategy.NONE else 1 retry_count = 0 output = None while retry_count < max_retries and output is None: try: # Si ce n'est pas la première tentative et que nous avons une stratégie de contexte if retry_count > 0 and context_strategy != ContextStrategy.NONE: print(f"Tentative {retry_count+1}/{max_retries} avec une stratégie de contexte réduite") # Réduire la taille du contexte à chaque tentative reduced_context_length = context_length // (retry_count + 1) fallback_strategy = ContextStrategy.LATEST if retry_count == 1 else ContextStrategy.NONE previous = get_context(step_outputs, fallback_strategy, reduced_context_length) # Générer le texte output = generate_text(step["prompt"], previous, context_strategy != ContextStrategy.NONE) except Exception as e: print(f"⚠️ Erreur lors de la génération: {e}") retry_count += 1 time.sleep(1) # Pause avant de réessayer if output is None: retry_count += 1 if output: step_outputs.append({"title": step["title"], "output": output}) print(f"Section '{step['title']}' générée ({len(output)} caractères)") else: print(f"❌ Échec de la génération pour '{step['title']}'") # 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_text = "\n\n".join(f"# {s['title']}\n\n{s['output']}" for s in ordered_outputs) # Écrire le rapport dans un fichier try: output_file.write_text(report_text, encoding="utf-8") print(f"\nRapport final généré dans '{output_file}'") except IOError as e: print(f"❌ Erreur lors de l'écriture du fichier de sortie: {e}") 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.SUMMARIZE.value, help=f"Stratégie de gestion du contexte (défaut: {ContextStrategy.SUMMARIZE.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)") args = parser.parse_args() # 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)