本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
模型定制的代码示例
以下代码示例展示了如何准备基本数据集、设置权限、创建自定义模型、查看输出文件、购买模型吞吐量以及对模型运行推理。您可以根据自己的特定用例修改这些代码片段。
-
准备训练数据集。
-
创建一个包含以下一行的训练数据集文件,并将其命名为
train.jsonl。
{"prompt": "what is AWS", "completion": "it's Amazon Web Services"}
-
为训练数据创建一个 S3 存储桶,为输出数据创建另一个 S3 存储桶(名称必须是唯一的)。
-
将
train.jsonl 上传到训练数据
桶中。
-
-
创建访问您的培训的策略,并将其附加到具有 Amazon Bedrock 信任关系的 IAM 角色。选择与您选择的方法相对应的选项卡,然后按照步骤操作。
- Console
-
-
创建 S3 策略。
-
导航到 IAM 控制台 https://console.aws.amazon.com/iam
,然后从左侧导航窗格中选择策略。 -
选择创建策略,然后选择 JSON 以打开策略编辑器。
-
粘贴以下策略,将
$ {training-bucket}
和$ {output-bucket}
替换为您的存储桶名称,然后选择下一步。{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::
${training-bucket}
", "arn:aws:s3:::${training-bucket}
/*" ] }, { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:PutObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::${output-bucket}
", "arn:aws:s3:::${output-bucket}
/*" ] } ] } -
命名策略
MyFineTuningDataAccess
并选择创建策略。
-
-
创建 IAM 角色并附加策略。
-
从左侧导航窗格中选择 “角色”,然后选择 “创建角色”。
-
选择自定义信任策略,粘贴以下策略,然后选择下一步。
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "bedrock.amazonaws.com" }, "Action": "sts:AssumeRole" } ] }
-
搜索您创建的
MyFineTuningDataAccess
策略,选中该复选框,然后选择下一步。 -
为角色命名
MyCustomizationRole
并选择创建角色
。
-
-
- CLI
-
-
创建一个名为
BedrockTrust.json
的文件并将以下策略粘贴到其中。{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "bedrock.amazonaws.com" }, "Action": "sts:AssumeRole" } ] }
-
创建另一个名为
MyFineTuningDataAccess.json
的文件并将以下策略粘贴到其中,将 $ {training-bucket} 和 $ {output-bucket}
替换为您的存储桶名称
。{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::
${training-bucket}
", "arn:aws:s3:::${training-bucket}
/*" ] }, { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:PutObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::${training-bucket}
", "arn:aws:s3:::${training-bucket}
/*" ] } ] } -
在终端中,导航到包含您创建的策略的文件夹。
-
CreateRole请求创建名为的 IAM 角色
MyCustomizationRole
并附加您创建的BedrockTrust.json
信任策略。aws iam create-role \ --role-name MyCustomizationRole \ --assume-role-policy-document file://BedrockTrust.json
-
使用您创建的
MyFineTuningDataAccess.js
on 文件CreatePolicy请求创建 S3 数据访问策略。该策略Arn
的响应会返回一个。aws iam create-policy \ --policy-name MyFineTuningDataAccess \ --policy-document file://myFineTuningDataAccess.json
-
AttachRolePolicy请求将 S3 数据访问策略附加到您的角色,将
policy-arn
替换为上一步响应中的 ARN:aws iam attach-role-policy \ --role-name MyCustomizationRole \ --policy-arn
${policy-arn}
-
- Python
-
-
运行以下代码,CreateRole请求创建名为的 IAM 角色
MyCustomizationRole
并CreatePolicy请求创建名为的 S3 数据访问策略MyFineTuningDataAccess
。对于 S3 数据访问策略,请将$ {training-bucket} 和 $ {output-
bucket} 替换为您的 S3 存储桶
名称。import boto3 import json iam = boto3.client("iam") iam.create_role( RoleName="MyCustomizationRole", AssumeRolePolicyDocument=json.dumps({ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "bedrock.amazonaws.com" }, "Action": "sts:AssumeRole" } ] }) ) iam.create_policy( PolicyName="MyFineTuningDataAccess", PolicyDocument=json.dumps({ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::
${training-bucket}
", "arn:aws:s3:::${training-bucket}
/*" ] }, { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:PutObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::${output-bucket}
", "arn:aws:s3:::${output-bucket}
/*" ] } ] }) ) -
响应
Arn
中返回。运行以下代码片段AttachRolePolicy发出请求,将$ {policy-arn}
替换为返回的。Arn
iam.attach_role_policy( RoleName="MyCustomizationRole", PolicyArn="
${policy-arn}
" )
-
-
选择一种语言以查看调用模型自定义 API 操作的代码示例。
- CLI
-
首先,创建一个名为
FineTuningData.json
的文本文件。将 JSON 代码从下方复制到文本文件中,将$ {training-bucket} 和 $ {output-
bucket} 替换为您的 S3 存储桶
名称。{ "trainingDataConfig": { "s3Uri": "s3://
${training-bucket}
/train.jsonl" }, "outputDataConfig": { "s3Uri": "s3://${output-bucket}
" } }要提交模型自定义作业,请在终端中导航到包含
FineTuningData.json
的文件夹,然后在命令行中运行以下命令,将$ {your-customization-role-arn}
替换为您设置的模型自定义角色。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=.0005,learningRateWarmupSteps=0 \ --cli-input-json file://FineTuningData.json响应返回一个
jobarn。
留出一些时间来完成任务。您可以使用以下命令检查其状态。aws bedrock get-model-customization-job \ --job-identifier "
jobArn
"当
status
为时COMPLETE
,你可以在响应trainingMetrics
中看到。您可以通过运行以下命令将构件下载到当前文件夹,将
aet.et-bucket 替换为您的输出存储桶
名称,将 jobID 替换为自定义任务的 ID(最后一个斜杠之后的序列)。jobArn
aws s3 cp s3://
${output-bucket}
/model-customization-job-jobId
. --recursive使用以下命令为您的自定义模型购买无承诺预置吞吐量。
注意
此次购买将按小时向您收费。使用控制台查看不同选项的估算价格。
aws bedrock create-provisioned-model-throughput \ --model-id MyCustomModel \ --provisioned-model-name MyProvisionedCustomModel \ --model-units 1
响应返回 a
provisionedModelArn
。留出一段时间来创建预配置吞吐量。要检查其状态,请在以下命令provisioned-model-id
中提供已配置模型的名称或 ARN。aws bedrock get-provisioned-model-throughput \ --provisioned-model-id
${provisioned-model-arn}
如果
status
是InService
,则可以使用以下命令对自定义模型进行推理。您必须提供预配置模型的 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 - Python
-
运行以下代码片段以提交微调作业。将
$ {your-customization-role-arn}
替换为您设置的 ARN,将 $ {training-bucket} 和 $ {output-bucket}
替换为您的 S3 存储桶名称
。MyCustomizationRole
import boto3 import json 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": ".0005", "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')响应返回一个
jobarn。
留出一些时间来完成任务。您可以使用以下命令检查其状态。bedrock.get_model_customization_job(jobIdentifier=jobArn).get('status')
当
status
为时COMPLETE
,你可以在GetModelCustomizationJob响应trainingMetrics
中看到。您也可以按照下载对象中的步骤下载指标。使用以下命令为您的自定义模型购买无承诺预置吞吐量。
response_pt = bedrock.create_provisioned_model_throughput( modelId="MyCustomModel", provisionedModelName="MyProvisionedCustomModel" modelUnits="1" ) provisionedModelArn = response_pt.get('provisionedModelArn')
响应返回 a
provisionedModelArn
。留出一段时间来创建预配置吞吐量。要检查其状态,请在以下命令provisionedModelId
中提供已配置模型的名称或 ARN。bedrock.get_provisioned_model_throughput(provisionedModelId=provisionedModelArn)
如果
status
是InService
,则可以使用以下命令对自定义模型进行推理。您必须提供预配置模型的 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 = 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()