Ajuste una aplicación de IA generativa para Amazon Bedrock utilizando decoradores de Amazon SageMaker Pipeline

Creación de una canalización de implementación para inteligencia artificial generativa (IA) La implementación de aplicaciones de IA a gran escala es un desafío formidable debido a las complejidades y los requisitos únicos de estos sistemas. Los modelos de IA generativa evolucionan constantemente y se lanzan nuevas versiones y actualizaciones con frecuencia. Esto hace que la gestión e implementación de estas actualizaciones en un flujo de implementación a gran escala, al mismo tiempo que se proporciona coherencia y se minimiza el tiempo de inactividad, sea una tarea importante. Las aplicaciones de IA generativa requieren la ingesta, el preprocesamiento y el formateo continuos de grandes cantidades de datos de diversas fuentes. Construir flujos de datos robustos que puedan manejar esta carga de trabajo de manera confiable y eficiente a gran escala es un desafío considerable. Monitorear el rendimiento, el sesgo y las implicaciones éticas de los modelos de IA generativa en entornos de producción es una tarea crucial.

Para lograrlo a gran escala, se necesitan inversiones significativas en recursos, experiencia y colaboración multifuncional entre múltiples personas, como científicos de datos o desarrolladores de aprendizaje automático (ML) que se centran en el desarrollo de modelos de ML e ingenieros de operaciones de aprendizaje automático (MLOps) que se centran en los aspectos únicos de los proyectos de IA/ML y ayudan a mejorar el tiempo de entrega, reducir los defectos y hacer que la ciencia de datos sea más productiva. En esta publicación, le mostramos cómo convertir el código Python que ajusta un modelo de IA generativa en La roca madre del Amazonas de archivos locales a un flujo de trabajo reutilizable utilizando Canalizaciones de Amazon SageMaker Decoradores. Puedes utilizar Canalizaciones de creación de modelos de Amazon SageMaker para colaborar entre múltiples equipos de IA/ML.

Canalizaciones de SageMaker

Puede utilizar SageMaker Pipelines para definir y organizar los distintos pasos involucrados en el ciclo de vida de ML, como el preprocesamiento de datos, el entrenamiento de modelos, la evaluación y la implementación. Esto agiliza el proceso y proporciona coherencia en las diferentes etapas del proceso. SageMaker Pipelines puede gestionar el control de versiones de modelos y el seguimiento de linaje. Realiza un seguimiento automático de los artefactos, hiperparámetros y metadatos del modelo, lo que le ayuda a reproducir y auditar las versiones del modelo.

La función de decorador SageMaker Pipelines ayuda a convertir el código ML local escrito como un programa Python en uno o más pasos de canalización. Debido a que se puede acceder a Amazon Bedrock como una API, los desarrolladores que no saben Amazon SageMaker puede implementar una aplicación de Amazon Bedrock o ajustar Amazon Bedrock escribiendo un programa Python normal.

Puede escribir su función de ML como lo haría para cualquier proyecto de ML. Después de probarla localmente o como un trabajo de capacitación, un científico de datos o un profesional que sea experto en SageMaker puede convertir la función en un paso de canalización de SageMaker agregando un @step decorador.

Descripción general de la solución

SageMaker Model Building Pipelines es una herramienta para crear pipelines de ML que aprovecha la integración directa con SageMaker. Gracias a esta integración, puede crear un pipeline para orquestación mediante una herramienta que se encarga de gran parte de la creación y administración de pasos por usted.

A medida que pasa de las fases piloto y de prueba a la implementación de modelos de IA generativos a escala, deberá aplicar prácticas de DevOps a las cargas de trabajo de ML. SageMaker Pipelines está integrado con SageMaker, por lo que no necesita interactuar con ningún otro servicio de AWS. Tampoco necesita administrar ningún recurso porque SageMaker Pipelines es un servicio completamente administrado, lo que significa que crea y administra recursos por usted. Estudio Amazon SageMaker ofrece un entorno para gestionar la experiencia de SageMaker Pipelines de extremo a extremo. La solución de esta publicación muestra cómo se puede tomar el código Python que se escribió para preprocesar, ajustar y probar un modelo de lenguaje grande (LLM) mediante las API de Amazon Bedrock y convertirlo en un pipeline de SageMaker para mejorar la eficiencia operativa de ML.

La solución tiene tres pasos principales:

1. Escriba código Python para preprocesar, entrenar y probar un LLM en Amazon Bedrock.
2. Agregar @step funciones decoradas para convertir el código Python en una canalización de SageMaker.
3. Cree y ejecute el pipeline de SageMaker.

El siguiente diagrama ilustra el flujo de trabajo de la solución.

Prerrequisitos

Si solo desea ver el código del cuaderno, puede verlo cuaderno en GitHub.

Si eres nuevo en AWS, primero debes… crear y configurar una cuenta de AWSEntonces lo harás Configurar SageMaker Studio en su cuenta de AWS. Cree un espacio JupyterLab dentro de SageMaker Studio para ejecutar la aplicación JupyterLab.

Cuando esté en el espacio JupyterLab de SageMaker Studio, complete los siguientes pasos:

1. En el menú Archivo, elija Nuevo y Terminal para abrir una nueva terminal.
2. En la terminal, ingrese el siguiente código:

   git clone https://github.com/aws/amazon-sagemaker-examples.git

3. Verás el llamador de la carpeta amazon-sagemaker-examples en el panel Explorador de archivos de SageMaker Studio.
4. Abrir la carpeta amazon-sagemaker-examples/sagemaker-pipelines/step-decorator/bedrock-examples.
5. Abre el cuaderno fine_tune_bedrock_step_decorator.ipynb.

Este cuaderno contiene todo el código para esta publicación y puedes ejecutarlo de principio a fin.

Explicación del código del cuaderno

El portátil utiliza el valor predeterminado Servicio de almacenamiento simple de Amazon (Amazon S3) depósito para el usuario. El depósito S3 predeterminado sigue el patrón de nombres s3://sagemaker-{Region}-{your-account-id}Si aún no existe, se creará automáticamente.

Utiliza el SageMaker Studio predeterminado. Gestión de identidad y acceso de AWS Rol (IAM) para el usuario. Si su rol de usuario de SageMaker Studio no tiene acceso de administrador, debe agregar los permisos necesarios al rol.

Para obtener más información, consulte lo siguiente:

Crea una sesión de SageMaker y obtiene el depósito S3 predeterminado y el rol de IAM:

sagemaker_session = sagemaker.session.Session()
region = sagemaker_session.boto_region_name

bucket_name = sagemaker_session.default_bucket()
role_arn = sagemaker.get_execution_role() 
...

Utilice Python para preprocesar, entrenar y probar un LLM en Amazon Bedrock

Para comenzar, necesitamos descargar datos y preparar un LLM en Amazon Bedrock. Para ello, utilizamos Python.

Cargar datos

Nosotros usamos el Conjunto de datos de CNN/DailyMail de Hugging Face para ajustar el modelo. El conjunto de datos CNN/DailyMail es un conjunto de datos en idioma inglés que contiene más de 300.000 artículos de noticias únicos escritos por periodistas de CNN y Daily Mail. El conjunto de datos sin procesar incluye los artículos y sus resúmenes para entrenamiento, validación y prueba. Antes de que podamos usar el conjunto de datos, debe formatearse para incluir el mensaje. Vea el siguiente código:

def add_prompt_to_data(dataset):

    datapoints = []
    
    for datapoint in dataset:
        # Add insruction prompt to each CNN article
        # and add prefix 'response:' to the article summary.
        temp_dict = {}
        temp_dict['prompt'] = instruction + datapoint['article']
        temp_dict['completion'] = 'response:\n\n' + datapoint['highlights']
        datapoints.append(temp_dict)
    return datapoints

def data_load(ds_name: str, ds_version: str) -> tuple:

    dataset = load_dataset(ds_name, ds_version)
    datapoints_train = add_prompt_to_data(dataset['train'])
    datapoints_valid = add_prompt_to_data(dataset['validation'])
    datapoints_test = add_prompt_to_data(dataset['test'])
    ...

Datos divididos

Divida el conjunto de datos en entrenamiento, validación y prueba. Para esta publicación, restringimos el tamaño de cada fila a 3000 palabras y seleccionamos 100 filas para entrenamiento, 10 para validación y 5 para prueba. Puede seguir el cuaderno en GitHub Para más detalles.

def data_split(step_load_result: tuple)  -> tuple:

    train_lines = reduce_dataset_size(step_load_result[0], 3000, 100)
    validation_lines = reduce_dataset_size(step_load_result[1], 3000, 10)
    test_lines = reduce_dataset_size(step_load_result[2], 3000, 5)
    
    ...

    return train_lines, validation_lines, test_lines

Subir datos a Amazon S3

A continuación, convertimos los datos al formato JSONL y cargamos los archivos de entrenamiento, validación y prueba en Amazon S3:

def upload_file_to_s3(bucket_name: str, file_names: tuple,
                        s3_key_names: tuple):
    import boto3
    s3_client = boto3.client('s3')
    for i in range(len(file_names)):
        s3_client.upload_file(file_names[i], bucket_name, s3_key_names[i])
    ...
    
def data_upload_to_s3(data_split_response: tuple, bucket_name: str) -> tuple:

    dataset_folder = "fine-tuning-datasets"

    if not os.path.exists(dataset_folder):
        os.makedirs(dataset_folder)

    abs_path = os.path.abspath(dataset_folder)
    train_file = write_jsonl_file(abs_path, 'train-cnn.jsonl', data_split_response[0])
    val_file = write_jsonl_file(abs_path, 'validation-cnn.jsonl', data_split_response[1])
    test_file = write_jsonl_file(abs_path, 'test-cnn.jsonl', data_split_response[2])

    file_names = train_file, val_file, test_file

    s3_keys = f'{dataset_folder}/train/train-cnn.jsonl', f'{dataset_folder}/validation/validation-cnn.jsonl', f'{dataset_folder}/test/test-cnn.jsonl'

    upload_file_to_s3(bucket_name, file_names, s3_keys)
    
    ...

Entrenar el modelo

Ahora que los datos de entrenamiento están cargados en Amazon S3, es hora de Ajustar un modelo Bedrock de Amazon utilizando el conjunto de datos CNN/DailyMail. Ajustamos los Texto Lite de Amazon Titan modelo proporcionado por Amazon Bedrock para un caso de uso de resumen. definir los hiperparámetros Para afinar y lanzar el trabajo de entrenamiento:

    hyper_parameters = {
        "epochCount": "2",
        "batchSize": "1",
        "learningRate": "0.00003",
    }
...

    training_job_response = bedrock.create_model_customization_job(
        customizationType = "FINE_TUNING",
        jobName = training_job_name,
        customModelName = custom_model_name,
        roleArn = role_arn,
        baseModelIdentifier = "amazon.titan-text-lite-v1:0:4k",
        hyperParameters = hyper_parameters,
        trainingDataConfig = training_data_config,
        validationDataConfig = validation_data_config,
        outputDataConfig = output_data_config
    )
...
    model_id = bedrock.get_custom_model(modelIdentifier=custom_model_name)['modelArn']

    print(f'Model id: {model_id}')
    return model_id

Crear rendimiento aprovisionado

Rendimiento se refiere a la cantidad y la tasa de entradas y salidas que un modelo procesa y devuelve. Puede comprar Provisioned Throughput para aprovisionar recursos dedicados en lugar de rendimiento a pedido, que podría tener fluctuaciones de rendimiento. Para modelos personalizados, debe comprar Rendimiento aprovisionado para poder usarlo. Ver Rendimiento aprovisionado para Amazon Bedrock Para más información.

def create_prov_thruput(model_id: str, provisioned_model_name: str) -> str:

    bedrock = boto3.client(service_name="bedrock")

    provisioned_model_id = bedrock.create_provisioned_model_throughput(
                modelUnits=1,
                provisionedModelName=provisioned_model_name,
                modelId=model_id
                )['provisionedModelArn']
    ...

    return provisioned_model_id

Probar el modelo

Ahora es el momento de invocar y probar el modelo. Usamos el indicador de tiempo de ejecución de Amazon Bedrock del conjunto de datos de prueba junto con el ID del rendimiento aprovisionado que se configuró en el paso anterior y parámetros de inferencia como maxTokenCount, stopSequence, temperature, y top:

...
def test_model (provisioned_model_id: str) -> tuple:

    s3.download_file(s3_bucket, s3_key, 'test-cnn.jsonl')

...
    body = json.dumps(
        {
            "inputText": prompt,
            "textGenerationConfig": {
                "maxTokenCount": 2048,
                "stopSequences": ['User:'],
                "temperature": 0,
                "topP": 0.9
            }
        }
    )

    accept="application/json"
    contentType="application/json"

    bedrock_runtime = boto3.client(service_name="bedrock-runtime")

    fine_tuned_response = bedrock_runtime.invoke_model(body=body,
                                        modelId=provisioned_model_id,
                                        accept=accept,
                                        contentType=contentType)

    fine_tuned_response_body = json.loads(fine_tuned_response.get('body').read())
    summary = fine_tuned_response_body["results"][0]["outputText"]

    return prompt, summary

Decore funciones con @step que convierte funciones de Python en pasos de canalización de SageMaker

El @step El decorador es una función que convierte su código ML local en uno o más pasos de canalización. Puede escribir su función ML como lo haría para cualquier proyecto ML y luego crear una canalización convirtiendo funciones Python en pasos de canalización utilizando el @step decorador, creando dependencias entre esas funciones para crear un gráfico de canalización o un gráfico acíclico dirigido (DAG), y pasando los nodos de hoja de ese gráfico como una lista de pasos a la canalización. Para crear un paso usando el @step decorador, anota la función con @stepCuando se invoca esta función, recibe la Retorno retrasado Salida del paso anterior del pipeline como entrada. Una instancia contiene la información sobre todos los pasos anteriores definidos en la función que forman el DAG del pipeline de SageMaker.

En el cuaderno ya agregamos el @step Decorador al comienzo de cada definición de función en la celda donde se definió la función, como se muestra en el código siguiente. El código de la función provendrá del programa de Python de ajuste fino que estamos tratando de convertir aquí en una canalización de SageMaker.

@step(
name="data-load-step",
keep_alive_period_in_seconds = 300,
)
def data_load(ds_name: str, ds_version: str) -> tuple:
...
return datapoints_train, datapoints_valid, datapoints_test

@step(
name = "data-split-step",
keep_alive_period_in_seconds = 300,
)
def data_split(step_load_result: tuple)  -> tuple:
...
return train_lines, validation_lines, test_lines

@step(
name = "data-upload-to-s3-step",
keep_alive_period_in_seconds=300,
)
def data_upload_to_s3(data_split_response: tuple, bucket_name: str) -> tuple:
...
return f's3://{bucket_name}/{s3_keys[0]}', f's3://{bucket_name}/{s3_keys[1]}', f's3://{bucket_name}/{s3_keys[2]}'

@step(
name = "model-training-step",
keep_alive_period_in_seconds=300,
)
def train(custom_model_name: str,
training_job_name: str,
step_data_upload_to_s3_result: tuple) -> str:
...
return model_id

@step(
name = "create-provisioned-throughput-step",
keep_alive_period_in_seconds=300,
)
def create_prov_thruput(model_id: str, provisioned_model_name: str) -> str:
...
return provisioned_model_id

@step(
name = "model-testing-step",
keep_alive_period_in_seconds=300,
)
def test_model (provisioned_model_id: str) -> tuple:
...
return prompt, summary

Crear y ejecutar la canalización de SageMaker

Para unirlo todo, conectamos la tubería definida @step funciones en una secuencia de varios pasos. Luego, enviamos y ejecutamos la secuencia:

pipeline_name = "bedrock-fine-tune-pipeline"
...
data_load_response = data_load(param1, param2)

data_split_response = data_split(data_load_response)

data_upload_to_s3_response = data_upload_to_s3(data_split_response, bucket_name)

train_response = train(custom_model_name, training_job_name, data_upload_to_s3_response)

create_prov_thruput_response = create_prov_thruput(train_response, provisioned_model_name)

test_model_response = test_model(create_prov_thruput_response)

pipeline = Pipeline(
    name=pipeline_name,
    steps=[test_model_response],
    parameters=[param1, param2]
    )
...
execution = pipeline.start()

Una vez ejecutada la canalización, puede enumerar los pasos de la canalización para recuperar el conjunto de datos completo de resultados:

execution.list_steps()

[{'StepName': 'model-testing-step',
  ...
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-model-testing-step-rnUvvmGxgn'}},
  ... 
 {'StepName': 'create-provisioned-throughput-step',
  ...  
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-create-provisioned-t-vmNdXHTaH3'}},
  ...  
 {'StepName': 'model-training-step',
  ...
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-model-training-step-t3vmuAmWf6'}},
  ... 
 {'StepName': 'data-upload-to-s3-step',
  ... 
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-data-upload-to-s3-st-cDKe6fJYtf'}},
  ...  
 {'StepName': 'data-split-step',
  ...
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-data-split-step-ciIP7t0tTq'}},
  ...
 {'StepName': 'data-load-step',
  ... 
  'StepStatus': 'Succeeded',
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:xxxxxxxx:training-job/pipelines-a6lnarybitw1-data-load-step-swEWNYi5mK'}},

Puede realizar un seguimiento del linaje de una canalización de SageMaker ML en SageMaker Studio. El seguimiento del linaje en SageMaker Studio se centra en un DAG. El DAG representa los pasos de una canalización. Desde el DAG, puede realizar un seguimiento del linaje de un paso a otro. El siguiente diagrama muestra los pasos de la canalización de ajuste fino de Amazon Bedrock. Para obtener más información, consulte Ver la ejecución de una canalización.

Al elegir un paso en el menú desplegable Seleccionar paso, puede centrarse en una parte específica del gráfico. Puede ver registros detallados de cada paso del pipeline en Registros de Amazon CloudWatch.

Limpiar

Para limpiar y evitar incurrir en cargos, siga las instrucciones de limpieza detalladas en el Repositorio de GitHub Para eliminar lo siguiente:

• El rendimiento aprovisionado de Amazon Bedrock
• El modelo del cliente
• El oleoducto Sagemaker
• El objeto de Amazon S3 que almacena el conjunto de datos ajustado

Conclusión

MLOps se centra en optimizar, automatizar y supervisar los modelos de ML a lo largo de su ciclo de vida. La creación de un flujo de trabajo de MLOps sólido exige una colaboración interdisciplinaria. Los científicos de datos, los ingenieros de ML, el personal de TI y los equipos de DevOps deben trabajar juntos para poner en funcionamiento los modelos desde la investigación hasta la implementación y el mantenimiento. SageMaker Pipelines le permite crear y gestionar flujos de trabajo de ML a la vez que ofrece capacidades de almacenamiento y reutilización para los pasos del flujo de trabajo.

En esta publicación, le mostramos un ejemplo que utiliza los decoradores de pasos de SageMaker para convertir un programa Python para crear un modelo de Amazon Bedrock personalizado en una canalización de SageMaker. Con SageMaker Pipelines, obtiene los beneficios de un flujo de trabajo automatizado que se puede configurar para ejecutarse según un cronograma en función de los requisitos para volver a entrenar el modelo. También puede usar SageMaker Pipelines para agregar funciones útiles, como el seguimiento de linaje y la capacidad de administrar y visualizar todo su flujo de trabajo desde el entorno de SageMaker Studio.

AWS ofrece soluciones de ML administradas, como Amazon Bedrock y SageMaker, para ayudarlo a implementar y servir modelos básicos listos para usar existentes o crear y ejecutar sus propios modelos personalizados.

Consulte los siguientes recursos para obtener más información sobre los temas tratados en esta publicación:

Acerca de los autores

Neel Sendas Neel es gerente técnico principal de cuentas en Amazon Web Services. Trabaja con clientes empresariales para diseñar, implementar y escalar aplicaciones en la nube para lograr sus objetivos comerciales. Trabajó en varios casos de uso de ML, que van desde la detección de anomalías hasta la calidad predictiva de productos para la optimización de la fabricación y la logística. Cuando no está ayudando a los clientes, se dedica al golf y al baile de salsa.

Ashish Rawat es un arquitecto de soluciones sénior especialista en IA/ML en Amazon Web Services, con sede en Atlanta, Georgia. Ashish tiene una amplia experiencia en arquitectura de TI empresarial y desarrollo de software, incluidos IA/ML e IA generativa. Es fundamental para orientar a los clientes para resolver desafíos comerciales complejos y crear una ventaja competitiva utilizando los servicios de IA/ML de AWS.