Esta página foi traduzida pela API Cloud Translation.

Pesquisa por similaridade com o Spanner e a Vertex AI

1. Introdução

Com os avanços recentes no aprendizado profundo, é possível representar textos e outros dados de uma forma que capture o significado semântico. Isso levou a uma nova abordagem de pesquisa, chamada de pesquisa vetorial, que usa representações vetoriais de texto (conhecidas como embeddings) para encontrar documentos mais relevantes para a consulta do usuário. A pesquisa vetorial é preferível em relação à pesquisa tradicional para aplicativos como pesquisa de vestuário, em que os usuários geralmente pesquisam itens pela descrição, estilo ou contexto, e não pelo produto exato ou nome da marca. Podemos integrar o banco de dados do Cloud Spanner à pesquisa de vetores para fazer a correspondência por semelhança de vetores. Ao usar o Spanner e a Pesquisa Vetorial juntos, os clientes podem criar uma integração poderosa que combina a disponibilidade, a confiabilidade e a escala do Spanner e os recursos avançados de pesquisa por similaridade da Pesquisa de vetores 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ê é cientista de dados de um varejista de moda que está tentando acompanhar tendências, pesquisas de produtos e recomendações em constante mudança. O desafio é que você tem recursos e silos de dados limitados. Esta postagem do blog demonstra como implementar um caso de uso de recomendação de vestuário usando a abordagem de pesquisa por similaridade nos dados de vestuário.As etapas a seguir são abordadas:

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

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

Índice da Pesquisa de vetor do Spanner:

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

Execução de consultas de usuário no índice:

Quando um usuário insere a descrição de uma peça de vestuário, o app gera os embeddings em tempo real usando a API Text Embeddings. Em seguida, ele é enviado como entrada para a API Vector Search para encontrar 10 descrições relevantes de produtos no índice e exibir a imagem correspondente.

Visão geral da arquitetura

A arquitetura do aplicativo de pesquisa de vetores do Spanner é mostrada no diagrama de duas partes a seguir:

Índice do Spanner para pesquisa vetorial:

App cliente para executar consultas de usuários no índice:

O que você criará

Spanner para índice de vetores:

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

Execução de consultas de usuário no índice:

Um aplicativo da Web que permite aos usuários inserir descrições textuais de vestuário, realiza pesquisas de similaridade usando o endpoint do índice implantado e retorna as roupas mais próximas à entrada.

Como funciona

Quando o usuário insere uma descrição em texto de uma peça de vestuário, 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 vestuário para encontrar as descrições de produtos mais relevantes no índice. As descrições do produto e as imagens correspondentes são exibidas para o usuário. O fluxo de trabalho geral é o seguinte:

gerar embeddings para dados armazenados no Spanner;
Exportar e fazer upload de embeddings para um índice da Pesquisa Vetorial.
Consultar o índice da Pesquisa Vetorial em busca de itens semelhantes executando 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 do 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.
Verificar se todas as APIs necessárias (Cloud Spanner, Vertex AI, Google Cloud Storage) estão ativadas
Você usará o Cloud Shell, um ambiente de linha de comando em execução no Google Cloud que vem pré-carregado com o gcloud. Consulte a documentação para saber mais sobre os comandos e o uso da gcloud. Se o projeto não estiver definido, use este comando:

gcloud config set project <YOUR_PROJECT_ID>

Para começar, navegue até a página do Cloud Spanner com seu projeto ativo do Google Cloud

3. Back-end: crie a fonte de dados e os embeddings do Spanner

Nesse 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 o 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 com o comando INSERT SQL

Para acessar os scripts de inserção de dados de amostra, clique aqui.

Criar um modelo de embeddings de texto

Isso é necessário para gerar embeddings do 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 inserir aqueles gerados. Em um aplicativo de banco de dados 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, então crie uma tabela separada para embeddings.

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 em massa e os embeddings estão prontos, vamos criar um índice de pesquisa vetorial e um endpoint para armazenar os embeddings que vão ajudar a realizar a pesquisa de vetores.

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

Crie um bucket do Cloud Storage

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

Configurar o Cloud Workflow

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

Crie 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ções na guia do console da seção Criar um índice para atualização em lote na página de gerenciamento de índices. Na pasta que é transmitida para sumDeltaUri, crie um arquivo vazio chamado empty.json porque você não poderia criar um índice sem esse arquivo. Isso cria um índice vazio.

Se você já tem um índice, pule esta etapa. O fluxo de trabalho vai substituir o í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 uma etapa posterior, depois de exportar os dados vetoriais para o Cloud Storage.

Clone este repositório git: há várias maneiras de clonar um repositório git. Uma delas é executar o comando a seguir 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

Esta pasta contém dois arquivos

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

Configure 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 detalhes do seu projeto. Nesse caso, seu JSON precisa 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 nova conta de serviço e conceder a ela um ou mais papéis do IAM com as permissões mínimas necessárias para gerenciar o serviço. Os papéis a seguir são necessários para configurar o fluxo de trabalho e exportar dados do Spanner (embeddings) para o índice da Pesquisa Vetorial:

Conta de serviço do Cloud Workflow:

Ele usa a conta de serviço padrão do Compute Engine.

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

Para acionar um job do Dataflow: Administrador do Dataflow, Worker do Dataflow.

Para representar uma conta de serviço do worker do Dataflow: Service Account User.

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 de worker do Dataflow:

Se você usar uma conta de serviço configurada manualmente, precisará incluir 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 no GCS Container Registry selecionado: Proprietário do bucket do GCS Storage.

Implantar o fluxo de trabalho do Cloud

Implante o arquivo yaml do fluxo de trabalho no projeto do Google Cloud. É possível configurar a região ou o local em que o fluxo de trabalho será executado quando 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 ficará visível 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.

Depois disso, execute o fluxo de trabalho para iniciar a exportação de dados.

Execute o fluxo de trabalho do Cloud

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

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

A execução aparece na guia "Execuções" do app Workflows. Isso deve carregar seus dados no banco de dados da Pesquisa Vetorial e indexá-los.

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

5. Implantar o índice da Pesquisa Vetorial

Implantar o índice em um endpoint

Siga as etapas abaixo para implantar o índice:

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

Como alternativa, você pode examinar este notebook para implantá-lo em um endpoint (pule para a parte de implantação do notebook). Após a implantação, anote o ID do índice implantado e o URL do endpoint.

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

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

Usaremos o SDK do Python aiplatform para chamar a API Embeddings e também para 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 grádio para demonstrar o aplicativo de IA que estamos criando de maneira 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 app da Web após a entrada do usuário, invoque a API Embeddings. Vamos usar o modelo de incorporação de texto: textembedding-gecko@latest

O método abaixo invoca o modelo de embedding de texto e retorna os embeddings vetoriais 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")

O resultado será semelhante ao mostrado abaixo (como a imagem está cortada na altura, não é possível ver toda a resposta vetorial):

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 de vetores para chamar o endpoint do índice e mostrar o resultado com as 10 correspondências mais próximas para a resposta de embedding correspondente ao texto de entrada do usuário.

Na definição de método abaixo para Pesquisa de vetor, 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 notará a chamada ao método spanner_read_data. Vamos analisar isso na próxima etapa.

Definir 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 vizinhos mais próximos 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)

Ela deve 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 Vetorial

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 será exibido abaixo:

Imagem: link

Assista ao vídeo com o 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 não quiser excluir o projeto, exclua a instância do Spanner acessando a que você acabou de criar para o 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 Vetorial, cancelar a implantação do endpoint e do índice e excluir o índice.

8. Conclusão

Parabéns! Você concluiu a implementação do Spanner: pesquisa de vetores da Vertex até

Criar fontes de dados e embeddings do Spanner para aplicativos provenientes do banco de dados do Spanner.
Criando um índice do banco de dados da Pesquisa Vetorial.
Integrar dados vetoriais do Spanner à pesquisa de vetores usando o Dataflow e jobs do Workflows.
Implantando o índice em um endpoint.
Por fim, invocar a pesquisa de vetores na entrada do usuário em uma implementação com tecnologia Python do SDK da Vertex AI.

Fique à vontade para estender a implementação ao seu caso de uso ou improvisar o caso de uso atual com novos recursos. Saiba mais sobre os recursos de machine learning do Spanner aqui.