1. Introduzione
I recenti progressi nel deep learning hanno reso possibile la rappresentazione di testo e altri dati in modo da acquisire il significato semantico. Ciò ha portato a un nuovo approccio alla ricerca, chiamato ricerca vettoriale, che utilizza rappresentazioni vettoriali del testo (note come incorporamenti) per trovare i documenti più pertinenti per la query di un utente. La ricerca vettoriale è preferita rispetto alla ricerca tradizionale per applicazioni come la ricerca di capi di abbigliamento, dove gli utenti spesso cercano gli articoli in base alla descrizione, allo stile o al contesto anziché in base al nome esatto del prodotto o del brand. Possiamo integrare il database Cloud Spanner con Vector Search per eseguire la corrispondenza di similitudine vettoriale. Utilizzando Spanner e Vector Search insieme, i clienti possono creare una potente integrazione che combina la disponibilità, l'affidabilità e la scalabilità di Spanner e le funzionalità di ricerca di somiglianze avanzate di Vertex AI Vector Search. Questa ricerca viene eseguita confrontando le incorporazioni di elementi nell'indice di Vector Search e restituisce le corrispondenze più simili.
Caso d'uso
Immagina di essere un data scientist di un rivenditore di moda che cerca di stare al passo con i rapidi cambiamenti delle tendenze, delle ricerche di prodotti e dei consigli. Il problema è che disponi di risorse e silos di dati limitati. Questo blog post mostra come implementare un caso d'uso di consigli di abbigliamento utilizzando un approccio di ricerca di somiglianza sui dati relativi all'abbigliamento.Vengono illustrati i seguenti passaggi:
- Dati provenienti da Spanner
- Vettori generati per i dati di abbigliamento utilizzando ML.PREDICT e archiviati in Spanner
- Dati vettoriali Spanner integrati con Vector Search utilizzando Dataflow e job di flusso di lavoro
- Ricerca vettoriale eseguita per trovare corrispondenze di somiglianza per l'input inserito dall'utente
Creeremo un'applicazione web demo per eseguire la ricerca di capi di abbigliamento in base al testo inserito dall'utente. L'applicazione consente agli utenti di cercare capi di abbigliamento inserendo una descrizione testuale.
Spanner all'indice di ricerca vettoriale:
I dati per la ricerca di articoli di abbigliamento vengono archiviati in Spanner. Richiamo l'API Vertex AI Embeddings nel costrutto ML.PREDICT direttamente dai dati Spanner. Dopodiché sfrutteremo i job Dataflow e Workflow che caricano collettivamente questi dati (inventario e incorporamenti) nella Vector Search di Vertex AI e aggiornano l'indice.
Esecuzione di query utente sull'indice:
Quando un utente inserisce una descrizione di abbigliamento, l'app genera gli incorporamenti in tempo reale utilizzando l'API Text Embeddings. Questo viene poi inviato come input all'API Vector Search per trovare 10 descrizioni di prodotto pertinenti nell'indice e viene visualizzata l'immagine corrispondente.
Panoramica dell'architettura
L'architettura dell'applicazione Spanner-Vector Search è mostrata nel seguente diagramma in due parti:
Da Spanner all'indice di ricerca vettoriale: 
App client per eseguire query utente sull'indice:
Cosa creerai
Da Spanner a indice vettoriale:
- Database Spanner per archiviare e gestire i dati di origine e gli incorporamenti corrispondenti
- Un job di flusso di lavoro che carica collettivamente i dati (ID e incorporamenti) nel database Vertex AI Vector Search.
- Un'API Vector Search utilizzata per trovare descrizioni di prodotti pertinenti dall'indice.
Esecuzione di query utente sull'indice:
- Un'applicazione web che consente agli utenti di inserire descrizioni testuali di capi di abbigliamento, esegue una ricerca di somiglianze utilizzando l'endpoint indice di cui è stato eseguito il deployment e restituisce i capi di abbigliamento più vicini all'input.
Come funziona
Quando un utente inserisce una descrizione testuale di un capo di abbigliamento, l'applicazione web invia la descrizione all'API Vector Search. L'API Vector Search utilizza poi gli incorporamenti delle descrizioni dei capi di abbigliamento per trovare le descrizioni dei prodotti più pertinenti nell'indice. Le descrizioni dei prodotti e le immagini corrispondenti vengono quindi mostrate all'utente. Il flusso di lavoro generale è il seguente:
- Genera incorporamenti per i dati archiviati in Spanner.
- Esporta e carica gli incorporamenti in un indice di Vector Search.
- Esegui una query sull'indice di Vector Search per trovare elementi simili eseguendo una ricerca per il vicino più prossimo.
2. Requisiti
Prima di iniziare
- Nella pagina del selettore dei progetti della console Google Cloud, seleziona o crea un progetto Google Cloud.
- Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come controllare se la fatturazione è abilitata su un progetto
- Assicurati che tutte le API necessarie (Cloud Spanner, Vertex AI, Google Cloud Storage) siano abilitate.
- Utilizzerai Cloud Shell, un ambiente a riga di comando in esecuzione su Google Cloud in cui è precaricato gcloud. Consulta la documentazione per i comandi e l'utilizzo di gcloud. Se il progetto non è configurato, utilizza il comando seguente per impostarlo:
gcloud config set project <YOUR_PROJECT_ID>
- Per iniziare, vai alla pagina Cloud Spanner con il tuo progetto Google Cloud attivo
3. Backend: creare l'origine dati e gli incorporamenti di Spanner
In questo caso d'uso, il database Spanner ospita l'inventario di capi di abbigliamento con le immagini e la descrizione corrispondenti. Assicurati di generare incorporamenti per la descrizione testuale e di archiviarli nel database Spanner come ARRAY<float64>.
- Crea i dati Spanner
Crea un'istanza denominata "spanner-vertex" e un database denominato "spanner-vertex-embeddings". Crea una tabella utilizzando il 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);
- Inserisci i dati nella tabella utilizzando il comando INSERT SQL
Gli script per i dati di esempio sono disponibili qui.
- Crea modello di incorporamenti di testo
Questa operazione è necessaria per poter generare incorporamenti per i contenuti nell'input. Di seguito è riportato il DDL per lo stesso:
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');
- Generare incorporamenti di testo per i dati di origine
Crea una tabella in cui archiviare gli incorporamenti e inserisci quelli generati. In un'applicazione di database reale, il carico dei dati su Spanner fino al passaggio 2 sarebbe transazionale. Per mantenere invariate le best practice di progettazione, preferisco mantenere le tabelle transazionali normalizzate, in modo da creare una tabella separata per gli incorporamenti.
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)
) ;
Ora che i contenuti collettivi e gli incorporamenti sono pronti, creiamo un indice e un endpoint di ricerca vettoriale per archiviare gli incorporamenti che aiuteranno a eseguire la ricerca vettoriale.
4. Job flusso di lavoro: esportazione dati Spanner in Vector Search
- Crea un bucket Cloud Storage
Questo è necessario per archiviare gli incorporamenti da Spanner in un bucket GCS in un formato json che Vector Search prevede come input. Crea un bucket nella stessa regione dei dati in Spanner. Se necessario, crea una cartella, ma crea principalmente un file vuoto denominato empty.json.
- Configura Cloud Workflows
Per configurare un'esportazione batch da Spanner a un indice di Vertex AI Vector Search:
Crea un indice vuoto:
Assicurati che l'indice di Vector Search si trovi nella stessa regione del bucket Cloud Storage e dei dati. Segui gli 11 passaggi delle istruzioni nella scheda Console della sezione Crea un indice per l'aggiornamento batch nella pagina Gestisci indici. Nella cartella che viene passata a contentDeltaUri, crea un file vuoto denominato empty.json, perché non potrai creare un indice senza questo file. Viene creato un indice vuoto.
Se disponi già di un indice, puoi saltare questo passaggio. Il flusso di lavoro sovrascriverà l'indice.
Nota: non puoi eseguire il deployment di un indice vuoto in un endpoint. Quindi rinviamo il passaggio del deployment in un endpoint a un passaggio successivo, dopo l'esportazione dei dati vettoriali in Cloud Storage.
Clona questo repository git: esistono diversi modi per clonare un repository git, uno dei quali è eseguire il seguente comando utilizzando l'interfaccia a riga di comando GitHub. Esegui questi due comandi dal terminale Cloud Shell:
gh repo clone cloudspannerecosystem/spanner-ai
cd spanner-ai/vertex-vector-search/workflows
Questa cartella contiene due file
batch-export.yaml: questa è la definizione del flusso di lavoro.sample-batch-input.json: questo è un esempio dei parametri di input del flusso di lavoro.
Configura input.json dal file di esempio: copia innanzitutto il file json di esempio,
cp sample-batch-input.json input.json
Quindi, modifica input.json con i dettagli del progetto. In questo caso, il file JSON dovrebbe avere il seguente aspetto:
{
"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>>"
}
}
Autorizzazioni di configurazione
Per gli ambienti di produzione, consigliamo vivamente di creare un nuovo account di servizio e assegnargli uno o più ruoli IAM contenenti le autorizzazioni minime richieste per la gestione del servizio. I ruoli seguenti sono necessari per configurare il flusso di lavoro per l'esportazione dei dati da Spanner (incorporamenti) all'indice di Vector Search:
Account di servizio Cloud Workflow:
Per impostazione predefinita, utilizza l'account di servizio predefinito di Compute Engine.
Se utilizzi un account di servizio configurato manualmente, devi includere i seguenti ruoli:
Per attivare un job Dataflow: Amministratore Dataflow, worker Dataflow.
Per impersonare un account di servizio worker Dataflow: Utente account di servizio.
Per scrivere i log: Writer log.
Per attivare la ricreazione di Vertex AI Vector Search: Vertex AI User.
Account di servizio worker Dataflow:
Se utilizzi un account di servizio configurato manualmente, devi includere i seguenti ruoli:
Per gestire Dataflow: Amministratore Dataflow, Worker Dataflow. Per leggere i dati da Spanner: Lettore database Cloud Spanner. Accesso in scrittura sul Container Registry GCS selezionato: Proprietario bucket di archiviazione GCS.
- Esegui il deployment del flusso di lavoro Cloud
Eseguire il deployment del file YAML del flusso di lavoro nel tuo progetto Google Cloud. Puoi configurare la regione o la località in cui verrà eseguito il flusso di lavoro quando verrà eseguito.
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"
Il flusso di lavoro ora dovrebbe essere visibile nella pagina Flussi di lavoro della console Google Cloud.
Nota: puoi anche creare il flusso di lavoro ed eseguirne il deployment dalla console Google Cloud. Segui le istruzioni nella console Cloud. Per la definizione del flusso di lavoro, copia e incolla il contenuto di batch-export.yaml.
Al termine, esegui il flusso di lavoro in modo che inizi l'esportazione dei dati.
- Eseguire il flusso di lavoro cloud
Esegui questo comando per eseguire il flusso di lavoro:
gcloud workflows execute vector-export-workflow --data="$(cat input.json)"
L'esecuzione dovrebbe essere visualizzata nella scheda Esecuzioni di Workflows. In questo modo i dati dovrebbero essere caricati nel database di Vector Search e indicizzarli.
Nota: puoi eseguire l'operazione anche dalla console utilizzando il pulsante Esegui. Segui le istruzioni e, per l'input, copia e incolla i contenuti del file input.json personalizzato.
5. Esegui il deployment dell'indice di Vector Search
Esegui il deployment dell'indice in un endpoint
Per eseguire il deployment dell'indice, segui questi passaggi:
- Nella pagina Indici di ricerca vettoriale, dovresti vedere un pulsante DEPLOY accanto all'indice che hai appena creato nel passaggio 2 della sezione precedente. In alternativa, puoi andare alla pagina delle informazioni dell'indice e fare clic sul pulsante ESEGUI IL DEPLOYMENT SU ENDPOINT.
- Fornisci le informazioni necessarie ed esegui il deployment dell'indice su un endpoint.
In alternativa, puoi esaminare questo blocco note per eseguire il deployment in un endpoint (vai alla parte del blocco note relativa al deployment). Dopo il deployment, prendi nota dell'ID indice e dell'URL dell'endpoint di cui è stato eseguito il deployment.
6. Frontend: dai dati utente alla ricerca vettoriale
Creiamo un'applicazione Python semplice con un'esperienza utente basata su gradio per testare rapidamente la nostra implementazione: puoi fare riferimento all'implementazione qui per implementare questa app demo nel tuo blocco note colab.
- Utilizzeremo l'SDK python aiplatform per chiamare l'API Embeddings e anche per richiamare l'endpoint indice di 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()
- Utilizzeremo gradio per una dimostrazione dell'applicazione di IA che stiamo creando in modo rapido e semplice, con un'interfaccia utente. Riavvia il runtime prima di implementare questo passaggio.
!pip install gradio
import gradio as gr
- Dall'app web, dopo l'input dell'utente, richiamare l'API Embeddings; utilizzeremo il modello di incorporamento del testo: textembedding-gecko@latest
Il metodo riportato di seguito richiama il modello di incorporamento di testo e restituisce gli incorporamenti vettoriali per il testo inserito dall'utente:
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
Testa
text_embedding("red shorts for girls")
Dovresti vedere un output simile al seguente (tieni presente che l'immagine è ritagliata in altezza, quindi non puoi vedere l'intera risposta vettoriale):
- Dichiara l'ID indice e l'ID endpoint di cui è stato eseguito il deployment
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')
- Definisci il metodo Vector Search per chiamare l'endpoint indice e mostrare il risultato con le 10 corrispondenze più vicine per la risposta di incorporamento corrispondente al testo di input dell'utente.
Nella definizione del metodo riportata di seguito per la ricerca vettoriale, si noti che viene richiamato il metodo Find_neighbors per identificare i 10 vettori più vicini.
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
Noterai anche il richiamo al metodo spanner_read_data. Vediamola nel prossimo passaggio.
- Definisci l'implementazione del metodo dati di lettura di Spanner che richiama il metodo Esegui_sql per estrarre le immagini corrispondenti agli ID dei vettori dei vicini più vicini restituiti dall'ultimo passaggio.
!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)
Dovrebbe restituire gli URL delle immagini corrispondenti ai vettori scelti.
- Infine, mettiamo insieme le parti in un'interfaccia utente e attiviamo il processo di ricerca vettoriale.
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)
Dovresti vedere il risultato come mostrato di seguito:
Immagine: link
Guarda il video del risultato: qui.
7. Esegui la pulizia
Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo post, segui questi passaggi:
- Nella console Google Cloud, vai alla pagina Gestisci risorse.
- Nell'elenco dei progetti, seleziona il progetto che vuoi eliminare, quindi fai clic su Elimina.
- Nella finestra di dialogo, digita l'ID progetto, quindi fai clic su Chiudi per eliminare il progetto.
- Se non vuoi eliminare il progetto, elimina l'istanza Spanner accedendo all'istanza appena creata per il progetto e fai clic sul pulsante ELIMINA ISTANZA nell'angolo in alto a destra della pagina della panoramica dell'istanza.
- Puoi anche accedere all'indice di Vector Search, annullare il deployment dell'endpoint e dell'indice ed eliminare l'indice.
8. Conclusione
Complimenti! Hai completato l'implementazione di Spanner - Vertex Vector Search entro il giorno
- Creazione dell'origine dati e degli incorporamenti di Spanner per le applicazioni provenienti dal database Spanner.
- Creazione dell'indice del database Vector Search in corso.
- Integrazione dei dati vettoriali da Spanner a Vector Search utilizzando Dataflow e job di flusso di lavoro.
- Deployment dell'indice in un endpoint.
- Infine, è stato richiamato Vector Search sull'input dell'utente in un'implementazione basata su Python dell'SDK Vertex AI.
Sentiti libero di estendere l'implementazione al tuo caso d'uso o di improvvisare il caso d'uso attuale con nuove funzionalità. Scopri di più sulle funzionalità di machine learning di Spanner qui.