"""É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()