Multimodal AI#

"""
Extract structured data from a clinical lab report PDF.
Context: HMS scenario — processing inbound lab results from an external
reference lab that sends PDF reports rather than HL7/FHIR interfaces.
The extracted structured data is ingested into the Epic EHR as manual results.

Dependencies:
    anthropic>=0.49.0
    pillow>=10.2.0
    python-dotenv>=1.0.0
    pdf2image>=1.16.0 (requires poppler)
"""
import anthropic
import base64
import json
from pathlib import Path
from typing import Any


def pdf_page_to_base64(pdf_path: Path, page_number: int = 0) -> str:
    """Convert a PDF page to base64-encoded PNG for LLM processing."""
    try:
        from pdf2image import convert_from_path
    except ImportError:
        raise ImportError("Install pdf2image: pip install pdf2image (requires poppler)")

    images = convert_from_path(str(pdf_path), first_page=page_number + 1, last_page=page_number + 1)
    if not images:
        raise ValueError(f"No page {page_number} in {pdf_path}")

    import io
    buffer = io.BytesIO()
    images[0].save(buffer, format="PNG")
    return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")


def extract_lab_results(
    pdf_path: Path,
    client: anthropic.Anthropic,
) -> dict[str, Any]:
    """
    Extract structured lab results from a PDF lab report.
    Returns structured data suitable for EHR ingestion.

    Note: This function processes PHI (lab results are PHI).
    Ensure the Anthropic API call is made within the HIPAA BAA boundary.
    PHI is NOT logged. Results are not cached.
    """
    image_data = pdf_page_to_base64(pdf_path, page_number=0)

    response = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=2048,
        system=(
            "Extract laboratory results from medical lab report images. "
            "Return ONLY valid JSON with this structure:\n"
            "{\n"
            '  "patient_name": "string",\n'
            '  "dob": "YYYY-MM-DD",\n'
            '  "collection_date": "YYYY-MM-DD",\n'
            '  "ordering_provider": "string",\n'
            '  "lab_results": [\n'
            '    {"test_name": "...", "result": "...", "unit": "...", '
            '"reference_range": "...", "flag": "H|L|C|null"}\n'
            "  ],\n"
            '  "critical_values": ["list of tests with critical flags"]\n'
            "}\n"
            "Flag meanings: H=High, L=Low, C=Critical. "
            "If a field is not visible, use null."
        ),
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": image_data,
                    },
                },
                {
                    "type": "text",
                    "text": "Extract all laboratory results from this lab report.",
                },
            ],
        }],
    )

    raw_output = response.content[0].text.strip()
    return json.loads(raw_output)

"""
Process a scanned clinical document (e.g., a handwritten referral letter).
Context: HMS receives scanned referral letters from community physicians.
This pipeline converts the scan to structured text and extracts key clinical data.
"""
import anthropic
import base64
from pathlib import Path
from dataclasses import dataclass


@dataclass
class ReferralExtract:
    referring_provider: str | None
    referring_facility: str | None
    reason_for_referral: str | None
    relevant_history: str | None
    current_medications: list[str]
    urgency: str  # "routine" | "urgent" | "emergent"
    requested_specialty: str | None
    patient_summary: str


def process_scanned_referral(
    image_path: Path,
    client: anthropic.Anthropic,
) -> ReferralExtract:
    """
    Extract structured data from a scanned physician referral letter.
    Handles handwritten and typed formats.
    """
    image_bytes = image_path.read_bytes()
    image_data = base64.standard_b64encode(image_bytes).decode("utf-8")

    # Determine media type from extension
    suffix = image_path.suffix.lower()
    media_type_map = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png"}
    media_type = media_type_map.get(suffix, "image/jpeg")

    response = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=(
            "You are processing a scanned physician referral letter. "
            "Extract all clinically relevant information. "
            "If handwriting is unclear, indicate uncertainty with [?]. "
            "Return JSON matching the ReferralExtract schema: "
            '{"referring_provider": "...", "referring_facility": "...", '
            '"reason_for_referral": "...", "relevant_history": "...", '
            '"current_medications": ["..."], '
            '"urgency": "routine|urgent|emergent", '
            '"requested_specialty": "...", '
            '"patient_summary": "2-3 sentence clinical summary"}'
        ),
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "base64", "media_type": media_type, "data": image_data},
                },
                {"type": "text", "text": "Extract the referral information from this document."},
            ],
        }],
    )

    import json
    data = json.loads(response.content[0].text.strip())
    return ReferralExtract(**data)

"""
Multimodal RAG: combine a clinical image with retrieved clinical guidelines.
Context: HMS radiology workflow — a chest X-ray is flagged by radiology AI
as potentially showing pneumonia. The clinical AI combines the image with
retrieved clinical guidelines to provide the ordering physician with
context-aware clinical decision support.

NOTE: This is NOT a diagnostic AI. It is a clinical knowledge support tool.
The image interpretation provided to the LLM is from a separate, cleared
radiology AI. The LLM provides clinical knowledge context, not diagnosis.
"""
import anthropic
import base64
from pathlib import Path


def clinical_image_support(
    image_path: Path,
    radiology_ai_findings: str,  # Output from FDA-cleared radiology AI
    clinical_question: str,
    retrieved_guidelines: list[str],  # From RAG pipeline
    client: anthropic.Anthropic,
) -> str:
    """
    Combine radiology AI findings, the CXR image, and retrieved guidelines
    to provide clinical knowledge context to the ordering physician.

    IMPORTANT: This tool does NOT provide a diagnosis. It provides:
    - Context from clinical guidelines relevant to the AI findings
    - Suggested clinical questions to consider
    - Relevant clinical decision support information

    All clinical decisions remain with the ordering physician.
    """
    image_data = base64.standard_b64encode(image_path.read_bytes()).decode("utf-8")

    guidelines_context = "\n\n---\n\n".join(
        f"Guideline {i+1}:\n{g}" for i, g in enumerate(retrieved_guidelines)
    )

    response = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        system=(
            "You are a clinical knowledge assistant. You provide relevant clinical "
            "guideline information to support physician clinical decision-making. "
            "You do NOT diagnose. You do NOT interpret radiological images. "
            "A separate FDA-cleared radiology AI has provided preliminary findings. "
            "Your role is to provide relevant clinical guideline context based on "
            "those findings. Always defer to the ordering physician's clinical judgment. "
            "Always include: 'Clinical decision-making remains with the ordering physician.'"
        ),
        messages=[{
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": image_data,
                    },
                },
                {
                    "type": "text",
                    "text": (
                        f"Radiology AI preliminary findings: {radiology_ai_findings}\n\n"
                        f"Clinical question: {clinical_question}\n\n"
                        f"Retrieved Clinical Guidelines:\n\n{guidelines_context}\n\n"
                        "Please provide relevant clinical guideline context based on "
                        "the radiology AI findings. Do not interpret the image directly."
                    ),
                },
            ],
        }],
    )

    return response.content[0].text

def dicom_to_png_for_llm(
    dicom_path: Path,
    window_center: int = 40,   # Soft tissue window for CXR
    window_width: int = 400,
) -> bytes:
    """
    Convert DICOM image to PNG for LLM processing.
    Applies windowing for appropriate tissue visualization.

    IMPORTANT: Strip DICOM metadata tags containing PHI before
    sending to any external API. DICOM metadata includes patient
    name, DOB, MRN, and other identifiers.
    """
    import pydicom
    import numpy as np
    from PIL import Image
    import io

    ds = pydicom.dcmread(str(dicom_path))
    pixel_array = ds.pixel_array.astype(float)

    # Apply HU windowing
    min_val = window_center - window_width / 2
    max_val = window_center + window_width / 2
    pixel_array = np.clip(pixel_array, min_val, max_val)
    pixel_array = ((pixel_array - min_val) / window_width * 255).astype(np.uint8)

    image = Image.fromarray(pixel_array)
    if image.mode != "RGB":
        image = image.convert("RGB")

    buffer = io.BytesIO()
    image.save(buffer, format="PNG")
    return buffer.getvalue()