使用 Spanner 和 Vertex AI 进行相似性搜索

1. 简介

深度学习方面的最新进展使得我们能够以可捕捉语义含义的方式表示文本和其他数据。这催生了一种新的搜索方法，称为向量搜索，该方法使用文本的向量表示形式（称为嵌入）来查找与用户查询最相关的文档。对于服装搜索等应用，向量搜索比传统搜索更受欢迎，因为用户通常会根据商品说明、款式或情境来搜索商品，而不是根据确切的产品或品牌名称来搜索。我们可以将 Cloud Spanner 数据库与 Vector Search 集成，以执行向量相似度匹配。通过将 Spanner 与 Vector Search 搭配使用，客户可以创建强大的集成解决方案，将 Spanner 的可用性、可靠性和可伸缩性与 Vertex AI Vector Search 的高级相似度搜索功能相结合。此搜索通过比较 Vector Search 索引中商品的嵌入来执行，并返回最相似的匹配项。

使用场景

假设您是某时尚零售商的数据科学家，正努力跟上快速变化的趋势、产品搜索和推荐。但难点在于，您的资源有限，并且存在数据孤岛。这篇博文演示了如何使用服装数据的相似性搜索方法来实现服装推荐用例。其中涵盖了以下步骤：

数据源自 Spanner
使用 ML.PREDICT 为服装数据生成的向量，并存储在 Spanner 中
使用 Dataflow 和工作流作业将 Spanner 向量数据与 Vector Search 集成
执行 Vector Search 以查找用户输入内容的相似匹配项

我们将构建一个演示 Web 应用，以根据用户输入的文本执行服装搜索。该应用允许用户通过输入文字说明来搜索服装。

Spanner 到 Vector Search 索引：

服装搜索的数据存储在 Spanner 中。我们将直接从 Spanner 数据中调用 ML.PREDICT 构造中的 Vertex AI Embeddings API。然后，我们将利用 Dataflow 和 Workflow 作业将这些数据（商品目录和嵌入）批量上传到 Vertex AI 的 Vector Search 并刷新索引。

在索引上运行用户查询：

当用户输入服装说明时，应用会使用文本嵌入 API 实时生成嵌入。然后，将此内容作为输入发送到 Vector Search API，以从索引中查找 10 个相关的商品说明，并显示相应的图片。

架构概览

Spanner-Vector Search 搜索应用的架构如下图所示（分为两部分）：

从 Spanner 到 Vector Search 索引：

用于在索引上运行用户查询的客户端应用：

构建内容

从 Spanner 到向量索引：

用于存储和管理源数据及相应嵌入的 Spanner 数据库
一种将数据（ID 和嵌入）批量上传到 Vertex AI Vector Search 数据库的工作流作业。
一种 Vector Search API，用于从索引中查找相关商品描述。

在索引上运行用户查询：

一个 Web 应用，允许用户输入服装的文字说明，使用已部署的索引端点执行相似性搜索，并返回与输入最接近的服装。

具体用法

当用户输入服装的文字说明时，Web 应用会将该说明发送到 Vector Search API。然后，Vector Search API 会使用服装说明的嵌入从索引中找到相关度最高的产品说明。然后，系统会向用户显示商品说明和对应的图片。一般工作流程如下：

为存储在 Spanner 中的数据生成嵌入。
将嵌入导出并上传到 Vector Search 索引。
通过执行最近邻搜索，查询 Vector Search 索引以查找类似商品。

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 页面，开始使用 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');

为源数据生成文本嵌入

创建一个用于存储嵌入的表，并插入生成的嵌入。在实际的数据库应用中，将数据加载到 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

创建 Cloud Storage 存储分区

这是必需的，用于以向量搜索预期作为输入的 JSON 格式将 Spanner 中的嵌入存储在 GCS 存储分区中。在 Spanner 中与您的数据相同的区域中创建存储分区。如果需要，可以在其中创建一个文件夹，但主要是在其中创建一个名为 empty.json 的空文件。

设置 Cloud Workflow

如需设置从 Spanner 到 Vertex AI Vector Search 索引的批量导出，请执行以下操作：

创建空索引：

确保 Vector Search 索引与 Cloud Storage 存储分区和数据位于同一区域。按照管理索引页面上为批量更新创建索引部分中的控制台选项卡下的 11 个步骤进行操作。在传递给 contentsDeltaUri 的文件夹中，创建一个名为 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 Admin、Dataflow Worker。

如需模拟 Dataflow 工作器服务账号，您需要拥有：服务账号用户角色。

写入日志：Logs Writer。

如需触发 Vertex AI Vector Search 重建，请执行以下操作：Vertex AI 用户。

Dataflow Worker 服务账号：

如果您使用手动配置的服务账号，则必须添加以下角色：

如需管理 Dataflow，请使用 Dataflow Admin 和 Dataflow Worker 角色。如需从 Spanner 读取数据，请使用 Cloud Spanner Database Reader。对所选 GCS Container Registry 的写入权限：GCS 存储分区所有者。

部署 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 中的“执行”标签页中。这应会将您的数据加载到 Vector Search 数据库并为其编制索引。

注意：您还可以使用“执行”按钮从控制台执行代码。按照提示操作，然后复制并粘贴自定义 input.json 的内容作为输入。

5. 部署 Vector Search 索引

将索引部署到端点

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

在向量搜索索引页面上，您应该会在刚刚在上一部分的第 2 步中创建的索引旁边看到一个“部署”按钮。或者，您也可以前往索引信息页面，然后点击“部署到端点”按钮。
提供必要的信息，并将索引部署到端点。

或者，您也可以查看此笔记本，了解如何将其部署到端点（跳到笔记本的部署部分）。部署完成后，请记下已部署的索引 ID 和端点网址。

6. 前端：用户数据到 Vector Search

我们来构建一个简单的 Python 应用，其中包含由 Gradio 提供支持的用户体验，以便快速测试我们的实现：您可以参考此处的实现，在自己的 Colab 笔记本中实现此演示版应用。

我们将使用 aiplatform 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()

我们将使用 Gradio 演示如何通过用户界面快速轻松地构建 AI 应用。在实现此步骤之前，请先重启运行时。

!pip install gradio
import gradio as gr

在用户输入后，从 Web 应用调用 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')

定义 Vector Search 方法以调用索引端点，并显示与用户输入文本对应的嵌入响应的 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 的调用。我们将在下一步中介绍。

定义 Spanner 读取数据方法实现，该实现会调用 execute_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)

它应返回与所选向量对应的图片的网址。

最后，让我们将各个部分整合到用户界面中，并触发 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 账号收取费用，请按照以下步骤操作：

在 Google Cloud 控制台中，前往管理资源页面。
在项目列表中，选择要删除的项目，然后点击“删除”。
在对话框中输入项目 ID，然后点击“关停”以删除项目。
如果您不想删除项目，请删除 Spanner 实例，方法是前往您为此项目刚刚创建的实例，然后点击实例概览页面右上角的“删除实例”按钮。
您还可以前往 Vector Search 索引，取消部署端点和索引，然后删除索引。

8. 总结

恭喜！您已成功完成 Spanner - Vertex Vector Search 实现，

为源自 Spanner 数据库的应用创建 Spanner 数据源和嵌入内容。
创建 Vector Search 数据库索引。
使用 Dataflow 和 Workflow 作业将 Spanner 中的向量数据集成到 Vector Search 中。
将索引部署到端点。
最后，在基于 Python 的 Vertex AI SDK 实现中，针对用户输入调用 Vector Search。

您可以根据自己的使用情形随意扩展实现，也可以通过新功能即兴创作当前的使用情形。如需详细了解 Spanner 的机器学习功能，请点击此处。