Cavell SDK — CSV Extraction Demo¶

Extract structured FHIR resources from clinical notes stored in a CSV file.

The SDK works in two steps:

1. Seed — organizations, practitioners, and patients into FHIR
2. Extract — clinical notes, 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.

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 = "notes.csv"  # Adjust to your CSV path

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 = "notes.csv" # Adjust to your CSV path 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¶

Map your CSV column headers to the SDK fields. Set any optional column to None if your CSV doesn't have it.

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 -> its 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.
# Map a label to a CSV column name. Add or remove entries as needed.
# 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 -> its 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. # Map a label to a CSV column name. Add or remove entries as needed. # 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

all_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,
)

print(f"{len(all_documents)} documents")

from cavell_client import Document all_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, ) print(f"{len(all_documents)} documents")

Data profile¶

In [ ]:

from collections import Counter

MAX_BAR = 40  # max width of histogram bars


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


# --- Notes per patient ---
notes_per_patient = Counter(d.patient_identifier for d in all_documents)
counts = sorted(notes_per_patient.values())
n_patients = len(counts)
total_notes = sum(counts)
avg = total_notes / n_patients
mid = n_patients // 2
median = counts[mid] if n_patients % 2 else (counts[mid - 1] + counts[mid]) / 2

print(f"Patients: {n_patients}")
print(
    f"Notes per patient:"
    f"  min={counts[0]}  max={counts[-1]}"
    f"  avg={avg:.1f}  median={median}"
)

# Bucket into ranges of 5
step = 5
lo = (counts[0] // step) * step + 1
hi = ((counts[-1] // step) + 1) * step
buckets = []
for start in range(lo, hi + 1, step):
    end = start + step - 1
    c = sum(1 for v in counts if start <= v <= end)
    if c:
        buckets.append((f"{start}\u2013{end}", c))
max_c = max(c for _, c in buckets)

print("\n  notes/patient | patients")
for label, c in buckets:
    print(f"  {label:>11}   | {c:>4}  {bar(c, max_c)}")

# --- Note length (characters) ---
lengths = sorted(len(d.text) for d in all_documents)
n = len(lengths)
avg_len = sum(lengths) / n
mid = n // 2
median_len = lengths[mid] if n % 2 else (lengths[mid - 1] + lengths[mid]) / 2

print(
    f"\nNote length (chars):"
    f"  min={lengths[0]:,}  max={lengths[-1]:,}"
    f"  avg={avg_len:,.0f}  median={median_len:,.0f}"
)
print(f"Total content: {sum(lengths):,} chars across {n} notes")

edges = [0, 500, 1_000, 2_000, 5_000, 10_000]
len_buckets = []
for i in range(len(edges)):
    lo_e = edges[i]
    if i + 1 < len(edges):
        hi_e = edges[i + 1]
        if lo_e:
            label = f"{lo_e + 1:,}\u2013{hi_e:,}"
        else:
            label = f"\u2264{hi_e:,}"
        c = sum(1 for x in lengths if lo_e < x <= hi_e)
    else:
        label = f"{lo_e + 1:,}+"
        c = sum(1 for x in lengths if x > lo_e)
    if c:
        len_buckets.append((label, c))
max_lc = max(c for _, c in len_buckets)

print("\n  length (chars) | notes")
for label, c in len_buckets:
    print(f"  {label:>14}  | {c:>4}  {bar(c, max_lc)}")

from collections import Counter MAX_BAR = 40 # max width of histogram bars def bar(n, max_n): w = round(n / max_n * MAX_BAR) if max_n else 0 return "\u2588" * w # --- Notes per patient --- notes_per_patient = Counter(d.patient_identifier for d in all_documents) counts = sorted(notes_per_patient.values()) n_patients = len(counts) total_notes = sum(counts) avg = total_notes / n_patients mid = n_patients // 2 median = counts[mid] if n_patients % 2 else (counts[mid - 1] + counts[mid]) / 2 print(f"Patients: {n_patients}") print( f"Notes per patient:" f" min={counts[0]} max={counts[-1]}" f" avg={avg:.1f} median={median}" ) # Bucket into ranges of 5 step = 5 lo = (counts[0] // step) * step + 1 hi = ((counts[-1] // step) + 1) * step buckets = [] for start in range(lo, hi + 1, step): end = start + step - 1 c = sum(1 for v in counts if start <= v <= end) if c: buckets.append((f"{start}\u2013{end}", c)) max_c = max(c for _, c in buckets) print("\n notes/patient | patients") for label, c in buckets: print(f" {label:>11} | {c:>4} {bar(c, max_c)}") # --- Note length (characters) --- lengths = sorted(len(d.text) for d in all_documents) n = len(lengths) avg_len = sum(lengths) / n mid = n // 2 median_len = lengths[mid] if n % 2 else (lengths[mid - 1] + lengths[mid]) / 2 print( f"\nNote length (chars):" f" min={lengths[0]:,} max={lengths[-1]:,}" f" avg={avg_len:,.0f} median={median_len:,.0f}" ) print(f"Total content: {sum(lengths):,} chars across {n} notes") edges = [0, 500, 1_000, 2_000, 5_000, 10_000] len_buckets = [] for i in range(len(edges)): lo_e = edges[i] if i + 1 < len(edges): hi_e = edges[i + 1] if lo_e: label = f"{lo_e + 1:,}\u2013{hi_e:,}" else: label = f"\u2264{hi_e:,}" c = sum(1 for x in lengths if lo_e < x <= hi_e) else: label = f"{lo_e + 1:,}+" c = sum(1 for x in lengths if x > lo_e) if c: len_buckets.append((label, c)) max_lc = max(c for _, c in len_buckets) print("\n length (chars) | notes") for label, c in len_buckets: print(f" {label:>14} | {c:>4} {bar(c, max_lc)}")

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 makes batching safe — batching an unsorted list would let a later batch carry notes older than ones already persisted, which lose their updates to the chronology guard. Lower BATCH_SIZE to bound how much work an interruption loses; each batch costs one FHIR query per patient in it, so prefer larger values.

Within each patient, documents are always processed in date order — this matters because earlier notes provide clinical context for later ones. 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 = 500  # documents per batch; None processes everything in one call

BATCH_SIZE = 500 # 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
    cost = pipeline.total_cost
    print(f"Total: {total_done}/{total_all} succeeded, ${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 cost = pipeline.total_cost print(f"Total: {total_done}/{total_all} succeeded, ${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. 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:

1. Step 7 — fresh pipeline, re-seeds all patients
2. Step 8 — extract (only deleted patient's docs are re-processed)

If you forget to re-seed, extract() will catch the missing patient and raise RuntimeError before making any API calls.

In [ ]:

DELETE_MRNS = ["MRN-12345"]  # 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-12345"] # 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?")