AIサービスのコーディングとデプロイを手動で行う

最終更新: 2025年3月25日
AIサービスのコーディングとデプロイを手動で行う

AIサービスの展開における手動コーディングのアプローチでは、AIサービスの展開と管理を行うためのカスタムコードを記述する必要があります。 このアプローチでは、 デプロイメントを完全に制御でき、特定の要件を満たすためのカスタマイズが可能です。

プロセスの概要

次の図は、AIサービスのコーディングプロセスを示しています。

プロジェクト内のAIサービスと接続を含むノートブックを作成できます。 AIサービスは、お客様のRAGアプリケーションのロジックをキャプチャし、コードのデプロイ可能な単位である生成機能を含んでいます。 生成関数はデプロイメントスペースに昇格され、 デプロイメントの作成に使用されます。 デプロイメントは、他のアプリケーションからアクセス可能なREST APIエンドポイントとして公開されます。 デプロイしたAIサービスを推論に使用するには、REST APIエンドポイントにリクエストを送信します。 展開されたAIサービスはリクエストを処理し、応答を返します。

手動コーディングのユースケース

AIサービスの作成と展開のためのタスク

AIサービスの作成、展開、管理には、以下の手順に従います

  1. AIサービスを作成する : Python を使用して、ノートブックでAIサービスを定義します。 AIサービスとして展開するには、AIサービスが特定の要件を満たす必要があります。
  2. AIサービスのテスト :AIサービスのコーディングロジックをローカルでテストします。
  3. AIサービス資産を作成する: After you create and test the AI service, you must package the AI service as a deployable asset.
  4. AIサービス資産を展開する: Deploy the AI service asset as an online or a batch deployment.
  5. AI デプロイメントテスト: オンライン推論またはバッチスコアリング用にデプロイされたAIサービスをテストします。
  6. AIサービスの管理: Access and update the deployment details. ユーザーインターフェースまたはプログラムから、 デプロイメントの規模を変更または削除します。

ノートブックでAIサービスを作成する

AIサービスを展開するには、ノートブックに直接AIサービスを作成することができます。 Python でAIサービスを定義する必要があり、特定の要件を満たす必要があります。 To deploy an AI service, you must create a watsonx.ai Runtime repository asset and upload the Python file to the asset.

Python クライアントライブラリを使用したAIサービスの定義

watsonx.ai Python クライアントライブラリを使用してノートブックでAIサービスを定義するには、以下の手順に従います

  1. Python でAIサービスを利用するには、 ibm-watsonx-ai Python SDKをインストールします

    pip install ibm-watsonx-ai
    
  2. After you install the Python client library, initialize the client and set the default deployment space:

    from ibm_watsonx_ai import APIClient
    from ibm_watsonx_ai import Credentials
    
    credentials = Credentials(
        url=url, api_key=apikey
    )
    
    client = APIClient(credentials)
    client.set.default_space(space_id=space_id)
    
  3. Python でAIサービスを定義するには、以下のレイアウトを使用してください。 ユースケースに応じて、少なくとも以下の機能のうち1つをネストされた関数として含める必要があります

    • generate()
    • generate_stream()
    • generate_batch()

    詳細は、 AIサービス作成の要件をご覧ください。

    def basic_generate_demo(context, model="google/flan-t5-xl", **parameters):
     # "parameters" is a reserved argument and will be enabled in future
    
     # generate token from task credentials api
     task_token = context.generate_token()
    
     def generate(context):
         user_token = context.get_token()  # extract token from header
         user_headers = context.get_headers()
         json_body = context.get_json()
    
         # example 1: json
         return {
             "headers": {
                 "Content-Type": "application/json",
                 "user-custom-header": "my-header-x1",
             },
             "body": {
                 "model": model
             },
         }
    
     def generate_stream(context):
         user_token = context.get_token()  # extract token from header
         user_headers = context.get_headers()
         json_body = context.get_json()
    
         # return a generator
         data_to_stream = json_body.get("sse", "Default message!")
         for x in data_to_stream:
             yield x
    
     def generate_batch(input_data_references, output_data_reference):
         # generate token from task credentials api
         task_token = context.generate_token()
         # do something.
         # ...
    
     return generate, generate_stream, generate_batch
    

AIサービスの定義に必要な要件

The AI service captures the logic of your generative AI use case (such as a Retrieval-augmented generation application) and handles the REST API call to the deployment endpoint /ml/v4/deployments.

AIサービスを定義するには、以下のガイドラインに従ってください

  • Python 関数を作成します。 関数には任意の名前を指定できます。 機能パラメータの詳細については 、 watsonx.ai のREST APIドキュメントを参照してください。

  • お客様の用途に応じて、展開する Python 機能には、そのスコープ内に、少なくとも以下の機能のうち1つをネストされた機能として含める必要があります

    • generate()
    • generate_stream()
    • generate_batch()
  • watsonx.ai Python クライアントライブラリを使用して、外部関数への参照を含む Python 関数を保存する場合、外部関数のスコープ(そのネストされた関数を含む)内のコードのみが保存されます。 外側の関数のスコープ外のコードは保存されないため、関数をデプロイする際には利用できません。

generate() 関数の定義に関するガイドライン

generate() 機能を使用して、認証トークンを処理することができます。 この関数は、推論エンドポイント /ml/v4/deployments/{id_or_name}/ai_service へのRESTコールを処理します。

AIサービス内で generate() 機能を定義するには、以下のガイドラインに従ってください

  • 関数を定義するには、 generate という名前を使用する必要があります。
  • generate() 関数には、 context という1つの引数しか指定できません。
  • generate() 関数は、 dict (辞書)のデータ型の値を返さなければなりません。
  • オプション :必要に応じて、 body または header キーを指定することができます。

def generate(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    return {
        "headers": {
             "Content-Type": "text/plain"
        },
        "body": "Hello WatsonX"
    }

generate_stream() 関数の定義に関するガイドライン

生成型AIのユースケースでストリーミングが必要な場合は、 generate_stream() 機能を使用できます。 この関数は、サーバー送信イベント(SSE)推論エンドポイント POST /ml/v4/deployments/{id_or_name}/ai_service_stream へのRESTコールを処理します。

AIサービス内で generate_stream() 機能を定義するには、以下のガイドラインに従ってください

  • 関数を定義するには、 generate_stream という名前を使用する必要があります。
  • generate_stream() 関数には、 context という1つの引数しか指定できません。

def generate_stream(context):
    user_token = context.get_token()
    headers = context.get_headers()
    json_body = context.get_json()

    for x in ["Hello", "WatsonX", "!"]:
        yield x

出力

id: 1
event: message
data: Hello

id: 2
event: message
data: WatsonX

id: 3
event: message
data: !

id: 4
event: eos

generate_batch() 関数の定義に関するガイドライン

generate_batch() 機能は、バッチ推論を必要とするユースケースで使用できます。 この関数は、ジョブエンドポイント /ml/v4/deployments_jobs へのREST APIコールを処理します。

AIサービス内で generate_batch() 機能を定義するには、以下のガイドラインに従ってください

  • 関数を定義するには、 generate_batch() という名前を使用する必要があります。

def generate_batch(input_data_references: list[dict], output_data_reference: dict):
    # context from outer function is visible
    batch_token = context.generate_token()
    print(f"batch_token: {batch_token[-5:]}", flush=True)
    print(
        f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
        flush=True,
    )

AIサービスを作成するためのサンプルコード

サンプルコードでは、AIサービス deployable_ai_service_f1 を定義しています。 REST API リクエストが /ml/v4/deployments エンドポイントに送信されると、 deployable_ai_service_f1 が呼び出されます。 この関数は、JSON形式の入力ペイロードを受け取り、そのスコープの一部として以下のネストされた関数を含みます

  • generate(): /ml/v4/deployments/{id_or_name}/ai_service エンドポイントにREST APIコールを実行します。 これはコンテキストオブジェクトを受け取り、トークン、ヘッダー、およびJSONボディを抽出して、JSONボディ内のモードキーに基づいて応答を返します。 レスポンスのフォーマットは、オプションのカスタムヘッダー付きのJSON、バイト、または文字列とすることができます。
  • generate_stream()SSE(サーバー送信イベント)推論エンドポイント /ml/v4/deployments/{id_or_name}/ai_service_stream にREST APIコールを実行します。 これはコンテクストオブジェクトを受け取り、トークン、ヘッダー、JSON ボディを抽出して、 eos (ストリームの終了)で示される SSE イベントのストリームを返します。
  • generate_batch(): ジョブエンドポイント /ml/v4/deployments_jobs にREST APIコールを実行します。 リクエストJSONボディから input_data_referencesoutput_data_reference を取り込み、バッチトークンを生成し、入力および出力データの参照をログに記録します。
def deployable_ai_service_f1(context, params={"k1": "v1"}, **custom):
    """
    The outer function handles the REST call to the deployment endpoint
    POST /ml/v4/deployments

        context.generate_token() - generate a token from the task credentials

    To use `generate` and `generate_stream`, the deployment has to be ONLINE
    To use `generate_batch`, the deployment has to be BATCH
    """
    task_token = context.generate_token()
    print(f"outer function: {task_token[-5:]}", flush=True)

    def generate(context) -> dict:
        """
        The `generate` function handles the REST call to the inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate function should return a dict
        The following optional keys are supported currently
        - body
        - headers

        This particular example accepts a json body of the format:
        { "mode" : <value> }

        Depending on the <value> of the mode, it will return different response

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"my_generate: {user_token=}", flush=True)
        print(f"request headers: {headers=}", flush=True)
        print(f"json body: {json_body=}", flush=True)

        match json_body.get("mode", "no-match"):
            case "json":
                # response Content-Type is "application/json"
                return {
                    "headers": {
                        "Content-Type": "application/json",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "json-no-header":
                # response Content-Type is "application/json"
                return {
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "json-custom-header":
                # response Content-Type is "text/plain; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "text/plain; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": {
                        "user_token": user_token[-5:],
                        "task_token": task_token[-5:],
                        "json_body": json_body,
                        "params": params,
                        "custom": custom,
                    },
                }
            case "bytes":
                # response Content-Type is "application/octet-stream"
                return {
                    "headers": {
                        "Content-Type": "application/octet-stream",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": b"12345678910",
                }
            case "bytes-no-header":
                # response Content-Type is 'text/html; charset=utf-8'
                return {
                    "body": b"12345678910",
                }
            case "bytes-custom-header":
                # response Content-Type is "text/plain; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "text/plain; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": b"12345678910",
                }
            case "str":
                # response Content-Type is "text/plain"
                return {
                    "headers": {
                        "Content-Type": "text/plain",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "str-no-header":
                # response Content-Type is "text/html; charset=utf-8"
                return {
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "str-custom-header":
                # response Content-Type is "application/octet-stream; charset=utf-8; test-2"
                return {
                    "headers": {
                        "Content-Type": "application/octet-stream; charset=utf-8; test-2",
                        "User-Defined-Head": "x-genai",
                    },
                    "body": f"Hello WatsonX: {json_body}",
                }
            case "negative-str-return":
                # Bad request
                return "Should give 400 bad request"
            case _:
                # response Content-Type is "text/html; charset=utf-8"
                return {"body": "No match"}

    def generate_stream(context):
        """
        The generate_stream function handles the REST call to the SSE inference endpoint
        POST /ml/v4/deployments/{id_or_name}/ai_service_stream

            context.get_token()     - get the Bearer token from the header of the request
            context.get_json()      - get the body of the request
            context.get_headers()   - get the headers of the request

        The generate_stream function be a python `generator` with yield
        The data in yield will the "data" for the SSE event

        Example: The following request json
            { "sse": ["Hello" , "", "WatsonX"," ", "!"]}
        will return the following stream of events
            --------------
            id: 1
            event: message
            data: Hello

            id: 2
            event: message
            data:

            id: 3
            event: message
            data: WatsonX

            id: 4
            event: message
            data:  

            id: 5
            event: message
            data: !

            id: 6
            event: eos
            ---------------
        The end of the stream will be marked by the event "eos"

        """
        user_token = context.get_token()
        headers = context.get_headers()
        json_body = context.get_json()
        print(f"generate_stream: {user_token=}", flush=True)
        print(f"generate_stream: {headers=}", flush=True)
        print(f"generate_stream: {json_body=}", flush=True)

        import time
        for x in json_body.get("sse", ["default", "message"]):
            time.sleep(1)
            yield x

    def generate_batch(input_data_references: list[dict], output_data_reference: dict) -> None:
        """
        The generate_batch function handles the REST jobs endpoint
        POST /ml/v4/deployments_jobs

            Arguments to the function are from the json body of the request to jobs
            - input_data_references : scoring.input_data_references
            - output_data_reference : scoring.output_data_reference

        context.generate_token() : can access context object
        from outer function scope if token is required
        """
        batch_token = context.generate_token()
        print(f"batch_token: {batch_token[-5:]}", flush=True)
        print(
            f"generate_batch:\n{input_data_references=}\n{output_data_reference=}",
            flush=True,
        )

    return generate, generate_stream, generate_batch

AIサービスのテスト

AIサービスを作成した後、 watsonx.ai Python クライアントライブラリを使用して、AIサービスのコーディングロジックをテストすることができます。

Python クライアントライブラリを使用したAIサービスのテスト

watsonx.ai Python クライアントライブラリの RuntimeContext クラスを使用して、AI サービスのロジックをローカルでテストするには、以下の手順に従います

  1. Python クライアントライブラリの RuntimeContext クラスを使用して、AI サービスをローカルでテストします

    from ibm_watsonx_ai.deployments import RuntimeContext
    
    context = RuntimeContext(
        api_client=client, request_payload_json={}
    )
    
    # custom is optional argument which is specified during the time of creation of deployment
    custom_object = {"space_id": space_id}
    
    generate, generate_stream, generate_batch = basic_generate_demo(context, **custom_object)
    
    

    詳細については、 watsonx.ai Python AIサービスで を使用するためのクライアントライブラリドキュメント RuntimeContext をご覧ください。

  2. お客様の用途に応じて、 generate()generate_stream() 、または generate_batch() の機能を以下のようにテストすることができます

    • generate() 関数をテストするには:

      context.request_payload_json = { "test": "ai_service inference payload"}
      print(generate(context))
      
    • generate_stream() 関数をテストするには:

      context.request_payload_json = {"sse": ["ai_service_stream", "inference", "test"]}
      for data in generate_stream(context):
          print(data)
      
    • generate_batch() 関数をテストするには:

      input_data_references = [
          {
              "type": "connection_asset",
              "connection": {"id": "2d07a6b4-8fa9-43ab-91c8-befcd9dab8d2"},
              "location": {
                  "bucket": "wml-v4-fvt-batch-pytorch-connection-input",
                  "file_name": "testing-123",
              },
          }
      ]
      output_data_reference = {
          "type": "data_asset",
          "location": {"name": "nb-pytorch_output.zip"},
      }
      
      generate_batch(input_data_references, output_data_reference)
      

AIサービス資産の作成

To deploy an AI service, you must create a repository asset in watsonx.ai Runtime that contains the AI service and upload the Python file to the asset.

AIサービス資産作成に必要な要件

VSCode、 Eclipse、 PyCharm, などの統合開発環境(IDE)を使用して生成型AIアプリケーションを構築する際には、AIサービスを保存するための Python ファイルを作成する必要があります。 関数を定義した後、AI サービスを圧縮して gzip アーカイブ( .gz ファイル形式)を作成する必要があります。

When you use the watsonx.ai Python client library to create your AI service asset, the library automatically stores the function in gzip archive for you. However, when you create an AI service asset by using the REST API, you must follow the process of manually compressing your Python file in a gzip archive.

You must use the runtime-24.1-py3.11 software specification to create and deploy an AI service asset that is coded in Python.

Creating AI service assets with Python client library

You can use the store_ai_service function of the watsonx.ai Python client library to create an AI service asset.

The following code sample shows how to create an AI service asset by using the Python client library:

documentation_request = {
    "application/json": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "parameters": {
                "properties": {
                    "max_new_tokens": {"type": "integer"},
                    "top_p": {"type": "number"},
                },
                "required": ["max_new_tokens", "top_p"],
            },
        },
        "required": ["query"],
    }
}

documentation_response = {
    "application/json": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "properties": {"query": {"type": "string"}, "result": {"type": "string"}},
        "required": ["query", "result"],
    }
}


meta_props = {
    client.repository.AIServiceMetaNames.NAME: "AI service example",
    client.repository.AIServiceMetaNames.DESCRIPTION: "This is AI service function",
    client.repository.AIServiceMetaNames.SOFTWARE_SPEC_ID: client.software_specifications.get_id_by_name(
        "runtime-24.1-py3.11"
    ),
    client.repository.AIServiceMetaNames.REQUEST_DOCUMENTATION: documentation_request,
    client.repository.AIServiceMetaNames.RESPONSE_DOCUMENTATION: documentation_response,
}

stored_ai_service_details = client.repository.store_ai_service(
    basic_generate_demo, meta_props
)

ai_service_id = client.repository.get_ai_service_id(stored_ai_service_details)
print("The AI service asset id:", ai_service_id)
注:
  • REQUEST_DOCUMENTATIONRESPONSE_DOCUMENTATION のパラメータはオプションです。 これらのパラメータを使用して、 generate および generate_stream 関数のリクエストとレスポンスのスキーマを保存することができます。
  • 関数呼び出し client.repository.store_ai_service は、AI サービス関数 basic_generate_demo を内部的に gzip ファイルに保存します。

詳細については、 watsonx.ai Python client library documentation for creating an AI service asset をご覧ください。

REST APIで 資産を作成する

You can use the /ml/v4/ai_services REST API endpoint to create the AI services asset in the watsonx.ai Runtime repository. 詳細については 、 watsonx.ai のREST APIドキュメントをご覧ください。

AIサービス資産の展開

お客様の用途に応じて、 デプロイメントスペースから 資産のオンラインまたはデプロイメントを作成することができます。 watsonx.ai のREST APIまたは Python のクライアントライブラリを使用して、プログラムでAIサービスを展開します。

AIサービスのデプロイメント

ユースケースに応じて、AI サービス資産をオンライン デプロイメントまたはバッチデプロイメントとしてデプロイできます。 AIサービスで使用する機能に基づいて、 デプロイメントを選択します。

  • You must create an オンライン deployment for your AI service asset for online scoring (AI service contains the generate() function) or streaming applications (AI service contains the generate_stream() function).
  • You must create a バッチ deployment for your AI service asset for batch scoring applications (AI service contains the generate_batch() function).

前提条件

  1. AI サービスを展開するには、タスクの認証情報を設定する必要があります。 wx 詳細については、 タスクの認証情報を追加するを参照のこと。
  2. AIサービスの資産 デプロイメントスペースにプロモートする必要があります。

Python クライアントライブラリを使用したAIサービスの展開

You can create an online or a batch deployment for your AI service asset by using the Python client library.

デプロイメントの作成

The following example shows how to create an online deployment for your AI service by using the watsonx.ai Python client library:

deployment_details = client.deployments.create(
    artifact_id=ai_service_id,
    meta_props={
        client.deployments.ConfigurationMetaNames.NAME: "ai-service - online test",
        client.deployments.ConfigurationMetaNames.ONLINE: {},
        client.deployments.ConfigurationMetaNames.HARDWARE_SPEC: {
            "id": client.hardware_specifications.get_id_by_name("XS")
        },
    },
)
deployment_id = client.deployments.get_uid(deployment_details)
print("The deployment id:", deployment_id)

デプロイメントの作成

The following example shows how to create a batch deployment for your AI service by using the watsonx.ai Python client library:

deployment_details = client.deployments.create(
    artifact_id=ai_service_id,
    meta_props={
        client.deployments.ConfigurationMetaNames.NAME: f"ai-service - batch",
        client.deployments.ConfigurationMetaNames.BATCH: {},
        client.deployments.ConfigurationMetaNames.HARDWARE_SPEC: {
            "id": client.hardware_specifications.get_id_by_name("XS")
        },
    },
)
deployment_id = client.deployments.get_uid(deployment_details)
print("The batch deployment id:", deployment_id)

REST API による AI サービスの展開

You can use the /ml/v4/deployments watsonx.ai REST API endpoint to create an online or a batch deployment for your AI service asset.

デプロイメントの作成

次の例では、REST API を使用して AI サービスのオンラインデプロイメントを作成する方法を示します

# POST /ml/v4/deployments
response = requests.post(
    f'{HOST}/ml/v4/deployments?version={VERSION}',
    headers=headers,
    verify=False,
    json={
        "space_id": space_id,
        "name": "genai flow online",
        "custom": {
            "key1": "value1",
            "key2": "value2",
            "model": "meta-llama/llama-3-8b-instruct"
        },
        "asset": {
            "id": asset_id
        },
        "online": {}
    }
)

デプロイメントの作成

次の例では、REST API を使用して AI サービスのデプロイメントを作成する方法を示します

response = requests.post(
    f'{HOST}/ml/v4/deployments?version={VERSION}',
    headers=headers,
    verify=False,
    json={
        "hardware_spec": {
          "id": "........",
          "num_nodes": 1
        },
        "space_id": space_id,
        "name": "ai service batch dep",
        "custom": {
            "key1": "value1",
            "key2": "value2",
            "model": "meta-llama/llama-3-8b-instruct"
        },
        "asset": {
            "id": asset_id
        },
        "batch": {}
    }
)
print(f'POST {HOST}/ml/v4/deployments?version={VERSION}', response.status_code)
print(json.dumps(response.json(), indent=2))

dep_id = response.json()["metadata"]["id"]

print(f"{dep_id=}")

AIサービスの作成と展開のためのサンプルノートブック

AI サービスをプログラムで作成および展開する方法については、以下のサンプルノートブックを参照してください

AIサービス用サンプルノートブック
サンプル名 フレームワーク 示される手法
watsonx と meta-llama/llama-3-2-11b-vision-instruct を使用してAIサービスとして実行する Python 環境設定
AIサービスを作成
AIサービスの機能をローカルでテスト
AIサービスをデプロイ
AIサービスの実行
watsonx、 Elasticsearch、 LangChain を使用して質問に回答する(RAG) LangChain 環境設定
テスト用データセットをダウンロードします。
watsonx で基盤モデルを定義する
Elasticsearch への接続情報を設定する
質問に対する検索強化応答を生成する
AIサービスの作成
AIサービス機能をローカルでテストする
AIサービスの導入
watsonx と meta-llama/llama-3-1-70b-instruct を使用してAIサービスを作成する LangGraph 環境設定
AIサービスを作成
AIサービスの機能をローカルでテスト
AIサービスをデプロイ
AIサービスの実行

親トピック: コードによるAIサービスの展開