refactor: split cpam_response → cpam_rag, cpam_context, cpam_validation
Découpe le monolithe cpam_response.py (1207L) en 3 modules spécialisés : - cpam_rag.py : recherche RAG ciblée (5 requêtes, dédup) - cpam_context.py : construction prompt, définitions CIM-10, bio summary - cpam_validation.py : grounding, références, codes fermée, adversariale Le cpam_response.py reste orchestrateur (~230L) avec re-exports backward-compat. Mocks des tests mis à jour pour cibler les bons modules. Ajout RULE-CPAM-CORRECTION-LOOP dans base.yaml. 748 tests passent. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dom
376
src/control/cpam_validation.py
Normal file
376
src/control/cpam_validation.py
Normal file
@@ -0,0 +1,376 @@
|
||||
"""Validation et formatage des réponses CPAM (grounding, adversariale, codes)."""
|
||||
|
||||
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 ..medical.ollama_client import call_anthropic, call_ollama
|
||||
from ..prompts import CPAM_ADVERSARIAL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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 _validate_references(parsed: dict, sources: list[dict]) -> list[str]:
|
||||
"""Vérifie que les références citées correspondent aux sources RAG fournies.
|
||||
|
||||
Returns:
|
||||
Liste d'avertissements pour les références non vérifiables.
|
||||
"""
|
||||
warnings = []
|
||||
refs = parsed.get("references")
|
||||
if not refs or not isinstance(refs, list):
|
||||
return warnings
|
||||
|
||||
# Construire un set des documents sources disponibles
|
||||
source_docs = set()
|
||||
for src in sources:
|
||||
doc_name = src.get("document", "")
|
||||
source_docs.add(doc_name)
|
||||
# Ajouter les noms lisibles aussi
|
||||
readable = {
|
||||
"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(doc_name, "")
|
||||
if readable:
|
||||
source_docs.add(readable)
|
||||
source_docs.add(readable.lower())
|
||||
|
||||
if not source_docs:
|
||||
return warnings
|
||||
|
||||
for ref in refs:
|
||||
if not isinstance(ref, dict):
|
||||
continue
|
||||
doc = ref.get("document", "")
|
||||
if doc and not any(sd in doc.lower() or doc.lower() in sd.lower() for sd in source_docs if sd):
|
||||
warnings.append(f"Référence non vérifiable : {doc}")
|
||||
logger.warning("CPAM : référence non vérifiable « %s »", doc)
|
||||
|
||||
return warnings
|
||||
|
||||
|
||||
# Regex pour capturer les codes CIM-10 (ex: K81.0, E87, Z45.80)
|
||||
_CIM10_CODE_RE = re.compile(r"\b([A-Z]\d{2}\.?\d{0,2})\b")
|
||||
|
||||
|
||||
def _validate_codes_in_response(
|
||||
parsed: dict,
|
||||
dossier: DossierMedical,
|
||||
controle: ControleCPAM,
|
||||
) -> list[str]:
|
||||
"""Vérifie que les codes CIM-10 cités dans la réponse sont dans le périmètre du dossier.
|
||||
|
||||
Construit une whitelist à partir du dossier (DP, DAS) et de l'UCR (dp_ucr, da_ucr, dr_ucr),
|
||||
puis extrait tous les codes CIM-10 des champs textuels de la réponse LLM.
|
||||
La comparaison se fait par préfixe 3 caractères (ex: K81 matche K81.0 et K81.09).
|
||||
|
||||
Returns:
|
||||
Liste de warnings pour les codes hors périmètre.
|
||||
"""
|
||||
# 1. Construire la whitelist (préfixes 3 chars)
|
||||
whitelist_prefixes: set[str] = set()
|
||||
|
||||
def _add_code(raw: str) -> None:
|
||||
raw = raw.strip()
|
||||
if not raw:
|
||||
return
|
||||
norm = normalize_code(raw)
|
||||
if norm and len(norm) >= 3:
|
||||
whitelist_prefixes.add(norm[:3])
|
||||
|
||||
# Codes du dossier
|
||||
if dossier.diagnostic_principal and dossier.diagnostic_principal.cim10_suggestion:
|
||||
_add_code(dossier.diagnostic_principal.cim10_suggestion)
|
||||
for das in dossier.diagnostics_associes:
|
||||
if das.cim10_suggestion:
|
||||
_add_code(das.cim10_suggestion)
|
||||
|
||||
# Codes de l'UCR
|
||||
for field in (controle.dp_ucr, controle.da_ucr, controle.dr_ucr):
|
||||
if not field:
|
||||
continue
|
||||
for raw in re.split(r"[,;\s]+", field.strip()):
|
||||
_add_code(raw)
|
||||
|
||||
if not whitelist_prefixes:
|
||||
return []
|
||||
|
||||
# 2. Extraire les codes CIM-10 de la réponse LLM (hors citations RAG)
|
||||
text_fields = []
|
||||
for key in (
|
||||
"analyse_contestation",
|
||||
"contre_arguments_medicaux",
|
||||
"contre_arguments_asymetrie",
|
||||
"contre_arguments_reglementaires",
|
||||
"conclusion",
|
||||
):
|
||||
val = parsed.get(key)
|
||||
if val and isinstance(val, str):
|
||||
text_fields.append(val)
|
||||
|
||||
# Preuves du dossier — valeurs
|
||||
preuves = parsed.get("preuves_dossier")
|
||||
if preuves and isinstance(preuves, list):
|
||||
for p in preuves:
|
||||
if isinstance(p, dict):
|
||||
v = p.get("valeur", "")
|
||||
if v and isinstance(v, str):
|
||||
text_fields.append(v)
|
||||
|
||||
combined_text = "\n".join(text_fields)
|
||||
found_codes = _CIM10_CODE_RE.findall(combined_text)
|
||||
|
||||
if not found_codes:
|
||||
return []
|
||||
|
||||
# 3. Comparer par préfixe 3 chars
|
||||
warnings: list[str] = []
|
||||
seen_warned: set[str] = set()
|
||||
|
||||
for raw_code in found_codes:
|
||||
norm = normalize_code(raw_code)
|
||||
if not norm or len(norm) < 3:
|
||||
continue
|
||||
prefix = norm[:3]
|
||||
if prefix in whitelist_prefixes:
|
||||
continue
|
||||
if norm in seen_warned:
|
||||
continue
|
||||
seen_warned.add(norm)
|
||||
is_valid, label = validate_code(norm)
|
||||
label_str = f" ({label})" if is_valid and label else ""
|
||||
warnings.append(f"Code {norm}{label_str} hors périmètre dossier/UCR")
|
||||
logger.warning("CPAM : code %s%s absent du dossier et de l'UCR", norm, label_str)
|
||||
|
||||
return warnings
|
||||
|
||||
|
||||
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)
|
||||
|
||||
dp_ucr_line = f"DP UCR : {controle.dp_ucr}" if controle.dp_ucr else ""
|
||||
da_ucr_line = f"DA UCR : {controle.da_ucr}" if controle.da_ucr else ""
|
||||
|
||||
prompt = CPAM_ADVERSARIAL.format(
|
||||
response_json=response_json,
|
||||
factual_section=factual_section,
|
||||
normes_section=normes_section,
|
||||
dp_ucr_line=dp_ucr_line,
|
||||
da_ucr_line=da_ucr_line,
|
||||
)
|
||||
|
||||
logger.debug(" Validation adversariale")
|
||||
result = call_ollama(prompt, temperature=0.0, max_tokens=800, role="validation")
|
||||
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 _build_correction_prompt(
|
||||
original_prompt: str,
|
||||
original_response: dict,
|
||||
adversarial_result: dict,
|
||||
) -> str:
|
||||
"""Construit un prompt de correction en injectant les erreurs détectées.
|
||||
|
||||
Args:
|
||||
original_prompt: Le prompt d'argumentation initial.
|
||||
original_response: La réponse LLM originale (dict).
|
||||
adversarial_result: Le résultat de la validation adversariale.
|
||||
|
||||
Returns:
|
||||
Prompt de correction prêt à envoyer au LLM.
|
||||
"""
|
||||
import json as _json
|
||||
|
||||
erreurs = adversarial_result.get("erreurs", [])
|
||||
erreurs_text = "\n".join(f" {i}. {e}" for i, e in enumerate(erreurs, 1))
|
||||
|
||||
# Résumé compact de la réponse problématique
|
||||
summary_fields = {}
|
||||
for key in ("analyse_contestation", "contre_arguments_medicaux",
|
||||
"contre_arguments_asymetrie", "contre_arguments_reglementaires",
|
||||
"conclusion"):
|
||||
val = original_response.get(key)
|
||||
if val and isinstance(val, str):
|
||||
# Tronquer chaque champ à 400 chars
|
||||
summary_fields[key] = val[:400] + ("..." if len(val) > 400 else "")
|
||||
|
||||
try:
|
||||
response_summary = _json.dumps(summary_fields, ensure_ascii=False, indent=2)
|
||||
except (TypeError, ValueError):
|
||||
response_summary = str(summary_fields)
|
||||
|
||||
correction_block = (
|
||||
"\n\n=== CORRECTION REQUISE — ERREURS DÉTECTÉES DANS TA RÉPONSE PRÉCÉDENTE ===\n"
|
||||
f"{erreurs_text}\n\n"
|
||||
f"RÉPONSE PRÉCÉDENTE (À CORRIGER) :\n{response_summary}\n\n"
|
||||
"Corrige UNIQUEMENT les erreurs ci-dessus. Conserve les parties correctes.\n"
|
||||
"Réponds avec le même format JSON."
|
||||
)
|
||||
|
||||
return original_prompt + correction_block
|
||||
|
||||
|
||||
def _format_response(parsed: dict, ref_warnings: list[str] | None = None) -> str:
|
||||
"""Formate la réponse LLM en texte lisible."""
|
||||
sections = []
|
||||
|
||||
analyse = parsed.get("analyse_contestation")
|
||||
if analyse:
|
||||
sections.append(f"ANALYSE DE LA CONTESTATION\n{analyse}")
|
||||
|
||||
accord = parsed.get("points_accord")
|
||||
if accord and accord.lower() not in ("aucun", "non applicable", "n/a", ""):
|
||||
sections.append(f"POINTS D'ACCORD\n{accord}")
|
||||
|
||||
# Nouveaux champs structurés par axe
|
||||
contre_med = parsed.get("contre_arguments_medicaux")
|
||||
if contre_med:
|
||||
sections.append(f"CONTRE-ARGUMENTS MÉDICAUX\n{contre_med}")
|
||||
|
||||
# Preuves du dossier (nouveau champ structuré)
|
||||
preuves = parsed.get("preuves_dossier")
|
||||
if preuves and isinstance(preuves, list):
|
||||
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", "")
|
||||
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))
|
||||
|
||||
contre_asym = parsed.get("contre_arguments_asymetrie")
|
||||
if contre_asym:
|
||||
sections.append(f"ASYMÉTRIE D'INFORMATION\n{contre_asym}")
|
||||
|
||||
contre_regl = parsed.get("contre_arguments_reglementaires")
|
||||
if contre_regl:
|
||||
sections.append(f"CONTRE-ARGUMENTS RÉGLEMENTAIRES\n{contre_regl}")
|
||||
|
||||
# Fallback : ancien champ unique (réponses en cache existantes)
|
||||
if not contre_med and not contre_asym and not contre_regl:
|
||||
contre = parsed.get("contre_arguments")
|
||||
if contre:
|
||||
sections.append(f"CONTRE-ARGUMENTS\n{contre}")
|
||||
|
||||
# Références structurées (nouveau format liste) ou ancien format string
|
||||
refs = parsed.get("references")
|
||||
if refs:
|
||||
if isinstance(refs, list):
|
||||
ref_lines = []
|
||||
for r in refs:
|
||||
if isinstance(r, dict):
|
||||
doc = r.get("document", "")
|
||||
page = r.get("page", "")
|
||||
citation = r.get("citation", "")
|
||||
ref_lines.append(f"- [{doc}, p.{page}] {citation}")
|
||||
else:
|
||||
ref_lines.append(f"- {r}")
|
||||
if ref_lines:
|
||||
sections.append(f"REFERENCES\n" + "\n".join(ref_lines))
|
||||
else:
|
||||
sections.append(f"REFERENCES\n{refs}")
|
||||
|
||||
conclusion = parsed.get("conclusion")
|
||||
if conclusion:
|
||||
sections.append(f"CONCLUSION\n{conclusion}")
|
||||
|
||||
# Avertissements sur les références non vérifiables
|
||||
if ref_warnings:
|
||||
warning_text = "\n".join(f"- {w}" for w in ref_warnings)
|
||||
sections.append(f"AVERTISSEMENT — REFERENCES NON VÉRIFIÉES\n{warning_text}")
|
||||
|
||||
return "\n\n".join(sections)
|
||||
Reference in New Issue
Block a user