优化生成式 AI 应用程序在 Amazon Bedrock 的部署

关键要点

在构建生成式人工智能AI应用程序的部署管道时，面临诸多复杂性与独特要求。本文将向您展示如何利用 Amazon SageMaker Pipelines 装饰器将 Python 代码转变为可重用的工作流，进而高效地进行模型微调，确保资源的有效利用和团队的协作。

构建生成式人工智能AI应用程序的部署管道是一项艰巨的任务，因为此类系统涉及复杂的功能和独特的要求。生成式 AI 模型持续演进，频繁发布新版本和更新，这使得在大规模部署管道中管理和部署这些更新，确保一致性且最小化停机时间变得尤为重要。此外，生成式 AI 应用需要不断从各类源中摄取、预处理和格式化大量数据，构建能够可靠且高效处理这些工作负载的强大数据管道是一大挑战。同时，监控生成式 AI 模型在生产环境中的性能、偏见和伦理影响也是一项重要任务。

实现这一目标需要在资源、专业知识和跨职能合作方面进行重大投资，涉及数据科学家、机器学习ML开发人员以及关注 AI/ML 项目独特性的机器学习运维MLOps工程师等多个角色。在这篇文章中，我们将展示如何将本地文件中的 Python 代码转换成可以利用 Amazon SageMaker Pipelines 装饰器重用的工作流，从而优化生成式 AI 模型在 Amazon Bedrock 的微调过程。您还可以通过使用 Amazon SageMaker Model Building Pipelines 在多个 AI/ML 团队之间进行协作。

SageMaker Pipelines

您可以借助 SageMaker Pipelines 定义和协调机器学习生命周期中的各个步骤，如数据预处理、模型训练、评估和部署。这简化了流程，并为管道不同阶段提供一致性。SageMaker Pipelines 能够处理模型版本控制和谱系追踪，它自动跟踪模型工件、超参数和元数据，帮助您重现和审计模型版本。

SageMaker Pipelines 的装饰器特性可以将本地点 ML 代码转换成一个或多个管道步骤。由于 Amazon Bedrock 可以通过 API 访问，因此即使是对 Amazon SageMaker 不熟悉的开发人员也可以编写常规 Python 程序来实现 Amazon Bedrock 应用或进行微调。

您可以像任何 ML 项目那样编写 ML 函数。在经过本地测试或作为训练作业后，经验丰富的数据科学家或 SageMaker 实践者可以通过添加 @step 装饰器将该函数转换为 SageMaker 管道步骤。

解决方案概述

SageMaker 模型构建管道是一个利用直接 SageMaker 集成构建 ML 管道的工具。借助此集成，您可以使用一个处理大量步骤创建和管理的工具，轻松创建管道。

随着您从试点和测试阶段转向大规模部署生成式 AI 模型，您需要将 DevOps 实践应用于 ML 工作负载。SageMaker Pipelines 与 SageMaker 集成，因此您无需与其他任何 AWS 服务交互。此外，由于 SageMaker Pipelines 是一项完全托管的服务，您无需管理任何资源，这意味着它会为您创建和管理资源。 Amazon SageMaker Studio 提供一个环境，以管理端到端的 SageMaker 管道体验。本文中的解决方案展示了如何将编写用于预处理、微调和测试大型语言模型LLM的 Python 代码，利用 Amazon Bedrock API 转换为 SageMaker 管道，从而提高 ML 运营效率。

该解决方案包含三个主要步骤：

使用 Python 编写代码以在 Amazon Bedrock 中预处理、训练和测试 LLM。添加 @step 装饰的函数，将 Python 代码转换为 SageMaker 管道。创建并运行 SageMaker 管道。

下图说明了解决方案工作流。

先决条件

如果您只想查看笔记本代码，可以参考 GitHub 上的笔记本。

如果您是 AWS 的新用户，首先需要创建并设置一个 AWS 账户，之后在您的 AWS 账户中设置 SageMaker Studio，创建一个 JupyterLab 空间以运行 JupyterLab 应用程序。

在 SageMaker Studio JupyterLab 空间中，完成以下步骤：

在文件菜单上选择新建并打开终端。

在终端中输入以下代码：

bashgit clone https//githubcom/aws/amazonsagemakerexamplesgit

您会在 SageMaker Studio 文件资源管理器窗格中看到名为 amazonsagemakerexamples 的文件夹。

打开文件夹 amazonsagemakerexamples/sagemakerpipelines/stepdecorator/bedrockexamples。打开笔记本 finetunebedrockstepdecoratoripynb。

该笔记本包含了本篇文章的所有代码，您可以从头到尾运行它。

笔记本代码解释

该笔记本使用用户的默认 Amazon 简单存储服务 (Amazon S3) 存储桶。默认的 S3 存储桶遵循命名模式 s3//sagemaker{Region}{youraccountid}，如果它尚不存在，则会自动创建。

一元免费机场

它使用 SageMaker Studio 默认的 AWS 身份和访问管理IAM角色。如果您的 SageMaker Studio 用户角色没有管理员访问权限，则需要向该角色添加必要的权限。

有关更多信息，请参见以下链接：

CreateTrainingJob API Execution Role Permissions基于身份的 Amazon Bedrock 策略示例为模型自定义创建服务角色IAM 访问管理

它会创建一个 SageMaker 会话并获取默认 S3 存储桶和 IAM 角色：

pythonsagemakersession = sagemakersessionSession()region = sagemakersessionbotoregionname

bucketname = sagemakersessiondefaultbucket()rolearn = sagemakergetexecutionrole()

使用 Amazon SageMaker Pipeline 装饰器微调 Amazon Bedrock

使用 Python 在 Amazon Bedrock 中预处理、训练和测试 LLM

首先，我们需要下载数据并准备 Amazon Bedrock 中的 LLM。这一步将使用 Python 完成。

加载数据

我们使用 CNN/DailyMail 数据集来微调模型。CNN/DailyMail 数据集是一个包含超过 300000 篇由 CNN 和 Daily Mail 的记者撰写的独特新闻文章的英文数据集。原始数据集包括用于训练、验证和测试的文章及其摘要。在使用该数据集前，我们需要对其进行格式化以包括提示。以下代码演示了这个过程：

pythondef addprompttodata(dataset) datapoints = []

for datapoint in dataset    # 为每篇 CNN 文章添加提示    # 并为文章摘要添加前缀 response     tempdict = {}    tempdict[prompt] = instruction  datapoint[article]    tempdict[completion] = responsenn  datapoint[highlights]    datapointsappend(tempdict)return datapoints

def dataload(dsname str dsversion str) gt tuple dataset = loaddataset(dsname dsversion) datapointstrain = addprompttodata(dataset[train]) datapointsvalid = addprompttodata(dataset[validation]) datapointstest = addprompttodata(dataset[test])

数据拆分

然后将数据集拆分为训练集、验证集和测试集。为便于说明，本文章限制每行最多 3000 个单词，并选择 100 行用于训练，10 行用于验证，5 行用于测试。有关更多详细信息，请参见 GitHub 上的笔记本。

pythondef datasplit(steploadresult tuple) gt tuple trainlines = reducedatasetsize(steploadresult[0] 3000 100) validationlines = reducedatasetsize(steploadresult[1] 3000 10) testlines = reducedatasetsize(steploadresult[2] 3000 5)

return trainlines validationlines testlines

将数据上传到 Amazon S3

接下来，我们将数据转换为 JSONL 格式并将训练、验证和测试文件上传到 Amazon S3：

pythondef uploadfiletos3(bucketname str filenames tuple s3keynames tuple) import boto3 s3client = boto3client(s3) for i in range(len(filenames)) s3clientuploadfile(filenames[i] bucketname s3keynames[i])

def datauploadtos3(datasplitresponse tuple bucketname str) gt tuple datasetfolder = finetuningdatasets

if not ospathexists(datasetfolder)    osmakedirs(datasetfolder)abspath = ospathabspath(datasetfolder)trainfile = writejsonlfile(abspath traincnnjsonl datasplitresponse[0])valfile = writejsonlfile(abspath validationcnnjsonl datasplitresponse[1])testfile = writejsonlfile(abspath testcnnjsonl datasplitresponse[2])filenames = trainfile valfile testfiles3keys = f{datasetfolder}/train/traincnnjsonl f{datasetfolder}/validation/validationcnnjsonl f{datasetfolder}/test/testcnnjsonluploadfiletos3(bucketname filenames s3keys)

训练模型

上传训练数据后，就可以使用 CNN/DailyMail 数据集微调 Amazon Bedrock 模型。我们微调的是 Amazon Titan Text Lite 模型，用于摘要生成。我们定义超参数并启动训练作业：

pythonhyperparameters = { epochCount 2 batchSize 1 learningRate 000003}

trainingjobresponse = bedrockcreatemodelcustomizationjob( customizationType=FINETUNING jobName=trainingjobname customModelName=custommodelname roleArn=rolearn baseModelIdentifier=amazontitantextlitev104k hyperParameters=hyperparameters trainingDataConfig=trainingdataconfig validationDataConfig=validationdataconfig outputDataConfig=outputdataconfig)

modelid = bedrockgetcustommodel(modelIdentifier=custommodelname)[modelArn]

print(fModel id {modelid})return modelid

创建预配置吞吐量

吞吐量指的是模型处理和返回的输入与输出的数量和速率。您可以购买预配置吞吐量，以便为模型分配专用资源，而不是根据需求进行吞吐量的变化。在自定义模型中，您必须购买预配置吞吐量才能使用它。有关更多信息，请参阅 Amazon Bedrock 的预配置吞吐量。

pythondef createprovthruput(modelid str provisionedmodelname str) gt str bedrock = boto3client(servicename=bedrock)

provisionedmodelid = bedrockcreateprovisionedmodelthroughput(            modelUnits=1            provisionedModelName=provisionedmodelname            modelId=modelid            )[provisionedModelArn]return provisionedmodelid

测试模型

现在可以调用并测试模型。我们结合来自测试数据集的 Amazon Bedrock 运行时提示，以及在之前步骤中设置的预配置吞吐量 ID 和推理参数，如 maxTokenCount stopSequence temperature 以及 top：

pythondef testmodel(provisionedmodelid str) gt tuple s3downloadfile(s3bucket s3key testcnnjsonl)

body = jsondumps(    {        inputText prompt        textGenerationConfig {            maxTokenCount 2048            stopSequences [User]            temperature 0            topP 09        }    })accept = application/jsoncontentType = application/jsonbedrockruntime = boto3client(servicename=bedrockruntime)finetunedresponse = bedrockruntimeinvokemodel(body=body                                    modelId=provisionedmodelid                                    accept=accept                                    contentType=contentType)finetunedresponsebody = jsonloads(finetunedresponseget(body)read())summary = finetunedresponsebody[results][0][outputText]return prompt summary

使用 @step 装饰器将函数转换为 SageMaker 管道步骤

@step 装饰器是将本地 ML 代码转换为一个或多个管道步骤的功能。您可以按照任何 ML 项目的方式编写 ML 函数，然后通过使用 @step 装饰器将 Python 函数转换为管道步骤，创建这些函数之间的依赖关系，从而形成一个管道图或有向无环图DAG，并将该图的叶节点作为步骤列表传递给管道。要创建一个步骤，只需使用 @step 装饰器注释该函数即可。当此函数被调用时，它会接收上一个管道步骤的 DelayedReturn 输出作为输入。该实例保存了函数内定义的所有之前步骤的信息，形成 SageMaker 管道 DAG。

在笔记本中，我们已经在每个函数定义前添加了 @step 装饰器，如下代码所示。该函数的代码来自于我们试图转换为 SageMaker 管道的微调 Python 程序。

python@step(name=dataloadstepkeepaliveperiodinseconds=300)def dataload(dsname str dsversion str) gt tuple return datapointstrain datapointsvalid datapointstest

@step(name=datasplitstepkeepaliveperiodinseconds=300)def datasplit(steploadresult tuple) gt tuple return trainlines validationlines testlines

@step(name=datauploadtos3stepkeepaliveperiodinseconds=300)def datauploadtos3(datasplitresponse tuple bucketname str) gt tuple return fs3//{bucketname}/{s3keys[0]} fs3//{bucketname}/{s3keys[1]} fs3//{bucketname}/{s3keys[2]}

@step(name=modeltrainingstepkeepaliveperiodinseconds=300)def train(custommodelname strtrainingjobname strstepdatauploadtos3result tuple) gt str return modelid

@step(name=createprovisionedthroughputstepkeepaliveperiodinseconds=300)def createprovthruput(modelid str provisionedmodelname str) gt str return provisionedmodelid

@step(name=modeltestingstepkeepaliveperiodinseconds=300)def testmodel(provisionedmodelid str) gt tuple return prompt summary