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

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

generate() — 同期推論

generate(): /ml/v4/deployments/{id_or_name}/ai_service エンドポイントにREST APIコールを実行します。 これはコンテキストオブジェクトを受け取り、トークン、ヘッダー、およびJSONボディを抽出して、JSONボディ内のモードキーに基づいて応答を返します。 レスポンスのフォーマットは、オプションのカスタムヘッダー付きのJSON、バイト、または文字列とすることができます。

リクエストJSONには必要な値を任意に含めることができます。例えば：

{ "content-type": "<value>" }

モードは応答形式を決定します。 異なるコンテンツタイプ向けの簡略版については、最小限の例を含む以下のサンプルコードを参照してください：

def deployable_ai_service_f1(context, params={"k1": "v1"}):
    """
    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:
        # context is mandatory parameter for ai service
        """
        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 request header
        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:
          { "content-type" : <value> }
        Depending on the <value> of the content type, it will return different responses.
        """
        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("content-type", "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:],  # updated from gen_token
                        "json_body": json_body,
                        "params": params,
                    },
                }

適用可能なその他の事例：

Content-Type: json-no-header:

case "json-no-header":
    return {
        "body": {
            "json_body": json_body,
            "user_token": user_token[-5:],
            "task_token": task_token[-5:],
        }
    }

Content-Type: json-custom-header:

case "json-custom-header":
    return {
        "headers": {"Content-Type": "text/plain; charset=utf-8"},
        "body": {
            "json_body": json_body,
            "user_token": user_token[-5:],
            "task_token": task_token[-5:],
        },
    }

Content-Type: バイト:

case "bytes":
    return {
        "headers": {"Content-Type": "application/octet-stream"},
        "body": b"12345678910",
    }

Content-Type: str:

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

Content-Type: ネガティブ文字列リターン:

case "negative-str-return":
    return "Should give 400 bad request"

デフォルトケース:

case _:
    return {"body": "No match"}

generate_stream() — SSEストリーミング

generate_stream()SSE（サーバー送信イベント）推論エンドポイント /ml/v4/deployments/{id_or_name}/ai_service_stream にREST APIコールを実行します。 これはコンテクストオブジェクトを受け取り、トークン、ヘッダー、JSON ボディを抽出して、 eos （ストリームの終了）で示される SSE イベントのストリームを返します。

generate_stream() 「sse」リストの各要素をyieldでストリーム処理する:

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 request header
    context.get_json()    - get the body of the request
    context.get_headers() - get the headers of the request

    The generate_stream function must be a Python `generator` with yield.
    The data in `yield` will be 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

generate_batch() — バッチ処理

generate_batch(): ジョブエンドポイント /ml/v4/deployments_jobs にREST APIコールを実行します。 リクエストJSONボディから input_data_references と output_data_reference を取り込み、バッチトークンを生成し、入力および出力データの参照をログに記録します。

generate_batch() 入力と出力の参照を受け取り、それらをログに記録します:

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 a 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

REST APIで 資産を作成する

/ml/v4/ai_services REST API エンドポイントを使用して、 watsonx.ai Runtime リポジトリに AI サービス資産作成できます。 詳細については 、 watsonx.ai のREST APIドキュメントをご覧ください。

デプロイメントの作成

次の例では、 watsonx.ai Python クライアント ライブラリを使用して、AI サービスのオンラインデプロイメント作成する方法を示します：

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)

デプロイメントの作成

次の例では、 watsonx.ai Python クライアント ライブラリを使用して、AI サービスのバッチデプロイメント作成する方法を示します：

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 サービスのオンラインデプロイメントを作成する方法を示します

# 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": {}
    }
)