From f7d87f2602fbfb0984f808b59509f255563ad096 Mon Sep 17 00:00:00 2001 From: dom Date: Wed, 18 Feb 2026 18:16:34 +0100 Subject: [PATCH] =?UTF-8?q?feat:=20pipeline=20CPAM=20multi-pass=20+=20gard?= =?UTF-8?q?e-fous=20qualit=C3=A9=20(solutions=201+2+3+6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Solution 1 : injection déterministe des définitions CIM-10 dans le prompt - Solution 2 : grounding tagué [BIO-N], [IMG-N], [TRT-N], [ACTE-N] avec validation - Solution 3 : pipeline 2 passes (extraction structurée → argumentation) - Solution 6 : validation adversariale LLM post-génération - Normes bio injectées dans les tags (NORMAL/ÉLEVÉ/BAS avec norme de référence) - Cross-check DAS/biologie détecte les incohérences (leucocytose vs leucocytes bas) - Contexte patient : flags pédiatrie, patient âgé, admission urgence - Dossiers pauvres : avertissement explicite au lieu de spéculation - Validation adversariale enrichie avec normes bio de référence - 75 tests CPAM (612 total), 0 régression Co-Authored-By: Claude Opus 4.6 --- src/control/cpam_response.py | 519 ++++++++++++++++++++++++++- tests/test_cpam_response.py | 665 +++++++++++++++++++++++++++++++++-- 2 files changed, 1136 insertions(+), 48 deletions(-) diff --git a/src/control/cpam_response.py b/src/control/cpam_response.py index db94d87..049faee 100644 --- a/src/control/cpam_response.py +++ b/src/control/cpam_response.py @@ -7,6 +7,7 @@ import re from ..config import ControleCPAM, DossierMedical, RAGSource from ..medical.cim10_dict import normalize_code, validate_code +from ..medical.cim10_extractor import BIO_NORMALS from ..medical.ollama_client import call_anthropic, call_ollama logger = logging.getLogger(__name__) @@ -162,12 +163,242 @@ def _get_code_label(code_str: str) -> str: return "\n " + "\n ".join(labels) +def _get_cim10_definitions( + dossier: DossierMedical, + controle: ControleCPAM, +) -> str: + """Construit une section de définitions CIM-10 déterministes pour tous les codes en jeu. + + Collecte les codes depuis : + - Le dossier : DP (cim10_suggestion) + DAS (cim10_suggestion) + - L'UCR : dp_ucr, da_ucr, dr_ucr + + Returns: + Texte formaté pour injection dans le prompt, ou "" si aucun code résolu. + """ + codes_seen: dict[str, str] = {} # code normalisé → rôle (pour affichage) + + # Codes du dossier (établissement) + if dossier.diagnostic_principal and dossier.diagnostic_principal.cim10_suggestion: + code = dossier.diagnostic_principal.cim10_suggestion + codes_seen[normalize_code(code)] = "DP établissement" + for das in dossier.diagnostics_associes: + if das.cim10_suggestion: + norm = normalize_code(das.cim10_suggestion) + if norm not in codes_seen: + codes_seen[norm] = "DAS établissement" + + # Codes de l'UCR (CPAM) + for field, role in [ + (controle.dp_ucr, "DP proposé UCR"), + (controle.da_ucr, "DA proposé UCR"), + (controle.dr_ucr, "DR proposé UCR"), + ]: + if not field: + continue + for raw in re.split(r"[,;\s]+", field.strip()): + raw = raw.strip() + if not raw: + continue + norm = normalize_code(raw) + if norm not in codes_seen: + codes_seen[norm] = role + + if not codes_seen: + return "" + + # Résoudre les libellés + lines = [] + for norm_code, role in codes_seen.items(): + is_valid, label = validate_code(norm_code) + if is_valid and label: + lines.append(f" {norm_code} — {label} [{role}]") + else: + lines.append(f" {norm_code} — (code non trouvé dans le dictionnaire) [{role}]") + + if not lines: + return "" + + return ( + "\nDÉFINITIONS CIM-10 — RÉFÉRENCE (source : dictionnaire officiel) :\n" + + "\n".join(lines) + ) + + +def _build_tagged_context(dossier: DossierMedical) -> tuple[str, dict[str, str]]: + """Construit un contexte clinique avec des tags de référence pour le grounding. + + Chaque élément clinique reçoit un tag unique ([BIO-1], [IMG-1], [TRT-1], [ACTE-1]) + que le LLM doit citer dans ses preuves pour garantir la traçabilité. + + Returns: + (texte tagué pour injection dans le prompt, dict tag → contenu original) + """ + tag_map: dict[str, str] = {} + lines: list[str] = [] + + # Biologie (avec normes de référence pour éviter les hallucinations) + for i, b in enumerate(dossier.biologie_cle, 1): + if not b.valeur: + continue + tag = f"BIO-{i}" + # Interpréter la valeur par rapport aux normes connues + norm_info = "" + if b.test in BIO_NORMALS: + lo, hi = BIO_NORMALS[b.test] + try: + val = float(b.valeur.replace(",", ".").split()[0]) + if val > hi: + norm_info = f" — ÉLEVÉ (norme {lo}-{hi})" + elif val < lo: + norm_info = f" — BAS (norme {lo}-{hi})" + else: + norm_info = f" — NORMAL (norme {lo}-{hi})" + except (ValueError, AttributeError): + pass + content = f"{b.test}: {b.valeur}{norm_info}" + tag_map[tag] = content + lines.append(f" [{tag}] {content}") + + # Imagerie + for i, im in enumerate(dossier.imagerie, 1): + tag = f"IMG-{i}" + conclusion = f" — {im.conclusion}" if im.conclusion else "" + content = f"{im.type}{conclusion}" + tag_map[tag] = content + lines.append(f" [{tag}] {content}") + + # Traitements + for i, t in enumerate(dossier.traitements_sortie[:10], 1): + tag = f"TRT-{i}" + posologie = f" {t.posologie}" if t.posologie else "" + content = f"{t.medicament}{posologie}" + tag_map[tag] = content + lines.append(f" [{tag}] {content}") + + # Actes CCAM + for i, a in enumerate(dossier.actes_ccam, 1): + tag = f"ACTE-{i}" + code = f" ({a.code_ccam_suggestion})" if a.code_ccam_suggestion else "" + content = f"{a.texte}{code}" + tag_map[tag] = content + lines.append(f" [{tag}] {content}") + + if not lines: + return "", tag_map + + text = "ÉLÉMENTS CLINIQUES RÉFÉRENCÉS (cite le tag [XX-N] dans tes preuves) :\n" + "\n".join(lines) + return text, tag_map + + +def _validate_grounding(response_data: dict, tag_map: dict[str, str]) -> list[str]: + """Vérifie que les références dans preuves_dossier correspondent à des tags existants. + + Returns: + Liste de warnings pour les références inventées. + """ + if not tag_map: + return [] + + warnings: list[str] = [] + preuves = response_data.get("preuves_dossier") + if not preuves or not isinstance(preuves, list): + return warnings + + for p in preuves: + if not isinstance(p, dict): + continue + ref = p.get("ref", "") + if not ref: + continue + if ref not in tag_map: + valeur = p.get("valeur", "?") + warnings.append(f"Preuve [{ref}] non traçable (« {valeur} »)") + logger.warning("Grounding : preuve [%s] introuvable dans les tags du dossier", ref) + + return warnings + + +def _check_das_bio_coherence(dossier: DossierMedical) -> list[str]: + """Vérifie la cohérence entre les textes DAS et les valeurs biologiques. + + Détecte les contradictions comme "leucocytose" dans un DAS alors que + les leucocytes sont bas, ou "anémie" alors que l'hémoglobine est normale. + + Returns: + Liste de warnings pour les incohérences détectées. + """ + if not dossier.diagnostics_associes or not dossier.biologie_cle: + return [] + + # Patterns DAS → (test bio attendu, direction attendue) + _DAS_BIO_CHECKS: dict[str, tuple[str, str]] = { + "leucocytose": ("Leucocytes", "high"), + "leucopénie": ("Leucocytes", "low"), + "leucopenie": ("Leucocytes", "low"), + "thrombocytose": ("Plaquettes", "high"), + "thrombocytopénie": ("Plaquettes", "low"), + "thrombocytopenie": ("Plaquettes", "low"), + "thrombopénie": ("Plaquettes", "low"), + "thrombopenie": ("Plaquettes", "low"), + "anémie": ("Hémoglobine", "low"), + "anemie": ("Hémoglobine", "low"), + "polyglobulie": ("Hémoglobine", "high"), + "hyperkaliémie": ("Potassium", "high"), + "hypokaliémie": ("Potassium", "low"), + } + + # Indexer les valeurs bio disponibles + bio_values: dict[str, float] = {} + for b in dossier.biologie_cle: + if b.test and b.valeur: + try: + bio_values[b.test] = float(b.valeur.replace(",", ".").split()[0]) + except (ValueError, AttributeError): + pass + + warnings: list[str] = [] + for das in dossier.diagnostics_associes: + texte_lower = (das.texte or "").lower() + for pattern, (bio_test, direction) in _DAS_BIO_CHECKS.items(): + if pattern not in texte_lower: + continue + if bio_test not in bio_values or bio_test not in BIO_NORMALS: + continue + val = bio_values[bio_test] + lo, hi = BIO_NORMALS[bio_test] + if direction == "high" and val <= hi: + warnings.append( + f"INCOHÉRENCE : DAS « {das.texte} » ({das.cim10_suggestion or '?'}) " + f"mais {bio_test} = {val} est NORMAL (norme {lo}-{hi})" + ) + elif direction == "low" and val >= lo: + warnings.append( + f"INCOHÉRENCE : DAS « {das.texte} » ({das.cim10_suggestion or '?'}) " + f"mais {bio_test} = {val} est NORMAL (norme {lo}-{hi})" + ) + + if warnings: + for w in warnings: + logger.warning(" DAS/bio : %s", w) + + return warnings + + def _build_cpam_prompt( dossier: DossierMedical, controle: ControleCPAM, sources: list[dict], -) -> str: - """Construit le prompt pour la contre-argumentation CPAM.""" + extraction: dict | None = None, +) -> tuple[str, dict[str, str]]: + """Construit le prompt pour la contre-argumentation CPAM. + + Args: + extraction: Résultat optionnel de la passe 1 (extraction structurée). + + Returns: + (prompt texte, tag_map pour validation grounding) + """ # Résumé du dossier médical dossier_lines = [] @@ -203,7 +434,19 @@ def _build_cpam_prompt( patient_info.append(sejour.sexe) if sejour.age is not None: patient_info.append(f"{sejour.age} ans") + if sejour.age < 18: + patient_info.append("(PÉDIATRIE — codage pédiatrique applicable)") + elif sejour.age >= 80: + patient_info.append("(patient âgé — comorbidités fréquentes)") dossier_lines.append(f"- Patient : {', '.join(patient_info)}") + if sejour.mode_entree: + mode_label = sejour.mode_entree + if "urgence" in mode_label.lower() or "urgent" in mode_label.lower(): + dossier_lines.append(f"- Mode d'entrée : {mode_label} (ADMISSION EN URGENCE)") + else: + dossier_lines.append(f"- Mode d'entrée : {mode_label}") + if sejour.mode_sortie: + dossier_lines.append(f"- Mode de sortie : {sejour.mode_sortie}") if sejour.imc is not None: dossier_lines.append(f"- IMC : {sejour.imc}") @@ -290,6 +533,30 @@ def _build_cpam_prompt( codes_contestes.append(f"Actes proposés par UCR : {controle.actes_ucr}") codes_str = "\n".join(codes_contestes) if codes_contestes else "Aucun code spécifique proposé" + # Définitions CIM-10 déterministes (tous les codes en jeu) + definitions_str = _get_cim10_definitions(dossier, controle) + + # Contexte clinique tagué pour le grounding + tagged_context, tag_map = _build_tagged_context(dossier) + if tagged_context: + tagged_str = f"\n\n{tagged_context}" + else: + tagged_str = ( + "\n\nATTENTION — DOSSIER PAUVRE EN ÉLÉMENTS CLINIQUES :\n" + "Aucune biologie, imagerie, traitement ou acte CCAM disponible.\n" + "Ne spécule PAS sur des éléments absents. Signale explicitement " + "le manque de données au lieu d'inventer des preuves." + ) + + # Vérification cohérence DAS / biologie + das_bio_warnings = _check_das_bio_coherence(dossier) + if das_bio_warnings: + tagged_str += ( + "\n\nALERTES COHÉRENCE DAS / BIOLOGIE (incohérences détectées dans le dossier) :\n" + + "\n".join(f" - {w}" for w in das_bio_warnings) + + "\n Prends en compte ces incohérences dans ton analyse." + ) + # Sources RAG sources_text = "" for i, src in enumerate(sources, 1): @@ -306,7 +573,36 @@ def _build_cpam_prompt( sources_text += f"--- Source {i}: {doc_name}{code_info}{page_info} ---\n" sources_text += (src.get("extrait", "")[:800]) + "\n\n" - return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en contentieux T2A. + # Section pré-analyse (résultat passe 1, si disponible) + extraction_str = "" + if extraction: + ext_lines = [] + comp = extraction.get("comprehension_contestation") + if comp: + ext_lines.append(f"Compréhension : {comp}") + elems = extraction.get("elements_cliniques_pertinents", []) + if elems and isinstance(elems, list): + elem_strs = [] + for e in elems: + if isinstance(e, dict): + elem_strs.append(f" - [{e.get('tag', '?')}] {e.get('pertinence', '')}") + if elem_strs: + ext_lines.append("Éléments pertinents :\n" + "\n".join(elem_strs)) + accords = extraction.get("points_accord_potentiels", []) + if accords and isinstance(accords, list): + ext_lines.append("Points d'accord potentiels : " + " ; ".join(str(a) for a in accords)) + codes = extraction.get("codes_en_jeu", {}) + if codes and isinstance(codes, dict): + diff = codes.get("difference_cle", "") + if diff: + ext_lines.append(f"Différence clé entre les codages : {diff}") + if ext_lines: + extraction_str = ( + "\nPRÉ-ANALYSE (extraction automatique — à utiliser comme base) :\n" + + "\n".join(ext_lines) + ) + + prompt = f"""Tu es un médecin DIM (Département d'Information Médicale) expert en contentieux T2A. Tu dois produire une analyse ÉQUILIBRÉE ET CRÉDIBLE de la contestation CPAM, puis contre-argumenter en mobilisant trois axes : médical, asymétrie d'information, et réglementaire. IMPORTANT — CRÉDIBILITÉ DE L'ANALYSE : @@ -320,6 +616,7 @@ Chaque argument doit désigner précisément quel code est défendu ou contesté DOSSIER MÉDICAL DE L'ÉTABLISSEMENT : {dossier_str} {asymetrie_str} +{tagged_str} OBJET DU DÉSACCORD : {controle.titre} @@ -330,12 +627,19 @@ DÉCISION UCR : {controle.decision_ucr} CODES CONTESTÉS : {codes_str} +{definitions_str} SOURCES RÉGLEMENTAIRES (Guide méthodologique, CIM-10) : {sources_text} +{extraction_str} CONSIGNES : +CONTEXTE CLINIQUE : +- Prends en compte l'ÂGE du patient (pédiatrie < 18 ans, personne âgée >= 80 ans), le MODE D'ENTRÉE (urgence vs programmé), et la DURÉE DE SÉJOUR pour contextualiser ton analyse +- En pédiatrie, les normes biologiques et les codages peuvent différer de l'adulte +- Une admission en urgence implique un contexte clinique aigu qui influence le choix du DP + ÉTAPE 1 — ANALYSE HONNÊTE (avant de contre-argumenter) : - Identifie ce que la CPAM a compris correctement dans le dossier - Reconnais les points où leur raisonnement est fondé, même partiellement @@ -343,9 +647,9 @@ CONSIGNES : AXE MÉDICAL : - Analyse le bien-fondé médical du codage de l'établissement -- CITE les éléments cliniques EXACTS du dossier : valeurs bio précises (ex: CRP 180 mg/L), résultats imagerie verbatim, traitements avec molécules et posologies +- CITE les éléments cliniques EXACTS du dossier en utilisant les tags [XX-N] fournis (ex: [BIO-1] CRP 180 mg/L) - Confronte l'argumentation CPAM aux sources CIM-10 et Guide Méthodologique fournies -- Ne mentionne que les éléments réellement présents dans le dossier fourni +- Ne mentionne AUCUN élément qui ne figure pas dans les éléments référencés ci-dessus AXE ASYMÉTRIE D'INFORMATION : - La CPAM a fondé son analyse uniquement sur le CRH et les codes transmis @@ -371,7 +675,7 @@ Réponds UNIQUEMENT avec un objet JSON au format suivant : "points_accord": "Points CONCRETS où la CPAM a raison ou partiellement raison (JAMAIS 'Aucun' — il y a toujours au moins un point légitime à reconnaître)", "contre_arguments_medicaux": "Argumentation médicale en faveur du codage, en expliquant pourquoi les points d'accord ne suffisent pas à invalider le codage", "preuves_dossier": [ - {{"element": "biologie|imagerie|traitement|acte|clinique", "valeur": "valeur exacte du dossier", "signification": "explication clinique"}} + {{"ref": "BIO-1", "element": "biologie|imagerie|traitement|acte|clinique", "valeur": "valeur exacte du dossier", "signification": "explication clinique"}} ], "contre_arguments_asymetrie": "Éléments cliniques que la CPAM n'avait pas et qui justifient le codage", "contre_arguments_reglementaires": "Erreurs d'interprétation réglementaire de la CPAM, avec citations verbatim des sources", @@ -380,6 +684,7 @@ Réponds UNIQUEMENT avec un objet JSON au format suivant : ], "conclusion": "Synthèse en citant EXPLICITEMENT les codes CIM-10 défendus (ex: DP Z45.80 — libellé) : points reconnus à la CPAM, puis pourquoi ce codage précis est néanmoins justifié" }}""" + return prompt, tag_map def _validate_references(parsed: dict, sources: list[dict]) -> list[str]: @@ -446,10 +751,12 @@ def _format_response(parsed: dict, ref_warnings: list[str] | None = None) -> str preuves_lines = [] for p in preuves: if isinstance(p, dict): + ref = p.get("ref", "") elem = p.get("element", "") valeur = p.get("valeur", "") signif = p.get("signification", "") - preuves_lines.append(f"- [{elem}] {valeur} → {signif}") + ref_prefix = f"[{ref}] " if ref else "" + preuves_lines.append(f"- {ref_prefix}[{elem}] {valeur} → {signif}") if preuves_lines: sections.append(f"PREUVES DU DOSSIER\n" + "\n".join(preuves_lines)) @@ -497,6 +804,166 @@ def _format_response(parsed: dict, ref_warnings: list[str] | None = None) -> str return "\n\n".join(sections) +def _validate_adversarial( + response_data: dict, + tag_map: dict[str, str], + controle: ControleCPAM, +) -> dict | None: + """Validation adversariale — vérifie la cohérence de la contre-argumentation. + + Un appel LLM de relecture critique vérifie : + 1. Les valeurs cliniques citées correspondent aux éléments tagués du dossier + 2. La conclusion est cohérente avec l'argumentation + 3. Les points d'accord ne contredisent pas la contre-argumentation + 4. Les codes CIM-10 cités sont cohérents + + Returns: + dict {"coherent": bool, "erreurs": list[str], "score_confiance": int} ou None si échec. + """ + import json as _json + + # Construire le résumé des éléments factuels disponibles + if tag_map: + factual_lines = "\n".join(f" [{tag}] {content}" for tag, content in tag_map.items()) + factual_section = f"ÉLÉMENTS FACTUELS DU DOSSIER :\n{factual_lines}" + else: + factual_section = "ÉLÉMENTS FACTUELS DU DOSSIER : aucun élément tagué disponible" + + # Sérialiser la réponse LLM de façon compacte + try: + response_json = _json.dumps(response_data, ensure_ascii=False, indent=None) + # Tronquer si trop long pour le prompt de validation + if len(response_json) > 3000: + response_json = response_json[:3000] + "..." + except (TypeError, ValueError): + logger.warning("Validation adversariale : impossible de sérialiser la réponse") + return None + + # Normes biologiques pour vérifier les interprétations + normes_lines = [] + for test, (lo, hi) in BIO_NORMALS.items(): + normes_lines.append(f" {test}: {lo}-{hi}") + normes_section = "NORMES BIOLOGIQUES DE RÉFÉRENCE :\n" + "\n".join(normes_lines) + + prompt = f"""Tu es un relecteur critique. Vérifie la cohérence de cette contre-argumentation CPAM. + +RÉPONSE GÉNÉRÉE : +{response_json} + +{factual_section} + +{normes_section} + +CODES CONTESTÉS : +{f"DP UCR : {controle.dp_ucr}" if controle.dp_ucr else ""} +{f"DA UCR : {controle.da_ucr}" if controle.da_ucr else ""} + +Vérifie STRICTEMENT : +1. Chaque valeur bio/imagerie/traitement citée dans les preuves existe dans les éléments factuels +2. Si une valeur bio est qualifiée de "élevée", "basse" ou "anormale", vérifie qu'elle est RÉELLEMENT hors normes selon les normes ci-dessus (ex: CRP 5 = NORMAL, pas élevé) +3. La conclusion est cohérente avec l'argumentation développée +4. Les points d'accord ne contredisent pas les contre-arguments +5. Les codes CIM-10 mentionnés dans la conclusion sont cohérents avec le reste + +Réponds UNIQUEMENT en JSON : +{{ + "coherent": true ou false, + "erreurs": ["description précise de chaque incohérence trouvée"], + "score_confiance": 0 à 10 +}}""" + + logger.debug(" Validation adversariale") + result = call_ollama(prompt, temperature=0.0, max_tokens=800) + if result is None: + result = call_anthropic(prompt, temperature=0.0, max_tokens=800) + if result is None: + logger.warning(" Validation adversariale échouée — LLM indisponible") + return None + + coherent = result.get("coherent", True) + erreurs = result.get("erreurs", []) + score = result.get("score_confiance", -1) + + if not coherent and erreurs: + logger.warning(" Validation adversariale : %d incohérence(s) détectée(s) (score %s/10)", + len(erreurs), score) + for e in erreurs: + logger.warning(" - %s", e) + else: + logger.info(" Validation adversariale OK (score %s/10)", score) + + return result + + +def _extraction_pass( + dossier: DossierMedical, + controle: ControleCPAM, +) -> dict | None: + """Passe 1 — Extraction structurée du contexte avant argumentation. + + Prompt court centré sur la compréhension de la contestation et l'extraction + des éléments cliniques pertinents. Pas de rédaction argumentative. + + Returns: + dict structuré ou None si le LLM échoue. + """ + # Résumé dossier compact + dp_str = "" + if dossier.diagnostic_principal: + dp = dossier.diagnostic_principal + code = f" ({dp.cim10_suggestion})" if dp.cim10_suggestion else "" + dp_str = f"{dp.texte}{code}" + elif controle.dp_ucr: + dp_str = f"code {controle.dp_ucr} (codé par l'établissement)" + + das_str = ", ".join( + f"{d.texte} ({d.cim10_suggestion})" if d.cim10_suggestion else d.texte + for d in dossier.diagnostics_associes + ) + + # Contexte tagué (réutilise la même fonction) + tagged_text, _ = _build_tagged_context(dossier) + + prompt = f"""Tu es un médecin DIM expert. Analyse cette contestation CPAM sans argumenter. + +DOSSIER : +- DP : {dp_str or "Non extrait"} +- DAS : {das_str or "Aucun"} +{tagged_text} + +CONTESTATION CPAM : +Titre : {controle.titre} +Argument : {controle.arg_ucr} +Décision : {controle.decision_ucr} +{f"DP proposé UCR : {controle.dp_ucr}" if controle.dp_ucr else ""} +{f"DA proposés UCR : {controle.da_ucr}" if controle.da_ucr else ""} + +Réponds UNIQUEMENT en JSON : +{{ + "comprehension_contestation": "Résumé factuel : que conteste la CPAM et pourquoi", + "elements_cliniques_pertinents": [ + {{"tag": "BIO-1 ou texte libre", "pertinence": "en quoi cet élément est pertinent pour le codage contesté"}} + ], + "points_accord_potentiels": ["points où la CPAM a partiellement raison"], + "codes_en_jeu": {{ + "dp_etablissement": "code + libellé", + "dp_ucr": "code + libellé si proposé", + "difference_cle": "explication de la différence entre les deux codages" + }} +}}""" + + logger.debug(" Passe 1 — extraction structurée") + result = call_ollama(prompt, temperature=0.0, max_tokens=1500) + if result is None: + result = call_anthropic(prompt, temperature=0.0, max_tokens=1500) + if result is not None: + logger.info(" Passe 1 OK : %d éléments cliniques extraits", + len(result.get("elements_cliniques_pertinents", []))) + else: + logger.warning(" Passe 1 échouée — fallback single-pass") + return result + + def generate_cpam_response( dossier: DossierMedical, controle: ControleCPAM, @@ -513,14 +980,17 @@ def generate_cpam_response( logger.info("CPAM : génération contre-argumentation pour OGC %d — %s", controle.numero_ogc, controle.titre) - # 1. Recherche RAG ciblée + # 1. Passe 1 — Extraction structurée (compréhension avant argumentation) + extraction = _extraction_pass(dossier, controle) + + # 2. Recherche RAG ciblée sources = _search_rag_for_control(controle, dossier) logger.info(" RAG : %d sources trouvées", len(sources)) - # 2. Construction du prompt - prompt = _build_cpam_prompt(dossier, controle, sources) + # 3. Construction du prompt (passe 2 — argumentation) + prompt, tag_map = _build_cpam_prompt(dossier, controle, sources, extraction) - # 3. Appel LLM — Ollama (modèle par défaut) > Haiku fallback + # 4. Appel LLM — Ollama (modèle par défaut) > Haiku fallback result = call_ollama(prompt, temperature=0.1, max_tokens=4000) if result is not None: logger.info(" Contre-argumentation via Ollama") @@ -530,7 +1000,7 @@ def generate_cpam_response( if result is not None: logger.info(" Contre-argumentation via Anthropic Haiku") - # 4. Conversion des sources RAG + # 5. Conversion des sources RAG rag_sources = [ RAGSource( document=s.get("document", ""), @@ -545,13 +1015,32 @@ def generate_cpam_response( logger.warning(" LLM non disponible — contre-argumentation non générée") return "", None, rag_sources - # 5. Validation des références + # 6. Validation des références RAG ref_warnings = _validate_references(result, sources) if ref_warnings: logger.warning(" CPAM : %d référence(s) non vérifiable(s)", len(ref_warnings)) - # 6. Formater la réponse - text = _format_response(result, ref_warnings) + # 7. Validation grounding (preuves traçables vers le dossier) + grounding_warnings = _validate_grounding(result, tag_map) + if grounding_warnings: + logger.warning(" CPAM : %d preuve(s) non traçable(s)", len(grounding_warnings)) + + # 8. Validation adversariale (cohérence factuelle) + adversarial_warnings: list[str] = [] + validation = _validate_adversarial(result, tag_map, controle) + if validation and not validation.get("coherent", True): + erreurs = validation.get("erreurs", []) + score = validation.get("score_confiance", "?") + for e in erreurs: + if isinstance(e, str) and e.strip(): + adversarial_warnings.append(f"Incohérence détectée : {e}") + if adversarial_warnings: + adversarial_warnings.append(f"Score de confiance : {score}/10") + + all_warnings = ref_warnings + grounding_warnings + adversarial_warnings + + # 9. Formater la réponse + text = _format_response(result, all_warnings) logger.info(" Contre-argumentation générée (%d caractères)", len(text)) return text, result, rag_sources diff --git a/tests/test_cpam_response.py b/tests/test_cpam_response.py index 3e99a39..8fdb7a6 100644 --- a/tests/test_cpam_response.py +++ b/tests/test_cpam_response.py @@ -17,9 +17,15 @@ from src.config import ( ) from src.control.cpam_response import ( _build_cpam_prompt, + _build_tagged_context, + _check_das_bio_coherence, + _extraction_pass, _format_response, + _get_cim10_definitions, _get_code_label, _search_rag_for_control, + _validate_adversarial, + _validate_grounding, _validate_references, generate_cpam_response, ) @@ -88,7 +94,7 @@ class TestBuildPrompt: def test_prompt_contains_dossier_info(self): dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "Cholécystite aiguë" in prompt assert "K81.0" in prompt @@ -98,7 +104,7 @@ class TestBuildPrompt: def test_prompt_contains_cpam_argument(self): dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert controle.arg_ucr in prompt assert controle.decision_ucr in prompt @@ -106,7 +112,7 @@ class TestBuildPrompt: def test_prompt_contains_codes_contestes(self): dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "DA proposés par UCR : K56.0" in prompt @@ -117,7 +123,7 @@ class TestBuildPrompt: {"document": "guide_methodo", "page": 64, "extrait": "Texte du guide..."}, {"document": "cim10", "code": "K56.0", "extrait": "Iléus paralytique..."}, ] - prompt = _build_cpam_prompt(dossier, controle, sources) + prompt, _ = _build_cpam_prompt(dossier, controle, sources) assert "Guide Méthodologique MCO 2026" in prompt assert "CIM-10 FR 2026" in prompt @@ -126,7 +132,7 @@ class TestBuildPrompt: def test_prompt_contains_three_axes(self): dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "AXE MÉDICAL" in prompt assert "AXE ASYMÉTRIE D'INFORMATION" in prompt @@ -135,7 +141,7 @@ class TestBuildPrompt: def test_prompt_contains_traitements_imagerie_when_present(self): dossier = _make_dossier_complet() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "Augmentin IV 3g/j" in prompt assert "Morphine SC" in prompt @@ -148,7 +154,7 @@ class TestBuildPrompt: def test_prompt_asymetrie_section_when_data_present(self): dossier = _make_dossier_complet() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "ÉLÉMENTS DU DOSSIER NON TRANSMIS À LA CPAM" in prompt assert "CRP: 180 mg/L (anormale)" in prompt @@ -162,14 +168,14 @@ class TestBuildPrompt: diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), ) controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "ÉLÉMENTS DU DOSSIER NON TRANSMIS À LA CPAM" not in prompt def test_prompt_json_format_new_fields(self): dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "contre_arguments_medicaux" in prompt assert "contre_arguments_asymetrie" in prompt @@ -179,7 +185,7 @@ class TestBuildPrompt: """Le prompt renforcé demande des preuves exactes.""" dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "CITE" in prompt assert "EXACTS" in prompt @@ -188,7 +194,7 @@ class TestBuildPrompt: """Le prompt interdit les références inventées.""" dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "INTERDICTION ABSOLUE" in prompt @@ -196,7 +202,7 @@ class TestBuildPrompt: """Le format JSON demandé inclut preuves_dossier.""" dossier = _make_dossier() controle = _make_controle() - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "preuves_dossier" in prompt @@ -206,7 +212,7 @@ class TestBuildPrompt: """Les codes contestés affichent le libellé CIM-10.""" dossier = _make_dossier() controle = _make_controle() # da_ucr="K56.0" - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "Iléus paralytique" in prompt assert "DA proposés par UCR" in prompt @@ -220,7 +226,7 @@ class TestBuildPrompt: numero_ogc=1, titre="Test", arg_ucr="Test", decision_ucr="Rejet", dp_ucr="Z99.9", da_ucr=None, ) - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "Z99.9" in prompt # Pas de crash @@ -239,7 +245,7 @@ class TestBuildPrompt: numero_ogc=1, titre="Désaccord DP", arg_ucr="Test", decision_ucr="Rejet", dp_ucr="Z45.8", da_ucr=None, ) - prompt = _build_cpam_prompt(dossier, controle, []) + prompt, _ = _build_cpam_prompt(dossier, controle, []) assert "codé par l'établissement" in prompt assert "contesté par la CPAM" in prompt @@ -395,16 +401,27 @@ class TestGenerateResponse: @patch("src.control.cpam_response.call_anthropic") @patch("src.control.cpam_response._search_rag_for_control") def test_generate_success_ollama_cpam(self, mock_rag, mock_anthropic, mock_ollama): - """Ollama disponible → utilisé en premier, retourne triplet.""" + """Ollama disponible → 3 passes (extraction + argumentation + validation).""" mock_rag.return_value = [ {"document": "guide_methodo", "page": 64, "extrait": "Texte guide"}, ] - mock_ollama.return_value = { - "analyse_contestation": "Analyse...", - "contre_arguments_medicaux": "Contre-arguments médicaux...", - "contre_arguments_asymetrie": "Asymétrie...", - "conclusion": "Conclusion...", - } + call_count = {"n": 0} + + def ollama_side_effect(prompt, temperature=0.1, max_tokens=4000): + call_count["n"] += 1 + if call_count["n"] == 1: + return {"comprehension_contestation": "Extraction...", "elements_cliniques_pertinents": [], "points_accord_potentiels": [], "codes_en_jeu": {}} + elif call_count["n"] == 2: + return { + "analyse_contestation": "Analyse...", + "contre_arguments_medicaux": "Contre-arguments médicaux...", + "contre_arguments_asymetrie": "Asymétrie...", + "conclusion": "Conclusion...", + } + else: + return {"coherent": True, "erreurs": [], "score_confiance": 9} + + mock_ollama.side_effect = ollama_side_effect dossier = _make_dossier() controle = _make_controle() @@ -414,26 +431,37 @@ class TestGenerateResponse: assert "Contre-arguments médicaux..." in text assert response_data is not None assert response_data["analyse_contestation"] == "Analyse..." - assert response_data["conclusion"] == "Conclusion..." assert len(sources) == 1 assert sources[0].document == "guide_methodo" - mock_ollama.assert_called_once() + # 3 appels Ollama : extraction + argumentation + validation + assert mock_ollama.call_count == 3 mock_anthropic.assert_not_called() @patch("src.control.cpam_response.call_ollama") @patch("src.control.cpam_response.call_anthropic") @patch("src.control.cpam_response._search_rag_for_control") def test_generate_fallback_haiku(self, mock_rag, mock_anthropic, mock_ollama): - """Ollama indisponible → fallback Haiku, retourne triplet.""" + """Ollama indisponible → fallback Haiku pour les 3 passes.""" mock_rag.return_value = [ {"document": "guide_methodo", "page": 64, "extrait": "Texte guide"}, ] mock_ollama.return_value = None - mock_anthropic.return_value = { - "analyse_contestation": "Analyse Haiku...", - "contre_arguments_medicaux": "Contre-args Haiku...", - "conclusion": "Conclusion Haiku...", - } + call_count = {"n": 0} + + def anthropic_side_effect(prompt, temperature=0.1, max_tokens=4000): + call_count["n"] += 1 + if call_count["n"] == 1: + return {"comprehension_contestation": "Extraction Haiku...", "elements_cliniques_pertinents": [], "points_accord_potentiels": [], "codes_en_jeu": {}} + elif call_count["n"] == 2: + return { + "analyse_contestation": "Analyse Haiku...", + "contre_arguments_medicaux": "Contre-args Haiku...", + "conclusion": "Conclusion Haiku...", + } + else: + return {"coherent": True, "erreurs": [], "score_confiance": 8} + + mock_anthropic.side_effect = anthropic_side_effect dossier = _make_dossier() controle = _make_controle() @@ -442,9 +470,10 @@ class TestGenerateResponse: assert "Contre-args Haiku..." in text assert response_data is not None - assert response_data["contre_arguments_medicaux"] == "Contre-args Haiku..." - mock_ollama.assert_called_once() - mock_anthropic.assert_called_once() + # Ollama appelé 3 fois mais retourne None + assert mock_ollama.call_count == 3 + # Anthropic appelé 3 fois en fallback + assert mock_anthropic.call_count == 3 @patch("src.control.cpam_response.call_ollama") @patch("src.control.cpam_response.call_anthropic") @@ -688,3 +717,573 @@ class TestSearchRagForControl: if "Iléus réflexe" in c[0][0] and "Cholécystite aiguë" in c[0][0] ] assert len(clinique_queries) >= 1 + + +class TestGetCim10Definitions: + """Tests pour l'injection déterministe des définitions CIM-10.""" + + @patch("src.control.cpam_response.validate_code") + @patch("src.control.cpam_response.normalize_code", side_effect=lambda c: c.upper()) + def test_definitions_injected_in_prompt(self, mock_norm, mock_valid): + """La section DÉFINITIONS CIM-10 apparaît dans le prompt avec les libellés.""" + mock_valid.side_effect = lambda c: { + "K81.0": (True, "Cholécystite aiguë"), + "K56.0": (True, "Iléus paralytique et obstruction intestinale"), + }.get(c, (False, "")) + + dossier = _make_dossier() # DP=K81.0, DAS=K56.0 + controle = _make_controle() # da_ucr="K56.0" + + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "DÉFINITIONS CIM-10" in prompt + assert "dictionnaire officiel" in prompt + assert "Cholécystite aiguë" in prompt + assert "Iléus paralytique" in prompt + assert "DP établissement" in prompt + + @patch("src.control.cpam_response.validate_code") + @patch("src.control.cpam_response.normalize_code", side_effect=lambda c: c.upper()) + def test_definitions_include_dp_and_ucr_codes(self, mock_norm, mock_valid): + """Les codes du dossier ET de l'UCR sont tous inclus.""" + mock_valid.side_effect = lambda c: { + "K81.0": (True, "Cholécystite aiguë"), + "K56.0": (True, "Iléus paralytique"), + "Z45.8": (True, "Ajustement d'un dispositif implantable"), + }.get(c, (False, "")) + + dossier = _make_dossier() # DP=K81.0, DAS=K56.0 + controle = ControleCPAM( + numero_ogc=1, titre="Test", arg_ucr="Test", + decision_ucr="Rejet", dp_ucr="Z45.8", da_ucr="K56.0", + ) + + result = _get_cim10_definitions(dossier, controle) + + # Codes dossier + assert "K81.0" in result + assert "DP établissement" in result + assert "K56.0" in result + # Codes UCR + assert "Z45.8" in result + assert "DP proposé UCR" in result + assert "DA proposé UCR" in result or "DAS établissement" in result + + @patch("src.control.cpam_response.validate_code", return_value=(False, "")) + @patch("src.control.cpam_response.normalize_code", side_effect=lambda c: c.upper()) + def test_definitions_graceful_when_code_unknown(self, mock_norm, mock_valid): + """Un code inconnu ne crashe pas, affiche un message explicite.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostic_principal=None, + ) + controle = ControleCPAM( + numero_ogc=1, titre="Test", arg_ucr="Test", + decision_ucr="Rejet", dp_ucr="Z99.9", da_ucr=None, + ) + + result = _get_cim10_definitions(dossier, controle) + + assert "Z99.9" in result + assert "non trouvé" in result + + def test_definitions_empty_when_no_codes(self): + """Aucun code → chaîne vide.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostic_principal=None, + ) + controle = ControleCPAM( + numero_ogc=1, titre="Test", arg_ucr="Test", + decision_ucr="Rejet", dp_ucr=None, da_ucr=None, + ) + + result = _get_cim10_definitions(dossier, controle) + + assert result == "" + + +class TestBuildTaggedContext: + """Tests pour le contexte clinique tagué (grounding).""" + + def test_tagged_context_bio_img_trt(self): + """Les tags BIO, IMG, TRT, ACTE sont correctement générés.""" + dossier = _make_dossier_complet() + text, tag_map = _build_tagged_context(dossier) + + assert "[BIO-1]" in text + assert "CRP" in text + assert "BIO-1" in tag_map + assert "[IMG-1]" in text + assert "Scanner abdominal" in text + assert "IMG-1" in tag_map + assert "[TRT-1]" in text + assert "Augmentin IV" in text + assert "TRT-1" in tag_map + assert "[ACTE-1]" in text + assert "Cholécystectomie" in text + assert "ACTE-1" in tag_map + + def test_tagged_context_bio_norms_annotated(self): + """Les valeurs bio sont annotées avec les normes de référence.""" + dossier = DossierMedical( + source_file="test.pdf", + biologie_cle=[ + BiologieCle(test="CRP", valeur="5", anomalie=False), + BiologieCle(test="CRP", valeur="180", anomalie=True), + BiologieCle(test="Hémoglobine", valeur="8.5", anomalie=True), + ], + ) + text, tag_map = _build_tagged_context(dossier) + + # CRP 5 = normal (norme 0-5) + assert "NORMAL" in tag_map.get("BIO-1", "") + # CRP 180 = élevé + assert "ÉLEVÉ" in tag_map.get("BIO-2", "") + # Hb 8.5 = bas (norme 12-17) + assert "BAS" in tag_map.get("BIO-3", "") + + def test_tagged_context_empty_dossier(self): + """Dossier sans données cliniques → texte vide, tag_map vide.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), + ) + text, tag_map = _build_tagged_context(dossier) + + assert text == "" + assert tag_map == {} + + def test_tagged_context_in_prompt(self): + """Le contexte tagué apparaît dans le prompt généré.""" + dossier = _make_dossier_complet() + controle = _make_controle() + prompt, tag_map = _build_cpam_prompt(dossier, controle, []) + + assert "ÉLÉMENTS CLINIQUES RÉFÉRENCÉS" in prompt + assert "[BIO-1]" in prompt + assert "[IMG-1]" in prompt + assert len(tag_map) > 0 + + def test_poor_dossier_warning_in_prompt(self): + """Dossier sans bio/imagerie → avertissement dans le prompt.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), + sejour=Sejour(sexe="M", age=70), + ) + controle = _make_controle() + prompt, tag_map = _build_cpam_prompt(dossier, controle, []) + + assert "DOSSIER PAUVRE" in prompt + assert "Ne spécule PAS" in prompt + assert len(tag_map) == 0 + + +class TestValidateGrounding: + """Tests pour la validation des preuves grounded.""" + + def test_grounding_valid_refs(self): + """Toutes les refs existent → 0 warnings.""" + tag_map = {"BIO-1": "CRP: 180 mg/L", "IMG-1": "Scanner abdominal"} + response_data = { + "preuves_dossier": [ + {"ref": "BIO-1", "element": "biologie", "valeur": "CRP 180 mg/L", "signification": "inflammation"}, + {"ref": "IMG-1", "element": "imagerie", "valeur": "Scanner", "signification": "confirme"}, + ] + } + warnings = _validate_grounding(response_data, tag_map) + assert len(warnings) == 0 + + def test_grounding_invented_ref(self): + """Ref inventée [BIO-99] → warning détecté.""" + tag_map = {"BIO-1": "CRP: 180 mg/L"} + response_data = { + "preuves_dossier": [ + {"ref": "BIO-99", "element": "biologie", "valeur": "Albumine 15 g/L", "signification": "inventé"}, + ] + } + warnings = _validate_grounding(response_data, tag_map) + assert len(warnings) == 1 + assert "BIO-99" in warnings[0] + + def test_grounding_no_tag_map_no_validation(self): + """Pas de tag_map (dossier vide) → pas de validation.""" + response_data = { + "preuves_dossier": [ + {"ref": "BIO-1", "element": "biologie", "valeur": "test", "signification": "test"}, + ] + } + warnings = _validate_grounding(response_data, {}) + assert len(warnings) == 0 + + def test_grounding_no_ref_field_ok(self): + """Preuves sans champ ref (ancien format) → pas de warning.""" + tag_map = {"BIO-1": "CRP: 180 mg/L"} + response_data = { + "preuves_dossier": [ + {"element": "biologie", "valeur": "CRP 180 mg/L", "signification": "inflammation"}, + ] + } + warnings = _validate_grounding(response_data, tag_map) + assert len(warnings) == 0 + + def test_format_response_with_ref(self): + """Le formatage inclut le tag ref dans les preuves.""" + parsed = { + "contre_arguments_medicaux": "Arguments...", + "preuves_dossier": [ + {"ref": "BIO-1", "element": "biologie", "valeur": "CRP 180 mg/L", "signification": "inflammation"}, + ], + "conclusion": "Conclusion...", + } + text = _format_response(parsed) + + assert "[BIO-1]" in text + assert "[biologie]" in text + assert "CRP 180 mg/L" in text + + +class TestCheckDasBioCoherence: + """Tests pour la vérification cohérence DAS / biologie.""" + + def test_leucocytose_with_low_leucocytes(self): + """DAS 'leucocytose' mais leucocytes bas → incohérence détectée.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostics_associes=[ + Diagnostic(texte="Leucocytose", cim10_suggestion="D72.8"), + ], + biologie_cle=[ + BiologieCle(test="Leucocytes", valeur="3", anomalie=True), + ], + ) + warnings = _check_das_bio_coherence(dossier) + + assert len(warnings) == 1 + assert "Leucocytose" in warnings[0] + assert "NORMAL" in warnings[0] + + def test_anemie_with_normal_hb(self): + """DAS 'anémie' mais Hb normale → incohérence détectée.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostics_associes=[ + Diagnostic(texte="Anémie ferriprive", cim10_suggestion="D50.9"), + ], + biologie_cle=[ + BiologieCle(test="Hémoglobine", valeur="14.5", anomalie=False), + ], + ) + warnings = _check_das_bio_coherence(dossier) + + assert len(warnings) == 1 + assert "anémie" in warnings[0].lower() or "Anémie" in warnings[0] + + def test_coherent_das_bio_no_warnings(self): + """DAS 'anémie' avec Hb basse → pas d'incohérence.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostics_associes=[ + Diagnostic(texte="Anémie", cim10_suggestion="D64.9"), + ], + biologie_cle=[ + BiologieCle(test="Hémoglobine", valeur="8.5", anomalie=True), + ], + ) + warnings = _check_das_bio_coherence(dossier) + + assert len(warnings) == 0 + + def test_no_bio_no_crash(self): + """Pas de biologie → pas de crash, pas de warnings.""" + dossier = DossierMedical( + source_file="test.pdf", + diagnostics_associes=[ + Diagnostic(texte="Leucocytose", cim10_suggestion="D72.8"), + ], + ) + warnings = _check_das_bio_coherence(dossier) + + assert len(warnings) == 0 + + def test_coherence_warnings_in_prompt(self): + """Les incohérences DAS/bio apparaissent dans le prompt.""" + dossier = DossierMedical( + source_file="test.pdf", + sejour=Sejour(sexe="M", age=65), + diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), + diagnostics_associes=[ + Diagnostic(texte="Thrombocytose", cim10_suggestion="D75.9"), + ], + biologie_cle=[ + BiologieCle(test="Plaquettes", valeur="200", anomalie=False), + ], + ) + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "ALERTES COHÉRENCE DAS / BIOLOGIE" in prompt + assert "Thrombocytose" in prompt + assert "NORMAL" in prompt + + +class TestPatientContext: + """Tests pour le contexte patient dans le prompt.""" + + def test_pediatric_flag(self): + """Patient < 18 ans → mention pédiatrie dans le prompt.""" + dossier = DossierMedical( + source_file="test.pdf", + sejour=Sejour(sexe="F", age=9), + diagnostic_principal=Diagnostic(texte="Appendicite", cim10_suggestion="K35.8"), + ) + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "PÉDIATRIE" in prompt + assert "9 ans" in prompt + + def test_elderly_flag(self): + """Patient >= 80 ans → mention patient âgé.""" + dossier = DossierMedical( + source_file="test.pdf", + sejour=Sejour(sexe="M", age=85), + diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), + ) + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "patient âgé" in prompt + assert "85 ans" in prompt + + def test_emergency_admission(self): + """Admission en urgence → flag dans le prompt.""" + dossier = DossierMedical( + source_file="test.pdf", + sejour=Sejour(sexe="M", age=50, mode_entree="Autres admissions urgentes"), + diagnostic_principal=Diagnostic(texte="Test", cim10_suggestion="Z00"), + ) + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "ADMISSION EN URGENCE" in prompt + + def test_context_consigne_in_prompt(self): + """Le prompt contient une consigne sur le contexte clinique.""" + dossier = _make_dossier() + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, []) + + assert "CONTEXTE CLINIQUE" in prompt + assert "ÂGE" in prompt + assert "MODE D'ENTRÉE" in prompt + + +class TestExtractionPass: + """Tests pour la passe 1 — extraction structurée.""" + + @patch("src.control.cpam_response.call_ollama") + def test_extraction_pass_returns_structured_json(self, mock_ollama): + """Passe 1 retourne les champs attendus.""" + mock_ollama.return_value = { + "comprehension_contestation": "La CPAM conteste le DAS K56.0", + "elements_cliniques_pertinents": [ + {"tag": "BIO-1", "pertinence": "CRP élevée confirme inflammation"} + ], + "points_accord_potentiels": ["Le CRH est succinct"], + "codes_en_jeu": { + "dp_etablissement": "K81.0 — Cholécystite aiguë", + "dp_ucr": "", + "difference_cle": "contestation porte sur le DAS, pas le DP", + }, + } + + dossier = _make_dossier() + controle = _make_controle() + result = _extraction_pass(dossier, controle) + + assert result is not None + assert "comprehension_contestation" in result + assert len(result["elements_cliniques_pertinents"]) == 1 + mock_ollama.assert_called_once() + + @patch("src.control.cpam_response.call_anthropic", return_value=None) + @patch("src.control.cpam_response.call_ollama", return_value=None) + def test_extraction_pass_failure_returns_none(self, mock_ollama, mock_anthropic): + """Passe 1 échoue → retourne None (fallback single-pass).""" + dossier = _make_dossier() + controle = _make_controle() + result = _extraction_pass(dossier, controle) + + assert result is None + + @patch("src.control.cpam_response.call_ollama") + def test_extraction_injected_in_prompt(self, mock_ollama): + """Le résultat de passe 1 est injecté dans le prompt de passe 2.""" + extraction = { + "comprehension_contestation": "La CPAM conteste le DAS K56.0", + "elements_cliniques_pertinents": [ + {"tag": "BIO-1", "pertinence": "CRP élevée"} + ], + "points_accord_potentiels": ["Le CRH est succinct"], + "codes_en_jeu": { + "difference_cle": "contestation porte sur le DAS", + }, + } + + dossier = _make_dossier() + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, [], extraction) + + assert "PRÉ-ANALYSE" in prompt + assert "La CPAM conteste le DAS K56.0" in prompt + assert "CRP élevée" in prompt + assert "contestation porte sur le DAS" in prompt + + def test_prompt_without_extraction(self): + """Sans extraction, pas de section PRÉ-ANALYSE.""" + dossier = _make_dossier() + controle = _make_controle() + prompt, _ = _build_cpam_prompt(dossier, controle, [], None) + + assert "PRÉ-ANALYSE" not in prompt + + @patch("src.control.cpam_response.call_ollama") + @patch("src.control.cpam_response._search_rag_for_control") + def test_generate_calls_three_passes(self, mock_rag, mock_ollama): + """L'orchestrateur appelle extraction + argumentation + validation.""" + call_count = {"n": 0} + + def ollama_side_effect(prompt, temperature=0.1, max_tokens=4000): + call_count["n"] += 1 + if call_count["n"] == 1: + return { + "comprehension_contestation": "Contestation DAS", + "elements_cliniques_pertinents": [], + "points_accord_potentiels": [], + "codes_en_jeu": {}, + } + elif call_count["n"] == 2: + return { + "analyse_contestation": "Analyse...", + "contre_arguments_medicaux": "Arguments...", + "conclusion": "Conclusion...", + } + else: + return {"coherent": True, "erreurs": [], "score_confiance": 9} + + mock_ollama.side_effect = ollama_side_effect + mock_rag.return_value = [] + + dossier = _make_dossier() + controle = _make_controle() + text, response_data, sources = generate_cpam_response(dossier, controle) + + # 3 appels Ollama : extraction + argumentation + validation + assert mock_ollama.call_count == 3 + assert response_data is not None + assert "Arguments..." in text + + +class TestValidateAdversarial: + """Tests pour la validation adversariale.""" + + @patch("src.control.cpam_response.call_ollama") + def test_coherent_response_no_warnings(self, mock_ollama): + """Réponse cohérente → coherent=true, pas de warnings dans le texte.""" + mock_ollama.return_value = {"coherent": True, "erreurs": [], "score_confiance": 9} + + tag_map = {"BIO-1": "CRP: 180 mg/L"} + response_data = { + "analyse_contestation": "Analyse...", + "preuves_dossier": [{"ref": "BIO-1", "valeur": "CRP 180 mg/L"}], + "conclusion": "Conclusion...", + } + controle = _make_controle() + + result = _validate_adversarial(response_data, tag_map, controle) + + assert result is not None + assert result["coherent"] is True + assert len(result["erreurs"]) == 0 + + @patch("src.control.cpam_response.call_ollama") + def test_hallucinated_bio_detected(self, mock_ollama): + """Valeur bio halluccinée → coherent=false avec erreur.""" + mock_ollama.return_value = { + "coherent": False, + "erreurs": ["CRP citée à 250 mg/L mais le dossier indique 180 mg/L"], + "score_confiance": 3, + } + + tag_map = {"BIO-1": "CRP: 180 mg/L"} + response_data = { + "preuves_dossier": [{"ref": "BIO-1", "valeur": "CRP 250 mg/L"}], + "conclusion": "Conclusion...", + } + controle = _make_controle() + + result = _validate_adversarial(response_data, tag_map, controle) + + assert result is not None + assert result["coherent"] is False + assert len(result["erreurs"]) == 1 + assert "CRP" in result["erreurs"][0] + + @patch("src.control.cpam_response.call_anthropic", return_value=None) + @patch("src.control.cpam_response.call_ollama", return_value=None) + def test_adversarial_failure_graceful(self, mock_ollama, mock_anthropic): + """LLM indisponible → retourne None, pas de crash.""" + tag_map = {"BIO-1": "CRP: 180 mg/L"} + response_data = {"conclusion": "Conclusion..."} + controle = _make_controle() + + result = _validate_adversarial(response_data, tag_map, controle) + + assert result is None + + @patch("src.control.cpam_response.call_ollama") + @patch("src.control.cpam_response._search_rag_for_control") + def test_adversarial_warnings_in_output(self, mock_rag, mock_ollama): + """Incohérences détectées → avertissements dans le texte formaté.""" + call_count = {"n": 0} + + def ollama_side_effect(prompt, temperature=0.1, max_tokens=4000): + call_count["n"] += 1 + if call_count["n"] == 1: + return {"comprehension_contestation": "Extraction", "elements_cliniques_pertinents": [], "points_accord_potentiels": [], "codes_en_jeu": {}} + elif call_count["n"] == 2: + return { + "analyse_contestation": "Analyse...", + "contre_arguments_medicaux": "Arguments...", + "conclusion": "Conclusion...", + } + else: + return { + "coherent": False, + "erreurs": ["Antibiotiques mentionnés mais absents du dossier"], + "score_confiance": 4, + } + + mock_ollama.side_effect = ollama_side_effect + mock_rag.return_value = [] + + dossier = _make_dossier() + controle = _make_controle() + text, response_data, sources = generate_cpam_response(dossier, controle) + + assert "Antibiotiques mentionnés" in text + assert "Score de confiance" in text + + def test_adversarial_empty_tag_map(self): + """Dossier sans tags → validation fonctionne quand même.""" + with patch("src.control.cpam_response.call_ollama") as mock_ollama: + mock_ollama.return_value = {"coherent": True, "erreurs": [], "score_confiance": 7} + + result = _validate_adversarial( + {"conclusion": "Test"}, {}, _make_controle() + ) + + assert result is not None + assert result["coherent"] is True