Spaces:
Sleeping
Sleeping
| # Ce fichier poc/tests/test_corpus_quality.py | |
| # Ce script utilise le jeu de données annoté de 14 questions pour calculer un score de précision. | |
| # Didier : fichier adapté au projet14 Objectif1 | |
| # REVISION SEMANTIQUE : Passage à une évaluation hybride "Dual-Arbiter" (Cosinus + LLM-as-a-Judge) | |
| # Ce script utilise le jeu de données annoté de 14 questions pour calculer un score de précision métier sans faille. | |
| """ | |
| Module d'Évaluation de la Qualité RAG (Phase 5 - Performance - Projet14 objectif1). | |
| Ce script constitue le 'Juge de Paix' final de notre assistant de caisse virtuel. | |
| Pour pallier les angles morts de la similarité cosinus (incapable de distinguer un bouton F10 d'un F9), | |
| cette version implémente une validation croisée : | |
| 1. Proximité vectorielle globale (Mistral AI Embeddings). | |
| 2. Vérification sémantique stricte des commandes UI par un LLM évaluateur (LLM-as-a-Judge). | |
| Esprit : On traque impitoyablement le syndrome du chatbot "poli mais incompétent". Si une manipulation | |
| logicielle ou une règle de rupture pour le hors-périmètre n'est pas respectée, le verdict s'effondre en ÉCHEC. | |
| """ | |
| import os | |
| import sys | |
| import json | |
| import csv | |
| import re | |
| import numpy as np | |
| from datetime import datetime | |
| from dotenv import load_dotenv | |
| # --- Configuration des chemins --- | |
| current_dir = os.path.dirname(os.path.abspath(__file__)) | |
| parent_dir = os.path.dirname(current_dir) | |
| sys.path.append(parent_dir) | |
| from langchain_mistralai import ChatMistralAI, MistralAIEmbeddings | |
| from langchain_community.vectorstores import FAISS | |
| from langchain_core.prompts import ChatPromptTemplate | |
| from langchain_core.output_parsers import StrOutputParser | |
| from langchain_core.runnables import RunnablePassthrough | |
| from utils.logging_config import setup_logging | |
| from utils.config import MISTRAL_API_KEY, FAISS_INDEX_DIR, CHAT_MODEL, EMBEDDING_MODEL | |
| logger = setup_logging() | |
| load_dotenv() | |
| # Constantes d'évaluation | |
| THRESHOLD = 0.80 # Seuil cosinus de tolérance globale | |
| DATASET_PATH = os.path.join(current_dir, "eval_dataset.json") | |
| LOGS_PATH = os.path.join(current_dir, "eval_logs.csv") | |
| def load_ground_truth(): | |
| """ | |
| Charge le jeu de données annoté depuis tests/eval_dataset.json. | |
| Didier : Conserve la structure exacte demandée (category, question, expected_answer). | |
| """ | |
| if not os.path.exists(DATASET_PATH): | |
| logger.error(f"Fichier de test de la Vérité Terrain introuvable : {DATASET_PATH}") | |
| return [] | |
| with open(DATASET_PATH, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| def format_docs(docs): | |
| """ | |
| Formate les documents de procédures extraits de l'armoire FAISS pour le LLM. | |
| """ | |
| formatted = [] | |
| for d in docs: | |
| source = d.metadata.get('source_file', 'Fiche Procédure LR') | |
| process = d.metadata.get('process', 'Inconnu') | |
| content = f"Procédure [{process}] : {d.page_content}" | |
| formatted.append(f"--- SOURCE : {source} ---\n{content}") | |
| return "\n\n".join(formatted) | |
| def build_rag_chain(): | |
| """ | |
| Reconstruit la chaîne RAG en mode 'Headless' pour l'évaluation de précision. | |
| """ | |
| embeddings = MistralAIEmbeddings(mistral_api_key=MISTRAL_API_KEY, model=EMBEDDING_MODEL) | |
| vector_store = FAISS.load_local(str(FAISS_INDEX_DIR), embeddings, allow_dangerous_deserialization=True) | |
| retriever = vector_store.as_retriever(search_kwargs={"k": 5}) | |
| # Température basse pour la stabilité | |
| llm = ChatMistralAI(model=CHAT_MODEL, mistral_api_key=MISTRAL_API_KEY, temperature=0.1) | |
| template = """Tu es un conseiller de caisse virtuel expert pour l'interface du logiciel de caisse LR. | |
| Tu dois guider l'utilisateur pas-à-pas en te basant UNIQUEMENT sur le contexte de procédures fourni. | |
| Règles de conformité strictes (Esprit) : | |
| 1. Base-toi exclusivement sur les connaissances extraites de l'armoire FAISS. N'invente aucune manipulation. | |
| 2. Respecte rigoureusement les boutons et menus de l'interface LR (ex: Chaîne des ventes, bouton Espèce, Règlement multiple F8, F10). | |
| 3. Si la question concerne un problème technique matériel (imprimante ticket en panne), un message d'erreur inconnu (code barre inconnu), une recherche de stock magasin vs web, ou un écran bloqué après action, tu dois STRICTEMENT déclencher la rupture et répondre : | |
| "Je ne peux pas t’aider pour résoudre ce problème. Contacte le support technique." OU "Je ne peux pas t’aider pour résoudre ce problème. Contacte le.a responsable du magasin" (selon la procédure exacte). | |
| 4. Si l'information n'est pas disponible dans le contexte, indique poliment que tu ne peux pas l'aider et de contacter le responsable. | |
| Contexte (Procédures de Caisse LR) : | |
| {context} | |
| Question du conseiller/caissier : | |
| {question} | |
| Réponse pas-à-pas exacte et professionnelle (mots-clés de l'interface respectés) :""" | |
| prompt_template = ChatPromptTemplate.from_template(template) | |
| rag_chain = ( | |
| {"context": retriever | format_docs, "question": RunnablePassthrough()} | |
| | prompt_template | |
| | llm | |
| | StrOutputParser() | |
| ) | |
| return rag_chain | |
| def calculate_cosine_similarity(text1, text2, embeddings_model): | |
| """ | |
| Calcule la similarité mathématique cosinus entre deux textes via mistral-embed. | |
| """ | |
| vec1 = embeddings_model.embed_query(text1) | |
| vec2 = embeddings_model.embed_query(text2) | |
| dot_product = np.dot(vec1, vec2) | |
| norm_vec1 = np.linalg.norm(vec1) | |
| norm_vec2 = np.linalg.norm(vec2) | |
| return dot_product / (norm_vec1 * norm_vec2) | |
| def run_llm_judge(question, expected, generated, llm_evaluator) -> dict: | |
| """ | |
| Didier : Le Juge Sémantique Métier (LLM-as-a-Judge). | |
| Analyse si les actions critiques (boutons, raccourcis, redirections de pannes) | |
| sont rigoureusement respectées sans biais de politesse ou d'illusion vectorielle. | |
| """ | |
| judge_template = """Tu es un auditeur qualité senior pour les processus de caisse chez LR. | |
| Ton rôle est de vérifier la stricte validité de la réponse d'un agent conversationnel face à la Vérité Terrain attendue. | |
| Voici les données de l'évaluation : | |
| - Question posée : {question} | |
| - Réponse attendue (Vérité Terrain) : {expected} | |
| - Réponse générée par le Chatbot : {generated} | |
| Critères d'évaluation impitoyables : | |
| 1. CONFORMITÉ UI : Les raccourcis clavier cruciaux (ex: F10, F8), boutons exacts (ex: Espèce) ou menus logiciels (ex: Chaîne des ventes) présents dans la Vérité Terrain sont-ils EXACTEMENT mentionnés dans la réponse générée ? (Une confusion F10/F9 ou l'oubli du bouton équivaut à un 'NON'). | |
| 2. GESTION DU HORS-PÉRIMÈTRE / RUPTURE : Si la Vérité Terrain stipule que l'agent ne peut pas aider et doit rediriger ("Je ne peux pas t'aider... Contacte le support / le responsable"), la réponse générée applique-t-elle strictement cette consigne de blocage sans inventer de solution technique ? | |
| Renvoie ta réponse sous ce format strict de 3 lignes : | |
| CONFORMITE_UI: [OUI ou NON] | |
| RUPTURE_RESPECTEE: [OUI, NON ou N/A si la question est dans le périmètre] | |
| VERDICT_FINAL: [SUCCÈS ou ÉCHEC] | |
| """ | |
| prompt = ChatPromptTemplate.from_template(judge_template) | |
| chain = prompt | llm_evaluator | StrOutputParser() | |
| try: | |
| response = chain.invoke({ | |
| "question": question, | |
| "expected": expected, | |
| "generated": generated | |
| }) | |
| # Extraction des métriques du Juge par Regex | |
| ui_match = re.search(r"CONFORMITE_UI:\s*(OUI|NON)", response) | |
| rupture_match = re.search(r"RUPTURE_RESPECTEE:\s*(OUI|NON|N/A)", response) | |
| verdict_match = re.search(r"VERDICT_FINAL:\s*(SUCCÈS|ÉCHEC)", response) | |
| return { | |
| "ui_conformity": ui_match.group(1) if ui_match else "NON", | |
| "rupture_status": rupture_match.group(1) if rupture_match else "N/A", | |
| "judge_verdict": verdict_match.group(1) if verdict_match else "ÉCHEC" | |
| } | |
| except Exception as e: | |
| logger.error(f"Erreur lors de l'arbitrage du LLM Judge : {e}") | |
| return {"ui_conformity": "NON", "rupture_status": "NON", "judge_verdict": "ÉCHEC"} | |
| def run_evaluation_cycle(): | |
| """ | |
| Orchestre le passage des tests de caisse avec le double arbitrage Cosinus + LLM Judge. | |
| """ | |
| logger.info("=== Lancement de l'évaluation RAG : Juge de Paix Hybride Caisse LR (Phase 5) ===") | |
| dataset = load_ground_truth() | |
| if not dataset: | |
| return | |
| rag_chain = build_rag_chain() | |
| embeddings = MistralAIEmbeddings(mistral_api_key=MISTRAL_API_KEY, model=EMBEDDING_MODEL) | |
| # Didier : Utilisation du modèle de chat standard pour exécuter l'évaluation critique | |
| llm_evaluator = ChatMistralAI(model=CHAT_MODEL, mistral_api_key=MISTRAL_API_KEY, temperature=0.0) | |
| results = [] | |
| success_count = 0 | |
| for idx, test_case in enumerate(dataset, 1): | |
| logger.info(f"Test {idx}/{len(dataset)}: [{test_case['category']}] -> Question: {test_case['question'][:40]}...") | |
| # 1. Génération de la réponse | |
| generated_answer = rag_chain.invoke(test_case["question"]) | |
| # 2. Arbitre 1 : Similarité Cosinus (Géométrique) | |
| cos_score = calculate_cosine_similarity(generated_answer, test_case["expected_answer"], embeddings) | |
| cos_status = "SUCCÈS" if cos_score >= THRESHOLD else "ÉCHEC" | |
| # 3. Arbitre 2 : LLM-as-a-Judge (Sémantique & Règles métier métiers) | |
| judge_metrics = run_llm_judge(test_case["question"], test_case["expected_answer"], generated_answer, llm_evaluator) | |
| # 4. Sentence Finale : Alignement impératif des deux mondes | |
| if cos_status == "SUCCÈS" and judge_metrics["judge_verdict"] == "SUCCÈS": | |
| final_status = "SUCCÈS" | |
| success_count += 1 | |
| else: | |
| final_status = "ÉCHEC" | |
| logger.info(f"-> Cosinus: {cos_score:.4f} ({cos_status}) | Judge UI: {judge_metrics['ui_conformity']} | Verdict Final: {final_status}") | |
| # 5. Stockage enrichi des métriques pour le tableau de bord | |
| results.append({ | |
| "timestamp": datetime.now().isoformat(), | |
| "category": test_case["category"], | |
| "question": test_case["question"], | |
| "expected_answer": test_case["expected_answer"], | |
| "generated_answer": generated_answer, | |
| "similarity_score": round(cos_score, 4), | |
| "ui_conformity": judge_metrics["ui_conformity"], | |
| "rupture_status": judge_metrics["rupture_status"], | |
| "status": final_status | |
| }) | |
| # Précision globale de la session | |
| accuracy = (success_count / len(dataset)) * 100 | |
| logger.info(f"=== BILAN DE CERTIFICATION : Précision sémantique globale = {accuracy:.2f}% ===") | |
| # Écriture persistante dans le fichier CSV d'évaluation | |
| file_exists = os.path.isfile(LOGS_PATH) | |
| fieldnames = ["timestamp", "category", "question", "expected_answer", "generated_answer", "similarity_score", "ui_conformity", "rupture_status", "status"] | |
| with open(LOGS_PATH, 'a', newline='', encoding='utf-8') as f: | |
| writer = csv.DictWriter(f, fieldnames=fieldnames) | |
| if not file_exists: | |
| writer.writeheader() | |
| writer.writerows(results) | |
| logger.info(f"=== Tableau de bord d'évaluation mis à jour dans {LOGS_PATH} ===") | |
| if __name__ == "__main__": | |
| run_evaluation_cycle() |