Initial commit - Code education RAG Space

.gitignore

@@ -0,0 +1,21 @@
+# Python
+__pycache__/
+*.pyc
+
+# Local venv (si jamais)
+.venv/
+venv/
+llm_code_education_env/
+
+# Modèles (jamais dans git)
+models/
+*.gguf
+*.bin
+models/*.gguf
+
+
+# Secrets
+.env
+
+# OS
+.DS_Store

app.py

@@ -0,0 +1,216 @@
+# app.py — Gradio UI for hf-code-education (CPU / Hugging Face Spaces)
+# This file must NOT change the validated RAG logic.
+# It only calls src/rag_core.py:answer_query(query).
+# to launch http://localhost:7860
+
+
+import os
+import sys
+import traceback
+import gradio as gr
+from huggingface_hub import hf_hub_download
+
+
+def ensure_model_present():
+    os.makedirs("models", exist_ok=True)
+    local_path = os.path.join("models", "mistral.gguf")
+    if os.path.exists(local_path):
+        return
+
+    repo_id = os.environ.get("MODEL_REPO_ID")
+    filename = os.environ.get("MODEL_FILENAME", "mistral.gguf")
+
+    if not repo_id:
+        raise RuntimeError(
+            "Modèle GGUF absent (models/mistral.gguf) et variable MODEL_REPO_ID non définie."
+        )
+
+    downloaded = hf_hub_download(repo_id=repo_id, filename=filename)
+    import shutil
+    shutil.copyfile(downloaded, local_path)
+
+ensure_model_present()
+
+
+
+# Ensure we can import src/rag_core.py without requiring src/ to be a package
+ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+SRC_DIR = os.path.join(ROOT_DIR, "src")
+if SRC_DIR not in sys.path:
+    sys.path.insert(0, SRC_DIR)
+
+# Import the validated core
+try:
+    import rag_core  # src/rag_core.py
+except Exception as e:
+    raise RuntimeError(
+        "Impossible d'importer src/rag_core.py. "
+        "Vérifie que le fichier existe bien et qu'il s'appelle exactement rag_core.py."
+    ) from e
+
+
+def _format_result(result) -> str:
+    """
+    Formats output robustly WITHOUT assuming a strict schema.
+    We do NOT modify any RAG logic, just display what comes back.
+    """
+    if result is None:
+        return "Aucune réponse (result=None)."
+
+    # Most common: a string answer
+    if isinstance(result, str):
+        return result
+
+    # If the core returns a dict-like object
+    if isinstance(result, dict):
+        # Try common keys while staying generic
+        parts = []
+        if "mode" in result:
+            parts.append(f"Mode: {result['mode']}")
+        if "answer" in result:
+            parts.append(str(result["answer"]))
+        elif "response" in result:
+            parts.append(str(result["response"]))
+        else:
+            # Fallback: dump dict (readable)
+            parts.append(str(result))
+
+        # Optional: sources / context / citations
+        for k in ["sources", "citations", "articles", "context_used", "context"]:
+            if k in result and result[k]:
+                parts.append(f"\n\n---\n{k}:\n{result[k]}")
+        return "\n\n".join(parts)
+
+    # If the core returns a tuple/list (e.g., (answer, meta))
+    if isinstance(result, (tuple, list)):
+        return "\n\n".join([str(x) for x in result])
+
+    # Fallback
+    return str(result)
+
+
+def chat_once(user_query: str) -> str:
+    """
+    Single-shot call to the validated RAG core.
+    """
+    q = (user_query or "").strip()
+    if not q:
+        return "Entre une question ou une demande (vide = rien à traiter)."
+
+    try:
+        # IMPORTANT: Do not change rag_core logic; just call it.
+        result = rag_core.answer_query(q)
+        return _format_result(result)
+    except Exception:
+        # Show error transparently (useful on HF Spaces logs)
+        err = traceback.format_exc()
+        return "Erreur côté application (pas côté utilisateur):\n\n" + err
+
+
+CSS = """
+/* Police sérieuse, institutionnelle */
+:root {
+  --font-sans: Inter, "Source Sans 3", Roboto, "Segoe UI", Arial, sans-serif;
+}
+
+body, .gradio-container {
+  font-family: var(--font-sans) !important;
+  font-size: 15px;
+  line-height: 1.5;
+}
+
+/* Titres plus sobres */
+h1, h2, h3 {
+  font-weight: 600;
+  letter-spacing: -0.01em;
+}
+
+/* Page un peu plus compacte */
+.gradio-container {
+  max-width: 980px !important;
+}
+
+/* Réponse : hauteur max + scroll */
+#answer textarea {
+  max-height: 360px !important;
+  overflow-y: auto !important;
+  font-size: 14px;
+  line-height: 1.55;
+}
+
+/* Moins d'espacement vertical */
+.wrap {
+  gap: 0.6rem !important;
+}
+"""
+
+
+with gr.Blocks(
+    title="Assistant Code de l’éducation (RAG)",
+    css=CSS,
+    theme=gr.themes.Soft(),
+) as demo:
+    gr.Markdown(
+        """
+# Assistant Code de l’éducation
+Cet outil recherche dans le Code de l’éducation (version du 7 janvier 2026) et répond uniquement à partir des articles retrouvés.
+
+### Ce que l’outil fait
+- Cite des articles (ou liste des articles pertinents)
+- Répond à une question si le texte nécessaire est présent dans les articles retrouvés
+
+### Ce que l’outil ne fait pas
+- N’invente pas : si le contexte est insuffisant, il refuse et le dit clairement
+- Ne remplace pas une validation juridique
+
+Conseil : pour une citation exacte, demande “Donne l’intégralité de l’article …”.
+        """.strip()
+    )
+
+    gr.Markdown(
+    """
+> **Information importante**  
+> Lors du premier lancement, l’application peut nécessiter **1 à 2 minutes** d’initialisation.  
+> Ensuite, l’utilisation est immédiate.  
+> En cas d’utilisation simultanée, les demandes sont traitées **successivement** afin de garantir la fiabilité des réponses.
+    """.strip()
+)
+
+
+    with gr.Row():
+        inp = gr.Textbox(
+            label="Votre demande",
+            placeholder="Ex : Donne l’intégralité de l’article D454-14",
+            lines=2,
+            max_lines=4,
+        )
+
+    with gr.Row():
+        out = gr.Textbox(
+            label="Réponse",
+            elem_id="answer",
+            lines=10,
+            max_lines=14,
+        )
+
+    with gr.Row():
+        btn = gr.Button("Répondre", variant="primary")
+        clear = gr.Button("Effacer")
+
+    btn.click(chat_once, inputs=inp, outputs=out)
+    clear.click(lambda: ("", ""), outputs=[inp, out])
+
+    with gr.Accordion("Exemples de requêtes", open=False):
+        gr.Examples(
+            examples=[
+                "Donne l'intégralité de l'article D454-14",
+                "Liste les articles qui parlent de l'obligation scolaire",
+                "Quelles sont les conditions de nomination d'un chef d'établissement ? Cite uniquement les articles fournis.",
+            ],
+            inputs=inp,
+        )
+
+
+# HF Spaces expects launch on 0.0.0.0:7860
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)

requirements.txt

@@ -0,0 +1,13 @@
+gradio==4.44.1
+faiss-cpu==1.8.0.post1
+numpy==1.26.4
+sentence-transformers==3.0.1
+llama-cpp-python==0.3.7
+langchain-community
+torch
+langchain-huggingface
+huggingface_hub
+
+
+
+

src/rag_core.py

@@ -0,0 +1,381 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+rag_core.py
+
+Transposition FIDÈLE de rag_chat_llama.py (mêmes règles, mêmes seuils, même prompt,
+même validation anti-hallucination), mais sans boucle interactive : on expose
+une fonction answer_query(question) utilisable par une app Hugging Face.
+
+ROUTAGE AUTO :
+- FULLTEXT : demande "texte exact / intégral / article X" => impression exacte depuis JSONL (SANS LLM)
+- LIST     : demande "quels articles parlent ..." => liste articles + extrait (SANS LLM)
+- QA       : RAG => LLM (llama/Mistral) + prompt strict + VALIDATION (anti-hallucinations)
+
+Prérequis :
+- data/chunks_articles.jsonl (article-level)
+- db/faiss_code_edu_by_article (FAISS)
+- models/mistral.gguf (GGUF)
+"""
+
+import json
+import re
+from pathlib import Path
+from typing import List, Tuple, Optional, Dict, Iterable, Any
+
+from langchain_community.vectorstores import FAISS
+from langchain_huggingface import HuggingFaceEmbeddings
+from llama_cpp import Llama
+
+
+# -------------------- CONFIG --------------------
+CHUNKS_PATH = Path("data/chunks_articles.jsonl")
+DB_DIR = Path("db/faiss_code_edu_by_article")
+
+EMBED_MODEL = "sentence-transformers/all-MiniLM-L6-v2"
+MODEL_NAME = "mistral:latest"
+
+TOP_K_FETCH = 30            # nb de docs candidats récupérés
+TOP_K_FINAL = 4            # nb max envoyés au LLM
+SCORE_THRESHOLD = 1.10      # à ajuster (voir affichage des scores)
+MAX_CHARS_PER_DOC = 800
+SNIPPET_CHARS = 260
+
+# Déclencheurs FULLTEXT
+FULLTEXT_TRIGGERS = [
+    "contenu exact", "texte exact", "texte intégral", "texte integral",
+    "intégral", "integral", "cite intégralement", "cite integralement",
+    "donne l'intégralité", "donne l'integralite", "recopie", "reproduis",
+    "affiche l'article", "donne l'article", "donne moi l'article",
+]
+
+# Déclencheurs LIST
+LIST_TRIGGERS = [
+    "quels articles", "quelles dispositions", "articles parlent",
+    "articles qui parlent", "articles sur", "donne les articles",
+    "cite les articles", "références", "references",
+]
+
+# Regex article id
+ARTICLE_ID_RE = re.compile(
+    r"\b(?:article\s+)?([LDR]\s?\d{1,4}(?:[.-]\d+){0,4})\b",
+    flags=re.IGNORECASE
+)
+
+EPLE_RE = re.compile(r"\bEPLE\b", flags=re.IGNORECASE)
+
+# Pour valider les sorties "Articles cités : ..."
+ARTICLES_CITES_RE = re.compile(r"Articles cités\s*:\s*(.*)$", flags=re.IGNORECASE | re.MULTILINE)
+
+# -------------------- LLM INIT (FIDÈLE) --------------------
+llm = Llama(
+    model_path="models/mistral.gguf",  # Mistral GGUF
+    n_ctx=2048,
+    n_threads=10,
+    n_batch=128,
+    verbose=False,
+)
+
+
+def llm_generate(prompt: str) -> str:
+    out = llm.create_chat_completion(
+        messages=[{"role": "user", "content": prompt}],
+        temperature=0.1,
+        max_tokens=200,
+    )
+    return out["choices"][0]["message"]["content"].strip()
+
+
+# -------------------- UTILS (FIDÈLES) --------------------
+
+def normalize_article_id(raw: str) -> str:
+    s = raw.strip().upper().replace(" ", "")
+    s = s.replace(".", "-")
+    return s
+
+
+def extract_article_id(q: str) -> Optional[str]:
+    m = ARTICLE_ID_RE.search(q)
+    if not m:
+        return None
+    return normalize_article_id(m.group(1))
+
+
+def is_fulltext_request(q: str) -> bool:
+    ql = q.lower()
+    if any(t in ql for t in FULLTEXT_TRIGGERS):
+        return True
+    aid = extract_article_id(q)
+    if aid and len(ql) <= 25:
+        return True
+    return False
+
+
+def is_list_request(q: str) -> bool:
+    ql = q.lower()
+    return any(t in ql for t in LIST_TRIGGERS)
+
+
+def dedupe_keep_order(items: Iterable[str]) -> List[str]:
+    seen = set()
+    out = []
+    for x in items:
+        if x not in seen:
+            out.append(x)
+            seen.add(x)
+    return out
+
+
+def safe_snippet(text: str, n: int) -> str:
+    t = " ".join((text or "").split())
+    if len(t) <= n:
+        return t
+    return t[:n].rstrip() + "…"
+
+
+def load_article_text(article_id: str) -> Optional[str]:
+    if not CHUNKS_PATH.exists():
+        raise FileNotFoundError(f"Fichier chunks introuvable : {CHUNKS_PATH}")
+
+    with CHUNKS_PATH.open("r", encoding="utf-8") as f:
+        for line in f:
+            if not line.strip():
+                continue
+            obj = json.loads(line)
+            aid = normalize_article_id(obj.get("article_id", ""))
+            if aid == article_id:
+                return (obj.get("text") or "").strip()
+    return None
+
+
+def load_vectorstore() -> FAISS:
+    if not DB_DIR.exists():
+        raise FileNotFoundError(f"Index FAISS introuvable : {DB_DIR}")
+    embeddings = HuggingFaceEmbeddings(model_name=EMBED_MODEL)
+    return FAISS.load_local(str(DB_DIR), embeddings, allow_dangerous_deserialization=True)
+
+
+def retrieve_scored(vs: FAISS, query: str) -> List[Tuple[object, float]]:
+    """
+    Retourne liste (Document, score). Plus le score est PETIT, plus c'est proche (distance).
+    """
+    return vs.similarity_search_with_score(query, k=TOP_K_FETCH)
+
+
+def filter_docs(scored: List[Tuple[object, float]]) -> List[Tuple[object, float]]:
+    """
+    Filtre simple par seuil + garde TOP_K_FINAL.
+    """
+    kept = [(d, s) for (d, s) in scored if s <= SCORE_THRESHOLD]
+    if not kept:
+        # fallback : au moins TOP_K_FINAL meilleurs, sinon tu refuses trop souvent
+        kept = sorted(scored, key=lambda x: x[1])[:TOP_K_FINAL]
+    else:
+        kept = sorted(kept, key=lambda x: x[1])[:TOP_K_FINAL]
+    return kept
+
+
+def build_context(scored_docs: List[Tuple[object, float]]) -> Tuple[str, List[str], Dict[str, str], Dict[str, float]]:
+    used = []
+    by_id: Dict[str, str] = {}
+    by_score: Dict[str, float] = {}
+
+    blocks = []
+    for d, s in scored_docs:
+        aid = d.metadata.get("article_id", "UNKNOWN")
+        aid_norm = normalize_article_id(aid)
+        used.append(aid_norm)
+
+        txt = (d.page_content or "").strip()
+        by_id[aid_norm] = txt
+        by_score[aid_norm] = float(s)
+
+        if len(txt) > MAX_CHARS_PER_DOC:
+            txt = txt[:MAX_CHARS_PER_DOC].rstrip() + "\n[.]"
+
+        blocks.append(f"[{aid_norm}]\n{txt}")
+
+    used = dedupe_keep_order(used)
+    return "\n\n".join(blocks), used, by_id, by_score
+
+
+def eple_context_ok(question: str, by_id: Dict[str, str]) -> bool:
+    """
+    Si la question contient "EPLE", on veut que le contexte contienne explicitement
+    des indices "collège/lycée/établissement public local d'enseignement".
+    """
+    if not EPLE_RE.search(question):
+        return True
+
+    joined = "\n".join(by_id.values()).lower()
+    signals = [
+        "établissement public local d'enseignement",
+        "etablissement public local d'enseignement",
+        "collège", "college", "lycée", "lycee",
+        "chef d'établissement", "chef d'etablissement",
+    ]
+    return any(sig in joined for sig in signals)
+
+
+def extract_cited_articles(answer: str) -> List[str]:
+    m = ARTICLES_CITES_RE.search(answer)
+    if not m:
+        return []
+    tail = m.group(1).strip()
+    if not tail:
+        return []
+    parts = re.split(r"[,\s]+", tail)
+    out = []
+    for p in parts:
+        p = p.strip()
+        if not p:
+            continue
+        # tolère "D422-15." ou "[D422-15]"
+        p = p.strip("[]().;:")
+        if ARTICLE_ID_RE.match(p) or re.match(r"^[LDR]\d", p, flags=re.I):
+            out.append(normalize_article_id(p))
+    return dedupe_keep_order(out)
+
+
+def validate_answer(answer: str, allowed_articles: List[str]) -> bool:
+    cited = extract_cited_articles(answer)
+    allowed_set = set(allowed_articles)
+
+    # si le LLM ne cite rien => on refuse (sinon il peut raconter)
+    if not cited:
+        return False
+
+    # interdit de citer un article non présent dans la liste autorisée
+    if any(c not in allowed_set for c in cited):
+        return False
+
+    return True
+
+
+def build_prompt(question: str, context: str, allowed_articles: List[str]) -> str:
+    allowed = ", ".join(allowed_articles)
+
+    return f"""Tu es un assistant juridique spécialisé dans le Code de l'éducation (France).
+
+RÈGLES ABSOLUES (non négociables) :
+1) Tu réponds UNIQUEMENT à partir du CONTEXTE fourni ci-dessous.
+2) Tu n'inventes rien, tu ne complètes pas, tu ne "supposes" pas. Interdiction d'utiliser :
+   "on peut supposer", "il est possible que", "on peut déduire", "probablement", etc.
+3) Si le CONTEXTE ne permet pas de répondre, tu dis exactement :
+   "Je ne peux pas répondre avec certitude à partir des articles fournis."
+4) Tu DOIS citer uniquement des articles présents dans la liste autorisée :
+   {allowed}
+5) Attention au sigle EPLE :
+   - EPLE = établissement public local d'enseignement (collèges/lycées).
+   - Ne confonds pas avec d'autres établissements.
+   Si le CONTEXTE ne traite pas clairement des EPLE au sens collèges/lycées, tu refuses de conclure.
+
+QUESTION :
+{question}
+
+CONTEXTE :
+{context}
+
+FORMAT DE SORTIE OBLIGATOIRE :
+- Une réponse courte et factuelle.
+- Dernière ligne STRICTE : "Articles cités : A, B, C" (uniquement parmi la liste autorisée).
+"""
+
+
+# -------------------- CORE API (HF) --------------------
+_REFUSAL = "Je ne peux pas répondre avec certitude à partir des articles fournis."
+
+# cache pour éviter de recharger FAISS à chaque call
+_VS: Optional[FAISS] = None
+
+
+def get_vectorstore() -> FAISS:
+    global _VS
+    if _VS is None:
+        _VS = load_vectorstore()
+    return _VS
+
+
+def answer_query(q: str) -> Dict[str, Any]:
+    """
+    API équivalente à la boucle interactive de rag_chat_llama.py.
+
+    Retourne un dict structuré :
+    - mode: "FULLTEXT" | "LIST" | "QA"
+    - answer: str (réponse finale ou refus)
+    - articles: liste des articles récupérés (pour debug/affichage)
+    - scores: dict {article: score} (pour debug/affichage)
+    - snippets: (LIST) dict {article: snippet}
+    - fulltext: (FULLTEXT) texte exact
+    """
+    q = (q or "").strip()
+    if not q:
+        return {"mode": "QA", "answer": _REFUSAL, "articles": [], "scores": {}}
+
+    vs = get_vectorstore()
+
+    # --- FULLTEXT ---
+    aid = extract_article_id(q)
+    if aid and is_fulltext_request(q):
+        txt = load_article_text(aid)
+        if not txt:
+            return {
+                "mode": "FULLTEXT",
+                "answer": f"Je ne trouve pas l'article {aid} dans {CHUNKS_PATH}.",
+                "articles": [],
+                "scores": {},
+                "fulltext": None,
+            }
+        return {
+            "mode": "FULLTEXT",
+            "answer": txt,
+            "articles": [aid],
+            "scores": {},
+            "fulltext": txt,
+        }
+
+    # --- RETRIEVE (scored) ---
+    scored = retrieve_scored(vs, q)
+    scored = filter_docs(scored)
+    context, articles, by_id, by_score = build_context(scored)
+
+    # --- LIST ---
+    if is_list_request(q):
+        snippets = {a: safe_snippet(by_id.get(a, ""), SNIPPET_CHARS) for a in articles}
+        return {
+            "mode": "LIST",
+            "answer": "",
+            "articles": articles,
+            "scores": by_score,
+            "snippets": snippets,
+        }
+
+    # --- EPLE safety gate ---
+    if not eple_context_ok(q, by_id):
+        return {
+            "mode": "QA",
+            "answer": _REFUSAL,
+            "articles": articles,
+            "scores": by_score,
+        }
+
+    # --- QA (LLM) ---
+    prompt = build_prompt(q, context, articles)
+    answer = llm_generate(prompt)
+
+    # --- VALIDATION ---
+    if not validate_answer(answer, articles):
+        return {
+            "mode": "QA",
+            "answer": _REFUSAL,
+            "articles": articles,
+            "scores": by_score,
+        }
+
+    return {
+        "mode": "QA",
+        "answer": answer,
+        "articles": articles,
+        "scores": by_score,
+    }