from PyAnnote.audio import Pipeline

def model_fn(model_dir):
# Load the model from the specified model directory
model = Pipeline.from_pretrained(
"PyAnnote/speaker-diarization-3.1",
use_auth_token="Replace-with-the-Hugging-face-auth-token")
return model

with open("model/code/requirements.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")

!tar zcvf model.tar.gz *

import sagemaker
import boto3

sess = sagemaker.Session()

sagemaker_session_bucket = None
if sagemaker_session_bucket is None and sess is not None:
    sagemaker_session_bucket = sess.default_bucket()

try:
    role = sagemaker.get_execution_role()
except ValueError:
    iam = boto3.client("iam")
    role = iam.get_role(RoleName="sagemaker_execution_role")["Role"]["Arn"]

sess = sagemaker.Session(default_bucket=sagemaker_session_bucket)

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

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

from sagemaker.huggingface.model 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 Model Class
huggingface_model = HuggingFaceModel(
    model_data=s3_location,  # path to your model and script
    role=role,  # iam role with permissions to create an Endpoint
    transformers_version="4.17",  # transformers version used
    pytorch_version="1.10",  # pytorch version used
    py_version="py38",  # python version used
)

# create async endpoint configuration
async_config = AsyncInferenceConfig(
    output_path=s3_path_join(
        "s3://", sagemaker_session_bucket, "async_inference/output"
    ),  # Where our results will be stored
    # Add nofitication SNS if needed
    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,
)

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

config = WaiterConfig(
  max_attempts=10, #  number of attempts
  delay=10#  time in seconds to wait between attempts
  )
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"]

# Common class representing application autoscaling for SageMaker 
client = boto3.client('application-autoscaling') 

# This is the format in which application autoscaling references the endpoint
resource_id='endpoint/' + <endpoint_name> + '/variant/' + <'variant1'> 

# Define and register your endpoint variant
response = client.register_scalable_target(
    ServiceNamespace='sagemaker', 
    ResourceId=resource_id,
    ScalableDimension='sagemaker:variant:DesiredInstanceCount', # The number of EC2 instances for your Amazon SageMaker model endpoint variant.
    MinCapacity=0,
    MaxCapacity=5
)

Se você deseja excluir o endpoint, use o seguinte código:

async_predictor.delete_endpoint(async_endpoint_name)

Benefícios da implantação assíncrona de endpoint

Esta solução oferece os seguintes benefícios:

• A solução pode lidar com eficiência com arquivos de áudio múltiplos ou grandes.
• Este exemplo usa uma única instância para demonstração. Se quiser usar esta solução para centenas ou milhares de vídeos e usar um endpoint assíncrono para processar em várias instâncias, você pode usar um política de escala automática, projetado para um grande número de documentos de origem. O escalonamento automático ajusta dinamicamente o número de instâncias provisionadas para um modelo em resposta a alterações na sua carga de trabalho.
• A solução otimiza recursos e reduz a carga do sistema, separando tarefas de longa duração da inferência em tempo real.

Conclusão

Nesta postagem, fornecemos uma abordagem direta para implantar o modelo de diarização de alto-falante do Hugging Face no SageMaker usando scripts Python. O uso de um endpoint assíncrono fornece um meio eficiente e escalonável de fornecer previsões de diarização como um serviço, acomodando solicitações simultâneas de maneira integrada.

Comece hoje mesmo com a diarização assíncrona de alto-falantes para seus projetos de áudio. Entre em contato nos comentários se tiver alguma dúvida sobre como colocar seu próprio endpoint de diarização assíncrona em funcionamento.

Sobre os autores

Sanjay Tiwary é um arquiteto de soluções especializado em IA/ML que passa seu tempo trabalhando com clientes estratégicos para definir requisitos de negócios, fornecer sessões L300 em torno de casos de uso específicos e projetar aplicativos e serviços de IA/ML que sejam escalonáveis, confiáveis ​​e de alto desempenho. Ele ajudou a lançar e dimensionar o serviço Amazon SageMaker baseado em IA/ML e implementou diversas provas de conceito usando os serviços de IA da Amazon. Ele também desenvolveu a plataforma de análise avançada como parte da jornada de transformação digital.

Kiran Challapalli é um desenvolvedor de negócios de tecnologia profunda no setor público da AWS. Ele tem mais de 8 anos de experiência em IA/ML e 23 anos de experiência geral em desenvolvimento de software e vendas. Kiran ajuda empresas do setor público em toda a Índia a explorar e co-criar soluções baseadas em nuvem que usam tecnologias de IA, ML e IA generativa, incluindo grandes modelos de linguagem.