このページは Cloud Translation API によって翻訳されました。

Spanner と Vertex AI を使用した類似検索

1. はじめに

ディープラーニングの最近の進歩により、テキストやその他のデータをセマンティックな意味を捉えて表現できるようになりました。これにより、ベクトル検索と呼ばれる新しい検索アプローチが生まれました。ベクトル検索は、テキストのベクトル表現（エンべディングと呼ばれる）を使用して、ユーザーのクエリに最も関連性の高いドキュメントを見つけます。ベクトル検索は、アパレル検索などのアプリケーションで従来の検索よりも好まれます。アパレル検索では、ユーザーは正確な商品名やブランド名ではなく、説明、スタイル、コンテキストでアイテムを検索することがよくあります。Cloud Spanner データベースとベクトル検索を統合して、ベクトル類似性マッチングを実行できます。Spanner とベクトル検索を併用することで、Spanner の可用性、信頼性、スケールと、Vertex AI Vector Search の高度な類似検索機能を組み合わせたパワフルなインテグレーションを実現できます。この検索は、ベクトル検索インデックス内のアイテムのエンベディングを比較し、最も類似した一致を返すことによって実行されます。

ユースケース

あなたはファッション小売業者のデータサイエンティストで、急速に変化するトレンド、商品検索、レコメンデーションに対応しようとしているとします。課題は、限られたリソースとデータサイロしかないことです。このブログ投稿では、アパレルデータに対して類似検索アプローチを使用して、アパレルレコメンデーションのユースケースを実装する方法について説明します。手順は次のとおりです。

Spanner をソースとするデータ
ML.PREDICT を使用してアパレルデータ用に生成され、Spanner に保存されるベクトル
Dataflow ジョブとワークフロージョブを使用してベクトル検索と統合した Spanner ベクトルデータ
ユーザー入力との類似度の一致を検出するためにベクトル検索が実行されました

ユーザー入力テキストに基づいてアパレル検索を実行するデモウェブアプリケーションを構築します。このアプリケーションでは、ユーザーがテキストの説明を入力してアパレルを検索できます。

Spanner からベクトル検索へのインデックス:

アパレル検索のデータは Spanner に保存されます。ML.PREDICT コンストラクトの Vertex AI Embeddings API を、Spanner データから直接呼び出します。次に、Dataflow ジョブとワークフロージョブを活用して、このデータ（インベントリとエンベディング）を Vertex AI のベクトル検索に一括アップロードし、インデックスを更新します。

インデックスに対するユーザークエリの実行:

ユーザーがアパレルの説明を入力すると、アプリは Text Embeddings API を使用してリアルタイムでエンべディングを生成します。これを Vector Search API に入力として送信し、インデックスから関連する 10 個の商品説明を検索し、対応する画像を表示します。

アーキテクチャの概要

Spanner ベクトル検索アプリケーションのアーキテクチャを、次の 2 部構成の図に示します。

Spanner からベクトル検索へのインデックス:

インデックスに対してユーザークエリを実行するクライアントアプリ:

作成する内容

Spanner から Vector へのインデックス:

ソースデータと対応するエンべディングを保存および管理するための Spanner データベース
データ（ID とエンベディング）を Vertex AI Vector Search データベースに一括アップロードするワークフロージョブ。
Vector Search API の 1 つ。インデックスから関連する商品の説明を見つけるために使用されます。

インデックスに対するユーザークエリの実行:

ユーザーがアパレル商品の説明テキストを入力できるウェブアプリケーション。デプロイされたインデックスエンドポイントを使用して類似性検索を実行し、最も近いアパレルを入力に返します。

仕組み

ユーザーがアパレル商品の説明を入力すると、ウェブアプリケーションがその説明を Vector Search API に送信します。次に、Vector Search API がアパレル商品説明のエンベディングを使用して、インデックスから最も関連性の高い商品説明を見つけます。商品の説明と対応する画像がユーザーに表示されます。一般的なワークフローは次のとおりです。

Spanner に保存されたデータのエンベディングを生成する。
エンベディングをベクトル検索インデックスにエクスポートしてアップロードする。
最近傍探索を実行して、ベクトル検索のインデックスに対してクエリを実行し、類似アイテムの検索を行います。

2. 必要なもの

ブラウザ（Chrome、Firefox など）
課金を有効にした Google Cloud プロジェクト

始める前に

Google Cloud コンソールの [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。
Cloud プロジェクトに対して課金が有効になっていることを確認します。詳しくは、プロジェクトで課金が有効になっているかどうかを確認する方法をご覧ください。
必要なすべての API（Cloud Spanner、Vertex AI、Google Cloud Storage）が有効になっていることを確認します。
Cloud Shell を使用します。Cloud Shell は Google Cloud で実行されるコマンドライン環境で、gcloud があらかじめ組み込まれています。gcloud のコマンドと使用方法については、ドキュメントをご覧ください。プロジェクトが設定されていない場合は、次のコマンドを使用して設定します。

gcloud config set project <YOUR_PROJECT_ID>

開始するには、アクティブな Google Cloud プロジェクトの [Cloud Spanner] ページに移動します。

3. バックエンド: Spanner データソースとエンベディングを作成する

このユースケースでは、Spanner データベースに、対応する画像と説明を含むアパレルの在庫が格納されます。テキストによる説明のエンベディングを生成し、ARRAY<float64> として Spanner データベースに保存してください。

Spanner データを作成する

「spanner-vertex」という名前のインスタンスを作成します。「spanner-vertex-embeddings」という名前のデータベースを使用します。DDL を使用してテーブルを作成します。

CREATE TABLE
  apparels ( id NUMERIC,
    category STRING(100),
    sub_category STRING(50),
    uri STRING(200),
    content STRING(2000),
    embedding ARRAY<FLOAT64>
    )
PRIMARY KEY
  (id);

INSERT SQL を使用してテーブルにデータを挿入する

サンプルデータの挿入スクリプトについては、こちらをご覧ください。

テキストエンベディングモデルを作成する

これは、入力のコンテンツに対するエンベディングを生成できるようにするために必要です。同じ場合の DDL は次のとおりです。

CREATE MODEL text_embeddings INPUT(content STRING(MAX))
OUTPUT(
  embeddings
    STRUCT<
      statistics STRUCT<truncated BOOL, token_count FLOAT64>,
      values ARRAY<FLOAT64>>
)
REMOTE OPTIONS (
  endpoint = '//aiplatform.googleapis.com/projects/abis-345004/locations/us-central1/publishers/google/models/textembedding-gecko');

ソースデータのテキストエンベディングを生成する

エンベディングを保存するテーブルを作成し、生成されたエンベディングを挿入します。実際のデータベースアプリケーションでは、ステップ 2 までの Spanner へのデータ読み込みはトランザクションアプリケーションになります。設計のベストプラクティスを維持するために、トランザクションテーブルを正規化しておき、エンベディング用に別のテーブルを作成することをおすすめします。

CREATE TABLE apparels_embeddings (id string(100), embedding ARRAY<FLOAT64>)
PRIMARY KEY (id);

INSERT INTO apparels_embeddings(id, embeddings) 
SELECT CAST(id as string), embeddings.values
FROM ML.PREDICT(
  MODEL text_embeddings,
  (SELECT id, content from apparels)
) ;

一括コンテンツとエンベディングの準備ができたら、ベクトル検索のインデックスとエンドポイントを作成して、ベクトル検索の実行に役立つエンベディングを保存します。

4. ワークフロージョブ: ベクトル検索への Spanner データのエクスポート

Cloud Storage バケットを作成する

これは、Spanner からのエンベディングを、ベクトル検索で入力として想定される JSON 形式で GCS バケットに保存するために必要です。Spanner のデータと同じリージョンにバケットを作成します。必要に応じてフォルダ内にフォルダを作成しますが、主に empty.json という空のファイルを作成します。

Cloud ワークフローの設定

Spanner から Vertex AI Vector Search インデックスへのバッチエクスポートを設定するには:

空のインデックスを作成します。

ベクトル検索インデックスが Cloud Storage バケットおよびデータと同じリージョンにあることを確認します。[インデックスの管理] ページの [バッチアップデート用のインデックスの作成] セクションの [コンソール] タブで、11 の手順に沿って操作します。このファイルがないとインデックスは作成できないため、contentDeltaUri に渡されるフォルダに empty.json という空のファイルを作成します。これにより、空のインデックスが作成されます。

すでにインデックスがある場合は、この手順をスキップできます。ワークフローによってインデックスが上書きされます。

注: 空のインデックスをエンドポイントにデプロイすることはできません。そのため、ベクトルデータを Cloud Storage にエクスポートした後、エンドポイントにデプロイするステップを後のステップに延期します。

この Git リポジトリのクローンを作成する: Git リポジトリのクローンを作成する方法は複数あります。1 つの例として、GitHub CLI を使用して次のコマンドを実行します。Cloud Shell ターミナルから次の 2 つのコマンドを実行します。

gh repo clone cloudspannerecosystem/spanner-ai

cd spanner-ai/vertex-vector-search/workflows

このフォルダには 2 つのファイルが含まれています

batch-export.yaml: これはワークフローの定義です。
sample-batch-input.json: これは、ワークフロー入力パラメータのサンプルです。

サンプルファイルから input.json を設定する: まず、サンプルの JSON をコピーします。

cp sample-batch-input.json input.json

次に、input.json を編集してプロジェクトの詳細を追加します。この場合、JSON は次のようになります。

{
  "project_id": "<<YOUR_PROJECT>>",
  "location": "<<us-central1>>",
  "dataflow": {
    "temp_location": "gs://<<YOUR_BUCKET>>/<<FOLDER_IF_ANY>>/workflow_temp"
  },
  "gcs": {
    "output_folder": "gs://<<YOUR_BUCKET>>/<<FOLDER_IF_ANY>>/workflow_output"
  },
  "spanner": {
    "instance_id": "spanner-vertex",
    "database_id": "spanner-vertex-embeddings",
    "table_name": "apparels_embeddings",
    "columns_to_export": "embedding,id"
  },
  "vertex": {
    "vector_search_index_id": "<<YOUR_INDEX_ID>>"
  }
}

権限を設定する

本番環境では、新しいサービスアカウントを作成し、サービスの管理に必要な最小限の権限を含む 1 つ以上の IAM ロールを付与することを強くおすすめします。Spanner（エンベディング）からベクトル検索インデックスにデータをエクスポートするワークフローを設定するには、次のロールが必要です。

Cloud Workflow サービスアカウント:

デフォルトでは、Compute Engine のデフォルトのサービスアカウントが使用されます。

手動で構成されたサービスアカウントを使用する場合は、次のロールを含める必要があります。

Dataflow ジョブをトリガーする: Dataflow 管理者、Dataflow ワーカー

Dataflow ワーカーサービスアカウントの権限を借用するには: サービスアカウントユーザー

ログを書き込むには: ログ書き込み。

Vertex AI Vector Search の再構築をトリガーするには、Vertex AI ユーザーを使用します。

Dataflow ワーカーサービスアカウント:

手動で構成されたサービスアカウントを使用する場合は、次のロールを含める必要があります。

Dataflow を管理する: Dataflow 管理者、Dataflow ワーカーSpanner からデータを読み取る: Cloud Spanner データベース読み取り。選択した GCS Container Registry に対する書き込みアクセス権: GCS ストレージバケットオーナー。

Cloud ワークフローをデプロイする

ワークフロー yaml ファイルを Google Cloud プロジェクトにデプロイします。実行時にワークフローが実行されるリージョンまたはロケーションを構成できます。

gcloud workflows deploy vector-export-workflow --source=batch-export.yaml --location="us-central1" [--service account=<service_account>]

or 

gcloud workflows deploy vector-export-workflow --source=batch-export.yaml --location="us-central1"

これで、Google Cloud コンソールの [ワークフロー] ページにワークフローが表示されます。

注: ワークフローは Google Cloud コンソールから作成してデプロイすることもできます。Cloud コンソールのプロンプトに沿って操作します。ワークフロー定義のため、batch-export.yaml の内容をコピーして貼り付けます。

完了したら、データのエクスポートが開始されるようにワークフローを実行します。

Cloud Workflow を実行する

次のコマンドを実行してワークフローを実行します。

gcloud workflows execute vector-export-workflow --data="$(cat input.json)"

実行内容が Workflows の [Executions] タブに表示されます。これにより、データがベクトル検索データベースに読み込まれ、インデックスが作成されます。

注: [実行] ボタンを使用して、コンソールから実行することもできます。画面の指示に沿って入力し、カスタマイズした input.json の内容をコピーして貼り付けます。

5. ベクトル検索インデックスのデプロイ

インデックスをエンドポイントにデプロイする

インデックスをデプロイする手順は次のとおりです。

[ベクトル検索のインデックス] ページで、前のセクションのステップ 2 で作成したインデックスの横に [デプロイ] ボタンが表示されます。または、インデックスの情報ページに移動して、[エンドポイントにデプロイ] ボタンをクリックします。
必要な情報を入力して、インデックスをエンドポイントにデプロイします。

このノートブックを参照してエンドポイントにデプロイすることもできます（ノートブックのデプロイ部分に進みます）。デプロイしたら、デプロイしたインデックス ID とエンドポイント URL をメモします。

6. フロントエンド: ベクトル検索へのユーザーデータ

gradio を活用した UX を使用して簡単な Python アプリケーションを構築し、実装を迅速にテストしましょう。このデモアプリを独自の colab ノートブックに実装するには、こちらの実装を参照してください。

Embeddings API とベクトル検索のインデックスエンドポイントの呼び出しには、aiplatform Python SDK を使用します。

# [START aiplatform_sdk_embedding]
!pip install google-cloud-aiplatform==1.35.0 --upgrade --quiet --user


import vertexai
vertexai.init(project=PROJECT_ID, location="us-central1")


from vertexai.language_models import TextEmbeddingModel


import sys
if "google.colab" in sys.modules:
    # Define project information
    PROJECT_ID = " "  # Your project id
    LOCATION = " "  # Your location 


    # Authenticate user to Google Cloud
    from google.colab import auth
    auth.authenticate_user()

gradio を使用して、ユーザーインターフェースを使用して迅速かつ簡単に構築している AI アプリケーションのデモを行います。このステップを実装する前にランタイムを再起動します。

!pip install gradio
import gradio as gr

ユーザー入力時にウェブアプリから Embeddings API を呼び出し、テキストエンベディングモデル（textembedding-gecko@latest）を使用する

以下のメソッドは、テキストエンベディングモデルを呼び出して、ユーザーが入力したテキストのベクトルエンベディングを返します。

def text_embedding(content) -> list:
    """Text embedding with a Large Language Model."""
    model = TextEmbeddingModel.from_pretrained("textembedding-gecko@latest")
    embeddings = model.get_embeddings(content)
    for embedding in embeddings:
        vector = embedding.values
        #print(f"Length of Embedding Vector: {len(vector)}")
    return vector

テスト

text_embedding("red shorts for girls")

次のような出力が表示されます（ただし、画像の高さがトリミングされるため、ベクトルレスポンス全体は表示されません）。

デプロイされるインデックス ID とエンドポイント ID を宣言する

from google.cloud import aiplatform
DEPLOYED_INDEX_ID = "spanner_vector1_1702366982123"

#Vector Search Endpoint
index_endpoint = aiplatform.MatchingEngineIndexEndpoint('projects/273845608377/locations/us-central1/indexEndpoints/2021628049526620160')

インデックスエンドポイントを呼び出すベクトル検索メソッドを定義して、ユーザー入力テキストに対応するエンベディングレスポンスに最も近い 10 件の結果を表示します。

以下のベクトル検索のメソッド定義では、10 個の最近傍のベクトルを識別するために find_neighbors メソッドが呼び出されています。

def vector_search(content) -> list:
  result = text_embedding(content)
  #call_vector_search_api(content)
  index_endpoint = aiplatform.MatchingEngineIndexEndpoint('projects/273845608377/locations/us-central1/indexEndpoints/2021628049526620160')
  # run query
  response = index_endpoint.find_neighbors(
      deployed_index_id = DEPLOYED_INDEX_ID,
      queries = [result],
      num_neighbors = 10
  )
  out = []
  # show the results
  for idx, neighbor in enumerate(response[0]):
      print(f"{neighbor.distance:.2f} {spanner_read_data(neighbor.id)}")
      out.append(f"{spanner_read_data(neighbor.id)}")
  return out

spanner_read_data メソッドの呼び出しにも注目してください。次のステップで見てみましょう。

run_sql メソッドを呼び出す Spanner 読み取りデータメソッドの実装を定義して、前のステップで返された最近傍ベクトルの ID に対応する画像を抽出します。

!pip install google-cloud-spanner==3.36.0


from google.cloud import spanner


instance_id = "spanner-vertex"
database_id = "spanner-vertex-embeddings"
projectId = PROJECT_ID
client = spanner.Client()
client.project = projectId
instance = client.instance(instance_id)
database = instance.database(database_id)
def spanner_read_data(id):
    query = "SELECT uri FROM apparels where id = " + id
    outputs = []
    with database.snapshot() as snapshot:
        results = snapshot.execute_sql(query)


        for row in results:
            #print(row)
            #output = "ID: {}, CONTENT: {}, URI: {}".format(*row)
            output = "{}".format(*row)
            outputs.append(output)


    return "\n".join(outputs)

選択したベクトルに対応する画像の URL が返されます。

最後に、ユーザーインターフェース上でこれらの要素をまとめ、ベクトル検索プロセスをトリガーしましょう。

from PIL import Image


def call_search(query):
  response = vector_search(query)
  return response


input_text = gr.Textbox(label="Enter your query. Examples: Girls Tops White Casual, Green t-shirt girls, jeans shorts, denim skirt etc.")
output_texts = [gr.Image(label="") for i in range(10)]
demo = gr.Interface(fn=call_search, inputs=input_text, outputs=output_texts, live=True)
resp = demo.launch(share = True)

次のような結果が表示されます。

画像: リンク

結果の動画はこちらでご覧いただけます。

7. クリーンアップ

この投稿で使用したリソースについて、Google Cloud アカウントに課金されないようにするには、次の操作を行います。

Google Cloud コンソールで、[リソースの管理] ページに移動します。
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。
プロジェクトを削除しない場合は、このプロジェクト用に作成したインスタンスに移動し、インスタンスの概要ページの右上隅にある [インスタンスを削除] ボタンをクリックして Spanner インスタンスを削除します。
ベクトル検索のインデックスに移動し、エンドポイントとインデックスのデプロイ解除、インデックスの削除を行うこともできます。

8. おわりに

これで、「Spanner - Vertex Vector Search」の実装を

Spanner データベースをソースとするアプリケーションの Spanner データソースとエンベディングを作成する。
ベクトル検索データベースインデックスを作成しています。
Dataflow とワークフロージョブを使用して、Spanner からベクトル検索にベクトルデータを統合する。
エンドポイントにインデックスをデプロイしています。
最後に、Python を利用した Vertex AI SDK の実装で、ユーザー入力でベクトル検索を呼び出します。

独自のユースケースに実装を拡張することも、新機能を追加して現在のユースケースに即応することも可能です。Spanner の ML 機能について詳しくは、こちらをご覧ください。