搭配 Spanner 和 Vertex AI 進行相似搜尋

1. 簡介

近年來深度學習技術的進步，現在能以擷取語意含意的方式呈現文字和其他資料。這個轉變形成了一種全新的搜尋方法，稱為向量搜尋，其透過文字向量表示法 (稱為嵌入) 來尋找與使用者查詢最相關的文件。相較於傳統搜尋應用程式 (例如服飾搜尋)，Vector Search 更偏好根據描述、風格或情境來搜尋商品，而非使用確切產品或品牌名稱進行搜尋。我們能夠整合 Cloud Spanner 資料庫與 Vector Search，以執行向量相似度比對。將 Spanner 和 Vector Search 搭配使用，客戶就能創造強大的整合體驗，結合 Spanner 的可用性、可靠性與規模，以及 Vertex AI Vector Search 的進階相似度搜尋功能。這項搜尋方法是比較 Vector Search 索引中項目的「嵌入」，然後傳回最相似的相符項目。

用途

假設您是時尚零售商的數據資料學家，正設法跟上瞬息萬變的趨勢、產品搜尋和推薦內容。難題在於您的資源與資料孤島有限。這篇網誌文章示範如何使用服飾資料相似度的搜尋方法，導入服飾推薦的應用實例。以下步驟將說明以下步驟：

1. 資料來源：Spanner
2. 使用 ML.PREDICT 針對服飾資料產生的向量，並儲存在 Spanner 中
3. 使用 Dataflow 和工作流程工作將 Spanner 向量資料與 Vector Search 整合
4. 已執行 Vector Search，尋找使用者輸入內容的相似度比對結果

我們會建立示範網頁應用程式，根據使用者輸入的文字執行服飾搜尋。使用者可在此應用程式輸入文字說明，搜尋服飾相關資訊。

從 Spanner 到 Vector Search 索引：

服飾搜尋資料會儲存在 Spanner 中。我們會在 ML.PREDICT 中，直接透過 Spanner 資料叫用 Vertex AI Embeddings API。接著，我們會運用 Dataflow 和工作流程工作，將這些資料 (庫存和嵌入) 大量上傳至 Vertex AI 的 Vector Search，並重新整理索引。

對索引執行使用者查詢：

使用者輸入服飾說明時，應用程式會使用 Text Embeddings API 即時產生嵌入項目。接著，系統會將這個錯誤做為輸入內容傳送至 Vector Search API，從索引中尋找 10 則相關產品說明，並顯示對應的圖片。

架構總覽

以下 2 個部分圖表顯示 Spanner - Vector Search 應用程式的架構：

從 Spanner 到 Vector Search 索引：

在索引上執行使用者查詢的用戶端應用程式：

建構項目

Spanner 到向量索引：

• 用於儲存及管理來源資料和對應嵌入的 Spanner 資料庫
• 可將資料 (ID 和嵌入) 大量上傳至 Vertex AI Vector Search 資料庫的工作流程工作。
• 用於從索引中尋找相關產品說明的 Vector Search API。

對索引執行使用者查詢：

• 這個網頁應用程式可讓使用者輸入服飾文字說明、使用已部署的索引端點執行相似度搜尋，並將最近期的服飾傳回至輸入內容。

運作方式

使用者輸入服飾的文字說明時，網頁應用程式會將說明傳送至 Vector Search API。接著，Vector Search API 會使用服飾說明的嵌入項目，從索引中找出最相關的產品說明。接著，使用者就會看到產品說明和對應的圖片。一般工作流程如下所示：

1. 為 Spanner 中儲存的資料產生嵌入。
2. 匯出嵌入並上傳至 Vector Search 索引。
3. 執行最鄰近搜尋，在 Vector Search 索引查詢類似項目。

2. 需求條件

• 瀏覽器，例如 Chrome 或 Firefox
• 已啟用計費功能的 Google Cloud 專案

事前準備

1. 前往 Google Cloud 控制台的專案選取器頁面，選取或建立 Google Cloud 專案
2. 確認 Cloud 專案已啟用計費功能。瞭解如何檢查專案是否已啟用帳單功能
3. 確認已啟用所有必要的 API，包括 Cloud Spanner、Vertex AI、Google Cloud Storage
4. 您將使用 Cloud Shell，這是在 Google Cloud 中執行的指令列環境，已預先載入 gcloud。如要查看 gcloud 指令和使用方式，請參閱說明文件。如果尚未設定專案，請使用下列指令進行設定：

gcloud config set project <YOUR_PROJECT_ID>

5. 前往「Cloud Spanner」頁面，即可開始使用有效的 Google Cloud 專案

3. 後端：建立 Spanner 資料來源和嵌入

在此用途中，Spanner 資料庫包含服飾目錄、對應的圖片和說明。請務必產生文字說明的嵌入，並將其以 ARRAY<float64> 儲存在 Spanner 資料庫中。

1. 建立 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);

2. 使用 INSERT SQL 將資料插入資料表

您可以從這裡插入範例資料的指令碼。

3. 建立文字嵌入模型

此為必要步驟，我們才能為輸入內容產生嵌入內容。以下是相同資料的 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');

4. 為來源資料產生文字嵌入

建立資料表來儲存嵌入，並插入產生的嵌入。在實際資料庫應用程式中，Spanner 至步驟 2 的資料載入作業會是交易性質。為了維持設計最佳做法，我偏好將交易資料表保持正規化，因此請另外建立嵌入資料表。

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

現在大量內容和嵌入已準備就緒，讓我們建立 Vector Search 索引和端點來儲存有助於執行 Vector Search 的嵌入。

4. 工作流程工作：將 Spanner 資料匯出至 Vector Search

1. 建立 Cloud Storage 值區

這會以 Vector Search 預期做為輸入內容的 JSON 格式，以 JSON 格式將 Spanner 的嵌入儲存至 GCS 值區。在與 Spanner 資料相同的區域建立值區。視需要在資料夾中建立資料夾，但主要會在其中建立名為 empty.json 的空白檔案。

2. 設定 Cloud Workflow

如要設定從 Spanner 匯出至 Vertex AI Vector Search 索引的批次匯出作業，請按照下列步驟操作：

建立空白索引：

確認 Vector Search 索引與 Cloud Storage 值區和資料位於相同的區域。按照「管理索引」頁面中的「為批次更新建立索引」部分，按照控制台分頁的 11 個步驟操作。在傳遞至 contentDeltaUri 的資料夾中，建立一個空白檔案 (稱為 empty.json)，因為您需要這個檔案才能建立索引。這會建立一個空白索引。

如果您已有索引，可以略過這個步驟。工作流程會覆寫索引。

注意：您無法將空白索引部署至端點。因此，將向量資料匯出至 Cloud Storage 後，我們將延遲部署至端點的步驟。

複製這個 Git 存放區：您可以透過多種方式複製 Git 存放區，其中一種方法是使用 GitHub CLI 執行下列指令。透過 Cloud Shell 終端機執行下列 2 指令：

gh repo clone cloudspannerecosystem/spanner-ai

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

這個資料夾包含兩個檔案

• 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>>"
  }
}

設定權限

在實際工作環境中，強烈建議您建立新的服務帳戶，並授予一或多個 IAM 角色 (含有管理服務所需的最低權限)。需要下列角色來設定工作流程，以便將資料從 Spanner (嵌入) 匯出至 Vector Search 索引：

Cloud Workflow 服務帳戶：

預設會使用 Compute Engine 預設服務帳戶。

如果您使用手動設定的服務帳戶，就必須加入下列角色：

如要觸發 Dataflow 工作：Dataflow 管理員、Dataflow 工作站。

模擬 Dataflow 工作站服務帳戶：服務帳戶使用者。

如要寫入記錄：記錄寫入者。

如要觸發 Vertex AI Vector Search 重新建構程序：請參閱 Vertex AI User。

Dataflow 工作站服務帳戶：

如果您使用手動設定的服務帳戶，就必須加入下列角色：

管理 Dataflow：Dataflow 管理員、Dataflow Worker。如要從 Spanner 讀取資料：Cloud Spanner 資料庫讀取者。具備所選 GCS Container Registry 的寫入權限：GCS Storage 值區擁有者。

3. 部署 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 控制台的「Workflows」頁面中。

注意：您也可以透過 Google Cloud 控制台建立及部署工作流程。按照 Cloud 控制台中的提示操作。針對工作流程定義，複製及貼上 batch-export.yaml 的內容。

完成後，請執行工作流程，資料匯出作業就會開始。

4. 執行 Cloud Workflow

執行下列指令來執行工作流程：

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

執行作業應會顯示在 Workflows 的「 Executions」分頁中。這樣應該就會將資料載入 Vector Search 資料庫並建立索引。

注意：您也可以從控制台使用「執行」按鈕執行。按照提示和輸入欄位，複製及貼上自訂 input.json 的內容。

5. 部署 Vector Search 索引

將索引部署至端點

您可以按照下列步驟部署索引：

1. 在「Vector Search 索引」頁面中，您應該會在上一節步驟 2 建立的索引旁看到「DEPLOY」按鈕。或者，您也可以瀏覽索引資訊頁面，並按一下「DEPLOY TO ENDPOINT」按鈕。
2. 提供必要資訊，並將索引部署至端點。

或者，您也可以查看這個筆記本，將其部署至端點 (略過筆記本的部署部分)。部署完成後，請記下已部署的索引 ID 和端點網址。

6. 前端：將使用者資料轉移至 Vector Search

讓我們建構一個以 gRadio 提供使用者體驗的簡易 Python 應用程式，快速測試我們的實作結果：您可以參考這裡的實作方式，在專屬的 colab 筆記本中實作這個示範應用程式。

1. 我們會使用 AI Platform Python SDK 呼叫 Embeddings API，以及叫用 Vector Search 索引端點。

# [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()

2. 我們會使用 gRadio 來展示我們透過使用者介面，輕鬆快速地建構的 AI 應用程式。請先重新啟動執行階段，再執行這個步驟。

!pip install gradio
import gradio as gr

3. 在使用者輸入內容時叫用 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")

您應該會看到類似下方的輸出內容 (請注意，圖片高度會遭到裁剪，因此您無法看見整個向量回應)：

4. 宣告已部署的索引 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')

5. 定義 Vector Search 方法以呼叫索引端點，並顯示與使用者輸入內容相對應的嵌入回應最接近 10 個相符項目的結果。

在下列 Vector Search 方法定義中，請注意，系統會叫用 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 方法的呼叫。下一個步驟將說明。

6. 定義 Spanner 讀取資料方法的實作方式，叫用 run_sql 方法時，擷取與最後一個步驟中最鄰近向量的 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)

應該會傳回與所選向量相對應的圖片網址。

7. 最後，讓我們把各部分放到使用者介面中，並觸發 Vector Search 程序

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 帳戶收取本文中所用資源的費用，請按照下列步驟操作：

1. 在 Google Cloud 控制台中，前往「管理資源」頁面。
2. 在專案清單中選取要刪除的專案，然後按一下「刪除」。
3. 在對話方塊中輸入專案 ID，然後按一下「關閉」以刪除專案。
4. 如果您不想刪除專案，請前往您剛為這項專案建立的執行個體，然後按一下執行個體總覽頁面右上角的「刪除執行個體」按鈕，即可刪除 Spanner 執行個體。
5. 您也可以查看 Vector Search 索引、取消部署端點和索引，以及刪除索引。

8. 結語

恭喜！您已成功透過以下服務完成 Spanner - Vertex Vector Search 實作程序：

1. 針對取自 Spanner 資料庫的應用程式建立 Spanner 資料來源和嵌入。
2. 正在建立 Vector Search 資料庫索引。
3. 使用 Dataflow 和工作流程工作，將向量資料從 Spanner 整合至 Vector Search。
4. 正在將索引部署至端點。
5. 最後，以採用 Python 技術的 Vertex AI SDK 實作方式，在使用者輸入上叫用 Vector Search。

您可以依據自己的用途延伸導入作業，或利用新功能改善目前的應用實例。如要進一步瞭解 Spanner 的機器學習功能，請按這裡。