例：Vertex AI モデルの統合

以下のドキュメントでは、Google の Vertex AI でホストされているモデルへのカスタム接続の例として、設定とモデルアダプターを提供しています。

段階的なガイドについては、モデルアダプターの作成方法と外部ホストされたモデルへの接続の作成方法のドキュメントを参照してください。

例：Vertex AI タブラー型モデルアダプター

まず、Code Repositories アプリケーション内のモデルアダプターライブラリを使用して、モデルアダプターを公開してタグ付けします。以下のモデルアダプターは、Vertex AI SDK for Python およびフレームワークを使用して、Vertex AI でホストされているモデルへの接続を設定します。以下のコードは、Python 3.8.17、pandas 1.5.3、 google-cloud-aiplatform 1.32.0、google-auth 2.23.0、google-auth-oauthlib 1.1.0 のバージョンでテストされました。

このモデルアダプターは以下の仮定を置いています：

このモデルアダプターは、提供されるデータがタブラー型であることを前提としています。
このモデルアダプターは、入力データを JSON にシリアライズし、このデータをホストされた Vertex AI モデルに送信します。
このモデルアダプターは、レスポンスが JSON から pandas データフレームにデシリアライズ可能であることを前提としています。
このモデルアダプターは、Vertex AI クライアントを構築するために 4 つの入力を取ります。
- region_name - 接続設定として提供されます
- project_id - 接続設定として提供されます
- endpoint_id - 接続設定として提供されます
- google_application_credentials - 資格情報として提供されます

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import palantir_models as pm
import models_api.models_api_executable as executable_api

from google.oauth2 import service_account
from google.cloud import aiplatform
from google.api_core import exceptions

import json
import pandas as pd
import logging
from typing import Optional

logger = logging.getLogger(__name__)


class VertexAITabularAdapter(pm.ExternalModelAdapter):
  """
  :display-name: Vertex AI Tabular Model Adapter
  :description: Vertex AIモデルのデフォルトのモデルアダプター。タブラー形式の入力と出力データを想定しています。
  """

  def __init__(self, project_id, region_name, endpoint_id, google_application_credentials):
    self.endpoint_id = endpoint_id
    # google_application_credentialsはGoogleが提供する秘密鍵ファイルの有効な文字列表現であることが期待されます
    credentials = service_account.Credentials.from_service_account_info(
      json.loads(google_application_credentials),
      scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    aiplatform.init(project=project_id, location=region_name, credentials=credentials)

  @classmethod
  def init_external(cls, external_context) -> "pm.ExternalModelAdapter":
    project_id = external_context.external_model.connection_configuration["project_id"]
    region_name = external_context.external_model.connection_configuration["region_name"]
    endpoint_id = external_context.external_model.connection_configuration["endpoint_id"]
    google_application_credentials = external_context.resolved_credentials["google_application_credentials"]
    return cls(
      project_id,
      region_name,
      endpoint_id,
      google_application_credentials
    )

  @classmethod
  def api(cls):
    inputs = {"df_in": pm.Pandas()}
    outputs = {"df_out": pm.Pandas()}
    return inputs, outputs

  def predict(self, df_in):
    instances = df_in.to_dict(orient='records')
    try:
      endpoint = aiplatform.Endpoint(endpoint_name=self.endpoint_id)
    except ValueError as error:
      logger.error("エンドポイントオブジェクトの初期化でエラーが発生しました。入力されたendpoint_id、project_id、region_nameを再確認してください。")
      raise error
    try:
      # モデルからの出力はjsonシリアライズ可能と仮定されます
      # 結果が実行者にとって大きすぎる場合、これはOOMを引き起こす可能性があります
      prediction_result = endpoint.predict(instances=instances)
    except exceptions.Forbidden as error:
      logger.error("推論の実行中にエラーが発生しました。提供されたgoogle_application_credentialsには十分な権限がありません。")
      raise error
    except exceptions.BadRequest as error:
      logger.error("推論の実行中にエラーが発生しました。入力データフレームを再確認してください。")
      raise error
    return pd.json_normalize(prediction_result.predictions)