Dom/t2a_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: quality_tier CPAM (A/B/C) + requires_review + warnings catégorisés

Browse Source 
- ControleCPAM enrichi : quality_tier, requires_review, quality_warnings
- _assess_quality_tier() : classification basée sur score adversarial + warnings
  - Tier C (requires_review) : score <4, code hors périmètre, >2 preuves non traçables
  - Tier B : score 4-6, warnings mineurs
  - Tier A : score >=7, 0 critique
- _format_response() : bandeau "REVUE MANUELLE REQUISE" pour tier C,
  sections CRITIQUES/MINEURS séparées
- Badge qualité dans le viewer CPAM (vert A / orange B / rouge C)
- 17 tests : tier A/B/C, bandeau, séparation warnings, backward compat

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
dom
2026-02-20 11:01:21 +01:00
parent 77ffbc56d4
commit 5d5f119057
5 changed files with 404 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/config.py
View File

@@ -729,6 +729,9 @@ class ControleCPAM(BaseModel):
contre_argumentation: Optional[str] = None
response_data: Optional[dict] = None
sources_reponse: list[RAGSource] = Field(default_factory=list)
quality_tier: Optional[str] = None # "A" | "B" | "C"
requires_review: bool = False
quality_warnings: list[str] = Field(default_factory=list)
# --- Qualité / Vetos (contestabilité) ---

20
src/control/cpam_response.py
View File

@@ -27,6 +27,7 @@ from .cpam_validation import (
_validate_codes_in_response,
_build_correction_prompt,
_format_response,
_assess_quality_tier,
)
# Backward compat — sera retiré dans un commit futur
@@ -38,7 +39,7 @@ from .cpam_context import ( # noqa: F401
_build_bio_summary,
_check_das_bio_coherence,
)
from .cpam_validation import _CIM10_CODE_RE, _validate_adversarial as _validate_adversarial # noqa: F401
from .cpam_validation import _CIM10_CODE_RE, _validate_adversarial as _validate_adversarial, _assess_quality_tier as _assess_quality_tier # noqa: F401
logger = logging.getLogger(__name__)
@@ -220,8 +221,23 @@ def generate_cpam_response(
all_warnings = ref_warnings + grounding_warnings + code_warnings + adversarial_warnings
# 8c. Classification qualité (A/B/C)
tier, needs_review, cat_warnings = _assess_quality_tier(
result, ref_warnings, grounding_warnings, code_warnings, validation,
)
controle.quality_tier = tier
controle.requires_review = needs_review
controle.quality_warnings = cat_warnings
logger.info(" Qualité CPAM : tier %s, requires_review=%s, %d warnings",
tier, needs_review, len(cat_warnings))
# 9. Formater la réponse
text = _format_response(result, all_warnings)
text = _format_response(
result,
ref_warnings=all_warnings,
quality_tier=tier,
categorized_warnings=cat_warnings,
)
logger.info(" Contre-argumentation générée (%d caractères)", len(text))
return text, result, rag_sources

99
src/control/cpam_validation.py
View File

@@ -300,10 +300,91 @@ def _build_correction_prompt(
return original_prompt + correction_block
def _format_response(parsed: dict, ref_warnings: list[str] | None = None) -> str:
def _assess_quality_tier(
parsed: dict,
ref_warnings: list[str],
grounding_warnings: list[str],
code_warnings: list[str],
adversarial_result: dict | None,
) -> tuple[str, bool, list[str]]:
"""Évalue le tier qualité (A/B/C) et le flag requires_review.
Classification :
- Tier C (requires_review=True) :
score adversarial < 4 OU code_warnings > 0 OU grounding_warnings > 2
- Tier B :
score adversarial 4-6 OU ref_warnings > 0 OU grounding_warnings 1-2
- Tier A :
score adversarial >= 7, 0 warning critique, <= 1 warning mineur
Returns:
(tier, requires_review, categorized_warnings)
"""
categorized: list[str] = []
score = adversarial_result.get("score_confiance", -1) if adversarial_result else -1
has_critical = False
minor_count = 0
# --- Warnings critiques ---
for w in code_warnings:
categorized.append(f"[CRITIQUE] {w}")
has_critical = True
if score != -1 and score <= 3:
categorized.append(f"[CRITIQUE] Score adversarial très bas : {score}/10")
has_critical = True
if len(grounding_warnings) > 2:
for w in grounding_warnings:
categorized.append(f"[CRITIQUE] {w}")
has_critical = True
elif grounding_warnings:
for w in grounding_warnings:
categorized.append(f"[MINEUR] {w}")
minor_count += 1
# --- Warnings mineurs ---
for w in ref_warnings:
categorized.append(f"[MINEUR] {w}")
minor_count += 1
if adversarial_result and not adversarial_result.get("coherent", True):
for e in adversarial_result.get("erreurs", []):
if isinstance(e, str) and e.strip():
categorized.append(f"[MINEUR] Incohérence : {e}")
minor_count += 1
if score != -1 and 4 <= score <= 6:
categorized.append(f"[MINEUR] Score adversarial moyen : {score}/10")
minor_count += 1
# --- Classification ---
if has_critical or (score != -1 and score < 4):
tier = "C"
requires_review = True
elif minor_count > 0 or (score != -1 and 4 <= score <= 6):
tier = "B"
requires_review = False
else:
tier = "A"
requires_review = False
return tier, requires_review, categorized
def _format_response(
parsed: dict,
ref_warnings: list[str] | None = None,
quality_tier: str | None = None,
categorized_warnings: list[str] | None = None,
) -> str:
"""Formate la réponse LLM en texte lisible."""
sections = []
# Bandeau qualité si tier C
if quality_tier == "C":
sections.append("⚠ REVUE MANUELLE REQUISE (Qualité : C)")
analyse = parsed.get("analyse_contestation")
if analyse:
sections.append(f"ANALYSE DE LA CONTESTATION\n{analyse}")
@@ -368,8 +449,20 @@ def _format_response(parsed: dict, ref_warnings: list[str] | None = None) -> str
if conclusion:
sections.append(f"CONCLUSION\n{conclusion}")
# Avertissements sur les références non vérifiables
if ref_warnings:
# Avertissements catégorisés (nouveau format)
if categorized_warnings:
critiques = [w for w in categorized_warnings if w.startswith("[CRITIQUE]")]
mineurs = [w for w in categorized_warnings if w.startswith("[MINEUR]")]
if critiques:
sections.append(
"AVERTISSEMENTS CRITIQUES\n" + "\n".join(f"- {w}" for w in critiques)
)
if mineurs:
sections.append(
"AVERTISSEMENTS MINEURS\n" + "\n".join(f"- {w}" for w in mineurs)
)
elif ref_warnings:
# Fallback ancien format
warning_text = "\n".join(f"- {w}" for w in ref_warnings)
sections.append(f"AVERTISSEMENT — REFERENCES NON VÉRIFIÉES\n{warning_text}")

12
src/viewer/templates/cpam.html
View File

@@ -33,6 +33,7 @@
<tr>
<th>Dossier</th>
<th>OGC</th>
<th>Qualité</th>
<th>Titre</th>
<th>Décision</th>
<th>Codes contestés</th>
@@ -51,6 +52,17 @@
{% endif %}
</td>
<td style="font-weight:600;">{{ c.ctrl.numero_ogc }}</td>
<td style="text-align:center;">
{% if c.ctrl.quality_tier == 'A' %}
<span class="badge" style="background:#2ecc71;color:#fff;font-weight:700;font-size:0.8rem;padding:3px 10px;">A</span>
{% elif c.ctrl.quality_tier == 'B' %}
<span class="badge" style="background:#f39c12;color:#fff;font-weight:700;font-size:0.8rem;padding:3px 10px;">B</span>
{% elif c.ctrl.quality_tier == 'C' %}
<span class="badge" style="background:#e74c3c;color:#fff;font-weight:700;font-size:0.8rem;padding:3px 10px;">C</span>
{% else %}
<span style="color:#94a3b8;font-size:0.7rem;">—</span>
{% endif %}
</td>
<td style="max-width:200px;">{{ c.ctrl.titre }}</td>
<td>
{% if 'retient' in c.ctrl.decision_ucr|lower %}

277
tests/test_cpam_response.py
View File

@@ -30,6 +30,7 @@ from src.control.cpam_response import (
_validate_codes_in_response,
_validate_grounding,
_validate_references,
_assess_quality_tier,
generate_cpam_response,
)
@@ -1287,7 +1288,7 @@ class TestValidateAdversarial:
text, response_data, sources = generate_cpam_response(dossier, controle)
assert "Antibiotiques mentionnés" in text
assert "Score de confiance" in text
assert "Score adversarial" in text
def test_adversarial_empty_tag_map(self):
"""Dossier sans tags → validation fonctionne quand même."""
@@ -1438,7 +1439,7 @@ class TestBuildBioSummary:
dossier = DossierMedical(
source_file="test.pdf",
biologie_cle=[
BiologieCle(test="Ferritine", valeur="15 µg/L", anomalie=True),
BiologieCle(test="Vitamine D", valeur="15 ng/mL", anomalie=True),
],
)
summary = _build_bio_summary(dossier)
@@ -1677,3 +1678,275 @@ class TestCorrectionLoop:
assert "CRP citée à 250" in correction
assert "Prompt d'argumentation original" in correction
assert "Corrige UNIQUEMENT" in correction
class TestAssessQualityTier:
"""Tests pour la classification qualité CPAM (A/B/C)."""
def test_tier_a_no_warnings_high_score(self):
"""0 warning, score adversarial >= 7 → tier A, requires_review=False."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[],
code_warnings=[],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 9},
)
assert tier == "A"
assert review is False
assert len(warnings) == 0
def test_tier_b_ref_warnings(self):
"""Warnings de référence → tier B."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=["Référence non vérifiable : Manuel Inventé"],
grounding_warnings=[],
code_warnings=[],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 8},
)
assert tier == "B"
assert review is False
assert any("[MINEUR]" in w for w in warnings)
def test_tier_b_medium_adversarial_score(self):
"""Score adversarial 4-6 → tier B."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[],
code_warnings=[],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 5},
)
assert tier == "B"
assert review is False
def test_tier_b_one_grounding_warning(self):
"""1 preuve non traçable → tier B (mineur)."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=["Preuve [BIO-99] non traçable"],
code_warnings=[],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 8},
)
assert tier == "B"
assert review is False
assert any("[MINEUR]" in w for w in warnings)
def test_tier_c_code_warnings(self):
"""Code hors périmètre → tier C, requires_review=True."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[],
code_warnings=["Code Z45.8 hors périmètre dossier/UCR"],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 7},
)
assert tier == "C"
assert review is True
assert any("[CRITIQUE]" in w for w in warnings)
def test_tier_c_low_adversarial_score(self):
"""Score adversarial < 4 → tier C."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[],
code_warnings=[],
adversarial_result={"coherent": False, "erreurs": ["Bio inventée"], "score_confiance": 2},
)
assert tier == "C"
assert review is True
assert any("[CRITIQUE]" in w for w in warnings)
def test_tier_c_many_grounding_warnings(self):
"""3+ preuves non traçables → tier C (critique)."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[
"Preuve [BIO-1] non traçable",
"Preuve [BIO-2] non traçable",
"Preuve [BIO-3] non traçable",
],
code_warnings=[],
adversarial_result={"coherent": True, "erreurs": [], "score_confiance": 7},
)
assert tier == "C"
assert review is True
def test_tier_a_no_adversarial(self):
"""Pas de validation adversariale (None) + 0 warnings → tier A."""
tier, review, warnings = _assess_quality_tier(
parsed={},
ref_warnings=[],
grounding_warnings=[],
code_warnings=[],
adversarial_result=None,
)
assert tier == "A"
assert review is False
class TestFormatResponseCategorized:
"""Tests pour le formatage avec warnings catégorisés et quality_tier."""
def test_tier_c_banner(self):
"""Tier C → bandeau REVUE MANUELLE REQUISE."""
text = _format_response(
{"conclusion": "Conclusion..."},
quality_tier="C",
categorized_warnings=["[CRITIQUE] Code hors périmètre"],
)
assert "REVUE MANUELLE REQUISE" in text
assert "Qualité : C" in text
assert "AVERTISSEMENTS CRITIQUES" in text
def test_tier_a_no_banner(self):
"""Tier A → pas de bandeau."""
text = _format_response(
{"conclusion": "Conclusion..."},
quality_tier="A",
categorized_warnings=[],
)
assert "REVUE MANUELLE REQUISE" not in text
def test_warnings_separated(self):
"""Warnings critiques et mineurs dans des sections distinctes."""
text = _format_response(
{"conclusion": "Conclusion..."},
quality_tier="C",
categorized_warnings=[
"[CRITIQUE] Code Z45.8 hors périmètre",
"[MINEUR] Référence non vérifiable",
],
)
assert "AVERTISSEMENTS CRITIQUES" in text
assert "AVERTISSEMENTS MINEURS" in text
assert text.index("CRITIQUES") < text.index("MINEURS")
def test_backward_compat_old_ref_warnings(self):
"""Sans categorized_warnings, fallback sur ref_warnings."""
text = _format_response(
{"conclusion": "Conclusion..."},
ref_warnings=["Référence non vérifiable : X"],
)
assert "AVERTISSEMENT — REFERENCES NON VÉRIFIÉES" in text
class TestCheckDasBioCoherenceExtended:
"""Tests pour les nouveaux patterns DAS/bio (Phase 5)."""
def test_sepsis_with_normal_crp(self):
"""DAS 'sepsis' mais CRP normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Sepsis sévère", cim10_suggestion="A41.9"),
],
biologie_cle=[
BiologieCle(test="CRP", valeur="3", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
assert any("Sepsis" in w or "sepsis" in w for w in warnings)
def test_infarctus_with_normal_troponine(self):
"""DAS 'infarctus' mais troponine normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Infarctus du myocarde", cim10_suggestion="I21.9"),
],
biologie_cle=[
BiologieCle(test="Troponine", valeur="0.01", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
def test_infarctus_with_high_troponine_ok(self):
"""DAS 'infarctus' + troponine élevée → pas d'incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Infarctus du myocarde", cim10_suggestion="I21.9"),
],
biologie_cle=[
BiologieCle(test="Troponine", valeur="0.5", anomalie=True),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) == 0
def test_denutrition_with_normal_albumine(self):
"""DAS 'dénutrition' mais albumine normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Dénutrition sévère", cim10_suggestion="E43"),
],
biologie_cle=[
BiologieCle(test="Albumine", valeur="42", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
def test_hypothyroidie_with_normal_tsh(self):
"""DAS 'hypothyroïdie' mais TSH normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Hypothyroïdie", cim10_suggestion="E03.9"),
],
biologie_cle=[
BiologieCle(test="TSH", valeur="2.5", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
def test_diabete_with_normal_glycemie(self):
"""DAS 'diabète' mais glycémie normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Diabète de type 2", cim10_suggestion="E11.9"),
],
biologie_cle=[
BiologieCle(test="Glycémie", valeur="4.5", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
def test_embolie_pulmonaire_with_normal_d_dimeres(self):
"""DAS 'embolie pulmonaire' mais D-dimères normaux → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Embolie pulmonaire", cim10_suggestion="I26.9"),
],
biologie_cle=[
BiologieCle(test="D-dimères", valeur="200", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1
def test_insuffisance_renale_with_normal_creatinine(self):
"""DAS 'insuffisance rénale' mais créatinine normale → incohérence."""
dossier = DossierMedical(
source_file="test.pdf",
diagnostics_associes=[
Diagnostic(texte="Insuffisance rénale aiguë", cim10_suggestion="N17.9"),
],
biologie_cle=[
BiologieCle(test="Créatinine", valeur="80", anomalie=False),
],
)
warnings = _check_das_bio_coherence(dossier)
assert len(warnings) >= 1