Pesquisa por similaridade com o Spanner e a Vertex AI

1. Introdução

Os avanços recentes no aprendizado profundo possibilitaram representar textos e outros dados de uma forma que captura o significado semântico. Isso levou a uma nova abordagem de pesquisa, chamada pesquisa vetorial, que usa representações vetoriais de texto (conhecidas como embeddings) para encontrar os documentos mais relevantes para a consulta de um usuário. A pesquisa vetorial é preferível à pesquisa tradicional em aplicativos como a pesquisa de vestuário, em que os usuários geralmente procuram itens pela descrição, estilo ou contexto, em vez de nomes exatos de produtos ou marcas. Podemos integrar o banco de dados do Cloud Spanner à Pesquisa vetorial para realizar a correspondência de similaridade de vetores. Ao usar o Spanner e a pesquisa vetorial juntos, os clientes podem criar uma integração eficiente que combina a disponibilidade, a confiabilidade e a escalonabilidade do Spanner com os recursos avançados de pesquisa por similaridade da pesquisa vetorial da Vertex AI. Essa pesquisa é realizada comparando embeddings de itens no índice da pesquisa vetorial e retornando as correspondências mais semelhantes.

Caso de uso

Imagine que você é um cientista de dados em uma loja de moda tentando acompanhar as tendências, as pesquisas de produtos e as recomendações que mudam rapidamente. O desafio é que você tem recursos limitados e silos de dados. Esta postagem do blog mostra como implementar um caso de uso de recomendação de vestuário usando uma abordagem de pesquisa de similaridade em dados de vestuário.As seguintes etapas são abordadas:

Dados extraídos do Spanner
Vetores gerados para os dados de vestuário usando ML.PREDICT e armazenados no Spanner
Dados vetoriais do Spanner integrados à pesquisa vetorial usando jobs do Dataflow e de fluxo de trabalho
Pesquisa vetorial realizada para encontrar uma correspondência de similaridade para a entrada inserida pelo usuário

Vamos criar um aplicativo da Web de demonstração para realizar pesquisas de roupas com base na entrada do usuário. O aplicativo permite que os usuários pesquisem roupas inserindo uma descrição de texto.

Spanner para índice de pesquisa vetorial:

Os dados da pesquisa de vestuário são armazenados no Spanner. Vamos invocar a API Embeddings da Vertex AI na estrutura ML.PREDICT diretamente dos dados do Spanner. Em seguida, vamos usar os jobs do Dataflow e do Workflow que fazem upload em massa desses dados (inventário e embeddings) para a pesquisa vetorial da Vertex AI e atualizam o índice.

Executar consultas do usuário no índice:

Quando um usuário insere uma descrição de roupa, o app gera os embeddings em tempo real usando a API Text Embeddings. Em seguida, ela é enviada como entrada para a API Vector Search, que encontra 10 descrições de produtos relevantes no índice e mostra a imagem correspondente.

Visão geral da arquitetura

A arquitetura do aplicativo Spanner-Vector Search é mostrada no diagrama de duas partes a seguir:

Spanner para índice de pesquisa vetorial:

App cliente para executar consultas do usuário no índice:

O que você vai criar

Spanner para índice vetorial:

Banco de dados do Spanner para armazenar e gerenciar dados de origem e os embeddings correspondentes.
Um job do Workflow que faz upload em massa de dados (ID e embeddings) no banco de dados da Vertex AI Vector Search.
Uma API Vector Search usada para encontrar descrições de produtos relevantes no índice.

Executar consultas do usuário no índice:

Um aplicativo da Web que permite aos usuários inserir descrições de texto de roupas, realiza uma pesquisa de similaridade usando o endpoint de índice implantado e retorna as roupas mais próximas da entrada.

Como funciona

Quando um usuário insere uma descrição textual de uma peça de roupa, o aplicativo da Web envia a descrição para a API Vector Search. Em seguida, a API Vector Search usa os embeddings das descrições de roupas para encontrar as descrições de produtos mais relevantes no índice. As descrições e imagens correspondentes são exibidas para o usuário. O fluxo de trabalho geral é o seguinte:

Gere embeddings para dados armazenados no Spanner.
Exporte e faça upload de embeddings para um índice da Pesquisa de vetor.
Consulte o índice da Pesquisa de vetores para encontrar itens semelhantes realizando uma pesquisa de vizinho mais próximo.

2. Requisitos

Use um navegador, como o Chrome ou o Firefox.
Tenha um projeto do Google Cloud com o faturamento ativado.

Antes de começar

No console do Google Cloud, na página de seletor de projetos, selecione ou crie um projeto do Google Cloud.
Verifique se o faturamento está ativado para seu projeto do Cloud. Saiba como verificar se o faturamento está ativado em um projeto.
Verifique se todas as APIs necessárias (Cloud Spanner, Vertex AI, Google Cloud Storage) estão ativadas.
Você vai usar o Cloud Shell, um ambiente de linha de comando executado no Google Cloud que vem pré-carregado com gcloud. Consulte a documentação para ver o uso e os comandos gcloud. Se o projeto não estiver definido, use este comando:

gcloud config set project <YOUR_PROJECT_ID>

Navegue até a página Cloud Spanner com seu projeto ativo do Google Cloud para começar.

3. Back-end: crie sua fonte de dados e incorporações do Spanner

Neste caso de uso, o banco de dados do Spanner armazena o inventário de roupas com as imagens e descrições correspondentes. Gere embeddings para a descrição do texto e armazene-os no banco de dados do Spanner como ARRAY<float64>.

Criar os dados do Spanner

Crie uma instância chamada "spanner-vertex" e um banco de dados chamado "spanner-vertex-embeddings". Crie uma tabela usando a 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);

Inserir dados na tabela usando o SQL INSERT

Os scripts de inserção para dados de amostra estão disponíveis aqui.

Criar modelo de embeddings de texto

Isso é necessário para que possamos gerar embeddings para o conteúdo na entrada. Confira abaixo o DDL para o mesmo:

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

Gerar embeddings de texto para os dados de origem

Crie uma tabela para armazenar os embeddings e insira os embeddings gerados. Em um aplicativo de banco de dados do mundo real, o carregamento de dados no Spanner até a etapa 2 seria transacional. Para manter as práticas recomendadas de design intactas, prefiro manter as tabelas transacionais normalizadas, criando uma tabela separada para incorporações.

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

Agora que o conteúdo e os embeddings em massa estão prontos, vamos criar um índice e um endpoint de pesquisa vetorial para armazenar os embeddings que vão ajudar a realizar a pesquisa vetorial.

4. Trabalho do fluxo de trabalho: exportação de dados do Spanner para a Pesquisa de vetores

Criar um bucket do Cloud Storage

Isso é necessário para armazenar embeddings do Spanner em um bucket do GCS em um formato JSON que a pesquisa vetorial espera como entrada. Crie um bucket na mesma região dos seus dados no Spanner. Crie uma pasta dentro, se necessário, mas principalmente crie um arquivo vazio chamado empty.json nela.

Configurar o Cloud Workflow

Para configurar uma exportação em lote do Spanner para um índice da pesquisa vetorial da Vertex AI:

Criar um índice vazio:

Verifique se o índice de pesquisa vetorial está na mesma região que o bucket do Cloud Storage e os dados. Siga as 11 etapas de instrução na guia do console na seção Criar um índice para atualização em lote na página "Gerenciar índices". Na pasta transmitida para contentsDeltaUri, crie um arquivo vazio chamado "empty.json", porque não é possível criar um índice sem ele. Isso cria um índice vazio.

Se você já tiver um índice, pule esta etapa. O fluxo de trabalho vai substituir seu índice.

Observação: não é possível implantar um índice vazio em um endpoint. Por isso, vamos adiar a etapa de implantação em um endpoint para depois de exportar os dados de vetor para o Cloud Storage.

Clone este repositório git: há várias maneiras de clonar um repositório git. Uma delas é executar o seguinte comando usando a CLI do GitHub. Execute os dois comandos abaixo no terminal do Cloud Shell:

gh repo clone cloudspannerecosystem/spanner-ai

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

Essa pasta contém dois arquivos

batch-export.yaml: esta é a definição do fluxo de trabalho.
sample-batch-input.json: este é um exemplo dos parâmetros de entrada do fluxo de trabalho.

Configurar input.json do arquivo de amostra:primeiro, copie o JSON de amostra.

cp sample-batch-input.json input.json

Em seguida, edite input.json com os detalhes do seu projeto. Nesse caso, o JSON vai ficar assim:

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

Configurar permissões

Para ambientes de produção, recomendamos criar uma conta de serviço e conceder a ela um ou mais papéis do IAM que contenham as permissões mínimas necessárias para gerenciar o serviço. Os seguintes papéis são necessários para configurar o fluxo de trabalho de exportação de dados do Spanner (embeddings) para o índice da Pesquisa vetorial:

Conta de serviço do Cloud Workflow:

Por padrão, ele usa a conta de serviço padrão do Compute Engine.

Se você usar uma conta de serviço configurada manualmente, inclua os seguintes papéis:

Para acionar um job do Dataflow: Administrador do Dataflow, worker do Dataflow

Para personificar uma conta de serviço do worker do Dataflow: Usuário da conta de serviço.

Para gravar registros: Gravador de registros.

Para acionar a recriação da pesquisa de vetor da Vertex AI: usuário da Vertex AI

Conta de serviço do worker do Dataflow:

Se você usar uma conta de serviço configurada manualmente, inclua os seguintes papéis:

Para gerenciar o Dataflow: Administrador do Dataflow, Worker do Dataflow. Para ler dados do Spanner: Leitor de banco de dados do Cloud Spanner. Acesso de gravação ao Container Registry do GCS selecionado: Proprietário do bucket de armazenamento do GCS.

Implantar o fluxo de trabalho do Cloud

Implante o arquivo YAML de fluxo de trabalho no seu projeto na nuvem do Google. É possível configurar a região ou o local em que o fluxo de trabalho será executado.

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"

O fluxo de trabalho agora vai aparecer na página "Fluxos de trabalho" no console do Google Cloud.

Observação: também é possível criar e implantar o fluxo de trabalho no console do Google Cloud. Siga as instruções no console do Cloud. Para a definição do fluxo de trabalho, copie e cole o conteúdo de batch-export.yaml.

Quando isso for concluído, execute o fluxo de trabalho para iniciar a exportação de dados.

Executar o fluxo de trabalho do Cloud

Execute o comando a seguir para executar o fluxo de trabalho:

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

A execução vai aparecer na guia "Execuções" do Workflows. Isso vai carregar e indexar seus dados no banco de dados da pesquisa de vetor.

Observação: também é possível executar no console usando o botão "Executar". Siga as instruções e, para a entrada, copie e cole o conteúdo do seu input.json personalizado.

5. Implantar o índice de pesquisa de vetor

Implantar o índice em um endpoint

Siga as etapas abaixo para implantar o índice:

Na página Índices de pesquisa de vetor, você vai encontrar um botão "IMPLANTAR" ao lado do índice que acabou de criar na etapa 2 da seção anterior. Como alternativa, navegue até a página de informações do índice e clique no botão "IMPLANTAR NO ENDPOINT".
Forneça as informações necessárias e implante o índice em um endpoint.

Outra opção é consultar este notebook para implantar em um endpoint (pule para a parte de implantação do notebook). Depois da implantação, anote o ID do índice implantado e o URL do endpoint.

6. Front-end: dados do usuário para a pesquisa de vetor

Vamos criar um aplicativo simples em Python com uma UX baseada em gradio para testar rapidamente nossa implementação. Consulte a implementação aqui para implementar esse app de demonstração no seu próprio notebook colab.

Vamos usar o SDK aiplatform do Python para chamar a API Embeddings e invocar o endpoint do índice da Pesquisa vetorial.

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

Vamos usar o Gradio para demonstrar o aplicativo de IA que estamos criando de forma rápida e fácil com uma interface do usuário. Reinicie o ambiente de execução antes de implementar esta etapa.

!pip install gradio
import gradio as gr

No web app, com base na entrada do usuário, invoque a API Embeddings. Vamos usar o modelo de embedding de texto: textembedding-gecko@latest

O método abaixo invoca o modelo de embedding de texto e retorna os embeddings de vetor para o texto inserido pelo usuário:

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

Realizar o teste

text_embedding("red shorts for girls")

Você vai receber uma saída semelhante à abaixo. A imagem está cortada na altura para que não seja possível ver toda a resposta do vetor.

Declarar o ID do índice implantado e o ID do endpoint

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

Defina o método de pesquisa vetorial para chamar o endpoint do índice e mostrar o resultado com as 10 correspondências mais próximas para a resposta de incorporação correspondente ao texto de entrada do usuário.

Na definição do método abaixo para a pesquisa vetorial, observe que o método "find_neighbors" é invocado para identificar os 10 vetores mais próximos.

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

Você também vai notar a chamada para o método spanner_read_data. Vamos analisar isso na próxima etapa.

Defina a implementação do método de leitura de dados do Spanner que invoca o método "execute_sql" para extrair as imagens correspondentes aos IDs dos vetores de vizinho mais próximo retornados da última etapa.

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

Ele vai retornar os URLs das imagens correspondentes aos vetores escolhidos.

Por fim, vamos juntar as peças em uma interface do usuário e acionar o processo de pesquisa de vetor.

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)

O resultado vai aparecer como mostrado abaixo:

Imagem: link

Assista o vídeo de resultado aqui.

7. Limpar

Para evitar cobranças na sua conta do Google Cloud pelos recursos usados nesta postagem, siga estas etapas:

No console do Google Cloud, acesse a página Gerenciar recursos.
Na lista de projetos, selecione o projeto que você quer excluir e clique em Excluir.
Na caixa de diálogo, digite o ID do projeto e clique em "Encerrar" para excluí-lo.
Se você não quiser excluir o projeto, exclua a instância do Spanner. Para isso, navegue até a instância que você acabou de criar para este projeto e clique no botão "EXCLUIR INSTÂNCIA" no canto superior direito da página de visão geral da instância.
Você também pode navegar até o índice da Pesquisa de vetor, desimplantar o endpoint e o índice e excluir o índice.

8. Conclusão

Parabéns! Você concluiu a implementação da Pesquisa vetorial do Spanner - Vertex ao

Criação de fonte de dados e incorporações do Spanner para aplicativos originados do banco de dados do Spanner.
Criando o índice do banco de dados da Pesquisa de vetor.
Integração de dados de vetores do Spanner à pesquisa de vetores usando jobs do Dataflow e do Workflow.
Implantação do índice em um endpoint.
Por fim, invocando a pesquisa vetorial na entrada do usuário em uma implementação do SDK da Vertex AI com tecnologia Python.

Você pode estender a implementação para seu próprio caso de uso ou improvisar o caso de uso atual com novos recursos. Saiba mais sobre os recursos de machine learning do Spanner aqui.