Skip to content

Build a CDS Hooks Service for Discharge Summarization

This tutorial shows you how to build a CDS service that integrates with EHR systems. We'll automatically summarize discharge notes and return actionable recommendations using the CDS Hooks standard.

Check out the full working example here!

Setup

Make sure you have a Hugging Face API token and set it as the HUGGINGFACEHUB_API_TOKEN environment variable.

import getpass
import os

if not os.getenv("HUGGINGFACEHUB_API_TOKEN"):
    os.environ["HUGGINGFACEHUB_API_TOKEN"] = getpass.getpass(
      "Enter your token: "
    )

If you are using a chat model, make sure you have the necessary langchain packages installed.

pip install langchain langchain-huggingface

Initialize the pipeline

First, we'll create a summarization pipeline with domain-specific prompting for discharge workflows. You can choose between:

  • Transformer models fine-tuned for clinical summarization (like google/pegasus-xsum)
  • Large Language Models with custom clinical prompting (like zephyr-7b-beta)

For LLM approaches, we'll use LangChain for better prompting.

from healthchain.pipeline import SummarizationPipeline

pipeline = SummarizationPipeline.from_model_id(
  "google/pegasus-xsum", source="huggingface", task="summarization"
  )

from healthchain.pipeline import SummarizationPipeline

from langchain_huggingface.llms import HuggingFaceEndpoint
from langchain_huggingface import ChatHuggingFace
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

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.\n'''{text}'''
"""
prompt = PromptTemplate.from_template(template)

chain = prompt | model | StrOutputParser()

pipeline = SummarizationPipeline.load(chain, source="langchain")

The SummarizationPipeline automatically:

  • Parses FHIR resources from CDS Hooks requests
  • Extracts clinical text from discharge documents
  • Formats outputs as CDS cards according to the CDS Hooks specification

Add the CDS FHIR Adapter

The CdsFhirAdapter converts between CDS Hooks requests and HealthChain's Document format. This makes it easy to work with FHIR data in CDS workflows.

from healthchain.io import CdsFhirAdapter

cds_adapter = CdsFhirAdapter()

# Parse the CDS request to a Document object
cds_adapter.parse(request)

# Format the Document object back to a CDS response
cds_adapter.format(doc)

What it does:

  • Parses FHIR resources from CDS requests
  • Extracts text from DocumentReference resources
  • Formats responses as CDS cards

Set up the CDS service

Now let's create the CDS service. HealthChainAPI gives you discovery endpoints, validation, and docs automatically.

from healthchain.gateway import HealthChainAPI, CDSHooksService
from healthchain.models import CDSRequest, CDSResponse
from healthchain.io import CdsFhirAdapter

def create_pipeline():
    """Build the discharge summarization pipeline"""
    # Configure your pipeline (using previous examples)
    return pipeline

def create_app():
    """Create the CDS Hooks application"""
    pipeline = create_pipeline()
    adapter = CdsFhirAdapter()

    # Initialize the CDS service
    cds_service = CDSHooksService()

    # Define the CDS service function
    @cds_service.hook("encounter-discharge", id="discharge-summary")
    def handle_discharge_summary(request: CDSRequest) -> CDSResponse:
        """Process discharge summaries with AI"""
        # Parse CDS request to internal Document format
        doc = adapter.parse(request)

        # Process through AI pipeline
        processed_doc = pipeline(doc)

        # Format response with CDS cards
        response = adapter.format(processed_doc)
        return response

    # Register the service with the API gateway
    app = HealthChainAPI(title="Discharge Summary CDS Service")
    app.register_service(cds_service)

    return app

Test with sample clinical data

Let's test the service with some sample discharge notes using the sandbox utility and the CdsDataGenerator:

from healthchain.data_generators import CdsDataGenerator

data_generator = CdsDataGenerator()
data = data_generator.generate(
  free_text_path="data/discharge_notes.csv", column_name="text"
)
print(data.model_dump())
# {
#    "prefetch": {
#        "entry": [
#        {
#            "resource": {
#                "resourceType": "Bundle",
#                ...
#            }
#        }
#    ]
# }

The data generator returns a Prefetch object, which ensures that the data is parsed correctly by the CDS service.

Run the complete example

Run the service with uvicorn:

import uvicorn

app = create_app()

uvicorn.run(app)

What happens when you run this

Workflow Overview

  • URL: http://localhost:8000/
  • Service discovery: /cds-services
  • CDS endpoint: /cds-services/discharge-summary
  • API docs: /docs
  • Receives CDS Hooks requests from EHR systems
  • Summarizes discharge notes using AI
  • Returns CDS cards with clinical recommendations
{
  "cards": [
    {
      "summary": "Discharge Transportation",
      "indicator": "info",
      "source": {
        "label": "HealthChain Discharge Assistant"
      },
      "detail": "ā€¢ Transport arranged for 11:00 HRs\nā€¢ Requires bariatric ambulance and 2 crew members\nā€¢ Confirmation number: TR-2024-001"
    },
    {
      "summary": "Medication Management",
      "indicator": "warning",
      "source": {
        "label": "HealthChain Discharge Assistant"
      },
      "detail": "ā€¢ Discharge medications: Apixaban 5mg, Baclofen 20mg MR\nā€¢ New anticoagulation card prepared\nā€¢ Collection by daughter scheduled"
    }
  ]
}