feat: résumé clinique enrichi + preuves cliniques + validation QC batch

Améliore la qualité du codage CIM-10 sur 3 axes :
- Contexte clinique enrichi (interprétations bio, traitements indicatifs, marqueurs sévérité)
- Preuves cliniques structurées par diagnostic (evidence linking dans le prompt LLM)
- Validation batch post-codage (1 appel LLM/dossier, ajustement confiance, alertes QC)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dom
2026-02-17 21:47:27 +01:00
parent dbc5bdbaf4
commit 94fa4e5f3b
7 changed files with 988 additions and 16 deletions

View File

@@ -92,6 +92,12 @@ class Sejour(BaseModel):
taille: Optional[float] = None taille: Optional[float] = None
class PreuveClinique(BaseModel):
type: str # "biologie" | "imagerie" | "traitement" | "acte" | "clinique"
element: str # "CRP 180 mg/L"
interpretation: str # "syndrome inflammatoire majeur"
class Diagnostic(BaseModel): class Diagnostic(BaseModel):
texte: str texte: str
cim10_suggestion: Optional[str] = None cim10_suggestion: Optional[str] = None
@@ -99,6 +105,7 @@ class Diagnostic(BaseModel):
justification: Optional[str] = None justification: Optional[str] = None
raisonnement: Optional[str] = None raisonnement: Optional[str] = None
sources_rag: list[RAGSource] = Field(default_factory=list) sources_rag: list[RAGSource] = Field(default_factory=list)
preuves_cliniques: list[PreuveClinique] = Field(default_factory=list)
est_cma: Optional[bool] = None est_cma: Optional[bool] = None
est_cms: Optional[bool] = None est_cms: Optional[bool] = None
niveau_severite: Optional[str] = None # "leger" | "modere" | "severe" | "non_evalue" niveau_severite: Optional[str] = None # "leger" | "modere" | "severe" | "non_evalue"

View File

@@ -150,6 +150,10 @@ def extract_medical_info(
# Post-processing : retirer DAS dont le code est identique au DP # Post-processing : retirer DAS dont le code est identique au DP
_remove_das_equal_dp(dossier) _remove_das_equal_dp(dossier)
# Post-processing : validation justifications (QC batch)
if use_rag:
_validate_justifications(dossier)
# Post-processing : traçabilité source (page + extrait) # Post-processing : traçabilité source (page + extrait)
if page_tracker: if page_tracker:
_apply_source_tracking(dossier, page_tracker, search_text) _apply_source_tracking(dossier, page_tracker, search_text)
@@ -1019,3 +1023,110 @@ def _apply_source_tracking(dossier: DossierMedical, page_tracker, search_text: s
if tracked: if tracked:
logger.info(" Traçabilité source : %d/%d diagnostics localisés", tracked, len(all_diags)) logger.info(" Traçabilité source : %d/%d diagnostics localisés", tracked, len(all_diags))
def _validate_justifications(dossier: DossierMedical) -> None:
"""Validation croisée de tous les diagnostics via un appel LLM unique.
Vérifie la cohérence, les preuves cliniques et la spécificité des codes.
Ajuste la confiance si la justification est faible et ajoute des alertes QC.
"""
try:
from .ollama_client import call_ollama
from .clinical_context import build_enriched_context, format_enriched_context
except ImportError:
logger.warning("Module clinical_context non disponible pour la validation QC")
return
all_diags: list[tuple[str, Diagnostic]] = []
if dossier.diagnostic_principal:
all_diags.append(("DP", dossier.diagnostic_principal))
for das in dossier.diagnostics_associes:
all_diags.append(("DAS", das))
if not all_diags:
return
# Construire le résumé des codes à valider
codes_section = ""
for i, (type_diag, diag) in enumerate(all_diags, 1):
code = diag.cim10_suggestion or "?"
justif = (diag.justification or "")[:150]
preuves = ", ".join(p.element for p in diag.preuves_cliniques[:3]) or "aucune"
codes_section += f"{i}. [{type_diag}] {code}{diag.texte}\n"
codes_section += f" Justification: {justif}\n"
codes_section += f" Preuves: {preuves}\n\n"
ctx = build_enriched_context(dossier)
ctx_str = format_enriched_context(ctx)
prompt = f"""Tu es un médecin DIM contrôleur qualité PMSI.
Vérifie la cohérence et la justification de ce codage complet.
DOSSIER CLINIQUE :
{ctx_str}
CODAGE À VALIDER :
{codes_section}
Pour CHAQUE code, vérifie :
1. Existe-t-il une preuve clinique concrète dans le dossier ?
2. Le code est-il le plus spécifique possible ?
3. Y a-t-il des conflits ou redondances avec d'autres codes ?
Réponds avec un JSON :
{{
"validations": [
{{
"numero": 1,
"code": "X99.9",
"verdict": "maintenir|reclasser|supprimer",
"confidence_recommandee": "high|medium|low",
"commentaire": "explication courte"
}}
],
"alertes_globales": ["..."]
}}"""
try:
result = call_ollama(prompt, temperature=0.1, max_tokens=2500)
except Exception:
logger.warning("Erreur lors de l'appel Ollama pour validation QC", exc_info=True)
return
if result is None:
return
# Appliquer les ajustements
validations = result.get("validations", [])
for v in validations:
if not isinstance(v, dict):
continue
num = v.get("numero")
if not isinstance(num, int) or num < 1 or num > len(all_diags):
continue
type_diag, diag = all_diags[num - 1]
conf = v.get("confidence_recommandee")
verdict = v.get("verdict")
commentaire = v.get("commentaire", "")
if conf in ("high", "medium", "low") and conf != diag.cim10_confidence:
old = diag.cim10_confidence
diag.cim10_confidence = conf
if old and conf != old:
dossier.alertes_codage.append(
f"QC: {type_diag} {diag.cim10_suggestion} confiance {old}\u2192{conf} \u2014 {commentaire}"
)
if verdict == "supprimer" and type_diag == "DAS":
dossier.alertes_codage.append(
f"QC: DAS {diag.cim10_suggestion} ({diag.texte}) à reconsidérer \u2014 {commentaire}"
)
alertes_globales = result.get("alertes_globales", [])
for a in alertes_globales:
if isinstance(a, str) and a.strip():
dossier.alertes_codage.append(f"QC: {a}")
logger.info(" QC batch : %d validations, %d alertes globales",
len(validations), len(alertes_globales))

View File

@@ -0,0 +1,315 @@
"""Enrichissement du contexte clinique pour les prompts LLM.
Interprète les données brutes (biologie, traitements, séjour) en informations
cliniques structurées pour améliorer la qualité du codage CIM-10.
"""
from __future__ import annotations
from ..config import DossierMedical
from .cim10_extractor import BIO_NORMALS
# Seuils d'interprétation biologique (test → liste de (seuil, direction, interprétation))
# Ordre décroissant : le premier seuil franchi donne l'interprétation
BIO_INTERPRETATIONS: dict[str, list[tuple[float, str, str]]] = {
"CRP": [
(100, "high", "syndrome inflammatoire majeur"),
(20, "high", "syndrome inflammatoire modéré"),
(5, "high", "syndrome inflammatoire mineur"),
],
"Lipasémie": [
(180, "high", "pancréatite biologique (>3N)"),
(60, "high", "élévation modérée de la lipase"),
],
"ASAT": [
(200, "high", "cytolyse hépatique majeure (>5N)"),
(80, "high", "cytolyse hépatique modérée (>2N)"),
],
"ALAT": [
(200, "high", "cytolyse hépatique majeure (>5N)"),
(80, "high", "cytolyse hépatique modérée (>2N)"),
],
"Bilirubine totale": [
(50, "high", "ictère franc"),
(17, "high", "hyperbilirubinémie modérée"),
],
"Hémoglobine": [
(7, "low", "anémie sévère (transfusion probable)"),
(10, "low", "anémie modérée"),
],
"Créatinine": [
(300, "high", "insuffisance rénale sévère"),
(150, "high", "insuffisance rénale modérée"),
],
"Plaquettes": [
(50, "low", "thrombopénie sévère"),
(100, "low", "thrombopénie modérée"),
],
"Leucocytes": [
(20, "high", "hyperleucocytose majeure (infection, inflammation)"),
(2, "low", "leucopénie sévère (aplasie, immunodépression)"),
],
}
# Médicaments → condition implicite (clé en lowercase)
TREATMENT_INDICATORS: dict[str, str] = {
"insuline": "diabète insulino-traité",
"metformine": "diabète type 2",
"héparine": "anticoagulation (risque thromboembolique)",
"enoxaparine": "anticoagulation (HBPM)",
"lovenox": "anticoagulation (HBPM)",
"warfarine": "anticoagulation au long cours (AVK)",
"fluindione": "anticoagulation au long cours (AVK)",
"amoxicilline": "antibiothérapie",
"ceftriaxone": "antibiothérapie IV",
"tazocilline": "antibiothérapie large spectre IV",
"morphine": "analgésie palier 3 (douleur sévère)",
"oxycodone": "analgésie palier 3 (douleur sévère)",
"oxygène": "oxygénothérapie (insuffisance respiratoire)",
"furosémide": "insuffisance cardiaque / rétention hydrique",
"lasilix": "insuffisance cardiaque / rétention hydrique",
}
def interpret_bio_value(test: str, value_str: str, is_abnormal: bool | None) -> str | None:
"""Retourne l'interprétation clinique d'une valeur bio, ou None si normale."""
if test not in BIO_INTERPRETATIONS:
return None
try:
val = float(value_str.replace(",", ".").replace(" ", ""))
except (ValueError, AttributeError):
return None
# Si la valeur est normale (pas anormale), pas d'interprétation
if is_abnormal is False:
return None
thresholds = BIO_INTERPRETATIONS[test]
for seuil, direction, interpretation in thresholds:
if direction == "high" and val >= seuil:
return interpretation
if direction == "low" and val <= seuil:
return interpretation
return None
def detect_treatment_indicators(traitements: list) -> list[dict]:
"""Retourne les conditions implicites détectées via les traitements.
Args:
traitements: Liste d'objets Traitement ou de dicts avec clé 'medicament'.
Returns:
Liste de dicts {medicament, condition}.
"""
results = []
seen_conditions: set[str] = set()
for t in traitements:
med = t.medicament if hasattr(t, "medicament") else t.get("medicament", "")
med_lower = med.lower().strip()
for keyword, condition in TREATMENT_INDICATORS.items():
if keyword in med_lower and condition not in seen_conditions:
results.append({"medicament": med, "condition": condition})
seen_conditions.add(condition)
break
return results
def detect_severity_markers(dossier: DossierMedical) -> list[str]:
"""Détecte les marqueurs de sévérité globaux."""
markers = []
duree = dossier.sejour.duree_sejour
if duree is not None:
if duree > 14:
markers.append(f"séjour prolongé ({duree} jours)")
elif duree > 7:
markers.append(f"séjour >7 jours ({duree} jours)")
age = dossier.sejour.age
if age is not None:
if age >= 80:
markers.append(f"patient très âgé ({age} ans)")
elif age >= 70:
markers.append(f"patient âgé ({age} ans)")
imc = dossier.sejour.imc
if imc is not None:
if imc >= 40:
markers.append(f"obésité morbide (IMC {imc})")
elif imc >= 30:
markers.append(f"obésité (IMC {imc})")
if dossier.complications:
markers.append(f"{len(dossier.complications)} complication(s)")
return markers
def build_enriched_context(dossier: DossierMedical) -> dict:
"""Construit le contexte clinique enrichi (appel unique par dossier).
Returns:
Dict avec les clés : patient, duree_sejour, antecedents,
biologie (avec interprétations), imagerie, complications,
dp_texte, das_codes_existants, interpretations_bio,
conditions_traitements, marqueurs_severite.
"""
# Données de base (compatibles avec l'ancien format)
ctx: dict = {
"sexe": dossier.sejour.sexe,
"age": dossier.sejour.age,
"duree_sejour": dossier.sejour.duree_sejour,
"imc": dossier.sejour.imc,
"antecedents": dossier.antecedents[:5],
"biologie_cle": [(b.test, b.valeur, b.anomalie) for b in dossier.biologie_cle],
"imagerie": [(i.type, (i.conclusion or "")[:200]) for i in dossier.imagerie],
"complications": dossier.complications,
}
# Interprétations biologiques
interpretations = []
for b in dossier.biologie_cle:
interp = interpret_bio_value(b.test, b.valeur or "", b.anomalie)
if interp:
# Ajouter l'unité si connue
unit = ""
if b.test in ("CRP",):
unit = " mg/L"
elif b.test in ("Lipasémie", "ASAT", "ALAT", "GGT", "PAL"):
unit = " UI/L"
elif b.test in ("Bilirubine totale", "Créatinine"):
unit = " µmol/L"
elif b.test in ("Hémoglobine",):
unit = " g/dL"
elif b.test in ("Plaquettes", "Leucocytes"):
unit = " G/L"
interpretations.append({
"test": b.test,
"valeur": f"{b.valeur}{unit}",
"interpretation": interp,
})
ctx["interpretations_bio"] = interpretations
# Conditions implicites via traitements
ctx["conditions_traitements"] = detect_treatment_indicators(dossier.traitements_sortie)
# Marqueurs de sévérité
ctx["marqueurs_severite"] = detect_severity_markers(dossier)
return ctx
def format_enriched_context(context: dict) -> str:
"""Formate le contexte enrichi en texte structuré pour le prompt.
Inclut les mêmes sections que l'ancien _format_contexte() PLUS :
interprétations bio, conditions implicites traitements, marqueurs sévérité.
"""
lines = []
# Patient
sexe = context.get("sexe")
age = context.get("age")
imc = context.get("imc")
patient_parts = []
if sexe:
patient_parts.append(sexe)
if age:
patient_parts.append(f"{age} ans")
if imc:
patient_parts.append(f"IMC {imc}")
if patient_parts:
lines.append(f"- Patient : {', '.join(str(p) for p in patient_parts)}")
# Durée de séjour
duree = context.get("duree_sejour")
if duree:
lines.append(f"- Durée séjour : {duree} jours")
# Antécédents
antecedents = context.get("antecedents")
if antecedents:
lines.append(f"- Antécédents : {', '.join(antecedents[:5])}")
# Biologie (avec normes)
biologie = context.get("biologie_cle")
if biologie:
bio_parts = []
for b in biologie:
test, valeur, anomalie = (
b if isinstance(b, (list, tuple))
else (b.get("test"), b.get("valeur"), b.get("anomalie"))
)
norme_str = ""
if test in BIO_NORMALS:
lo, hi = BIO_NORMALS[test]
lo_s = int(lo) if lo == int(lo) else lo
hi_s = int(hi) if hi == int(hi) else hi
norme_str = f" [N: {lo_s}-{hi_s}]"
marker = " (\u2191)" if anomalie else ""
bio_parts.append(f"{test} {valeur}{norme_str}{marker}")
lines.append(f"- Biologie : {', '.join(bio_parts)}")
# Imagerie
imagerie = context.get("imagerie")
if imagerie:
for img in imagerie:
img_type, conclusion = (
img if isinstance(img, (list, tuple))
else (img.get("type"), img.get("conclusion"))
)
if conclusion:
lines.append(f"- Imagerie : {img_type}{conclusion[:200]}")
# Complications
complications = context.get("complications")
if complications:
lines.append(f"- Complications : {', '.join(complications)}")
# DP du séjour
dp_texte = context.get("dp_texte")
if dp_texte:
lines.append(f"- DP du séjour : {dp_texte}")
# DAS déjà codés
das_codes = context.get("das_codes_existants")
if das_codes:
lines.append(f"- DAS déjà codés : {', '.join(das_codes)}")
# --- Sections enrichies ---
# Interprétations biologiques
interpretations = context.get("interpretations_bio", [])
if interpretations:
interp_parts = [
f"{i['test']} {i['valeur']} \u2192 {i['interpretation']}"
for i in interpretations
]
lines.append(f"\nINTERPRÉTATION CLINIQUE :")
lines.append(f"- Biologie : {' ; '.join(interp_parts)}")
# Conditions implicites via traitements
conditions = context.get("conditions_traitements", [])
if conditions:
cond_parts = [
f"{c['medicament']} \u2192 {c['condition']}"
for c in conditions
]
if not interpretations:
lines.append(f"\nINTERPRÉTATION CLINIQUE :")
lines.append(f"- Traitements indicatifs : {' ; '.join(cond_parts)}")
# Marqueurs de sévérité
marqueurs = context.get("marqueurs_severite", [])
if marqueurs:
if not interpretations and not conditions:
lines.append(f"\nINTERPRÉTATION CLINIQUE :")
lines.append(f"- Marqueurs de sévérité : {', '.join(marqueurs)}")
return "\n".join(lines) if lines else "Non précisé"

View File

@@ -6,12 +6,13 @@ import logging
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
from ..config import ( from ..config import (
ActeCCAM, Diagnostic, DossierMedical, RAGSource, ActeCCAM, Diagnostic, DossierMedical, PreuveClinique, RAGSource,
OLLAMA_CACHE_PATH, OLLAMA_MAX_PARALLEL, OLLAMA_MODEL, OLLAMA_CACHE_PATH, OLLAMA_MAX_PARALLEL, OLLAMA_MODEL,
EMBEDDING_MODEL, RERANKER_MODEL, EMBEDDING_MODEL, RERANKER_MODEL,
) )
from .cim10_dict import normalize_code, validate_code as cim10_validate, fallback_parent_code from .cim10_dict import normalize_code, validate_code as cim10_validate, fallback_parent_code
from .cim10_extractor import BIO_NORMALS from .cim10_extractor import BIO_NORMALS
from .clinical_context import build_enriched_context, format_enriched_context
from .ccam_dict import validate_code as ccam_validate from .ccam_dict import validate_code as ccam_validate
from .ollama_client import call_ollama, parse_json_response from .ollama_client import call_ollama, parse_json_response
from .ollama_cache import OllamaCache from .ollama_cache import OllamaCache
@@ -347,7 +348,7 @@ def _build_prompt(texte: str, sources: list[dict], contexte: dict, est_dp: bool
sources_text += (src.get("extrait", "")[:800]) + "\n\n" sources_text += (src.get("extrait", "")[:800]) + "\n\n"
type_diag = "DP (diagnostic principal)" if est_dp else "DAS (diagnostic associé significatif)" type_diag = "DP (diagnostic principal)" if est_dp else "DAS (diagnostic associé significatif)"
ctx_str = _format_contexte(contexte) ctx_str = format_enriched_context(contexte)
return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage PMSI. return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage PMSI.
Tu dois coder le diagnostic suivant en respectant STRICTEMENT les règles de l'ATIH. Tu dois coder le diagnostic suivant en respectant STRICTEMENT les règles de l'ATIH.
@@ -377,7 +378,10 @@ Réponds UNIQUEMENT avec un objet JSON au format suivant, sans aucun texte avant
"regle_pmsi": "conformité aux règles PMSI pour un {type_diag} (guide méthodologique)", "regle_pmsi": "conformité aux règles PMSI pour un {type_diag} (guide méthodologique)",
"code": "X99.9", "code": "X99.9",
"confidence": "high ou medium ou low", "confidence": "high ou medium ou low",
"justification": "explication courte en français" "justification": "explication courte en français",
"preuves_cliniques": [
{{"type": "biologie|imagerie|traitement|acte|clinique", "element": "élément concret du dossier", "interpretation": "signification clinique justifiant le code"}}
]
}}""" }}"""
@@ -398,7 +402,7 @@ def _build_prompt_ccam(texte: str, sources: list[dict], contexte: dict) -> str:
sources_text += f"--- Source {i}: {doc_name}{code_info}{page_info} ---\n" sources_text += f"--- Source {i}: {doc_name}{code_info}{page_info} ---\n"
sources_text += (src.get("extrait", "")[:800]) + "\n\n" sources_text += (src.get("extrait", "")[:800]) + "\n\n"
ctx_str = _format_contexte(contexte) ctx_str = format_enriched_context(contexte)
return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage CCAM PMSI. return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage CCAM PMSI.
Tu dois coder l'acte chirurgical/médical suivant en respectant STRICTEMENT la nomenclature CCAM. Tu dois coder l'acte chirurgical/médical suivant en respectant STRICTEMENT la nomenclature CCAM.
@@ -498,6 +502,20 @@ def _apply_llm_result_diagnostic(diagnostic: Diagnostic, llm_result: dict) -> No
if raisonnement: if raisonnement:
diagnostic.raisonnement = raisonnement diagnostic.raisonnement = raisonnement
# Stocker les preuves cliniques
preuves = llm_result.get("preuves_cliniques", [])
if preuves and isinstance(preuves, list):
for p in preuves:
if isinstance(p, dict) and p.get("element"):
try:
diagnostic.preuves_cliniques.append(PreuveClinique(
type=p.get("type", "clinique"),
element=p["element"],
interpretation=p.get("interpretation", ""),
))
except Exception:
pass
def enrich_diagnostic( def enrich_diagnostic(
diagnostic: Diagnostic, diagnostic: Diagnostic,
@@ -621,7 +639,7 @@ def enrich_acte(acte: ActeCCAM, contexte: dict, cache: OllamaCache | None = None
def _build_prompt_das_extraction(text: str, contexte: dict, existing_das: list[str], dp_texte: str) -> str: def _build_prompt_das_extraction(text: str, contexte: dict, existing_das: list[str], dp_texte: str) -> str:
"""Construit le prompt pour l'extraction LLM de DAS supplémentaires.""" """Construit le prompt pour l'extraction LLM de DAS supplémentaires."""
ctx_str = _format_contexte(contexte) ctx_str = format_enriched_context(contexte)
existing_str = "\n".join(f"- {d}" for d in existing_das) if existing_das else "Aucun" existing_str = "\n".join(f"- {d}" for d in existing_das) if existing_das else "Aucun"
return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage PMSI. return f"""Tu es un médecin DIM (Département d'Information Médicale) expert en codage PMSI.
@@ -723,16 +741,7 @@ def enrich_dossier(dossier: DossierMedical) -> None:
""" """
cache = OllamaCache(OLLAMA_CACHE_PATH, OLLAMA_MODEL) cache = OllamaCache(OLLAMA_CACHE_PATH, OLLAMA_MODEL)
contexte = { contexte = build_enriched_context(dossier)
"sexe": dossier.sejour.sexe,
"age": dossier.sejour.age,
"duree_sejour": dossier.sejour.duree_sejour,
"imc": dossier.sejour.imc,
"antecedents": dossier.antecedents[:5],
"biologie_cle": [(b.test, b.valeur, b.anomalie) for b in dossier.biologie_cle],
"imagerie": [(i.type, (i.conclusion or "")[:200]) for i in dossier.imagerie],
"complications": dossier.complications,
}
# Phase 1 : DP seul (le contexte DAS en dépend) # Phase 1 : DP seul (le contexte DAS en dépend)
if dossier.diagnostic_principal: if dossier.diagnostic_principal:

View File

@@ -223,6 +223,16 @@
{% if dp.justification %} {% if dp.justification %}
<div style="margin-top:0.5rem;font-size:0.8rem;color:#475569;">{{ dp.justification }}</div> <div style="margin-top:0.5rem;font-size:0.8rem;color:#475569;">{{ dp.justification }}</div>
{% endif %} {% endif %}
{% if dp.preuves_cliniques %}
<details style="margin-top:0.5rem;">
<summary style="font-size:0.8rem;color:#0369a1;cursor:pointer;font-weight:600;">Preuves cliniques ({{ dp.preuves_cliniques|length }})</summary>
<ul style="margin:0.25rem 0 0 0;padding-left:1.2rem;font-size:0.8rem;">
{% for p in dp.preuves_cliniques %}
<li style="margin-bottom:0.15rem;"><span class="badge" style="background:#e0f2fe;color:#0369a1;font-size:0.7rem;">{{ p.type }}</span> {{ p.element }} <span style="color:#64748b;">&rarr; {{ p.interpretation }}</span></li>
{% endfor %}
</ul>
</details>
{% endif %}
{% if dp.raisonnement %} {% if dp.raisonnement %}
<details style="margin-top:0.5rem;"> <details style="margin-top:0.5rem;">
<summary>Raisonnement LLM</summary> <summary>Raisonnement LLM</summary>
@@ -275,7 +285,18 @@
</details> </details>
{% endif %} {% endif %}
</td> </td>
<td style="font-size:0.8rem;color:#475569;">{{ das.justification or '' }}</td> <td style="font-size:0.8rem;color:#475569;">
{{ das.justification or '' }}
{% if das.preuves_cliniques %}
<details style="margin-top:0.3rem;"><summary style="font-size:0.7rem;color:#0369a1;cursor:pointer;">preuves ({{ das.preuves_cliniques|length }})</summary>
<ul style="margin:0.15rem 0 0 0;padding-left:1rem;font-size:0.75rem;">
{% for p in das.preuves_cliniques %}
<li><span style="font-weight:600;color:#0369a1;">[{{ p.type }}]</span> {{ p.element }} <span style="color:#64748b;">&rarr; {{ p.interpretation }}</span></li>
{% endfor %}
</ul>
</details>
{% endif %}
</td>
</tr> </tr>
{% if das.raisonnement %} {% if das.raisonnement %}
<tr> <tr>

View File

@@ -0,0 +1,264 @@
"""Tests pour le module d'enrichissement du contexte clinique."""
import pytest
from src.medical.clinical_context import (
BIO_INTERPRETATIONS,
TREATMENT_INDICATORS,
interpret_bio_value,
detect_treatment_indicators,
detect_severity_markers,
build_enriched_context,
format_enriched_context,
)
from src.config import (
BiologieCle,
Diagnostic,
DossierMedical,
Imagerie,
Sejour,
Traitement,
)
# --- interpret_bio_value ---
class TestInterpretBioValue:
def test_crp_major(self):
assert interpret_bio_value("CRP", "180", True) == "syndrome inflammatoire majeur"
def test_crp_moderate(self):
assert interpret_bio_value("CRP", "45", True) == "syndrome inflammatoire modéré"
def test_crp_minor(self):
assert interpret_bio_value("CRP", "8", True) == "syndrome inflammatoire mineur"
def test_crp_normal(self):
assert interpret_bio_value("CRP", "3", False) is None
def test_lipase_pancreatite(self):
assert interpret_bio_value("Lipasémie", "450", True) == "pancréatite biologique (>3N)"
def test_lipase_moderee(self):
assert interpret_bio_value("Lipasémie", "90", True) == "élévation modérée de la lipase"
def test_hemoglobine_severe(self):
assert interpret_bio_value("Hémoglobine", "6.5", True) == "anémie sévère (transfusion probable)"
def test_hemoglobine_moderee(self):
assert interpret_bio_value("Hémoglobine", "9", True) == "anémie modérée"
def test_plaquettes_severe(self):
assert interpret_bio_value("Plaquettes", "30", True) == "thrombopénie sévère"
def test_leucocytes_high(self):
assert interpret_bio_value("Leucocytes", "25", True) == "hyperleucocytose majeure (infection, inflammation)"
def test_leucocytes_low(self):
assert interpret_bio_value("Leucocytes", "1.5", True) == "leucopénie sévère (aplasie, immunodépression)"
def test_creatinine_severe(self):
assert interpret_bio_value("Créatinine", "350", True) == "insuffisance rénale sévère"
def test_unknown_test(self):
assert interpret_bio_value("Glycémie", "2.5", True) is None
def test_invalid_value(self):
assert interpret_bio_value("CRP", "positive", True) is None
def test_comma_separator(self):
assert interpret_bio_value("Hémoglobine", "6,5", True) == "anémie sévère (transfusion probable)"
def test_bilirubine_ictere(self):
assert interpret_bio_value("Bilirubine totale", "55", True) == "ictère franc"
def test_asat_cytolyse_majeure(self):
assert interpret_bio_value("ASAT", "250", True) == "cytolyse hépatique majeure (>5N)"
def test_asat_cytolyse_moderee(self):
assert interpret_bio_value("ASAT", "100", True) == "cytolyse hépatique modérée (>2N)"
# --- detect_treatment_indicators ---
class TestDetectTreatmentIndicators:
def test_insuline(self):
traitements = [Traitement(medicament="INSULINE LANTUS 20UI")]
result = detect_treatment_indicators(traitements)
assert len(result) == 1
assert result[0]["condition"] == "diabète insulino-traité"
def test_antibiotique_iv(self):
traitements = [Traitement(medicament="CEFTRIAXONE 1g IV")]
result = detect_treatment_indicators(traitements)
assert len(result) == 1
assert result[0]["condition"] == "antibiothérapie IV"
def test_multiple(self):
traitements = [
Traitement(medicament="Metformine 1000mg"),
Traitement(medicament="Enoxaparine 4000UI"),
Traitement(medicament="Paracétamol 1g"),
]
result = detect_treatment_indicators(traitements)
assert len(result) == 2
conditions = {r["condition"] for r in result}
assert "diabète type 2" in conditions
assert "anticoagulation (HBPM)" in conditions
def test_no_match(self):
traitements = [Traitement(medicament="Paracétamol 1g")]
result = detect_treatment_indicators(traitements)
assert result == []
def test_dedup_conditions(self):
traitements = [
Traitement(medicament="Enoxaparine 4000UI"),
Traitement(medicament="Lovenox 4000UI"),
]
result = detect_treatment_indicators(traitements)
# Les deux sont HBPM, mais une seule condition
assert len(result) == 1
def test_dict_input(self):
traitements = [{"medicament": "morphine 10mg"}]
result = detect_treatment_indicators(traitements)
assert len(result) == 1
assert result[0]["condition"] == "analgésie palier 3 (douleur sévère)"
# --- detect_severity_markers ---
class TestDetectSeverityMarkers:
def test_sejour_prolonge(self):
dossier = DossierMedical(sejour=Sejour(duree_sejour=20))
markers = detect_severity_markers(dossier)
assert any("séjour prolongé" in m for m in markers)
def test_sejour_gt7(self):
dossier = DossierMedical(sejour=Sejour(duree_sejour=10))
markers = detect_severity_markers(dossier)
assert any("séjour >7 jours" in m for m in markers)
def test_patient_tres_age(self):
dossier = DossierMedical(sejour=Sejour(age=85))
markers = detect_severity_markers(dossier)
assert any("très âgé" in m for m in markers)
def test_patient_age(self):
dossier = DossierMedical(sejour=Sejour(age=72))
markers = detect_severity_markers(dossier)
assert any("patient âgé" in m for m in markers)
def test_obesite_morbide(self):
dossier = DossierMedical(sejour=Sejour(imc=42.0))
markers = detect_severity_markers(dossier)
assert any("obésité morbide" in m for m in markers)
def test_complications(self):
dossier = DossierMedical(complications=["Fièvre", "Hématome"])
markers = detect_severity_markers(dossier)
assert any("2 complication(s)" in m for m in markers)
def test_no_markers(self):
dossier = DossierMedical(sejour=Sejour(age=45, duree_sejour=3))
markers = detect_severity_markers(dossier)
assert markers == []
# --- build_enriched_context ---
class TestBuildEnrichedContext:
def test_basic_context(self):
dossier = DossierMedical(
sejour=Sejour(sexe="M", age=65, duree_sejour=5),
biologie_cle=[
BiologieCle(test="CRP", valeur="150", anomalie=True),
],
traitements_sortie=[
Traitement(medicament="Ceftriaxone 1g"),
],
)
ctx = build_enriched_context(dossier)
assert ctx["sexe"] == "M"
assert ctx["age"] == 65
assert len(ctx["interpretations_bio"]) == 1
assert ctx["interpretations_bio"][0]["interpretation"] == "syndrome inflammatoire majeur"
assert len(ctx["conditions_traitements"]) == 1
assert ctx["conditions_traitements"][0]["condition"] == "antibiothérapie IV"
def test_no_abnormal_bio(self):
dossier = DossierMedical(
biologie_cle=[
BiologieCle(test="CRP", valeur="3", anomalie=False),
],
)
ctx = build_enriched_context(dossier)
assert ctx["interpretations_bio"] == []
# --- format_enriched_context ---
class TestFormatEnrichedContext:
def test_with_interpretations(self):
ctx = {
"sexe": "F",
"age": 70,
"imc": None,
"duree_sejour": 10,
"antecedents": ["HTA", "Diabète"],
"biologie_cle": [("CRP", "180", True)],
"imagerie": [],
"complications": [],
"dp_texte": None,
"das_codes_existants": None,
"interpretations_bio": [
{"test": "CRP", "valeur": "180 mg/L", "interpretation": "syndrome inflammatoire majeur"},
],
"conditions_traitements": [
{"medicament": "Insuline", "condition": "diabète insulino-traité"},
],
"marqueurs_severite": ["séjour >7 jours (10 jours)"],
}
result = format_enriched_context(ctx)
assert "Patient : F, 70 ans" in result
assert "Durée séjour : 10 jours" in result
assert "HTA" in result
assert "CRP 180" in result
assert "INTERPRÉTATION CLINIQUE" in result
assert "syndrome inflammatoire majeur" in result
assert "diabète insulino-traité" in result
assert "séjour >7 jours" in result
def test_empty_context(self):
ctx = {
"sexe": None, "age": None, "imc": None,
"duree_sejour": None, "antecedents": [],
"biologie_cle": [], "imagerie": [],
"complications": [], "dp_texte": None,
"das_codes_existants": None,
"interpretations_bio": [],
"conditions_traitements": [],
"marqueurs_severite": [],
}
result = format_enriched_context(ctx)
assert result == "Non précisé"
def test_backward_compat_bio_format(self):
"""Le format bio tuple (test, valeur, anomalie) doit rester compatible."""
ctx = {
"sexe": None, "age": None, "imc": None,
"duree_sejour": None, "antecedents": [],
"biologie_cle": [("CRP", "180", True)],
"imagerie": [],
"complications": [],
"dp_texte": None,
"das_codes_existants": None,
"interpretations_bio": [],
"conditions_traitements": [],
"marqueurs_severite": [],
}
result = format_enriched_context(ctx)
assert "CRP 180" in result
assert "(\u2191)" in result

245
tests/test_justification.py Normal file
View File

@@ -0,0 +1,245 @@
"""Tests pour la validation batch des justifications (QC post-codage)."""
from unittest.mock import patch, MagicMock
import pytest
from src.config import (
Diagnostic,
DossierMedical,
PreuveClinique,
Sejour,
BiologieCle,
)
class TestPreuveClinique:
def test_create(self):
p = PreuveClinique(type="biologie", element="CRP 180 mg/L", interpretation="syndrome inflammatoire majeur")
assert p.type == "biologie"
assert p.element == "CRP 180 mg/L"
assert p.interpretation == "syndrome inflammatoire majeur"
def test_diagnostic_with_preuves(self):
d = Diagnostic(
texte="Pancréatite aiguë",
cim10_suggestion="K85.9",
preuves_cliniques=[
PreuveClinique(type="biologie", element="Lipasémie 450 UI/L", interpretation="pancréatite biologique"),
PreuveClinique(type="imagerie", element="TDM: pancréatite stade D", interpretation="confirmation"),
],
)
assert len(d.preuves_cliniques) == 2
assert d.preuves_cliniques[0].type == "biologie"
def test_diagnostic_default_empty_preuves(self):
d = Diagnostic(texte="Test")
assert d.preuves_cliniques == []
def test_serialization_round_trip(self):
d = Diagnostic(
texte="Test",
preuves_cliniques=[
PreuveClinique(type="clinique", element="fièvre 39°C", interpretation="syndrome infectieux"),
],
)
data = d.model_dump()
assert data["preuves_cliniques"][0]["type"] == "clinique"
d2 = Diagnostic(**data)
assert d2.preuves_cliniques[0].element == "fièvre 39°C"
class TestApplyLlmResultPreuves:
"""Teste le stockage des preuves cliniques dans _apply_llm_result_diagnostic."""
def test_preuves_stored(self):
from src.medical.rag_search import _apply_llm_result_diagnostic
diag = Diagnostic(texte="Pneumopathie")
llm_result = {
"code": "J18.9",
"confidence": "high",
"justification": "Pneumopathie confirmée",
"preuves_cliniques": [
{"type": "biologie", "element": "CRP 120 mg/L", "interpretation": "syndrome inflammatoire"},
{"type": "imagerie", "element": "Radio thorax: opacité", "interpretation": "foyer pulmonaire"},
],
}
_apply_llm_result_diagnostic(diag, llm_result)
assert len(diag.preuves_cliniques) == 2
assert diag.preuves_cliniques[0].type == "biologie"
assert diag.preuves_cliniques[1].element == "Radio thorax: opacité"
def test_preuves_empty_list(self):
from src.medical.rag_search import _apply_llm_result_diagnostic
diag = Diagnostic(texte="Test")
llm_result = {"code": "K85.9", "confidence": "medium", "preuves_cliniques": []}
_apply_llm_result_diagnostic(diag, llm_result)
assert diag.preuves_cliniques == []
def test_preuves_missing(self):
from src.medical.rag_search import _apply_llm_result_diagnostic
diag = Diagnostic(texte="Test")
llm_result = {"code": "K85.9", "confidence": "medium"}
_apply_llm_result_diagnostic(diag, llm_result)
assert diag.preuves_cliniques == []
def test_preuves_malformed_skipped(self):
from src.medical.rag_search import _apply_llm_result_diagnostic
diag = Diagnostic(texte="Test")
llm_result = {
"code": "K85.9",
"confidence": "high",
"preuves_cliniques": [
{"type": "bio"}, # manque 'element' → ignoré
{"type": "imagerie", "element": "TDM ok", "interpretation": "normal"},
"not a dict", # ignoré
],
}
_apply_llm_result_diagnostic(diag, llm_result)
assert len(diag.preuves_cliniques) == 1
assert diag.preuves_cliniques[0].element == "TDM ok"
class TestValidateJustifications:
"""Teste la fonction _validate_justifications."""
@patch("src.medical.ollama_client.call_ollama")
def test_confidence_adjusted(self, mock_ollama):
from src.medical.cim10_extractor import _validate_justifications
mock_ollama.return_value = {
"validations": [
{
"numero": 1,
"code": "K85.9",
"verdict": "maintenir",
"confidence_recommandee": "high",
"commentaire": "bien justifié",
},
{
"numero": 2,
"code": "I10",
"verdict": "maintenir",
"confidence_recommandee": "low",
"commentaire": "pas de preuve tensionnelle",
},
],
"alertes_globales": [],
}
dossier = DossierMedical(
sejour=Sejour(sexe="M", age=60),
diagnostic_principal=Diagnostic(
texte="Pancréatite aiguë",
cim10_suggestion="K85.9",
cim10_confidence="medium",
),
diagnostics_associes=[
Diagnostic(
texte="HTA",
cim10_suggestion="I10",
cim10_confidence="high",
),
],
)
_validate_justifications(dossier)
# DP: medium → high
assert dossier.diagnostic_principal.cim10_confidence == "high"
# DAS: high → low
assert dossier.diagnostics_associes[0].cim10_confidence == "low"
# Alertes de confiance
assert any("QC:" in a and "I10" in a for a in dossier.alertes_codage)
@patch("src.medical.ollama_client.call_ollama")
def test_das_supprimer_alerte(self, mock_ollama):
from src.medical.cim10_extractor import _validate_justifications
mock_ollama.return_value = {
"validations": [
{
"numero": 1,
"code": "K85.9",
"verdict": "maintenir",
"confidence_recommandee": "high",
"commentaire": "ok",
},
{
"numero": 2,
"code": "R10.4",
"verdict": "supprimer",
"confidence_recommandee": "low",
"commentaire": "symptôme couvert par le DP",
},
],
"alertes_globales": ["Vérifier la spécificité du DP"],
}
dossier = DossierMedical(
diagnostic_principal=Diagnostic(
texte="Pancréatite aiguë",
cim10_suggestion="K85.9",
cim10_confidence="high",
),
diagnostics_associes=[
Diagnostic(
texte="Douleur abdominale",
cim10_suggestion="R10.4",
cim10_confidence="medium",
),
],
)
_validate_justifications(dossier)
# Le DAS n'est pas supprimé automatiquement, mais une alerte est ajoutée
assert any("à reconsidérer" in a for a in dossier.alertes_codage)
assert any("Vérifier la spécificité" in a for a in dossier.alertes_codage)
@patch("src.medical.ollama_client.call_ollama")
def test_ollama_returns_none(self, mock_ollama):
from src.medical.cim10_extractor import _validate_justifications
mock_ollama.return_value = None
dossier = DossierMedical(
diagnostic_principal=Diagnostic(
texte="Test",
cim10_suggestion="K85.9",
cim10_confidence="high",
),
)
_validate_justifications(dossier)
assert dossier.alertes_codage == []
def test_no_diags(self):
from src.medical.cim10_extractor import _validate_justifications
dossier = DossierMedical()
_validate_justifications(dossier)
assert dossier.alertes_codage == []
@patch("src.medical.ollama_client.call_ollama")
def test_invalid_validation_nums_skipped(self, mock_ollama):
from src.medical.cim10_extractor import _validate_justifications
mock_ollama.return_value = {
"validations": [
{"numero": 0, "code": "X", "verdict": "supprimer", "confidence_recommandee": "low", "commentaire": "oob"},
{"numero": 99, "code": "Y", "verdict": "supprimer", "confidence_recommandee": "low", "commentaire": "oob"},
{"numero": "abc", "code": "Z", "verdict": "supprimer", "confidence_recommandee": "low", "commentaire": "type"},
],
"alertes_globales": [],
}
dossier = DossierMedical(
diagnostic_principal=Diagnostic(texte="T", cim10_suggestion="A00", cim10_confidence="high"),
)
_validate_justifications(dossier)
# Aucune modification, tous les numéros sont invalides
assert dossier.diagnostic_principal.cim10_confidence == "high"
assert dossier.alertes_codage == []