293 lines
14 KiB
Python
293 lines
14 KiB
Python
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)
|