Amazon Bedrock의 프로비저닝된 처리량에 대한 코드 샘플

다음 코드 예제는 AWS CLI 및 Python SDK를 사용하여 프로비저닝된 처리량을 생성, 사용, 관리하는 방법을 보여줍니다.

AWS CLI

터미널에서 다음 명령어를 실행하여 Anthropic Claude v2.1 모델에서 사용자 지정된 사용자 지정 모델을 MyPT 기반으로 MyCustomModel 커밋이 필요 없는 프로비저닝된 처리량을 생성하십시오.

aws bedrock create-provisioned-model-throughput \
   --model-units 1 \
   --provisioned-model-name MyPT \
   --model-id arn:aws:bedrock:us-east-1::custom-model/anthropic.claude-v2:1:200k/MyCustomModel

응답은 a를 반환합니다. provisioned-model-arn 생성이 완료될 때까지 잠시 기다려 주세요. 상태를 확인하려면 다음 명령과 같이 프로비저닝된 모델의 이름 또는 ARN을 provisioned-model-id 입력합니다.

aws bedrock get-provisioned-model-throughput \
    --provisioned-model-id MyPT

프로비저닝된 처리량의 이름을 변경하고 v2.1에서 사용자 지정된 다른 모델에 연결합니다. Anthropic Claude

aws bedrock update-provisioned-model-throughput \
    --provisioned-model-id MyPT \ 
    --desired-provisioned-model-name MyPT2 \
    --desired-model-id arn:aws:bedrock:us-east-1::custom-model/anthropic.claude-v2:1:200k/MyCustomModel2

다음 명령어를 사용하여 업데이트된 프로비저닝 모델을 사용하여 추론을 실행합니다. UpdateProvisionedModelThroughput응답에 반환된 프로비저닝된 모델의 ARN을 다음과 같이 제공해야 합니다. model-id 출력은 현재 폴더의 output.txt 파일에 기록됩니다.

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

다음 명령을 사용하여 프로비저닝된 처리량을 삭제합니다. 프로비저닝된 처리량에 대해서는 더 이상 요금이 청구되지 않습니다.

aws bedrock delete-provisioned-model-throughput 
  --provisioned-model-id MyPT2

Python (Boto)

다음 코드 스니펫을 실행하여 Anthropic Claude v2.1 모델에서 사용자 지정된 사용자 지정 모델을 MyPT 기반으로 약정 없는 프로비저닝된 처리량을 생성하십시오. MyCustomModel

import boto3 
                    
bedrock = boto3.client(service_name='bedrock')
bedrock.create_provisioned_model_throughput(
    modelUnits=1,
    provisionedModelName='MyPT', 
    modelId='arn:aws:bedrock:us-east-1::custom-model/anthropic.claude-v2:1:200k/MyCustomModel' 
)

provisionedModelArn응답은 a를 반환합니다. 생성이 완료될 때까지 잠시 기다려 주세요. 다음 코드 스니펫으로 상태를 확인할 수 있습니다. 프로비저닝된 처리량의 이름 또는 응답에서 반환된 ARN의 이름을 다음과 같이 CreateProvisionedModelThroughput제공할 수 있습니다. provisionedModelId

bedrock.get_provisioned_model_throughput(provisionedModelId='MyPT')

프로비저닝된 처리량의 이름을 변경하고 v2.1에서 사용자 지정된 다른 모델에 연결합니다. Anthropic Claude 그런 다음 GetProvisionedModelThroughput요청을 보내고 프로비저닝된 모델의 ARN을 추론에 사용할 변수에 저장합니다.

bedrock.update_provisioned_model_throughput(
    provisionedModelId='MyPT',
    desiredProvisionedModelName='MyPT2',
    desiredModelId='arn:aws:bedrock:us-east-1::custom-model/anthropic.claude-v2:1:200k/MyCustomModel2'
)
                
arn_MyPT2 = bedrock.get_provisioned_model_throughput(provisionedModelId='MyPT2').get('provisionedModelArn')

다음 명령어를 사용하여 업데이트된 프로비저닝 모델로 추론을 실행합니다. 프로비저닝된 모델의 ARN을 로 제공해야 합니다. 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 = arn_myPT2

        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()

다음 코드 스니펫을 사용하여 프로비저닝된 처리량을 삭제하십시오. 프로비저닝된 처리량에 대해서는 더 이상 요금이 부과되지 않습니다.

bedrock.delete_provisioned_model_throughput(provisionedModelId='MyPT2')