Se usó la API de Cloud Translation para traducir esta página.

Similarity Search con Spanner y Vertex AI

1. Introducción

Los avances recientes en el aprendizaje profundo hicieron posible la representación de texto y otros datos de una manera que capte el significado semántico. Esto ha llevado a un nuevo enfoque para la búsqueda, llamado búsqueda de vectores, que usa representaciones vectoriales de texto (conocidas como incorporaciones) para encontrar los documentos más relevantes para la consulta de un usuario. La búsqueda de vectores es preferible a la búsqueda tradicional en aplicaciones como la búsqueda de indumentaria, donde los usuarios suelen buscar artículos por su descripción, estilo o contexto en lugar de por los nombres exactos de productos o marcas. Podemos integrar la base de datos de Cloud Spanner a la búsqueda de vectores para realizar una coincidencia de similitud de vectores. Cuando los clientes usan Spanner y la búsqueda de vectores en conjunto, los clientes pueden crear una integración potente que combina la disponibilidad, la confiabilidad y la escala de Spanner y las capacidades avanzadas de búsqueda de similitud de Vertex AI Vector Search. Esta búsqueda se realiza comparando las incorporaciones de elementos en el índice de la búsqueda de vectores y mostrando las coincidencias más similares.

Caso de uso

Imagina que eres un científico de datos en un minorista de moda y tratas de mantenerte al día con las tendencias, búsquedas de productos y recomendaciones que cambian rápidamente. El desafío es que tienes recursos y silos de datos limitados. Esta entrada de blog muestra cómo implementar un caso de uso de recomendación de indumentaria mediante el enfoque de búsqueda de similitud en los datos de indumentaria.Se abordan los siguientes pasos:

Datos provenientes de Spanner
Vectores generados para los datos de indumentaria con ML.PREDICT y almacenados en Spanner
Datos vectoriales de Spanner integrados en la búsqueda de vectores mediante el flujo de datos y los trabajos de flujo de trabajo
Se realizó una búsqueda de vectores para encontrar una coincidencia de similitud con las entradas ingresadas por el usuario

Crearemos una aplicación web de demostración para realizar una búsqueda de indumentaria basada en el texto de entrada del usuario. La aplicación permite que los usuarios busquen indumentaria ingresando una descripción de texto.

Spanner al índice de búsqueda de vectores:

Los datos de la búsqueda de indumentaria se almacenan en Spanner. Invocaremos la API de Embeddings de Vertex AI en la construcción ML.PREDICT directamente a partir de los datos de Spanner. Luego, aprovecharemos los trabajos de Dataflow y flujo de trabajo que suben de forma masiva estos datos (inventario y, también, incorporaciones) a la búsqueda de vectores de Vertex AI y actualiza el índice.

Cómo ejecutar consultas de usuarios en el índice:

Cuando un usuario ingresa una descripción de indumentaria, la app genera las incorporaciones en tiempo real con la API de Text Embeddings. Esto se envía como entrada a la API de Vector Search para encontrar 10 descripciones de productos relevantes del índice y muestra la imagen correspondiente.

Descripción general de la arquitectura

La arquitectura de la aplicación de búsqueda de vectores de Spanner se muestra en el siguiente diagrama de 2 partes:

Spanner al índice de búsqueda de vectores:

App cliente para ejecutar consultas de usuarios en el índice:

Qué compilarás

Spanner al índice vectorial:

Base de datos de Spanner para almacenar y administrar datos de origen y las incorporaciones correspondientes
Un trabajo de flujo de trabajo que sube datos de forma masiva (ID y también incorporaciones) a la base de datos de Vertex AI Vector Search.
Una API de búsqueda de vectores que se usa para encontrar descripciones de productos relevantes en el índice.

Cómo ejecutar consultas de usuarios en el índice:

Una aplicación web que permite que los usuarios ingresen descripciones de texto de indumentaria, realiza una búsqueda de similitud usando el extremo de índice implementado y muestra la indumentaria más cercana a la entrada.

Cómo funciona

Cuando un usuario ingresa una descripción de texto de indumentaria, la aplicación web envía la descripción a la API de búsqueda de vectores. Luego, la API de Vector Search usa las incorporaciones de las descripciones de indumentaria para encontrar las descripciones de productos más relevantes del índice. Luego, se muestran al usuario las descripciones y las imágenes correspondientes del producto. El flujo de trabajo general es el siguiente:

Generar incorporaciones para datos almacenados en Spanner
Exportar y subir incorporaciones a un índice de búsqueda de vectores
Consulta el índice de búsqueda de vectores en busca de elementos similares realizando una búsqueda de vecino más cercano.

2. Requisitos

Un navegador, como Chrome o Firefox.
Un proyecto de Google Cloud con la facturación habilitada.

Antes de comenzar

En la consola de Google Cloud, en la página del selector de proyectos, selecciona o crea un proyecto de Google Cloud.
Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén más información para verificar si la facturación está habilitada en un proyecto.
Asegúrate de que todas las APIs necesarias (Cloud Spanner, Vertex AI, Google Cloud Storage) estén habilitadas.
Usarás Cloud Shell, un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado en gcloud. Consulta la documentación para ver los comandos de gcloud y su uso. Si tu proyecto no está configurado, usa el siguiente comando para hacerlo:

gcloud config set project <YOUR_PROJECT_ID>

Para comenzar, navega a la página Cloud Spanner con tu proyecto activo de Google Cloud.

3. Backend: Crea la fuente de datos y las incorporaciones de Spanner

En este caso de uso, la base de datos de Spanner aloja el inventario de indumentaria con las imágenes y la descripción correspondientes. Asegúrate de generar incorporaciones para la descripción de texto y almacenarlas en tu base de datos de Spanner como ARRAY<float64>.

Crea los datos de Spanner

Crea una instancia llamada “spanner-vertex” y una base de datos llamada “spanner-vertex-embeddings”. Crea una tabla con el 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);

Inserta datos en la tabla con INSERT SQL

Las secuencias de comandos de inserción para los datos de muestra están disponibles aquí.

Crea un modelo de incorporaciones de texto

Esto es necesario para que podamos generar incorporaciones del contenido de la entrada. A continuación, se incluye el DDL de lo siguiente:

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

Genera incorporaciones de texto para los datos de origen

Crea una tabla para almacenar las incorporaciones y, luego, insertar las que se generaron. En una aplicación de base de datos real, la carga de datos en Spanner hasta el paso 2 sería transaccional. Para mantener intactas las prácticas recomendadas de diseño, prefiero mantener las tablas transaccionales normalizadas, por lo que crear una tabla separada para las incorporaciones.

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

Ahora que el contenido masivo y las incorporaciones están listos, vamos a crear un índice y un extremo de búsqueda de vectores para almacenar las incorporaciones que ayudarán a realizar la búsqueda de vectores.

4. Trabajo de flujo de trabajo: exportación de datos de Spanner a la búsqueda de vectores

Crea un bucket de Cloud Storage

Esto es necesario para almacenar las incorporaciones de Spanner en un bucket de GCS en un formato JSON que la búsqueda de vectores espera como entrada. Crea un bucket en la misma región que tus datos en Spanner. Si es necesario, crea una carpeta dentro de la carpeta, pero, en especial, crea un archivo vacío llamado empty.json.

Configura el flujo de trabajo en la nube

Para configurar una exportación por lotes de Spanner a un índice de búsqueda de vectores de Vertex AI, sigue estos pasos:

Crea un índice vacío:

Asegúrate de que el índice de búsqueda de vectores esté en la misma región que tu bucket de Cloud Storage y los datos. Sigue los 11 pasos de instrucción en la pestaña de la consola de la sección Crea un índice para la actualización por lotes en la página Administrar índices. En la carpeta que se pasa a contentDeltaUri, crea un archivo vacío llamado empty.json, ya que no podrás crear un índice sin este archivo. Esto crea un índice vacío.

Si ya tienes un índice, puedes omitir este paso. El flujo de trabajo reemplazará tu índice.

Nota: No puedes implementar un índice vacío en un extremo. Así que postergamos el paso de la implementación en un extremo a un paso posterior, después de exportar los datos vectoriales a Cloud Storage.

Clona este repositorio de Git: Hay varias formas de clonar un repositorio de Git. Una forma de hacerlo es ejecutar el siguiente comando con la CLI de GitHub. Ejecuta los siguientes 2 comandos desde la terminal de Cloud Shell:

gh repo clone cloudspannerecosystem/spanner-ai

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

Esta carpeta contiene dos archivos

batch-export.yaml: Esta es la definición del flujo de trabajo.
sample-batch-input.json: Esta es una muestra de los parámetros de entrada del flujo de trabajo.

Configura input.json desde el archivo de muestra: Primero, copia el archivo JSON de muestra.

cp sample-batch-input.json input.json

Luego, edita input.json con los detalles de tu proyecto. En este caso, tu archivo json debería verse de la siguiente manera:

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

Permisos de configuración

Para los entornos de producción, te recomendamos crear una nueva cuenta de servicio y otorgarle uno o más roles de IAM que contengan los permisos mínimos necesarios para administrar el servicio. Los siguientes roles son necesarios para configurar el flujo de trabajo para exportar datos de Spanner (incorporaciones) al índice de la búsqueda de vectores:

Cuenta de servicio de Cloud Workflow:

De forma predeterminada, usa la cuenta de servicio predeterminada de Compute Engine.

Si usas una cuenta de servicio configurada de forma manual, debes incluir los siguientes roles:

Para activar un trabajo de Dataflow: Dataflow Admin, Dataflow Worker.

Para actuar en nombre de una cuenta de servicio de trabajador de Dataflow, consulta Usuario de cuenta de servicio.

Para escribir registros, sigue estos pasos: Escritor de registros.

Para activar la recompilación de Vertex AI Vector Search, haz lo siguiente: Usuario de Vertex AI.

Cuenta de servicio de trabajador de Dataflow:

Si usas una cuenta de servicio configurada de forma manual, debes incluir los siguientes roles:

Para administrar Dataflow: Administrador de Dataflow, Trabajador de Dataflow. Para leer datos de Spanner, usa el siguiente comando: Lector de bases de datos de Cloud Spanner. Acceso de escritura sobre el Container Registry de GCS seleccionado: GCS Storage Bucket Owner.

Implementa el flujo de trabajo en la nube

Implementa el archivo yaml de flujo de trabajo en tu proyecto de Google Cloud. Puedes configurar la región o ubicación en la que se ejecutará el flujo de trabajo cuando se ejecute.

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"

El flujo de trabajo ahora debería ser visible en la página Flujos de trabajo en la consola de Google Cloud.

Nota: También puedes crear e implementar el flujo de trabajo desde la consola de Google Cloud. Sigue las indicaciones que aparecen en la consola de Cloud. Para la definición del flujo de trabajo, copia y pega el contenido de batch-export.yaml.

Una vez que se complete este proceso, ejecuta el flujo de trabajo para que comience la exportación de datos.

Ejecuta el flujo de trabajo en la nube

Ejecuta el siguiente comando para ejecutar el flujo de trabajo:

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

La ejecución debería aparecer en la pestaña Ejecuciones en Flujos de trabajo. Esto debería cargar tus datos en la base de datos de la búsqueda de vectores y, luego, indexarlos.

Nota: También puedes realizar la ejecución desde la consola con el botón Ejecutar. Sigue las indicaciones y, para la entrada, copia y pega el contenido del input.json personalizado.

5. Cómo implementar el índice de búsqueda de vectores

Implementa el índice en un extremo

Puedes seguir los pasos que se indican a continuación para implementar el índice:

En la página de índices de búsqueda de vectores, deberías ver un botón IMPLOY junto al índice que acabas de crear en el paso 2 de la sección anterior. También puedes navegar a la página de información del índice y hacer clic en el botón IMPLEMENTAR EN EL EXTREMO.
Proporciona la información necesaria y, luego, implementa el índice en un extremo.

Como alternativa, puedes ver este notebook para implementarlo en un extremo (pasa a la parte de implementación del notebook). Una vez implementado, toma nota del ID del índice implementado y la URL del extremo.

6. Frontend: Datos del usuario en la búsqueda de vectores

Compilemos una aplicación de Python simple con una UX con tecnología de graduación para probar rápidamente nuestra implementación. Puedes consultar la implementación aquí para implementar esta app de demostración en tu propio notebook de colab.

Usaremos el SDK de Python de aiplatform para llamar a la API de incorporaciones y también invocar el extremo del índice de la búsqueda de vectores.

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

Usaremos gradio para demostrar la aplicación de IA que estamos creando de forma rápida y fácil con una interfaz de usuario. Reinicia el entorno de ejecución antes de implementar este paso.

!pip install gradio
import gradio as gr

Desde la app web tras la entrada del usuario, invoca la API de Embeddings, usaremos el modelo de incorporación de texto: textembedding-gecko@latest

El siguiente método invoca el modelo de incorporación de texto y muestra las incorporaciones vectoriales del texto ingresado por el usuario:

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

Pruébalo

text_embedding("red shorts for girls")

Deberías ver un resultado similar al siguiente (ten en cuenta que la imagen se recortó en altura, por lo que no puedes ver toda la respuesta vectorial):

Declara el ID del índice implementado y el ID del extremo

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

Define el método de búsqueda de vectores para llamar al extremo de índice y mostrar el resultado con las 10 coincidencias más cercanas para la respuesta de incorporación correspondiente al texto de entrada del usuario.

En la siguiente definición de método para la búsqueda de vectores, ten en cuenta que el método find_nexts se invoca para identificar los 10 vectores más cercanos.

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

También observarás la llamada al método spanner_read_data. Analicemos esto en el siguiente paso.

Define la implementación del método de lectura de datos de Spanner que invoca el método actions_sql para extraer las imágenes correspondientes a los IDs de los vectores vecinos más cercanos que se mostraron en el paso anterior.

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

Se deben mostrar las URL de las imágenes correspondientes a los vectores elegidos.

Por último, reunimos las piezas en una interfaz de usuario y activemos el proceso de búsqueda de vectores

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)

Deberías ver el resultado como se muestra a continuación:

Imagen: Vínculo

Mira el video de resultados aquí.

7. Limpia

Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en esta publicación:

En la consola de Google Cloud, ve a la página Administrar recursos.
En la lista de proyectos, elige el proyecto que deseas borrar y haz clic en Borrar.
En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrarlo.
Si no quieres borrar el proyecto, borra la instancia de Spanner. Para ello, navega a la instancia que acabas de crear para este proyecto y haz clic en el botón BORRAR INSTANCIA en la esquina superior derecha de la página de descripción general de la instancia.
También puedes navegar al índice de la Búsqueda de vectores, anular la implementación del extremo y el índice, y borrar el índice.

8. Conclusión

¡Felicitaciones! Completaste correctamente la implementación de Spanner: búsqueda de vectores de Vertex del

Crear una fuente de datos de Spanner y las incorporaciones para aplicaciones que provienen de la base de datos de Spanner
Se está creando el índice de la base de datos de la Búsqueda de vectores.
Integrar datos vectoriales de Spanner a la búsqueda de vectores mediante trabajos de Dataflow y de flujo de trabajo
Implementar un índice en un extremo
Por último, invoca la búsqueda de vectores en la entrada del usuario en una implementación del SDK de Vertex AI con tecnología de Python.

No dudes en extender la implementación a tu propio caso de uso o improvisar el caso de uso actual con funciones nuevas. Obtén más información sobre las capacidades de aprendizaje automático de Spanner aquí.