| """É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 |
| from sklearn.metrics.pairwise import cosine_similarity |
|
|
| from app.crag import data_query |
| from app.crag.generator import GENERATOR_SYSTEM_PROMPT, REFUSAL_MESSAGE |
| from app.crag.graph import run_crag |
| from app.crag.llm_client import complete |
| from app.crag.retriever import retrieve_internal_docs |
| from app.crag.verifier import verify_answer |
|
|
| 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() |
|
|
