Wyszukiwanie podobieństw w usługach Spanner i Vertex AI

1. Wprowadzenie

Ostatnie postępy w dziedzinie deep learning umożliwiły przedstawianie tekstu i innych danych w sposób, który odzwierciedla znaczenie semantyczne. Doprowadziło to do powstania nowego podejścia do wyszukiwania, zwanego wyszukiwaniem wektorowym, które wykorzystuje wektorowe reprezentacje tekstu (tzw. wektory dystrybucyjne) do znajdowania dokumentów najbardziej odpowiednich dla zapytania użytkownika. Wyszukiwanie wektorowe jest preferowane w przypadku aplikacji takich jak wyszukiwanie odzieży, w których użytkownicy często wyszukują produkty według opisu, stylu lub kontekstu, a nie według dokładnych nazw produktów lub marek. Możemy zintegrować bazę danych Cloud Spanner z wyszukiwaniem wektorowym, aby przeprowadzać dopasowywanie wektorów pod względem podobieństwa. Dzięki połączeniu Spannera i wyszukiwania wektorowego klienci mogą utworzyć zaawansowaną integrację, która łączy dostępność, niezawodność i skalę Spannera z zaawansowanymi możliwościami wyszukiwania podobieństw w wyszukiwaniu wektorowym Vertex AI. Wyszukiwanie to polega na porównaniu wektorów dystrybucyjnych elementów w indeksie wyszukiwania wektorowego i zwróceniu najbardziej podobnych wyników.

Przypadek użycia

Wyobraź sobie, że jesteś badaczem danych w sklepie odzieżowym i starasz się nadążyć za szybko zmieniającymi się trendami, wyszukiwaniami produktów i rekomendacjami. Problem polega na tym, że masz ograniczone zasoby i silosy danych. W tym poście na blogu pokazujemy, jak wdrożyć przypadek użycia rekomendacji odzieży za pomocą wyszukiwania podobieństw na podstawie danych o odzieży.Opisujemy te kroki:

Dane pochodzące z usługi Spanner
Wektory wygenerowane dla danych odzieżowych za pomocą funkcji ML.PREDICT i przechowywane w Spannerze
Dane wektorowe Spannera zintegrowane z wyszukiwaniem wektorowym za pomocą zadań przepływu pracy i Dataflow
Wyszukiwanie wektorowe przeprowadzone w celu znalezienia podobieństwa do danych wpisanych przez użytkownika

Utworzymy demonstracyjną aplikację internetową, która będzie wyszukiwać odzież na podstawie danych wejściowych użytkownika. Aplikacja umożliwia użytkownikom wyszukiwanie odzieży na podstawie opisu tekstowego.

Spanner do indeksu wyszukiwania wektorowego:

Dane dotyczące wyszukiwania odzieży są przechowywane w usłudze Spanner. Będziemy wywoływać interfejs Vertex AI Embeddings w konstrukcji ML.PREDICT bezpośrednio z danych Spannera. Następnie wykorzystamy zadania Dataflow i Workflow, które zbiorczo przesyłają te dane (asortyment i wektory dystrybucyjne) do wyszukiwania wektorowego w Vertex AI i odświeżają indeks.

Uruchamianie zapytań użytkowników w indeksie:

Gdy użytkownik wpisze opis odzieży, aplikacja wygeneruje osadzenia w czasie rzeczywistym za pomocą interfejsu Text Embeddings API. Następnie jest on wysyłany jako dane wejściowe do interfejsu Vector Search API, aby znaleźć w indeksie 10 odpowiednich opisów produktów i wyświetlić odpowiednie obrazy.

Omówienie architektury

Architekturę aplikacji Spanner – wyszukiwanie wektorowe przedstawia poniższy 2-częściowy diagram:

Spanner do indeksu wyszukiwania wektorowego:

Aplikacja klienta do wykonywania zapytań użytkowników w indeksie:

Co utworzysz

Spanner do indeksu wektorowego:

Baza danych Spanner do przechowywania danych źródłowych i odpowiadających im wektorów dystrybucyjnych oraz zarządzania nimi.
Zadanie Workflow, które zbiorczo przesyła dane (identyfikator i wektory dystrybucyjne) do bazy danych wyszukiwania wektorowego Vertex AI.
Interfejs Vector Search API, który służy do wyszukiwania w indeksie trafnych opisów produktów.

Uruchamianie zapytań użytkowników w indeksie:

Aplikacja internetowa, która umożliwia użytkownikom wpisywanie tekstowych opisów odzieży, przeprowadza wyszukiwanie podobieństw za pomocą wdrożonego punktu końcowego indeksu i zwraca odzież najbardziej podobną do danych wejściowych.

Jak to działa

Gdy użytkownik wpisze tekstowy opis odzieży, aplikacja internetowa wyśle go do interfejsu Vector Search API. Interfejs Vector Search API wykorzystuje następnie wektory opisów odzieży, aby znaleźć w indeksie najtrafniejsze opisy produktów. Opisy produktów i odpowiednie zdjęcia są następnie wyświetlane użytkownikowi. Ogólny proces wygląda tak:

Generowanie wektorów dystrybucyjnych dla danych przechowywanych w usłudze Spanner.
Eksportuj i przesyłaj wektory dystrybucyjne do indeksu wyszukiwania wektorowego.
Wykonywanie zapytań do indeksu wyszukiwania wektorowego o podobne elementy przez wyszukiwanie najbliższych sąsiadów.

2. Wymagania

przeglądarka, np. Chrome lub Firefox;
projekt Google Cloud z włączonymi płatnościami;

Zanim zaczniesz

W konsoli Google Cloud na stronie selektora projektu wybierz lub utwórz projekt Google Cloud.
Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie włączone są płatności
Sprawdź, czy wszystkie niezbędne interfejsy API (Cloud Spanner, Vertex AI, Google Cloud Storage) są włączone.
Użyjesz Cloud Shell, czyli środowiska wiersza poleceń działającego w Google Cloud, które jest wstępnie załadowane narzędziem gcloud. Informacje o poleceniach gcloud i ich użyciu znajdziesz w dokumentacji. Jeśli projekt nie jest ustawiony, użyj tego polecenia, aby go ustawić:

gcloud config set project <YOUR_PROJECT_ID>

Aby rozpocząć, otwórz stronę Cloud Spanner w aktywnym projekcie w chmurze Google.

3. Backend: tworzenie źródła danych Spanner i wektorów

W tym przypadku baza danych Spanner zawiera asortyment odzieży z odpowiednimi zdjęciami i opisami. Wygeneruj wektory dystrybucyjne opisu tekstowego i zapisz je w bazie danych Spanner jako ARRAY<float64>.

Tworzenie danych Spanner

Utwórz instancję o nazwie „spanner-vertex” i bazę danych o nazwie „spanner-vertex-embeddings”. Utwórz tabelę za pomocą 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);

Wstawianie danych do tabeli za pomocą instrukcji INSERT SQL

Skrypty wstawiania przykładowych danych są dostępne tutaj.

Tworzenie modelu wektorów dystrybucyjnych tekstu

Jest to wymagane, abyśmy mogli wygenerować osadzenia dla treści w danych wejściowych. Poniżej znajdziesz DDL dla tego samego:

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

Generowanie wektorów tekstowych dla danych źródłowych

Utwórz tabelę do przechowywania wektorów dystrybucyjnych i wstaw wygenerowane wektory. W rzeczywistej aplikacji bazy danych wczytywanie danych do usługi Spanner do kroku 2 byłoby transakcyjne. Aby zachować sprawdzone metody projektowania, wolę, aby tabele transakcyjne były znormalizowane, dlatego tworzę osobną tabelę z wektorami.

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

Gdy treści zbiorcze i wektory dystrybucyjne są gotowe, utwórzmy indeks wyszukiwania wektorowego i punkt końcowy do przechowywania wektorów dystrybucyjnych, które pomogą w przeprowadzaniu wyszukiwania wektorowego.

4. Zadanie przepływu pracy: eksportowanie danych Spannera do wyszukiwania wektorowego

Utworzenie zasobnika Cloud Storage

Jest to wymagane do przechowywania osadzeń z Spannera w zasobniku GCS w formacie JSON, którego Vector Search oczekuje jako danych wejściowych. Utwórz zasobnik w tym samym regionie co dane w usłudze Spanner. W razie potrzeby utwórz w nim folder, ale przede wszystkim utwórz w nim pusty plik o nazwie empty.json.

Konfigurowanie przepływu pracy w Cloud

Aby skonfigurować eksportowanie zbiorcze ze Spannera do indeksu wyszukiwania wektorowego w Vertex AI:

Utwórz pusty indeks:

Sprawdź, czy indeks wyszukiwania wektorowego znajduje się w tym samym regionie co zasobnik Cloud Storage i dane. Wykonaj 11 kroków instrukcji w sekcji Tworzenie indeksu do aktualizacji zbiorczej na karcie konsoli na stronie zarządzania indeksami. W folderze przekazanym do parametru contentsDeltaUri utwórz pusty plik o nazwie empty.json, ponieważ bez tego pliku nie można utworzyć indeksu. Spowoduje to utworzenie pustego indeksu.

Jeśli masz już indeks, możesz pominąć ten krok. Przepływ pracy zastąpi Twój indeks.

Uwaga: nie możesz wdrożyć pustego indeksu w punkcie końcowym. Dlatego odkładamy wdrożenie go w punkcie końcowym na później, po wyeksportowaniu danych wektorowych do Cloud Storage.

Sklonuj to repozytorium Git: repozytorium Git można sklonować na wiele sposobów. Jednym z nich jest uruchomienie tego polecenia za pomocą interfejsu wiersza poleceń GitHub. Uruchom w terminalu Cloud Shell te 2 polecenia:

gh repo clone cloudspannerecosystem/spanner-ai

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

Ten folder zawiera 2 pliki

batch-export.yaml: Jest to definicja przepływu pracy.
sample-batch-input.json: to przykładowe parametry wejściowe przepływu pracy.

Skonfiguruj plik input.json na podstawie pliku przykładowego: najpierw skopiuj przykładowy plik JSON.

cp sample-batch-input.json input.json

Następnie zmień input.json, wpisując szczegóły projektu. W takim przypadku plik JSON powinien wyglądać tak:

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

Konfigurowanie uprawnień

W przypadku środowisk produkcyjnych zdecydowanie zalecamy utworzenie nowego konta usługi i przypisanie do niego co najmniej 1 roli uprawnień, która zawiera minimalne uprawnienia wymagane do zarządzania usługą. Aby skonfigurować przepływ pracy do eksportowania danych ze Spannera (wektorów dystrybucyjnych) do indeksu wyszukiwania wektorowego, potrzebujesz tych ról:

Konto usługi Cloud Workflow:

Domyślnie używa domyślnego konta usługi Compute Engine.

Jeśli używasz ręcznie skonfigurowanego konta usługi, musisz uwzględnić te role:

Aby uruchomić zadanie Dataflow: administrator Dataflow, zasób roboczy Dataflow.

Aby podawać się za konto usługi instancji roboczej Dataflow, musisz mieć uprawnienie Użytkownik konta usługi.

Zapisywanie logów: Zapisujący logi.

Aby wywołać ponowne tworzenie indeksu wyszukiwania wektorowego w Vertex AI: użytkownik Vertex AI.

Konto usługi instancji roboczej Dataflow:

Jeśli używasz ręcznie skonfigurowanego konta usługi, musisz uwzględnić te role:

Do zarządzania przepływem danych: Administrator Dataflow, Zasób roboczy Dataflow. Aby odczytać dane z Spannera: Odczytujący bazę danych Cloud Spanner. Uprawnienia do zapisu w wybranym rejestrze kontenerów GCS: właściciel zasobnika GCS.

Wdrażanie przepływu pracy w Cloud

wdrożyć plik YAML przepływu pracy w projekcie Google Cloud; Możesz skonfigurować region lub lokalizację, w której przepływ pracy będzie uruchamiany.

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"

Przepływ pracy powinien być teraz widoczny na stronie Przepływy pracy w konsoli Google Cloud.

Uwaga: przepływ pracy możesz też utworzyć i wdrożyć w konsoli Google Cloud. Postępuj zgodnie z instrukcjami wyświetlanymi w konsoli Cloud. W przypadku definicji przepływu pracy skopiuj i wklej zawartość pliku batch-export.yaml.

Gdy to zrobisz, uruchom przepływ pracy, aby rozpocząć eksportowanie danych.

Wykonanie przepływu pracy w Cloud Workflow

Aby uruchomić przepływ pracy, wpisz to polecenie:

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

Wykonanie powinno pojawić się na karcie Wykonania w sekcji Przepływy pracy. Spowoduje to załadowanie danych do bazy danych wyszukiwania wektorowego i ich zindeksowanie.

Uwaga: możesz też uruchomić skrypt z konsoli, klikając przycisk Wykonaj. Postępuj zgodnie z instrukcjami, a jako dane wejściowe skopiuj i wklej zawartość dostosowanego pliku input.json.

5. Wdrażanie indeksu wyszukiwania wektorowego

Wdróż indeks w punkcie końcowym

Aby wdrożyć indeks, wykonaj te czynności:

Na stronie Indeksy wyszukiwania wektorowego obok indeksu utworzonego w kroku 2 poprzedniej sekcji powinien być widoczny przycisk WDROŻ. Możesz też przejść na stronę informacji o indeksie i kliknąć przycisk WDROŻ DO PUNKTU KOŃCOWEGO.
Podaj niezbędne informacje i wdroż indeks w punkcie końcowym.

Możesz też przejrzeć ten notatnik, aby wdrożyć go w punkcie końcowym (przejdź do części notatnika dotyczącej wdrażania). Po wdrożeniu zanotuj wdrożony identyfikator indeksu i adres URL punktu końcowego.

6. Frontend: dane użytkownika do wyszukiwania wektorowego

Aby szybko przetestować naszą implementację, utwórzmy prostą aplikację w języku Python z interfejsem użytkownika opartym na bibliotece Gradio. Implementację znajdziesz tutaj. Możesz ją wykorzystać do utworzenia aplikacji w wersji demonstracyjnej we własnym notatniku Colab.

Do wywoływania interfejsu Embeddings API i punktu końcowego indeksu wyszukiwania wektorowego będziemy używać pakietu aiplatform python sdk.

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

Użyjemy biblioteki Gradio, aby szybko i łatwo zaprezentować tworzoną przez nas aplikację AI za pomocą interfejsu. Zanim wykonasz ten krok, uruchom środowisko wykonawcze ponownie.

!pip install gradio
import gradio as gr

Wywołaj interfejs Embeddings API z aplikacji internetowej po wprowadzeniu danych wejściowych użytkownika. Użyjemy modelu wektorów dystrybucyjnych tekstu: textembedding-gecko@latest.

Poniższa metoda wywołuje model wektorów dystrybucyjnych tekstu i zwraca wektory dystrybucyjne dla tekstu wpisanego przez użytkownika:

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

Przetestuj

text_embedding("red shorts for girls")

Dane wyjściowe powinny wyglądać podobnie do tych poniżej (pamiętaj, że obraz jest przycięty w pionie, więc nie widać całej odpowiedzi wektorowej):

Zadeklaruj identyfikator wdrożonego indeksu i identyfikator punktu końcowego.

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

Zdefiniuj metodę wyszukiwania wektorowego, aby wywołać punkt końcowy indeksu i wyświetlić wynik z 10 najbliższymi dopasowaniami do odpowiedzi osadzania odpowiadającej danym wejściowym użytkownika.

W definicji metody wyszukiwania wektorowego poniżej zwróć uwagę, że metoda find_neighbors jest wywoływana w celu zidentyfikowania 10 najbliższych wektorów.

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

Zauważysz też wywołanie metody spanner_read_data. Przyjrzyjmy się temu w następnym kroku.

Zdefiniuj implementację metody odczytu danych Spanner, która wywołuje metodę execute_sql, aby wyodrębnić obrazy odpowiadające identyfikatorom wektorów najbliższych sąsiadów zwróconych w ostatnim kroku.

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

Powinien zwracać adresy URL obrazów odpowiadających wybranym wektorom.

Na koniec połączmy wszystkie elementy w interfejsie użytkownika i uruchommy proces wyszukiwania wektorowego.

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)

Wynik powinien wyglądać tak:

Obraz: link

Obejrzyj film z wynikami: tutaj.

7. Czyszczenie danych

Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby użyte w tym poście, wykonaj te czynności:

W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
Z listy projektów wybierz projekt do usunięcia, a potem kliknij Usuń.
W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.
Jeśli nie chcesz usuwać projektu, usuń instancję Spanner. W tym celu przejdź do instancji utworzonej w tym projekcie i w prawym górnym rogu strony przeglądu instancji kliknij przycisk USUŃ INSTANCJĘ.
Możesz też przejść do indeksu wyszukiwania wektorowego, wycofać wdrożenie punktu końcowego i indeksu oraz usunąć indeks.

8. Podsumowanie

Gratulacje! Wdrożenie usługi Spanner – Vertex Vector Search zostało ukończone.

Tworzenie źródła danych i wektorów osadzania Spannera na potrzeby aplikacji, które pochodzą z bazy danych Spannera.
Tworzenie indeksu bazy danych wyszukiwania wektorowego.
Integrowanie danych wektorowych ze Spannera z wyszukiwarką wektorową za pomocą zadań Dataflow i Workflow.
Wdrażanie indeksu w punkcie końcowym.
Na koniec wywołaj wyszukiwanie wektorowe na podstawie danych wejściowych użytkownika w implementacji pakietu SDK Vertex AI opartej na Pythonie.

Możesz rozszerzyć implementację na własne potrzeby lub ulepszyć obecne zastosowanie za pomocą nowych funkcji. Więcej informacji o funkcjach uczenia maszynowego w usłudze Spanner znajdziesz tutaj.