Ejemplos de código para la personalización de modelos

En primer lugar, crea un archivo de texto llamado. FineTuningData.json Copia el código JSON que aparece a continuación en el archivo de texto, sustituyendo ${training-bucket} y por ${output-bucket} los nombres de los buckets de S3.

{
    "trainingDataConfig": {
        "s3Uri": "s3://${training-bucket}/train.jsonl"
    },
    "outputDataConfig": {
        "s3Uri": "s3://${output-bucket}"
    }
}

Para enviar un trabajo de personalización de modelos, navegue hasta la carpeta que contiene FineTuningData.json una terminal y ejecute el siguiente comando en la línea de comandos, ${your-customization-role-arn} sustituyéndolo por el rol de personalización del modelo que haya configurado.

aws bedrock create-model-customization-job \
    --customization-type FINE_TUNING \
    --base-model-identifier arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-text-express-v1 \
    --role-arn ${your-customization-role-arn} \
    --job-name MyFineTuningJob \
    --custom-model-name MyCustomModel \
    --hyper-parameters epochCount=1,batchSize=1,learningRate=.0001,learningRateWarmupSteps=0 \
    --cli-input-json file://FineTuningData.json
                

La respuesta devuelve un jobArn. El trabajo puede tardar unos minutos en completarse. Puede comprobar su estado con el siguiente comando.

aws bedrock get-model-customization-job \
    --job-identifier "jobArn"

Cuando el status sea COMPLETE, podrá ver trainingMetrics en la respuesta. Para descargar los artefactos a la carpeta actual, ejecute el siguiente comando y aet.et-bucket sustitúyalos por el nombre del bucket de salida y jobId por el identificador del trabajo de personalización (la secuencia que sigue a la última barra deljobArn).

aws s3 cp s3://${output-bucket}/model-customization-job-jobId . --recursive

Adquiera un rendimiento aprovisionado sin compromiso para el modelo personalizado con el siguiente comando.

aws bedrock create-provisioned-model-throughput \
    --model-id MyCustomModel \
    --provisioned-model-name MyProvisionedCustomModel \
    --model-units 1

La respuesta devuelve un provisionedModelArn. Debe esperar unos minutos a que se cree el rendimiento aprovisionado. Para comprobar el estado, proporcione el nombre o el ARN del modelo aprovisionado como provisioned-model-id en el siguiente comando.

aws bedrock get-provisioned-model-throughput \
    --provisioned-model-id ${provisioned-model-arn}

Si el status es InService, puede ejecutar una inferencia con su modelo personalizado con el siguiente comando. Debe proporcionar el ARN del modelo aprovisionado como model-id. El resultado se graba en un archivo con el nombre output.txt de la carpeta actual.

aws bedrock-runtime invoke-model \
    --model-id ${provisioned-model-arn} \
    --body '{"inputText": "What is AWS?", "textGenerationConfig": {"temperature": 0.5}}' \
    --cli-binary-format raw-in-base64-out \
    output.txt

Ejecute el siguiente fragmento de código para enviar un trabajo de afinación. ${your-customization-role-arn}Sustitúyalo por el ARN del depósito MyCustomizationRole que configuró y ${training-bucket} sustituyó y por los nombres de sus ${output-bucket} buckets de S3.

import boto3

bedrock = boto3.client(service_name='bedrock')
    
# Set parameters
customizationType = "FINE_TUNING"
baseModelIdentifier = "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-text-express-v1"
roleArn = "${your-customization-role-arn}"
jobName = "MyFineTuningJob"
customModelName = "MyCustomModel"
hyperParameters = {
        "epochCount": "1",
        "batchSize": "1",
        "learningRate": ".0001",
        "learningRateWarmupSteps": "0"
    }
trainingDataConfig = {"s3Uri": "s3://${training-bucket}/myInputData/train.jsonl"}
outputDataConfig = {"s3Uri": "s3://${output-bucket}/myOutputData"}

# Create job
response_ft = bedrock.create_model_customization_job(
    jobName=jobName, 
    customModelName=customModelName,
    roleArn=roleArn,
    baseModelIdentifier=baseModelIdentifier,
    hyperParameters=hyperParameters,
    trainingDataConfig=trainingDataConfig,
    outputDataConfig=outputDataConfig
)

jobArn = response_ft.get('jobArn')

La respuesta devuelve un jobArn. El trabajo puede tardar unos minutos en completarse. Puede comprobar su estado con el siguiente comando.

bedrock.get_model_customization_job(jobIdentifier=jobArn).get('status')

Cuando el status esCOMPLETE, puedes verlo trainingMetrics en la GetModelCustomizationJobrespuesta. También puede seguir los pasos que se indican en Descarga de objetos para descargar las métricas.

Adquiera un rendimiento aprovisionado sin compromiso para el modelo personalizado con el siguiente comando.

response_pt = bedrock.create_provisioned_model_throughput(
    modelId="MyCustomModel",
    provisionedModelName="MyProvisionedCustomModel",
    modelUnits="1"
)

provisionedModelArn = response_pt.get('provisionedModelArn')

La respuesta devuelve un provisionedModelArn. Debe esperar unos minutos a que se cree el rendimiento aprovisionado. Para comprobar el estado, proporcione el nombre o el ARN del modelo aprovisionado como provisionedModelId en el siguiente comando.

bedrock.get_provisioned_model_throughput(provisionedModelId=provisionedModelArn)

Si el status es InService, puede ejecutar una inferencia con su modelo personalizado con el siguiente comando. Debe proporcionar el ARN del modelo aprovisionado como modelId.

import json
import logging
import boto3

from botocore.exceptions import ClientError


class ImageError(Exception):
    "Custom exception for errors returned by the model"

    def __init__(self, message):
        self.message = message


logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


def generate_text(model_id, body):
    """
    Generate text using your provisioned custom model.
    Args:
        model_id (str): The model ID to use.
        body (str) : The request body to use.
    Returns:
        response (json): The response from the model.
    """

    logger.info(
        "Generating text with your provisioned custom model %s", model_id)

    brt = boto3.client(service_name='bedrock-runtime')

    accept = "application/json"
    content_type = "application/json"

    response = brt.invoke_model(
        body=body, modelId=model_id, accept=accept, contentType=content_type
    )
    response_body = json.loads(response.get("body").read())

    finish_reason = response_body.get("error")

    if finish_reason is not None:
        raise ImageError(f"Text generation error. Error is {finish_reason}")

    logger.info(
        "Successfully generated text with provisioned custom model %s", model_id)

    return response_body


def main():
    """
    Entrypoint for example.
    """
    try:
        logging.basicConfig(level=logging.INFO,
                            format="%(levelname)s: %(message)s")

        model_id = provisionedModelArn

        body = json.dumps({
            "inputText": "what is AWS?"
        })

        response_body = generate_text(model_id, body)
        print(f"Input token count: {response_body['inputTextTokenCount']}")

        for result in response_body['results']:
            print(f"Token count: {result['tokenCount']}")
            print(f"Output text: {result['outputText']}")
            print(f"Completion reason: {result['completionReason']}")

    except ClientError as err:
        message = err.response["Error"]["Message"]
        logger.error("A client error occurred: %s", message)
        print("A client error occured: " +
              format(message))
    except ImageError as err:
        logger.error(err.message)
        print(err.message)

    else:
        print(
            f"Finished generating text with your provisioned custom model {model_id}.")


if __name__ == "__main__":
    main()