Ajoute un producteur de questions OpenAI

Author: pie-leroy

pyproject.toml

@@ -8,6 +8,7 @@ dependencies = [
      "docling",
      "filelock>=3.20.3",
      "httpx",
+     "numpy",
      "openai",
      "json-repair",
      "pandas==2.3.2",
@@ -17,6 +18,8 @@ dependencies = [
      "reportlab",
      "requests",
      "respx",
+     "scikit-learn",
+     "sentence-transformers",
      "types-reportlab",
      "urllib3>=2.5.1",
      "aiohttp>=3.13.3",

src/guides/generateur_question/compteurs_thematique.py

@@ -0,0 +1,79 @@
+from abc import ABC, abstractmethod
+from typing import Any
+
+import numpy as np
+from sklearn.cluster import AgglomerativeClustering  # type: ignore[import-untyped]
+
+from guides.generateur_question.utils import _decoupe_en_phrases, _charge_encodeur
+
+
+class CompteurThematiques(ABC):
+    @abstractmethod
+    def nombre_topics(self, paragraphe: str) -> int:
+        raise NotImplementedError
+
+
+class CompteursThematique(CompteurThematiques):
+    def __init__(
+        self,
+        *,
+        modele_hf: str = "BAAI/bge-m3",
+        seuil_distance: float = 0.35,
+        min_topics: int = 1,
+        max_topics: int = 10,
+        min_phrases: int = 2,
+        encodeur: Any | None = None,
+    ) -> None:
+        self.modele_hf = modele_hf
+        self.seuil_distance = seuil_distance
+        self.min_topics = min_topics
+        self.max_topics = max_topics
+        self.min_phrases = min_phrases
+        self.encodeur = encodeur
+
+    def nombre_topics(self, paragraphe: str) -> int:
+        phrases = _decoupe_en_phrases(paragraphe)
+        if len(phrases) == 0:
+            return 0
+        if len(phrases) < self.min_phrases:
+            return 1
+        encodeur = (
+            self.encodeur
+            if self.encodeur is not None
+            else _charge_encodeur(self.modele_hf)
+        )
+        vecteurs = encodeur.encode(
+            phrases,
+            normalize_embeddings=True,
+            convert_to_numpy=True,
+            show_progress_bar=False,
+        ).astype(np.float32)
+
+        distances = 1.0 - (vecteurs @ vecteurs.T)
+        np.fill_diagonal(distances, 0.0)
+
+        createur_de_cluster = AgglomerativeClustering(
+            n_clusters=None,
+            metric="precomputed",
+            linkage="average",
+            distance_threshold=self.seuil_distance,
+        )
+        etiquettes = createur_de_cluster.fit_predict(distances)
+
+        n = int(len(set(etiquettes.tolist())))
+        n = max(self.min_topics, min(self.max_topics, n))
+        return n
+
+
+def calcule_nombre_questions(
+    paragraphe: str, compteur_thematiques: CompteurThematiques | None = None
+) -> int:
+    compteur = (
+        compteur_thematiques
+        if compteur_thematiques is not None
+        else CompteursThematique()
+    )
+    n_topics = compteur.nombre_topics(paragraphe)
+    if n_topics == 0:
+        return 0
+    return max(3, min(10, n_topics))

src/guides/generateur_question/producteur_questions.py

@@ -0,0 +1,54 @@
+from typing import Any, cast
+
+from openai import OpenAI
+from openai.types.chat import ChatCompletion, ChatCompletionMessageParam
+
+from configuration import recupere_configuration
+from guides.generateur_question.compteurs_thematique import calcule_nombre_questions
+from guides.generateur_question.utils import (
+    _charge_prompt_systeme,
+    parse_questions_depuis_contenu,
+)
+
+
+class ProducteurQuestionsOpenAI:
+    def __init__(
+        self,
+        *,
+        client: Any | None = None,
+        modele_generation: str | None = None,
+        temperature: float = 0.0,
+    ) -> None:
+        configuration = recupere_configuration().albert
+        self.client = (
+            client
+            if client is not None
+            else OpenAI(base_url=configuration.url, api_key=configuration.cle_api)
+        )
+        self.modele_generation = (
+            modele_generation if modele_generation is not None else configuration.modele
+        )
+        self.temperature = temperature
+
+    def __call__(self, paragraphe: str) -> list[str]:
+        n_questions = calcule_nombre_questions(paragraphe)
+        messages: list[ChatCompletionMessageParam] = [
+            {"role": "system", "content": _charge_prompt_systeme()},
+            {
+                "role": "user",
+                "content": (
+                    f"Génère exactement {n_questions} questions.\n"
+                    f"Paragraphe :\n{paragraphe}"
+                ),
+            },
+        ]
+        completion = self.client.chat.completions.create(
+            model=self.modele_generation,
+            messages=messages,
+            temperature=self.temperature,
+            stream=False,
+            n=1,
+        )
+        completion = cast(ChatCompletion, completion)
+        contenu = (completion.choices[0].message.content or "").strip()
+        return parse_questions_depuis_contenu(contenu)

src/guides/generateur_question/utils.py

@@ -0,0 +1,62 @@
+import json
+import re
+from functools import lru_cache
+from pathlib import Path
+
+import numpy as np
+
+
+def _charge_prompt_systeme() -> str:
+    racine_projet = Path(__file__).resolve().parents[3]
+    chemin_prompt = racine_projet / "tempaltes" / "prompt_generateur_questions.txt"
+    return chemin_prompt.read_text(encoding="utf-8").strip()
+
+
+def _extrait_objet_json(texte: str) -> str:
+    t = (texte or "").strip()
+    t = re.sub(r"^\s*```(?:json)?\s*", "", t, flags=re.IGNORECASE)
+    t = re.sub(r"\s*```\s*$", "", t)
+    t = t.lstrip("\ufeff")
+    match = re.search(r"\{.*\}", t, flags=re.DOTALL)
+    if not match:
+        raise ValueError("Aucun objet JSON détecté dans la sortie du modèle.")
+    return match.group(0)
+
+
+def parse_questions_depuis_contenu(contenu: str) -> list[str]:
+    bloc = _extrait_objet_json(contenu)
+    obj = json.loads(bloc)
+    questions = obj.get("questions", [])
+    if not isinstance(questions, list):
+        return []
+    return [q.strip() for q in questions if isinstance(q, str) and q.strip()]
+
+
+def _compte_mots(texte: str) -> int:
+    return len(re.findall(r"\b\w+\b", texte, flags=re.UNICODE))
+
+
+def _decoupe_en_phrases(texte: str) -> list[str]:
+    texte = re.sub(r"\s+", " ", (texte or "").strip())
+    if not texte:
+        return []
+    phrases = re.split(r"(?<=[.!?])\s+", texte)
+    return [p.strip() for p in phrases if p.strip()]
+
+
+def _compte_phrases(texte: str) -> int:
+    phrases = re.split(r"[.!?]\s+", texte.strip())
+    phrases = [p for p in phrases if p.strip()]
+    return max(1, len(phrases))
+
+
+def _normalise_l2(m: np.ndarray) -> np.ndarray:
+    normes = np.linalg.norm(m, axis=1, keepdims=True) + 1e-12
+    return m / normes
+
+
+@lru_cache(maxsize=4)
+def _charge_encodeur(modele_hf: str):
+    from sentence_transformers import SentenceTransformer
+
+    return SentenceTransformer(modele_hf)

tempaltes/prompt_generateur_questions.txt

@@ -0,0 +1,60 @@
+Tu es un composant de génération de questions pour un système RAG en cybersécurité (ANSSI).
+
+Mission :
+À partir d’un paragraphe, générer EXACTEMENT N questions réalistes qu’un utilisateur pourrait formuler pour retrouver ce paragraphe via recherche sémantique.
+
+Règles de sortie (obligatoires) :
+- Retourner UNIQUEMENT un JSON STRICT sur UNE seule ligne, sans texte avant/après, sans Markdown, sans ```fences```.
+- Une seule clé autorisée : "questions".
+- Format exact : {"questions":["...","..."]}
+
+Contraintes de contenu :
+1) Langue : français.
+2) Chaque élément de "questions" est une UNIQUE phrase interrogative et se termine par "?".
+3) Répondabilité (answerability) : chaque question doit être répondable uniquement à partir du paragraphe.
+   - Interdit : exiger une source externe, une interprétation juridique non présente, ou un contexte absent.
+4) Autoportance : aucune question ne doit dépendre d’un contexte externe.
+   - Interdit : pronoms/référents non résolus ("ça", "cela", "ce cas", "cette méthode", "ils") sans nom explicite.
+5) Un seul axe par question :
+   - Interdit : combiner deux thèmes indépendants dans une même question (ex : "périmètre ET coopération").
+   - Si deux thèmes apparaissent, produire deux questions distinctes.
+6) Non-duplication : pas de doublons ni de paraphrases quasi identiques.
+
+Couverture attendue (diversité) :
+- La liste doit couvrir des angles différents parmi :
+  - définition / explication
+  - mécanisme / fonctionnement
+  - risques / menaces / attaques
+  - limites / contournements / échecs
+  - bonnes pratiques / recommandations
+  - causes / conséquences
+  - comparaison / alternatives
+
+Optimisation retrieval (longueur et densité de signal) :
+7) Répartition :
+   - 30–40% de "requêtes courtes" (6 à 10 mots) de type moteur de recherche, terminées par "?".
+   - Les autres questions sont concises : 8 à 16 mots maximum.
+8) Requêtes courtes : elles doivent rester interprétables.
+   - Elles doivent contenir soit (a) un verbe, soit (b) un noyau d’intention explicite
+     (ex : "entités concernées", "changements majeurs", "objectif", "rôle", "périmètre", "exigences").
+   - Interdit : suites nominales vagues sans intention (ex : "Rôle X dans Y ?" si cela devient ambigu).
+9) Élagage :
+   - Supprimer les détails non discriminants qui dégradent le retrieval : dates, jugements, cadrages vagues,
+     formulations verbeuses ("est-il considéré comme", "en matière de", "à l’échelle ...").
+   - Ne conserver que : sujet + intention + 1 à 3 termes/entités clés présents dans le paragraphe.
+10) Robustesse lexicale :
+   - Conserver les termes techniques du paragraphe.
+   - Ajouter au plus 1 synonyme utile par question quand pertinent (ex : "coffre-fort de mots de passe" / "gestionnaire de mots de passe").
+   - Conserver les acronymes, et développer uniquement si le développement est explicitement présent dans le paragraphe.
+
+Mise en avant des recommandations ANSSI :
+11) Si le paragraphe contient une mention de recommandation "R" suivie d’un ou plusieurs chiffres (ex : "R1", "R32", "R33", "R34") :
+   - Générer au moins UNE question dédiée par recommandation détectée.
+   - La question doit citer explicitement la recommandation (ex : "Que dit la recommandation R33 sur ... ?").
+   - Interdit : inventer le contenu complet de la recommandation si le paragraphe ne le détaille pas.
+
+Nettoyage obligatoire (anti-notes, anti-citations, anti-marquage éditorial) :
+12) Interdire et supprimer dans les questions :
+   - toute référence bibliographique ou note : tout motif entre crochets [ ... ] (ex : [14], [1], [12–14]).
+   - tout astérisque "*" et tout texte entre astérisques (ex : *obligatoire*).
+   - "cf.", "voir", "référence", "guide", "article", ou toute mention de source externe.

tests/guides/producteur_questions/test_producteur_questions.py

@@ -0,0 +1,122 @@
+from types import SimpleNamespace
+
+import pytest
+
+from guides.generateur_question.producteur_questions import (
+    ProducteurQuestionsOpenAI,
+    _charge_prompt_systeme,
+    calcule_nombre_questions,
+    parse_questions_depuis_contenu,
+)
+from guides.generateur_question.compteurs_thematique import (
+    CompteurThematiques,
+    CompteursThematique,
+)
+from guides.generateur_question.utils import (
+    _extrait_objet_json,
+    _compte_mots,
+    _compte_phrases,
+)
+
+
+class FauxClientOpenAI:
+    def __init__(self, contenu: str):
+        self.contenu = contenu
+        self.appels: list[dict] = []
+        self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._cree))
+
+    def _cree(self, **kwargs):
+        self.appels.append(kwargs)
+        return SimpleNamespace(
+            choices=[SimpleNamespace(message=SimpleNamespace(content=self.contenu))]
+        )
+
+
+def test_producteur_questions_openai_retourne_les_questions_json():
+    contenu = """```json
+                {"questions": ["Question 1 ?", "Question 2 ?"]}
+                ```"""
+    client = FauxClientOpenAI(contenu)
+    producteur = ProducteurQuestionsOpenAI(
+        client=client,
+        modele_generation="modele-test",
+    )
+
+    paragraphe = "Un paragraphe test simple."
+    resultats = producteur(paragraphe)
+
+    assert resultats == ["Question 1 ?", "Question 2 ?"]
+    assert len(client.appels) == 1
+    assert client.appels[0]["model"] == "modele-test"
+
+
+def test_charge_prompt_systeme_charge_le_contenu_du_fichier():
+    prompt = _charge_prompt_systeme()
+    assert "Tu es un composant de génération de questions" in prompt
+
+
+def test_extrait_objet_json_retire_les_fences():
+    contenu = '```json\n{"questions": ["Q1 ?"]}\n```'
+
+    extrait = _extrait_objet_json(contenu)
+
+    assert extrait == '{"questions": ["Q1 ?"]}'
+
+
+def test_parse_questions_depuis_contenu_extrait_les_questions():
+    contenu = '```json\n{"questions": ["Q1 ?", "Q2 ?"]}\n```'
+
+    resultats = parse_questions_depuis_contenu(contenu)
+
+    assert resultats == ["Q1 ?", "Q2 ?"]
+
+
+def test_compte_mots_compte_les_mots():
+    assert _compte_mots("Un texte simple.") == 3
+
+
+def test_compte_phrases_compte__une_phrase():
+    assert _compte_phrases("Une première phrase.") == 1
+
+
+def test_compte_phrases_compte_au_moins_une_phrase():
+    assert _compte_phrases("") == 1
+
+
+def test_compte_phrases_compte_les_phrases():
+    assert _compte_phrases("Phrase une. Phrase deux ? Phrase trois !") == 3
+
+
+class CompteurThematiquesDeTest(CompteurThematiques):
+    def __init__(self, topics: int):
+        self.topics = topics
+
+    def nombre_topics(self, _: str) -> int:
+        return self.topics
+
+
+@pytest.mark.parametrize(
+    ("topics", "attendu"),
+    [
+        (0, 0),
+        (1, 3),
+        (2, 3),
+        (5, 5),
+        (10, 10),
+        (12, 10),
+    ],
+)
+def test_calcule_nombre_questions_est_borne(topics: int, attendu: int):
+    compteur = CompteurThematiquesDeTest(topics)
+    resultat = calcule_nombre_questions("peu importe", compteur_thematiques=compteur)
+    assert resultat == attendu
+
+
+def test_compteur_topics_retourne_0_si_paragraphe_vide():
+    compteur = CompteursThematique()
+    assert compteur.nombre_topics("") == 0
+
+
+def test_compteur_topics_retourne_1_si_pas_assez_de_phrases():
+    compteur = CompteursThematique()
+    assert compteur.nombre_topics("Une seule phrase.") == 1