Blame for eval/evaluate.py

"""Évalue le pipeline Corrective RAG face à un jeu de questions annotées, et le
compare à un baseline "RAG classique" naïf (top-1 sans grading, sans
correction, sans routage SQL, jamais de refus).

Usage :
    python -m eval.evaluate
    # ou, depuis la racine du projet :
    python eval/evaluate.py

Écrit eval/metrics_latest.json, lu par GET /api/metrics.

Important : sans HF_TOKEN (LLM_PROVIDER=mock), les scores sont structurels
mais non représentatifs. Configurez HF_TOKEN et LLM_PROVIDER=hf pour une
évaluation réelle. EMBEDDING_BACKEND=auto active fastembed.
"""
from __future__ import annotations

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

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from sklearn.feature_extraction.text import CountVectorizer  # noqa: E402
from sklearn.metrics.pairwise import cosine_similarity  # noqa: E402

from app.crag import data_query  # noqa: E402
from app.crag.generator import GENERATOR_SYSTEM_PROMPT, REFUSAL_MESSAGE  # noqa: E402
from app.crag.graph import run_crag  # noqa: E402
from app.crag.llm_client import complete  # noqa: E402
from app.crag.retriever import retrieve_internal_docs  # noqa: E402
from app.crag.verifier import verify_answer  # noqa: E402

QUESTIONS_PATH = Path(__file__).parent / "eval_questions.json"
OUTPUT_PATH = Path(__file__).parent / "metrics_latest.json"

REFUSAL_MARKERS = ("ne peux pas répondre", "ne permettent pas", "ne peux pas confirmer")


def looks_like_refusal(answer: str) -> bool:
    lowered = answer.lower()
    return any(marker in lowered for marker in REFUSAL_MARKERS)


def text_similarity(a: str, b: str) -> float:
    if not a.strip() or not b.strip():
        return 0.0
    try:
        vectorizer = CountVectorizer(binary=True).fit([a, b])
        vectors = vectorizer.transform([a, b])
        return float(cosine_similarity(vectors[0], vectors[1])[0][0])
    except ValueError:
        return 0.0


def run_baseline(question: str) -> dict:
    """RAG classique naïf : top-1 chunk, aucun grading, aucune correction,
    jamais de refus, jamais de routage vers la donnée structurée."""
    chunks = retrieve_internal_docs(question, top_k=1)
    context_text = chunks[0]["text"] if chunks else ""
    doc_ids = [chunks[0]["doc_id"]] if chunks else []
    user_prompt = f"Question :\n{question}\n\nContexte :\n{context_text or '(vide)'}"
    answer = complete(GENERATOR_SYSTEM_PROMPT, user_prompt, json_mode=False).strip()
    verification = verify_answer(question, answer, context_text or "(vide)")
    return {
        "answer": answer,
        "doc_ids": doc_ids,
        "supported": verification.supported,
        "used_structured_data": False,
        "structured_result": None,
        "fallback_triggered": False,
    }


def evaluate_pipeline(questions: list[dict], runner_name: str, run_fn) -> dict:
    precisions, recalls, faithfulness_scores, relevancy_scores = [], [], [], []
    hit_rates = []
    latencies = []
    abstention_correct = 0
    structured_correct, structured_total = 0, 0
    fallback_count = 0

    for item in questions:
        started = time.perf_counter()
        result = run_fn(item["question"])
        latencies.append((time.perf_counter() - started) * 1000)

        expected = set(item.get("expected_relevant_doc_ids", []))
        retrieved = set(result["doc_ids"])

        if expected:
            precisions.append(
                len(retrieved & expected) / len(retrieved) if retrieved else 0.0
            )
            recalls.append(len(retrieved & expected) / len(expected))
            hit_rates.append(1.0 if retrieved & expected else 0.0)

        faithfulness_scores.append(1.0 if result["supported"] else 0.0)
        relevancy_scores.append(
            text_similarity(item["question"], result["answer"])
        )

        abstained = (
            looks_like_refusal(result["answer"])
            or result["answer"] == REFUSAL_MESSAGE
        )
        should_answer = item["should_answer"]
        if (should_answer and not abstained) or (
            not should_answer and abstained
        ):
            abstention_correct += 1

        if item.get("expects_structured"):
            structured_total += 1
            if result["used_structured_data"]:
                structured_correct += 1

        if result.get("fallback_triggered"):
            fallback_count += 1

    n = len(questions)
    return {
        "context_precision": (
            round(sum(precisions) / len(precisions), 4) if precisions else None
        ),
        "context_recall": (
            round(sum(recalls) / len(recalls), 4) if recalls else None
        ),
        "retrieval_hit_rate": (
            round(sum(hit_rates) / len(hit_rates), 4) if hit_rates else None
        ),
        "faithfulness": round(sum(faithfulness_scores) / n, 4),
        "answer_relevancy": round(sum(relevancy_scores) / n, 4),
        "abstention_accuracy": round(abstention_correct / n, 4),
        "sql_execution_accuracy": (
            round(structured_correct / structured_total, 4)
            if structured_total
            else None
        ),
        "fallback_rate": round(fallback_count / n, 4),
        "avg_latency_ms": round(sum(latencies) / len(latencies), 1),
        "_runner": runner_name,
    }


def run_crag_wrapper(question: str) -> dict:
    state = run_crag(question)
    verification = state.get("verification") or {"supported": True}
    return {
        "answer": state.get("final_answer", ""),
        "doc_ids": list(dict.fromkeys(state.get("sources", []))),
        "supported": bool(verification.get("supported", True)),
        "used_structured_data": bool(state.get("used_structured_data")),
        "structured_result": state.get("structured_result"),
        "fallback_triggered": state.get("decision") != "CORRECT",
    }


def main() -> None:
    questions = json.loads(QUESTIONS_PATH.read_text(encoding="utf-8"))
    started = time.time()

    print(f"Évaluation de {len(questions)} questions — RAG classique (baseline)…")
    classic_metrics = evaluate_pipeline(questions, "rag_classique", run_baseline)

    print(f"Évaluation de {len(questions)} questions — Corrective RAG…")
    crag_metrics = evaluate_pipeline(questions, "corrective_rag", run_crag_wrapper)

    output = {
        "n_questions": len(questions),
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "duration_seconds": round(time.time() - started, 1),
        "rag_classique": {k: v for k, v in classic_metrics.items() if not k.startswith("_")},
        "corrective_rag": {k: v for k, v in crag_metrics.items() if not k.startswith("_")},
    }
    OUTPUT_PATH.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8")
    print(f"Métriques écrites dans {OUTPUT_PATH}")
    print(json.dumps(output, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    main()