RAG avancé sur Milvus : recherche hybride, HyDE, réécriture des requêtes et réclassement.
Construire progressivement un pipeline RAG pour Milvus : vecteurs denses et BM25, recherche hybride RRF, filtres de métadonnées, réécriture des requêtes, HyDE, décomposition et réclassement par cross-encoder.
Sources: Beyond Semantic Search: The Complete Guide to Advanced RAG with Milvus | Sagar Patil
Un système RAG de type proof-of-concept ne prend que quelques minutes : on divise un PDF en fragments, on les passe à un modèle d’incorporation, puis on transmet les résultats correspondants à un LLM. C’est là que cette configuration basée uniquement sur la recherche vectorielle rencontre des difficultés. Les utilisateurs réels ne formulent pas leurs requêtes comme le font les moteurs de données bien organisés. Ils utilisent des formulations vagues, omettent des mots-clés et regroupent plusieurs questions en une seule ; une simple recherche sémantique rate donc une grande partie de ces informations. Lorsque la récupération échoue, l’LLM doit résumer un contexte irrelevant, ce qui entraîne des hallucinations subtiles ou des réponses inutiles.
Compléter l’écart entre une recherche sémantique de base et un pipeline de récupération en plusieurs étapes nécessite plusieurs couches. La solution présentée ci-dessous utilise Milvus. Elle commence par des recherches vectorielles denses, ajoute une recherche hybride, puis combine quatre techniques (filtrage des métadonnées, réécriture de la requête, HyDE et décomposition de la requête) avant de terminer par un réclassement à l’aide d’un cross-encoder. Chaque étape est accompagnée de code fonctionnel.
Pourquoi RAG existe
Des modèles comme GPT-4 apprennent à partir de données Internet générales jusqu’à une date butoir de connaissance. Les documents privés, les wikis internes et les politiques d’entreprise restent inaccessibles. Si l’on demande à un modèle brut quelle est la politique de congés de l’entreprise, il répond soit qu’il ne le sait pas, soit qu’il invente une réponse plausible mais fausse.
RAG (Retrieval-Augmented Generation) divise cette tâche en deux étapes :
- Récupérer les passages les plus pertinents de vos documents à l’aide d’une recherche sémantique.
- Générer une réponse en fournissant ces passages au LLM en tant que contexte.
Le flux global se présente comme suit :
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 qualité de la réponse finale dépend presque entièrement de la qualité de la récupération des données, donc réussir cette récupération est essentiel.
Le rôle des bases de données vectorielles
Les bases de données conventionnelles comparent des valeurs exactes, de sorte qu’une requête WHERE text LIKE '%leave%' ignore les documents contenant des termes tels que « droit aux vacances » ou « absence médicale », même s’ils ont le même sens. En revanche, une base de données vectorielle stocke chaque document sous forme de vecteur numérique à haute dimension (un embedding) et effectue des recherches en fonction de la similarité sémantique, c’est-à-dire de la proximité entre le sens de la requête et celui de chaque fragment de texte. Milvus convient parfaitement à ce rôle pour les raisons suivantes :
- C’est un logiciel open source qui peut être hébergé en interne, ce qui permet de préserver la confidentialité des données.
- Milvus Lite fonctionne à partir d’un seul fichier
.db, ce qui évite le besoin d’un serveur pour le développement.
Embeddings denses et sparses
Le pipeline utilise deux types de vecteurs, il est donc utile de les connaître avant d’écrire du code. Un embedding dense est un vecteur de longueur fixe, par exemple composé de 1024 nombres, généré par un réseau neuronal. Toutes les dimensions contiennent des valeurs, et l’entraînement apprend au modèle à placer les phrases ayant un sens similaire proches l’une de l’autre :
"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
Les vecteurs denses captent bien le sens et les synonymes. Des modèles tels que text-embedding-3-large d’OpenAI en génèrent. Un embedding dispersé est un vecteur très grand dont la plupart des éléments sont nuls. Chaque position non nulle représente un mot du vocabulaire, et sa valeur correspond à l’importance de ce mot, c’est-à-dire un poids de type TF-IDF :
"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}
BM25 est l’algorithme sparsifié classique et il excelle dans la correspondance exacte des mots-clés. Il permet de détecter les noms propres, les noms de produits et les termes techniques que la recherche dense peut manquer, faute de voisin sémantique proche.
La recherche hybride utilise les deux approches, en tirant parti des forces de chacune d’elles.
Mise en place et configuration
Installez d’abord les dépendances :
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
Puis les importations :
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')
Gardez tous les paramètres réglables regroupés en haut du fichier :
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
Une taille de bloc de 500 caractères avec une superposition de 100 caractères convient comme point de départ pour les documents simples. Le texte narratif se prête mieux à des blocs plus grands, de 800 à 1000 caractères, tandis que la documentation technique dense nécessite des blocs plus petits, de 200 à 300 caractères. La superposition permet de garantir que les phrases coupées par une limite de bloc puissent toujours être retrouvées.
Le processus d’indexation
L’indexation s’exécute une fois, ou à nouveau lorsque les documents changent. Elle transforme les documents bruts en vecteurs et les charge dans Milvus. Le chemin est le suivant :
PDF → Pages → Chunks → Dense Embeddings
→ Sparse Embeddings
→ Milvus Collection
Tout d’abord, initialisez les modèles et effectuez la connexion :
# 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.")
Ensuite, chargez le PDF dans un objet document par page :
# 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 divise d’abord par des sauts de paragraphe, puis par des caractères de nouvelle ligne, et enfin par des phrases, ce qui permet de conserver autant que possible les phrases liées ensemble dans un même groupe. Viennent ensuite les embeddings pour les deux types de vecteurs :
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.")
L’appel bm25_ef.fit(texts) est essentiel. BM25 apprend le vocabulaire ainsi que les fréquences des termes à partir de votre propre corpus, et c’est ce qui permet à cette méthode de fonctionner efficacement. Un modèle BM25 générique en remplacement diminue considérablement la qualité de la correspondance des mots-clés. Voici maintenant la collection, avec son schéma et ses index :
# 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.")
Deux choix d’index sont importants ici :
FLATpour les vecteurs denses. Il effectue une recherche par voisin le plus proche exacte, très précise mais plus lente à grande échelle. Il convient bien pour le développement et pour des ensembles de données comptant environ moins de 100K d’éléments ;HNSWdevrait le remplacer en environnement de production.SPARSE_INVERTED_INDEXpour les vecteurs sparses. C’est le seul type d’index valide pour les données sparses dans Milvus, et il fonctionne comme un index inversé classique, de la même manière que les moteurs de recherche indexent les documents par mots.
Finalement, préparez les enregistrements et insérez-les :
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)}")
Le outil sparse_to_dict est nécessaire car BM25 génère une matrice sparse de scipy, tandis que Milvus exige un dictionnaire Python classique de type {int: float}. Il gère les lignes sparses 2D, les tableaux 1D ainsi que les dictionnaires déjà convertis. Un piège courant : il faut toujours appeler client.load_collection() avant de rechercher. Les index sont stockés sur disque dans Milvus et chargés en RAM sur demande ; sans cette appel, la recherche échoue ou renvoie un résultat vide.
Recherche dense en tant que référence
Lorsque les données sont indexées, la recherche la plus simple consiste à intégrer la requête et à trouver les vecteurs de documents les plus proches par similarité cosinus :
# ════════════════════════════════════════════════════════════
# 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")
Sous la métrique COSINE, les scores vont de -1 à 1. Une valeur supérieure à 0,7 indique généralement une correspondance sémantique forte, et les meilleurs résultats se situent typiquement entre 0,6 et 0,95. La recherche dense suffit pour les questions en langue naturelle dont les réponses se trouvent dans un espace sémantique proche, comme demander des explications sur les avantages du congé de paternité ou ce que prévoit le code de conduite. Elle rencontre des difficultés lorsque la requête contient des codes spécifiques, des acronymes ou des noms propres que le modèle d’embedding a rarement vus.
Recherche hybride avec RRF
La recherche hybride exécute en même temps une recherche dense et une recherche BM25 sparse, puis fusionne les deux listes de classement grâce à la Reciprocal Rank Fusion (RRF). L’idée est simple : le score RRF d’un document additionne les valeurs 1 / (k + rang) provenant de chaque liste, où k=60 sert de constante d’aplanissement, de sorte que les documents bien classés dans les deux listes finissent en tête.
# ════════════════════════════════════════════════════════════
# 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")
À propos des scores : la valeur hit['distance'] provenant de hybrid_search représente le score de classement RRF fusionné et non une similarité cosinus brute. Plus cette valeur est élevée, mieux c’est, bien que son intervalle ne soit pas limité à 0 et 1, ce qui signifie qu’elle ne doit pas être comparée aux scores de similarité cosinus obtenus via une recherche ordinaire search(). Pour les systèmes RAG en production, la recherche hybride est la valeur par défaut recommandée. BM25 ne provoque quasiment aucune latence et améliore de manière significative le taux de rappel pour les requêtes riches en mots-clés.
Même la récupération hybride peut échouer dans certaines situations, et quatre techniques permettent de gérer les modes d’échec les plus courants.
Filtrage des métadonnées
Le problème. Lorsqu’une collection couvre plusieurs documents, périodes ou départements, la recherche vectorielle peut retourner des extraits semblant pertinents mais provenant de la source incorrecte, telle qu’une version obsolète d’une politique, le manuel d’un autre département ou des pages hors du champ de la question.
Fonctionnement. Milvus applique une expression de filtre scalaire avant la recherche vectorielle, de sorte que seules les lignes correspondantes atteignent l’étape d’ANN. Il ne s’agit pas d’un filtre post-traitement. Le filtre réduit d’abord le nombre de candidats, ce qui permet de préserver le taux de rappel au sein du sous-ensemble filtré.
# ════════════════════════════════════════════════════════════
# 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]}")
Les expressions de filtrage suivent cette syntaxe :
'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
Tout système RAG multi-documentaire doit inclure un filtrage des métadonnées. Lorsque les utilisateurs travaillent dans un cadre précis, comme uniquement le manuel de 2024 ou uniquement les fichiers financiers, le filtre doit être déduit automatiquement de ce cadre.
Réécriture des requêtes
Le problème. Les utilisateurs écrivent de manière informelle, par exemple "combien de jours de congé ai-je ?", tandis que les documents utilisent un langage formel tel que "droit au congé annuel" ou "politique d’accumulation des jours de vacances". La recherche dense comble en partie cette lacune, et une bonne réécriture améliore considérablement la qualité des résultats obtenus ainsi que le taux de rappel global.
Fonctionnement. Un LLM rédige plusieurs versions formelles de la requête originale, riches en mots-clés ; chaque version est soumise à une recherche distincte, et le RRF fusionne et élimine les doublons des résultats obtenus.
# ════════════════════════════════════════════════════════════
# 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 exemple de transformation :
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"
Chaque variante élargit la surface de recherche. Un document qui n’est pas parmi les 5 meilleurs pour la requête initiale peut se classer deuxième avec la troisième variante, et RRF met en évidence les documents qui apparaissent de manière constante à travers différentes requêtes.
HyDE (Hypothetical Document Embeddings)
Le problème. Une question courte et sa réponse se trouvent en des endroits différents dans l’espace d’incorporation. « Combien de jours de maladie ? » est une question et s’incorpore près d’autres questions. Une disposition stipulant que les employés ont droit à 7 jours de congé maladie payé par an s’incorpore près d’autres dispositions similaires. Ce fossé réduit la précision.
Fonctionnement. Au lieu d’incorporer la question directement, le LLM rédige un document de réponse hypothétique qui est ensuite intégré. Cette réponse inventée apparaît à proximité des documents de réponses réels, même si ses faits spécifiques sont erronés. Seule son embedding est utilisée et son texte est éliminé, de sorte que les détails fabriqués n’atteignent jamais l’utilisateur.
# ════════════════════════════════════════════════════════════
# 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 est particulièrement utile pour les requêtes courtes et vagues, comme le processus d’évaluation des salaires, en cas de lacunes lexicales entre les questions et les documents, ou encore avec un langage très informel. Il est cependant superflu pour des requêtes précises riches en mots-clés où la recherche hybride fonctionne déjà bien, ainsi que lorsque la latence des LLM est un problème, car HyDE ajoute une appel supplémentaire à l’LLM par requête.
Décomposition de la requête
Le problème. Une question complexe et composée de plusieurs parties ne peut pas être répondue à partir d’un seul passage. Poser des questions sur la politique de congés et sur l’impact de l’évaluation de performance sur le salaire concerne deux sujets situés dans deux sections distinctes. Une recherche vectorielle unique se concentre soit sur un sujet et en néglige un autre, soit elle fournit des résultats médiocres pour les deux.
Fonctionnement. Un LLM divise la question en sous-questions atomiques, chacune étant recherchée séparément ; RRF fusionne ensuite les résultats, et le dernier appel à l’LLM rassemble tous les fragments pertinents.
# ════════════════════════════════════════════════════════════
# 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 exemple de décomposition :
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?"
Chaque sous-question récupère séparément les extraits dont elle a besoin, et l’ensemble fusionné couvre toutes les parties de la question originale, ce que aucune recherche individuelle ne pourrait accomplir.
Réclassement avec des cross-encoders
Jusqu’à présent, tout a été récupéré à l’aide de bi-encoders : la requête et les documents sont encodés de manière indépendante, et la similarité est déterminée par un produit scalaire ou une distance cosinus. Ceux-ci sont rapides, mais ce n’est pas la méthode la plus précise pour évaluer la pertinence. Les cross-encoders, quant à eux, analysent ensemble une requête et un document en tant que paire pour évaluer leur adéquation, en tenant compte de leur interaction complète, ce qui les rend bien plus précis. Ils sont cependant moins rapides, trop lents pour des milliers de documents mais efficaces avec un petit ensemble de 20 à 50 candidats. D’où ce schéma : récupérer un grand nombre de résultats avec un bi-encoder, puis réclasser les résultats en utilisant un cross-encoder.
# ============================================================
# 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")
Les scores des cross-encoders sont des logits non normalisés, et non des probabilités. Une valeur plus élevée indique une pertinence plus forte ; ainsi, 8,0 est nettement supérieur à 2,0, mais l’échelle absolue varie d’un modèle à l’autre. Le modèle utilisé, ms-marco-MiniLM-L12-v2, est une solution puissante et légère, entraînée sur l’ensemble de données MS MARCO Passage Ranking, et il gère bien les requêtes sous forme de questions concernant des extraits de documents.
Comparaison des techniques
Une configuration par défaut adaptée à la production est :
- Recherche hybride en tant que méthode de récupération de base toujours active.
- Filtrage des métadonnées chaque fois que le champ d’action de l’utilisateur peut être déterminé.
- Reranking par cross-encoder sur les 20 meilleurs candidats avant qu’ils n’atteignent le LLM.
Le processus se déroule comme suit :
- Recherche dense est le point de départ : rapide, indépendante des mots-clés et efficace avec les synonymes.
- Recherche hybride constitue l’amélioration suivante, car l’ajout de vecteurs épars BM25 améliore de manière fiable le taux de rappel avec peu plus de complexité.
- Filtrage, réécriture, HyDE et décomposition ciblent chacun un mode de défaillance spécifique. Ne les appliquez pas tous aveuglément. Choisissez en fonction des documents et des modèles de requêtes réellement générés par les utilisateurs.
- Réclassement via un encodeur croisé vaut presque toujours la peine d’être ajouté. Il s’exécute localement, ajoute peu de latence sur un petit ensemble de candidats et améliore de manière fiable le contexte final.
Finalement, c’est la qualité de la récupération qui détermine la qualité de la réponse. Un LLM ne peut être que aussi bon que le contexte qui lui est fourni, et chacune des techniques mentionnées vise à rendre ce contexte plus pertinent, complet et fiable. Tout s’exécute localement sur Milvus Lite, l’API d’OpenAI étant la seule dépendance externe pour les embeddings et la génération.