Files
rpa_vision_v3/core/llm/t2a_decision.py
Dom 9872f4510c feat(t2a): build_dpi_enriched - extraction déterministe horaires + classifications cliniques
Préprocesseur Python qui injecte un bloc FAITS_CALCULÉS en tête du DPI
avant l'appel LLM, pour neutraliser l'hallucination de durée (bug "23h"
sur cas MOREL, confusion avec "depuis 23h" de l'Observ. IDE Urg).

Extrait depuis le bandeau Easily Assure et la Synthèse Urgences :
- âge (dateutil.relativedelta)
- date admission / sortie + durée passage (format humain + décimal)
- CCMU / GEMSA libellé complet (parser multi-ligne)
- priorité IAO, mode de venue / médicalisation / mode d'entrée
- diagnostic principal
- decision_terrain + orientation_terrain (metadata only, jamais injectés
  dans le prompt pour ne pas biaiser le LLM)

Retour tuple (dpi_enriched, metadata) pour permettre les garde-fous
serveur Python ↔ LLM au commit 2.

Robustesse :
- re.search 1re occurrence + WARNING si bandeau divergent multi-occurrences
- Synthèse Urgences priorité sur bandeau pour dates
- Valeur exigée sur même ligne que label (évite capture de section title)
- Cas négatif (horaires absents) → "NON CALCULABLE" + parsing_warnings
- Jamais de crash, retour tuple toujours valide

Tests : 4/4 verts (golden MOREL string + metadata, négatif sortie absente,
DPI vide). Pas de régression sur tests/integration/test_t2a_extract.py.

Brief complet : docs/handoffs/2026-05-12_brief_S1_build_dpi_enriched.md

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 18:49:49 +02:00

457 lines
19 KiB
Python

"""Aide à la décision de facturation urgences T2A/PMSI via LLM local.
Décide si un passage aux urgences relève :
- du FORFAIT_URGENCE (passage simple, retour à domicile)
- de la REQUALIFICATION_HOSPITALISATION (séjour MCO, valorisation 1k-5k€+)
Le prompt impose une extraction littérale des faits du DPI (pas d'invention)
et une modulation honnête de la confiance. Validé sur 15 DPI synthétiques :
qwen2.5:7b atteint 100 % d'accuracy en ~5 s/cas avec 4,7 Go VRAM.
Voir docs/clients/ght_sud_95/ et demo/facturation_urgences/RESULTATS.md pour le
bench comparatif des 11 LLMs évalués.
"""
from __future__ import annotations
import json
import logging
import os
import re
import time
import urllib.error
import urllib.request
from datetime import datetime
from typing import Any, Dict, Optional, Tuple
from dateutil.relativedelta import relativedelta
logger = logging.getLogger(__name__)
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434/api/generate")
DEFAULT_MODEL = os.environ.get("T2A_MODEL", "qwen2.5:7b")
DEFAULT_TIMEOUT = 60 # secondes
PROMPT_TEMPLATE = """Tu es médecin DIM (Département d'Information Médicale), expert en facturation T2A/PMSI aux urgences hospitalières en France.
Analyse le dossier patient ci-dessous pour déterminer si le passage relève :
- FORFAIT_URGENCE : passage simple, retour à domicile, sans surveillance prolongée ni soins continus
- REQUALIFICATION_HOSPITALISATION : séjour MCO requis selon les 3 critères PMSI/ATIH
LES 3 CRITÈRES UHCD (au moins 2 sur 3 validés ⇒ REQUALIFICATION) :
1. Pathologie potentiellement évolutive (instabilité hémodynamique, terrain à risque, traitement nécessitant adaptation)
2. Surveillance médicale et paramédicale prolongée (constantes itératives, observations IDE/médecin, durée > 6 h)
3. Examens complémentaires ou actes thérapeutiques (biologie, imagerie, sutures, gestes techniques)
INSTRUCTIONS STRICTES :
1. N'utilise QUE des éléments littéralement présents dans le dossier patient. N'invente AUCUN critère.
2. Pour CHAQUE critère (1, 2, 3), tu DOIS produire un texte de preuve qui contient AU MOINS UNE CITATION LITTÉRALE du dossier entre guillemets français « ... ». Exemple : « FC à 110 bpm, TA 92/60 ».
3. Si le critère est NON validé, ne renvoie JAMAIS un fallback creux : explique factuellement ce qui manque, en citant le dossier (ex: « Sortie à H+2 », « Aucun acte technique au compte-rendu »).
4. Le texte de chaque preuve fait 2-3 phrases : (i) la citation littérale, (ii) l'analyse PMSI, (iii) la conclusion validé/non validé.
5. Calcule la durée totale du passage en heures (admission → sortie/transfert) à partir des horaires du dossier.
6. Module ta confiance honnêtement :
- "elevee" uniquement si tous les indices convergent
- "moyenne" si éléments ambivalents
- "faible" si information manquante ou très atypique
Réponds STRICTEMENT en JSON valide, sans texte avant ni après :
{{
"duree_passage_heures": <nombre>,
"elements_pour_hospitalisation": [<phrases littéralement extraites du dossier>],
"elements_pour_forfait": [<phrases littéralement extraites du dossier>],
"decision": "FORFAIT_URGENCE" | "REQUALIFICATION_HOSPITALISATION",
"decision_court": "UHCD" | "Forfait Urgences",
"preuve_critere1": "<2-3 phrases incluant AU MOINS UNE citation littérale entre « » (motif, symptôme, terrain à risque, traitement). Si non validé : factualise ce qui manque en citant le dossier.>",
"critere1_valide": true | false,
"preuve_critere2": "<2-3 phrases incluant AU MOINS UNE citation littérale entre « » (constantes, observations IDE, durée surveillance). Si non validé : factualise.>",
"critere2_valide": true | false,
"preuve_critere3": "<2-3 phrases incluant AU MOINS UNE citation littérale entre « » (actes/examens : biologie, imagerie, suture, etc.). Si non validé : factualise.>",
"critere3_valide": true | false,
"justification": "<2-3 phrases synthétiques s'appuyant explicitement sur les preuves ci-dessus, avec au moins une citation>",
"confiance": "elevee" | "moyenne" | "faible"
}}
DOSSIER PATIENT :
{dpi}
"""
def analyze_dpi(
dpi_text: str,
model: str = DEFAULT_MODEL,
timeout: int = DEFAULT_TIMEOUT,
ollama_url: str = OLLAMA_URL,
) -> Dict[str, Any]:
"""Soumet un DPI urgences à un LLM Ollama et retourne la décision JSON.
Args:
dpi_text: Texte du dossier patient (concaténation des onglets ou DPI brut).
model: Modèle Ollama à utiliser (default qwen2.5:7b — 100% accuracy bench).
timeout: Timeout HTTP en secondes.
ollama_url: Endpoint Ollama (default localhost:11434/api/generate).
Returns:
Dict avec :
decision: "FORFAIT_URGENCE" | "REQUALIFICATION_HOSPITALISATION"
elements_pour_hospitalisation: List[str]
elements_pour_forfait: List[str]
duree_passage_heures: float
justification: str
confiance: "elevee" | "moyenne" | "faible"
_elapsed_s: float (latence)
_model: str
En cas d'erreur :
{"_error": str, "_elapsed_s": float} (réseau / Ollama indisponible)
{"_parse_error": True, "_raw": str, "_elapsed_s": float} (JSON invalide)
"""
payload = {
"model": model,
"prompt": PROMPT_TEMPLATE.format(dpi=dpi_text),
"stream": False,
"format": "json",
"keep_alive": "5m",
"options": {
"temperature": 0.1,
"num_predict": 1500,
"num_ctx": 16384,
},
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
ollama_url,
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
t0 = time.time()
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
body = json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, TimeoutError, ConnectionError) as e:
elapsed = round(time.time() - t0, 1)
logger.warning("analyze_dpi: Ollama indisponible (%s) après %.1fs", e, elapsed)
return {"_error": str(e), "_elapsed_s": elapsed, "_model": model}
elapsed = time.time() - t0
raw_response = body.get("response", "").strip()
raw_thinking = body.get("thinking", "").strip()
candidates = [raw_response]
if not raw_response and raw_thinking:
last_close = raw_thinking.rfind("}")
last_open = raw_thinking.rfind("{", 0, last_close)
if last_open != -1 and last_close != -1:
candidates.append(raw_thinking[last_open:last_close + 1])
parsed = None
for cand in candidates:
cleaned = cand
if cleaned.startswith("```"):
cleaned = cleaned.split("\n", 1)[-1]
if cleaned.endswith("```"):
cleaned = cleaned.rsplit("```", 1)[0]
cleaned = cleaned.strip()
try:
parsed = json.loads(cleaned)
break
except json.JSONDecodeError:
continue
if parsed is None:
return {
"_parse_error": True,
"_raw": (raw_response or raw_thinking)[:500],
"_elapsed_s": round(elapsed, 1),
"_model": model,
}
parsed["_elapsed_s"] = round(elapsed, 1)
parsed["_model"] = model
parsed["_eval_count"] = body.get("eval_count")
return parsed
# ---------------------------------------------------------------------------
# build_dpi_enriched — extraction déterministe horaires + classifications
# ---------------------------------------------------------------------------
#
# Voir docs/handoffs/2026-05-12_brief_S1_build_dpi_enriched.md pour le contexte
# complet (bug hallucination durée "23h" sur cas MOREL, plan P0 [S1]/[S2]).
# Libellés connus de la section "Synthèse Urgences" (Easily Assure).
# Servent d'ancres pour le parsing : la valeur d'un champ s'étend depuis son
# libellé jusqu'au prochain libellé connu (gère le wrap multi-ligne CCMU/GEMSA).
# Trier par longueur décroissante côté regex pour éviter qu'un préfixe partiel
# (ex: "Décision médicale" matche "Médecin de la décision médicale").
_LIBELLES_SYNTHESE = [
"Episode - Date",
"Mode de transport à l'arrivée",
"Médicalisation du transport",
"Mode d'entrée",
"Origine du transfert",
"Date d'orientation",
"IAO",
"Priorité",
"Episode - Sous-type",
"Circonstances",
"Motif de prise en charge",
"Observ. IDE Urg",
"Médecin de la prise en charge médicale",
"Date de la prise en charge médicale",
"CCMU",
"GEMSA",
"Diagnostics",
"Médecin de la décision médicale",
"Date de décision médicale",
"Décision médicale",
"Orientation du patient",
"US de destination",
]
def _extract_synthese_field(text: str, label: str) -> Optional[str]:
"""Extrait la valeur du champ `label` dans la Synthèse Urgences.
Stratégie : trouve la 1re ligne qui commence par `label`, capture la suite
jusqu'au prochain libellé connu (en début de ligne) ou fin de texte.
Gère le wrap multi-ligne (CCMU 3 + libellé long sur plusieurs lignes).
Normalise les espaces multiples internes en un seul espace.
"""
other_labels = sorted(
(l for l in _LIBELLES_SYNTHESE if l != label),
key=len,
reverse=True,
)
other_alt = "|".join(re.escape(l) for l in other_labels)
# Exigence : la valeur doit commencer sur la même ligne que le label
# ([^\n]). Cela évite qu'une ligne titre de section ("Décision médicale\n"
# seule) capture la ligne suivante (qui serait un autre champ).
# Le wrap multi-ligne CCMU/GEMSA reste supporté : 1er char sur la même
# ligne que le label, suite via DOTALL jusqu'au prochain libellé connu.
pattern = (
rf"^[\s\|]*{re.escape(label)}[ \t]*[:\|]?[ \t]+"
rf"([^\n].*?)(?=\n[\s\|]*(?:{other_alt})\b|\Z)"
)
m = re.search(pattern, text, re.DOTALL | re.MULTILINE)
if not m:
return None
value = re.sub(r"\s+", " ", m.group(1)).strip()
return value or None
def _parse_dt(date_str: str, time_str: str) -> Optional[datetime]:
"""Parse 'DD/MM/YYYY' + 'HH:MM' → datetime. Retourne None si échec."""
try:
return datetime.strptime(f"{date_str} {time_str}", "%d/%m/%Y %H:%M")
except ValueError:
return None
def _parse_synthese_datetime(value: Optional[str]) -> Optional[datetime]:
"""Parse une valeur de type 'JJ/MM/AAAA à HH:MM' (format Synthèse Urgences).
Tolère absence du séparateur ' à ' (OCR peut linéariser autrement).
"""
if not value:
return None
m = re.search(r"(\d{2}/\d{2}/\d{4})\s*(?:à\s*)?(\d{2}:\d{2})", value)
if not m:
return None
return _parse_dt(m.group(1), m.group(2))
def build_dpi_enriched(dpi_raw: str) -> Tuple[str, Dict[str, Any]]:
"""Enrichit le DPI brut avec un bloc FAITS_CALCULÉS en tête.
Extrait de manière déterministe (Python, pas LLM) :
- âge du patient (depuis date de naissance bandeau + date d'admission)
- durée totale du passage (depuis horaires admission/sortie)
- CCMU, GEMSA, priorité IAO, mode de venue, diagnostic principal
- décision médicale terrain + orientation (metadata uniquement, NON injectés
dans le bloc FAITS_CALCULÉS pour ne pas biaiser le LLM)
Le bloc FAITS_CALCULÉS est concaténé en tête du DPI retourné.
Le metadata dict permet au handler serveur d'effectuer les garde-fous
Python ↔ LLM au commit 2.
Args:
dpi_raw: DPI brut concaténé (5 onglets OCR scroll auto + bandeau répété).
Returns:
Tuple (dpi_enriched, metadata).
dpi_enriched: str — FAITS_CALCULÉS + "\\n\\n" + dpi_raw.
metadata: dict — toutes les valeurs Python extraites (None si parsing
échoué pour un champ), + parsing_warnings: list[str].
"""
metadata: Dict[str, Any] = {
"age_ans": None,
"date_admission": None,
"date_sortie": None,
"duree_heures_decimales": None,
"ccmu": None,
"gemsa": None,
"priorite_iao": None,
"mode_venue": None,
"mode_medicalisation": None,
"mode_entree": None,
"diagnostic_principal": None,
"decision_terrain": None,
"orientation_terrain": None,
"parsing_warnings": [],
}
# ── 1. Bandeau Easily Assure (re.search = 1re occurrence) ────────────────
# Le bandeau est répété ~5 fois dans dpi_raw (un par extract_text_scroll
# de chaque onglet). On prend la 1re occurrence et on signale une éventuelle
# divergence inter-occurrences (symptôme d'OCR instable).
date_naissance: Optional[datetime] = None
date_adm_bandeau: Optional[datetime] = None
date_sortie_bandeau: Optional[datetime] = None
m_naissance = re.search(r"\(e\)\s+le\s+(\d{2}/\d{2}/\d{4})", dpi_raw)
if m_naissance:
try:
date_naissance = datetime.strptime(m_naissance.group(1), "%d/%m/%Y")
except ValueError:
metadata["parsing_warnings"].append(
f"date_naissance non parsable : {m_naissance.group(1)!r}"
)
arrivees = re.findall(
r"Arriv[ée]e\s*:\s*(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2})", dpi_raw
)
if arrivees:
date_adm_bandeau = _parse_dt(*arrivees[0])
if len({a for a in arrivees}) > 1:
logger.warning(
"[build_dpi_enriched] Bandeau détecté %d fois avec divergences "
"sur Arrivée — prise de la 1re occurrence : %s",
len(arrivees), arrivees[0],
)
sorties = re.findall(
r"Sortie\s*:\s*(\d{2}/\d{2}/\d{4})\s+(\d{2}:\d{2})", dpi_raw
)
if sorties:
date_sortie_bandeau = _parse_dt(*sorties[0])
if len({s for s in sorties}) > 1:
logger.warning(
"[build_dpi_enriched] Bandeau détecté %d fois avec divergences "
"sur Sortie — prise de la 1re occurrence : %s",
len(sorties), sorties[0],
)
# ── 2. Synthèse Urgences (priorité sur le bandeau) ───────────────────────
syn_episode_date = _parse_synthese_datetime(
_extract_synthese_field(dpi_raw, "Episode - Date")
)
syn_pec_medicale = _parse_synthese_datetime(
_extract_synthese_field(dpi_raw, "Date de la prise en charge médicale")
)
syn_decision_medicale = _parse_synthese_datetime(
_extract_synthese_field(dpi_raw, "Date de décision médicale")
)
# Priorité d'admission : Episode-Date > PEC médicale > bandeau Arrivée
metadata["date_admission"] = (
syn_episode_date or syn_pec_medicale or date_adm_bandeau
)
# Priorité de sortie : Date décision médicale > bandeau Sortie
metadata["date_sortie"] = syn_decision_medicale or date_sortie_bandeau
# Champs structurés Synthèse Urgences (libellé complet, multi-ligne ok)
metadata["ccmu"] = _extract_synthese_field(dpi_raw, "CCMU")
metadata["gemsa"] = _extract_synthese_field(dpi_raw, "GEMSA")
metadata["priorite_iao"] = _extract_synthese_field(dpi_raw, "Priorité")
metadata["mode_venue"] = _extract_synthese_field(
dpi_raw, "Mode de transport à l'arrivée"
)
metadata["mode_medicalisation"] = _extract_synthese_field(
dpi_raw, "Médicalisation du transport"
)
metadata["mode_entree"] = _extract_synthese_field(dpi_raw, "Mode d'entrée")
metadata["diagnostic_principal"] = _extract_synthese_field(dpi_raw, "Diagnostics")
metadata["decision_terrain"] = _extract_synthese_field(dpi_raw, "Décision médicale")
metadata["orientation_terrain"] = _extract_synthese_field(
dpi_raw, "US de destination"
)
# ── 3. Calculs dérivés ───────────────────────────────────────────────────
if date_naissance and metadata["date_admission"]:
metadata["age_ans"] = relativedelta(
metadata["date_admission"], date_naissance
).years
elif date_naissance is None:
metadata["parsing_warnings"].append("date_naissance non détectée dans bandeau")
duree_format_humain: Optional[str] = None
if metadata["date_admission"] and metadata["date_sortie"]:
delta = metadata["date_sortie"] - metadata["date_admission"]
total_seconds = delta.total_seconds()
if total_seconds <= 0:
metadata["parsing_warnings"].append(
f"durée invalide : sortie {metadata['date_sortie']} <= "
f"admission {metadata['date_admission']}"
)
else:
metadata["duree_heures_decimales"] = round(total_seconds / 3600, 2)
heures = int(total_seconds // 3600)
minutes = int((total_seconds % 3600) // 60)
duree_format_humain = f"{heures} heures et {minutes} minutes"
else:
if metadata["date_admission"] is None:
metadata["parsing_warnings"].append("date_admission non détectée")
if metadata["date_sortie"] is None:
metadata["parsing_warnings"].append("date_sortie non détectée")
logger.warning(
"[build_dpi_enriched] Durée non calculable — admission=%s sortie=%s",
metadata["date_admission"], metadata["date_sortie"],
)
# ── 4. Construction du bloc FAITS_CALCULÉS ───────────────────────────────
lignes = ["FAITS_CALCULÉS (déterministes, ne pas recalculer) :"]
if metadata["age_ans"] is not None:
lignes.append(f"- Âge du patient : {metadata['age_ans']} ans")
if metadata["date_admission"]:
lignes.append(
f"- Date admission : "
f"{metadata['date_admission'].strftime('%d/%m/%Y à %H:%M')}"
)
if metadata["date_sortie"]:
lignes.append(
f"- Date sortie : "
f"{metadata['date_sortie'].strftime('%d/%m/%Y à %H:%M')}"
)
if duree_format_humain and metadata["duree_heures_decimales"] is not None:
lignes.append(
f"- Durée totale du passage : {duree_format_humain} "
f"(soit {metadata['duree_heures_decimales']} heures décimales)"
)
elif metadata["date_admission"] is None or metadata["date_sortie"] is None:
lignes.append(
"- Durée totale du passage : NON CALCULABLE (horaires non détectés)"
)
if metadata["ccmu"]:
lignes.append(f"- CCMU : {metadata['ccmu']}")
if metadata["gemsa"]:
lignes.append(f"- GEMSA : {metadata['gemsa']}")
if metadata["priorite_iao"]:
lignes.append(f"- Priorité IAO : {metadata['priorite_iao']}")
if metadata["mode_venue"] or metadata["mode_medicalisation"]:
mode_parts = [
p for p in (metadata["mode_venue"], metadata["mode_medicalisation"]) if p
]
lignes.append(f"- Mode de venue : {', '.join(mode_parts)}")
if metadata["mode_entree"]:
lignes.append(f"- Mode d'entrée : {metadata['mode_entree']}")
if metadata["diagnostic_principal"]:
lignes.append(f"- Diagnostic principal : {metadata['diagnostic_principal']}")
bloc_faits = "\n".join(lignes)
dpi_enriched = f"{bloc_faits}\n\n{dpi_raw}"
return dpi_enriched, metadata