RAG avanzado en Milvus: búsqueda híbrida, HyDE, reescritura de consultas y reclasificación.

Un sistema RAG de prueba de concepto se crea en minutos: se divide un PDF en fragmentos, estos se pasan por un modelo de embedding y las coincidencias se envían a un LLM. Es en la fase de producción donde este esquema basado únicamente en búsquedas vectoriales se detiene. Los usuarios reales no formulan consultas como lo harían los motores de datos organizados. Escriben frases vagas, omiten palabras clave y combinan varias preguntas en una sola, y una única búsqueda semántica pierde gran parte de esa información. Cuando la recuperación falla, el LLM debe resumir contexto irrelevante, lo que genera alucinaciones sutiles o respuestas que no son de ayuda.

Cerrar la brecha entre la búsqueda semántica básica y un pipeline de recuperación multietapa requiere varios niveles. La implementación presentada a continuación utiliza Milvus. Comienza con búsquedas vectoriales densas, agrega búsqueda híbrida, luego aplica cuatro técnicas (filtrado de metadatos, reescritura de consultas, HyDE y descomposición de consultas) y finaliza con un reclasificación mediante cross-encoder. Cada paso incluye código funcional.

Por qué existe RAG

Modelos como GPT-4 aprenden a partir de datos generales de Internet hasta un punto determinado en el tiempo. Los documentos privados, las wikis internas y las políticas de la empresa quedan fuera de su alcance. Si le preguntas a un modelo en bruto cuál es la política de vacaciones de la empresa, o bien dice que no lo sabe o inventa una respuesta plausible pero incorrecta.

RAG (Retrieval-Augmented Generation) divide el proceso en dos pasos:

1. Recuperar los fragmentos más relevantes de tus documentos mediante búsqueda semántica.
2. Generar una respuesta al pasar esos fragmentos al LLM como contexto.

El flujo general es el siguiente:

User Question
     │
     ▼
[Embed the question]  →  query vector
     │
     ▼
[Search Vector DB]  →  top-K relevant document chunks
     │
     ▼
[LLM prompt: "Given these passages, answer: {question}"]
     │
     ▼
  Accurate, Grounded Answer

La calidad de la respuesta final depende casi en su totalidad de la calidad de la recuperación, por lo que acertar en esta fase es lo más importante.

Donde entra en juego una base de datos vectorial

Las bases de datos convencionales comparan valores exactos, por lo que una consulta WHERE text LIKE '%leave%' ignora documentos que mencionan “vacation entitlement” o “medical absence”, a pesar de tener el mismo significado. Una base de datos vectorial, en cambio, almacena cada documento como un vector numérico de alta dimensión (un embedding) y realiza búsquedas por similitud semántica, es decir, por cuán cercano es el significado de la consulta al del contenido buscado. Milvus es adecuado para esta función por las siguientes razones:

• Es de código abierto y se puede alojar de forma autónoma, lo que mantiene los datos privados.
• Milvus Lite funciona con un único archivo .db, por lo que no se necesita servidor para su desarrollo.

La búsqueda híbrida es nativa, lo que permite que los vectores densos (semánticos) y dispersos (por palabras clave) se combinen en una sola consulta.

Se puede escalar desde una computadora portátil hasta miles de millones de vectores sin necesidad de modificar el código de la aplicación.

Embebidos densos y dispersos

El proceso utiliza dos tipos de vectores, por lo que es útil conocerlos antes de escribir código. Un embebido denso es un vector de longitud fija, por ejemplo, con 1024 números, generado por una red neuronal. Todas las dimensiones contienen valores, y el entrenamiento enseña al modelo a colocar cerca aquellas oraciones con significados similares:

"sick leave policy"           →  [0.12, -0.87, 0.34, 0.56, ...]  (1024 numbers)
"medical absence entitlement" →  [0.13, -0.85, 0.31, 0.54, ...]  ← very close
"quarterly revenue target"    →  [0.91,  0.23, -0.67, 0.02, ...] ← far away

Los vectores densos capturan bien el significado y los sinónimos. Modelos como text-embedding-3-large de OpenAI los generan. Un embedding disperso es un vector enorme en el que la mayoría de las entradas son cero. Cada posición distinta de cero representa una palabra del vocabulario, y su valor corresponde a la importancia de esa palabra, un peso al estilo TF-IDF:

"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}

BM25 es el algoritmo disperso clásico y destaca en la coincidencia exacta de palabras clave. Detecta nombres propios, nombres de productos y términos técnicos que la búsqueda densa puede pasar por alto debido a la falta de un vecino semántico cercano.

La búsqueda híbrida utiliza ambos enfoques, aprovechando las ventajas de cada uno.

Configuración e instalación

Primero, instale las dependencias:

pip install --upgrade pymilvus
pip install "pymilvus[model]"
pip install sentence-transformers
pip install langchain-text-splitters
pip install langchain-openai
pip install langchain-community
pip install scipy
pip install nltk

Luego, las importaciones:

import uuid
from tqdm import tqdm
from pymilvus import (
    MilvusClient, DataType,
    AnnSearchRequest, RRFRanker
)
from pymilvus.model.sparse import BM25EmbeddingFunction
from pymilvus.model.sparse.bm25.tokenizers import build_default_analyzer
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
import scipy.sparse as sp
import re, json
import nltk
nltk.download('stopwords')

Mantenga todos los parámetros ajustables juntos en la parte superior del archivo:

PDF_PATH        = "./data/sample_employee_handbook.pdf" # path of you document
COLLECTION_NAME = "rag_documents_hybrid"
MILVUS_DB_PATH  = "./db/milvus_demo.db"
API_KEY         = "sk-..."
EMBEDDING_MODEL = "text-embedding-3-large"
EMBEDDING_DIM   = 1024
CHUNK_SIZE      = 500
CHUNK_OVERLAP   = 100
TOP_K           = 5

En cuanto al tamaño de los bloques: 500 caracteres con una superposición de 100 caracteres sirve como punto de partida para documentos simples. El texto narrativo funciona mejor con bloques más grandes de 800 a 1000 caracteres, mientras que la documentación técnica densa se beneficia de bloques más pequeños de 200 a 300 caracteres. La superposición asegura que las oraciones cortadas por los límites de un bloque puedan seguir encontrándose.

El proceso de indexación

La indexación se ejecuta una vez, o nuevamente cuando los documentos cambian. Convierte los documentos en bruto en vectores y los carga en Milvus. La ruta es:

PDF  →  Pages  →  Chunks  →  Dense Embeddings
                           →  Sparse Embeddings
                           →  Milvus Collection

Primero, inicialice los modelos y conéctelos:

# Dense embedding model here we'll be using OpenAI's embedding model
embedding_obj = OpenAIEmbeddings(
    model=EMBEDDING_MODEL,
    api_key=API_KEY,
    dimensions=EMBEDDING_DIM
)
# Milvus Lite - single file, no server needed
client = MilvusClient(MILVUS_DB_PATH)
print("Models and DB connection ready.")

Luego cargue el PDF en un objeto de documento por página:

# Load PDF — one Document object per page
loader = PyPDFLoader(PDF_PATH)
documents = loader.load()
print(f"Loaded {len(documents)} pages.")

# Split into overlapping chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP,
    separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks.")

RecursiveCharacterTextSplitter divide primero en saltos de párrafo, luego en saltos de línea y después en oraciones, lo que mantiene juntas las oraciones relacionadas en un mismo grupo tanto como sea posible. A continuación vienen los embeddings para ambos tipos de vectores:

texts = [doc.page_content for doc in chunks]

# Dense embeddings - one API call for the entire corpus
print("Generating dense embeddings...")
dense_embeddings = embedding_obj.embed_documents(texts)
print(f"Dense dimension: {len(dense_embeddings[0])}")

# Sparse embeddings - BM25 must be fit on YOUR corpus first
print("Fitting BM25 on corpus...")
analyzer = build_default_analyzer(language="en") # for this you will require nltk-stopwords
bm25_ef = BM25EmbeddingFunction(analyzer)
bm25_ef.fit(texts)  # Builds vocabulary from your documents
sparse_embeddings = bm25_ef.encode_documents(texts)
print("Sparse embeddings generated.")

La llamada bm25_ef.fit(texts) es crucial. BM25 aprende el vocabulario y las frecuencias de los términos a partir de su propio corpus, y por eso funciona. Un modelo BM25 genérico en su lugar reduce notablemente la calidad de la coincidencia con palabras clave. Ahora, la colección, con su esquema e índices:

# Drop and recreate for a clean state
if COLLECTION_NAME in client.list_collections():
    client.drop_collection(COLLECTION_NAME)

# Define schema
schema = client.create_schema()
schema.add_field("id",            DataType.VARCHAR,           is_primary=True, max_length=100)
schema.add_field("vector",        DataType.FLOAT_VECTOR,      dim=EMBEDDING_DIM)
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
schema.add_field("text",          DataType.VARCHAR,           max_length=65535)
schema.add_field("page_number",   DataType.INT64)
schema.add_field("source",        DataType.VARCHAR,           max_length=500)
schema.add_field("chunk_id",      DataType.INT64)

# Create the collection
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)

# Build indexes separately
index_params = client.prepare_index_params()

index_params.add_index(
    field_name="vector",
    index_type="FLAT",          # Exact search - swap to HNSW for production
    metric_type="COSINE"
)

index_params.add_index(
    field_name="sparse_vector",
    index_type="SPARSE_INVERTED_INDEX",
    metric_type="IP"            # Inner Product is the only valid metric for sparse
)

client.create_index(collection_name=COLLECTION_NAME, index_params=index_params)

print("Collection and indexes created.")

Aquí son importantes dos opciones de índice:

• FLAT para vectores densos. Realiza una búsqueda exacta de vecinos más cercanos, totalmente precisa pero más lenta a gran escala. Es adecuada para el desarrollo y para conjuntos de datos con menos de aproximadamente 100K elementos, mientras que HNSW debería reemplazarla en entornos de producción.
• SPARSE_INVERTED_INDEX para vectores dispersos. Es el único tipo de índice válido para datos dispersos en Milvus y funciona como un índice invertido clásico, de la misma manera en que los motores de búsqueda indexan documentos por palabras.

Finalmente, prepare los registros e insértelos:

def sparse_to_dict(s_emb) -> dict:
    """Convert a scipy sparse row into Milvus-compatible {index: value} dict."""
    if sp.issparse(s_emb):
        coo = s_emb.tocoo()
        return {int(col): float(val) for col, val in zip(coo.col, coo.data)}
    elif isinstance(s_emb, dict):
        return s_emb
    else:
        return {int(i): float(v) for i, v in enumerate(s_emb) if v != 0.0}

# Build the records list
data = []
for idx, (chunk, d_emb) in enumerate(tqdm(zip(chunks, dense_embeddings), total=len(chunks))):
    sparse_dict = sparse_to_dict(sparse_embeddings[idx])
    if not sparse_dict:
        print(f"Warning: empty sparse vector at chunk {idx}, skipping.")
        continue
    data.append({
        "id":            str(uuid.uuid4()),
        "vector":        d_emb,
        "sparse_vector": sparse_dict,
        "text":          chunk.page_content,
        "page_number":   int(chunk.metadata.get("page", -1)),
        "source":        PDF_PATH,
        "chunk_id":      idx
    })

# Insert into Milvus
res = client.insert(collection_name=COLLECTION_NAME, data=data)
print(f"Inserted {res['insert_count']} records.")

# Load into memory - required before any search operation
client.load_collection(COLLECTION_NAME)
print(f"Load state: {client.get_load_state(COLLECTION_NAME)}")

Se necesita la función auxiliar sparse_to_dict porque BM25 genera una matriz dispersa de scipy, mientras que Milvus requiere un diccionario normal de Python {int: float}. Esta función maneja filas dispersas 2D, arrays 1D y diccionarios que ya están convertidos. Un problema común: siempre se debe llamar a client.load_collection() antes de realizar búsquedas. Los índices se almacenan en el disco en Milvus y se cargan en la RAM según sea necesario; por lo tanto, sin esa llamada, la búsqueda fallará o devolverá resultados vacíos.

Búsqueda densa como referencia

Con los datos indexados, la búsqueda más sencilla consiste en incluir la consulta y encontrar los vectores del documento más cercano mediante la similitud coseno:

# ════════════════════════════════════════════════════════════
# Dense Vector Search
# ════════════════════════════════════════════════════════════

query = "What is the leave policy?"

# Step 1: Embed the query using the same model used at index time
query_dense_embedding = embedding_obj.embed_query(query)

# Step 2: Search
results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_dense_embedding],
    anns_field="vector",
    search_param={"metric_type": "COSINE"},
    limit=TOP_K,
    output_fields=["text", "page_number", "source"]
)

# Step 3: Display results
for idx, hit in enumerate(results[0], start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | Cosine Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
    print(f"  {entity['text'][:300]}\n")

Bajo la métrica COSINE, las puntuaciones van de -1 a 1. Un valor superior a 0.7 suele indicar una coincidencia semántica fuerte, y los resultados principales suelen situarse entre 0.6 y 0.95. La búsqueda densa es suficiente para preguntas en lenguaje natural cuyas respuestas se encuentran en un espacio semántico cercano, como pedir una explicación sobre los beneficios de la licencia parental o qué establece el código de conducta. Esta técnica tiene dificultades cuando la consulta contiene códigos específicos, acrónimos o nombres propios que el modelo de embedding rara vez ha visto.

Búsqueda híbrida con RRF

La búsqueda híbrida ejecuta al mismo tiempo una búsqueda densa y una búsqueda BM25 dispersa, y luego fusiona ambas listas ordenadas mediante Reciprocal Rank Fusion (RRF). La idea es sencilla: la puntuación RRF de un documento suma 1 / (k + rank) de cada lista, donde k=60 funciona como constante de suavizado, de modo que los documentos bien clasificados en ambas listas terminan en la parte superior.

# ════════════════════════════════════════════════════════════
# Hybrid Search (Dense + Sparse)
# ════════════════════════════════════════════════════════════

query = "leave policy?"

# Dense query vector
query_dense = embedding_obj.embed_query(query)

# Sparse query vector - uses the same BM25 model fitted on the corpus
sparse_raw  = bm25_ef.encode_queries([query])
sparse_dict = sparse_to_dict(sparse_raw[0])
print(f"Sparse query terms: {len(sparse_dict)}")  # Should be > 0

# Build two separate ANN search requests
dense_req = AnnSearchRequest(
    data=[query_dense],
    anns_field="vector",
    param={"metric_type": "COSINE"},
    limit=TOP_K
)
sparse_req = AnnSearchRequest(
    data=[sparse_dict],
    anns_field="sparse_vector",
    param={"metric_type": "IP"},
    limit=TOP_K
)

# Execute hybrid search with RRF fusion
results = client.hybrid_search(
    collection_name=COLLECTION_NAME,
    reqs=[dense_req, sparse_req],
    ranker=RRFRanker(k=60),
    limit=TOP_K,
    output_fields=["text", "page_number", "source"]
)
for idx, hit in enumerate(results[0], start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | RRF Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
    print(f"  {entity['text'][:300]}\n")

Sobre las puntuaciones: el valor hit['distance'] de hybrid_search corresponde a la puntuación de clasificación RRF fusionada y no a la similitud coseno en bruto. Cuanto mayor sea, mejor, aunque su rango no está limitado entre 0 y 1, lo que significa que no debe compararse con las puntuaciones de coseno obtenidas mediante una search() regular. Para sistemas RAG en producción, la búsqueda híbrida es el valor predeterminado recomendado. BM25 añade apenas algo de latencia y mejora notablemente la recuperación en consultas con muchas palabras clave.

Incluso la recuperación híbrida puede fallar en situaciones específicas, y cuatro técnicas abordan los modos de fallo más comunes.

Filtrado de metadatos

El problema. Cuando una colección abarca varios documentos, períodos o departamentos, la búsqueda vectorial puede devolver fragmentos que parecen relevantes pero provienen de la fuente incorrecta, como una versión obsoleta de una política, el manual de otro departamento o páginas fuera del alcance de la pregunta.

Cómo funciona. Milvus aplica una expresión de filtro escalar antes de la búsqueda vectorial, de modo que solo las filas que coinciden llegan a la etapa del ANN. No se trata de un filtro posterior. El filtro reduce primero el conjunto de candidatos, lo que mantiene la tasa de recuperación dentro del subconjunto filtrado.

# ════════════════════════════════════════════════════════════
# METADATA FILTERING
# ════════════════════════════════════════════════════════════

def search_with_metadata_filter(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    page_range: tuple = None,
    source_file: str = None,
    top_k: int = 5
):
    filter_parts = []

    if page_range:
        lo, hi = page_range
        filter_parts.append(f"page_number >= {lo} && page_number <= {hi}")

    if source_file:
        filter_parts.append(f'source == "{source_file}"')

    filter_expr = " && ".join(filter_parts) if filter_parts else None

    print(f"\n[Metadata Filter] Query : '{query}'")
    print(f"[Metadata Filter] Filter: {filter_expr or 'None (unfiltered)'}")

    results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=query,
        top_k=top_k,
        filters=filter_expr
    )
    return results


# ── Run ──────────────────────────────────────────────────────
meta_results = search_with_metadata_filter(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query      = "What is the leave policy?",
    page_range = (1, 30),
    source_file= None,
    top_k      = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nMETADATA-FILTERED RESULTS")
print("=" * 55)
if not meta_results or not meta_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(meta_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Las expresiones de filtrado siguen esta sintaxis:

'page_number >= 1 && page_number <= 30'      # page range
'source == "hr_policy.pdf"'                  # exact source
'category in ["leave", "performance"]'       # in a list
'source like "hr%"'                          # prefix match

Cualquier sistema RAG de múltiples documentos debe incluir filtrado de metadatos. Si los usuarios trabajan dentro de un ámbito específico, como únicamente el manual de 2024 o solo los archivos financieros, se debe derivar automáticamente el filtro a partir de ese ámbito.

Reescritura de consultas

El problema. Los usuarios escriben de forma informal, como “¿cuántos días libres tengo?”, mientras que los documentos utilizan un lenguaje formal como “derecho a vacaciones anuales” o “política de acumulación de días de vacaciones”. La búsqueda densa ayuda a cerrar parte de esa brecha, y una buena reescritura mejora considerablemente la recuperación de resultados escasos y el recuerdo general.

Cómo funciona. Un LLM elabora varias reescrituras formales y ricas en palabras clave de la consulta original; cada variante se procesa mediante una búsqueda independiente, y RRF fusiona y elimina las duplicadas entre los resultados obtenidos.

# ════════════════════════════════════════════════════════════
# QUERY REWRITING
# ════════════════════════════════════════════════════════════

import re, json

REWRITE_PROMPT = """You are an expert at reformulating search queries to improve document retrieval.

Given a user query, produce {n} alternative search queries that:
- Use formal, document-style language
- Include relevant keywords and synonyms
- Cover different angles of the same question

User query: {query}

Respond ONLY with a JSON array of strings. Example:
["rewritten query 1", "rewritten query 2", "rewritten query 3"]"""


def rewrite_query(query: str, n: int = 3) -> list[str]:
    prompt   = REWRITE_PROMPT.format(query=query, n=n)
    response = llm.invoke(prompt)
    raw      = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
    try:
        variants = json.loads(raw)
        return [query] + variants       # always keep the original
    except json.JSONDecodeError:
        print("Warning: Could not parse rewrites, using original query only.")
        return [query]


def search_with_query_rewriting(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    n_rewrites: int = 3,
    top_k: int = 5
):
    variants = rewrite_query(query, n=n_rewrites)

    print(f"\n[Query Rewriting] Original  : '{query}'")
    for i, v in enumerate(variants[1:], 1):
        print(f"[Query Rewriting] Variant {i} : '{v}'")

    seen_ids    = {}
    rank_scores = {}

    for variant in variants:
        results = hybrid_search(
            client, collection_name, embedding_obj, bm25_ef,
            query_text=variant,
            top_k=top_k
        )
        if not results or not results[0]:
            continue
        for rank, hit in enumerate(results[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
    return [merged]


# ── Run ──────────────────────────────────────────────────────
rewrite_results = search_with_query_rewriting(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query      = "What is the leave policy?",
    n_rewrites = 3,
    top_k      = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nQUERY-REWRITTEN RESULTS")
print("=" * 55)
if not rewrite_results or not rewrite_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(rewrite_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Un ejemplo de transformación:

Input:  "how many days off do I get?"

Output variants:
  1. "annual leave entitlement number of days employee handbook"
  2. "vacation days accrual policy full-time employee"
  3. "paid time off PTO allowance per calendar year"

Cada variante amplía la superficie de búsqueda. Un documento que no está entre los 5 primeros resultados para la consulta original podría ocupar el segundo lugar con la tercera variante, y RRF hace visible a los documentos que aparecen de manera consistente en distintas consultas.

HyDE (Hypothetical Document Embeddings)

El problema. Una pregunta breve y su respuesta se encuentran en lugares diferentes dentro del espacio de embeddings. “¿Cuántos días de enfermedad?” es una pregunta y se integra cerca de otras preguntas. Una disposición normativa que indica que los empleados reciben 7 días de licencia por enfermedad remunerada al año se integra cerca de otras declaraciones normativas. Esa brecha reduce la precisión.

Cómo funciona. En lugar de incrustar la pregunta, el LLM escribe un documento de respuesta hipotético, y ese es el que se incrusta. La respuesta inventada aparece cerca de los documentos de respuesta reales, incluso si sus hechos específicos son incorrectos. Solo se utiliza su incrustación y su texto se descarta, por lo que los detalles inventados nunca llegan al usuario.

# ════════════════════════════════════════════════════════════
# HyDE (Hypothetical Document Embeddings)
# ════════════════════════════════════════════════════════════

HYDE_PROMPT = """You are a corporate policy document writer.

Write a 2-3 paragraph excerpt from an official HR policy or company document
that would DIRECTLY ANSWER the following question.
Write in formal document style. Do not mention the question itself.

Question: {query}

Document excerpt:"""


def generate_hypothetical_document(query: str) -> str:
    response = llm.invoke(HYDE_PROMPT.format(query=query))
    return response.content.strip()


def search_with_hyde(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    top_k: int = 5
):
    hypothetical_doc = generate_hypothetical_document(query)

    print(f"\n[HyDE] Query            : '{query}'")
    print(f"[HyDE] Hypothetical doc :\n  {hypothetical_doc[:300]}...\n")

    # Search using the hypothetical document's embedding
    hyde_results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=hypothetical_doc,    # embed the answer, not the question
        top_k=top_k
    )

    # Also search with the original query and merge both via RRF
    original_results = hybrid_search(
        client, collection_name, embedding_obj, bm25_ef,
        query_text=query,
        top_k=top_k
    )

    seen_ids    = {}
    rank_scores = {}

    for result_set in [hyde_results, original_results]:
        if not result_set or not result_set[0]:
            continue
        for rank, hit in enumerate(result_set[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
    return [merged]


# ── Run ──────────────────────────────────────────────────────
hyde_results = search_with_hyde(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query = "What is the leave policy?",
    top_k = 5
)

# ── Print Results ─────────────────────────────────────────────
print("\nHyDE RESULTS")
print("=" * 55)
if not hyde_results or not hyde_results[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(hyde_results[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

HyDE es de mayor ayuda con consultas cortas y vagas, como los procesos de revisión salarial, las diferencias en vocabulario entre las preguntas y los documentos, y la redacción muy informal. Es un exceso para consultas precisas y ricas en palabras clave, donde la búsqueda híbrida ya funciona bien, así como cuando importa la latencia de los LLM, ya que HyDE añade una llamada adicional al LLM por consulta.

Descomposición de la consulta

El problema. Una pregunta compleja y multifacética no puede responderse con un solo texto. Preguntar sobre la política de vacaciones y cómo afecta la evaluación de desempeño al salario abarca dos temas en dos secciones distintas. Una búsqueda vectorial simple o bien se centra en un tema y pasa por alto el otro, o bien devuelve resultados mediocres para ambos.

Cómo funciona. Un LLM divide la pregunta en subpreguntas atómicas; cada una se busca por separado, RRF fusiona los resultados, y la última llamada al LLM recibe todos los fragmentos relevantes.

# ════════════════════════════════════════════════════════════
# QUERY DECOMPOSITION
# ════════════════════════════════════════════════════════════

DECOMPOSE_PROMPT = """You are an expert at breaking down complex questions for document retrieval.

Decompose the following question into 2-4 simple, self-contained sub-questions.
Each sub-question should target a single distinct piece of information.

Complex question: {query}

Respond ONLY with a JSON array of strings. Example:
["sub-question 1", "sub-question 2", "sub-question 3"]"""


def decompose_query(query: str) -> list[str]:
    response = llm.invoke(DECOMPOSE_PROMPT.format(query=query))
    raw      = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        print("Warning: Could not parse decomposition, using original query.")
        return [query]


def search_with_decomposition(
    client, collection_name, embedding_obj, bm25_ef,
    query: str,
    top_k: int = 5
):
    sub_questions = decompose_query(query)

    print(f"\n[Decomposition] Original query : '{query}'")
    for i, sq in enumerate(sub_questions, 1):
        print(f"[Decomposition] Sub-question {i} : '{sq}'")

    per_subquery_results = {}
    seen_ids             = {}
    rank_scores          = {}

    for sq in sub_questions:
        results = hybrid_search(
            client, collection_name, embedding_obj, bm25_ef,
            query_text=sq,
            top_k=top_k
        )
        per_subquery_results[sq] = results

        if not results or not results[0]:
            continue
        for rank, hit in enumerate(results[0], start=1):
            hit_id = hit["id"]
            rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
            if hit_id not in seen_ids:
                seen_ids[hit_id] = hit

    merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]

    # Per sub-question breakdown
    print("\n── Per Sub-question Results ──")
    for sq, res in per_subquery_results.items():
        print(f"\n  SUB-QUERY: '{sq[:60]}'")
        if res and res[0]:
            for i, hit in enumerate(res[0], start=1):
                print(f"    {i}. Page {hit['entity']['page_number']} | Score {hit['distance']:.4f} | {hit['entity']['text'][:150]}")

    return {"per_subquery": per_subquery_results, "merged": [merged]}


# ── Run ──────────────────────────────────────────────────────
decomp_results = search_with_decomposition(
    client, COLLECTION_NAME, embedding_obj, bm25_ef,
    query = "What is the leave policy and how does it affect salary deductions?",
    top_k = 5
)

# ── Print Merged Results ──────────────────────────────────────
print("\nDECOMPOSED — MERGED FINAL RESULTS")
print("=" * 55)
merged_hits = decomp_results["merged"]
if not merged_hits or not merged_hits[0]:
    print("No results returned.")
else:
    for idx, hit in enumerate(merged_hits[0], start=1):
        entity = hit["entity"]
        print(f"\nRank  : {idx}")
        print(f"Score : {hit['distance']:.4f}")
        print(f"Page  : {entity['page_number']}")
        print(f"Text  :\n{entity['text'][:400]}")

Un ejemplo de descomposición:

Input:  "What is the leave policy and how does performance review affect salary?"
Sub-questions:
  1. "What is the annual leave policy?"
  2. "How many sick days are employees entitled to?"
  3. "How does performance review affect salary?"
  4. "What is the performance review schedule?"

Cada subpregunta obtiene por sí sola los fragmentos de texto que necesita, y el conjunto resultante abarca todas las partes de la pregunta original, algo que ninguna búsqueda individual podría lograr.

Reclasificación con codificadores cruzados

Hasta ahora, todo se ha recuperado con bi-encoders: la consulta y los documentos se incrustan de forma independiente, y la similitud proviene del producto escalar o de la distancia coseno. Son rápidos, pero no representan la forma más precisa de evaluar la relevancia. Los codificadores cruzados, en cambio, leen una consulta y un documento juntos como un par y evalúan cuán bien se corresponden, teniendo en cuenta su interacción completa, lo que los hace mucho más precisos. Sin embargo, sacrifican velocidad, ya que son demasiado lentos para miles de documentos pero rápidos con un pequeño grupo de candidatos de 20 a 50. De ahí el patrón: recuperar una amplia gama de resultados con un bi-encoder y reclasificarlos con un codificador cruzado.

# ============================================================
# RERANKING WITH CROSS-ENCODER
# ============================================================

from sentence_transformers import CrossEncoder

# Huggingface: cross-encoder/ms-marco-MiniLM-L12-v2
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L12-v2")
query       = "What is the leave policy?"
RETRIEVAL_K = 20   # fetch more than you need
FINAL_K     = 5    # rerank down to this

# Step 1: Broad retrieval - fetch 20 candidates
query_dense = embedding_obj.embed_query(query)

results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_dense],
    anns_field="vector",
    search_param={"metric_type": "COSINE"},
    limit=RETRIEVAL_K,
    output_fields=["text", "page_number", "source"]
)

hits = results[0]
print(f"Retrieved {len(hits)} candidates for reranking.")

# Step 2: Score each (query, chunk) pair with the cross-encoder
pairs        = [[query, hit["entity"]["text"]] for hit in hits]
rerank_scores = cross_encoder.predict(pairs)

# Step 3: Sort by cross-encoder score
for hit, score in zip(hits, rerank_scores):
    hit["rerank_score"] = float(score)
reranked = sorted(hits, key=lambda x: x["rerank_score"], reverse=True)[:FINAL_K]

# Step 4: Display
for idx, hit in enumerate(reranked, start=1):
    entity = hit["entity"]
    print(f"Rank {idx} | Rerank: {hit['rerank_score']:.4f} | Vector: {hit['distance']:.4f}")
    print(f"  Page {entity['page_number']}: {entity['text'][:300]}\n")

Las puntuaciones del cross-encoder son logits no normalizados, no probabilidades. Un valor más alto indica una relevancia mayor, por lo que 8.0 supera ampliamente a 2.0, pero la escala absoluta difiere entre los modelos. El modelo utilizado, ms-marco-MiniLM-L12-v2, es una opción potente y ligera entrenada en el conjunto de datos de clasificación de pasajes MS MARCO, y maneja bien las consultas en formato de pregunta sobre pasajes de documentos.

Comparación de las técnicas

Una configuración por defecto adecuada para producción es:

1. Búsqueda híbrida como método de recuperación básico siempre activo.
2. Filtrado de metadatos cada vez que se puede determinar el alcance del usuario.
3. Reclasificación con cross-encoder de los 20 candidatos principales antes de que lleguen al LLM.

Reescritura de consultas o HyDE se añaden para dominios con una gran brecha léxica entre el idioma del usuario y el idioma del documento.

El progreso es el siguiente:

• Búsqueda densa es el punto de partida: rápida, independiente de las palabras clave y eficaz con sinónimos.
• Búsqueda híbrida representa la mejora, ya que la adición de vectores dispersos BM25 mejora de manera fiable el recuerdo con poca complejidad adicional.
• Filtrado, reescritura, HyDE y descomposición abordan cada uno un modo de fallo específico. No los aplique todos ciegamente; elija según los documentos y los patrones de consulta que generan realmente los usuarios.
• Reclasificación con codificador cruzado casi siempre vale la pena añadirla. Se ejecuta localmente, añade poca latencia en un conjunto pequeño de candidatos y mejora de manera fiable el contexto final.

Al final, la calidad de la recuperación determina la calidad de la respuesta. Un LLM solo puede ser tan bueno como el contexto que se le proporciona, y cada una de las técnicas mencionadas existe para hacer que ese contexto sea más relevante, completo y fiable. Todo se ejecuta localmente en Milvus Lite, con la API de OpenAI como única dependencia externa para los embeddings y la generación.