Deploy the Hugging Face (PyAnnote) speaker diarization mannequin to Amazon SageMaker as an asynchronous endpoint

from PyAnnote.audio import Pipeline

def model_fn(model_dir):
# Load the mannequin from the desired mannequin listing
mannequin = Pipeline.from_pretrained(
"PyAnnote/speaker-diarization-3.1",
use_auth_token="Change-with-the-Hugging-face-auth-token")
return mannequin

with open("mannequin/code/necessities.txt", "w") as f:
    f.write("transformers==4.25.1n")
    f.write("boto3n")
    f.write("PyAnnote.audion")
    f.write("soundfilen")
    f.write("librosan")
    f.write("onnxruntimen")
    f.write("wgetn")
    f.write("pandas")

import sagemaker
import boto3

sess = sagemaker.Session()

sagemaker_session_bucket = None
if sagemaker_session_bucket is None and sess shouldn't be None:
    sagemaker_session_bucket = sess.default_bucket()

attempt:
    position = sagemaker.get_execution_role()
besides ValueError:
    iam = boto3.shopper("iam")
    position = iam.get_role(RoleName="sagemaker_execution_role")["Role"]["Arn"]

sess = sagemaker.Session(default_bucket=sagemaker_session_bucket)

print(f"sagemaker position arn: {position}")
print(f"sagemaker bucket: {sess.default_bucket()}")
print(f"sagemaker session area: {sess.boto_region_name}")

s3_location = f"s3://{sagemaker_session_bucket}/whisper/mannequin/mannequin.tar.gz"
!aws s3 cp mannequin.tar.gz $s3_location

from sagemaker.huggingface.mannequin import HuggingFaceModel
from sagemaker.async_inference.async_inference_config import AsyncInferenceConfig
from sagemaker.s3 import s3_path_join
from sagemaker.utils import name_from_base

async_endpoint_name = name_from_base("custom-asyc")

# create Hugging Face Mannequin Class
huggingface_model = HuggingFaceModel(
    model_data=s3_location,  # path to your mannequin and script
    position=position,  # iam position with permissions to create an Endpoint
    transformers_version="4.17",  # transformers model used
    pytorch_version="1.10",  # pytorch model used
    py_version="py38",  # python model used
)

# create async endpoint configuration
async_config = AsyncInferenceConfig(
    output_path=s3_path_join(
        "s3://", sagemaker_session_bucket, "async_inference/output"
    ),  # The place our outcomes shall be saved
    # Add nofitication SNS if wanted
    notification_config={
        # "SuccessTopic": "PUT YOUR SUCCESS SNS TOPIC ARN",
        # "ErrorTopic": "PUT YOUR ERROR SNS TOPIC ARN",
    },  #  Notification configuration
)

env = {"MODEL_SERVER_WORKERS": "2"}

# deploy the endpoint endpoint
async_predictor = huggingface_model.deploy(
    initial_instance_count=1,
    instance_type="ml.xx",
    async_inference_config=async_config,
    endpoint_name=async_endpoint_name,
    env=env,
)

# Change with a path to audio object in S3
from sagemaker.async_inference import WaiterConfig
res = async_predictor.predict_async(knowledge=knowledge)
print(f"Response output path: {res.output_path}")
print("Begin Polling to get response:")

config = WaiterConfig(
  max_attempts=10, #  variety of makes an attempt
  delay=10#  time in seconds to attend between makes an attempt
  )
res.get_result(config)
#import waiterconfig

[0.9762308998, 8.9049235993, "SPEAKER_01"]

[9.533106961, 12.1646859083, "SPEAKER_01"]

[13.1324278438, 13.9303904924, "SPEAKER_00"]

[14.3548387097, 26.1884550085, "SPEAKER_00"]

[27.2410865874, 28.2258064516, "SPEAKER_01"]

[28.3446519525, 31.298811545, "SPEAKER_01"]

# Frequent class representing software autoscaling for SageMaker 
shopper = boto3.shopper('application-autoscaling') 

# That is the format wherein software autoscaling references the endpoint
resource_id='endpoint/' + <endpoint_name> + '/variant/' + <'variant1'> 

# Outline and register your endpoint variant
response = shopper.register_scalable_target(
    ServiceNamespace="sagemaker", 
    ResourceId=resource_id,
    ScalableDimension='sagemaker:variant:DesiredInstanceCount', # The variety of EC2 situations to your Amazon SageMaker mannequin endpoint variant.
    MinCapacity=0,
    MaxCapacity=5
)

If you wish to delete an endpoint, use the next code:

async_predictor.delete_endpoint(async_endpoint_name)

Benefits of implementing asynchronous endpoints

This answer has the next benefits:

• This answer can effectively course of a number of or giant audio recordsdata.
• This instance makes use of a single occasion for demonstration functions. When you use this answer for lots of or hundreds of movies, processed throughout a number of situations utilizing asynchronous endpoints, you should use autoscaling insurance policies designed for big numbers of supply paperwork. Autoscaling dynamically adjusts the variety of situations provisioned to your mannequin as your workload modifications.
• This answer optimizes assets and reduces system load by separating long-running duties from real-time inference.

conclusion

On this publish, I offered a easy method to deploying a Hugging Face speaker diarization mannequin to SageMaker utilizing a Python script. Asynchronous endpoints present an environment friendly and scalable means to supply diarization prediction as a service and seamlessly accommodate concurrent requests.

Get began with asynchronous speaker diarization to your audio initiatives at the moment. You probably have any questions on getting your personal asynchronous diarization endpoint up and operating, tell us within the feedback.

In regards to the writer

sanjay tiwary AI/ML Specialist Options Architects work with strategic prospects to outline enterprise necessities, ship L300 classes on particular use instances, and ship scalable, dependable, and performant AI/ML options. I spend my time designing ML functions and companies. He helped launch and scale his Amazon SageMaker service powered by AI/ML and carried out a number of proofs of idea utilizing Amazon AI companies. He additionally developed a complicated analytics platform as a part of the digital transformation.

Kiran Charapalli is a deep know-how enterprise developer within the AWS public sector. He has over 8 years of expertise in his AI/ML and 23 years of expertise in software program improvement and gross sales normally. Kiran helps public sector enterprises throughout India discover and co-create cloud-based options that use generative AI applied sciences, together with AI, ML, and large-scale language fashions.