Como implementar fluxos de trabalho do Databricks e o Atlas Vector Search para melhorar a precisão da pesquisa de comércio eletrônico
Vittal Pai, Ashwin Gangadhar, Francesco Baldissera6 min read • Published Sep 18, 2024 • Updated Sep 18, 2024
Avalie esse Tutorial
No vasto domínio do comércio eletrônico, a capacidade dos clientes de pesquisar com rapidez e precisão uma ampla variedade de produtos é fundamental. Atlas Vector Search está surgindo como um ponto de virada neste espaço, oferecendo uma abordagem refinada para pesquisa que vai além da mera correspondência de palavras-chave. Vamos nos aprofundar em sua implementação usando MongoDB Atlas, Atlas Vector Search e Databricks.
- Repositório do GitHub para pesquisa aprimorada por IA e pesquisa vetorial (o código está agrupado para maior clareza)
Em um tutorial anterior, Aprenda a construir soluções de pesquisa de varejo aprimoradas por IA com MongoDB e Databricks, mostramos como a integração do MongoDB e Databricks fornece uma solução abrangente para o setor de varejo, combinando processamento de dados em tempo real, orquestração de fluxo de trabalho, aprendizado de máquina, funções de dados personalizadas e recursos avançados de pesquisa como forma de otimizar o gerenciamento do catálogo de produtos e aprimorar as interações com os clientes.
Neste tutorial, construiremos a solução Vector Atlas Search sobre a base de código do tutorial anterior. Confira o repositório do Github para obter a solução completa.
O diagrama abaixo representa o fluxo de trabalho do Databricks para indexar dados do atp (disponível para promessa), imagens, prd_desc (desconto do produto), prd_score (pontuação do produto) e coleções de preços. Essas collections também fazem parte do tutorial mencionado anteriormente, portanto, consulte novamente se precisar acessá-las.
Na MongoDB Atlas plataforma , podemos usar change streams e o MongoDB connector do Spark para o mover dados das collections para uma nova collection chamada Catálogo. A partir disso, usaremos um transformador de texto para criar o
Catalog Final Collection. Isto nos permitirá criar um corpus de dados indexados e incorporados em vetores que será utilizado posteriormente como o dicionário do Atlas Search. Chamaremos isso de collection catalog_final_myn. Isso será mostrado mais adiante depois de incorporarmos os nomes dos produtos.A coleção final do catálogo incluirá o status disponível para promessa de cada produto, suas imagens, o desconto do produto, a pontuação de relevância do produto e o preço, além do nome vetorizado ou incorporado do produto para o qual direcionaremos nosso mecanismo de busca vetorial.
Na imagem abaixo, explicamos como éofluxo de trabalho do Databricks. Ele consiste em dois trabalhos que são separados em dois notebooks, respectivamente. Examinaremos cada um dos notebooks a seguir.
A primeira etapa é ingestão de dados das collection mencionadas anteriormente usando o spark.readStream método do MongoDB Connector for Spark. O código abaixo faz parte do bloco de anotações que definiremos como um trabalho usando o Databricks Workflows. Você pode aprender mais sobre os blocos deanotações Databricks seguindo o tutorial.
1 atp = spark.readStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "atp_status_myn").\ option('spark.mongodb.change.stream.publish.full.document.only','true').\ option('spark.mongodb.aggregation.pipeline',[]).\ option("forceDeleteTempCheckpointLocation", "true").load() atp = atp.drop("_id") atp.writeStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "catalog_myn").\ option('spark.mongodb.operationType', "update").\ option('spark.mongodb.upsertDocument', True).\ option('spark.mongodb.idFieldList', "id").\ option("forceDeleteTempCheckpointLocation", "true").\ option("checkpointLocation", "/tmp/retail-atp-myn4/_checkpoint/").\ outputMode("append").\ start()
Esta parte do bloco de anotações lê as alterações de dados da collection atp_status_myn no banco de dados do Atlas Search, descarta o campo _id e depois grava (ou atualiza) os dados processados na collection catálogo_myn no mesmo banco de dados.
Observe como está a leitura da collection
atp_status_myn, que já tem a codificação hot (valores booleanos se o produto estiver disponível ou não) do tutorial anterior. Dessa forma, garantimos que incorporamos apenas os dados dos produtos que estão disponíveis em nosso estoque.Consulte o bloco de notas completo em nosso repositório do Github se quiser saber mais sobre toda a ingestão e transformações de dados realizadas durante esse estágio.
Usando uma combinação de bibliotecas Python e operações do PySpark para processar dados da collection MongoDB do Catálogo, vamos transformá-los, vetorizá-los e escrever os dados transformados de volta na collection do Catálogo Final. Além disso, construiremos nossa lógica de negócios de pesquisa de aplicativos.
Começamos usando o comando mágico %pip, que é específico para notebooks Jupyter e ambientes IPython. Os pacotes necessários são:
- pymongo: Um driver Python para MongoDB.
- tqdm: Uma biblioteca para exibir barras de progresso.
- sentence-transformers: Uma biblioteca para incorporações de última geração de frases, textos e imagens.
Primeiro, vamos usar o pip para instalar esses pacotes em nosso bloco de anotações Databricks:
1 %pip install pymongo tqdm sentence-transformers
Continuamos o bloco de anotações com o seguinte código:
1 model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
Aqui carregamos um modelo pré-treinado da biblioteca de transformadores de frases. Este modelo será usado para converter texto em embeddings ou vetores.
A próxima etapa é trazer os dados do catálogo do MongoDB Atlas e Atlas Search collection. Isto como uma continuação do mesmo caderno:
1 catalog_status = spark.readStream.format("mongodb").\ 2 option('spark.mongodb.connection.uri', MONGO_CONN).\ 3 option('spark.mongodb.database', "search").\ 4 option('spark.mongodb.collection', "catalog_myn").\ 5 option('spark.mongodb.change.stream.publish.full.document.only','true').\ 6 option('spark.mongodb.aggregation.pipeline',[]).\ 7 option("forceDeleteTempCheckpointLocation", "true").load()
Com este código, configuramos uma leitura estruturada de streaming da
catalog_myn coleção no search banco de dados do MongoDB. Os dados resultantes são armazenados no catalog_status DataFrame no Spark. A operação de leitura é configurada para buscar o documento completo do change stream do MongoDB e não aplica nenhuma agregação.O código do notebook continua com:
1 #Calculating new column called discounted price using the F decorator 2 3 catalog_status = catalog_status.withColumn("discountedPrice", F.col("price") * F.col("pred_price")) 4 5 #One hot encoding of the atp_status column 6 7 catalog_status = catalog_status.withColumn("atp", (F.col("atp").cast("boolean") & F.lit(1).cast("boolean")).cast("integer")) 8 9 #Running embeddings of the product titles with the get_vec function 10 11 catalog_status.withColumn("vec", get_vec("title")) 12 13 #Dropping _id column and creating a new final catalog collection with checkpointing 14 15 catalog_status = catalog_status.drop("_id") 16 catalog_status.writeStream.format("mongodb").\ 17 option('spark.mongodb.connection.uri', MONGO_CONN).\ 18 option('spark.mongodb.database', "search").\ 19 option('spark.mongodb.collection', "catalog_final_myn").\ 20 option('spark.mongodb.operationType', "update").\ 21 option('spark.mongodb.idFieldList', "id").\ 22 option("forceDeleteTempCheckpointLocation", "true").\ 23 option("checkpointLocation", "/tmp/retail-atp-myn5/_checkpoint/").\ 24 outputMode("append").\ 25 start()
Com esta última parte do código, calculamos uma nova coluna chamada DiscountedPrice como produto do preço previsto. Em seguida, realizamos a codificação one-hot na coluna de status atp, vetorizamos o título do produto e mesclamos tudo de volta em uma collection de catálogo final.
Agora que temos nossa coleção de catálogos com seus devidos embeddings, é hora de construirmos o Vector Search Index usando MongoDB Atlas Search.
Aqui definiremos como os dados devem ser armazenados e indexados para uma pesquisa eficiente. Para configurar o índice, você pode inserir o snippet no MongoDB Atlas navegando até a página inicial do cluster e clicando na aba “Search:
Em seguida, você pode clicar em "Create Index. " Certifique-se de selecionar "JSON Editor ":
Cole o trecho JSON abaixo no Editor JSON. Certifique-se de selecionar o banco de dados e a coleção corretos! No nosso caso, o nome da coleção é
catalog_final_myn. Consulte o código completo no repositório para ver a aparência do índice completo e como você pode uni-lo ao restante dos parâmetros para o tutorial de pesquisa aprimorada por IA.1 { 2 "mappings": { 3 "fields": { 4 "vec": [ 5 { 6 "dimensions": 384, 7 "similarity": "cosine", 8 "type": "knnVector" 9 } 10 ] 11 } 12 } 13 }
No código acima, o campo vec é do tipo knnVector, projetado para pesquisa de vetores. Ele indica que cada vetor tem 384 dimensões e usa a similaridade de cosseno para determinar a proximidade do vetor. Isso é fundamental para a pesquisa semântica, em que o objetivo é encontrar resultados que sejam contextualmente ou semanticamente relacionados.
Ao implementar esses parâmetros de indexação, aceleramos os tempos de recuperação. Especialmente com dados vetoriais de alta dimensão, pois os vetores brutos podem consumir uma quantidade significativa de armazenamento e reduzir o custo computacional de operações como cálculos de similaridade.
Em vez de comparar um vetor de consulta com cada vetor no conjunto de dados, a indexação permite que o sistema compare com um subconjunto, economizando recursos computacionais.
Navegue até nosso website de comércio eletrônicoLeAFYY, no qual realizaremos uma pesquisa pelas palavras-chave
tan bags. Você obterá estes resultados:
Como você pode ver, primeiro você obterá resultados que correspondem às palavras-chave tokenizadas específicas "tan" e "bags". Como resultado, isso fornecerá qualquer produto que contenha uma ou ambas as palavras-chave nos documentos de coleção do catálogo de produtos.
Porém, nem todos os resultados são bolsas ou de cor castanha. Você pode ver sapatos, carteiras, um vestido e uma calça. Isso pode ser frustrante para o cliente, levando-o a sair do site.
Agora, ative o Atlas Search clicando na caixa de seleção à esquerda do ícone de lupa na barra Atlas Search e execute novamente a query "tan bags ". Os resultados obtidos estão na imagem abaixo:
Como você pode ver na captura de tela, os resultados se tornaram mais relevantes para um consumidor. Nosso mecanismo de pesquisa é capaz de identificar produtos semelhantes entendendo o contexto de que “beige” é uma cor semelhante a “tan” e, portanto, mostrar esses produtos como alternativas.
Ao trabalhar com MongoDB Atlas e Databricks, podemos criar pipelines de transformação de dados em tempo real. Conseguimos isso aproveitando o MongoDB Connector para Spark para preparar nossos dados operacionais para vetorização e armazená-los de volta em nossas coleções do MongoDB Atlas. Essa abordagem nos permite desenvolver a lógica de pesquisa para nosso aplicativo de comércio eletrônico com o mínimo de sobrecarga operacional.
Além disso, o Atlas Vector Search fornece uma solução robusta para implementar recursos de pesquisa avançada, facilitando a entrega de uma ótima experiência de usuário de pesquisa para seus clientes. Ao compreender e integrar essas ferramentas, os desenvolvedores podem criar experiências de pesquisa que sejam rápidas, relevantes e fáceis de usar.
Não deixe de revisar o código completo em nosso repositório do Github. Entre em contato conosco para obter uma compreensão mais profunda de como criar soluções de pesquisa avançada para sua empresa de comércio eletrônico.