"""Construction du contexte et du prompt pour la contre-argumentation CPAM.""" from __future__ import annotations import logging import re from ..config import ControleCPAM, DossierMedical from ..medical.bio_normals import BIO_NORMALS from ..medical.cim10_dict import normalize_code, validate_code from ..prompts import CPAM_ARGUMENTATION logger = logging.getLogger(__name__) def _get_code_label(code_str: str) -> str: """Résout le libellé CIM-10 pour un ou plusieurs codes.""" codes = re.split(r"[,;\s]+", code_str.strip()) labels = [] for raw in codes: raw = raw.strip() if not raw: continue norm = normalize_code(raw) is_valid, label = validate_code(norm) if is_valid and label: labels.append(f"{norm} — {label}") else: labels.append(norm) if not labels: return "" if len(labels) == 1: parts = labels[0].split(" — ", 1) return f" — {parts[1]}" if len(parts) > 1 else "" 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 # Interprétations cliniques pour le résumé bio déterministe _BIO_INTERPRETATION: dict[str, dict[str, str]] = { # --- Hépatique / digestif --- "Lipasémie": {"high": "pancréatite probable", "low": "normal", "normal": "pas de pancréatite"}, "ASAT": {"high": "cytolyse hépatique", "low": "normal", "normal": "pas de cytolyse"}, "ALAT": {"high": "cytolyse hépatique", "low": "normal", "normal": "pas de cytolyse"}, "GGT": {"high": "cholestase/atteinte hépatique", "low": "normal", "normal": "pas de cholestase"}, "PAL": {"high": "cholestase/atteinte osseuse", "low": "normal", "normal": "pas de cholestase"}, "Bilirubine totale": {"high": "ictère/cholestase", "low": "normal", "normal": "pas d'ictère"}, "Bilirubine directe": {"high": "cholestase/obstruction biliaire", "low": "normal", "normal": "pas de cholestase"}, "LDH": {"high": "cytolyse/hémolyse", "low": "normal", "normal": "pas de cytolyse"}, # --- Inflammatoire --- "CRP": {"high": "infection/inflammation active", "low": "normal", "normal": "pas d'inflammation"}, "VS": {"high": "inflammation", "low": "normal", "normal": "pas d'inflammation"}, # --- Ionogramme --- "Sodium": {"high": "hypernatrémie", "low": "hyponatrémie", "normal": "natrémie normale"}, "Potassium": {"high": "hyperkaliémie", "low": "hypokaliémie", "normal": "kaliémie normale"}, # --- Hématologie --- "Hémoglobine": {"high": "polyglobulie", "low": "anémie", "normal": "pas d'anémie"}, "Plaquettes": {"high": "thrombocytose", "low": "thrombopénie", "normal": "numération normale"}, "Leucocytes": {"high": "hyperleucocytose", "low": "leucopénie", "normal": "numération normale"}, "TP": {"high": "normal", "low": "insuffisance hépatique/CIVD", "normal": "coagulation normale"}, "TCA": {"high": "hypocoagulabilité", "low": "normal", "normal": "coagulation normale"}, "Ferritine": {"high": "surcharge en fer/inflammation", "low": "carence en fer", "normal": "réserves en fer normales"}, # --- Rénal --- "Créatinine": {"high": "insuffisance rénale", "low": "normal", "normal": "fonction rénale conservée"}, "Urée": {"high": "insuffisance rénale/catabolisme", "low": "normal", "normal": "fonction rénale conservée"}, # --- Cardiologie --- "Troponine": {"high": "nécrose myocardique (SCA/IDM)", "low": "normal", "normal": "pas de souffrance myocardique"}, "BNP": {"high": "insuffisance cardiaque", "low": "normal", "normal": "pas d'insuffisance cardiaque"}, "NT-proBNP": {"high": "insuffisance cardiaque", "low": "normal", "normal": "pas d'insuffisance cardiaque"}, "D-dimères": {"high": "activation coagulation (EP/TVP possible)", "low": "normal", "normal": "EP/TVP peu probable"}, "INR": {"high": "hypocoagulabilité/surdosage AVK", "low": "hypercoagulabilité", "normal": "coagulation normale"}, "Fibrinogène": {"high": "inflammation/risque thrombotique", "low": "CIVD/insuffisance hépatique", "normal": "normal"}, # --- Infectiologie --- "Procalcitonine": {"high": "infection bactérienne", "low": "normal", "normal": "pas d'infection bactérienne"}, "Lactate": {"high": "hypoperfusion/choc", "low": "normal", "normal": "pas d'hypoperfusion"}, # --- Métabolisme --- "Glycémie": {"high": "hyperglycémie/diabète", "low": "hypoglycémie", "normal": "glycémie normale"}, "HbA1c": {"high": "diabète mal équilibré", "low": "normal", "normal": "équilibre glycémique correct"}, "Albumine": {"high": "déshydratation", "low": "dénutrition/insuffisance hépatique", "normal": "état nutritionnel conservé"}, "Acide urique": {"high": "hyperuricémie/goutte", "low": "normal", "normal": "uricémie normale"}, # --- Thyroïde --- "TSH": {"high": "hypothyroïdie", "low": "hyperthyroïdie", "normal": "fonction thyroïdienne normale"}, } def _build_bio_summary(dossier: DossierMedical) -> str: """Construit un résumé biologique déterministe à injecter dans le prompt. Chaque valeur bio est interprétée contre BIO_NORMALS avec une conclusion non ambiguë que le LLM ne doit pas modifier. Returns: Texte formaté ou "" si aucune biologie exploitable. """ if not dossier.biologie_cle: return "" lines: list[str] = [] for b in dossier.biologie_cle: if not b.valeur or b.test not in BIO_NORMALS: continue try: val = float(b.valeur.replace(",", ".").split()[0]) except (ValueError, AttributeError): continue lo, hi = BIO_NORMALS[b.test] if val > hi: status = "ÉLEVÉ" interp_key = "high" elif val < lo: status = "BAS" interp_key = "low" else: status = "NORMAL" interp_key = "normal" interp = _BIO_INTERPRETATION.get(b.test, {}).get(interp_key, "") interp_str = f" — {interp}" if interp else "" lines.append(f" ✓ {b.test} = {b.valeur} → {status} (norme {lo}-{hi}){interp_str}") if not lines: return "" return ( "FAITS BIOLOGIQUES VÉRIFIÉS (NE PAS MODIFIER ces interprétations) :\n" + "\n".join(lines) + "\n\nRÈGLE STRICTE : si tu cites une valeur biologique, tu DOIS utiliser " "l'interprétation ci-dessus.\n" "Ne qualifie JAMAIS une valeur NORMAL comme pathologique, " "ni une valeur ÉLEVÉ/BAS comme normale." ) 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]] = { # Hématologie "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"), "carence en fer": ("Ferritine", "low"), "carence martiale": ("Ferritine", "low"), # Ionogramme "hyperkaliémie": ("Potassium", "high"), "hypokaliémie": ("Potassium", "low"), "hypernatrémie": ("Sodium", "high"), "hyponatrémie": ("Sodium", "low"), "hyponatremie": ("Sodium", "low"), # Rénal "insuffisance rénale": ("Créatinine", "high"), "insuffisance renale": ("Créatinine", "high"), # Digestif "pancréatite": ("Lipasémie", "high"), "pancreatite": ("Lipasémie", "high"), # Infectiologie "sepsis": ("CRP", "high"), "choc septique": ("Lactate", "high"), # Cardiologie "infarctus": ("Troponine", "high"), "syndrome coronarien": ("Troponine", "high"), "embolie pulmonaire": ("D-dimères", "high"), "insuffisance cardiaque": ("BNP", "high"), # Métabolisme / nutrition "dénutrition": ("Albumine", "low"), "denutrition": ("Albumine", "low"), "diabète": ("Glycémie", "high"), "diabete": ("Glycémie", "high"), "hyperuricémie": ("Acide urique", "high"), "goutte": ("Acide urique", "high"), # Thyroïde "hypothyroïdie": ("TSH", "high"), "hypothyroidie": ("TSH", "high"), "hyperthyroïdie": ("TSH", "low"), "hyperthyroidie": ("TSH", "low"), # Coagulation "civd": ("Fibrinogène", "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], 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 = [] if dossier.diagnostic_principal: dp = dossier.diagnostic_principal dp_code = f" ({dp.cim10_suggestion})" if dp.cim10_suggestion else "" dossier_lines.append(f"- DP : {dp.texte}{dp_code}") elif controle.dp_ucr: dp_label = _get_code_label(controle.dp_ucr) dossier_lines.append( f"- DP : code {controle.dp_ucr}{dp_label} " f"(codé par l'établissement, contesté par la CPAM)" ) if dossier.diagnostics_associes: das_parts = [] for das in dossier.diagnostics_associes: code = f" ({das.cim10_suggestion})" if das.cim10_suggestion else "" das_parts.append(f"{das.texte}{code}") dossier_lines.append(f"- DAS : {', '.join(das_parts)}") if dossier.actes_ccam: actes = [f"{a.texte} ({a.code_ccam_suggestion})" if a.code_ccam_suggestion else a.texte for a in dossier.actes_ccam] dossier_lines.append(f"- Actes CCAM : {', '.join(actes)}") sejour = dossier.sejour if sejour.duree_sejour is not None: dossier_lines.append(f"- Durée séjour : {sejour.duree_sejour} jours") if sejour.sexe or sejour.age is not None: patient_info = [] if sejour.sexe: 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}") if dossier.biologie_cle: bio = [f"{b.test}: {b.valeur}" for b in dossier.biologie_cle[:5] if b.valeur] if bio: dossier_lines.append(f"- Biologie clé : {', '.join(bio)}") if dossier.imagerie: img_parts = [] for im in dossier.imagerie: conclusion = f" — {im.conclusion}" if im.conclusion else "" img_parts.append(f"{im.type}{conclusion}") dossier_lines.append(f"- Imagerie : {', '.join(img_parts)}") if dossier.traitements_sortie: trt_parts = [] for t in dossier.traitements_sortie[:10]: posologie = f" {t.posologie}" if t.posologie else "" trt_parts.append(f"{t.medicament}{posologie}") dossier_lines.append(f"- Traitements de sortie : {', '.join(trt_parts)}") if dossier.antecedents: dossier_lines.append(f"- Antécédents : {', '.join(a.texte for a in dossier.antecedents[:10])}") if dossier.complications: dossier_lines.append(f"- Complications : {', '.join(c.texte for c in dossier.complications)}") dossier_str = "\n".join(dossier_lines) if dossier_lines else "Non disponible" # Section asymétrie : éléments que la CPAM n'avait pas asymetrie_lines = [] if dossier.biologie_cle: bio_details = [] for b in dossier.biologie_cle if len(dossier.biologie_cle) <= 10 else dossier.biologie_cle[:10]: anomalie = " (anormale)" if b.anomalie else "" if b.valeur: bio_details.append(f"{b.test}: {b.valeur}{anomalie}") if bio_details: asymetrie_lines.append(f"- Biologie : {', '.join(bio_details)}") if dossier.imagerie: img_details = [] for im in dossier.imagerie: conclusion = f" — {im.conclusion}" if im.conclusion else "" img_details.append(f"{im.type}{conclusion}") if img_details: asymetrie_lines.append(f"- Imagerie : {', '.join(img_details)}") if dossier.traitements_sortie: trt_details = [] for t in dossier.traitements_sortie[:10]: posologie = f" {t.posologie}" if t.posologie else "" trt_details.append(f"{t.medicament}{posologie}") if trt_details: asymetrie_lines.append(f"- Traitements : {', '.join(trt_details)}") if dossier.actes_ccam: actes_details = [ f"{a.texte} ({a.code_ccam_suggestion})" if a.code_ccam_suggestion else a.texte for a in dossier.actes_ccam ] if actes_details: asymetrie_lines.append(f"- Actes CCAM : {', '.join(actes_details)}") asymetrie_str = "" if asymetrie_lines: asymetrie_str = ( "\n\nÉLÉMENTS DU DOSSIER NON TRANSMIS À LA CPAM " "(l'UCR n'a eu que le CRH et les codes) :\n" + "\n".join(asymetrie_lines) ) # Codes contestés par la CPAM (avec libellés CIM-10 résolus) codes_contestes = [] if controle.dp_ucr: codes_contestes.append(f"DP proposé par UCR : {controle.dp_ucr}{_get_code_label(controle.dp_ucr)}") if controle.da_ucr: codes_contestes.append(f"DA proposés par UCR : {controle.da_ucr}{_get_code_label(controle.da_ucr)}") if controle.dr_ucr: codes_contestes.append(f"DR proposé par UCR : {controle.dr_ucr}{_get_code_label(controle.dr_ucr)}") if controle.actes_ucr: 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." ) # Résumé biologique déterministe (interprétations non modifiables par le LLM) bio_summary = _build_bio_summary(dossier) if bio_summary: tagged_str += f"\n\n{bio_summary}" # 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): doc_name = { "cim10": "CIM-10 FR 2026", "cim10_alpha": "CIM-10 Index Alphabétique 2026", "guide_methodo": "Guide Méthodologique MCO 2026", "ccam": "CCAM PMSI V4 2025", }.get(src.get("document", ""), src.get("document", "")) code_info = f" (code: {src['code']})" if src.get("code") else "" page_info = f" [page {src['page']}]" if src.get("page") else "" sources_text += f"--- Source {i}: {doc_name}{code_info}{page_info} ---\n" sources_text += (src.get("extrait", "")[:800]) + "\n\n" # 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 = CPAM_ARGUMENTATION.format( dossier_str=dossier_str, asymetrie_str=asymetrie_str, tagged_str=tagged_str, titre=controle.titre, arg_ucr=controle.arg_ucr, decision_ucr=controle.decision_ucr, codes_str=codes_str, definitions_str=definitions_str, sources_text=sources_text, extraction_str=extraction_str, ) return prompt, tag_map