HealthChain Python SDK

The Document class is a container designed to manage both clinical text and structured healthcare data. It is used for clinical notes, discharge summaries, patient records, and any data that combines unstructured text with structured FHIR resources.

A Document is organized around three primary attributes:

1. doc.text: The raw input clinical text.
2. doc.fhir: Manages FHIR resources, including clinical lists (problems, medications, allergies), Bundles, DocReference, and patient information.
3. doc.cds: Manages Clinical Decision Support (CDS) outputs, such as recommendation cards and suggested actions.

You can integrate your own NLP or ML libraries (like spaCy, HuggingFace, or LangChain) by processing doc.text and then updating the structured components via doc.update_problem_list() or by writing directly to doc.fhir.

from healthchain.io import Document

doc = Document("Patient presents with uncontrolled hypertension and Type 2 diabetes")
# doc.text contains the string
# doc.fhir and doc.cds provide structured management

A HealthChain Pipeline can be used to transform a FHIR Document into a structured plain-text context block. This ensures the LLM receives human-readable clinical information instead of raw JSON.

When a Document is initialized with a FHIR Bundle, you can use doc.fhir.get_resources("ResourceType") to extract specific resources.

Common resources to include:

• Condition
• Appointment
• CarePlan
• MedicationRequest
• Observation
• DocumentReference

from healthchain.pipeline import Pipeline
from healthchain.io.containers import Document

def create_pipeline() -> Pipeline:
    pipeline = Pipeline()

    @pipeline.add_node
    def format_context(doc: Document) -> Document:
        conditions = doc.fhir.get_resources("Condition")
        appointments = doc.fhir.get_resources("Appointment")
        careplans = doc.fhir.get_resources("CarePlan")

        lines = ["PATIENT CLINICAL CONTEXT"]
        if conditions:
            lines.append("\nDiagnoses:")
            for c in conditions:
                onset = c.onsetDateTime
                lines.append(
                    f"  - {c.code.text}" + (f" (since {onset})" if onset else "")
                )
        if appointments:
            lines.append("\nUpcoming Appointments:")
            for a in appointments:
                lines.append(f"  - {a.description}: {a.start}")
        if careplans:
            lines.append("\nCare Plan:")
            for cp in careplans:
                lines.append(f"  {cp.description}")

        doc.text = "\n".join(lines)
        return doc

    return pipeline

# Usage with a Document
doc = Document(data=bundle)
conditions = doc.fhir.get_resources("Condition")

The configuration system uses a hierarchical YAML structure. Key components include:

• defaults.yaml: Global default values and fallback values for FHIR resources.
• environments/: Environment-specific overrides (e.g., development.yaml, production.yaml).
• interop/: Format-specific settings (e.g., interop/cda/).
• mappings/: Terminology and code system mapping tables.
• templates/: Templates for generating healthcare documents.

CDA Configuration Loading: CDA configs are loaded from configs/interop/cda/. The directory structure determines the access key. For example:

• configs/interop/cda/document/ccd.yaml $\rightarrow$ cda.document.ccd
• configs/interop/cda/sections/problems.yaml $\rightarrow$ cda.sections.problems

defaults:
  common:
    id_prefix: "hc-"
  resources:
    Condition:
      clinicalStatus:
        coding:
          - system: "http://terminology.hl7.org/CodeSystem/condition-clinical"
            code: "unknown"
            display: "Unknown"

Transform handlers allow you to create custom API endpoints that process and enhance FHIR resources (e.g., adding AI insights) before returning them. They are registered using the @gateway.transform(ResourceType) decorator.

When a request is made to GET /fhir/transform/{ResourceType}/{id}?source={source}, the decorated function is executed.

Async Transform Example

@fhir_gateway.transform(Patient)
async def get_enhanced_patient_summary(id: str, source: str = None) -> Patient:
    async with fhir_gateway.modify(Patient, id, source=source) as patient:
        # ... perform logic (e.g., search for observations, run NLP) ...
        patient.extension = patient.extension or []
        patient.extension.append({
            "url": "http://healthchain.org/fhir/summary",
            "valueString": "AI Insight"
        })
        return patient

@fhir_gateway.transform(Patient)
async def get_enhanced_patient_summary(id: str, source: str = None) -> Patient:
    async with fhir_gateway.modify(Patient, id, source=source) as patient:
        # Logic here
        return patient