Cavell SDK — Hospitalization Extraction Demo¶

Extract structured FHIR resources from simulated hospital stays. Each stay is a discharge letter plus the supporting documents produced during that admission — admission notes, lab panels, imaging reports, operative notes, pathology and consults.

The dataset (hospitalizations.csv) is purpose-built to exercise downstream clinical coding of hospital stays. Every document of a stay shares one encounter_id, the admission identifier mapped to Document.encounter_id. The pipeline looks that Encounter up before each extraction and hands it to the API, which updates it in place — so a stay ends up as exactly one FHIR Encounter whose status and period follow the admission from the first note to the discharge letter, and every document's resources reference it. That is exactly how a clinical coder gathers all of a stay's documents (encounter=<Encounter/x>) to assign ICD-10 diagnoses and procedures.

Same two-step flow as the general CSV demo:

1. Seed — organizations, practitioners, and patients into FHIR
2. Extract — documents, processed per-patient in date order

Prerequisites:

• Docker running (docker compose up -d)
• A Prism API URL and an LLM Gateway key

All data in this demo is fully synthetic. Names, birth dates, identifiers, and clinical narratives were generated for demonstration purposes and do not describe real patients.

Dataset: curated hospital stays¶

108 documents across 17 stays and four earlier outpatient reviews (16 patients; MRN-20003 appears twice as a readmission). The outpatient reviews (V-101 for MRN-20003, V-102 for MRN-20009, V-103 for MRN-20002, V-104 for MRN-20010) predate those patients' admissions and carry their pre-existing medical history as a problem list with explicit statuses — active, resolved, inactive and in-remission conditions — under their own encounter ids. The cases are deliberately curated so the set covers the coding features a clinical coder must handle — a clean medical floor, surgical procedures (ICD-10-PCS), comorbidity-driven severity, a hospital-acquired (POA = N) complication, a worked-up provisional diagnosis, pathology-driven cancer staging, identified-organism sepsis, a readmission, a POA contrast pair (admitted for a COPD exacerbation, V-010, POA = Y, vs the same exacerbation arising in hospital, V-016, POA = N), and a combination-coding contrast (single infective exacerbation, V-010, → J44.0 alone, vs a distinct exacerbation plus a pneumococcal pneumonia, V-017, → J44.0 + J44.1 + the pneumonia code).

The CSV carries 8 extra rows beyond those 108, all held back at load time and submitted together in step 10. Six are earlier admissions for MRN-20002 — a November 2022 stay for pyelonephritis (V-903) and a September 2023 wrist fracture (V-902), both backwards in time and extracted against split context. Two are a later (July 2025) readmission for MRN-20017 (V-018) — forwards in time, and extracted normally. Step 10 submits them in one call and shows what the reconciliation gate did with each.

Stay	Scenario	Coding feature exercised	Documents
V-001	Community-acquired pneumonia, organism unidentified	Clean medical admission; Safe-path floor; physician query (J18.9 vs J15.x)	discharge letter, admission note, labs, sputum/blood culture (no growth), chest X-ray, progress note
V-002	Femoral-neck fracture → hemiarthroplasty	ICD-10-PCS procedure; osteoporosis as secondary	discharge letter, ED/admission note, hip X-ray, pre-op labs, operative note, post-op X-ray
V-003	Acute decompensated heart failure + T2DM + CKD3 + AFib	Multiple comorbidities (CC/MCC) → severity (SOI)	discharge letter, admission note, labs, ECG, CXR, echo, nephrology consult
V-004	Elective sigmoid colectomy + post-op DVT	POA = N complication vs POA = Y; procedure	discharge letter, admission note, operative note, pathology, post-op labs, Doppler US
V-005	Sepsis, source never confirmed	Inpatient provisional-diagnosis rule; physician query	discharge letter, admission note, blood/urine cultures, labs, CT abdomen, progress note
V-006	Caecal adenocarcinoma → right hemicolectomy	Pathology drives specific neoplasm + grade/stage; procedure	discharge letter, admission note, staging CT, pre-op labs, operative note, pathology, MDT note
V-007	Term pregnancy → emergency caesarean	Obstetric coding + delivery procedure + outcome	discharge letter, admission note, labs, intrapartum/CTG note, operative note, postnatal note
V-008	Acute appendicitis → laparoscopic appendectomy	Procedure + pathology	discharge letter, ED note, labs, CT, operative note, pathology
V-009	Inferior STEMI → primary PCI	Acute MI + cardiac PCS (stent)	discharge letter, ED/admission note, ECG, serial troponins, cath/PCI report, echo
V-010	COPD exacerbation + acute type 2 respiratory failure	Severity driver (J96.0x); POA = Y (exacerbation is the reason for admission)	discharge letter, admission note, ABG, labs, CXR
V-011	Acute ischaemic stroke → IV thrombolysis	Neuro; thrombolysis; imaging-driven	discharge letter, admission note, CT head, CT angiography, MRI, labs
V-012	E. coli urosepsis from acute pyelonephritis	Sepsis + identified organism; sequencing; POA = Y	discharge letter, admission note, blood culture, urine culture, labs, renal US
V-013	Diabetic foot ulcer with osteomyelitis → toe amputation	Comorbidity + complication chain + amputation procedure	discharge letter, admission note, wound culture, foot X-ray, labs, operative note, pathology
V-014	Readmission of the V-003 patient with AKI + hyperkalaemia	Readmission; prior-episode exclusion	discharge letter, admission note, labs, ECG, cardiology consult
V-015	Acute calculous cholecystitis → laparoscopic cholecystectomy	Procedure + pathology	discharge letter, ED note, RUQ ultrasound, LFT labs, operative note, pathology
V-016	Elective hip replacement; in-hospital COPD exacerbation (day 4)	THR procedure; POA = N (exacerbation arose in hospital) — mirror of V-010	discharge letter, admission/pre-op note, operative note, post-op labs, respiratory consult, chest X-ray
V-017	COPD with a distinct acute exacerbation plus a pneumococcal pneumonia	Combination coding: two distinct facets → J44.0 + J44.1 + pneumonia code (S. pneumoniae) — contrast with V-010's single infective exacerbation	discharge letter, admission note, chest X-ray (consolidation), labs, sputum/blood culture (S. pneumoniae), progress note

1. Install¶

In [ ]:

%pip install cavell-prism-client

%pip install cavell-prism-client

In [ ]:

import logging

logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s")
logging.getLogger("cavell_client").setLevel(logging.INFO)

import logging logging.basicConfig(level=logging.WARNING, format="%(levelname)s %(name)s: %(message)s") logging.getLogger("cavell_client").setLevel(logging.INFO)

2. Check FHIR server¶

In [ ]:

import httpx

FHIR_BASE_URL = "http://localhost:8090"

resp = httpx.get(f"{FHIR_BASE_URL}/fhir/metadata", timeout=5)
assert resp.status_code == 200, "FHIR server not reachable — run: docker compose up -d"
print(f"FHIR server OK at {FHIR_BASE_URL}")

import httpx FHIR_BASE_URL = "http://localhost:8090" resp = httpx.get(f"{FHIR_BASE_URL}/fhir/metadata", timeout=5) assert resp.status_code == 200, "FHIR server not reachable — run: docker compose up -d" print(f"FHIR server OK at {FHIR_BASE_URL}")

3. Load the CSV¶

In [ ]:

import csv

CSV_PATH = "hospitalizations.csv"  # curated hospital stays

with open(CSV_PATH, newline="", encoding="utf-8-sig") as f:
    rows = list(csv.DictReader(f))

print(f"{len(rows)} rows")
print(f"Columns: {list(rows[0].keys())}")
rows[0]

import csv CSV_PATH = "hospitalizations.csv" # curated hospital stays with open(CSV_PATH, newline="", encoding="utf-8-sig") as f: rows = list(csv.DictReader(f)) print(f"{len(rows)} rows") print(f"Columns: {list(rows[0].keys())}") rows[0]

4. Configure column mapping¶

Same 11 columns as the general CSV demo. The encounter_id column is mapped to Document.encounter_id: it is what makes every document of an admission update the same FHIR Encounter instead of each creating its own, so keep it mapped for this dataset.

In [ ]:

# Required
COL_PATIENT_ID = "patient_id"
COL_NOTE_TEXT = "note_text"
COL_NOTE_DATE = "note_date"  # must be YYYY-MM-DD
# Identifies the DocumentReference. Required: the resume-skip, the chronology
# watermark and failure reporting all key on it, so reuse your source system's
# identifier rather than generating a fresh one each run.
COL_NOTE_ID = "note_id"

# Optional — set to None if your CSV doesn't have these columns
COL_ENCOUNTER_ID = "encounter_id"  # the admission a note belongs to -> one Encounter
COL_PATIENT_NAME = "patient_name"  # stored on Patient FHIR resource
COL_BIRTH_DATE = "birth_date"  # stored on Patient FHIR resource
COL_GENDER = "gender"  # stored on Patient FHIR resource
COL_PRACTITIONER_ID = "practitioner_id"
COL_PRACTITIONER_NAME = "practitioner_name"  # required when practitioner_id is set

# Extra context columns — sent alongside each note to improve extraction.
# The pipeline sends the document date as its own payload field and
# auto-injects the practitioner name into meta — don't repeat either here
# into the meta, so don't duplicate those here.
META_COLUMNS = {
    "Department": "department",
}

# Organization — set to your facility identifier
ORG_ID = "DEMO-HOSPITAL"
ORG_NAME = "Demo Hospital"

# Required COL_PATIENT_ID = "patient_id" COL_NOTE_TEXT = "note_text" COL_NOTE_DATE = "note_date" # must be YYYY-MM-DD # Identifies the DocumentReference. Required: the resume-skip, the chronology # watermark and failure reporting all key on it, so reuse your source system's # identifier rather than generating a fresh one each run. COL_NOTE_ID = "note_id" # Optional — set to None if your CSV doesn't have these columns COL_ENCOUNTER_ID = "encounter_id" # the admission a note belongs to -> one Encounter COL_PATIENT_NAME = "patient_name" # stored on Patient FHIR resource COL_BIRTH_DATE = "birth_date" # stored on Patient FHIR resource COL_GENDER = "gender" # stored on Patient FHIR resource COL_PRACTITIONER_ID = "practitioner_id" COL_PRACTITIONER_NAME = "practitioner_name" # required when practitioner_id is set # Extra context columns — sent alongside each note to improve extraction. # The pipeline sends the document date as its own payload field and # auto-injects the practitioner name into meta — don't repeat either here # into the meta, so don't duplicate those here. META_COLUMNS = { "Department": "department", } # Organization — set to your facility identifier ORG_ID = "DEMO-HOSPITAL" ORG_NAME = "Demo Hospital"

5. Build SDK objects from CSV¶

Each cell below builds the objects for seeding.

Organizations and practitioners¶

In [ ]:

from cavell_client import Organization, Practitioner

organizations = [Organization(identifier=ORG_ID, name=ORG_NAME)]

practitioners = (
    Practitioner.from_rows(
        rows,
        columns={"identifier": COL_PRACTITIONER_ID, "name": COL_PRACTITIONER_NAME},
        organization_identifier=ORG_ID,
    )
    if COL_PRACTITIONER_ID
    else []
)

print(f"{len(organizations)} organizations, {len(practitioners)} practitioners")

from cavell_client import Organization, Practitioner organizations = [Organization(identifier=ORG_ID, name=ORG_NAME)] practitioners = ( Practitioner.from_rows( rows, columns={"identifier": COL_PRACTITIONER_ID, "name": COL_PRACTITIONER_NAME}, organization_identifier=ORG_ID, ) if COL_PRACTITIONER_ID else [] ) print(f"{len(organizations)} organizations, {len(practitioners)} practitioners")

Patients¶

In [ ]:

from cavell_client import Patient

_patient_columns = {"identifier": COL_PATIENT_ID}
if COL_PATIENT_NAME:
    _patient_columns["name"] = COL_PATIENT_NAME
if COL_BIRTH_DATE:
    _patient_columns["birth_date"] = COL_BIRTH_DATE
if COL_GENDER:
    _patient_columns["gender"] = COL_GENDER
if COL_PRACTITIONER_ID:
    _patient_columns["general_practitioners"] = COL_PRACTITIONER_ID

patients = Patient.from_rows(
    rows,
    columns=_patient_columns,
    managing_organization=ORG_ID,
)

print(f"{len(patients)} patients")

from cavell_client import Patient _patient_columns = {"identifier": COL_PATIENT_ID} if COL_PATIENT_NAME: _patient_columns["name"] = COL_PATIENT_NAME if COL_BIRTH_DATE: _patient_columns["birth_date"] = COL_BIRTH_DATE if COL_GENDER: _patient_columns["gender"] = COL_GENDER if COL_PRACTITIONER_ID: _patient_columns["general_practitioners"] = COL_PRACTITIONER_ID patients = Patient.from_rows( rows, columns=_patient_columns, managing_organization=ORG_ID, ) print(f"{len(patients)} patients")

Documents¶

In [ ]:

from cavell_client import Document

_documents = Document.from_rows(
    rows,
    columns={
        "text": COL_NOTE_TEXT,
        "patient_identifier": COL_PATIENT_ID,
        "date": COL_NOTE_DATE,
        "document_id": COL_NOTE_ID,
        "encounter_id": COL_ENCOUNTER_ID,
        "meta": META_COLUMNS,
        "practitioner_identifier": COL_PRACTITIONER_ID,
    },
    organization_identifier=ORG_ID,
)

# Held back from the main run and submitted together in step 10, which shows
# what an out-of-order note is given as context. Patients and practitioners are
# still built from every row, so both patients below are seeded normally either
# way.
#
# Backdated: two EARLIER admissions for MRN-20002, both behind her April 2024
# stay — these extract against context split at their own date. The November
# 2022 stay is the one that exercises all three outcomes; see step 10.
BACKDATED_IDS = {
    "note-9011",
    "note-9012",
    "note-9013",  # Nov 2022, pyelonephritis
    "note-9001",
    "note-9002",
    "note-9003",  # Sept 2023, wrist fracture
}
# Forward-dated: a LATER readmission for MRN-20017 (July 2025), ahead of his
# June 2025 stay — these extract normally.
FORWARD_IDS = {"note-9101", "note-9102"}
HELD_BACK_IDS = BACKDATED_IDS | FORWARD_IDS

all_documents = [d for d in _documents if d.document_id not in HELD_BACK_IDS]
backdated_documents = [d for d in _documents if d.document_id in BACKDATED_IDS]
forward_documents = [d for d in _documents if d.document_id in FORWARD_IDS]
held_back_documents = backdated_documents + forward_documents

for label, held, expected in (
    ("backdated", backdated_documents, BACKDATED_IDS),
    ("forward-dated", forward_documents, FORWARD_IDS),
):
    assert len(held) == len(expected), (
        f"expected {len(expected)} held-back {label} notes, found {len(held)} "
        f"— is the CSV up to date?"
    )

print(
    f"{len(all_documents)} documents ({len(held_back_documents)} held back: "
    f"{len(backdated_documents)} backdated, {len(forward_documents)} forward-dated)"
)

from cavell_client import Document _documents = Document.from_rows( rows, columns={ "text": COL_NOTE_TEXT, "patient_identifier": COL_PATIENT_ID, "date": COL_NOTE_DATE, "document_id": COL_NOTE_ID, "encounter_id": COL_ENCOUNTER_ID, "meta": META_COLUMNS, "practitioner_identifier": COL_PRACTITIONER_ID, }, organization_identifier=ORG_ID, ) # Held back from the main run and submitted together in step 10, which shows # what an out-of-order note is given as context. Patients and practitioners are # still built from every row, so both patients below are seeded normally either # way. # # Backdated: two EARLIER admissions for MRN-20002, both behind her April 2024 # stay — these extract against context split at their own date. The November # 2022 stay is the one that exercises all three outcomes; see step 10. BACKDATED_IDS = { "note-9011", "note-9012", "note-9013", # Nov 2022, pyelonephritis "note-9001", "note-9002", "note-9003", # Sept 2023, wrist fracture } # Forward-dated: a LATER readmission for MRN-20017 (July 2025), ahead of his # June 2025 stay — these extract normally. FORWARD_IDS = {"note-9101", "note-9102"} HELD_BACK_IDS = BACKDATED_IDS | FORWARD_IDS all_documents = [d for d in _documents if d.document_id not in HELD_BACK_IDS] backdated_documents = [d for d in _documents if d.document_id in BACKDATED_IDS] forward_documents = [d for d in _documents if d.document_id in FORWARD_IDS] held_back_documents = backdated_documents + forward_documents for label, held, expected in ( ("backdated", backdated_documents, BACKDATED_IDS), ("forward-dated", forward_documents, FORWARD_IDS), ): assert len(held) == len(expected), ( f"expected {len(expected)} held-back {label} notes, found {len(held)} " f"— is the CSV up to date?" ) print( f"{len(all_documents)} documents ({len(held_back_documents)} held back: " f"{len(backdated_documents)} backdated, {len(forward_documents)} forward-dated)" )

Data profile¶

Unlike the flat notes dataset, these documents cluster into hospital stays. The profile below shows documents per stay (each stay becomes one Encounter, updated by each of its documents in turn) and the mix of document types.

In [ ]:

from collections import Counter

MAX_BAR = 40


def bar(n, max_n):
    return "\u2588" * (round(n / max_n * MAX_BAR) if max_n else 0)


def doc_type(text):
    # The first line of each note is its type header, e.g.
    # "RADIOLOGY REPORT — Chest X-ray" or "DISCHARGE LETTER".
    head = text.strip().splitlines()[0]
    return head.split("\u2014")[0].strip()


# --- Stays and documents per stay ---
docs_per_stay = Counter(d.encounter_id for d in all_documents)
stay_counts = sorted(docs_per_stay.values())
n_stays = len(stay_counts)
stays_by_patient = Counter()
_seen = set()
for d in all_documents:
    key = (d.patient_identifier, d.encounter_id)
    if key not in _seen:
        _seen.add(key)
        stays_by_patient[d.patient_identifier] += 1
readmissions = {p: n for p, n in stays_by_patient.items() if n > 1}

held_back_stays = len({d.encounter_id for d in held_back_documents})
print(f"Stays (encounter_ids): {n_stays} in this run")
total_stays = n_stays + held_back_stays
print(f"  (+{held_back_stays} held back for step 10 = {total_stays} in the CSV)")
print(f"Patients: {len(stays_by_patient)}")
print(
    f"Documents per stay:"
    f"  min={stay_counts[0]}  max={stay_counts[-1]}"
    f"  avg={sum(stay_counts) / n_stays:.1f}"
)
print(f"Readmissions (patients with >1 stay): {readmissions}")

print("\n  documents | stays")
stay_hist = Counter(stay_counts)
max_sh = max(stay_hist.values())
for k in sorted(stay_hist):
    print(f"  {k:>9} | {stay_hist[k]:>3}  {bar(stay_hist[k], max_sh)}")

# --- Document types ---
types = Counter(doc_type(d.text) for d in all_documents)
max_t = max(types.values())
print("\n  document type                | count")
for label, c in types.most_common():
    print(f"  {label[:28]:<28} | {c:>3}  {bar(c, max_t)}")

# --- Note length (characters) ---
lengths = sorted(len(d.text) for d in all_documents)
n = len(lengths)
median_len = lengths[n // 2] if n % 2 else (lengths[n // 2 - 1] + lengths[n // 2]) / 2
print(
    f"\nNote length (chars):"
    f"  min={lengths[0]:,}  max={lengths[-1]:,}"
    f"  avg={sum(lengths) / n:,.0f}  median={median_len:,.0f}"
)
print(f"Total content: {sum(lengths):,} chars across {n} notes")

from collections import Counter MAX_BAR = 40 def bar(n, max_n): return "\u2588" * (round(n / max_n * MAX_BAR) if max_n else 0) def doc_type(text): # The first line of each note is its type header, e.g. # "RADIOLOGY REPORT — Chest X-ray" or "DISCHARGE LETTER". head = text.strip().splitlines()[0] return head.split("\u2014")[0].strip() # --- Stays and documents per stay --- docs_per_stay = Counter(d.encounter_id for d in all_documents) stay_counts = sorted(docs_per_stay.values()) n_stays = len(stay_counts) stays_by_patient = Counter() _seen = set() for d in all_documents: key = (d.patient_identifier, d.encounter_id) if key not in _seen: _seen.add(key) stays_by_patient[d.patient_identifier] += 1 readmissions = {p: n for p, n in stays_by_patient.items() if n > 1} held_back_stays = len({d.encounter_id for d in held_back_documents}) print(f"Stays (encounter_ids): {n_stays} in this run") total_stays = n_stays + held_back_stays print(f" (+{held_back_stays} held back for step 10 = {total_stays} in the CSV)") print(f"Patients: {len(stays_by_patient)}") print( f"Documents per stay:" f" min={stay_counts[0]} max={stay_counts[-1]}" f" avg={sum(stay_counts) / n_stays:.1f}" ) print(f"Readmissions (patients with >1 stay): {readmissions}") print("\n documents | stays") stay_hist = Counter(stay_counts) max_sh = max(stay_hist.values()) for k in sorted(stay_hist): print(f" {k:>9} | {stay_hist[k]:>3} {bar(stay_hist[k], max_sh)}") # --- Document types --- types = Counter(doc_type(d.text) for d in all_documents) max_t = max(types.values()) print("\n document type | count") for label, c in types.most_common(): print(f" {label[:28]:<28} | {c:>3} {bar(c, max_t)}") # --- Note length (characters) --- lengths = sorted(len(d.text) for d in all_documents) n = len(lengths) median_len = lengths[n // 2] if n % 2 else (lengths[n // 2 - 1] + lengths[n // 2]) / 2 print( f"\nNote length (chars):" f" min={lengths[0]:,} max={lengths[-1]:,}" f" avg={sum(lengths) / n:,.0f} median={median_len:,.0f}" ) print(f"Total content: {sum(lengths):,} chars across {n} notes")

6. Connect to FHIR and Cavell API¶

In [ ]:

import getpass
import os

from cavell_client import CavellClient, IngestionPipeline

# Point CAVELL_API_URL at your Prism deployment and provide your LLM Gateway
# key (prompted below when the CAVELL_API_KEY environment variable is unset).
CAVELL_API_URL = os.environ.get("CAVELL_API_URL", "https://prd.prism.cavell.app/api")
CAVELL_API_KEY = os.environ.get("CAVELL_API_KEY") or getpass.getpass(
    "LLM Gateway key: "
)

client = CavellClient(
    api_url=CAVELL_API_URL,
    api_key=CAVELL_API_KEY,
    fhir_base_url=FHIR_BASE_URL,
    fhir_api_path="/fhir",
)

print("Connected.")

tiers = client.list_tiers()
print("\nAvailable tiers:")
for t in tiers:
    default = " (default)" if t["default"] else ""
    print(f"  {t['name']}{default}")

import getpass import os from cavell_client import CavellClient, IngestionPipeline # Point CAVELL_API_URL at your Prism deployment and provide your LLM Gateway # key (prompted below when the CAVELL_API_KEY environment variable is unset). CAVELL_API_URL = os.environ.get("CAVELL_API_URL", "https://prd.prism.cavell.app/api") CAVELL_API_KEY = os.environ.get("CAVELL_API_KEY") or getpass.getpass( "LLM Gateway key: " ) client = CavellClient( api_url=CAVELL_API_URL, api_key=CAVELL_API_KEY, fhir_base_url=FHIR_BASE_URL, fhir_api_path="/fhir", ) print("Connected.") tiers = client.list_tiers() print("\nAvailable tiers:") for t in tiers: default = " (default)" if t["default"] else "" print(f" {t['name']}{default}")

7. Create pipeline and seed¶

In [ ]:

# Choose a tier from the list above
TIER = "low"

pipeline = IngestionPipeline(
    client,
    tier=TIER,
    max_concurrency=3,
    default_organization=ORG_ID,
)

pipeline.seed(
    organizations=organizations,
    patients=patients,
    practitioners=practitioners,
)
print("Done.")

# Choose a tier from the list above TIER = "low" pipeline = IngestionPipeline( client, tier=TIER, max_concurrency=3, default_organization=ORG_ID, ) pipeline.seed( organizations=organizations, patients=patients, practitioners=practitioners, ) print("Done.")

8. Extract documents¶

Safe to re-run — the pipeline automatically queries FHIR for already-processed documents and skips them, so you never get duplicates.

extract_all() processes the whole dataset: it sorts every document by ascending date, then works through it in batches of BATCH_SIZE. Sorting globally first is what keeps batching cheap — batching an unsorted list would let a later batch carry notes older than ones already persisted, and those take the slower split-context path rather than being read straight against the record as it stands.

Within each patient, documents are processed in date order — so for a stay the admission note and investigations provide context before the discharge letter is extracted. If a document fails mid-patient, the patient's remaining documents still process (a failed document persists nothing, so their context stays consistent). Re-running later picks the failed document up automatically, on the split-context path.

In [ ]:

BATCH_SIZE = 50  # documents per batch; None processes everything in one call

BATCH_SIZE = 50 # documents per batch; None processes everything in one call

In [ ]:

batch_ok = 0
batch_fail = 0
batch_cost = 0.0

for i, outcome in enumerate(
    pipeline.extract_all(all_documents, batch_size=BATCH_SIZE), 1
):
    if outcome.success:
        batch_ok += 1
        if outcome.extract_result and outcome.extract_result.usage:
            batch_cost += outcome.extract_result.usage.estimated_cost
    else:
        batch_fail += 1
    print(f"[{i}] {outcome}")

if batch_ok or batch_fail:
    total = batch_ok + batch_fail
    print(f"\nRun: {batch_ok}/{total} succeeded, ${batch_cost:.3f}")
    total_done = pipeline.documents_processed
    total_all = total_done + pipeline.documents_failed
    print(f"Total: {total_done}/{total_all} succeeded, ${pipeline.total_cost:.3f}")
else:
    print("No documents to process.")

batch_ok = 0 batch_fail = 0 batch_cost = 0.0 for i, outcome in enumerate( pipeline.extract_all(all_documents, batch_size=BATCH_SIZE), 1 ): if outcome.success: batch_ok += 1 if outcome.extract_result and outcome.extract_result.usage: batch_cost += outcome.extract_result.usage.estimated_cost else: batch_fail += 1 print(f"[{i}] {outcome}") if batch_ok or batch_fail: total = batch_ok + batch_fail print(f"\nRun: {batch_ok}/{total} succeeded, ${batch_cost:.3f}") total_done = pipeline.documents_processed total_all = total_done + pipeline.documents_failed print(f"Total: {total_done}/{total_all} succeeded, ${pipeline.total_cost:.3f}") else: print("No documents to process.")

9. Cost projection¶

In [ ]:

if pipeline.documents_processed > 0:
    avg = pipeline.total_cost / pipeline.documents_processed
    print(f"Average cost per document: ${avg:.4f}")
    n = pipeline.documents_processed
    print(f"Session total: {n} docs, ${pipeline.total_cost:.3f}")

if pipeline.documents_processed > 0: avg = pipeline.total_cost / pipeline.documents_processed print(f"Average cost per document: ${avg:.4f}") n = pipeline.documents_processed print(f"Session total: {n} docs, ${pipeline.total_cost:.3f}")

10. Going backwards in time: split context¶

Everything above moved forward in time. This step shows what happens when part of a batch doesn't.

Eight notes were held back at load time. This step submits all eight in one call:

Notes	Patient	Stay	Dated	Against that patient's newest extracted note	Context sent
note-9011–note-9013	MRN-20002 Margaret Doyle	V-903 pyelonephritis	November 2022	2024-04-09 — backwards	split at the note's own date
note-9001–note-9003	MRN-20002 Margaret Doyle	V-902 wrist fracture	September 2023	2024-04-09 — backwards	split at the note's own date
note-9101–note-9102	MRN-20017 Hugo Vermeulen	V-018 readmission	July 2025	2025-06-09 — forwards	the record as it stands

Her two earlier stays are genuinely earlier illnesses — a urinary infection in 2022 and a left wrist fracture in 2023 — while the stay already extracted is her April 2024 hip fracture. The clinical picture differed between them:

	Nov 2022 / Sept 2023 (held back)	March–April 2024 (already extracted)
Osteoporosis	suspected, DEXA requested	established
Bone protection	none started	alendronic acid started
Hypertension onset	stated: 2015	recorded, but with no onset

That difference is exactly the problem. Sending these notes against her current FHIR state would show the model "osteoporosis, confirmed" and "alendronic acid" — facts from months in their future — and anything they created would carry that contamination backwards.

So the pipeline splits the context instead. Every document is checked against its own patient's watermark — the date of the newest already-processed document for that patient — and a document older than it is sent:

• context — only what was on record on its own date, so the model sees the patient as her clinician did;
• future_context — the resources her April 2024 stay added, kept separate so the API can reconcile against them without reading them as history;
• out_of_order: true — so the API knows which is which.

Resources are sorted into those two by provenance: each is dated by the newest already-processed document that created or updated it, read from that document's DocumentReference.context.related. What matters is when a fact entered the record, not when it happened.

The two MRN-20017 notes move forwards, so they take the ordinary path in the same call.

Re-run steps 3 and 5 first if you loaded the CSV before these rows existed — step 3 re-reads the file and step 5 rebuilds the held-back lists. You do not need to re-run step 8; skip_processed=True means the 108 already-extracted notes are skipped. Re-running this step is safe too: all eight notes are filtered out as already-processed the second time round.

In [ ]:

WATCHED = {"MRN-20002": "backdated notes", "MRN-20017": "forward-dated notes"}
TRACKED_TYPES = ["Condition", "MedicationRequest", "Procedure"]

fhir_ids = {mrn: client.find_patient_id(mrn) for mrn in WATCHED}
for mrn, fhir_id in fhir_ids.items():
    print(f"{mrn} -> Patient/{fhir_id}   ({WATCHED[mrn]})")


def _label(resource):
    """Human-readable name for a clinical resource."""
    for field in ("code", "medicationCodeableConcept"):
        concept = resource.get(field) or {}
        text = concept.get("text") or "".join(
            c.get("display", "") for c in (concept.get("coding") or [])[:1]
        )
        if text:
            return text
    return (resource.get("medicationReference") or {}).get("display", "(no code text)")


def snapshot(patient_fhir_id):
    """Map "Type/id" -> (versionId, label, resource) for the resources we watch.

    The whole resource is kept so step 10 can show which element a merge filled,
    not merely that the version moved.
    """
    snap = {}
    for rtype in TRACKED_TYPES:
        for r in client.get_patient_resources(patient_fhir_id, rtype):
            snap[f"{r['resourceType']}/{r['id']}"] = (
                r.get("meta", {}).get("versionId") or "?",
                _label(r),
                r,
            )
    return snap


before = {mrn: snapshot(fhir_id) for mrn, fhir_id in fhir_ids.items()}

for mrn, snap in before.items():
    print(f"\n{mrn} — {len(snap)} tracked resources before the call:\n")
    for key, (version, label, _r) in sorted(snap.items(), key=lambda kv: kv[1][1]):
        print(f"  v{version:<3} {key:<26} {label[:52]}")

tag = {d.document_id: "backdated" for d in backdated_documents}
tag.update({d.document_id: "forward  " for d in forward_documents})

print(f"\nAbout to submit {len(held_back_documents)} held-back notes in one call:")
for doc in sorted(held_back_documents, key=lambda doc: doc.date):
    print(
        f"  [{tag[doc.document_id]}] {doc.document_id}  {doc.date}  "
        f"{doc.text.splitlines()[0][:40]}"
    )

WATCHED = {"MRN-20002": "backdated notes", "MRN-20017": "forward-dated notes"} TRACKED_TYPES = ["Condition", "MedicationRequest", "Procedure"] fhir_ids = {mrn: client.find_patient_id(mrn) for mrn in WATCHED} for mrn, fhir_id in fhir_ids.items(): print(f"{mrn} -> Patient/{fhir_id} ({WATCHED[mrn]})") def _label(resource): """Human-readable name for a clinical resource.""" for field in ("code", "medicationCodeableConcept"): concept = resource.get(field) or {} text = concept.get("text") or "".join( c.get("display", "") for c in (concept.get("coding") or [])[:1] ) if text: return text return (resource.get("medicationReference") or {}).get("display", "(no code text)") def snapshot(patient_fhir_id): """Map "Type/id" -> (versionId, label, resource) for the resources we watch. The whole resource is kept so step 10 can show which element a merge filled, not merely that the version moved. """ snap = {} for rtype in TRACKED_TYPES: for r in client.get_patient_resources(patient_fhir_id, rtype): snap[f"{r['resourceType']}/{r['id']}"] = ( r.get("meta", {}).get("versionId") or "?", _label(r), r, ) return snap before = {mrn: snapshot(fhir_id) for mrn, fhir_id in fhir_ids.items()} for mrn, snap in before.items(): print(f"\n{mrn} — {len(snap)} tracked resources before the call:\n") for key, (version, label, _r) in sorted(snap.items(), key=lambda kv: kv[1][1]): print(f" v{version:<3} {key:<26} {label[:52]}") tag = {d.document_id: "backdated" for d in backdated_documents} tag.update({d.document_id: "forward " for d in forward_documents}) print(f"\nAbout to submit {len(held_back_documents)} held-back notes in one call:") for doc in sorted(held_back_documents, key=lambda doc: doc.date): print( f" [{tag[doc.document_id]}] {doc.document_id} {doc.date} " f"{doc.text.splitlines()[0][:40]}" )

In [ ]:

outcomes = pipeline.extract_all(held_back_documents)

split = [o for o in outcomes if o.out_of_order]
extracted = [o for o in outcomes if o.success]

print(f"{len(extracted)} extracted, {len(split)} of them against split context\n")
for outcome in sorted(outcomes, key=lambda o: o.document_id or ""):
    print(outcome)

print(
    f"\nSession total is now ${pipeline.total_cost:.3f} across "
    f"{pipeline.documents_processed} documents, with "
    f"{pipeline.documents_failed} failed."
)

outcomes = pipeline.extract_all(held_back_documents) split = [o for o in outcomes if o.out_of_order] extracted = [o for o in outcomes if o.success] print(f"{len(extracted)} extracted, {len(split)} of them against split context\n") for outcome in sorted(outcomes, key=lambda o: o.document_id or ""): print(outcome) print( f"\nSession total is now ${pipeline.total_cost:.3f} across " f"{pipeline.documents_processed} documents, with " f"{pipeline.documents_failed} failed." )

In [ ]:

after = {mrn: snapshot(fhir_id) for mrn, fhir_id in fhir_ids.items()}

for mrn in WATCHED:
    print(f"{mrn}: {len(before[mrn])} resources before -> {len(after[mrn])} after")

# Both patients only ever gain. On a re-run every held-back note is filtered out
# as already-processed, so an unchanged snapshot is expected then too.
for mrn in WATCHED:
    assert after[mrn].keys() >= before[mrn].keys(), (
        f"extraction must not remove resources for {mrn}"
    )

for mrn, label in (
    ("MRN-20002", "backdated, split context"),
    ("MRN-20017", "forward-dated"),
):
    gained = after[mrn].keys() - before[mrn].keys()
    if gained:
        print(f"\n{mrn} ({label}): {len(gained)} new:\n")
        for key in sorted(gained, key=lambda k: after[mrn][k][1]):
            version, text, _r = after[mrn][key]
            print(f"  v{version:<3} {key:<26} {text[:52]}")
    else:
        print(f"\n{mrn} ({label}): nothing new — extracted on an earlier run")
        print("           and skip_processed filtered these notes out this time.")

print(
    "\nHer 2023 wrist fracture was extracted against the record as it stood in\n"
    "September 2023 — osteoporosis still only suspected, no bone protection —\n"
    "while everything her April 2024 stay added travelled as future_context."
)

after = {mrn: snapshot(fhir_id) for mrn, fhir_id in fhir_ids.items()} for mrn in WATCHED: print(f"{mrn}: {len(before[mrn])} resources before -> {len(after[mrn])} after") # Both patients only ever gain. On a re-run every held-back note is filtered out # as already-processed, so an unchanged snapshot is expected then too. for mrn in WATCHED: assert after[mrn].keys() >= before[mrn].keys(), ( f"extraction must not remove resources for {mrn}" ) for mrn, label in ( ("MRN-20002", "backdated, split context"), ("MRN-20017", "forward-dated"), ): gained = after[mrn].keys() - before[mrn].keys() if gained: print(f"\n{mrn} ({label}): {len(gained)} new:\n") for key in sorted(gained, key=lambda k: after[mrn][k][1]): version, text, _r = after[mrn][key] print(f" v{version:<3} {key:<26} {text[:52]}") else: print(f"\n{mrn} ({label}): nothing new — extracted on an earlier run") print(" and skip_processed filtered these notes out this time.") print( "\nHer 2023 wrist fracture was extracted against the record as it stood in\n" "September 2023 — osteoporosis still only suspected, no bone protection —\n" "while everything her April 2024 stay added travelled as future_context." )

What the gate decided¶

The reconciliation runs inside the extraction API, so what we can read back here is its effect on the record.

Every one of these six notes re-states things Margaret was already known to have — that is what clinical notes do, each one restating the history it needs. The extractors will happily propose all of them again, because each note is read as though it were her latest. These are the proposals the gate has to reject:

Re-stated by the backdated notes	Already on record from	Expected
Hypertension	April 2024 PMH and the March 2024 GP review, no onset recorded	merged — create dropped, and the "diagnosed in 2015" from Nov 2022 fills the onset
Hypothyroidism	April 2024 PMH	dropped — nothing to add
Amlodipine	April 2024 discharge meds	dropped
Levothyroxine	April 2024 discharge meds	dropped
Osteoporosis	April 2024, established on bone-health review	dropped — Sept 2023 only suspects it, which is not new information
Osteopenia	reported on the April 2024 hip films	dropped — the 2023 wrist films report it too
Acute pyelonephritis	the March 2024 GP review's problem list ("November 2022 — resolved")	merged — the same November 2022 episode, seen from inside it; the stay notes may fill the abatement
Left distal radius fracture	the same GP review ("September 2023 — healed")	dropped — the same injury, already on record as resolved history

The last two rows exist because of the pre-admission GP review (note-0107, March 2024): it lists both earlier illnesses as resolved history, and it is processed in the main pass, months before these backdated stays arrive. For the gate, an acute condition matches when the two sides happened at the same time — and here they did: the GP review dates each episode to the very month the backdated stay describes. Same episode, seen from inside it rather than in retrospect. (Without that GP review, both conditions would be genuinely new and the correct answer would be create — a real illness must never be dropped just because it is acute and old.)

Reject all eight and her chart keeps one of each. Miss them and she ends up with two hypertensions, two pyelonephritis episodes and two amlodipine prescriptions — the duplication that made reverse-chronological ingestion unsafe in the first place.

And the other direction, what the gate must not reject:

Genuinely new in the backdated notes	Expected
Ciprofloxacin, co-amoxiclav, IV fluids, the renal ultrasound (Nov 2022)	created — the GP review says only "resolved after antibiotics"; these notes carry the actual treatment
Wrist films, haematoma block, closed reduction, cast (Sept 2023)	created — the procedures of an injury the record only knows as healed

The hypertension row is the one worth watching. A backdated note is often the earliest account of something, and the later note that re-stated the fact never repeated the onset. Filling that gap is the only way an out-of-order document improves the record rather than merely avoiding harm — and the fill is one-way: an element that already has a value is never rewritten, whatever this note says.

In [ ]:

from collections import Counter

mrn = "MRN-20002"
was, now = before[mrn], after[mrn]

# The findings the backdated notes re-state that the record ALREADY held. These
# are the only ones the gate is actually tested on — the rest of her chart (the
# hemiarthroplasty, the fascia iliaca block, the hip films) is never mentioned
# by these notes, so it staying put proves nothing.
RESTATED = {
    "Hypertension": "merged",  # Nov 2022 supplies the 2015 onset
    "Hypothyroidism": "dropped",
    "Amlodipine": "dropped",
    "Levothyroxine": "dropped",
    "Osteoporosis": "dropped",  # Sept 2023 only suspects it
    "Osteopenia": "dropped",
    # Both illnesses below are on record BEFORE the backdated notes arrive:
    # the March 2024 GP review (note-0107) lists them as resolved history,
    # dated to the very months these stays describe. Same episodes.
    "Pyelonephritis": "merged",  # Nov 2022 may fill the abatement
    # "radius", not "fracture" — the 2024 hip fracture also matches "fracture"
    "Radius fracture": "dropped",
}


def matching(snap, name):
    """Resources whose label names this finding (wording varies per run)."""
    return {k: v for k, v in snap.items() if name.lower() in v[1].lower()}


print("RE-STATED BY THE BACKDATED NOTES — must not become a second copy:\n")
leaked = []
for name, expected in RESTATED.items():
    had, has = matching(was, name), matching(now, name)
    filled = ""
    if expected == "merged":
        for key, (version, _label, resource) in has.items():
            if key in had and version != had[key][0]:
                old = had[key][2]
                changed = {
                    f: v
                    for f, v in resource.items()
                    if f not in ("meta", "text") and old.get(f) != v
                }
                shown = ", ".join(f"{f}={v!r}" for f, v in changed.items())
                filled = f"  filled {shown}"
    verdict = "held" if len(has) == len(had) else f"LEAKED {len(had)} -> {len(has)}"
    if len(has) != len(had):
        leaked.append(name)
    print(f"  {name:<18} on record x{len(had)} -> x{len(has)}   {verdict}{filled}")

print(
    f"\n{len(RESTATED) - len(leaked)}/{len(RESTATED)} rejected as expected"
    + (f" — LEAKED: {', '.join(leaked)}" if leaked else "")
)

# And what the gate let through: findings with nothing on record to match.
created = {k: v for k, v in now.items() if k not in was}
print(f"\nCREATED — nothing on record matched these ({len(created)}):\n")
for key, (_v, label, _r) in sorted(created.items(), key=lambda kv: kv[1][1]):
    print(f"  {key:<26} {label[:50]}")

print(
    f"\nHer chart went from {len(was)} tracked resources to {len(now)}."
    "\n\nExtraction is a language model, so the exact wording and the peripheral"
    "\nresources vary between runs. What should not vary: the six above are"
    "\nrejected, hypertension gains the 2015 onset, and her 2022 pyelonephritis"
    "\nand 2023 wrist fracture are both recorded."
)

from collections import Counter mrn = "MRN-20002" was, now = before[mrn], after[mrn] # The findings the backdated notes re-state that the record ALREADY held. These # are the only ones the gate is actually tested on — the rest of her chart (the # hemiarthroplasty, the fascia iliaca block, the hip films) is never mentioned # by these notes, so it staying put proves nothing. RESTATED = { "Hypertension": "merged", # Nov 2022 supplies the 2015 onset "Hypothyroidism": "dropped", "Amlodipine": "dropped", "Levothyroxine": "dropped", "Osteoporosis": "dropped", # Sept 2023 only suspects it "Osteopenia": "dropped", # Both illnesses below are on record BEFORE the backdated notes arrive: # the March 2024 GP review (note-0107) lists them as resolved history, # dated to the very months these stays describe. Same episodes. "Pyelonephritis": "merged", # Nov 2022 may fill the abatement # "radius", not "fracture" — the 2024 hip fracture also matches "fracture" "Radius fracture": "dropped", } def matching(snap, name): """Resources whose label names this finding (wording varies per run).""" return {k: v for k, v in snap.items() if name.lower() in v[1].lower()} print("RE-STATED BY THE BACKDATED NOTES — must not become a second copy:\n") leaked = [] for name, expected in RESTATED.items(): had, has = matching(was, name), matching(now, name) filled = "" if expected == "merged": for key, (version, _label, resource) in has.items(): if key in had and version != had[key][0]: old = had[key][2] changed = { f: v for f, v in resource.items() if f not in ("meta", "text") and old.get(f) != v } shown = ", ".join(f"{f}={v!r}" for f, v in changed.items()) filled = f" filled {shown}" verdict = "held" if len(has) == len(had) else f"LEAKED {len(had)} -> {len(has)}" if len(has) != len(had): leaked.append(name) print(f" {name:<18} on record x{len(had)} -> x{len(has)} {verdict}{filled}") print( f"\n{len(RESTATED) - len(leaked)}/{len(RESTATED)} rejected as expected" + (f" — LEAKED: {', '.join(leaked)}" if leaked else "") ) # And what the gate let through: findings with nothing on record to match. created = {k: v for k, v in now.items() if k not in was} print(f"\nCREATED — nothing on record matched these ({len(created)}):\n") for key, (_v, label, _r) in sorted(created.items(), key=lambda kv: kv[1][1]): print(f" {key:<26} {label[:50]}") print( f"\nHer chart went from {len(was)} tracked resources to {len(now)}." "\n\nExtraction is a language model, so the exact wording and the peripheral" "\nresources vary between runs. What should not vary: the six above are" "\nrejected, hypertension gains the 2015 onset, and her 2022 pyelonephritis" "\nand 2023 wrist fracture are both recorded." )

11. Find a stay to code¶

After extraction, each encounter_id is one FHIR Encounter (identifier urn:cavell:encounter|<id>) that links the stay's DocumentReferences, Conditions and Procedures. List the discharge letters — each one is the natural entry point for downstream coding of its stay.

V-018 is in the list because step 10 extracted it. So is the September 2023 V-902 stay — those notes went backwards in time, but they were extracted against split context rather than turned away.

In [ ]:

# Discharge letter per stay (the entry point for coding). The V-018
# notes are held back from step 8 and extracted by step 10, so they
# have to be added back here.
def is_discharge(text):
    return text.strip().upper().startswith("DISCHARGE LETTER")


print("encounter note_id     patient      summary")
for d in all_documents + forward_documents:
    if is_discharge(d.text):
        reason = next(
            (
                ln.split(":", 1)[1].strip()
                for ln in d.text.splitlines()
                if ln.strip().startswith("Reason for admission")
            ),
            "",
        )
        print(
            f"{d.encounter_id:<9} {d.document_id:<11} "
            f"{d.patient_identifier:<12} {reason[:50]}"
        )

# Discharge letter per stay (the entry point for coding). The V-018 # notes are held back from step 8 and extracted by step 10, so they # have to be added back here. def is_discharge(text): return text.strip().upper().startswith("DISCHARGE LETTER") print("encounter note_id patient summary") for d in all_documents + forward_documents: if is_discharge(d.text): reason = next( ( ln.split(":", 1)[1].strip() for ln in d.text.splitlines() if ln.strip().startswith("Reason for admission") ), "", ) print( f"{d.encounter_id:<9} {d.document_id:<11} " f"{d.patient_identifier:<12} {reason[:50]}" )

12. Delete a patient's data¶

If a patient's data looks wrong, delete all their resources and re-extract. Cascade delete removes the patient and everything referencing them; organizations and practitioners are unaffected.

After deleting, re-run in order: Step 7 (re-seeds patients) then Step 8 (only the deleted patient's docs are re-processed).

In [ ]:

DELETE_MRNS = ["MRN-20001"]  # List the MRNs to delete

for mrn in DELETE_MRNS:
    fhir_id = client.find_patient_id(mrn)
    if fhir_id:
        client.delete_patient_resources(fhir_id)
        print(f"Deleted {mrn} ({fhir_id})")
    else:
        print(f"WARNING: {mrn} not found — already deleted?")

DELETE_MRNS = ["MRN-20001"] # List the MRNs to delete for mrn in DELETE_MRNS: fhir_id = client.find_patient_id(mrn) if fhir_id: client.delete_patient_resources(fhir_id) print(f"Deleted {mrn} ({fhir_id})") else: print(f"WARNING: {mrn} not found — already deleted?")