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

1. はじめに

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

ユースケース

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

Spanner から取得したデータ
ML.PREDICT を使用してアパレルデータ用に生成され、Spanner に保存されたベクトル
Dataflow ジョブとワークフロージョブを使用してベクトル検索と統合された Spanner ベクトルデータ
ユーザーが入力した入力の類似性一致を見つけるために実行されたベクトル検索

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

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

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

インデックスでユーザークエリを実行する:

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

アーキテクチャの概要

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

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

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

作成するアプリの概要

Spanner からベクトルインデックス:

ソースデータと対応するエンベディングを保存して管理する Spanner データベース
データを（ID とエンベディング）Vertex AI ベクトル検索データベースに一括アップロードするワークフロージョブ。
インデックスから関連する商品説明を検索するために使用される Vector Search API。

インデックスでユーザークエリを実行する:

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

仕組み

ユーザーがアパレルのテキスト説明を入力すると、ウェブアプリケーションはその説明を 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（Google Cloud で動作するコマンドライン環境）を使用します。この環境には gcloud がプリロードされています。gcloud コマンドとその使用方法については、ドキュメントをご覧ください。プロジェクトが設定されていない場合は、次のコマンドを使用して設定します。

gcloud config set project <YOUR_PROJECT_ID>

アクティブな Google Cloud プロジェクトで Cloud Spanner ページに移動して、開始します。

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

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

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 バケットを作成する

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

Cloud Workflow を設定する

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

空のインデックスを作成する:

Vector Search インデックスが、Cloud Storage バケットとデータと同じリージョンにあることを確認します。インデックスの管理ページの [バッチ更新用のインデックスを作成する] セクションの [コンソール] タブに記載されている 11 個の手順に沿って操作します。contentsDeltaUri に渡されるフォルダに、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（エンベディング）から Vector Search インデックスにデータをエクスポートするワークフローを設定するには、次のロールが必要です。

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

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

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

Dataflow ジョブをトリガーする: Dataflow 管理者、Dataflow ワークフロー。

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

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

Vertex AI ベクトル検索の再構築をトリガーするには: Vertex AI ユーザー。

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

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

Dataflow を管理する: Dataflow 管理者、Dataflow ワーカー。Spanner からデータを読み取るには、Cloud Spanner Database 読み取りロールが必要です。選択した GCS Container Registry に対する書き込みアクセス権: GCS Storage バケットオーナー。

Cloud Workflow をデプロイする

ワークフロー 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 の [実行] タブに表示されます。これにより、データがベクトル検索データベースに読み込まれ、インデックスが作成されます。

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

5. ベクトル検索インデックスをデプロイする

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

次の手順に沿ってインデックスをデプロイします。

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

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

6. フロントエンド: ユーザーデータをベクトル検索に送信します。

gradio を利用した UX を備えたシンプルな Python アプリケーションを構築して、実装をすばやくテストしましょう。こちらの実装を参照して、このデモアプリを独自の colab ノートブックに実装できます。

Embeddings API の呼び出しと Vector Search インデックスエンドポイントの呼び出しには、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 個の最も近い一致結果を表示します。

ベクトル検索の次のメソッド定義では、find_neighbors メソッドが呼び出されて、10 個の最近傍ベクトルが特定されます。

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 メソッドの呼び出しも確認できます。次のステップで見ていきましょう。

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

!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 ベクトル検索の実装が完了しました。

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

実装を独自のユースケースに拡張したり、現在のユースケースを新機能で即興的に変更したりしてもかまいません。Spanner の ML 機能の詳細については、こちらをご覧ください。