Criando Pipelines de Busca Prontos para Produção com Spark e Milvus
Construir um pipeline de busca vetorial escalável em produção não é tão fácil quanto construir seu protótipo. Ao trabalhar em um protótipo, frequentemente lidamos com apenas uma pequena quantidade de dados não estruturados. No entanto, ao levar o protótipo para produção, normalmente precisamos lidar com milhões ou bilhões de dados não estruturados e altos volumes de consultas. Portanto, é necessária uma solução robusta para a execução eficiente de operações comuns de um pipeline de busca vetorial, como ingestão de dados e recuperação de informações.
Em uma palestra recente, Jiang Chen, Head of Ecosystem & AI Platform na Zilliz, apresentou um processo passo a passo para construir um pipeline de busca vetorial eficiente e pronto para produção. Este artigo discutirá os principais pontos da palestra, que consistem em três tópicos:
Fluxo de trabalho da busca de informações em configurações tradicionais e de Retrieval Augmented Generation (RAG).
Construção de um pipeline de busca vetorial escalável em um sistema RAG com Milvus e Spark.
Conselhos para melhorar a qualidade do seu sistema RAG.
Sem mais delongas, vamos falar sobre o primeiro tópico principal e começamos com o fluxo de trabalho da busca de informações em uma configuração tradicional.
Fluxo de Trabalho de uma Busca de Informações Tradicional
Antes dos avanços do deep learning, os sistemas tradicionais de recuperação ou busca de informações dependiam fortemente de tags e rotulagem manual. Considere o caso de lojas online: elas dependiam de tags de produtos para oferecer aos clientes os produtos mais adequados de acordo com suas necessidades. Portanto, lojas online precisam de um pipeline de busca escalável e eficiente que lhes permita atender a um grande número de consultas de clientes todos os dias.
Para acomodar essa demanda, a arquitetura do sistema de busca tradicional normalmente é dividida em dois componentes: um para ingestão de dados offline e outro para atendimento de consultas online.
Ingestão de Dados Offline
O principal objetivo da ingestão de dados offline é carregar todos os dados em um banco de dados. Como primeira etapa desse processo, os dados são reunidos a partir de uma ou várias fontes, como documentos internos ou a Internet. Depois que conseguimos buscar os dados, podemos continuar com a marcação de dados. Por fim, os dados marcados podem ser indexados e carregados no banco de dados.
Vamos usar o exemplo da loja online para ilustrar o fluxo de trabalho. Podemos obter descrições de produtos da Internet rastreando a web. Em seguida, depois de termos as descrições, criamos tags que representam essas descrições de produtos, como 'clothes', 'dress', 'formal dress', 'party dress', etc. Então, construímos o índice das tags, descrições de produtos, preços e outros metadados antes de carregá-los em um banco de dados não estruturado. Por fim, enviamos esse banco de dados para o ambiente de atendimento.
Dois componentes de uma arquitetura tradicional de busca de informações
Atendimento de Consultas Online
O principal objetivo do segundo componente é atender às consultas dos clientes e realizar a recuperação de informações a partir do banco de dados que criamos no fluxo de trabalho anterior.
O processo começa no front-end e no compilador de consultas para sintetizar a consulta do usuário em um conjunto de tags. Em seguida, o sistema usará as tags geradas como entradas para a busca por similaridade. As principais k entradas no banco de dados com as palavras-chave ou tags mais semelhantes serão então buscadas e classificadas usando um algoritmo que varia com base nos casos de uso. O resultado classificado será finalmente retornado ao usuário.
A principal desvantagem dos sistemas tradicionais de recuperação e busca de informações é a falta de compreensão semântica. Tags ou rótulos criados manualmente não conseguiam capturar o significado semântico e a intenção da consulta do usuário, o que poderia levar a resultados de busca imprecisos. Além disso, marcar manualmente cada entrada seria trabalhoso se tivermos uma enorme quantidade de dados.
O Fluxo de Trabalho do RAG como a Nova Busca de Informações
Os avanços rápidos do deep learning mudaram significativamente o cenário dos processos de recuperação de informações. Com a ajuda de modelos de embedding e grandes modelos de linguagem (LLMs) como GPT, Claude, LLAMA e Mistral, o significado semântico da consulta de um usuário pode ser capturado de forma eficaz, eliminando a necessidade de criar manualmente rótulos ou tags para cada entrada de dados.
Com modelos de embedding, dados não estruturados podem ser transformados em embeddings vetoriais, que consistem em vetores n-dimensionais. A dimensionalidade do embedding depende do modelo usado. Esses embeddings carregam o significado semântico dos dados que representam e, portanto, a similaridade entre quaisquer dois embeddings pode ser facilmente calculada usando métricas como a distância do cosseno. A intuição é que embeddings que carregam significados semelhantes serão posicionados mais próximos uns dos outros no espaço vetorial.
Exemplo de embeddings vetoriais que carregam significado semântico semelhante em um espaço vetorial 2D
Depois de termos os embeddings, eles podem ser ingeridos diretamente em um banco de dados vetorial como o Milvus, concluindo a parte de ingestão de dados. Em seguida, o processo de recuperação de informações pode ser realizado.
Quando uma consulta de usuário é recebida, ela será transformada em um embedding usando o mesmo modelo de embedding durante a parte de ingestão de dados. Em seguida, o pipeline executa uma operação de busca vetorial e recupera os k embeddings mais semelhantes do banco de dados. No contexto de Geração Aumentada por Recuperação (RAG), esses embeddings semelhantes são então usados como contextos para que o LLM responda à consulta do usuário.
Fluxo de trabalho RAG
Pipeline de Busca Vetorial com Spark e Milvus
RAG é uma abordagem inovadora para melhorar a precisão das respostas geradas por um LLM, fornecendo a ele contextos relevantes recuperados de uma busca vetorial. No entanto, criar uma aplicação RAG pronta para produção é desafiador devido a problemas de escalabilidade.
Ao implantar uma aplicação RAG em produção, você provavelmente lidará com milhões ou até bilhões de dados não estruturados. Além disso, seu sistema RAG receberá milhares ou até mais consultas de clientes. Portanto, uma solução eficiente e escalável é necessária para lidar com esses problemas de forma eficaz, e é aqui que o Apache Spark pode ser útil.
Nesta seção, construiremos um pipeline de busca usando Milvus e Spark. Milvus é um banco de dados vetorial open-source que nos permite realizar busca vetorial em enormes quantidades de dados em segundos. Enquanto isso, Spark é um poderoso framework open-source de computação distribuída, particularmente útil para processar e analisar grandes conjuntos de dados de maneira rápida e eficiente.
Vamos começar instalando o Milvus. Existem várias maneiras de instalar o Milvus, mas se você quiser usar o Milvus em um ambiente de produção, é melhor instalar e executar o Milvus no Docker com os seguintes comandos:
# Baixar o script de instalação
$ curl -sfL <https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh> -o standalone_embed.sh
# Iniciar o contêiner Docker
$ bash standalone_embed.sh start
Como um banco de dados vetorial open-source, Milvus oferece integrações perfeitas com muitas ferramentas e frameworks de IA, facilitando a criação de aplicações baseadas em IA prontas para produção, como RAG. Apache Spark é um dos frameworks que podem ser usados junto com o Milvus para escalar de forma eficiente os processos de ingestão de dados e recuperação de consultas.
Exemplo de um fluxo de trabalho de pipeline de busca vetorial com Milvus e Spark
Como o Spark é um sistema de processamento distribuído, ele é capaz de distribuir tarefas de processamento de dados por vários computadores em um lote. Esse recurso acelera o processamento de dados ao lidar com grandes quantidades de dados, como ao implantar uma aplicação RAG em produção. Graças a essa integração, também podemos mover dados entre o Milvus e outros serviços de banco de dados, como MySQL.
Para instalar o Apache Spark, consulte a documentação de instalação mais recente. Depois de instalar o Spark, você também precisará instalar o arquivo jar spark-milvus.
wget <https://github.com/zilliztech/spark-milvus/raw/1.0.0-SNAPSHOT/output/spark-milvus-1.0.0-SNAPSHOT.jar>
Depois de baixar o arquivo jar spark-milvus, você pode adicioná-lo como uma dependência seguindo estas etapas:
# 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 agora estamos prontos para integrar o Milvus ao Spark. No exemplo a seguir, mostraremos como você pode ingerir dados do dataframe do Spark diretamente no 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()
}
No trecho de código fornecido acima, ingerimos um DataFrame do Spark com três campos: um ID, um texto e uma incorporação vetorial em uma coleção chamada "hello_spark_milvus". As incorporações consistem em vetores de 5 dimensões, e usamos o ID como a chave primária da nossa coleção.
Também precisamos fornecer várias informações de configuração sobre nosso banco de dados Milvus dentro do mapa milvusOptions:
milvus.hostandmilvus.port: servidor e porta do Milvus. Se você executar o Milvus no Docker, a porta padrão é 19530.milvus.collection.name: o nome da coleção dentro do banco de dados Milvus onde os dados serão ingeridos.milvus.collection.vectorField: o nome da coluna dos nossos dados que contém a incorporação vetorial.milvus.collection.vectorDim: a dimensionalidade da nossa incorporação vetorial.milvus.collection.primaryKeyField: o nome da coluna dos nossos dados que contém a chave primária.
Se você quiser saber mais sobre os diferentes tipos de opções do Milvus que você pode ajustar, confira a página de documentação do Milvus.
Agora que ingerimos dados no banco de dados Milvus, precisamos especificar o método de indexação para nossa coleção. O Milvus oferece suporte a vários métodos de indexação, como o índice Flat regular, Inverted Flat Index (IVF), Hierarchical Navigable Small World (HNSW) e muitos outros.
No exemplo a seguir, usaremos AUTOINDEX, que é uma versão personalizada do HNSW. Como métrica durante a busca vetorial, usaremos a distância 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)
Em seguida, precisamos carregar nossa coleção “hello_spark_milvus” antes de podermos realizar uma busca vetorial nela.
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)
Agora que ingerimos os dados no Milvus e criamos um índice, finalmente estamos prontos para realizar uma operação de busca vetorial. Usaremos a primeira linha do Spark DataFrame que criamos anteriormente como o vetor de entrada.
// 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)
Como você pode ver, precisamos fornecer várias chamadas de método dentro do método SearchParam.newBuilder() para realizar uma operação de busca vetorial, como:
.withCollectionName(): o nome da coleção onde a busca vetorial é realizada..withMetricType(): a métrica usada para realizar a busca vetorial..withOutFields(): os campos de saída em uma coleção para retornar o resultado..withVectors(): o vetor de entrada ou consulta..withVectorFieldName(): o campo em uma coleção que contém embeddings vetoriais..withTopK(): retorna as k principais entradas que têm os embeddings mais semelhantes ao vetor de consulta.
Em uma aplicação RAG, as k entradas mais semelhantes serão usadas como contextos a serem passados junto com a consulta para o LLM. Dessa forma, o LLM pode usar os contextos para gerar uma resposta precisa à consulta.
Há muitos casos de uso mais avançados nos quais você pode aproveitar a integração do Milvus com o Spark. Por exemplo, você pode ler dados do seu banco de dados regular, como MySQL, transformá-los em embeddings vetoriais e ingerir esses embeddings no Milvus. Você pode explorar esses casos de uso neste repositório do GitHub ou nesta demonstração em notebook do Milvus.
Um bom RAG vem de bons dados
A integração do Milvus com muitos toolkits e frameworks de IA simplifica o desenvolvimento de aplicações RAG (Retrieval Augmented Generation) prontas para produção. No entanto, depois de implantar um sistema RAG em produção, é importante monitorar continuamente a qualidade das respostas geradas pelo sistema.
Se a qualidade das respostas precisar ser melhorada, é crucial primeiro examinar os fundamentos antes de se aprofundar em algoritmos mais complexos. O foco principal deve estar na qualidade da fonte de dados usada pelo sistema RAG.
Ao avaliar a fonte de dados, considere as seguintes perguntas:
Temos os dados necessários para responder às consultas do usuário em nosso banco de dados?
Coletamos todos os dados necessários em nosso banco de dados?
Realizamos as etapas corretas de pré-processamento de dados antes de ingerir os dados em nosso banco de dados (por exemplo, análise de dados, limpeza de dados, chunking, uso de modelos de embedding apropriados)?
Depois de verificar a qualidade da fonte de dados, você pode então considerar melhorar a qualidade do sistema RAG a partir de uma perspectiva algorítmica. Há várias maneiras de melhorar o desempenho de um sistema RAG, como:
Usar modelos de embedding mais poderosos: Experimente diferentes modelos de embedding pré-treinados ou treinados sob medida para encontrar aquele que melhor captura as relações semânticas em seus dados.
Implementar roteamento de consultas e integração com ferramentas de terceiros: Se o modelo de embedding não for o problema, você pode melhorar o sistema RAG aplicando um agente para roteamento de consultas e integrando-o a ferramentas ou fontes de dados adicionais.
Ao focar nos fundamentos e iterar continuamente sobre a fonte de dados e os componentes algorítmicos, você pode garantir que sua aplicação RAG pronta para produção entregue respostas de alta qualidade aos seus usuários.
Conclusão
A integração perfeita do Milvus com vários frameworks como Spark facilita a criação e a implantação de uma aplicação escalável impulsionada por LLM. A capacidade do Spark de distribuir tarefas de processamento de dados entre vários computadores em lotes realmente acelera as operações de processamento de dados. Esse recurso é particularmente útil quando queremos ingerir grandes quantidades de dados em nosso banco de dados vetorial ou quando nossa aplicação está lidando com um grande número de consultas de usuários ao mesmo tempo.
Depois de ingerirmos dados em nosso banco de dados vetorial Milvus e recebermos a consulta de um usuário, podemos então realizar uma busca vetorial. Esse processo é crucial para buscar os contextos mais relevantes entre os dados dentro do nosso banco de dados, a fim de serem passados para uma LLM para gerar respostas altamente contextualizadas.
Continue lendo
How to Choose the Best Embedding Model for RAG in 2026: 10 Models Benchmarked
We benchmarked 10 embedding models on cross-modal, cross-lingual, long-document, and dimension compression tasks. See which one fits your RAG pipeline.
How Zilliz Ended Up at the Center of NVIDIA’s Unstructured Data Story at GTC 2026
If unstructured data is the context of AI, then the ceiling of AI applications will be set not just by models, but by how mature the infrastructure for unstructured data becomes.
Vector Databases vs. Graph Databases
Use a vector database for AI-powered similarity search; use a graph database for complex relationship-based queries and network analysis.