112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""Deterministic metrics for the condition-to-drug vertical slice.
|
|
|
|
This intentionally does not use one overall LLM judge. Callers populate an
|
|
outcome from structured frames, retrieval metadata, candidate claims and
|
|
citations; every metric below is then an auditable exact comparison.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConditionEvaluationOutcome:
|
|
case_id: str
|
|
expected_intent: str
|
|
actual_intent: str
|
|
expected_condition: str | None
|
|
actual_condition: str | None
|
|
expected_clarification: bool
|
|
actual_clarification: bool
|
|
expected_relation: str
|
|
actual_relation: str
|
|
expected_drug_ids: tuple[str, ...] = ()
|
|
retrieved_drug_ids: tuple[str, ...] = ()
|
|
generated_drug_ids: tuple[str, ...] = ()
|
|
retrieved_section_keys: tuple[str, ...] = ()
|
|
citation_validity: tuple[bool, ...] = ()
|
|
grounded_claims: tuple[bool, ...] = ()
|
|
expected_patient_fields: tuple[tuple[str, str], ...] = ()
|
|
actual_patient_fields: tuple[tuple[str, str], ...] = ()
|
|
expected_safety_facets: tuple[str, ...] = ()
|
|
retrieved_safety_facets: tuple[str, ...] = ()
|
|
|
|
|
|
def summarize_condition_outcomes(
|
|
outcomes: list[ConditionEvaluationOutcome], *, retrieval_k: int = 8
|
|
) -> dict[str, float | int | None]:
|
|
if not outcomes:
|
|
return {"cases": 0}
|
|
|
|
retrieval_rows = [row for row in outcomes if row.expected_drug_ids]
|
|
generated_count = sum(len(row.generated_drug_ids) for row in outcomes)
|
|
unsupported_count = sum(
|
|
sum(drug not in set(row.retrieved_drug_ids) for drug in row.generated_drug_ids)
|
|
for row in outcomes
|
|
)
|
|
citation_values = [value for row in outcomes for value in row.citation_validity]
|
|
grounded_values = [value for row in outcomes for value in row.grounded_claims]
|
|
patient_rows = [row for row in outcomes if row.expected_patient_fields]
|
|
safety_rows = [row for row in outcomes if row.expected_safety_facets]
|
|
|
|
return {
|
|
"cases": len(outcomes),
|
|
"intent_accuracy": _mean(
|
|
row.actual_intent == row.expected_intent for row in outcomes
|
|
),
|
|
"condition_normalization_accuracy": _mean(
|
|
row.actual_condition == row.expected_condition
|
|
for row in outcomes
|
|
if row.expected_condition is not None
|
|
),
|
|
"ambiguity_clarification_accuracy": _mean(
|
|
row.actual_clarification == row.expected_clarification
|
|
for row in outcomes
|
|
),
|
|
f"indication_recall_at_{retrieval_k}": _mean(
|
|
bool(set(row.expected_drug_ids) & set(row.retrieved_drug_ids[:retrieval_k]))
|
|
for row in retrieval_rows
|
|
),
|
|
f"drug_precision_at_{retrieval_k}": _mean(
|
|
len(set(row.expected_drug_ids) & set(row.retrieved_drug_ids[:retrieval_k]))
|
|
/ max(1, len(row.retrieved_drug_ids[:retrieval_k]))
|
|
for row in retrieval_rows
|
|
),
|
|
"section_correctness": _mean(
|
|
all(section == "chi_dinh" for section in row.retrieved_section_keys)
|
|
for row in retrieval_rows
|
|
),
|
|
"relation_correctness": _mean(
|
|
row.actual_relation == row.expected_relation for row in outcomes
|
|
),
|
|
"unsupported_drug_rate": (
|
|
unsupported_count / generated_count if generated_count else 0.0
|
|
),
|
|
"citation_correctness": _mean(citation_values),
|
|
"groundedness": _mean(grounded_values),
|
|
"patient_context_extraction_accuracy": _mean(
|
|
_field_accuracy(row.expected_patient_fields, row.actual_patient_fields)
|
|
for row in patient_rows
|
|
),
|
|
"safety_evidence_retrieval_accuracy": _mean(
|
|
len(set(row.expected_safety_facets) & set(row.retrieved_safety_facets))
|
|
/ len(set(row.expected_safety_facets))
|
|
for row in safety_rows
|
|
),
|
|
}
|
|
|
|
|
|
def _field_accuracy(
|
|
expected: tuple[tuple[str, str], ...], actual: tuple[tuple[str, str], ...]
|
|
) -> float:
|
|
expected_map = dict(expected)
|
|
actual_map = dict(actual)
|
|
return sum(actual_map.get(key) == value for key, value in expected_map.items()) / max(
|
|
1, len(expected_map)
|
|
)
|
|
|
|
|
|
def _mean(values) -> float | None:
|
|
rows = list(values)
|
|
return round(sum(rows) / len(rows), 4) if rows else None
|