Spaces:
Sleeping
Sleeping
| """Evaluation metrics for concept retrieval and code classification against | |
| expert-curated gold-standard phenotype codelists.""" | |
| from __future__ import annotations | |
| from dataclasses import asdict, dataclass | |
| import json | |
| import pandas as pd | |
| from src.data.preprocessing import standardize_code | |
| LABELS = ("Narrow", "Possible", "Exclude") | |
| class RetrievalMetrics: | |
| true_positives: int | |
| false_positives: int | |
| false_negatives: int | |
| sensitivity: float | |
| precision: float | |
| f1: float | |
| class ClassificationMetrics: | |
| accuracy: float | |
| cohens_kappa: float | |
| true_negatives: int | |
| sensitivity: float | |
| precision: float | |
| f1: float | |
| macro_sensitivity: float | |
| macro_precision: float | |
| macro_f1: float | |
| per_label: dict[str, dict[str, float]] | |
| class NarrowLossBreakdown: | |
| """Where available gold-Narrow codes were lost end to end.""" | |
| gold_narrow_available: int | |
| correctly_classified: int | |
| misclassified: int | |
| not_classified: int | |
| total_missed: int | |
| end_to_end_recall: float | |
| class GPTReviewEvaluation: | |
| """Gold-standard impact of the final low-confidence GPT review.""" | |
| candidates_reviewed: int | |
| labels_changed: int | |
| corrected_changes: int | |
| harmful_changes: int | |
| unchanged: int | |
| correct_before: int | |
| correct_after: int | |
| net_correct_change: int | |
| reviewed_accuracy_before: float | |
| reviewed_accuracy_after: float | |
| def _standardized_key(code: str, vocabulary: str) -> tuple[str, str]: | |
| return standardize_code(code, vocabulary), (vocabulary or "").strip().upper() | |
| def filter_records_by_vocabulary(records: list[dict], vocabularies: list[str]) -> list[dict]: | |
| """Keep only records whose ``vocabulary`` is in *vocabularies* (case-insensitive).""" | |
| vocab_set = {v.strip().upper() for v in vocabularies} | |
| return [r for r in records if (r.get("vocabulary") or "").strip().upper() in vocab_set] | |
| def filter_gold_by_vocabulary(gold_labels: pd.DataFrame, vocabularies: list[str]) -> pd.DataFrame: | |
| """Keep only gold-label rows whose ``vocabulary`` is in *vocabularies* (case-insensitive).""" | |
| vocab_set = {v.strip().upper() for v in vocabularies} | |
| mask = gold_labels["vocabulary"].str.strip().str.upper().isin(vocab_set) | |
| return gold_labels[mask].reset_index(drop=True) | |
| def filter_gold_by_available_codes( | |
| gold_labels: pd.DataFrame, | |
| available_codes: pd.DataFrame, | |
| ) -> pd.DataFrame: | |
| """Restrict gold labels to codes present in the loaded terminology corpus. | |
| This prevents codes from older or otherwise unavailable terminology | |
| versions from being counted as false negatives. | |
| """ | |
| required = {"code", "vocabulary"} | |
| missing = required - set(available_codes.columns) | |
| if missing: | |
| raise ValueError(f"Available code corpus missing required columns: {missing}") | |
| available_keys = { | |
| _standardized_key(row["code"], row["vocabulary"]) | |
| for _, row in available_codes.iterrows() | |
| } | |
| mask = gold_labels.apply( | |
| lambda row: _standardized_key(row["code"], row["vocabulary"]) in available_keys, | |
| axis=1, | |
| ) | |
| return gold_labels[mask].reset_index(drop=True) | |
| def narrow_loss_breakdown( | |
| classified: list[dict], | |
| gold_labels: pd.DataFrame, | |
| ) -> NarrowLossBreakdown: | |
| """Separate gold-Narrow losses before and during LLM classification. | |
| ``not_classified`` covers gold-Narrow codes that never reached the model | |
| (for example because adaptive stopping or a hard top-k limit ended the | |
| scan). ``misclassified`` covers codes that did reach the model but were | |
| assigned a label other than Narrow. | |
| """ | |
| gold_narrow_keys = { | |
| _standardized_key(row["code"], row["vocabulary"]) | |
| for _, row in gold_labels.iterrows() | |
| if str(row["label"]).strip().title() == "Narrow" | |
| } | |
| predictions_by_key = { | |
| _standardized_key(item["code"], item["vocabulary"]): | |
| str(item.get("label", "")).strip().title() | |
| for item in classified | |
| } | |
| correctly_classified = sum( | |
| predictions_by_key.get(key) == "Narrow" for key in gold_narrow_keys | |
| ) | |
| misclassified = sum( | |
| key in predictions_by_key and predictions_by_key[key] != "Narrow" | |
| for key in gold_narrow_keys | |
| ) | |
| not_classified = len(gold_narrow_keys - predictions_by_key.keys()) | |
| total_missed = misclassified + not_classified | |
| recall = correctly_classified / len(gold_narrow_keys) if gold_narrow_keys else 0.0 | |
| return NarrowLossBreakdown( | |
| gold_narrow_available=len(gold_narrow_keys), | |
| correctly_classified=correctly_classified, | |
| misclassified=misclassified, | |
| not_classified=not_classified, | |
| total_missed=total_missed, | |
| end_to_end_recall=round(recall, 4), | |
| ) | |
| def evaluate_gpt_review( | |
| stage2_classified: list[dict], | |
| final_classified: list[dict], | |
| gold_labels: pd.DataFrame, | |
| *, | |
| confidence_threshold: float, | |
| max_candidates: int | None = None, | |
| include_possible: bool = False, | |
| ) -> GPTReviewEvaluation: | |
| """Measure whether GPT's reviewed labels improved stage-2 decisions.""" | |
| gold_by_key = { | |
| _standardized_key(row["code"], row["vocabulary"]): ( | |
| row["label"] | |
| if include_possible or row["label"] != "Possible" | |
| else "Exclude" | |
| ) | |
| for _, row in gold_labels.iterrows() | |
| } | |
| final_by_key = { | |
| _standardized_key(item["code"], item["vocabulary"]): | |
| str(item.get("label", "")).strip().title() | |
| for item in final_classified | |
| } | |
| eligible = [ | |
| item | |
| for item in stage2_classified | |
| if float(item.get("confidence", 0.0)) < confidence_threshold | |
| ] | |
| eligible.sort(key=lambda item: float(item.get("confidence", 0.0))) | |
| selected = eligible[:max_candidates] if max_candidates is not None else eligible | |
| selected_keys = { | |
| _standardized_key(item["code"], item["vocabulary"]) for item in selected | |
| } | |
| reviewed = 0 | |
| changed = 0 | |
| corrected = 0 | |
| harmful = 0 | |
| correct_before = 0 | |
| correct_after = 0 | |
| for original in stage2_classified: | |
| key = _standardized_key(original["code"], original["vocabulary"]) | |
| if key not in selected_keys: | |
| continue | |
| if key not in final_by_key: | |
| continue | |
| reviewed += 1 | |
| original_label = str(original.get("label", "")).strip().title() | |
| final_label = final_by_key[key] | |
| if not include_possible: | |
| original_label = "Exclude" if original_label == "Possible" else original_label | |
| final_label = "Exclude" if final_label == "Possible" else final_label | |
| gold_label = gold_by_key.get(key, "Exclude") | |
| was_correct = original_label == gold_label | |
| is_correct = final_label == gold_label | |
| correct_before += was_correct | |
| correct_after += is_correct | |
| if original_label != final_label: | |
| changed += 1 | |
| corrected += is_correct and not was_correct | |
| harmful += was_correct and not is_correct | |
| return GPTReviewEvaluation( | |
| candidates_reviewed=reviewed, | |
| labels_changed=changed, | |
| corrected_changes=corrected, | |
| harmful_changes=harmful, | |
| unchanged=reviewed - changed, | |
| correct_before=correct_before, | |
| correct_after=correct_after, | |
| net_correct_change=correct_after - correct_before, | |
| reviewed_accuracy_before=round(correct_before / reviewed, 4) if reviewed else 0.0, | |
| reviewed_accuracy_after=round(correct_after / reviewed, 4) if reviewed else 0.0, | |
| ) | |
| def to_predicted_codelist(classified: list[dict]) -> pd.DataFrame: | |
| """Render final classifications in the same schema as the ground-truth AESI | |
| export (``coding_system``, ``code``, ``code_name``, ``concept``, | |
| ``concept_name``, ``tags``) so predicted and reference codelists are | |
| directly comparable/exportable side by side. | |
| Parameters | |
| ---------- | |
| classified : list[dict] | |
| Each dict must contain ``code``, ``vocabulary``, ``description``, and | |
| ``label`` keys (as produced by :func:`src.llm.classify.llm_classify`). | |
| Returns | |
| ------- | |
| pd.DataFrame | |
| AESI-compatible columns. When classification annotations are supplied, | |
| confidence, explanation, manual-review flag, and review reason are | |
| appended. ``concept`` is blank because the corpus has no UMLS mapping. | |
| """ | |
| base_columns = ["coding_system", "code", "code_name", "concept", "concept_name", "tags"] | |
| annotation_columns = ["confidence", "explanation", "manual_review", "review_reason"] | |
| include_annotations = any(any(column in item for column in annotation_columns) for item in classified) | |
| columns = base_columns + annotation_columns if include_annotations else base_columns | |
| mapping_columns = ["decision_origin", "mapping_provenance", "prior_label", "prior_confidence"] | |
| include_mapping = any(c.get("decision_origin") == "codemapper" for c in classified) | |
| if include_mapping: | |
| columns += mapping_columns | |
| return pd.DataFrame( | |
| [ | |
| { | |
| "coding_system": c["vocabulary"], | |
| "code": c["code"], | |
| "code_name": c.get("description", ""), | |
| "concept": "", | |
| "concept_name": c.get("description", ""), | |
| "tags": (c.get("label") or "").lower(), | |
| **( | |
| { | |
| "confidence": c.get("confidence", ""), | |
| "explanation": c.get("explanation", ""), | |
| "manual_review": bool(c.get("manual_review", False) or c.get("decision_origin") == "codemapper"), | |
| "review_reason": c.get("review_reason") or ( | |
| "included by CodeMapper mapping; EDF scope not independently assessed" | |
| if c.get("decision_origin") == "codemapper" else "" | |
| ), | |
| } | |
| if include_annotations | |
| else {} | |
| ), | |
| **({"decision_origin": c.get("decision_origin", "llm"), | |
| "mapping_provenance": json.dumps(c.get("mapping_provenance", [])), | |
| "prior_label": c.get("prior_label"), | |
| "prior_confidence": c.get("prior_confidence")} if include_mapping else {}), | |
| } | |
| for c in classified | |
| ], | |
| columns=columns, | |
| ) | |
| def evaluate_retrieval( | |
| retrieved_codes: list[dict], | |
| gold_labels: pd.DataFrame, | |
| narrow_only_as_positive: bool = False, | |
| ) -> RetrievalMetrics: | |
| """Evaluate concept retrieval as an information-retrieval task. | |
| By default, true positives are retrieved codes labeled 'Narrow' or | |
| 'Possible' in the gold standard. With ``narrow_only_as_positive=True``, only | |
| 'Narrow' is positive. | |
| False positives are retrieved codes labeled 'Exclude' in the gold standard, | |
| or codes absent from it entirely. | |
| False negatives are positive gold-standard codes that were not retrieved. | |
| Parameters | |
| ---------- | |
| retrieved_codes : list[dict] | |
| Each dict must contain ``code`` and ``vocabulary`` keys. | |
| gold_labels : pd.DataFrame | |
| Reference standard with ``code``, ``vocabulary``, ``label`` columns. | |
| Returns | |
| ------- | |
| RetrievalMetrics | |
| """ | |
| gold_by_key = { | |
| _standardized_key(row["code"], row["vocabulary"]): row["label"] | |
| for _, row in gold_labels.iterrows() | |
| } | |
| positive_labels = {"Narrow"} if narrow_only_as_positive else {"Narrow", "Possible"} | |
| positive_keys = {k for k, label in gold_by_key.items() if label in positive_labels} | |
| retrieved_keys = { | |
| _standardized_key(r["code"], r["vocabulary"]) for r in retrieved_codes | |
| } | |
| tp = len(retrieved_keys & positive_keys) | |
| fp = len(retrieved_keys - positive_keys) | |
| fn = len(positive_keys - retrieved_keys) | |
| sensitivity = tp / (tp + fn) if (tp + fn) > 0 else 0.0 | |
| precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 | |
| f1 = ( | |
| 2 * precision * sensitivity / (precision + sensitivity) | |
| if (precision + sensitivity) > 0 | |
| else 0.0 | |
| ) | |
| return RetrievalMetrics( | |
| true_positives=tp, | |
| false_positives=fp, | |
| false_negatives=fn, | |
| sensitivity=round(sensitivity, 4), | |
| precision=round(precision, 4), | |
| f1=round(f1, 4), | |
| ) | |
| def _cohens_kappa(confusion: dict[tuple[str, str], int], labels: tuple[str, ...], n: int) -> float: | |
| if n == 0: | |
| return 0.0 | |
| observed_agreement = sum(confusion.get((lbl, lbl), 0) for lbl in labels) / n | |
| row_totals = {lbl: sum(confusion.get((lbl, o), 0) for o in labels) for lbl in labels} | |
| col_totals = {lbl: sum(confusion.get((o, lbl), 0) for o in labels) for lbl in labels} | |
| expected_agreement = sum((row_totals[lbl] * col_totals[lbl]) / n for lbl in labels) / n | |
| if expected_agreement >= 1.0: | |
| return 1.0 | |
| return (observed_agreement - expected_agreement) / (1 - expected_agreement) | |
| def evaluate_classification( | |
| classified: list[dict], | |
| gold_labels: pd.DataFrame, | |
| include_possible: bool = True, | |
| ) -> ClassificationMetrics: | |
| """Evaluate code classification (Narrow/Possible/Exclude) against gold labels. | |
| Only codes present in both the classified set and gold standard are compared | |
| (codes absent from either side cannot be scored for classification agreement). | |
| Parameters | |
| ---------- | |
| classified : list[dict] | |
| Each dict must contain ``code``, ``vocabulary``, ``label`` keys. | |
| gold_labels : pd.DataFrame | |
| Reference standard with ``code``, ``vocabulary``, ``label`` columns. | |
| Returns | |
| ------- | |
| ClassificationMetrics | |
| """ | |
| labels = LABELS if include_possible else ("Narrow", "Exclude") | |
| gold_by_key = { | |
| _standardized_key(row["code"], row["vocabulary"]): ( | |
| row["label"] if include_possible or row["label"] != "Possible" else "Exclude" | |
| ) | |
| for _, row in gold_labels.iterrows() | |
| } | |
| confusion: dict[tuple[str, str], int] = {} | |
| correct = 0 | |
| total = 0 | |
| per_label_counts = {lbl: {"tp": 0, "tn": 0, "fp": 0, "fn": 0} for lbl in labels} | |
| for item in classified: | |
| key = _standardized_key(item["code"], item["vocabulary"]) | |
| gold_label = gold_by_key.get(key) | |
| pred_label = item["label"] | |
| if not include_possible and pred_label == "Possible": | |
| pred_label = "Exclude" | |
| if gold_label is None: | |
| # A retrieved code absent from the curated positive codelist is an | |
| # implicit Exclude. Apply this consistently to both correct | |
| # Exclude predictions and false-positive Narrow/Possible results. | |
| gold_label = "Exclude" | |
| total += 1 | |
| confusion[(gold_label, pred_label)] = confusion.get((gold_label, pred_label), 0) + 1 | |
| if pred_label == gold_label: | |
| correct += 1 | |
| for label in labels: | |
| gold_positive = gold_label == label | |
| predicted_positive = pred_label == label | |
| if gold_positive and predicted_positive: | |
| per_label_counts[label]["tp"] += 1 | |
| elif not gold_positive and not predicted_positive: | |
| per_label_counts[label]["tn"] += 1 | |
| elif predicted_positive: | |
| per_label_counts[label]["fp"] += 1 | |
| else: | |
| per_label_counts[label]["fn"] += 1 | |
| accuracy = correct / total if total > 0 else 0.0 | |
| kappa = _cohens_kappa(confusion, labels, total) | |
| per_label: dict[str, dict[str, float]] = {} | |
| for lbl in labels: | |
| tp = per_label_counts[lbl]["tp"] | |
| tn = per_label_counts[lbl]["tn"] | |
| fp = per_label_counts[lbl]["fp"] | |
| fn = per_label_counts[lbl]["fn"] | |
| precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 | |
| recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0 | |
| f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0 | |
| per_label[lbl] = { | |
| "tp": tp, | |
| "tn": tn, | |
| "fp": fp, | |
| "fn": fn, | |
| "precision": round(precision, 4), | |
| "recall": round(recall, 4), | |
| "f1": round(f1, 4), | |
| } | |
| macro_labels = [ | |
| label | |
| for label, counts in per_label_counts.items() | |
| if counts["tp"] + counts["fn"] > 0 | |
| ] or list(labels) | |
| return ClassificationMetrics( | |
| accuracy=round(accuracy, 4), | |
| cohens_kappa=round(kappa, 4), | |
| true_negatives=per_label["Narrow"]["tn"], | |
| sensitivity=per_label["Narrow"]["recall"], | |
| precision=per_label["Narrow"]["precision"], | |
| f1=per_label["Narrow"]["f1"], | |
| macro_sensitivity=round(sum(per_label[label]["recall"] for label in macro_labels) / len(macro_labels), 4), | |
| macro_precision=round(sum(per_label[label]["precision"] for label in macro_labels) / len(macro_labels), 4), | |
| macro_f1=round(sum(per_label[label]["f1"] for label in macro_labels) / len(macro_labels), 4), | |
| per_label=per_label, | |
| ) | |
| def evaluate( | |
| classified: list[dict], | |
| gold_labels: pd.DataFrame, | |
| narrow_only_as_positive: bool = False, | |
| include_possible: bool = True, | |
| ) -> dict: | |
| """Compute both retrieval and classification metrics for the full pipeline output. | |
| Parameters | |
| ---------- | |
| classified : list[dict] | |
| Final classified candidates, each with ``code``, ``vocabulary``, ``label`` keys. | |
| gold_labels : pd.DataFrame | |
| Reference standard with ``code``, ``vocabulary``, ``label`` columns. | |
| Returns | |
| ------- | |
| dict | |
| Dictionary with 'retrieval' and 'classification' metric sub-dictionaries. | |
| """ | |
| retrieval_metrics = evaluate_retrieval( | |
| classified, | |
| gold_labels, | |
| narrow_only_as_positive=narrow_only_as_positive, | |
| ) | |
| classification_metrics = evaluate_classification( | |
| classified, | |
| gold_labels, | |
| include_possible=include_possible, | |
| ) | |
| return { | |
| "retrieval": asdict(retrieval_metrics), | |
| "classification": { | |
| "accuracy": classification_metrics.accuracy, | |
| "cohens_kappa": classification_metrics.cohens_kappa, | |
| "true_negatives": classification_metrics.true_negatives, | |
| "sensitivity": classification_metrics.sensitivity, | |
| "precision": classification_metrics.precision, | |
| "f1": classification_metrics.f1, | |
| "macro_sensitivity": classification_metrics.macro_sensitivity, | |
| "macro_precision": classification_metrics.macro_precision, | |
| "macro_f1": classification_metrics.macro_f1, | |
| "per_label": classification_metrics.per_label, | |
| }, | |
| } | |