Búsqueda híbrida RAG con BM25 y fusión de rangos en Python

Una persona pregunta cómo resolver ERR-104, pero el recuperador devuelve una guía general en lugar del documento que nombra ese error. Otra describe el mismo problema sin mencionar su código; una búsqueda literal no encuentra la página correcta. Son fallos diferentes. La búsqueda híbrida combina candidatos léxicos y semánticos para inspeccionar y medir ambas señales.

Este artículo implementa un baseline pequeño de BM25 y reciprocal rank fusion (RRF) en Python. Complementa RAG con LangChain, que explica el pipeline, y la evaluación offline de RAG, que mide disponibilidad de evidencia. Aquí la decisión nueva es cómo combinar rankings de candidatos antes de entregar contexto al generador.

Decide qué debe aportar cada recuperador

BM25 utiliza frecuencia de términos, frecuencia en documentos y longitud del documento. Sirve para identificadores explícitos, nombres de productos y palabras que deben aparecer. La recuperación densa compara embeddings y puede encontrar paráfrasis sin las mismas palabras. Ninguna señal demuestra por sí sola que un pasaje responda la pregunta.

Por ejemplo, conserva la diferencia entre ERR-104 y ERR-105, y entre las versiones v2.1 y v2.0 de una API. La tokenización es parte del contrato de recuperación. Separar todos los identificadores por puntuación puede eliminar precisamente la distinción que necesitas preservar.

La configuración BM25 de Elasticsearch describe k1 como saturación de frecuencia de términos y b como normalización por longitud. Los valores usados abajo, 1.2 y 0.75, son un punto de partida reproducible, no parámetros ajustados a tu corpus.

Ejecuta un baseline léxico que puedas inspeccionar

El ejemplo necesita Python 3.11 y ninguna dependencia externa, servidor, descarga de modelos o clave de API. Guarda los siguientes bloques, en orden, como hybrid_search.py. Esta implementación recorre un corpus diminuto; usa un motor con índice para colecciones de producción.

import math
import re
import unicodedata
from collections import Counter, defaultdict


def tokens(text: str) -> list[str]:
    normalized = unicodedata.normalize("NFKC", text).casefold()
    return re.findall(r"\w+(?:[-.]\w+)*", normalized)


def bm25(query: str, documents: dict[str, str], limit: int = 3) -> list[str]:
    if type(limit) is not int or limit <= 0:
        raise ValueError("limit must be a positive integer")
    if not documents:
        return []
    counts = {key: Counter(tokens(text)) for key, text in documents.items()}
    lengths = {key: sum(count.values()) for key, count in counts.items()}
    average_length = sum(lengths.values()) / len(lengths)
    if average_length == 0:
        return []
    terms = set(tokens(query))
    document_frequency = {
        term: sum(term in count for count in counts.values()) for term in terms
    }
    scores = {}
    k1, b = 1.2, 0.75
    for key, count in counts.items():
        score = 0.0
        for term in terms:
            frequency = count[term]
            if frequency == 0:
                continue
            df = document_frequency[term]
            idf = math.log(1 + (len(documents) - df + 0.5) / (df + 0.5))
            denominator = frequency + k1 * (
                1 - b + b * lengths[key] / average_length
            )
            score += idf * frequency * (k1 + 1) / denominator
        if score > 0:
            scores[key] = score
    return sorted(scores, key=lambda key: (-scores[key], key))[:limit]

El patrón conserva versiones con puntos y códigos con guiones. La normalización Unicode y el cambio de mayúsculas/minúsculas resuelven diferencias sencillas de formato. No implementa stemming, sinónimos, analizadores específicos por idioma ni tratamiento especial de nombres como C++. Prueba consultas representativas antes de adoptar un tokenizador. contraseña y password todavía requieren un modelo semántico multilingüe o una estrategia explícita de traducción/sinónimos.

La fórmula es una variante educativa de BM25. Sigue la ponderación explicada en el artículo de Elastic sobre similitud, pero no reemplaza exactamente los analizadores y normas del índice de Lucene. Se omiten resultados con puntuación léxica cero: un identificador desconocido no debe producir una coincidencia léxica inventada.

Fusiona rangos, no puntuaciones de escalas diferentes

Una puntuación BM25 y una similitud de embeddings tienen escalas diferentes. Sumarlas con un peso arbitrario puede hacer que un recuperador domine por su rango numérico. RRF suma una contribución basada en la posición dentro de cada lista:

score(documento) = sum(1 / (smoothing + rank))

En este ejemplo los rangos empiezan en uno, las listas tienen igual peso y smoothing=60. La referencia RRF de Elasticsearch documenta la fusión basada en rangos. Otros motores pueden usar constantes o convenciones distintas; la referencia de consultas híbridas de Qdrant documenta actualmente otro valor por defecto. Configúralos explícitamente al comparar implementaciones.

def rrf(*rankings: list[str], limit: int = 3,
        smoothing: int = 60) -> list[str]:
    if type(limit) is not int or limit <= 0:
        raise ValueError("limit must be a positive integer")
    if type(smoothing) is not int or smoothing < 0:
        raise ValueError("smoothing must be a non-negative integer")
    scores: dict[str, float] = defaultdict(float)
    for ranking in rankings:
        if len(ranking) != len(set(ranking)):
            raise ValueError("A retriever returned duplicate chunk IDs")
        for rank, chunk_id in enumerate(ranking, start=1):
            scores[chunk_id] += 1 / (smoothing + rank)
    return sorted(scores, key=lambda key: (-scores[key], key))[:limit]


if __name__ == "__main__":
    documents = {
        "general": "Troubleshooting guide for failed account login.",
        "reset": "Reset an expired password using account recovery.",
        "err-104": "ERR-104 means the login token expired. Request a new token.",
        "err-105": "ERR-105 means the account is locked. Contact support.",
    }
    lexical = bm25("ERR-104", documents)
    # Fixed fixture: replace with ranked chunk IDs from your dense retriever.
    semantic = ["general", "reset", "err-104"]
    print("lexical:", lexical)
    print("fused:", rrf(lexical, semantic))

Ejecuta python hybrid_search.py. Imprime:

lexical: ['err-104']
fused: ['err-104', 'general', 'reset']

El ranking semántico es un fixture suministrado deliberadamente, no la salida de un modelo de embeddings. Permite reproducir la combinación y observar el fallo: el documento relevante estaba presente, pero en tercera posición. La fusión lo promueve porque aparece en ambas listas. Es una demostración de aritmética de rangos, no evidencia medida de que la búsqueda híbrida mejore tu sistema RAG.

Los IDs repetidos dentro de una lista generan un error para evitar que inflen una puntuación. El mismo ID en dos listas diferentes es esperado y recibe ambas contribuciones. Los empates se resuelven por ID para repetir experimentos; la política de desempate de producción también debe ser explícita.

La profundidad de candidatos va antes del tamaño final del contexto

Separa tres presupuestos: candidatos por recuperador, candidatos entregados al reranker y pasajes entregados al LLM. Recuperar 30 candidatos de cada señal y devolver cinco pasajes fusionados es diferente de recuperar cinco por señal. Esas cifras son ejemplos, no recomendaciones para cualquier índice.

Si el documento relevante falta en ambas listas, RRF no puede recuperarlo. Un reranker tampoco puede recuperar un documento que nunca recibe. Aumentar solamente la ventana de contexto final no corrige ese fallo.

Usa un ID de chunk estable en ambos índices y versiona el mapeo cuando cambie la segmentación. Aplica el mismo alcance autorizado de documentos antes de recuperar candidatos en cada rama. Filtrar solamente la lista fusionada puede exponer datos no autorizados en pasos intermedios y llenar el presupuesto con candidatos inaccesibles. RRF y las puntuaciones de relevancia no proporcionan autorización.

Evalúa el cambio sin mover la referencia

Crea casos revisados para códigos exactos, terminología habitual, paráfrasis, consultas en español/inglés y preguntas que el corpus no puede responder. Guarda IDs de chunks relevantes y el comportamiento de abstención esperado. Mantén un conjunto de desarrollo para ajustar y otro fijo para comparar.

Compara rankings léxicos, semánticos y fusionados con el mismo k final. Reporta también recall de candidatos antes de la fusión: distingue fallos de recuperación de fallos de ordenamiento. Registra latencia, resultados léxicos vacíos, cantidad de candidatos, documentos repetidos y presupuesto de contexto. Un recall agregado puede esconder una regresión en identificadores poco frecuentes.

El ejemplo ejecutable se comprobó con identificadores exactos, consultas sin coincidencias, documentos vacíos, IDs duplicados, salida de fusión estable y un candidato ausente. No incluye un corpus de producción, una ejecución de embeddings ni un benchmark de precisión. Usa el arnés de evaluación offline para añadir etiquetas reales revisadas antes de elegir una configuración.

Si los candidatos híbridos resuelven el fallo observado, añade un reranker solo cuando los casos etiquetados muestren que el orden todavía limita la respuesta. Si la evidencia correcta ya llega al generador, trabaja en selección de contexto y procedencia. El siguiente cambio útil es el que aborda el fallo medido, no el pipeline de recuperación más largo.