416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""Small clinical query contracts for condition-centric formulary retrieval.
|
|
|
|
This module contains no treatment knowledge and no disease-to-drug map. It
|
|
only preserves facts the clinician supplied, normalises a deliberately small
|
|
set of unambiguous Vietnamese aliases/abbreviations, and groups retrieved
|
|
evidence by safety facet. Medication candidates always originate in the
|
|
corpus's ``chi_dinh`` sections.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from enum import StrEnum
|
|
|
|
from .models import Evidence
|
|
from .text import normalize_name
|
|
|
|
|
|
class ConditionRelation(StrEnum):
|
|
INDICATION = "indication"
|
|
ADVERSE_EFFECT = "adverse_effect"
|
|
CONTRAINDICATION = "contraindication"
|
|
UNKNOWN = "unknown"
|
|
|
|
|
|
class CaseContextAction(StrEnum):
|
|
NONE = "none"
|
|
CONTINUE = "continue"
|
|
NEW = "new"
|
|
|
|
|
|
class CandidateStatus(StrEnum):
|
|
SUPPORTED = "supported"
|
|
SUPPORTED_WITH_CAUTION = "supported_with_caution"
|
|
REQUIRES_ADDITIONAL_INFORMATION = "requires_additional_information"
|
|
INSUFFICIENT_EVIDENCE = "insufficient_evidence"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConditionQuery:
|
|
original_query: str
|
|
normalized_condition: str
|
|
subtype: str | None = None
|
|
qualifiers: tuple[str, ...] = ()
|
|
ambiguous: bool = False
|
|
clarify_question: str | None = None
|
|
|
|
@property
|
|
def retrieval_text(self) -> str:
|
|
"""Most specific safe text to search, without inventing a subtype."""
|
|
if self.subtype and normalize_name(self.subtype) not in normalize_name(
|
|
self.normalized_condition
|
|
):
|
|
return f"{self.normalized_condition} {self.subtype}".strip()
|
|
return self.normalized_condition
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RenalContext:
|
|
description: str | None = None
|
|
ckd_stage: str | None = None
|
|
egfr: str | None = None
|
|
crcl: str | None = None
|
|
creatinine: str | None = None
|
|
|
|
@property
|
|
def present(self) -> bool:
|
|
return any((self.description, self.ckd_stage, self.egfr, self.crcl, self.creatinine))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HepaticContext:
|
|
description: str | None = None
|
|
child_pugh: str | None = None
|
|
ast: str | None = None
|
|
alt: str | None = None
|
|
bilirubin: str | None = None
|
|
|
|
@property
|
|
def present(self) -> bool:
|
|
return any((self.description, self.child_pugh, self.ast, self.alt, self.bilirubin))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PatientContext:
|
|
age_text: str | None = None
|
|
sex: str | None = None
|
|
weight_kg: float | None = None
|
|
primary_condition: str | None = None
|
|
comorbidities: tuple[str, ...] = ()
|
|
allergies: tuple[str, ...] = ()
|
|
previous_adverse_reactions: tuple[str, ...] = ()
|
|
current_medications: tuple[str, ...] = ()
|
|
pregnancy_status: str | None = None
|
|
breastfeeding: bool | None = None
|
|
renal: RenalContext = field(default_factory=RenalContext)
|
|
hepatic: HepaticContext = field(default_factory=HepaticContext)
|
|
relevant_labs: tuple[str, ...] = ()
|
|
treatment_history: tuple[str, ...] = ()
|
|
|
|
@property
|
|
def present(self) -> bool:
|
|
return any((
|
|
self.age_text,
|
|
self.sex,
|
|
self.weight_kg is not None,
|
|
self.primary_condition,
|
|
self.comorbidities,
|
|
self.allergies,
|
|
self.previous_adverse_reactions,
|
|
self.current_medications,
|
|
self.pregnancy_status,
|
|
self.breastfeeding is not None,
|
|
self.renal.present,
|
|
self.hepatic.present,
|
|
self.relevant_labs,
|
|
self.treatment_history,
|
|
))
|
|
|
|
@property
|
|
def requires_safety_review(self) -> bool:
|
|
"""Whether this condition lookup needs the bounded stage-2 path."""
|
|
return any((
|
|
self.age_text,
|
|
self.weight_kg is not None,
|
|
self.comorbidities,
|
|
self.allergies,
|
|
self.previous_adverse_reactions,
|
|
self.current_medications,
|
|
self.pregnancy_status,
|
|
self.breastfeeding is not None,
|
|
self.renal.present,
|
|
self.hepatic.present,
|
|
self.relevant_labs,
|
|
self.treatment_history,
|
|
))
|
|
|
|
def merged_with(self, earlier: "PatientContext") -> "PatientContext":
|
|
"""Prefer facts in the current turn and retain earlier case facts.
|
|
|
|
Tuple fields are unioned in conversation order. This is called only
|
|
after query understanding explicitly marks the turn as the same case;
|
|
it must never decide case continuity itself.
|
|
"""
|
|
return PatientContext(
|
|
age_text=self.age_text or earlier.age_text,
|
|
sex=self.sex or earlier.sex,
|
|
weight_kg=self.weight_kg if self.weight_kg is not None else earlier.weight_kg,
|
|
primary_condition=self.primary_condition or earlier.primary_condition,
|
|
comorbidities=_merge_tuple(earlier.comorbidities, self.comorbidities),
|
|
allergies=_merge_tuple(earlier.allergies, self.allergies),
|
|
previous_adverse_reactions=_merge_tuple(
|
|
earlier.previous_adverse_reactions, self.previous_adverse_reactions
|
|
),
|
|
current_medications=_merge_tuple(
|
|
earlier.current_medications, self.current_medications
|
|
),
|
|
pregnancy_status=self.pregnancy_status or earlier.pregnancy_status,
|
|
breastfeeding=(
|
|
self.breastfeeding
|
|
if self.breastfeeding is not None
|
|
else earlier.breastfeeding
|
|
),
|
|
renal=_merge_renal(self.renal, earlier.renal),
|
|
hepatic=_merge_hepatic(self.hepatic, earlier.hepatic),
|
|
relevant_labs=_merge_tuple(earlier.relevant_labs, self.relevant_labs),
|
|
treatment_history=_merge_tuple(
|
|
earlier.treatment_history, self.treatment_history
|
|
),
|
|
)
|
|
|
|
def safety_query(self) -> str:
|
|
"""Search text made only from supplied/normalised case facts."""
|
|
parts = [
|
|
*self.comorbidities,
|
|
*self.allergies,
|
|
*self.previous_adverse_reactions,
|
|
*self.current_medications,
|
|
*self.relevant_labs,
|
|
self.renal.description,
|
|
self.renal.ckd_stage,
|
|
self.renal.egfr,
|
|
self.renal.crcl,
|
|
self.renal.creatinine,
|
|
self.hepatic.description,
|
|
self.hepatic.child_pugh,
|
|
self.hepatic.ast,
|
|
self.hepatic.alt,
|
|
self.hepatic.bilirubin,
|
|
]
|
|
if self.pregnancy_status:
|
|
parts.extend(("mang thai", self.pregnancy_status))
|
|
if self.breastfeeding is True:
|
|
parts.append("cho con bú")
|
|
if self.allergies or self.previous_adverse_reactions:
|
|
parts.extend(("dị ứng", "quá mẫn"))
|
|
if self.renal.present:
|
|
parts.extend(("suy thận", "chức năng thận", "độ thanh thải creatinin"))
|
|
if self.hepatic.present:
|
|
parts.extend(("suy gan", "chức năng gan"))
|
|
if self.age_text:
|
|
parts.append(self.age_text)
|
|
digits = "".join(char if char.isdigit() else " " for char in self.age_text)
|
|
values = [int(item) for item in digits.split() if item.isdigit()]
|
|
if values and values[0] >= 65:
|
|
parts.append("người cao tuổi")
|
|
return ". ".join(str(item).strip() for item in parts if str(item or "").strip())
|
|
|
|
def interaction_query(self) -> str:
|
|
return ". ".join(self.current_medications)
|
|
|
|
def warning_query(self) -> str:
|
|
parts = [
|
|
*self.comorbidities,
|
|
*self.allergies,
|
|
*self.previous_adverse_reactions,
|
|
*self.relevant_labs,
|
|
self.renal.description,
|
|
self.renal.ckd_stage,
|
|
self.renal.egfr,
|
|
self.renal.crcl,
|
|
self.renal.creatinine,
|
|
self.hepatic.description,
|
|
self.hepatic.child_pugh,
|
|
self.hepatic.ast,
|
|
self.hepatic.alt,
|
|
self.hepatic.bilirubin,
|
|
self.age_text,
|
|
]
|
|
if self.allergies or self.previous_adverse_reactions:
|
|
parts.extend(("dị ứng", "quá mẫn"))
|
|
if self.renal.present:
|
|
parts.extend(("suy thận", "chức năng thận", "độ thanh thải creatinin"))
|
|
if self.hepatic.present:
|
|
parts.extend(("suy gan", "chức năng gan"))
|
|
if self.age_text:
|
|
parts.append("người cao tuổi")
|
|
return ". ".join(str(item).strip() for item in parts if str(item or "").strip())
|
|
|
|
def dosage_context_query(self) -> str:
|
|
parts = [
|
|
self.renal.description,
|
|
self.renal.ckd_stage,
|
|
self.renal.egfr,
|
|
self.renal.crcl,
|
|
self.renal.creatinine,
|
|
self.hepatic.description,
|
|
self.hepatic.child_pugh,
|
|
self.age_text,
|
|
]
|
|
if self.renal.present:
|
|
parts.extend(("suy thận", "độ thanh thải creatinin"))
|
|
if self.hepatic.present:
|
|
parts.append("suy gan")
|
|
if self.age_text:
|
|
parts.append("người cao tuổi")
|
|
return ". ".join(str(item).strip() for item in parts if str(item or "").strip())
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MedicationCandidateAssessment:
|
|
drug_id: str
|
|
drug_name: str
|
|
indication_supported: bool
|
|
indication_evidence: tuple[Evidence, ...] = ()
|
|
contraindication_evidence: tuple[Evidence, ...] = ()
|
|
precaution_evidence: tuple[Evidence, ...] = ()
|
|
interaction_evidence: tuple[Evidence, ...] = ()
|
|
renal_evidence: tuple[Evidence, ...] = ()
|
|
hepatic_evidence: tuple[Evidence, ...] = ()
|
|
pregnancy_evidence: tuple[Evidence, ...] = ()
|
|
breastfeeding_evidence: tuple[Evidence, ...] = ()
|
|
age_evidence: tuple[Evidence, ...] = ()
|
|
dose_evidence: tuple[Evidence, ...] = ()
|
|
status: CandidateStatus = CandidateStatus.SUPPORTED
|
|
|
|
@property
|
|
def evidence(self) -> tuple[Evidence, ...]:
|
|
ordered = (
|
|
self.indication_evidence
|
|
+ self.contraindication_evidence
|
|
+ self.precaution_evidence
|
|
+ self.interaction_evidence
|
|
+ self.renal_evidence
|
|
+ self.hepatic_evidence
|
|
+ self.pregnancy_evidence
|
|
+ self.breastfeeding_evidence
|
|
+ self.age_evidence
|
|
+ self.dose_evidence
|
|
)
|
|
output: list[Evidence] = []
|
|
seen: set[str] = set()
|
|
for item in ordered:
|
|
if item.evidence_id in seen:
|
|
continue
|
|
seen.add(item.evidence_id)
|
|
output.append(item)
|
|
return tuple(output)
|
|
|
|
|
|
class ConditionNormalizer:
|
|
"""Conservative terminology normalisation, never treatment mapping.
|
|
|
|
Only unambiguous aliases required by the professional UX are canonicalised.
|
|
Every unknown phrase is preserved, so an ambiguous abbreviation is never
|
|
silently expanded by this class.
|
|
"""
|
|
|
|
_ALIASES = {
|
|
"tha": "tăng huyết áp",
|
|
"cao huyet ap": "tăng huyết áp",
|
|
"tang huyet ap": "tăng huyết áp",
|
|
"gout": "gút",
|
|
"benh gout": "gút",
|
|
"benh gut": "gút",
|
|
"gut": "gút",
|
|
}
|
|
_BROAD = frozenset({"viem gan", "ung thu", "nhiem trung", "nhiem khuan"})
|
|
_BROAD_QUESTIONS = {
|
|
"viem gan": "Bạn đang hỏi viêm gan A, B, C hay loại viêm gan nào?",
|
|
"ung thu": "Bạn đang hỏi loại ung thư cụ thể nào?",
|
|
"nhiem trung": "Bạn đang hỏi nhiễm trùng ở vị trí nào và do tác nhân nào đã xác định?",
|
|
"nhiem khuan": "Bạn đang hỏi nhiễm khuẩn ở vị trí nào và do tác nhân nào đã xác định?",
|
|
}
|
|
|
|
def detect_known_alias(self, original_query: str) -> ConditionQuery | None:
|
|
"""Find a conservative condition alias anywhere in a clinician turn."""
|
|
text = normalize_name(original_query)
|
|
matches: list[tuple[int, int, str]] = []
|
|
for alias in self._ALIASES:
|
|
match = re.search(rf"(?:^| ){re.escape(alias)}(?:$| )", text)
|
|
if match:
|
|
matches.append((match.start(), -len(alias), alias))
|
|
if not matches:
|
|
return None
|
|
_, _, alias = min(matches)
|
|
return self.normalize(original_query, alias)
|
|
|
|
def detect_broad_question(self, original_query: str) -> ConditionQuery | None:
|
|
"""Recognise only a bare broad condition followed by a drug request.
|
|
|
|
A subtype/site between the broad noun and the request deliberately
|
|
prevents a match (for example ``nhiễm trùng đường tiết niệu``), so the
|
|
guard does not over-clarify a condition the clinician already narrowed.
|
|
"""
|
|
text = normalize_name(original_query)
|
|
request = (
|
|
r"(?:dung thuoc gi|nen dung thuoc nao|dieu tri (?:bang )?thuoc nao|"
|
|
r"co thuoc nao(?: dieu tri)?|thuoc nao dieu tri)"
|
|
)
|
|
for broad in sorted(self._BROAD, key=len, reverse=True):
|
|
patterns = (
|
|
rf"(?:benh )?{re.escape(broad)} {request}",
|
|
rf"thuoc nao (?:dieu tri )?(?:benh )?{re.escape(broad)}",
|
|
)
|
|
if any(re.fullmatch(pattern, text) for pattern in patterns):
|
|
return ConditionQuery(
|
|
original_query=original_query,
|
|
normalized_condition=broad,
|
|
ambiguous=True,
|
|
clarify_question=self._BROAD_QUESTIONS[broad],
|
|
)
|
|
return None
|
|
|
|
def normalize(
|
|
self,
|
|
original_query: str,
|
|
condition: str,
|
|
*,
|
|
subtype: str | None = None,
|
|
qualifiers: tuple[str, ...] = (),
|
|
ambiguous: bool = False,
|
|
clarify_question: str | None = None,
|
|
) -> ConditionQuery:
|
|
cleaned = " ".join(condition.split())
|
|
key = normalize_name(cleaned)
|
|
canonical = self._ALIASES.get(key, cleaned)
|
|
broad = normalize_name(canonical) in self._BROAD and not subtype
|
|
return ConditionQuery(
|
|
original_query=original_query,
|
|
normalized_condition=canonical,
|
|
subtype=subtype,
|
|
qualifiers=qualifiers,
|
|
ambiguous=ambiguous or broad,
|
|
clarify_question=(
|
|
clarify_question
|
|
or self._BROAD_QUESTIONS.get(normalize_name(canonical))
|
|
if ambiguous or broad
|
|
else None
|
|
),
|
|
)
|
|
|
|
|
|
def _merge_tuple(left: tuple[str, ...], right: tuple[str, ...]) -> tuple[str, ...]:
|
|
return tuple(dict.fromkeys((*left, *right)))
|
|
|
|
|
|
def _merge_renal(current: RenalContext, earlier: RenalContext) -> RenalContext:
|
|
return RenalContext(
|
|
description=current.description or earlier.description,
|
|
ckd_stage=current.ckd_stage or earlier.ckd_stage,
|
|
egfr=current.egfr or earlier.egfr,
|
|
crcl=current.crcl or earlier.crcl,
|
|
creatinine=current.creatinine or earlier.creatinine,
|
|
)
|
|
|
|
|
|
def _merge_hepatic(current: HepaticContext, earlier: HepaticContext) -> HepaticContext:
|
|
return HepaticContext(
|
|
description=current.description or earlier.description,
|
|
child_pugh=current.child_pugh or earlier.child_pugh,
|
|
ast=current.ast or earlier.ast,
|
|
alt=current.alt or earlier.alt,
|
|
bilirubin=current.bilirubin or earlier.bilirubin,
|
|
)
|