Spaces:
Sleeping
Sleeping
| """FastAPI backend for the HCC-MEAT Coder Space. | |
| Endpoints: | |
| POST /api/code — submit a note, receive ranked HCC candidates + MEAT evidence | |
| GET /api/gallery — return the preloaded demo gallery (note + precomputed result) | |
| GET /api/healthz — simple health probe | |
| GET /api/reference/hcc — static HCC label table for tooltip content in the UI | |
| The app keeps a single `Pipeline` instance in state and reuses it across requests. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| import traceback | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel, Field | |
| from app.api.metrics import snapshot as metrics_snapshot | |
| from app.pipeline import Pipeline | |
| from app.pipeline.types import CodingResult, Note | |
| _LOG = logging.getLogger("hcc_meat.api") | |
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s") | |
| GALLERY_PATH = Path(__file__).resolve().parents[2] / "tests" / "fixtures" / "regression_suite.json" | |
| UI_DIST = Path(__file__).resolve().parents[2] / "ui" / "dist" | |
| class CodeRequest(BaseModel): | |
| text: str = Field(..., min_length=1, max_length=50_000) | |
| note_id: str = Field(default="adhoc") | |
| class GalleryItem(BaseModel): | |
| case_id: str | |
| name: str | |
| tags: list[str] | |
| text: str | |
| result: CodingResult | |
| async def lifespan(app: FastAPI): | |
| """Build the pipeline and prime caches at startup. | |
| Each step is wrapped defensively: if a step fails, we log the traceback | |
| and leave the corresponding state empty so the app still boots. /api/status | |
| reports this as `degraded` and the UI banner surfaces which facet failed. | |
| A hard crash here would leave HF users with a 502 and no diagnostic path. | |
| """ | |
| app.state.startup_errors = {} | |
| try: | |
| app.state.pipeline = Pipeline() | |
| _LOG.info("Pipeline built") | |
| except Exception as e: # noqa: BLE001 | |
| app.state.pipeline = None | |
| app.state.startup_errors["pipeline"] = repr(e) | |
| _LOG.error("Pipeline build failed: %s\n%s", e, traceback.format_exc()) | |
| try: | |
| app.state.gallery = ( | |
| _build_gallery(app.state.pipeline) if app.state.pipeline is not None else [] | |
| ) | |
| _LOG.info("Gallery primed: %d items", len(app.state.gallery)) | |
| except Exception as e: # noqa: BLE001 | |
| app.state.gallery = [] | |
| app.state.startup_errors["gallery"] = repr(e) | |
| _LOG.error("Gallery build failed: %s\n%s", e, traceback.format_exc()) | |
| try: | |
| app.state.metrics = metrics_snapshot() if app.state.pipeline is not None else {} | |
| _LOG.info("Metrics snapshot computed") | |
| except Exception as e: # noqa: BLE001 | |
| app.state.metrics = {} | |
| app.state.startup_errors["metrics"] = repr(e) | |
| _LOG.error("Metrics snapshot failed: %s\n%s", e, traceback.format_exc()) | |
| # Kick off LLM MEAT warm-up in the background when selected. | |
| # Rule-based path serves until the LLM is ready; /api/status exposes the | |
| # ready flag so the UI (and smoke tests) can see warm state. | |
| import threading as _threading | |
| app.state.meat_llm = None | |
| extractor = getattr(app.state.pipeline, "meat_extractor", None) if app.state.pipeline else None | |
| primary = getattr(extractor, "primary", None) | |
| if primary is not None and hasattr(primary, "warm"): | |
| app.state.meat_llm = primary | |
| def _warm_meat_llm() -> None: | |
| try: | |
| primary.warm() | |
| _LOG.info("MEAT LLM ready: %s", getattr(primary, "model_name", "?")) | |
| except Exception as e: # noqa: BLE001 | |
| app.state.startup_errors["meat_llm"] = repr(e) | |
| _LOG.error( | |
| "MEAT LLM warm-up failed: %s\n%s", e, traceback.format_exc() | |
| ) | |
| _threading.Thread( | |
| target=_warm_meat_llm, name="meat-llm-warmup", daemon=True | |
| ).start() | |
| yield | |
| app = FastAPI( | |
| title="HCC-MEAT Coder", | |
| version="0.1.0", | |
| description=( | |
| "Suggests CMS-HCC v28 codes from a clinical note with MEAT-justified " | |
| "evidence, calibrated confidence, and an over-coding risk flag." | |
| ), | |
| lifespan=lifespan, | |
| ) | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| def healthz() -> dict: | |
| return {"status": "ok", "version": app.version} | |
| def status() -> dict: | |
| """Runtime diagnostic: are reference CSVs loaded, pipeline built, gallery primed? | |
| The UI polls this on startup so an HF-Space misdeploy (missing data dir, | |
| startup crash) surfaces as a visible banner instead of a silent blank state. | |
| """ | |
| pipeline: Pipeline | None = getattr(app.state, "pipeline", None) | |
| gallery: list[GalleryItem] = getattr(app.state, "gallery", []) or [] | |
| metrics: dict = getattr(app.state, "metrics", {}) or {} | |
| crosswalk = pipeline.crosswalk if pipeline else None | |
| ref_counts: dict = {} | |
| if crosswalk is not None: | |
| ref_counts = { | |
| "hcc_labels": len(crosswalk.all_hccs()), | |
| "hcc_crosswalk_icd10": len(crosswalk.all_icd10_in_crosswalk()), | |
| } | |
| ok = ( | |
| pipeline is not None | |
| and ref_counts.get("hcc_labels", 0) > 0 | |
| and ref_counts.get("hcc_crosswalk_icd10", 0) > 0 | |
| and len(gallery) > 0 | |
| ) | |
| meat_llm = getattr(app.state, "meat_llm", None) | |
| meat_llm_model = getattr(meat_llm, "model_name", None) if meat_llm else None | |
| meat_llm_ready = bool(meat_llm and meat_llm.is_ready()) if meat_llm else None | |
| return { | |
| "status": "ok" if ok else "degraded", | |
| "version": app.version, | |
| "pipeline_ready": pipeline is not None, | |
| "reference_counts": ref_counts, | |
| "gallery_size": len(gallery), | |
| "metrics_available": bool(metrics), | |
| "meat_llm_configured": meat_llm is not None, | |
| "meat_llm_model": meat_llm_model, | |
| "meat_llm_ready": meat_llm_ready, | |
| "startup_errors": getattr(app.state, "startup_errors", {}), | |
| } | |
| def code(req: CodeRequest) -> CodingResult: | |
| pipeline: Pipeline | None = getattr(app.state, "pipeline", None) | |
| if pipeline is None: | |
| raise HTTPException(503, "pipeline not ready — see /api/status") | |
| note = Note(note_id=req.note_id, text=req.text) | |
| return pipeline.run(note) | |
| def gallery() -> list[GalleryItem]: | |
| return app.state.gallery | |
| def metrics() -> dict: | |
| """Static snapshot built at startup. v0.1 derives from the regression fixture.""" | |
| return app.state.metrics | |
| def debug_pipeline(req: CodeRequest) -> dict: | |
| """Pipeline introspection: sections, spans, attributable spans, and | |
| per-candidate MEAT extraction trace. Used to diagnose MEAT-coverage gaps | |
| when the live Space disagrees with local expectations. | |
| """ | |
| pipeline: Pipeline | None = getattr(app.state, "pipeline", None) | |
| if pipeline is None: | |
| raise HTTPException(503, "pipeline not ready") | |
| from app.pipeline.components import a_segmenter | |
| from app.pipeline.components.c_negation import is_attributable | |
| from app.pipeline.components.f_meat import _sentences_with_offsets | |
| note = Note(note_id=req.note_id, text=req.text) | |
| note = a_segmenter.segment(note) | |
| spans = pipeline.ner(note) | |
| spans = pipeline.negation(note, spans) | |
| eligible = [s for s in spans if is_attributable(s)] | |
| links = pipeline.linker(eligible) | |
| candidates = pipeline.crosswalk.build_candidates( | |
| links, raf_segment=pipeline.config.raf_segment | |
| ) | |
| candidates = pipeline.crosswalk.apply_trumping(candidates) | |
| sentences = _sentences_with_offsets(note.text) | |
| extraction_trace: list[dict] = [] | |
| primary = getattr(pipeline.meat_extractor, "primary", None) | |
| primary_has_try = primary is not None and hasattr(primary, "try_extract") | |
| primary_ready = ( | |
| primary.is_ready() if primary is not None and hasattr(primary, "is_ready") else None | |
| ) | |
| # Single LLM pass per candidate to keep latency sane: capture raw LLM | |
| # output, then filter via the same substring rule Component I uses. No | |
| # double-calling pipeline.meat_extractor. | |
| for cand in candidates: | |
| llm_raw: list[dict] = [] | |
| kept: list[dict] = [] | |
| if primary_has_try and primary_ready: | |
| try: | |
| res = primary.try_extract(note, cand) | |
| if res is not None: | |
| for ev in res: | |
| in_note = ev.quote in note.text | |
| llm_raw.append( | |
| { | |
| "cat": ev.category.value, | |
| "quote": ev.quote, | |
| "is_substring": in_note, | |
| } | |
| ) | |
| if in_note: | |
| kept.append( | |
| {"cat": ev.category.value, "quote": ev.quote[:120]} | |
| ) | |
| except Exception as e: # noqa: BLE001 | |
| llm_raw = [{"error": f"{type(e).__name__}: {e}"}] | |
| extraction_trace.append( | |
| { | |
| "hcc": cand.hcc, | |
| "surface": [lnk.condition_span.text for lnk in cand.icd10_links], | |
| "llm_raw_output": llm_raw, | |
| "verified_evidence_preview": kept, | |
| } | |
| ) | |
| return { | |
| "sections": [ | |
| {"kind": s.kind, "start": s.start, "end": s.end} for s in note.sections | |
| ], | |
| "sentences": [{"start": s, "end": e, "text": t[:80]} for (s, e, t) in sentences], | |
| "all_spans": [ | |
| {"text": s.text, "type": s.span_type.value, "start": s.start, "end": s.end} | |
| for s in spans | |
| ], | |
| "eligible_spans": len(eligible), | |
| "candidates": [c.hcc for c in candidates], | |
| "extractor_chain": { | |
| "outer_type": type(pipeline.meat_extractor).__name__, | |
| "primary_type": type(primary).__name__ if primary else None, | |
| "primary_has_try_extract": primary_has_try, | |
| "primary_ready": primary_ready, | |
| }, | |
| "extraction_trace": extraction_trace, | |
| } | |
| def benchmark() -> dict: | |
| """Re-run every gallery note live (no precomputation) and summarize latency. | |
| Exercises the CPU-free-tier latency target (S2 in AGENT_BRIEF §3): p95 < 8s. | |
| """ | |
| import statistics | |
| import time | |
| pipeline: Pipeline | None = getattr(app.state, "pipeline", None) | |
| if pipeline is None: | |
| raise HTTPException(503, "pipeline not ready — see /api/status") | |
| items: list[GalleryItem] = app.state.gallery | |
| latencies: list[float] = [] | |
| timings: list[dict] = [] | |
| for item in items: | |
| t0 = time.perf_counter() | |
| pipeline.run(Note(note_id=item.case_id, text=item.text)) | |
| ms = (time.perf_counter() - t0) * 1000.0 | |
| latencies.append(ms) | |
| timings.append({"case_id": item.case_id, "ms": round(ms, 2), "chars": len(item.text)}) | |
| latencies.sort() | |
| summary = { | |
| "count": len(latencies), | |
| "mean_ms": round(statistics.fmean(latencies), 2) if latencies else 0.0, | |
| "median_ms": round(statistics.median(latencies), 2) if latencies else 0.0, | |
| "p95_ms": round(latencies[int(0.95 * (len(latencies) - 1))], 2) if latencies else 0.0, | |
| "max_ms": round(max(latencies), 2) if latencies else 0.0, | |
| "min_ms": round(min(latencies), 2) if latencies else 0.0, | |
| "target_p95_ms": 8000.0, | |
| "timings": timings, | |
| } | |
| return summary | |
| def reference_hcc() -> list[dict]: | |
| """Return HCC labels for UI tooltips + model card linking.""" | |
| import csv | |
| path = Path("data/reference/hcc_v28_labels.csv") | |
| if not path.exists(): | |
| raise HTTPException(500, "reference data missing") | |
| with path.open() as f: | |
| return list(csv.DictReader(f)) | |
| def _build_gallery(pipeline: Pipeline) -> list[GalleryItem]: | |
| if not GALLERY_PATH.exists(): | |
| return [] | |
| cases = json.loads(GALLERY_PATH.read_text())["cases"] | |
| out: list[GalleryItem] = [] | |
| for c in cases: | |
| note = Note(note_id=c["case_id"], text=c["note_text"]) | |
| result = pipeline.run(note) | |
| out.append( | |
| GalleryItem( | |
| case_id=c["case_id"], | |
| name=c["name"], | |
| tags=c.get("tags", []), | |
| text=c["note_text"], | |
| result=result, | |
| ) | |
| ) | |
| return out | |
| # Mount the built React UI *last* so /api/* routes win. | |
| if UI_DIST.exists(): | |
| app.mount("/", StaticFiles(directory=UI_DIST, html=True), name="ui") | |