Creare pipeline di ricerca pronte per la produzione con Spark e Milvus
Costruire una pipeline di ricerca vettoriale scalabile in produzione non è facile quanto costruirne il prototipo. Quando lavoriamo su un prototipo, spesso gestiamo solo una piccola quantità di dati non strutturati. Tuttavia, quando spostiamo il prototipo in produzione, in genere dobbiamo gestire milioni o miliardi di dati non strutturati e volumi elevati di query. Pertanto, è necessaria una soluzione robusta per l'esecuzione efficiente delle operazioni comuni della pipeline di ricerca vettoriale, come l'ingestione dei dati e il recupero delle informazioni.
In un recente intervento, Jiang Chen, Head of Ecosystem & AI Platform presso Zilliz, ha presentato un processo passo dopo passo per costruire una pipeline di ricerca vettoriale efficiente e pronta per la produzione. Questo articolo discuterà i punti principali dell'intervento, che consistono in tre argomenti:
Flusso di lavoro della ricerca di informazioni in contesti tradizionali e di Retrieval Augmented Generation (RAG).
Costruire una pipeline di ricerca vettoriale scalabile in un sistema RAG con Milvus e Spark.
Consigli per migliorare la qualità del tuo sistema RAG.
Senza ulteriori indugi, parliamo del primo argomento principale e iniziamo con il flusso di lavoro della ricerca di informazioni in un contesto tradizionale.
Flusso di lavoro di una ricerca di informazioni tradizionale
Prima dei progressi del deep learning, i sistemi tradizionali di recupero delle informazioni o di ricerca si basavano fortemente su tag ed etichettatura manuale. Considera il caso dei negozi online: si affidavano ai tag dei prodotti per offrire ai clienti i prodotti più adatti in base alle loro esigenze. Pertanto, i negozi online hanno bisogno di una pipeline di ricerca scalabile ed efficiente che consenta loro di servire ogni giorno un gran numero di query dei clienti.
Per soddisfare questa domanda, l'architettura del sistema di ricerca tradizionale è normalmente divisa in due componenti: uno per l'ingestione dei dati offline e uno per il servizio delle query online.
Ingestione dei dati offline
L'obiettivo principale dell'ingestione dei dati offline è caricare tutti i dati in un database. Come primo passo di questo processo, i dati vengono raccolti da una o più fonti, come documenti interni o Internet. Una volta che possiamo recuperare i dati, possiamo continuare con l'assegnazione dei tag. Infine, i dati con tag possono essere indicizzati e caricati nel database.
Usiamo l'esempio del negozio online per illustrare il flusso di lavoro. Possiamo ottenere descrizioni dei prodotti da Internet effettuando il crawling del web. Successivamente, una volta ottenute le descrizioni, creiamo tag che rappresentano tali descrizioni dei prodotti, come 'clothes', 'dress', 'formal dress', 'party dress', ecc. Poi, costruiamo l'indice dei tag, delle descrizioni dei prodotti, dei prezzi e di altri metadati prima di caricarli in un database non strutturato. Infine, inviamo questo database all'ambiente di servizio.
Due componenti di un'architettura di ricerca di informazioni tradizionale
Servizio delle query online
L'obiettivo principale del secondo componente è servire le query dei clienti ed eseguire il recupero delle informazioni dal database che abbiamo creato nel flusso di lavoro precedente.
Il processo parte dal front-end e dal compilatore di query per sintetizzare la query dell'utente in un insieme di tag. Successivamente, il sistema userà i tag generati come input per la ricerca di similarità. Le prime k voci nel database con le parole chiave o i tag più simili verranno quindi recuperate e classificate utilizzando un algoritmo che varia in base ai casi d'uso. Il risultato classificato verrà infine restituito all'utente.
Il principale svantaggio dei sistemi tradizionali di recupero delle informazioni e di ricerca è la mancanza di comprensione semantica. I tag o le etichette create manualmente non riuscivano a cogliere il significato semantico e l'intento della query dell'utente, il che poteva portare a risultati di ricerca imprecisi. Inoltre, assegnare manualmente tag a ogni voce sarebbe macchinoso se avessimo una quantità enorme di dati.
Il flusso di lavoro di RAG come nuova ricerca di informazioni
I rapidi progressi del deep learning hanno cambiato significativamente il panorama dei processi di recupero delle informazioni. Con l'aiuto dei modelli di embedding e dei modelli linguistici di grandi dimensioni (LLM) come GPT, Claude, LLAMA e Mistral, il significato semantico della query di un utente può essere catturato efficacemente, eliminando la necessità di creare manualmente etichette o tag per ogni voce di dati.
Con i modelli di embedding, i dati non strutturati possono essere trasformati in embedding vettoriali, che consistono in vettori n-dimensionali. La dimensionalità dell'embedding dipende dal modello utilizzato. Questi embedding portano il significato semantico dei dati che rappresentano e, pertanto, la somiglianza tra due embedding qualsiasi può essere facilmente calcolata usando metriche come la distanza coseno. L'intuizione è che gli embedding che portano significati simili saranno posizionati più vicini tra loro nello spazio vettoriale.
Esempio di embedding vettoriali che portano significato semantico simile in uno spazio vettoriale 2D
Una volta ottenuti gli embedding, possono essere direttamente importati in un database vettoriale come Milvus, completando la parte di ingestione dei dati. Successivamente, può essere condotto il processo di recupero delle informazioni.
Quando viene ricevuta una query utente, verrà trasformata in un embedding utilizzando lo stesso modello di embedding usato durante la parte di ingestione dei dati. Successivamente, la pipeline esegue un'operazione di ricerca vettoriale e recupera i primi k embedding più simili dal database. Nel contesto della Retrieval Augmented Generation (RAG), questi embedding simili vengono quindi utilizzati come contesti per l'LLM per rispondere alla query dell'utente.
Flusso di lavoro RAG
Pipeline di ricerca vettoriale con Spark e Milvus
La RAG è un approccio innovativo per migliorare l'accuratezza delle risposte generate da un LLM fornendogli contesti pertinenti recuperati da una ricerca vettoriale. Tuttavia, costruire un'applicazione RAG pronta per la produzione è impegnativo a causa dei problemi di scalabilità.
Quando distribuisci un'applicazione RAG in produzione, molto probabilmente avrai a che fare con milioni o persino miliardi di dati non strutturati. Inoltre, il tuo sistema RAG riceverà migliaia o persino più query dai clienti. Pertanto, è necessaria una soluzione efficiente e scalabile per gestire efficacemente questi problemi, ed è qui che Apache Spark può essere utile.
In questa sezione, costruiremo una pipeline di ricerca usando Milvus e Spark. Milvus è un database vettoriale open-source che ci consente di eseguire ricerche vettoriali su enormi quantità di dati in pochi secondi. Nel frattempo, Spark è un potente framework di calcolo distribuito open-source, particolarmente utile per elaborare e analizzare grandi dataset in modo rapido ed efficiente.
Iniziamo installando Milvus. Esistono diversi modi per installare Milvus, ma se vuoi usare Milvus in un ambiente di produzione, è meglio installare ed eseguire Milvus in Docker con i seguenti comandi:
# Scarica lo script di installazione
$ curl -sfL <https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh> -o standalone_embed.sh
# Avvia il container Docker
$ bash standalone_embed.sh start
In quanto database vettoriale open-source, Milvus offre integrazioni fluide con molti strumenti e framework di IA, rendendo facile costruire applicazioni basate sull'IA pronte per la produzione come la RAG. Apache Spark è uno dei framework che può essere utilizzato insieme a Milvus per scalare in modo efficiente i processi di ingestione dei dati e di recupero delle query.
Esempio di flusso di lavoro di una pipeline di ricerca vettoriale con Milvus e Spark
Poiché Spark è un sistema di elaborazione distribuita, è in grado di distribuire le attività di elaborazione dei dati su più computer in batch. Questa funzionalità accelera l'elaborazione dei dati quando si gestiscono enormi quantità di dati, ad esempio quando si distribuisce un'applicazione RAG in produzione. Grazie a questa integrazione, possiamo anche spostare i dati tra Milvus e altri servizi di database come MySQL.
Per installare Apache Spark, consulta la loro documentazione di installazione più recente. Una volta installato Spark, dovrai anche installare il file jar spark-milvus.
wget <https://github.com/zilliztech/spark-milvus/raw/1.0.0-SNAPSHOT/output/spark-milvus-1.0.0-SNAPSHOT.jar>
Una volta scaricato il file jar spark-milvus, puoi aggiungerlo come dipendenza seguendo questi passaggi:
# For pyspark
./bin/pyspark --jars spark-milvus-1.0.0-SNAPSHOT.jar
# For spark-shell
./bin/spark-shell --jars spark-milvus-1.0.0-SNAPSHOT.jar
E ora siamo pronti a integrare Milvus con Spark. Nell'esempio seguente, ti mostreremo come puoi importare dati dal dataframe Spark direttamente in Milvus.
import org.apache.spark.sql.{SaveMode, SparkSession}
import io.milvus.client.{MilvusClient, MilvusServiceClient}
import io.milvus.grpc.{DataType, FlushResponse, ImportResponse}
import io.milvus.param.bulkinsert.{BulkInsertParam, GetBulkInsertStateParam}
import io.milvus.param.collection.{CreateCollectionParam, DescribeCollectionParam, FieldType, FlushParam, LoadCollectionParam}
import io.milvus.param.dml.SearchParam
import io.milvus.param.index.CreateIndexParam
import io.milvus.param.{ConnectParam, IndexType, MetricType, R, RpcStatus}
import zilliztech.spark.milvus.{MilvusOptions, MilvusUtils}
import zilliztech.spark.milvus.MilvusOptions._
import org.apache.spark.SparkConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.log4j.Logger
import org.slf4j.LoggerFactory
import java.util
import scala.collection.JavaConverters._
object Hello extends App {
val spark = SparkSession.builder().master("local[*]")
.appName("HelloSparkMilvus")
.getOrCreate()
import spark.implicits._
// Create DataFrame
val sampleDF = Seq(
(1, "a", Seq(1.0,2.0,3.0,4.0,5.0)),
(2, "b", Seq(1.0,2.0,3.0,4.0,5.0)),
(3, "c", Seq(1.0,2.0,3.0,4.0,5.0)),
(4, "d", Seq(1.0,2.0,3.0,4.0,5.0))
).toDF("id", "text", "vec")
// set milvus options
val milvusOptions = Map(
"milvus.host" -> "localhost" -> uri,
"milvus.port" -> "19530",
"milvus.collection.name" -> "hello_spark_milvus",
"milvus.collection.vectorField" -> "vec",
"milvus.collection.vectorDim" -> "5",
"milvus.collection.primaryKeyField", "id"
)
sampleDF.write.format("milvus")
.options(milvusOptions)
.mode(SaveMode.Append)
.save()
}
Nel frammento di codice fornito sopra, abbiamo importato un Spark DataFrame con tre campi: un ID, un testo e un embedding vettoriale in una collection chiamata "hello_spark_milvus". Gli embedding sono costituiti da vettori a 5 dimensioni e abbiamo usato l'ID come chiave primaria della nostra collection.
Dobbiamo anche fornire diversi elementi di configurazione sul nostro database Milvus all'interno della mappa milvusOptions:
milvus.hostemilvus.port: server e porta Milvus. Se esegui Milvus in Docker, la porta predefinita è 19530.milvus.collection.name: il nome della collection all'interno del database Milvus in cui verranno importati i dati.milvus.collection.vectorField: il nome della colonna dei nostri dati che contiene l'embedding vettoriale.milvus.collection.vectorDim: la dimensionalità del nostro embedding vettoriale.milvus.collection.primaryKeyField: il nome della colonna dei nostri dati che contiene la chiave primaria.
Se desideri saperne di più sui diversi tipi di opzioni Milvus che puoi regolare, consulta la pagina della documentazione Milvus.
Ora che abbiamo ingerito i dati nel database Milvus, dobbiamo specificare il metodo di indicizzazione per la nostra collection. Milvus supporta vari metodi di indicizzazione, come il normale indice Flat, Inverted Flat Index (IVF), Hierarchical Navigable Small World (HNSW) e molti altri.
Nell’esempio seguente, useremo AUTOINDEX, che è una versione personalizzata di HNSW. Come metrica durante la ricerca vettoriale, useremo la distanza L2.
val username = <YOUR_MILVUS_USER>
val password = <YOUR_MILVUS_PASSWORD>
val connectParam: ConnectParam = ConnectParam.newBuilder
.withHost("localhost")
.withPort("19530")
.withAuthorization(username, password)
.build
val client: MilvusClient = new MilvusServiceClient(connectParam)
val createIndexParam = CreateIndexParam.newBuilder()
.withCollectionName("hello_spark_milvus")
.withIndexName("index_name")
.withFieldName("vec")
.withMetricType(MetricType.L2)
.withIndexType(IndexType.AUTOINDEX)
.build()
val createIndexR = client.createIndex(createIndexParam)
println(createIndexR)
Successivamente, dobbiamo caricare la nostra collection “hello_spark_milvus” prima di poter eseguire una ricerca vettoriale su di essa.
import io.milvus.param.collection.{CreateCollectionParam, DescribeCollectionParam, FieldType, FlushParam, LoadCollectionParam}
// Load collection, only loaded collection can be searched
val loadCollectionParam = LoadCollectionParam.newBuilder().withCollectionName("hello_spark_milvus").build()
val loadCollectionR = client.loadCollection(loadCollectionParam)
println(loadCollectionR)
Ora che abbiamo ingerito i dati in Milvus e creato un indice, finalmente siamo pronti a eseguire un’operazione di ricerca vettoriale. Useremo la prima riga del DataFrame Spark che abbiamo creato in precedenza come vettore di input.
// Search, use the first row of input dataframe as search vector
val fieldList: util.List[String] = new util.ArrayList[String]()
fieldList.add("vec")
val searchVectors = util.Arrays.asList(sampleDF.first().getList(2))
val searchParam = SearchParam.newBuilder()
.withCollectionName("hello_spark_milvus")
.withMetricType(MetricType.L2)
.withOutFields("text")
.withVectors(searchVectors)
.withVectorFieldName("vec")
.withTopK(2)
.build()
val searchParamR = client.search(searchParam)
println(searchParamR)
Come puoi vedere, dobbiamo fornire diverse chiamate di metodo all’interno del metodo SearchParam.newBuilder() per eseguire un’operazione di ricerca vettoriale, come:
.withCollectionName(): il nome della collection in cui viene eseguita la ricerca vettoriale..withMetricType(): la metrica utilizzata per eseguire la ricerca vettoriale..withOutFields(): i campi di output in una collection per restituire il risultato..withVectors(): il vettore di input o di query..withVectorFieldName(): il campo in una collection che contiene gli embedding vettoriali..withTopK(): restituisce le prime k voci che hanno gli embedding più simili al vettore di query.
In un’applicazione RAG, le prime k voci più simili verranno utilizzate come contesti da passare insieme alla query all’LLM. In questo modo, l’LLM può utilizzare i contesti per generare una risposta accurata alla query.
Esistono molti altri casi d'uso avanzati in cui puoi sfruttare l'integrazione di Milvus con Spark. Ad esempio, puoi leggere dati dal tuo database tradizionale come MySQL, trasformarli in embedding vettoriali e inserire tali embedding in Milvus. Puoi esplorare questi casi d'uso in questo repository GitHub o in questa demo notebook di Milvus.
Un buon RAG nasce da buoni dati
L'integrazione di Milvus con molti toolkit e framework AI semplifica lo sviluppo di applicazioni RAG (Retrieval Augmented Generation) pronte per la produzione. Tuttavia, dopo aver distribuito un sistema RAG in produzione, è importante monitorare continuamente la qualità delle risposte generate dal sistema.
Se la qualità delle risposte deve essere migliorata, è fondamentale esaminare prima gli elementi di base prima di immergersi in algoritmi più complessi. L'attenzione principale dovrebbe essere sulla qualità della fonte dati utilizzata dal sistema RAG.
Quando si valuta la fonte dati, considera le seguenti domande:
Disponiamo dei dati necessari per rispondere alle query dell'utente nel nostro database?
Abbiamo raccolto tutti i dati necessari nel nostro database?
Abbiamo eseguito i corretti passaggi di pre-elaborazione dei dati prima di inserire i dati nel nostro database (ad es., parsing dei dati, pulizia dei dati, chunking, utilizzo di modelli di embedding appropriati)?
Una volta verificata la qualità della fonte dati, puoi quindi considerare di migliorare la qualità del sistema RAG da una prospettiva algoritmica. Esistono diversi modi per migliorare le prestazioni di un sistema RAG, come:
Utilizzare modelli di embedding più potenti: Sperimenta con diversi modelli di embedding pre-addestrati o addestrati su misura per trovare quello che cattura meglio le relazioni semantiche nei tuoi dati.
Implementare il routing delle query e l'integrazione con strumenti di terze parti: Se il modello di embedding non è il problema, puoi migliorare il sistema RAG applicando un agent per il routing delle query e integrandolo con strumenti o fonti dati aggiuntivi.
Concentrandoti sugli elementii di base e iterando continuamente sulla fonte dati e sui componenti algoritmici, puoi garantire che la tua applicazione RAG pronta per la produzione offra risposte di alta qualità ai tuoi utenti.
Conclusione
La perfetta integrazione di Milvus con vari framework come Spark ci permette di costruire e distribuire facilmente un'applicazione scalabile basata su LLM. La capacità di Spark di distribuire attività di elaborazione dei dati su più computer in batch accelera davvero le operazioni di elaborazione dei dati. Questa funzionalità è particolarmente utile quando vogliamo inserire enormi quantità di dati nel nostro database vettoriale o quando la nostra applicazione gestisce contemporaneamente un gran numero di query degli utenti.
Una volta inseriti i dati nel nostro database vettoriale Milvus e ricevuta la query di un utente, possiamo quindi eseguire una ricerca vettoriale. Questo processo è cruciale per recuperare i contesti più rilevanti tra i dati all'interno del nostro database da passare a un LLM per generare risposte altamente contestualizzate.
Continua a leggere
Stop Building AI Data Infra for the Wrong Stage
Learn how AI data infrastructure should evolve from prototype to enterprise scale, and when Vector Lakebase becomes the right architecture for AI apps.
Zilliz Cloud Just Landed in Claude Code
The Zilliz Cloud Plugin brings the full power of Zilliz Cloud directly into your Claude Code terminal as natural-language conversations.
Data Deduplication at Trillion Scale: How to Solve the Biggest Bottleneck of LLM Training
Explore how MinHash LSH and Milvus handle data deduplication at the trillion-scale level, solving key bottlenecks in LLM training for improved AI model performance.