En este artículo, aprenderá cómo implementar una estrategia de búsqueda híbrida para sistemas RAG combinando la búsqueda léxica BM25 con la búsqueda semántica, fusionadas mediante Reciprocal Rank Fusion.
Los temas que cubriremos incluyen:
Por qué la búsqueda híbrida supera a la búsqueda léxica o semántica por sí sola en sistemas de generación de recuperación aumentada. Cómo implementar la búsqueda léxica BM25 y la búsqueda semántica de vectores densos como motores de recuperación independientes en Python. Cómo fusionar ambas clasificaciones utilizando Reciprocal Rank Fusion (RRF) para producir un resultado de recuperación final y equilibrado.
Vayamos directo a ello.
Implementación de búsqueda híbrida semántica-léxica en RAG
Introducción
La implementación de estrategias de búsqueda híbrida es un paso fundamental en la construcción de sistemas RAG (generación aumentada de recuperación) modernos, especialmente cuando se pasa de soluciones prototipo a soluciones listas para producción.
Hay pocos argumentos en contra de que la búsqueda semántica, impulsada por vectores densos o incrustaciones, que son representaciones numéricas de texto, sea increíblemente útil para comprender la semántica, los sinónimos y el contexto. Sin embargo, la búsqueda léxica basada en palabras clave con enfoques como BM25 cubre un pequeño punto ciego descuidado por la búsqueda semántica. Por lo tanto, combinar lo mejor de ambos mundos es la receta perfecta para llevar el mecanismo de recuperación de su sistema RAG un paso más allá.
Exploremos cómo implementar una estrategia de búsqueda híbrida a través de un ejemplo de codificación sencillo, que lo guiará en cada paso del proceso.
Nota: Si no está familiarizado con los sistemas RAG, es posible que la serie de artículos “Comprensión de RAG” le resulte muy útil para aprovechar al máximo esta lectura. En particular, recomiendo adquirir conocimientos sobre las bases de datos vectoriales primero a través de este artículo.
Implementación paso a paso
El primer paso es asegurarse de que estén instaladas todas las bibliotecas externas de Python necesarias, en particular estas tres:
!pip install rank_bm25 sentence-transformers requests
! pip install rank_bm25 sentence – transformers requests
rank_bm25: una implementación del algoritmo de búsqueda léxica BM25 para la recuperación de información (BM significa "Best Matching"). Transformadores de oraciones: proporciona modelos de lenguaje previamente entrenados para generar incrustaciones de texto. En un entorno real, es posible que ya tenga su propia base de datos de vectores que contenga muchos documentos incrustados y no la necesite, pero la usaremos aquí para simular la construcción de una base de datos de vectores de juguetes e ilustrar la búsqueda híbrida en ella. solicitudes: se utiliza para recuperar el paquete del conjunto de datos sin procesar de un repositorio público de conjuntos de datos de GitHub preparado para este ejemplo.
Con estos ingredientes a mano, comenzamos cargando el conjunto de datos y almacenando los textos sin formato en una lista (lo hacemos porque es un conjunto de datos pequeño).
import requests
import zipfile
import io
import os
# Downloading and extracting the dataset from the compressed file
url = “https://github.com/gakudo-ai/open-datasets/raw/refs/heads/main/asia_documents.zip”
response = requests.get(url)
with zipfile.ZipFile(io.BytesIO(response.content)) as z:
z.extractall(“asia_data”)
# Loading documents and getting their filenames
documents =[]nombres_doc =[]para archivo en os.listdir("asia_data"): if file.endswith(".txt"): with open(f"asia_data/{file}", "r", encoding="utf-8") as f: documentos.append(f.read()) doc_names.append(file) print(f"Documentos cargados {len(documents)} para la base de conocimientos.")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
solicitudes de importación
importar archivo zip
importar io
importar sistema operativo
# Descargar y extraer el conjunto de datos del archivo comprimido
URL = "https://github.com/gakudo-ai/open-datasets/raw/refs/heads/main/asia_documents.zip"
respuesta = solicitudes . obtener ( url )
con archivo zip . ZipFile ( io . BytesIO ( respuesta . contenido ) ) como z :
z . extraer todo ( "asia_data" )
# Cargando documentos y obteniendo sus nombres de archivo
documentos = [ ]
nombres_doc = [ ]
para archivar en os . listdir ( "asia_data" ) :
si archivo . termina con ( ".txt" ) :
con open ( f "asia_data/{file}" , "r" , codificación = "utf-8" ) como f :
documentos . agregar ( f . leer ( ) )
nombres_doc . adjuntar ( archivo )
print ( f "Documentos cargados {len(documentos)} para la base de conocimientos." )
El proceso de búsqueda híbrida se divide en tres etapas: dos de ellas se desarrollan en paralelo o de forma independiente entre sí. El tercero es donde ocurre la fusión de ambos enfoques, utilizando un método de fusión llamado Fusión de Rango Recíproco (RRF).
Primero cubramos la búsqueda léxica con BM25:
from rank_bm25 import BM25Okapi
# BM25 requires that each text is tokenized as a (sub)list of words
tokenized_corpus = [doc.lower().split() for doc in documents]
bm25 = BM25Okapi(tokenized_corpus)
def search_bm25(query, top_k=3):
tokenized_query = query.lower().split()
# Getting scores (lexical relevance to the query) for all documents
scores = bm25.get_scores(tokenized_query)
# Ranking documents by score
ranked_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
return ranked_indices[:top_k], scores
from rank_bm25 import BM25Okapi
# BM25 requires that each text is tokenized as a (sub)list of words
tokenized_corpus = [ doc . lower ( ) . split ( ) for doc in documents ]
bm25 = BM25Okapi ( tokenized_corpus )
def search_bm25 ( query , top_k = 3 ) :
tokenized_query = query . lower ( ) . split ( )
# Getting scores (lexical relevance to the query) for all documents
scores = bm25 . get_scores ( tokenized_query )
# Ranking documents by score
ranked_indices = sorted ( range ( len ( scores ) ) , key = lambda i : scores [ i ] , reverse = True )
return ranked_indices [ : top_k ] , scores
El proceso de búsqueda léxica se ha encapsulado en una función llamada search_bm25(). Esta función toma dos argumentos de entrada: una cadena que contiene la consulta del usuario al sistema RAG y el número de resultados principales a recuperar. La biblioteca rank_bm25 proporciona un método get_scores() que calcula, para cada documento (tratado como una colección de tokens), una puntuación de relevancia léxica. Luego clasificamos los documentos según su puntuación decreciente, seleccionamos el top-k y los devolvemos.
Mientras tanto, el motor de búsqueda semántica primero utiliza un modelo de transformador de oraciones para obtener vectores de incrustación para los textos y la consulta del usuario, luego aplica una métrica de similitud de vectores como la similitud de coseno para clasificar los textos por relevancia semántica y recuperar el k más relevante:
from sentence_transformers import SentenceTransformer, util
import torch
# Loading the pre-trained embedding model
model = SentenceTransformer(‘all-MiniLM-L6-v2’)
# Pre-compute embeddings for our corpus (our “Vector DB”)
# You do not need this step if you already have an external vector database:
# you may read and import your document vectors instead
doc_embeddings = model.encode(documents, convert_to_tensor=True)
def search_semantic(query, top_k=3):
# Embedding the user’s query into a vector
query_embedding = model.encode(query, convert_to_tensor=True)
# Calculating cosine similarity between the query and all documents
cosine_scores = util.cos_sim(query_embedding, doc_embeddings)[0]# Clasificación de documentos por similitud ranking_indices = torch.argsort(cosine_scores, descendente=True).tolist() return ranking_indices[:top_k], cosine_scores.tolist()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
de sentencia_transformers importar SentenceTransformer , útil
importar antorcha
# Cargando el modelo de incrustación previamente entrenado
modelo = SentenceTransformer ( 'todo-MiniLM-L6-v2' )
# Precalcular incrustaciones para nuestro corpus (nuestra "Vector DB")
# No necesitas este paso si ya tienes una base de datos vectorial externa:
# puedes leer e importar los vectores de tus documentos en su lugar
doc_embeddings = modelo . codificar ( documentos , convert_to_tensor = Verdadero )
def search_semantic ( consulta , top_k = 3 ) :
# Incrustar la consulta del usuario en un vector
consulta_incrustación = modelo . codificar ( consulta , convert_to_tensor = Verdadero )
# Calcular la similitud del coseno entre la consulta y todos los documentos
puntuaciones_coseno = útil . cos_sim ( query_embedding , doc_embeddings ) [ 0 ]
# Clasificación de documentos por similitud
índices_clasificados = antorcha . argsort ( coseno_puntuaciones , descendente = Verdadero ) . tolista ( )
devolver índices_clasificados [ : top_k ] , puntuaciones_coseno . tolista ( )
Es hora de ponerlo todo junto. Las dos puntuaciones calculadas para cada documento no pueden simplemente sumarse, porque operan en escalas numéricas muy diferentes. En lugar de ello, realizamos la fusión en función de rangos en lugar de puntuaciones brutas de similitud o relevancia. Para ello, RRF es el estándar de oro de la industria para fusionar información de clasificación: calcula una clasificación general para cada documento recompensando a aquellos que aparecen en posiciones altas en ambas listas. La lógica subyacente es algo similar a la del operador de media armónica en estadística.
El proceso general de búsqueda híbrida se implementa de la siguiente manera:
def hybrid_search(query, top_k=3):
# 1. Obtaining the two standalone search rankings
bm25_ranks, _ = search_bm25(query, top_k=len(documents))
semantic_ranks, _ = search_semantic(query, top_k=len(documents))
# 2. Applying RRF formula: RRF_score = 1 / (k + rank)
rrf_scores = {i: 0.0 for i in range(len(documents))}
k_constant = 60 # The value of 60 is a standard academic convention
# Adding RRF scores from BM25
for rank, doc_idx in enumerate(bm25_ranks):
rrf_scores[doc_idx] += 1.0 / (k_constant + rank + 1)
# Adding RRF scores from semantic search
for rank, doc_idx in enumerate(semantic_ranks):
rrf_scores[doc_idx] += 1.0 / (k_constant + rank + 1)
# 3. Sorting documents by their final fused RRF score
final_ranked_indices = sorted(rrf_scores.keys(), key=lambda idx: rrf_scores[idx], reverse=True)
return final_ranked_indices[:top_k], rrf_scores
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def hybrid_search ( query , top_k = 3 ) :
# 1. Obtaining the two standalone search rankings
bm25_ranks , _ = search_bm25 ( query , top_k = len ( documents ) )
semantic_ranks , _ = search_semantic ( query , top_k = len ( documents ) )
# 2. Applying RRF formula: RRF_score = 1 / (k + rank)
rrf_scores = { i : 0.0 for i in range ( len ( documents ) ) }
k_constant = 60 # The value of 60 is a standard academic convention
# Adding RRF scores from BM25
for rank , doc_idx in enumerate ( bm25_ranks ) :
rrf_scores [ doc_idx ] += 1.0 / ( k_constant + rank + 1 )
# Adding RRF scores from semantic search
for rank , doc_idx in enumerate ( semantic_ranks ) :
rrf_scores [ doc_idx ] += 1.0 / ( k_constant + rank + 1 )
# 3. Sorting documents by their final fused RRF score
índices_clasificados_finales = ordenado ( rrf_scores . claves ( ) , clave = lambda idx : rrf_scores [ idx ] , reverso = Verdadero )
devolver índices_clasificados_finales [ : top_k ] , puntuaciones_rrf
Ahora es el momento de probarlo todo. Formulemos una consulta de usuario y veamos qué resultados obtenemos.
query = “Which nation is best known for rice fields and paddies?”
print(f”— Query: ‘{query}’ —“)
# Testing Semantic (good at understanding aspects like “nation-wise nuances” and conceptual titles)
print(“nTop Semantic Results:”)
sem_indices, _ = search_semantic(query)
for idx in sem_indices:
print(f”- {doc_names[idx]}”)
# Testing BM25 (good at finding exact keyword-based matches like “rice”, “field”, “paddy”)
print(“nTop BM25 Results:”)
bm25_indices, _ = search_bm25(query)
for idx in bm25_indices:
print(f”- {doc_names[idx]}”)
# Testing Hybrid (balances both)
print(“nTop Hybrid (RRF) Results:”)
hybrid_indices, _ = hybrid_search(query)
for idx in hybrid_indices:
print(f”- {doc_names[idx]}”)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
query = “Which nation is best known for rice fields and paddies?”
print ( f “— Query: ‘{query}’ —“ )
# Testing Semantic (good at understanding aspects like “nation-wise nuances” and conceptual titles)
print ( “nTop Semantic Results:” )
sem_indices , _ = search_semantic ( query )
for idx in sem_indices :
print ( f “- {doc_names[idx]}” )
# Testing BM25 (good at finding exact keyword-based matches like “rice”, “field”, “paddy”)
print ( “nTop BM25 Results:” )
bm25_indices , _ = search_bm25 ( query )
for idx in bm25_indices :
print ( f “- {doc_names[idx]}” )
# Testing Hybrid (balances both)
print ( “nTop Hybrid (RRF) Results:” )
hybrid_indices , _ = hybrid_search ( query )
for idx in hybrid_indices :
print ( f “- {doc_names[idx]}” )
Los resultados no son excelentes en comparación con un sistema RAG de producción, pero tenga en cuenta que lo probamos en un pequeño conjunto de datos de nueve documentos. En ese contexto, el resultado es bastante razonable.
— Query: ‘Which nation is best known for rice fields and paddies?’ —
Top Semantic Results:
– Vietnam.txt
– South_Korea.txt
– Thailand.txt
Top BM25 Results:
– Indonesia.txt
– Japan.txt
– Philippines.txt
Top Hybrid (RRF) Results:
– Vietnam.txt
– Thailand.txt
– Indonesia.txt
— – Query : ‘Which nation is best known for rice fields and paddies?’ — –
Top Semantic Results :
– Vietnam . txt
– South_Korea . txt
– Thailand . txt
Top BM25 Results :
– Indonesia . txt
– Japan . txt
– Philippines . txt
Top Hybrid ( RRF ) Results :
– Vietnam . txt
– Thailand . txt
– Indonesia . txt
Prueba a modificar la consulta y sustituirla por otras relacionadas con templos, playas, montañas o cualquier otra cosa que se te ocurra al pensar en destinos orientales. ¿Puede encontrar un escenario en el que tanto los resultados semánticos como los resultados del BM25 sean muy consistentes entre sí?
Concluyendo
Este artículo lo guió en la implementación de un mecanismo de búsqueda híbrido para la etapa de recuperación de los sistemas RAG. Elegir no depender únicamente de la búsqueda semántica es una consideración importante al escalar las soluciones RAG a entornos de producción.