を使用して推論用の大規模なモデルをデプロイする TorchServe

このチュートリアルでは、 SageMaker TorchServe で Amazon に大規模なモデルをデプロイし、推論を提供する方法を示しますGPUs。この例では、OPT-30b モデルをml.g5インスタンスにデプロイします。これを変更して、他のモデルやインスタンスタイプと連携できます。例italicized placeholder textの を独自の情報に置き換えます。

TorchServe は、大規模な分散モデル推論のための強力なオープンプラットフォームです。 PyTorch、ネイティブ P iPPy DeepSpeed、 HuggingFaceおよび Accelerate などの一般的なライブラリをサポートすることで、分散された大規模モデルおよび分散されていないモデル推論シナリオ全体で一貫性APIsを維持する均一なハンドラーを提供します。詳細については、「 TorchServeの大規模モデル推論ドキュメント」を参照してください。

を使用した深層学習コンテナ TorchServe

TorchServe で を使用して大規模なモデルをデプロイするには SageMaker、 SageMaker 深層学習コンテナ () のいずれかを使用できますDLCs。デフォルトでは、 TorchServe はすべての AWS PyTorch にインストールされますDLCs。モデルのロード中に、P 、DeepspeediPPy、Accelerator などの大規模なモデルに合わせた特殊なライブラリをインストール TorchServe できます。

次の表は、 SageMaker DLCsを含む をすべて一覧表示しています TorchServe。

DLC cateogry	フレームワーク	ハードウェア	例 URL
SageMaker フレームワークコンテナ	PyTorch 2.0.0 以降	CPU, GPU	763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-inference:2.0.1-gpu-py310-cu118-ubuntu20.04-sagemaker
SageMaker Framework Graviton コンテナ	PyTorch 2.0.0 以降	CPU	763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-inference-graviton:2.0.1-cpu-py310-ubuntu20.04-sagemaker
StabilityAI 推論コンテナ	PyTorch 2.0.0 以降	GPU	763104351884.dkr.ecr.us-east-1.amazonaws.com/stabilityai-pytorch-inference:2.0.1-sgm0.1.0-gpu-py310-cu118-ubuntu20.04-sagemaker
Neuron コンテナ	PyTorch 1.13.1	Neuronx	763104351884.dkr.ecr.us-west-2.amazonaws.com/pytorch-inference-neuron:1.13.1-neuron-py310-sdk2.12.0-ubuntu20.04

使用開始

モデルをデプロイする前に、前提条件を完了してください。モデルパラメータを設定したり、ハンドラーコードをカスタマイズしたりすることもできます。

前提条件

開始するには、次の前提条件が整っていることを確認してください。

1. AWS アカウントにアクセスできることを確認します。がユーザーまたはIAMロールを介して AWS IAMアカウント AWS CLI にアクセスできるように環境を設定します。IAM ロールを使用することをお勧めします。個人アカウントでテストするために、IAMロールに次の管理アクセス許可ポリシーをアタッチできます。

   • AmazonEC2ContainerRegistryFullAccess

   • AmazonEC2FullAccess

   • AWSServiceRoleForAmazonEKSNodegroup

   • AmazonSageMakerFullAccess

   • AmazonS3FullAccess

   ロールへのIAMポリシーのアタッチの詳細については、AWS IAM「 ユーザーガイド」のIAM「ID アクセス許可の追加と削除」を参照してください。

2. 次の例に示すように、依存関係をローカルに設定します。

   1. のバージョン 2 をインストールします AWS CLI。

      # Install the latest AWS CLI v2 if it is not installed
      !curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" !unzip awscliv2.zip
      #Follow the instructions to install v2 on the terminal
      !cat aws/README.md

   2. SageMaker と Boto3 クライアントをインストールします。

      # If already installed, update your client
      #%pip install sagemaker pip --upgrade --quiet
      !pip install -U sagemaker
      !pip install -U boto
      !pip install -U botocore
      !pip install -U boto3

モデル設定とパラメータを設定する

TorchServe はtorchrun、モデル並列処理用の分散環境をセットアップします。 TorchServe には、大規模なモデルに対して複数のワーカーをサポートする機能があります。デフォルトでは、 はラウンドロビンアルゴリズム TorchServe を使用して、ホスト上のワーカーGPUsに割り当てます。モデル推論が大きい場合、各ワーカーGPUsに割り当てられた数は、model_config.yamlファイルでGPUs指定された数に基づいて自動的に計算されます。特定の時間IDsに表示されるGPUデバイスCUDA_VISIBLE_DEVICESを指定する環境変数 は、この数に基づいて設定されます。

例えば、ノードGPUsに 8 人いて、1 人のワーカーがノードGPUsに 4 人 () を必要とするとしますnproc_per_node=4。この場合、 は最初のワーカー (CUDA_VISIBLE_DEVICES="0,1,2,3") GPUsに 4 つ、2 番目のワーカー () GPUs に 4 つを TorchServe 割り当てますCUDA_VISIBLE_DEVICES="4,5,6,7”。

このデフォルトの動作に加えて、 はユーザーがワーカーGPUsに を指定するための柔軟性 TorchServe を提供します。例えば、モデル設定YAMLファイル deviceIds: [2,3,4,5]で変数 を設定し、 を設定するとnproc_per_node=2、 は最初のワーカーに を、2 番目のワーカーCUDA_VISIBLE_DEVICES="4,5”に CUDA_VISIBLE_DEVICES=”2,3”を TorchServe 割り当てます。

次のmodel_config.yaml例では、OPT-30b モデルのフロントエンドパラメータとバックエンドパラメータの両方を設定します。設定されたフロントエンドパラメータは、parallelType、deviceType、deviceIds および ですtorchrun。設定できるフロントエンドパラメータの詳細については、PyTorch GitHub ドキュメント を参照してください。バックエンド設定は、フリースタイルのカスタマイズを可能にするYAMLマップに基づいています。バックエンドパラメータについては、カスタムハンドラーコードで使用される DeepSpeed 設定および追加のパラメータを定義します。

# TorchServe front-end parameters
minWorkers: 1
maxWorkers: 1
maxBatchDelay: 100
responseTimeout: 1200
parallelType: "tp"
deviceType: "gpu"
# example of user specified GPU deviceIds
deviceIds: [0,1,2,3] # sets CUDA_VISIBLE_DEVICES

torchrun:
    nproc-per-node: 4

# TorchServe back-end parameters
deepspeed:
    config: ds-config.json
    checkpoint: checkpoints.json

handler: # parameters for custom handler code
    model_name: "facebook/opt-30b"
    model_path: "model/models--facebook--opt-30b/snapshots/ceea0a90ac0f6fae7c2c34bcb40477438c152546"
    max_length: 50
    max_new_tokens: 10
    manual_seed: 40

ハンドラーをカスタマイズする

TorchServe は、一般的なライブラリで構築された大規模なモデル推論用のベースハンドラーとハンドラーユーティリティを提供します。次の例は、カスタムハンドラークラス がハンドラーユーティリティ TransformersSeqClassifierHandlerを拡張BaseDeepSpeedHandlerして使用する方法を示しています。完全なコード例については、ドキュメントの custom_handler.pyコード PyTorch GitHubを参照してください。

class TransformersSeqClassifierHandler(BaseDeepSpeedHandler, ABC):
    """
    Transformers handler class for sequence, token classification and question answering.
    """

    def __init__(self):
        super(TransformersSeqClassifierHandler, self).__init__()
        self.max_length = None
        self.max_new_tokens = None
        self.tokenizer = None
        self.initialized = False

    def initialize(self, ctx: Context):
        """In this initialize function, the HF large model is loaded and
        partitioned using DeepSpeed.
        Args:
            ctx (context): It is a JSON Object containing information
            pertaining to the model artifacts parameters.
        """
        super().initialize(ctx)
        model_dir = ctx.system_properties.get("model_dir")
        self.max_length = int(ctx.model_yaml_config["handler"]["max_length"])
        self.max_new_tokens = int(ctx.model_yaml_config["handler"]["max_new_tokens"])
        model_name = ctx.model_yaml_config["handler"]["model_name"]
        model_path = ctx.model_yaml_config["handler"]["model_path"]
        seed = int(ctx.model_yaml_config["handler"]["manual_seed"])
        torch.manual_seed(seed)

        logger.info("Model %s loading tokenizer", ctx.model_name)

        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.tokenizer.pad_token = self.tokenizer.eos_token
        config = AutoConfig.from_pretrained(model_name)
        with torch.device("meta"):
            self.model = AutoModelForCausalLM.from_config(
                config, torch_dtype=torch.float16
            )
        self.model = self.model.eval()

        ds_engine = get_ds_engine(self.model, ctx)
        self.model = ds_engine.module
        logger.info("Model %s loaded successfully", ctx.model_name)
        self.initialized = True

    def preprocess(self, requests):
        """
        Basic text preprocessing, based on the user's choice of application mode.
        Args:
            requests (list): A list of dictionaries with a "data" or "body" field, each
                            containing the input text to be processed.
        Returns:
            tuple: A tuple with two tensors: the batch of input ids and the batch of
                attention masks.
        """

    def inference(self, input_batch):
        """
        Predicts the class (or classes) of the received text using the serialized transformers
        checkpoint.
        Args:
            input_batch (tuple): A tuple with two tensors: the batch of input ids and the batch
                                of attention masks, as returned by the preprocess function.
        Returns:
            list: A list of strings with the predicted values for each input text in the batch.
        """
        
    def postprocess(self, inference_output):
        """Post Process Function converts the predicted response into Torchserve readable format.
        Args:
            inference_output (list): It contains the predicted response of the input text.
        Returns:
            (list): Returns a list of the Predictions and Explanations.
        """

モデルのアーティファクトの準備

にモデルをデプロイする前に SageMaker、モデルアーティファクトをパッケージ化する必要があります。大きなモデルの場合、 引数 でツールを使用することをお勧めします。引数は--archive-format no-archiveモデルアーティファクトの PyTorch torch-model-archiver圧縮をスキップします。次の例では、すべてのモデルアーティファクトを という名前の新しいフォルダに保存しますopt/。

torch-model-archiver --model-name opt --version 1.0 --handler custom_handler.py --extra-files ds-config.json -r requirements.txt --config-file opt/model-config.yaml --archive-format no-archive

opt/ フォルダを作成したら、 PyTorch Download_model ツールを使用して OPT-30b モデルをフォルダにダウンロードします。

cd opt
python path_to/Download_model.py --model_path model --model_name facebook/opt-30b --revision main

最後に、モデルアーティファクトを Amazon S3 バケットにアップロードします。

aws s3 cp opt {your_s3_bucket}/opt --recursive

これで、Amazon S3 にモデルアーティファクトが保存され、エンドポイントにデプロイする準備が整いました SageMaker。

SageMaker Python を使用してモデルをデプロイする SDK

モデルアーティファクトを準備したら、 SageMaker ホスティングエンドポイントにモデルをデプロイできます。このセクションでは、単一の大きなモデルをエンドポイントにデプロイし、ストリーミングレスポンスの予測を行う方法について説明します。エンドポイントからのレスポンスのストリーミングの詳細については、「リアルタイムエンドポイントの呼び出し」を参照してください。

モデルをデプロイするには、次の手順を実行します。

1. 次の例に示すように、 SageMaker セッションを作成します。

   import boto3
   import sagemaker
   from sagemaker import Model, image_uris, serializers, deserializers
   
   boto3_session=boto3.session.Session(region_name="us-west-2")
   smr = boto3.client('sagemaker-runtime-demo')
   sm = boto3.client('sagemaker')
   role = sagemaker.get_execution_role()  # execution role for the endpoint
   sess= sagemaker.session.Session(boto3_session, sagemaker_client=sm, sagemaker_runtime_client=smr)  # SageMaker session for interacting with different AWS APIs
   region = sess._region_name  # region name of the current SageMaker Studio Classic environment
   account = sess.account_id()  # account_id of the current SageMaker Studio Classic environment
   
   # Configuration:
   bucket_name = sess.default_bucket()
   prefix = "torchserve"
   output_path = f"s3://{bucket_name}/{prefix}"
   print(f'account={account}, region={region}, role={role}, output_path={output_path}')

2. 次の例に示すように SageMaker、 で非圧縮モデルを作成します。

   from datetime import datetime
   
   instance_type = "ml.g5.24xlarge"
   endpoint_name = sagemaker.utils.name_from_base("ts-opt-30b")
   s3_uri = {your_s3_bucket}/opt
   
   model = Model(
       name="torchserve-opt-30b" + datetime.now().strftime("%Y-%m-%d-%H-%M-%S"),
       # Enable SageMaker uncompressed model artifacts
       model_data={
           "S3DataSource": {
                   "S3Uri": s3_uri,
                   "S3DataType": "S3Prefix",
                   "CompressionType": "None",
           }
       },
       image_uri=container,
       role=role,
       sagemaker_session=sess,
       env={"TS_INSTALL_PY_DEP_PER_MODEL": "true"},
   )
   print(model)

3. 次の例に示すように、モデルを Amazon EC2インスタンスにデプロイします。

   model.deploy(
       initial_instance_count=1,
       instance_type=instance_type,
       endpoint_name=endpoint_name,
       volume_size=512, # increase the size to store large model
       model_data_download_timeout=3600, # increase the timeout to download large model
       container_startup_health_check_timeout=600, # increase the timeout to load large model
   )

4. 次の例に示すように、ストリーミングレスポンスを処理するクラスを初期化します。

   import io
   
   class Parser:
       """
       A helper class for parsing the byte stream input. 
       
       The output of the model will be in the following format:
       ```
       b'{"outputs": [" a"]}\n'
       b'{"outputs": [" challenging"]}\n'
       b'{"outputs": [" problem"]}\n'
       ...
       ```
       
       While usually each PayloadPart event from the event stream will contain a byte array 
       with a full json, this is not guaranteed and some of the json objects may be split across
       PayloadPart events. For example:
       ```
       {'PayloadPart': {'Bytes': b'{"outputs": '}}
       {'PayloadPart': {'Bytes': b'[" problem"]}\n'}}
       ```
       
       This class accounts for this by concatenating bytes written via the 'write' function
       and then exposing a method which will return lines (ending with a '\n' character) within
       the buffer via the 'scan_lines' function. It maintains the position of the last read 
       position to ensure that previous bytes are not exposed again. 
       """
       
       def __init__(self):
           self.buff = io.BytesIO()
           self.read_pos = 0
           
       def write(self, content):
           self.buff.seek(0, io.SEEK_END)
           self.buff.write(content)
           data = self.buff.getvalue()
           
       def scan_lines(self):
           self.buff.seek(self.read_pos)
           for line in self.buff.readlines():
               if line[-1] != b'\n':
                   self.read_pos += len(line)
                   yield line[:-1]
                   
       def reset(self):
           self.read_pos = 0

5. 次の例に示すように、ストリーミングレスポンス予測をテストします。

   import json
   
   body = "Today the weather is really nice and I am planning on".encode('utf-8')
   resp = smr.invoke_endpoint_with_response_stream(EndpointName=endpoint_name, Body=body, ContentType="application/json")
   event_stream = resp['Body']
   parser = Parser()
   for event in event_stream:
       parser.write(event['PayloadPart']['Bytes'])
       for line in parser.scan_lines():
           print(line.decode("utf-8"), end=' ')

これで、モデルを SageMaker エンドポイントにデプロイし、レスポンスのためにモデルを呼び出すことができるはずです。リアルタイムエンドポイントの詳細については、 SageMaker 「」を参照してくださいシングルモデルエンドポイント。