Produktionsreife Suchpipelines mit Spark und Milvus
Der Aufbau einer skalierbaren Vektorsuch-Pipeline in der Produktion ist nicht so einfach wie der Aufbau ihres Prototyps. Bei der Arbeit an einem Prototyp haben wir es oft nur mit einer kleinen Menge unstrukturierter Daten zu tun. Wenn wir den Prototyp jedoch in die Produktion überführen, müssen wir typischerweise Millionen oder Milliarden unstrukturierter Daten und hohe Anfragevolumina bewältigen. Daher ist eine robuste Lösung für die effiziente Ausführung gängiger Operationen einer Vektorsuch-Pipeline wie Datenaufnahme und Informationsabruf erforderlich.
In einem aktuellen Vortrag stellte Jiang Chen, Head of Ecosystem & AI Platform bei Zilliz, einen schrittweisen Prozess zum Aufbau einer effizienten und produktionsreifen Vektorsuch-Pipeline vor. Dieser Artikel behandelt die wichtigsten Punkte des Vortrags, die aus drei Themen bestehen:
Workflow der Informationssuche in traditionellen und Retrieval Augmented Generation (RAG)-Umgebungen.
Aufbau einer skalierbaren Vektorsuch-Pipeline in einem RAG-System mit Milvus und Spark.
Ratschläge zur Verbesserung der Qualität Ihres RAG-Systems.
Ohne weitere Umschweife sprechen wir über das erste Hauptthema und beginnen mit dem Workflow der Informationssuche in einer traditionellen Umgebung.
Workflow einer traditionellen Informationssuche
Vor den Fortschritten des Deep Learning stützten sich traditionelle Informationsabruf- oder Suchsysteme stark auf Tags und manuelle Kennzeichnung. Betrachten wir den Fall von Online-Shops: Sie verließen sich auf Produkt-Tags, um Kunden die passendsten Produkte entsprechend ihren Bedürfnissen anzubieten. Daher benötigen Online-Shops eine skalierbare und effiziente Such-Pipeline, die es ihnen ermöglicht, täglich eine große Anzahl von Kundenanfragen zu bedienen.
Um dieser Nachfrage gerecht zu werden, wird die Architektur des traditionellen Suchsystems normalerweise in zwei Komponenten unterteilt: eine für die Offline-Datenaufnahme und eine für die Online-Anfragebearbeitung.
Offline-Datenaufnahme
Das Hauptziel der Offline-Datenaufnahme besteht darin, alle Daten in eine Datenbank zu laden. Als erster Schritt dieses Prozesses werden die Daten aus einer oder mehreren Quellen gesammelt, wie etwa internen Dokumenten oder dem Internet. Sobald wir die Daten abrufen können, können wir mit dem Tagging der Daten fortfahren. Schließlich können die getaggten Daten indexiert und in die Datenbank geladen werden.
Verwenden wir das Beispiel des Online-Shops, um den Workflow zu veranschaulichen. Wir können Produktbeschreibungen aus dem Internet beziehen, indem wir das Web crawlen. Sobald wir die Beschreibungen haben, erstellen wir als Nächstes Tags, die diese Produktbeschreibungen repräsentieren, wie 'clothes', 'dress', 'formal dress', 'party dress' usw. Dann erstellen wir den Index der Tags, Produktbeschreibungen, Preise und anderer Metadaten, bevor wir sie in eine unstrukturierte Datenbank laden. Schließlich übertragen wir diese Datenbank in die Serving-Umgebung.
Zwei Komponenten einer traditionellen Architektur für die Informationssuche
Online-Anfragebearbeitung
Das Hauptziel der zweiten Komponente besteht darin, die Anfragen der Kunden zu bedienen und den Informationsabruf aus der Datenbank durchzuführen, die wir im vorherigen Workflow erstellt haben.
Der Prozess beginnt beim Frontend und Query Compiler, um die Anfrage des Benutzers in eine Reihe von Tags zu synthetisieren. Als Nächstes verwendet das System die generierten Tags als Eingaben für die Ähnlichkeitssuche. Die Top-k-Einträge in der Datenbank mit den ähnlichsten Schlüsselwörtern oder Tags werden dann abgerufen und mithilfe eines Algorithmus gerankt, der je nach Anwendungsfall variiert. Das gerankte Ergebnis wird schließlich an den Benutzer zurückgegeben.
Der Hauptnachteil traditioneller Informationsabruf- und Suchsysteme ist das fehlende semantische Verständnis. Tags oder manuell erstellte Labels konnten die semantische Bedeutung und die Absicht der Benutzeranfrage nicht erfassen, was zu ungenauen Suchergebnissen führen kann. Außerdem wäre das manuelle Taggen jedes Eintrags umständlich, wenn wir eine riesige Datenmenge haben.
Der Workflow von RAG als neue Informationssuche
Die rasanten Fortschritte im Deep Learning haben die Landschaft der Informationsabrufprozesse erheblich verändert. Mithilfe von Embedding-Modellen und großen Sprachmodellen (LLMs) wie GPT, Claude, LLAMA und Mistral kann die semantische Bedeutung der Anfrage eines Benutzers effektiv erfasst werden, wodurch die manuelle Erstellung von Labels oder Tags für jeden Dateneintrag entfällt.
Mit Embedding-Modellen können unstrukturierte Daten in Vektor-Embeddings umgewandelt werden, die aus n-dimensionalen Vektoren bestehen. Die Dimensionalität des Embeddings hängt vom verwendeten Modell ab. Diese Embeddings tragen die semantische Bedeutung der Daten, die sie repräsentieren, und daher kann die Ähnlichkeit zwischen zwei beliebigen Embeddings mithilfe von Metriken wie der Kosinusdistanz leicht berechnet werden. Die Intuition dahinter ist, dass Embeddings mit ähnlichen Bedeutungen im Vektorraum näher beieinander platziert werden.
Beispiel für Vektor-Embeddings, die eine ähnliche semantische Bedeutung in einem 2D-Vektorraum tragen
Sobald wir die Embeddings haben, können sie direkt in eine Vektordatenbank wie Milvus aufgenommen werden, womit der Teil der Datenaufnahme abgeschlossen ist. Anschließend kann der Informationsabrufprozess durchgeführt werden.
Wenn eine Benutzeranfrage empfangen wird, wird sie während des Teils der Datenaufnahme mit demselben Embedding-Modell in ein Embedding umgewandelt. Als Nächstes führt die Pipeline eine Vektorsuchoperation aus und ruft die top k ähnlichsten Embeddings aus der Datenbank ab. Im Kontext von Retrieval Augmented Generation (RAG) werden diese ähnlichen Embeddings dann als Kontexte für das LLM verwendet, um die Anfrage des Benutzers zu beantworten.
RAG-Workflow
Vektorsuch-Pipeline mit Spark und Milvus
RAG ist ein neuartiger Ansatz, um die Genauigkeit der von einem LLM generierten Antworten zu verbessern, indem ihm relevante Kontexte bereitgestellt werden, die aus einer Vektorsuche abgerufen wurden. Der Aufbau einer produktionsreifen RAG-Anwendung ist jedoch aufgrund von Skalierbarkeitsproblemen eine Herausforderung.
Wenn Sie eine RAG-Anwendung in der Produktion bereitstellen, werden Sie es höchstwahrscheinlich mit Millionen oder sogar Milliarden unstrukturierter Daten zu tun haben. Zusätzlich wird Ihr RAG-System Tausende oder sogar noch mehr Anfragen von Kunden erhalten. Daher ist eine effiziente und skalierbare Lösung erforderlich, um diese Probleme effektiv zu bewältigen, und hier kann Apache Spark nützlich sein.
In diesem Abschnitt erstellen wir eine Such-Pipeline mit Milvus und Spark. Milvus ist eine Open-Source-Vektordatenbank, die es uns ermöglicht, Vektorsuchen auf riesigen Datenmengen in Sekundenschnelle durchzuführen. Spark ist derweil ein leistungsstarkes Open-Source-Framework für verteiltes Computing, das besonders nützlich ist, um große Datensätze schnell und effizient zu verarbeiten und zu analysieren.
Beginnen wir mit der Installation von Milvus. Es gibt mehrere Möglichkeiten, Milvus zu installieren, aber wenn Sie Milvus in einer Produktionsumgebung verwenden möchten, ist es am besten, wenn Sie Milvus in Docker mit den folgenden Befehlen installieren und ausführen:
# Download the installation script
$ curl -sfL <https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh> -o standalone_embed.sh
# Start the Docker container
$ bash standalone_embed.sh start
Als Open-Source-Vektordatenbank bietet Milvus nahtlose Integrationen mit vielen Tools und KI-Frameworks, wodurch es einfach wird, produktionsreife KI-gestützte Anwendungen wie RAG zu erstellen. Apache Spark ist eines der Frameworks, die zusammen mit Milvus verwendet werden können, um Datenaufnahme- und Anfrageabrufprozesse effizient zu skalieren.
Beispiel für einen Workflow einer Vektorsuch-Pipeline mit Milvus und Spark
Da Spark ein verteiltes Verarbeitungssystem ist, kann es Datenverarbeitungsaufgaben stapelweise auf mehrere Computer verteilen. Diese Funktion beschleunigt die Datenverarbeitung beim Umgang mit enormen Datenmengen, beispielsweise beim Bereitstellen einer RAG-Anwendung in der Produktion. Dank dieser Integration können wir auch Daten zwischen Milvus und anderen Datenbankdiensten wie MySQL verschieben.
Informationen zur Installation von Apache Spark finden Sie in deren neuester Installationsdokumentation. Sobald Sie Spark installiert haben, müssen Sie außerdem die spark-milvus-jar-Datei installieren.
wget <https://github.com/zilliztech/spark-milvus/raw/1.0.0-SNAPSHOT/output/spark-milvus-1.0.0-SNAPSHOT.jar>
Nachdem Sie die spark-milvus-jar-Datei heruntergeladen haben, können Sie sie mit den folgenden Schritten als Abhängigkeit hinzufügen:
# 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
Und nun sind wir bereit, Milvus mit Spark zu integrieren. Im folgenden Beispiel zeigen wir Ihnen, wie Sie Daten aus einem Spark DataFrame direkt in Milvus aufnehmen können.
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()
}
Im oben bereitgestellten Codeausschnitt haben wir einen Spark DataFrame mit drei Feldern in eine Collection namens "hello_spark_milvus" aufgenommen: eine ID, einen Text und ein Vektor-Embedding. Die Embeddings bestehen aus 5-dimensionalen Vektoren, und wir haben die ID als Primärschlüssel unserer Collection verwendet.
Außerdem müssen wir innerhalb der milvusOptions-Map mehrere Konfigurationselemente zu unserer Milvus-Datenbank bereitstellen:
milvus.hostundmilvus.port: Milvus-Server und -Port. Wenn Sie Milvus in Docker ausführen, ist der Standardport 19530.milvus.collection.name: der Collection-Name innerhalb der Milvus-Datenbank, in die die Daten aufgenommen werden.milvus.collection.vectorField: der Spaltenname unserer Daten, der das Vektor-Embedding enthält.milvus.collection.vectorDim: die Dimensionalität unseres Vektor-Embeddings.milvus.collection.primaryKeyField: der Spaltenname unserer Daten, der den Primärschlüssel enthält.
Wenn Sie mehr über die verschiedenen Milvus-Optionen erfahren möchten, die Sie anpassen können, sehen Sie sich die Milvus-Dokumentationsseite an.
Nachdem wir nun Daten in die Milvus-Datenbank aufgenommen haben, müssen wir die Indexierungsmethode für unsere Collection festlegen. Milvus unterstützt verschiedene Indexierungsmethoden, wie den regulären Flat-Index, Inverted Flat Index (IVF), Hierarchical Navigable Small World (HNSW) und viele mehr.
Im folgenden Beispiel verwenden wir AUTOINDEX, eine angepasste Version von HNSW. Als Metrik während der Vektorsuche verwenden wir die L2-Distanz.
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)
Als Nächstes müssen wir unsere „hello_spark_milvus“-Collection laden, bevor wir darauf eine Vektorsuche durchführen können.
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)
Nachdem wir nun die Daten in Milvus aufgenommen und einen Index erstellt haben, sind wir schließlich bereit, eine Vektorsuchoperation durchzuführen. Wir verwenden die erste Zeile des Spark DataFrame, den wir zuvor erstellt haben, als Eingabevektor.
// 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)
Wie Sie sehen können, müssen wir innerhalb der Methode SearchParam.newBuilder() mehrere Methodenaufrufe bereitstellen, um eine Vektorsuchoperation durchzuführen, wie zum Beispiel:
.withCollectionName(): der Name der Collection, in der die Vektorsuche durchgeführt wird..withMetricType(): die Metrik, die zur Durchführung der Vektorsuche verwendet wird..withOutFields(): die Ausgabefelder in einer Collection, die das Ergebnis zurückgeben..withVectors(): der Eingabe- oder Abfragevektor..withVectorFieldName(): das Feld in einer Collection, das Vektor-Embeddings enthält..withTopK(): gibt die obersten k Einträge zurück, die die ähnlichsten Embeddings zum Abfragevektor aufweisen.
In einer RAG-Anwendung werden die obersten k ähnlichsten Einträge als Kontexte verwendet, die zusammen mit der Abfrage an das LLM übergeben werden. Auf diese Weise kann das LLM die Kontexte verwenden, um eine genaue Antwort auf die Abfrage zu generieren.
Es gibt viele fortgeschrittenere Anwendungsfälle, in denen Sie die Milvus-Integration mit Spark nutzen können. Beispielsweise können Sie Daten aus Ihrer regulären Datenbank wie MySQL lesen, sie in Vektor-Embeddings umwandeln und diese Embeddings in Milvus aufnehmen. Sie können diese Anwendungsfälle in diesem GitHub-Repository oder in dieser Milvus-Notebook-Demo erkunden.
Gutes RAG entsteht aus guten Daten
Die Integration von Milvus mit vielen KI-Toolkits und Frameworks vereinfacht die Entwicklung produktionsreifer RAG-Anwendungen (Retrieval Augmented Generation). Nach der Bereitstellung eines RAG-Systems in der Produktion ist es jedoch wichtig, die Qualität der vom System generierten Antworten kontinuierlich zu überwachen.
Wenn die Antwortqualität verbessert werden muss, ist es entscheidend, zunächst die Grundlagen zu prüfen, bevor man sich mit komplexeren Algorithmen befasst. Der Hauptfokus sollte auf der Qualität der Datenquelle liegen, die vom RAG-System verwendet wird.
Berücksichtigen Sie bei der Bewertung der Datenquelle die folgenden Fragen:
Haben wir die erforderlichen Daten in unserer Datenbank, um die Abfragen des Benutzers zu beantworten?
Haben wir alle notwendigen Daten in unserer Datenbank gesammelt?
Haben wir die korrekten Datenvorverarbeitungsschritte durchgeführt, bevor wir die Daten in unsere Datenbank aufgenommen haben (z. B. Datenparsing, Datenbereinigung, Chunking, Verwendung geeigneter Embedding-Modelle)?
Sobald Sie die Qualität der Datenquelle überprüft haben, können Sie erwägen, die Qualität des RAG-Systems aus algorithmischer Perspektive zu verbessern. Es gibt mehrere Möglichkeiten, die Leistung eines RAG-Systems zu steigern, wie zum Beispiel:
Verwendung leistungsfähigerer Embedding-Modelle: Experimentieren Sie mit verschiedenen vortrainierten oder individuell trainierten Embedding-Modellen, um dasjenige zu finden, das die semantischen Beziehungen in Ihren Daten am besten erfasst.
Implementierung von Query Routing und Integration von Drittanbieter-Tools: Wenn das Embedding-Modell nicht das Problem ist, können Sie das RAG-System verbessern, indem Sie einen Agenten für Query Routing einsetzen und zusätzliche Tools oder Datenquellen integrieren.
Indem Sie sich auf die Grundlagen konzentrieren und die Datenquelle sowie die algorithmischen Komponenten kontinuierlich iterieren, können Sie sicherstellen, dass Ihre produktionsreife RAG-Anwendung Ihren Benutzern hochwertige Antworten liefert.
Fazit
Die nahtlose Integration von Milvus mit verschiedenen Frameworks wie Spark erleichtert es uns, eine skalierbare LLM-gestützte Anwendung zu erstellen und bereitzustellen. Die Fähigkeit von Spark, Datenverarbeitungsaufgaben stapelweise auf mehrere Computer zu verteilen, beschleunigt Datenverarbeitungsvorgänge erheblich. Diese Funktion ist besonders hilfreich, wenn wir riesige Datenmengen in unsere Vektordatenbank aufnehmen möchten oder wenn unsere Anwendung gleichzeitig mit einer großen Anzahl von Benutzerabfragen umgehen muss.
Sobald wir Daten in unsere Milvus-Vektordatenbank aufgenommen haben und die Abfrage eines Benutzers erhalten, können wir eine Vektorsuche durchführen. Dieser Prozess ist entscheidend, um aus den Daten in unserer Datenbank die relevantesten Kontexte abzurufen, die an ein LLM übergeben werden, um stark kontextualisierte Antworten zu generieren.
Weiterlesen
Why We Built Vector Lakebase: Rethinking Unstructured Data Architecture for AI
Vector Lakebase: a unified, lake-native data foundation for AI workloads — and an answer to what happens after vector databases succeed.
Zilliz Named "Highest Performer" and "Easiest to Use" in G2's Summer 2025 Grid® Report for Vector Databases
Zilliz shines in G2's Summer 2025 Grid® Report as both "Highest Performer" and "Easiest to Use," solving the performance-usability dilemma.
Why Deepseek is Waking up AI Giants Like OpenAI And Why You Should Care
Discover how DeepSeek R1's open-source AI model with superior reasoning capabilities and lower costs is disrupting the AI landscape and challenging tech giants like OpenAI.