Notes pratiques : Au-delà de la recherche sémantique : Le guide complet sur RAG avancé
Guide pratique pas à pas : Au-delà de la recherche sémantique – Le guide complet sur RAG avancé : contrats, vérifications et emplacements de code intégrables pour les équipes qui mettent en œuvre ce modèle.
Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : « Beyond Semantic Search: The Complete Guide to Advanced RAG with Milvus » | l’auteur. L’accent est mis sur des étapes opérationnelles, des vérifications explicites et du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner l’intention. Pour l’étape d’aperçu, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le parcours passe d’un environnement de démonstration à des environnements partagés.
Qu’est-ce que RAG et pourquoi existe-t-il ?
Lorsque vous travaillez sur la section « Qu’est-ce que RAG ? », notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans devoir lire l’ensemble du système. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
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
Le rôle d’une base de données vectorielle
Lorsque vous travaillez sur le chapitre « Le rôle d’une étape », notez d’abord le contrat : les entrées requises, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
Comprendre les embeddings : denses et spars
Lorsque vous travaillez sur la phase « Understanding Embeddings Dense », notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Changer fréquemment les prompts ne résout généralement pas un système de récupération insuffisant. Lorsque vous travaillez sur la phase « Understanding Embeddings Dense », notez d’abord les exigences : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les surprises financières lorsque le projet passe de l’environnement de démonstration à des environnements partagés.
Embeddings denses
La phase des Embeddings denses fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
"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
Embeddings esparses (BM25)
La phase de BM25 pour les embeddings spars fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours réussi et celui de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Séparez la politique de segmentation des données de la politique de récupération ; modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}
Pourquoi vous avez besoin des deux
La phase « Pourquoi vous avez besoin des deux » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent. La phase « Pourquoi vous avez besoin des deux » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de la démonstration aux environnements partagés.
Mise en place du projet et dépendances
Pendant l’étape de configuration du projet et des dépendances, définissez les entrées, le responsable de l’étape ainsi que les critères d’achèvement avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du système. Citez les passages qui servent de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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
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')
Configuration
Pendant l’étape de configuration, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’une mise en forme ultérieure. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
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
Construction du pipeline d’indexation
Pour l’étape de création du pipeline d’indexation, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un pipeline embrouillé. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour l’étape de création du pipeline d’indexation, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite les factures inattendues lorsque le processus passe de la démonstration à sha
environnements rouges.PDF → Pages → Chunks → Dense Embeddings
→ Sparse Embeddings
→ Milvus Collection
Étape 1 et 2 : Initialiser les modèles et établir la connexion
Lors de l’exécution des étapes 1 et 2 d’initialisation, notez d’abord le contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de maintenir l’honnêteté des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Mémorisez les instructions stables du système et les schémas des outils. Envoyer à nouveau un préambule identique est une cause fréquente de gaspillage.
# 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.")
Étape 3 et 4 : Charger et diviser le document en blocs
Lors de l’étape 3 et 4 relative au chargement, notez d’abord les exigences du contrat : les entrées requises, le signal de succès, ainsi que ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez en même temps le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
# 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.")
Étape 5 : Générer les deux types d’embeddings
Lors de l’étape 5 « Générer les deux versions », notez d’abord les spécifications : entrées requises, signal de succès et comportement en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une responsabilité précise plutôt qu’un processus embrouillé. Évaluez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Un changement fréquent de prompts ne résout généralement pas un système de récupération insuffisant.
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.")
Lors de l’étape 5 « Générer les deux versions », notez d’abord les spécifications : entrées requises, signal de succès et comportement en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les surprises financières lorsque le système passe de l’environnement de démonstration à des environnements partagés.
Étape 6 : Créer la collection avec schéma et index
L’étape 6 de création fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
# 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.")
Étapes 7 et 8 : Préparer les enregistrements et les insérer
La phase de préparation des étapes 7 et 8 fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours optimal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’une mise en forme ultérieure. Séparez la politique de segmentation de la politique de récupération : modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
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)}")
RAG de base : Recherche vectorielle dense
La phase Basic RAG Dense Vector fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’erreur doit pointer vers une seule responsabilité et non vers un processus embrouillé. Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les indicateurs de qualité changent. La phase Basic RAG Dense Vector fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe de la démonstration aux environnements partagés.
# ════════════════════════════════════════════════════════════
# 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")
Better RAG : Recherche hybride (Dense + Sparse)
Pour l’étape de recherche hybride RAG améliorée, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
# ════════════════════════════════════════════════════════════
# 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")
RAG avancé — Quatre techniques de récupération
Pour l’étape avancée de récupération RAG Four, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations ultérieures. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Filtrage des métadonnées
Pour l’étape de filtrage des métadonnées, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Citez les passages qui ont réellement servi de base à la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour l’étape de filtrage des métadonnées, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût évite les factures inattendues lorsque le processus passe de l’environnement de démonstration aux environnements partagés.
# ════════════════════════════════════════════════════════════
# 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]}")
'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
Réécriture des requêtes
Lors de la phase de réécriture des requêtes, notez d’abord les exigences : entrées nécessaires, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de maintenir l’honnêteté des modifications ultérieures du code. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les administrateurs peuvent auditer sans avoir à lire l’ensemble du système. Évaluez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
# ════════════════════════════════════════════════════════════
# 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]}")
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"
HyDE — Encodages hypothétiques de documents
Lors de la phase d’incorporation des documents hypothétiques HyDE, notez d’abord les exigences : entrées requises, signal de succès et comportement en cas d’échec partiel. Cette liste de contrôle permet de rester honnête lors des modifications ultérieures du code. Documentez à la fois le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Le simple changement de prompts ne résout que rarement un système de récupération insuffisant.
# ════════════════════════════════════════════════════════════
# 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]}")
Décomposition de la requête
Lors de la phase de décomposition des requêtes, notez d’abord le contrat : les entrées nécessaires, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Un changement fréquent de prompts ne résout que rarement un système de récupération insuffisant. Lors de la phase de décomposition des requêtes, notez d’abord le contrat : les entrées nécessaires, le signal de succès et ce qui se passe en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le système passe de l’environnement de démonstration à des environnements partagés.
# ════════════════════════════════════════════════════════════
# 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]}")
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?"
Reranking avec Cross-Encoder
La phase de reranking utilisant le Cross-Encoder fonctionne au mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez les paramètres de configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
# ============================================================
# 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")
Comparaison des techniques
La phase « Comment comparer les techniques » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript parfait, un cas d’échec et la note de rollback avant d’élargir le périmètre. Documentez en même temps le parcours optimal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Séparez la politique de segmentation de la politique de récupération : modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
Conclusion
La phase de Conclusion fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit pointer vers une seule responsabilité plutôt que vers un processus embrouillé. Séparez la politique de segmentation de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent. La phase de Conclusion fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Enregistrez les temps d’exécution ainsi que le coût en tokens ou requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés.
Liste de contrôle opérationnelle
La phase de liste de contrôle opérationnel fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un transcript idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre.
Considérez cette phase comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définez des vérifications de succès et refusez toute mise en œuvre partielle silencieuse.
Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité évoluent.
Ajoutez un test de base qui exerce le chemin critique dans l’environnement CI à l’aide de fixtures, et non d’API payantes en ligne, chaque fois que le budget le permet.
Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés.
Séparez la politique de segmentation des données de la politique de récupération. Modifier l’une ne doit pas obliger à réécrire l’autre lorsque les métriques de qualité changent.
Au préalable de promouvoir l’ensemble du système, figez les versions, conservez une transcription exemplaire pour le parcours critique, et définites clairement les étapes de réversion. Les environnements partagés nécessitent des limites de fréquence, des vérifications d’attribution et un responsable désigné pour la rotation des secrets. Préférez une fiabilité simple à des démonstrations originales mais peu fiables.
Note de batch pour c9664ffe2213 : gardez les clés du fournisseur hors du répertoire de code, fixez un plafond pour les tokens par session, et stockez les transcriptions à côté des fichiers d’évaluation afin que les remplacements ultérieurs de modèles restent comparables.
Note de déploiement 1 (c9664ffe2213) : fixez les images, définissez des budgets de requêtes, et vérifiez l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 2 (c9664ffe2213) : fixez les images, définissez des budgets de requêtes, et vérifiez l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 3 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 4 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 5 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 6 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 7 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 8 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des utilisateurs sur un environnement de test avant un déploiement plus large.
Note de déploiement 9 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des locataires sur un environnement canari avant un déploiement plus large.
Note de déploiement 10 (c9664ffe2213) : fixer les images, définir les budgets de requête et vérifier l’isolation des locataires sur un environnement canari avant un déploiement plus large.