Crea un flusso di lavoro di inferenza AI per l’imaging medico con MONAI Deploy su AWS
Crea un flusso di lavoro di inferenza AI per l'imaging medico utilizzando MONAI Deploy su AWS
Questo post è scritto a quattro mani insieme a Ming (Melvin) Qin, David Bericat e Brad Genereaux di NVIDIA.
Gli ricercatori e sviluppatori di intelligenza artificiale per l’imaging medico hanno bisogno di un framework scalabile per costruire, distribuire e integrare le loro applicazioni di intelligenza artificiale. AWS e NVIDIA si sono uniti per rendere questa visione una realtà. AWS, NVIDIA e altri partner costruiscono applicazioni e soluzioni per rendere l’assistenza sanitaria più accessibile, conveniente ed efficiente accelerando la connettività cloud dell’imaging aziendale. MONAI Deploy è uno dei moduli chiave all’interno di MONAI (Medical Open Network for Artificial Intelligence) sviluppato da un consorzio di leader accademici e industriali, tra cui NVIDIA. AWS HealthImaging (AHI) è un archivio di immagini mediche conforme a HIPAA, altamente scalabile, performante ed economico. Abbiamo sviluppato un connettore MONAI Deploy per AHI per integrare applicazioni di intelligenza artificiale per l’imaging medico con latenze di recupero delle immagini nell’ordine dei subsecondi a scalabilità supportata da API native del cloud. I modelli e le applicazioni MONAI AI possono essere ospitati su Amazon SageMaker, un servizio completamente gestito per distribuire modelli di apprendimento automatico (ML) su larga scala. SageMaker si occupa di configurare e gestire le istanze per le inferenze e fornisce metriche e log integrati per i punti finali che è possibile utilizzare per monitorare e ricevere avvisi. Offre anche una varietà di istanze GPU NVIDIA per l’inferenza dell’ML, nonché diverse opzioni di distribuzione dei modelli con scalabilità automatica, incluse l’inferenza in tempo reale, l’inferenza serverless, l’inferenza asincrona e la trasformazione batch.
In questo post, mostriamo come distribuire un pacchetto di applicazioni MONAI (MAP) con il connettore su AWS HealthImaging, utilizzando un endpoint multiservizio SageMaker per inferenza in tempo reale e inferenza asincrona. Queste due opzioni coprono la maggior parte dei casi d’uso di pipeline di inferenza di imaging medico quasi in tempo reale.
Panoramica della soluzione
Il diagramma seguente illustra l’architettura della soluzione.
- L’intelligenza artificiale generativa nella Ricerca si espande in oltre 120 nuovi paesi e territori.
- Promuovi la sostenibilità con Amazon CodeWhisperer
- Dr. Aengus Tran, CEO di Harrison.ai, sull’utilizzo dell’intelligenza artificiale come un correttore di bozze per i controlli di salute.
Prerequisiti
Completa i seguenti passaggi preliminari:
- Usa un account AWS in una delle seguenti regioni, dove è disponibile AWS HealthImaging: Virginia del Nord (
us-east-1), Oregon (us-west-2), Irlanda (eu-west-1) e Sydney (ap-southeast-2). - Crea un dominio di Amazon SageMaker Studio e un profilo utente con AWS Identity and Access Management (IAM) con i permessi per accedere a AWS HealthImaging.
- Abilita l’estensione JupyterLab v3 e installa l’estensione Imjoy-jupyter se desideri visualizzare immagini mediche interattivamente nel notebook SageMaker utilizzando itkwidgets.
Connettore MAP per AWS HealthImaging
AWS HealthImaging importa file DICOM P10 e li converte in ImageSets, che rappresentano in modo ottimizzato una serie DICOM. AHI fornisce l’accesso all’API ai metadati degli ImageSet e agli ImageFrames. I metadati contengono tutti gli attributi DICOM in un documento JSON. Gli ImageFrames vengono restituiti in formato lossless (HTJ2K JPEG2000 ad alta velocità), che può essere decodificato estremamente velocemente. Gli ImageSet possono essere recuperati utilizzando l’Interfaccia a linea di comando AWS (AWS CLI) o gli SDK di AWS.
MONAI è un framework di intelligenza artificiale per l’elaborazione delle immagini mediche che trasforma le scoperte della ricerca e le applicazioni di intelligenza artificiale in impatto clinico. MONAI Deploy è il flusso di lavoro di elaborazione che permette di eseguire l’intero processo, inclusi l’incapsulamento, il test, il rilascio e l’esecuzione di applicazioni di intelligenza artificiale per l’elaborazione delle immagini mediche in produzione clinica. Comprende il MONAI Deploy App SDK, MONAI Deploy Express, Workflow Manager e Informatics Gateway. Il MONAI Deploy App SDK fornisce algoritmi pronti all’uso e un framework per accelerare la creazione di applicazioni di intelligenza artificiale per l’elaborazione delle immagini mediche, nonché strumenti di utilità per incapsulare l’applicazione in un contenitore MAP. Le funzionalità integrate basate su standard dell’SDK dell’app consentono al MAP di integrarsi facilmente nelle reti IT sanitarie, il che richiede l’utilizzo degli standard come DICOM, HL7 e FHIR, e attraverso ambienti di data center e cloud. I MAP possono utilizzare operatori predefiniti e personalizzati per il caricamento delle immagini DICOM, la selezione delle serie, l’inferenza del modello e la post-elaborazione.
Abbiamo sviluppato un modulo Python utilizzando l’SDK di Python di AWS HealthImaging Boto3. Puoi installarlo tramite pip e utilizzare la funzione ausiliaria per recuperare le istanze Service-Object Pair (SOP) dei file DICOM come segue:
!pip install -q AHItoDICOMInterfacefrom AHItoDICOMInterface.AHItoDICOM import AHItoDICOMhelper = AHItoDICOM()instances = helper.DICOMizeImageSet(datastore_id=datastoreId , image_set_id=next(iter(imageSetIds)))Le istanze SOP di output possono essere visualizzate utilizzando il visualizzatore di immagini mediche interattivo 3D itkwidgets nel seguente notebook. La classe AHItoDICOM sfrutta processi multipli per recuperare i frame dei pixel da AWS HealthImaging in parallelo e decodificare i blobs binari HTJ2K utilizzando la libreria Python OpenJPEG. Gli ImageSetIds provengono dai file di output di un determinato lavoro di importazione di AWS HealthImaging. Dato il DatastoreId e l’ID del lavoro di importazione, è possibile recuperare l’ImageSetId, che corrisponde all’UID dell’istanza della serie DICOM, come segue:
imageSetIds = {}try: response = s3.head_object(Bucket=OutputBucketName, Key=f"output/{res_createstore['datastoreId']}-DicomImport-{res_startimportjob['jobId']}/job-output-manifest.json") if response['ResponseMetadata']['HTTPStatusCode'] == 200: data = s3.get_object(Bucket=OutputBucketName, Key=f"output/{res_createstore['datastoreId']}-DicomImport-{res_startimportjob['jobId']}/SUCCESS/success.ndjson") contents = data['Body'].read().decode("utf-8") for l in contents.splitlines(): isid = json.loads(l)['importResponse']['imageSetId'] if isid in imageSetIds: imageSetIds[isid]+=1 else: imageSetIds[isid]=1except ClientError: passCon ImageSetId, puoi recuperare separatamente i metadati dell’intestazione DICOM e i pixel dell’immagine utilizzando le funzioni native dell’API HealthImaging di AWS. L’esportatore DICOM aggrega gli intestazioni DICOM e i pixel dell’immagine nell’insieme di dati Pydicom, che può essere elaborato dall’operatore di caricamento dati DICOM della MAP DICOM. Utilizzando la funzione DICOMizeImageSet(), abbiamo creato un connettore per caricare i dati delle immagini da AWS HealthImaging, basato sull’operatore di caricamento dati DICOM MAP:
class AHIDataLoaderOperator(Operator): def __init__(self, ahi_client, must_load: bool = True, *args, **kwargs): self.ahi_client = ahi_client … def _load_data(self, input_obj: string): study_dict = {} series_dict = {} sop_instances = self.ahi_client.DICOMizeImageSet(input_obj['datastoreId'], input_obj['imageSetId'])Nel codice precedente, ahi_client è un’istanza della classe esportatore DICOM AHItoDICOM, con funzioni di recupero dei dati illustrate. Abbiamo incluso questo nuovo operatore di caricamento dati in un’applicazione AI di segmentazione 3D della milza creata da MONAI Deploy App SDK. Puoi prima esplorare come creare ed eseguire questa applicazione su un’istanza locale di notebook, e quindi distribuire questa applicazione MAP negli endpoint di inferenza gestiti da SageMaker.
Inferenza asincrona di SageMaker
Un endpoint di inferenza asincrona di SageMaker viene utilizzato per richieste con dimensioni di payload elevate (fino a 1 GB), tempi di elaborazione lunghi (fino a 15 minuti) ed esigenze di latenza quasi in tempo reale. Quando non ci sono richieste da elaborare, questa opzione di distribuzione può ridurre il numero di istanze a zero per risparmiare costi, il che è ideale per carichi di lavoro di inferenza di imaging medico basati su ML. Segui i passaggi nel notebook di esempio per creare e invocare l’endpoint di inferenza asincrona di SageMaker. Per creare un endpoint di inferenza asincrona, dovrai prima creare un modello SageMaker e una configurazione dell’endpoint. Per creare un modello SageMaker, dovrai caricare un pacchetto model.tar.gz con una struttura delle directory definita in un contenitore Docker. Il pacchetto model.tar.gz include un file model.ts di segmentazione della milza preaddestrato e un file inference.py personalizzato. Abbiamo utilizzato un contenitore predefinito con versioni del framework Python 3.8 e PyTorch 1.12.1 per caricare il modello ed eseguire le previsioni.
Nel file di inferenza personalizzato inference.py, istanziamo una classe helper AHItoDICOM dalla libreria AHItoDICOMInterface e la utilizziamo per creare un’istanza MAP nella funzione model_fn(), e eseguiamo l’applicazione MAP su ogni richiesta di inferenza nella funzione predict_fn():
from app import AISpleenSegAppfrom AHItoDICOMInterface.AHItoDICOM import AHItoDICOMhelper = AHItoDICOM()def model_fn(model_dir, context): … monai_app_instance = AISpleenSegApp(helper, do_run=False,path="/home/model-server")def predict_fn(input_data, model): with open('/home/model-server/inputImageSets.json', 'w') as f: f.write(json.dumps(input_data)) output_folder = "/home/model-server/output" if not os.path.exists(output_folder): os.makedirs(output_folder) model.run(input='/home/model-server/inputImageSets.json', output=output_folder, workdir='/home/model-server', model='/opt/ml/model/model.ts')Per invocare il punto finale asincrono, è necessario caricare i dati di input della richiesta su Amazon Simple Storage Service (Amazon S3), che è un file JSON che specifica l’ID del datastore AWS HealthImaging e l’ID dell’ImageSet su cui eseguire l’inferenza:
sess = sagemaker.Session()InputLocation = sess.upload_data('inputImageSets.json', bucket=sess.default_bucket(), key_prefix=prefix, extra_args={"ContentType": "application/json"})response = runtime_sm_client.invoke_endpoint_async(EndpointName=endpoint_name, InputLocation=InputLocation, ContentType="application/json", Accept="application/json")output_location = response["OutputLocation"]Il risultato può essere trovato anche su Amazon S3.
Inferenza in tempo reale multi-modello di SageMaker
I punti finali di inferenza in tempo reale di SageMaker soddisfano i requisiti interattivi a bassa latenza. Questa opzione può ospitare più modelli in un unico contenitore dietro un singolo punto finale, che è una soluzione scalabile ed economica per implementare diversi modelli di machine learning. Un punto finale multi-modello di SageMaker utilizza NVIDIA Triton Inference Server con GPU per eseguire inferenze su più modelli di deep learning in parallelo.
In questa sezione, ti guideremo attraverso i passaggi necessari per creare e invocare un punto finale multi-modello adattando il tuo contenitore di inferenza personalizzato nel seguente notebook di esempio. Diversi modelli possono essere serviti in un unico contenitore condiviso sulla stessa flotta di risorse. I punti finali multi-modello riducono i costi di implementazione e scalano l’inferenza dei modelli in base ai pattern di traffico verso il punto finale. Abbiamo utilizzato strumenti per sviluppatori AWS, tra cui Amazon CodeCommit, Amazon CodeBuild e Amazon CodePipeline, per creare il contenitore personalizzato per l’inferenza dei modelli SageMaker. Abbiamo preparato un model_handler.py per utilizzare il tuo stesso contenitore anziché il file inference.py nell’esempio precedente, e abbiamo implementato le funzioni initialize(), preprocess() e inference():
from app import AISpleenSegAppfrom AHItoDICOMInterface.AHItoDICOM import AHItoDICOMclass ModelHandler(object): def __init__(self): self.initialized = False self.shapes = None def initialize(self, context): self.initialized = True properties = context.system_properties model_dir = properties.get("model_dir") gpu_id = properties.get("gpu_id") helper = AHItoDICOM() self.monai_app_instance = AISpleenSegApp(helper, do_run=False, path="/home/model-server/") def preprocess(self, request): inputStr = request[0].get("body").decode('UTF8') datastoreId = json.loads(inputStr)['inputs'][0]['datastoreId'] imageSetId = json.loads(inputStr)['inputs'][0]['imageSetId'] with open('/tmp/inputImageSets.json', 'w') as f: f.write(json.dumps({"datastoreId": datastoreId, "imageSetId": imageSetId})) return '/tmp/inputImageSets.json' def inference(self, model_input): self.monai_app_instance.run(input=model_input, output="/home/model-server/output/", workdir="/home/model-server/", model=os.environ["model_dir"]+"/model.ts")Dopo che il contenitore è stato creato e caricato su Amazon Elastic Container Registry (Amazon ECR), è possibile creare un modello SageMaker con esso, insieme a diversi pacchetti di modelli (file tar.gz) in un percorso specifico di Amazon S3:
model_name = "DEMO-MONAIDeployModel" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())model_url = "s3://{}/{}/".format(bucket, prefix)container = "{}.dkr.ecr.{}.amazonaws.com/{}:dev".format( account_id, region, prefix )container = {"Image": container, "ModelDataUrl": model_url, "Mode": "MultiModel"}create_model_response = sm_client.create_model(ModelName=model_name, ExecutionRoleArn=role, PrimaryContainer=container)È importante notare che il model_url qui specifica solo il percorso di una cartella di file tar.gz e specificare quale pacchetto di modelli utilizzare per l’elaborazione quando si invoca il punto di estremità, come mostrato nel codice seguente:
Payload = {"inputs": [ {"datastoreId": datastoreId, "imageSetId": next(iter(imageSetIds))} ]}response = runtime_sm_client.invoke_endpoint(EndpointName=endpoint_name, ContentType="application/json", Accept="application/json", TargetModel="model.tar.gz", Body=json.dumps(Payload))Possiamo aggiungere più modelli al punto di estremità di elaborazione multi-modello esistente senza dover aggiornare il punto di estremità o crearne uno nuovo.
Pulizia
Non dimenticare di completare il passaggio Elimina le risorse di hosting nei quaderni lab-3 e lab-4 per eliminare i punti di estremità di inferenza di SageMaker. È anche consigliabile ridurre al minimo l’istanza del blocco note di SageMaker per risparmiare sui costi. Infine, puoi chiamare la funzione API di AWS HealthImaging o utilizzare la console di AWS HealthImaging per eliminare i set di immagini e i data store creati in precedenza:
for s in imageSetIds.keys(): medicalimaging.deleteImageSet(datastoreId, s)medicalimaging.deleteDatastore(datastoreId)Conclusione
In questo post, ti abbiamo mostrato come creare un connettore MAP per AWS HealthImaging, che può essere riutilizzato nelle applicazioni costruite con il MONAI Deploy App SDK, per integrarsi e accelerare il recupero dei dati di immagini da un archivio DICOM nativo basato su cloud per i carichi di lavoro di intelligenza artificiale per l’immagine medica. Il MONAI Deploy SDK può essere utilizzato per supportare le operazioni ospedaliere. Abbiamo anche dimostrato due opzioni di hosting per distribuire le applicazioni di intelligenza artificiale MAP su SageMaker su larga scala.
Consulta i quaderni di esempio nel repository GitHub per saperne di più su come distribuire applicazioni MONAI su SageMaker con immagini mediche archiviate in AWS HealthImaging. Per sapere cosa AWS può fare per te, contatta un rappresentante AWS.
Per ulteriori risorse, consulta quanto segue:
