File size: 4,479 Bytes
ba54ea9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Recap — FastAPI app entry point. Serves the React UI + JSON inference API.
GET / → static index.html (React via CDN, Babel-compiled JSX in browser)
GET /static/* → static assets (app.jsx, css)
GET /api/patients → list of patients with full event timelines
POST /api/answer → run the inference gateway and return a cited answer
GET /api/health → liveness + backend selection
"""
from pathlib import Path
from fastapi import FastAPI
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from recap.cases import load_case
from recap.config import load as load_config
from recap.demo_patient import build_demo_patient
from recap.inference import answer as answer_question
from recap.models import Patient
CFG = load_config()
ROOT = Path(__file__).parent
STATIC_DIR = ROOT / "static"
def _discover_cases() -> dict[str, Patient]:
cases: dict[str, Patient] = {}
cases_dir = Path(CFG.cases_dir)
if cases_dir.exists():
for d in sorted(cases_dir.iterdir()):
if (d / "manifest.json").exists():
try:
cases[d.name] = load_case(CFG.cases_dir, d.name)
except Exception as e: # noqa: BLE001 — keep one bad case from breaking the whole API
print(f"[recap] failed to load case {d.name}: {e}")
if not cases:
cases["demo"] = build_demo_patient()
return cases
PATIENTS: dict[str, Patient] = _discover_cases()
app = FastAPI(title="Recap", version="0.1.0")
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
class AnswerRequest(BaseModel):
patient_id: str
question: str
@app.get("/")
def index() -> FileResponse:
return FileResponse(STATIC_DIR / "index.html")
@app.get("/api/patients")
def list_patients() -> JSONResponse:
"""Serialize all loaded patients in a shape the React app expects."""
out = []
for pid, p in PATIENTS.items():
out.append({
"id": p.id,
"display_name": p.display_name,
"age": p.age,
"gender": p.gender,
"mrn": getattr(p, "mrn", None) or f"MRN-{abs(hash(p.id)) % 9999999:07d}",
"summary": _patient_summary(p),
"hook": _patient_hook(p),
"tags": _patient_tags(p),
"events": [_event_to_dict(e) for e in p.events],
})
return JSONResponse(out)
@app.post("/api/answer")
def answer(req: AnswerRequest) -> JSONResponse:
if req.patient_id not in PATIENTS:
return JSONResponse({"error": f"unknown patient {req.patient_id}"}, status_code=404)
p = PATIENTS[req.patient_id]
a = answer_question(req.question, p.events)
return JSONResponse({
"text": a.text,
"citations": [
{"source_id": c.source_id, "page": c.page, "snippet": c.snippet}
for c in a.citations
],
})
@app.get("/api/health")
def health() -> JSONResponse:
return JSONResponse({
"ok": True,
"backend": CFG.backend,
"patient_count": len(PATIENTS),
"patient_ids": list(PATIENTS.keys()),
})
# ─── Helpers ───────────────────────────────────────────────────────────
def _event_to_dict(e) -> dict:
return {
"id": e.id,
"date": e.date.date().isoformat(),
"category": e.category,
"title": e.title,
"source": e.source,
"body": e.body,
"page": e.metadata.get("page"),
"snippet": e.metadata.get("snippet"),
"flag": e.metadata.get("flag"),
}
def _patient_summary(p: Patient) -> str:
"""One-sentence dossier summary. Real cases override via manifest.summary later."""
n = len(p.events)
years = sorted({e.date.year for e in p.events})
span = f"{years[0]}–{years[-1]}" if years else "no record"
return f"{n} clinical events on file from {span}."
def _patient_hook(p: Patient) -> str:
return ""
def _patient_tags(p: Patient) -> list[str]:
"""Surface the most recent diagnosis titles as tag chips."""
dx = [e.title for e in sorted(p.events, key=lambda e: e.date, reverse=True) if e.category == "diagnosis"]
return dx[:3]
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860)
|