evaluate.py

"""
evaluate.py — Etape 3 : mesurer la qualite du systeme.

C'est la partie qui distingue une demo d'un vrai projet.
Sans mesure, on ne sait pas si une modification (taille des morceaux,
valeur de k, changement de modele) ameliore ou degrade le systeme.

Trois mesures, volontairement simples et explicables :

  1. RECALL@K  — la recherche ramene-t-elle le bon passage ?
     Pour chaque question de test, on verifie qu'au moins un des k morceaux
     retournes contient le mot-cle attendu. C'est la mesure la plus importante :
     si la recherche echoue, le meilleur LLM du monde ne peut pas repondre.

  2. TAUX DE CITATION — la reponse cite-t-elle sa source entre crochets ?
     Proxy simple de la "groundedness" : une reponse sans source est suspecte.

  3. TAUX DE REFUS CORRECT — sur des questions volontairement hors sujet,
     le systeme dit-il bien qu'il ne sait pas, au lieu d'inventer ?
     C'est le test anti-hallucination.

    python evaluate.py                 (recherche seule, rapide)
    python evaluate.py --with-llm      (recherche + generation, plus lent)
"""

from __future__ import annotations

import argparse
import json
import re
import sys
import time
from pathlib import Path

from llm import DEFAULT_MODEL, ask_ollama, build_prompt
from rag_core import Embedder, VectorStore

JEU_DE_TEST = "questions.json"


def charger_jeu_de_test(path: str) -> dict:
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(
            f"{path} introuvable. Cree-le sur ce modele :\n"
            '{\n'
            '  "dans_les_docs": [\n'
            '    {"question": "...", "mot_cle_attendu": "..."}\n'
            '  ],\n'
            '  "hors_sujet": ["question sans reponse dans les docs"]\n'
            '}'
        )
    return json.loads(p.read_text(encoding="utf-8"))


def contient(passages, mot_cle: str) -> bool:
    """Le mot-cle attendu apparait-il dans l'un des passages retrouves ?"""
    besoin = mot_cle.lower()
    return any(besoin in chunk.text.lower() for chunk, _ in passages)


def main() -> None:
    ap = argparse.ArgumentParser(description="Evalue le systeme RAG.")
    ap.add_argument("--k", type=int, default=4)
    ap.add_argument("--index", default="index")
    ap.add_argument("--questions", default=JEU_DE_TEST)
    ap.add_argument("--model", default=DEFAULT_MODEL)
    ap.add_argument("--with-llm", action="store_true",
                    help="evaluer aussi les reponses generees (plus lent)")
    args = ap.parse_args()

    try:
        jeu = charger_jeu_de_test(args.questions)
        store = VectorStore.load(args.index)
    except (FileNotFoundError, ValueError) as e:
        print(f"\n{e}")
        sys.exit(1)
    embedder = Embedder(store.model_name)

    print(f"Index : {len(store.chunks)} morceaux — k = {args.k}\n")

    # ---------------------------------------------------------------- 1. recall
    print("=" * 68)
    print("1. RECALL@K — la recherche trouve-t-elle le bon passage ?")
    print("=" * 68)

    cas = jeu.get("dans_les_docs", [])
    trouves = 0
    rangs = []

    for item in cas:
        q, mot = item["question"], item["mot_cle_attendu"]
        passages = store.search(embedder.encode([q])[0], k=args.k)
        ok = contient(passages, mot)
        trouves += ok

        rang = "-"
        for i, (chunk, _) in enumerate(passages, start=1):
            if mot.lower() in chunk.text.lower():
                rang = str(i)
                rangs.append(i)
                break

        print(f"  [{'OK ' if ok else 'RATE'}] rang {rang:>2}  {q[:58]}")

    recall = trouves / len(cas) if cas else 0.0
    print(f"\n  Recall@{args.k} : {trouves}/{len(cas)}  =  {recall:.0%}")
    if rangs:
        print(f"  Rang moyen du bon passage : {sum(rangs)/len(rangs):.1f}")

    # ------------------------------------------------- 2 & 3. qualite reponses
    if args.with_llm:
        print("\n" + "=" * 68)
        print("2. CITATION DES SOURCES")
        print("=" * 68)

        avec_source = 0
        t0 = time.time()
        for item in cas:
            q = item["question"]
            passages = store.search(embedder.encode([q])[0], k=args.k)
            rep = ask_ollama(build_prompt(q, passages), model=args.model)
            cite = bool(re.search(r"\[[^\]]+\]", rep))
            avec_source += cite
            print(f"  [{'OK ' if cite else 'SANS'}] {q[:58]}")

        print(f"\n  Reponses citant une source : "
              f"{avec_source}/{len(cas)} = {avec_source/len(cas):.0%}"
              if cas else "  (aucun cas)")

        print("\n" + "=" * 68)
        print("3. REFUS CORRECT — questions hors sujet")
        print("=" * 68)

        hors = jeu.get("hors_sujet", [])
        refus = 0
        for q in hors:
            passages = store.search(embedder.encode([q])[0], k=args.k)
            rep = ask_ollama(build_prompt(q, passages), model=args.model)
            a_refuse = "ne trouve pas" in rep.lower() or "pas la reponse" in rep.lower()
            refus += a_refuse
            print(f"  [{'OK    ' if a_refuse else 'INVENTE'}] {q[:56]}")
            if not a_refuse:
                print(f"          -> {rep[:110]}...")

        if hors:
            print(f"\n  Refus corrects : {refus}/{len(hors)} = {refus/len(hors):.0%}")
        print(f"\n  (generation : {time.time() - t0:.0f} s)")

    # ---------------------------------------------------------------- synthese
    print("\n" + "=" * 68)
    print("A RETENIR POUR EN PARLER")
    print("=" * 68)
    print("  - Un recall faible ne se corrige pas en changeant de LLM :")
    print("    il faut revoir le decoupage, le chevauchement ou la valeur de k.")
    print("  - Augmenter k ameliore le recall mais dilue le contexte,")
    print("    augmente le cout en tokens et peut degrader la reponse.")
    print("  - Note tes valeurs avant / apres chaque modification :")
    print("    c'est ce qui transforme un bricolage en demarche d'ingenieur.")


if __name__ == "__main__":
    main()