Add FHIR sources and LLM config to healthchain.yaml

cookbook/cds_discharge_summarizer_hf_chat.py

@@ -1,98 +1,100 @@
+#!/usr/bin/env python3
+"""
+Discharge Note Summarizer (LangChain + HuggingFace Chat)
+
+CDS Hooks service that summarises discharge notes using a HuggingFace
+chat model via LangChain (DeepSeek R1 by default).
+
+Requirements:
+    pip install healthchain langchain-core langchain-huggingface python-dotenv
+    # HUGGINGFACEHUB_API_TOKEN env var required
+
+Run:
+    python cookbook/cds_discharge_summarizer_hf_chat.py
+    # POST /cds/cds-services/discharge-summarizer
+    # Docs at: http://localhost:8000/docs
+"""
+
 import os
 import getpass
 
-from healthchain.gateway import HealthChainAPI, CDSHooksService
-from healthchain.pipeline import SummarizationPipeline
-from healthchain.models import CDSRequest, CDSResponse
+from dotenv import load_dotenv
 
 from langchain_huggingface.llms import HuggingFaceEndpoint
 from langchain_huggingface import ChatHuggingFace
 from langchain_core.prompts import PromptTemplate
 from langchain_core.output_parsers import StrOutputParser
 
-from dotenv import load_dotenv
+from healthchain.gateway import HealthChainAPI, CDSHooksService
+from healthchain.pipeline import SummarizationPipeline
+from healthchain.models import CDSRequest, CDSResponse
 
 load_dotenv()
 
 
-if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
-    os.environ["HUGGINGFACEHUB_API_TOKEN"] = getpass.getpass("Enter your token: ")
+def create_chain():
+    if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
+        os.environ["HUGGINGFACEHUB_API_TOKEN"] = getpass.getpass(
+            "Enter your HuggingFace token: "
+        )
 
-
-def create_summarization_chain():
     hf = HuggingFaceEndpoint(
         repo_id="deepseek-ai/DeepSeek-R1-0528",
         task="text-generation",
         max_new_tokens=512,
         do_sample=False,
         repetition_penalty=1.03,
     )
-
     model = ChatHuggingFace(llm=hf)
-
-    template = """
-    You are a discharge planning assistant for hospital operations.
-    Provide a concise, objective summary focusing on actionable items
-    for care coordination, including appointments, medications, and
-    follow-up instructions. Format as bullet points with no preamble.\n'''{text}'''
-    """
-    prompt = PromptTemplate.from_template(template)
-
+    prompt = PromptTemplate.from_template(
+        "You are a discharge planning assistant for hospital operations. "
+        "Provide a concise, objective summary focusing on actionable items "
+        "for care coordination, including appointments, medications, and "
+        "follow-up instructions. Format as bullet points with no preamble.\n'''{text}'''"
+    )
     return prompt | model | StrOutputParser()
 
 
-# Create the healthcare application
-app = HealthChainAPI(
-    title="Discharge Note Summarizer",
-    description="AI-powered discharge note summarization service",
-)
-
-chain = create_summarization_chain()
-pipeline = SummarizationPipeline.load(
-    chain, source="langchain", template_path="templates/cds_card_template.json"
-)
-
-# Create CDS Hooks service
-cds = CDSHooksService()
+def create_app() -> HealthChainAPI:
+    chain = create_chain()
+    pipeline = SummarizationPipeline.load(
+        chain, source="langchain", template_path="templates/cds_card_template.json"
+    )
+    cds = CDSHooksService()
 
+    @cds.hook("encounter-discharge", id="discharge-summarizer")
+    def discharge_summarizer(request: CDSRequest) -> CDSResponse:
+        return pipeline.process_request(request)
 
-@cds.hook("encounter-discharge", id="discharge-summarizer")
-def discharge_summarizer(request: CDSRequest) -> CDSResponse:
-    result = pipeline.process_request(request)
-    return result
+    app = HealthChainAPI(
+        title="Discharge Note Summarizer",
+        description="AI-powered discharge note summarization service",
+        port=8000,
+        service_type="cds-hooks",
+    )
+    app.register_service(cds, path="/cds")
+    return app
 
 
-# Register the CDS service
-app.register_service(cds, path="/cds")
+app = create_app()
 
 
 if __name__ == "__main__":
-    import uvicorn
     import threading
-
     from healthchain.sandbox import SandboxClient
 
-    # Start the API server in a separate thread
-    def start_api():
-        uvicorn.run(app, port=8000)
-
-    api_thread = threading.Thread(target=start_api, daemon=True)
+    api_thread = threading.Thread(target=app.run, daemon=True)
     api_thread.start()
 
-    # Create sandbox client and load test data
     client = SandboxClient(
         url="http://localhost:8000/cds/cds-services/discharge-summarizer",
         workflow="encounter-discharge",
     )
-    # Load discharge notes from CSV
     client.load_free_text(
         csv_path="data/discharge_notes.csv",
         column_name="text",
     )
-    # Send requests and get responses
     responses = client.send_requests()
-
-    # Save results
     client.save_results("./output/")
 
     try:

cookbook/cds_discharge_summarizer_hf_trf.py

@@ -1,71 +1,79 @@
+#!/usr/bin/env python3
+"""
+Discharge Note Summarizer (Transformer)
+
+CDS Hooks service that summarises discharge notes using a fine-tuned
+HuggingFace transformer model (PEGASUS).
+
+Requirements:
+    pip install healthchain transformers torch python-dotenv
+    # HUGGINGFACEHUB_API_TOKEN env var required
+
+Run:
+    python cookbook/cds_discharge_summarizer_hf_trf.py
+    # POST /cds/cds-services/discharge-summarizer
+    # Docs at: http://localhost:8000/docs
+"""
+
 import os
 import getpass
 
+from dotenv import load_dotenv
+
 from healthchain.gateway import HealthChainAPI, CDSHooksService
 from healthchain.pipeline import SummarizationPipeline
 from healthchain.models import CDSRequest, CDSResponse
 
-from dotenv import load_dotenv
-
 load_dotenv()
 
 
-if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
-    os.environ["HUGGINGFACEHUB_API_TOKEN"] = getpass.getpass("Enter your token: ")
-
-
-# Create the healthcare application
-app = HealthChainAPI(
-    title="Discharge Note Summarizer",
-    description="AI-powered discharge note summarization service",
-)
+def create_pipeline() -> SummarizationPipeline:
+    if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
+        os.environ["HUGGINGFACEHUB_API_TOKEN"] = getpass.getpass(
+            "Enter your HuggingFace token: "
+        )
+    return SummarizationPipeline.from_model_id(
+        "google/pegasus-xsum", source="huggingface", task="summarization"
+    )
 
-# Initialize pipeline
-pipeline = SummarizationPipeline.from_model_id(
-    "google/pegasus-xsum", source="huggingface", task="summarization"
-)
 
-# Create CDS Hooks service
-cds = CDSHooksService()
+def create_app() -> HealthChainAPI:
+    pipeline = create_pipeline()
+    cds = CDSHooksService()
 
+    @cds.hook("encounter-discharge", id="discharge-summarizer")
+    def discharge_summarizer(request: CDSRequest) -> CDSResponse:
+        return pipeline.process_request(request)
 
-@cds.hook("encounter-discharge", id="discharge-summarizer")
-def discharge_summarizer(request: CDSRequest) -> CDSResponse:
-    result = pipeline.process_request(request)
-    return result
+    app = HealthChainAPI(
+        title="Discharge Note Summarizer",
+        description="AI-powered discharge note summarization service",
+        port=8000,
+        service_type="cds-hooks",
+    )
+    app.register_service(cds, path="/cds")
+    return app
 
 
-# Register the CDS service
-app.register_service(cds, path="/cds")
+app = create_app()
 
 
 if __name__ == "__main__":
-    import uvicorn
     import threading
-
     from healthchain.sandbox import SandboxClient
 
-    # Start the API server in a separate thread
-    def start_api():
-        uvicorn.run(app, port=8000)
-
-    api_thread = threading.Thread(target=start_api, daemon=True)
+    api_thread = threading.Thread(target=app.run, daemon=True)
     api_thread.start()
 
-    # Create sandbox client and load test data
     client = SandboxClient(
         url="http://localhost:8000/cds/cds-services/discharge-summarizer",
         workflow="encounter-discharge",
     )
-    # Load discharge notes from CSV
     client.load_free_text(
         csv_path="data/discharge_notes.csv",
         column_name="text",
     )
-    # Send requests and get responses
     responses = client.send_requests()
-
-    # Save results
     client.save_results("./output/")
 
     try:

cookbook/multi_ehr_data_aggregation.py

@@ -13,7 +13,7 @@
 - Cerner Open Sandbox: No auth needed
 
 Run:
-- python data_aggregation.py
+    python cookbook/multi_ehr_data_aggregation.py
 """
 
 from typing import List
@@ -23,7 +23,7 @@
 from healthchain.fhir.r4b import Bundle, Condition, Annotation
 
 from healthchain.gateway import FHIRGateway, HealthChainAPI
-from healthchain.gateway.clients.fhir.base import FHIRAuthConfig
+from healthchain.gateway.clients import FHIRAuthConfig
 from healthchain.pipeline import Pipeline
 from healthchain.io.containers import Document
 from healthchain.fhir import merge_bundles
@@ -100,15 +100,16 @@ def get_unified_patient(patient_id: str, sources: List[str]) -> Bundle:
 
         return doc.fhir.bundle
 
-    app = HealthChainAPI()
-    app.register_gateway(gateway)
+    app = HealthChainAPI(
+        title="Multi-EHR Data Aggregation",
+        description="Aggregate patient data from multiple FHIR sources",
+        port=8888,
+        service_type="fhir-gateway",
+    )
+    app.register_gateway(gateway, path="/fhir")
 
     return app
 
 
 if __name__ == "__main__":
-    import uvicorn
-
-    app = create_app()
-    uvicorn.run(app, port=8888)
-    # Runs at: http://127.0.0.1:8888/
+    create_app().run()

cookbook/notereader_clinical_coding_fhir.py

@@ -4,13 +4,11 @@
 Demonstrates FHIR-native pipelines, legacy system integration, and multi-source data handling.
 
 Requirements:
-- pip install healthchain
-- pip install scispacy
-- pip install https://s3-us-west-2.amazonaws.com/ai2-s2-scispacy/releases/v0.5.4/en_core_sci_sm-0.5.4.tar.gz
-- pip install python-dotenv
+    pip install healthchain scispacy python-dotenv
+    pip install https://s3-us-west-2.amazonaws.com/ai2-s2-scispacy/releases/v0.5.4/en_core_sci_sm-0.5.4.tar.gz
 
 Run:
-- python notereader_clinical_coding_fhir.py  # Demo and start server
+    python cookbook/notereader_clinical_coding_fhir.py
 """
 
 import logging
@@ -21,7 +19,7 @@
 from healthchain.fhir import add_provenance_metadata
 from healthchain.gateway.api import HealthChainAPI
 from healthchain.gateway.fhir import FHIRGateway
-from healthchain.gateway.clients.fhir.base import FHIRAuthConfig
+from healthchain.gateway.clients import FHIRAuthConfig
 from healthchain.gateway.soap import NoteReaderService
 from healthchain.io import CdaAdapter, Document
 from healthchain.models import CdaRequest
@@ -104,7 +102,12 @@ def ai_coding_workflow(request: CdaRequest):
         return cda_response
 
     # Register services
-    app = HealthChainAPI(title="Epic CDI Service with FHIR integration")
+    app = HealthChainAPI(
+        title="Epic CDI Service",
+        description="Clinical document intelligence with FHIR and NoteReader integration",
+        port=8000,
+        service_type="fhir-gateway",
+    )
     app.register_gateway(fhir_gateway, path="/fhir")
     app.register_service(note_service, path="/notereader")
 
@@ -117,18 +120,12 @@ def ai_coding_workflow(request: CdaRequest):
 
 if __name__ == "__main__":
     import threading
-    import uvicorn
-
     from time import sleep
     from healthchain.sandbox import SandboxClient
 
-    # Start server
-    def run_server():
-        uvicorn.run(app, port=8000, log_level="warning")
-
-    server_thread = threading.Thread(target=run_server, daemon=True)
+    server_thread = threading.Thread(target=app.run, daemon=True)
     server_thread.start()
-    sleep(2)  # Wait for startup
+    sleep(2)
 
     # Create sandbox client for testing
     client = SandboxClient(

cookbook/sepsis_cds_hooks.py

@@ -115,7 +115,12 @@ def sepsis_alert(request: CDSRequest) -> CDSResponse:
 
         return CDSResponse(cards=[])
 
-    app = HealthChainAPI(title="Sepsis CDS Hooks")
+    app = HealthChainAPI(
+        title="Sepsis CDS Hooks",
+        description="Real-time sepsis risk alerts via CDS Hooks",
+        port=8000,
+        service_type="cds-hooks",
+    )
     app.register_service(cds, path="/cds")
 
     return app
@@ -126,15 +131,10 @@ def sepsis_alert(request: CDSRequest) -> CDSResponse:
 
 if __name__ == "__main__":
     import threading
-    import uvicorn
     from time import sleep
     from healthchain.sandbox import SandboxClient
 
-    # Start server
-    def run_server():
-        uvicorn.run(app, port=8000, log_level="warning")
-
-    server = threading.Thread(target=run_server, daemon=True)
+    server = threading.Thread(target=app.run, daemon=True)
     server.start()
     sleep(2)