1. Introduzione
I recenti progressi nel deep learning hanno reso possibile rappresentare il 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 embedding) per trovare i documenti più pertinenti per la query di un utente. La ricerca vettoriale è preferita alla ricerca tradizionale per applicazioni come la ricerca di abbigliamento, in cui gli utenti spesso cercano articoli in base alla descrizione, allo stile o al contesto anziché in base ai nomi esatti di prodotti o brand. Possiamo integrare il database Cloud Spanner con Vector Search per eseguire la corrispondenza della somiglianza vettoriale. Utilizzando insieme Spanner e Vector Search, i clienti possono creare un'integrazione potente che combina la disponibilità, l'affidabilità e la scalabilità di Spanner con le funzionalità avanzate di ricerca di similarità di Vertex AI Vector Search. Questa ricerca viene eseguita confrontando gli embedding degli elementi nell'indice di Vector Search e restituendo le corrispondenze più simili.
Caso d'uso
Immagina di essere un data scientist di un retailer di moda che cerca di stare al passo con le tendenze, le ricerche di prodotti e i consigli in rapida evoluzione. Il problema è che hai risorse e silo di dati limitati. Questo post del blog mostra come implementare un caso d'uso di raccomandazione di abbigliamento utilizzando l'approccio di ricerca per similarità sui dati di abbigliamento.Vengono trattati 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 job di flusso di dati e workflow
- Ricerca vettoriale eseguita per trovare una corrispondenza di similarità per l'input inserito dall'utente
Creeremo un'applicazione web demo per eseguire la ricerca 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 Vector Search:
I dati per la ricerca di abbigliamento sono archiviati in Spanner. Richiameremo l'API Vertex AI Embeddings nel costrutto ML.PREDICT direttamente dai dati Spanner. Poi utilizzeremo i job Dataflow e Workflow che caricano in blocco questi dati (inventario ed embedding) in Vertex AI Vector Search e aggiornano l'indice.
Esecuzione di query degli utenti sull'indice:
Quando un utente inserisce una descrizione di abbigliamento, l'app genera gli incorporamenti in tempo reale utilizzando l'API Text Embeddings. Queste informazioni vengono poi inviate come input all'API Vector Search per trovare 10 descrizioni di prodotto pertinenti dall'indice e visualizzare l'immagine corrispondente.
Panoramica dell'architettura
L'architettura dell'applicazione Spanner-Vector Search è mostrata nel seguente diagramma in due parti:
Spanner to Vector Search Index: 
App client per eseguire query utente sull'indice:
Cosa creerai
Spanner a indice vettoriale:
- Database Spanner per archiviare e gestire i dati di origine e gli incorporamenti corrispondenti
- Un job del flusso di lavoro che carica in blocco i dati (ID e incorporamenti) nel database Vertex AI Vector Search.
- Un'API Vector Search utilizzata per trovare descrizioni dei prodotti pertinenti nell'indice.
Esecuzione di query degli utenti sull'indice:
- Un'applicazione web che consente agli utenti di inserire descrizioni di abbigliamento in formato testo, esegue la ricerca di somiglianze utilizzando l'endpoint dell'indice di cui è stato eseguito il deployment e restituisce gli indumenti più simili 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 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 workflow generale è il seguente:
- Genera embedding per i dati archiviati in Spanner.
- Esporta e carica gli embedding in un indice Vector Search.
- Esegui query sull'indice di ricerca vettoriale per trovare elementi simili eseguendo una ricerca del vicino più prossimo.
2. Requisiti
Prima di iniziare
- Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
- Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per 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 in Google Cloud precaricato con gcloud. Consulta la documentazione per i comandi e l'utilizzo di gcloud. Se il progetto non è impostato, utilizza il seguente comando per impostarlo:
gcloud config set project <YOUR_PROJECT_ID>
- Vai alla pagina Cloud Spanner con il tuo progetto Google Cloud attivo per iniziare.
3. Backend: crea l'origine dati Spanner e gli incorporamenti
In questo caso d'uso, il database Spanner contiene l'inventario di capi di abbigliamento con le immagini e le descrizioni 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 INSERT SQL
Gli script per i dati di esempio sono disponibili qui.
- Crea il modello Text Embeddings
Questo è necessario 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 per archiviare gli embedding e inserisci gli embedding generati. In un'applicazione di database reale, il caricamento dei dati in Spanner fino al passaggio 2 sarebbe transazionale. Per mantenere intatte le best practice di progettazione, preferisco mantenere normalizzate le tabelle transazionali, quindi creare una tabella separata per gli embedding.
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 e gli embedding collettivi sono pronti, creiamo un indice e un endpoint Vector Search per archiviare gli embedding che aiuteranno a eseguire la ricerca vettoriale.
4. Job del flusso di lavoro: esportazione dei dati Spanner in Vector Search
- Crea un bucket Cloud Storage
Questo è necessario per archiviare gli incorporamenti di Spanner in un bucket GCS in un formato JSON che Vector Search prevede come input. Crea un bucket nella stessa regione dei tuoi dati in Spanner. Crea una cartella al suo interno, se necessario, ma soprattutto crea un file vuoto denominato empty.json.
- Configura Cloud Workflow
Per configurare un'esportazione batch da Spanner a un indice Vertex AI Vector Search:
Crea un indice vuoto:
Assicurati che l'indice di ricerca vettoriale si trovi nella stessa regione del bucket Cloud Storage e dei dati. Segui gli 11 passaggi delle istruzioni nella scheda della console della sezione Crea un indice per l'aggiornamento batch nella pagina Gestisci indici. Nella cartella passata a contentsDeltaUri, crea un file vuoto denominato empty.json perché non potresti creare un indice senza questo file. Viene creato un indice vuoto.
Se hai già 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. Pertanto, rimandiamo il passaggio di deployment a un endpoint a un passaggio successivo, dopo aver esportato i dati vettoriali in Cloud Storage.
Clona questo repository Git: esistono diversi modi per clonare un repository Git. Uno di questi consiste nell'eseguire il seguente comando utilizzando la CLI di GitHub. Esegui i due comandi riportati di seguito 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 workflow.sample-batch-input.json: Questo è un esempio dei parametri di input del workflow.
Configura input.json dal file di esempio:innanzitutto, copia il file JSON di esempio.
cp sample-batch-input.json input.json
Poi modifica input.json con i dettagli del tuo progetto. In questo caso, il file JSON dovrebbe essere simile al seguente:
{
"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>>"
}
}
Configurare le autorizzazioni
Per gli ambienti di produzione, ti consigliamo vivamente di creare un nuovo service account e di concedergli uno o più ruoli IAM che contengano le autorizzazioni minime richieste per la gestione del servizio. Per configurare il flusso di lavoro per esportare i dati da Spanner (incorporamenti) all'indice di Vector Search, sono necessari i seguenti ruoli:
Service account Cloud Workflow:
Per impostazione predefinita, utilizza l'account di servizio predefinito di Compute Engine.
Se utilizzi un service account configurato manualmente, devi includere i seguenti ruoli:
Per attivare un job Dataflow: Dataflow Admin, Dataflow Worker.
Per rappresentare un service account worker Dataflow: Utente service account.
Per scrivere log: Writer log.
Per attivare la ricompilazione di Vertex AI Vector Search: utente Vertex AI.
Account di servizio worker Dataflow:
Se utilizzi un service account configurato manualmente, devi includere i seguenti ruoli:
Per gestire Dataflow: Dataflow Admin, Dataflow Worker. Per leggere i dati da Spanner: Cloud Spanner Database Reader. Accesso in scrittura a Container Registry GCS selezionato: Proprietario bucket di archiviazione GCS
- Esegui il deployment del flusso di lavoro Cloud
Esegui 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 workflow.
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 workflow ora dovrebbe essere visibile nella pagina Workflows della console Google Cloud.
Nota: puoi anche creare ed eseguire il deployment del flusso di lavoro dalla console Google Cloud. Segui le istruzioni nella console Cloud. Per la definizione del flusso di lavoro, copia e incolla i contenuti di batch-export.yaml.
Una volta completata questa operazione, esegui il flusso di lavoro in modo che inizi l'esportazione dei dati.
- Esegui 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 in Workflows. In questo modo, i dati vengono caricati nel database Vector Search e indicizzati.
Nota: puoi anche eseguire l'operazione 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 Vector Search
Esegui il deployment dell'indice su un endpoint
Per eseguire il deployment dell'indice, segui questi passaggi:
- Nella pagina Indici di ricerca vettoriale, dovresti vedere un pulsante DEPLOY (Esegui il deployment) accanto all'indice che hai appena creato nel passaggio 2 della sezione precedente. In alternativa, puoi andare alla pagina delle informazioni sull'indice e fare clic sul pulsante ESEGUI IL DEPLOYMENT SU ENDPOINT.
- Fornisci le informazioni necessarie ed esegui il deployment dell'indice in un endpoint.
In alternativa, puoi consultare questo blocco note per eseguire il deployment su un endpoint (vai alla parte del deployment del blocco note). Una volta eseguito il deployment, prendi nota dell'ID indice e dell'URL dell'endpoint.
6. Frontend: dati utente per Vector Search
Creiamo una semplice applicazione Python 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 notebook Colab.
- Utilizzeremo l'SDK Python aiplatform per chiamare l'API Embeddings e anche per richiamare l'endpoint dell'indice 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 dimostrare l'applicazione AI 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 in base all'input dell'utente, richiama l'API Embeddings. Utilizzeremo il modello di incorporamento di 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
Testalo
text_embedding("red shorts for girls")
Dovresti visualizzare un output simile a quello riportato di seguito (tieni presente che l'immagine è ritagliata in altezza, quindi non puoi visualizzare l'intera risposta del vettore):
- Dichiara l'ID indice di cui è stato eseguito il deployment e l'ID 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')
- Definisci il metodo di ricerca vettoriale per chiamare l'endpoint dell'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, tieni presente 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 riferimento al metodo spanner_read_data. Vediamo come fare nel passaggio successivo.
- Definisci l'implementazione del metodo di lettura dei dati di Spanner che richiama il metodo execute_sql per estrarre le immagini corrispondenti agli ID dei vettori dei vicini più prossimi 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 i pezzi in un'interfaccia utente e attiviamo la procedura 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 mostrato di seguito:
Immagine: Link
Guarda il video dei risultati: 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 andando all'istanza che hai appena creato per questo progetto e fai clic sul pulsante ELIMINA ISTANZA nell'angolo in alto a destra della pagina di panoramica dell'istanza.
- Puoi anche accedere all'indice Vector Search, annullare il deployment dell'endpoint e dell'indice ed eliminare l'indice.
8. Conclusione
Complimenti! Hai completato correttamente l'implementazione di Spanner - Vertex Vector Search entro
- Creazione dell'origine dati Spanner e degli incorporamenti per le applicazioni provenienti dal database Spanner.
- Creazione dell'indice del database Vector Search.
- Integrazione dei dati vettoriali da Spanner a Vector Search utilizzando i job Dataflow e Workflow.
- Deployment dell'indice in un endpoint.
- Infine, richiama la ricerca vettoriale sull'input dell'utente in un'implementazione basata su Python dell'SDK Vertex AI.
Puoi estendere l'implementazione al tuo caso d'uso o improvvisare il caso d'uso attuale con nuove funzionalità. Scopri di più sulle funzionalità di machine learning di Spanner qui.