Notes pratiques : Bases de données vectorielles pour RAG en production : Indexation, Recherche hybride
Guide pas à pas des notes pratiques : Bases de données vectorielles pour RAG en production : Indexation, Recherche hybride : contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes utilisant ce modèle.
Les notes suivantes reconstituent une approche pratique pour aborder le sujet « Bases de données vectorielles pour RAG en production : indexation, recherche hybride et mise à l’échelle de la récupération ». L’accent est mis sur les contrats, les vérifications et les placeholders de code à insérer, plutôt que sur une présentation motivante. Lors de la phase d’aperçu, 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. 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 administrateurs peuvent auditer sans devoir lire l’ensemble du système.
La base de données vectorielle est l’algorithme de recherche
La base de données vectorielle fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Documentez en même temps le parcours optimal et celui de récupération. Les tentatives répétées, les contrôles humains et le traitement 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.
Recherche du voisin le plus proche exact : la référence qui ne peut pas être utilisée à grande échelle
La phase de recherche du voisin le plus proche exact 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. 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 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.
import faiss
import numpy as np
from typing import Tuple
def build_exact_index(
embeddings: np.ndarray,
use_cosine: bool = True
) -> faiss.IndexFlatIP:
"""
Build a FAISS flat index for exact nearest neighbour search.
embeddings: (N, D) float32 array.
use_cosine: If True, normalises a copy of the embeddings and uses inner
product (equivalent to cosine similarity). The caller's array
is not mutated.
Returns a FAISS flat index. Benchmark latency against your corpus and
latency SLO before deciding whether ANN indexing is necessary.
"""
dimension = embeddings.shape[1]
if use_cosine:
# Copy before normalising to avoid mutating the caller's array.
embeddings_copy = embeddings.astype(np.float32).copy()
faiss.normalize_L2(embeddings_copy)
index = faiss.IndexFlatIP(dimension)
index.add(embeddings_copy)
return index
else:
index = faiss.IndexFlatL2(dimension)
index.add(embeddings.astype(np.float32).copy())
return index
def search_exact(
index: faiss.IndexFlatIP,
query_vector: np.ndarray,
top_k: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Search the flat index. Returns (distances, indices).
query_vector must already be normalised if the index was built with
normalised embeddings.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
distances, indices = index.search(query, top_k)
return distances[0], indices[0]
HNSW : Pourquoi il est si courant dans la recherche vectorielle en production
La phase HNSW Why It Is 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. Traitez cette phase comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites des vérifications de succès et refusez toute complétion 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. La phase HNSW Why It Is 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. Conservez les configurations 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 graphe.
Comment le graphe est construit
Pour l’étape « Comment le graphique est généré », 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 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’une amélioration ultérieure. Citez les passages qui justifient réellement la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation.
Les paramètres qui définissent votre compromis entre taux de rappel et latence
Pour les paramètres qui définissent l’étape, il convient de spécifier les entrées, le responsable de cette étape ainsi que 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 avoir à deviner l’état caché. Préférer des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité et non un processus embrouillé. Citer 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.
import faiss
import numpy as np
from typing import Tuple
def build_hnsw_index(
embeddings: np.ndarray,
m: int = 32,
ef_construction: int = 200,
ef_search: int = 100,
use_cosine: bool = True
) -> faiss.IndexHNSWFlat:
"""
Build a FAISS HNSW index for approximate nearest neighbour search.
m: Graph connectivity parameter. Higher = better recall potential, more memory.
Starting range for banking policy corpora: 16 to 32. Benchmark your corpus.
ef_construction: Candidates explored during index build. Higher = better graph quality.
One-time cost at index build; does not affect query latency.
ef_search: Candidates explored at query time. Controls recall-latency trade-off.
Can be changed without rebuilding. Starting range: 50 to 200.
use_cosine: Normalise embeddings and use inner product (cosine similarity).
Note: FAISS HNSW does not support GPU acceleration. For GPU-accelerated ANN,
use IndexIVFPQ variants.
"""
dimension = embeddings.shape[1]
# Copy before normalising to avoid mutating the caller's array.
embeddings_to_index = embeddings.astype(np.float32).copy()
if use_cosine:
faiss.normalize_L2(embeddings_to_index)
index = faiss.IndexHNSWFlat(dimension, m, faiss.METRIC_INNER_PRODUCT)
else:
index = faiss.IndexHNSWFlat(dimension, m, faiss.METRIC_L2)
index.hnsw.efConstruction = ef_construction
index.hnsw.efSearch = ef_search
index.add(embeddings_to_index)
return index
def search_hnsw(
index: faiss.IndexHNSWFlat,
query_vector: np.ndarray,
top_k: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Search the HNSW index. Returns (scores, indices).
query_vector must be normalised if the index was built with normalised embeddings.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
scores, indices = index.search(query, top_k)
return scores[0], indices[0]
Exigences en mémoire de HNSW
Pour l’étape des exigences en matière de mémoire HNSW, 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é. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute complétion partielle silencieuse. 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 des exigences en matière de mémoire HNSW, 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.
IVF : Indexation inversée de fichiers pour des déploiements à ressources limitées
Lors de la phase d’indexation inversée de fichiers IVF, 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 à la fois 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 ultérieures. 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.
import faiss
import numpy as np
from typing import Tuple
def build_ivf_index(
embeddings: np.ndarray,
nlist: int = 1024,
nprobe: int = 64,
use_cosine: bool = True
) -> faiss.IndexIVFFlat:
"""
Build a FAISS IVF flat index.
nlist: Number of Voronoi cells. A common starting heuristic is sqrt(N),
where N is corpus size. For 100K vectors: 300-1000. For 1M: 1024-4096.
Validate empirically.
nprobe: Number of cells searched at query time. Higher = better recall, slower.
Set based on your recall benchmark results.
use_cosine: Use inner product on normalised vectors.
Requires training on a representative sample before adding vectors.
"""
dimension = embeddings.shape[1]
embeddings_to_index = embeddings.astype(np.float32).copy()
if use_cosine:
faiss.normalize_L2(embeddings_to_index)
quantiser = faiss.IndexFlatIP(dimension)
index = faiss.IndexIVFFlat(quantiser, dimension, nlist, faiss.METRIC_INNER_PRODUCT)
else:
quantiser = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFFlat(quantiser, dimension, nlist)
# Use a random representative sample for training. Using the first N records
# risks training on a non-representative slice if the corpus is ordered by
# date, jurisdiction, or document type.
n_available = len(embeddings_to_index)
desired_training_size = min(n_available, 40 * nlist)
rng = np.random.default_rng(seed=42)
training_indices = rng.choice(n_available, size=desired_training_size, replace=False)
training_sample = embeddings_to_index[training_indices]
index.train(training_sample)
index.nprobe = nprobe
index.add(embeddings_to_index)
return index
IVF avec quantification du produit
Lors du travail sur l’étape de quantification des produits dans le cadre de la FIV, 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é. Évaluez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
import faiss
import numpy as np
from typing import Tuple
def build_ivfpq_index(
embeddings: np.ndarray,
nlist: int = 1024,
m_subvectors: int = 8,
bits_per_code: int = 8,
nprobe: int = 64
) -> Tuple[faiss.IndexIVFPQ, faiss.IndexFlatIP]:
"""
Build a FAISS IVF-PQ index paired with a flat index for exact re-scoring.
m_subvectors: Number of sub-vectors. Must divide dimension evenly.
For 1536 dimensions: m=8 (192 dims each), m=16 (96 dims each).
Select based on the storage-recall trade-off for your corpus.
bits_per_code: Bits per sub-vector code. 8 bits = 256 centroids per sub-vector.
Lower bits = smaller code, larger recall degradation.
Returns (pq_index, flat_index).
Use pq_index to retrieve top-N candidates cheaply; use flat_index to re-score
those candidates with full float32 precision.
"""
dimension = embeddings.shape[1]
assert dimension % m_subvectors == 0, (
f"Dimension {dimension} must be divisible by m_subvectors {m_subvectors}"
)
norm_embeddings = embeddings.astype(np.float32).copy()
faiss.normalize_L2(norm_embeddings)
# Compressed IVF-PQ index for broad retrieval
quantiser = faiss.IndexFlatIP(dimension)
pq_index = faiss.IndexIVFPQ(
quantiser, dimension, nlist, m_subvectors, bits_per_code,
faiss.METRIC_INNER_PRODUCT
)
training_size = min(len(norm_embeddings), 50 * nlist)
rng = np.random.default_rng(seed=42)
training_indices = rng.choice(len(norm_embeddings), size=training_size, replace=False)
pq_index.train(norm_embeddings[training_indices])
pq_index.nprobe = nprobe
pq_index.add(norm_embeddings)
# Flat index for exact re-scoring of PQ candidates
flat_index = faiss.IndexFlatIP(dimension)
flat_index.add(norm_embeddings)
return pq_index, flat_index
def two_stage_search(
pq_index: faiss.IndexIVFPQ,
flat_index: faiss.IndexFlatIP,
query_vector: np.ndarray,
top_k: int = 10,
candidate_multiplier: int = 10
) -> Tuple[np.ndarray, np.ndarray]:
"""
Two-stage retrieval: broad PQ candidate recall followed by exact flat re-scoring.
Stage 1: IVF-PQ retrieves top_k * candidate_multiplier candidates cheaply.
Stage 2: The flat index re-scores those candidates with full float32 precision.
The flat index must have been built with the same normalised embeddings added
in the same corpus order so that IVF-PQ indices align to flat index positions.
candidate_multiplier: Higher values improve recall at higher latency cost.
"""
query = query_vector.reshape(1, -1).astype(np.float32)
faiss.normalize_L2(query)
n_candidates = top_k * candidate_multiplier
_, candidate_indices = pq_index.search(query, n_candidates)
valid_mask = candidate_indices[0] >= 0
valid_candidates = candidate_indices[0][valid_mask]
if len(valid_candidates) == 0:
return np.array([]), np.array([])
# Reconstruct candidate vectors from the flat index and score them exactly.
candidate_vectors = np.zeros(
(len(valid_candidates), flat_index.d), dtype=np.float32
)
for i, idx in enumerate(valid_candidates):
flat_index.reconstruct(int(idx), candidate_vectors[i])
exact_scores = (candidate_vectors @ query.T).flatten()
reranked_order = np.argsort(exact_scores)[::-1][:top_k]
final_indices = valid_candidates[reranked_order]
final_scores = exact_scores[reranked_order]
return final_scores, final_indices
Comparaison des bases de données vectorielles pour 2026
Lors de l’étape de comparaison des bases de données vectorielles, notez d’abord les conditions prévues : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Considérez cette étape 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 les terminations partielles silencieuses. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Les changements fréquents de prompts résolvent rarement un système de récupération insuffisant. Lors de l’étape de comparaison des bases de données vectorielles, notez d’abord les conditions prévues : entrées requises, signal de succès et conséquences en cas d’échec partiel. Cette liste de contrôle permet de garantir l’honnêteté des modifications ultérieures du code. Gardez 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 administrateurs peuvent auditer sans devoir lire l’ensemble du système.
FAISS
La phase FAISS fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple réussi, un cas d’échec ainsi que la note de réversion avant d’élargir le périmètre. Documentez en même temps le parcours optimal et celui de récupération. Les tentatives répétées, les contrôles humains et le traitement 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.
pgvector
La phase pgvector 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. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’erreur 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é changent.
-- Enable the pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;
-- Policy chunk table with vector and structured metadata
CREATE TABLE policy_chunks (
chunk_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
document_id TEXT NOT NULL,
document_version TEXT NOT NULL,
policy_id TEXT,
jurisdiction TEXT,
effective_date DATE,
section TEXT,
content_type TEXT NOT NULL,
chunk_text TEXT NOT NULL,
classification TEXT NOT NULL DEFAULT 'INTERNAL',
permitted_roles TEXT[] NOT NULL DEFAULT '{}',
embedding_model TEXT NOT NULL,
embedding vector(1536),
indexed_at TIMESTAMPTZ DEFAULT NOW()
);
-- HNSW index for cosine similarity retrieval
CREATE INDEX ON policy_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- Partial index for jurisdiction-scoped retrieval (common query pattern)
CREATE INDEX ON policy_chunks
USING hnsw (embedding vector_cosine_ops)
WHERE jurisdiction = 'EU';
-- Standard indexes for metadata filter columns
CREATE INDEX ON policy_chunks (policy_id);
CREATE INDEX ON policy_chunks (jurisdiction);
CREATE INDEX ON policy_chunks (classification);
CREATE INDEX ON policy_chunks (effective_date);
import psycopg2
import numpy as np
from typing import List, Dict, Optional
def search_policy_chunks(
query_embedding: List[float],
jurisdiction: Optional[str] = None,
classification_ceiling: str = "INTERNAL",
permitted_role: Optional[str] = None,
top_k: int = 10,
ef_search: int = 100,
connection_string: str = "postgresql://user:password@localhost:5432/rag_db"
) -> List[Dict]:
"""
Retrieve policy chunks from pgvector with jurisdiction and access filtering.
ef_search: Controls the HNSW recall-latency trade-off for this session.
Set per-session; does not require index rebuild.
"""
conn = psycopg2.connect(connection_string)
cur = conn.cursor()
cur.execute(f"SET hnsw.ef_search = {ef_search};")
classification_levels = {"PUBLIC": 0, "INTERNAL": 1, "CONFIDENTIAL": 2}
max_level = classification_levels.get(classification_ceiling, 1)
permitted_classifications = [
k for k, v in classification_levels.items() if v <= max_level
]
filters = ["classification = ANY(%s)"]
params: List = [permitted_classifications]
if jurisdiction:
filters.append("jurisdiction = %s")
params.append(jurisdiction)
if permitted_role:
filters.append("%s = ANY(permitted_roles) OR cardinality(permitted_roles) = 0")
params.append(permitted_role)
where_clause = " AND ".join(filters)
embedding_str = "[" + ",".join(str(x) for x in query_embedding) + "]"
query = f"""
SELECT
chunk_id,
document_id,
document_version,
policy_id,
jurisdiction,
effective_date,
section,
content_type,
chunk_text,
1 - (embedding <=> %s::vector) AS cosine_similarity
FROM policy_chunks
WHERE {where_clause}
ORDER BY embedding <=> %s::vector
LIMIT %s;
"""
params_with_embedding = [embedding_str] + params + [embedding_str, top_k]
cur.execute(query, params_with_embedding)
rows = cur.fetchall()
columns = [
"chunk_id", "document_id", "document_version", "policy_id",
"jurisdiction", "effective_date", "section", "content_type",
"chunk_text", "cosine_similarity"
]
results = [dict(zip(columns, row)) for row in rows]
cur.close()
conn.close()
return results
Qdrant
La phase Qdrant fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Traitez 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. La phase Qdrant fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Conservez les configurations 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.
from qdrant_client import QdrantClient
from qdrant_client.models import (
VectorParams, Distance, HnswConfigDiff,
PointStruct, Filter, FieldCondition, MatchValue, MatchAny,
SparseVectorParams, SparseIndexParams, SparseVector
)
from typing import List, Dict, Optional
client = QdrantClient(host="localhost", port=6333)
COLLECTION_NAME = "banking_policy"
DENSE_VECTOR_NAME = "dense"
SPARSE_VECTOR_NAME = "sparse"
def create_policy_collection(
dimension: int = 1536,
m: int = 16,
ef_construction: int = 200
) -> None:
"""
Create a Qdrant collection configured for both dense and sparse vectors.
"""
client.recreate_collection(
collection_name=COLLECTION_NAME,
vectors_config={
DENSE_VECTOR_NAME: VectorParams(
size=dimension,
distance=Distance.COSINE,
hnsw_config=HnswConfigDiff(
m=m,
ef_construct=ef_construction,
full_scan_threshold=10000
)
)
},
sparse_vectors_config={
SPARSE_VECTOR_NAME: SparseVectorParams(
index=SparseIndexParams(on_disk=False)
)
}
)
def upsert_policy_chunks(chunks: List[Dict]) -> None:
"""
Index policy chunks with dense vectors, sparse vectors, and metadata payloads.
Each chunk dict must contain:
chunk_id, dense_vector, sparse_indices, sparse_values,
chunk_text, document_id, document_version, policy_id,
jurisdiction, effective_date, content_type, classification,
permitted_roles
"""
points = [
PointStruct(
id=chunk["chunk_id"],
vector={
DENSE_VECTOR_NAME: chunk["dense_vector"],
SPARSE_VECTOR_NAME: SparseVector(
indices=chunk["sparse_indices"],
values=chunk["sparse_values"]
)
},
payload={
"chunk_text": chunk["chunk_text"],
"document_id": chunk["document_id"],
"document_version": chunk["document_version"],
"policy_id": chunk.get("policy_id"),
"jurisdiction": chunk.get("jurisdiction"),
"effective_date": chunk.get("effective_date"),
"content_type": chunk["content_type"],
"classification": chunk["classification"],
"permitted_roles": chunk.get("permitted_roles", []),
"status": "active"
}
)
for chunk in chunks
]
client.upsert(collection_name=COLLECTION_NAME, points=points)
def search_dense_filtered(
dense_query: List[float],
jurisdiction: Optional[str] = None,
permitted_classifications: List[str] = None,
top_k: int = 10,
score_threshold: float = 0.3
) -> List[Dict]:
"""
Dense vector search with integrated payload filtering.
Filtering is applied inside the HNSW graph traversal, not as a post-filter.
"""
if permitted_classifications is None:
permitted_classifications = ["PUBLIC", "INTERNAL"]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if jurisdiction:
must_conditions.append(
FieldCondition(key="jurisdiction", match=MatchValue(value=jurisdiction))
)
search_filter = Filter(must=must_conditions)
results = client.search(
collection_name=COLLECTION_NAME,
query_vector=(DENSE_VECTOR_NAME, dense_query),
query_filter=search_filter,
limit=top_k,
score_threshold=score_threshold,
with_payload=True
)
return [
{"chunk_id": hit.id, "score": hit.score, **hit.payload}
for hit in results
]
def search_hybrid_qdrant(
dense_query: List[float],
sparse_query_indices: List[int],
sparse_query_values: List[float],
jurisdiction: Optional[str] = None,
permitted_classifications: List[str] = None,
top_k: int = 10
) -> List[Dict]:
"""
Hybrid search using both dense and sparse vectors with access-control filtering.
This uses Qdrant's native prefetch-and-fuse API. Both the dense and sparse
signals contribute to retrieval. The Fusion.RRF strategy applies Reciprocal
Rank Fusion internally.
"""
from qdrant_client.models import Prefetch, FusionQuery, Fusion
if permitted_classifications is None:
permitted_classifications = ["PUBLIC", "INTERNAL"]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if jurisdiction:
must_conditions.append(
FieldCondition(key="jurisdiction", match=MatchValue(value=jurisdiction))
)
search_filter = Filter(must=must_conditions)
results = client.query_points(
collection_name=COLLECTION_NAME,
prefetch=[
Prefetch(
query=dense_query,
using=DENSE_VECTOR_NAME,
limit=top_k * 5,
filter=search_filter
),
Prefetch(
query=SparseVector(
indices=sparse_query_indices,
values=sparse_query_values
),
using=SPARSE_VECTOR_NAME,
limit=top_k * 5,
filter=search_filter
),
],
query=FusionQuery(fusion=Fusion.RRF),
limit=top_k,
with_payload=True
)
return [
{"chunk_id": hit.id, "score": hit.score, **hit.payload}
for hit in results.points
]
Weaviate
Pour l’étape Weaviate, 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 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 amélioration 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.
Milvus
Pour l’étape Milvus, 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 deviner l’état caché. 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é. 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.
ChromaDB
Pour l’étape ChromaDB, 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 deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute exécution partielle silencieuse. Citez les passages qui justifient réellement la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour l’étape ChromaDB, 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 deviner l’état caché. 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 avoir à lire l’ensemble du système.
Pinecone
Lorsque vous travaillez sur l’étape Pinecone, 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 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 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. Le remplacement des prompts résout rarement un système de récupération insuffisant.
Recherche hybride : combinaison de la récupération dense et sparsa
Lorsque vous travaillez sur l’étape de combinaison des recherches hybrides à densité élevée, notez d’abord les exigences : entrées requises, 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. 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. Un changement fréquent des prompts ne résout que rarement un système de récupération insuffisant.
Fusion du classement par rang réciproque pondéré
Lors du traitement de l’étape de fusion par rang réciproque pondéré, 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 garantir l’intégrité des modifications ultérieures du code. Considérez cette étape 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 exécution partielle silencieuse. 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. Lors du traitement de l’étape de fusion par rang réciproque pondéré, 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 garantir l’intégrité 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.
from typing import List, Dict, Tuple
from collections import defaultdict
def weighted_reciprocal_rank_fusion(
dense_results: List[Tuple[str, float]],
sparse_results: List[Tuple[str, float]],
k: int = 60,
dense_weight: float = 0.6,
sparse_weight: float = 0.4
) -> List[Tuple[str, float]]:
"""
Fuse dense vector search results with sparse BM25 results using weighted RRF.
dense_results: List of (chunk_id, dense_score) sorted by dense score descending.
sparse_results: List of (chunk_id, sparse_score) sorted by sparse score descending.
k: RRF constant. Higher k reduces the influence of top-ranked documents.
Conventional default: 60.
dense_weight / sparse_weight: Relative weights. Tune against your evaluation set.
For corpora with high-precision identifier queries, increase sparse_weight.
Returns fused list of (chunk_id, rrf_score) sorted by rrf_score descending.
"""
rrf_scores: Dict[str, float] = defaultdict(float)
for rank, (chunk_id, _) in enumerate(dense_results, start=1):
rrf_scores[chunk_id] += dense_weight * (1.0 / (k + rank))
for rank, (chunk_id, _) in enumerate(sparse_results, start=1):
rrf_scores[chunk_id] += sparse_weight * (1.0 / (k + rank))
return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
Ajustement du poids dense-spars
L’étape d’ajustement du poids dense-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 rollback avant d’élargir le périmètre. Documentez ensemble le parcours réussi 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 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.
Filtrage des métadonnées : périmètre de récupération et limite d’autorisation
La phase de filtrage des métadonnées et de délimitation du champ de recherche fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire idéal, un cas d’échec ainsi que la note de réversion avant d’élargir le champ d’application. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité plutôt qu’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 indicateurs de qualité changent.
from typing import List, Dict, Optional
from enum import Enum
class ClassificationLevel(Enum):
PUBLIC = 0
INTERNAL = 1
CONFIDENTIAL = 2
def build_access_filter(
user_classification_ceiling: str,
user_jurisdiction: Optional[str] = None,
user_roles: Optional[List[str]] = None
) -> Dict:
"""
Build a Qdrant-compatible filter dict enforcing access control rules.
user_classification_ceiling: Highest classification the user can see.
user_jurisdiction: If set, restrict to chunks applicable to that jurisdiction.
user_roles: If set, restrict to chunks permitted for those roles.
Integrate with your identity provider at request time, not at index time.
This filter represents one layer of the authorisation model; it does not
replace identity verification, audit logging, tenant isolation, or
downstream response controls.
"""
ceiling = ClassificationLevel[user_classification_ceiling].value
permitted = [
level.name
for level in ClassificationLevel
if level.value <= ceiling
]
must_conditions = [
{"key": "classification", "match": {"any": permitted}},
{"key": "status", "match": {"value": "active"}}
]
if user_jurisdiction:
must_conditions.append(
{"key": "jurisdiction", "match": {"value": user_jurisdiction}}
)
if user_roles:
# Chunks with empty permitted_roles are accessible to all roles.
must_conditions.append({
"should": [
{"key": "permitted_roles", "match": {"any": user_roles}},
{"is_empty": {"key": "permitted_roles"}}
]
})
return {"must": must_conditions}
Indexation incrémentale : Ajout de nouveaux documents sans réindexation complète
La phase d’indexation incrémentale « Ajout de nouveaux éléments » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire 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. Nommez les artefacts, définites des critères 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. La phase d’indexation incrémentale « Ajout de nouveaux éléments » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemplaire idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Gardez 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.
import logging
from typing import List, Dict
from datetime import datetime
logger = logging.getLogger(__name__)
class IncrementalIndexManager:
"""
Manages incremental updates to a Qdrant collection using soft deletion.
Production pattern:
1. New chunks are inserted immediately with status='active'.
2. Superseded chunks are marked status='deleted' (soft delete).
3. Retrieval filters exclude deleted chunks without graph rebuild.
4. Full rebuild is triggered on schedule or when deleted fraction exceeds threshold.
"""
def __init__(self, qdrant_client, collection_name: str):
self.client = qdrant_client
self.collection = collection_name
self.deleted_threshold = 0.15 # Rebuild when 15% of index is soft-deleted
def upsert_policy_version(
self,
new_chunks: List[Dict],
superseded_chunk_ids: List[str],
policy_id: str,
new_version: str
) -> Dict:
"""
Insert new policy version chunks and soft-delete superseded ones.
"""
if superseded_chunk_ids:
self.client.set_payload(
collection_name=self.collection,
payload={
"status": "deleted",
"deleted_at": datetime.now().isoformat(),
"superseded_by_version": new_version
},
points=superseded_chunk_ids
)
logger.info(
f"Soft-deleted {len(superseded_chunk_ids)} chunks "
f"from policy {policy_id}, superseded by version {new_version}"
)
from qdrant_client.models import PointStruct
points = [
PointStruct(
id=chunk["chunk_id"],
vector={"dense": chunk["dense_vector"]},
payload={
**{k: v for k, v in chunk.items()
if k not in ("chunk_id", "dense_vector")},
"status": "active",
"indexed_at": datetime.now().isoformat()
}
)
for chunk in new_chunks
]
self.client.upsert(collection_name=self.collection, points=points)
logger.info(
f"Inserted {len(new_chunks)} chunks for policy {policy_id} version {new_version}"
)
return {
"inserted": len(new_chunks),
"soft_deleted": len(superseded_chunk_ids),
"policy_id": policy_id,
"new_version": new_version
}
def should_rebuild(self) -> bool:
"""Check whether the fraction of soft-deleted vectors justifies a full rebuild."""
from qdrant_client.models import Filter, FieldCondition, MatchValue
total = self.client.get_collection(self.collection).vectors_count
deleted_filter = Filter(must=[
FieldCondition(key="status", match=MatchValue(value="deleted"))
])
deleted_count = self.client.count(
collection_name=self.collection,
count_filter=deleted_filter
).count
fraction = deleted_count / total if total > 0 else 0
logger.info(
f"Index health: {deleted_count}/{total} soft-deleted ({fraction:.1%})"
)
return fraction >= self.deleted_threshold
Intégration des modifications de version du modèle
Pour l’étape des modifications de version du modèle d’incorporation, 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 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. Préférez des sorties structurées avec validation de schéma plutôt que du texte libre lorsque l’étape suivante consiste en du code ou une appel à outil.
Sharding et réplication : échelle au-delà d’un seul nœud
Pour l’étape d’escalade par shardage et de réplication, définissez les entrées, le responsable de l’étape ainsi que 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 des états cachés. 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é. 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.
Stratégies de shardage
Pour l’étape des Stratégies de Shardage, 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é. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute exécution partielle silencieuse. Citez les passages qui justifient réellement la réponse. Sans citations, les opérateurs ne peuvent pas distinguer une hallucination d’un manque d’indexation. Pour l’étape des Stratégies de Shardage, 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 avoir à lire l’ensemble du système.
Planification de la capacité
Lors de l’étape de planification de la capacité, notez d’abord les éléments requis : les données nécessaires, 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 à la fois 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.
from dataclasses import dataclass
@dataclass
class VectorIndexCapacityPlan:
"""
Illustrative capacity model for HNSW vector indexes.
All figures are approximations for planning purposes.
Benchmark against your actual implementation and workload.
"""
n_vectors: int
dimension: int
hnsw_m: int = 16
replication_factor: int = 2
avg_payload_bytes: int = 2048
memory_headroom_factor: float = 1.5
def vector_storage_gb(self) -> float:
return (self.n_vectors * self.dimension * 4) / (1024 ** 3)
def hnsw_graph_gb(self) -> float:
# Approximate; actual graph overhead varies by implementation and configuration.
return (self.n_vectors * self.hnsw_m * 2 * 8) / (1024 ** 3)
def payload_storage_gb(self) -> float:
return (self.n_vectors * self.avg_payload_bytes) / (1024 ** 3)
def total_index_gb(self) -> float:
return self.vector_storage_gb() + self.hnsw_graph_gb() + self.payload_storage_gb()
def memory_per_replica_gb(self) -> float:
return self.total_index_gb() * self.memory_headroom_factor
def total_cluster_memory_gb(self) -> float:
# Each replica holds a full copy of the index.
return self.memory_per_replica_gb() * self.replication_factor
def report(self) -> str:
return (
f"Illustrative capacity model — {self.n_vectors:,} vectors at {self.dimension}d:\n"
f" Vector storage (approx): {self.vector_storage_gb():.2f} GB\n"
f" HNSW graph estimate: {self.hnsw_graph_gb():.2f} GB\n"
f" Payload storage (approx): {self.payload_storage_gb():.2f} GB\n"
f" Index footprint before overhead: {self.total_index_gb():.2f} GB\n"
f" Per-replica memory + headroom: {self.memory_per_replica_gb():.2f} GB\n"
f" Total cluster memory (approx): {self.total_cluster_memory_gb():.2f} GB\n"
f" ({self.replication_factor} replicas, each holding a full copy)\n"
f" Treat these as planning estimates, not deployment guarantees.\n"
f" Benchmark against your implementation before provisioning."
)
# Illustrative example: banking policy corpus
plan = VectorIndexCapacityPlan(
n_vectors=500_000,
dimension=1536,
hnsw_m=16,
replication_factor=2,
avg_payload_bytes=2048,
memory_headroom_factor=1.5
)
print(plan.report())
Le pipeline complet de récupération
Lorsque vous travaillez sur l’étape du « The Complete Retrieval Pipeline », 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. 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é. Mesurez le taux de rappel sur un ensemble de questions fixe avant d’ajuster les prompts. Changer fréquemment les prompts ne résout que rarement un système de récupération insuffisant.
User Query
↓
Query Embedding (dense + sparse)
↓
Metadata / Authorisation Constraints
↓
Dense Vector Retrieval (filtered HNSW)
+
Sparse Retrieval (BM25)
↓
Weighted RRF Fusion
↓
Candidate Documents with Provenance Metadata
↓
[Part 7: Reranking and Context Assembly]
↓
LLM Generation
import logging
import re
from typing import List, Dict, Optional
from dataclasses import dataclass
from collections import defaultdict
logger = logging.getLogger(__name__)
@dataclass
class RetrievalConfig:
dense_candidate_pool: int = 50
sparse_candidate_pool: int = 50
rrf_k: int = 60
dense_weight: float = 0.6
sparse_weight: float = 0.4
final_top_k: int = 10
score_threshold: float = 0.2
class BankingPolicyRetriever:
"""
Production retrieval pipeline for a regulated banking policy corpus.
Combines dense vector search, BM25 sparse retrieval, access filtering,
and weighted RRF score fusion.
Retrieval ends at the fused candidate list. Reranking and context assembly
are handled in Part 7.
"""
def __init__(
self,
qdrant_client,
collection_name: str,
embedding_pipeline,
bm25_index,
chunk_store: Dict[str, Dict],
config: Optional[RetrievalConfig] = None
):
self.client = qdrant_client
self.collection = collection_name
self.embedder = embedding_pipeline
self.bm25 = bm25_index
self.chunk_store = chunk_store
self.config = config or RetrievalConfig()
def retrieve(
self,
query: str,
user_classification_ceiling: str = "INTERNAL",
user_jurisdiction: Optional[str] = None,
user_roles: Optional[List[str]] = None
) -> List[Dict]:
"""
Full hybrid retrieval with access control.
Returns top-k chunks with provenance metadata, access-filtered
for the requesting user's classification ceiling and jurisdiction.
"""
from qdrant_client.models import Filter, FieldCondition, MatchValue, MatchAny
query_embedding = self.embedder.embed_query(query)
classification_levels = {"PUBLIC": 0, "INTERNAL": 1, "CONFIDENTIAL": 2}
ceiling = classification_levels.get(user_classification_ceiling, 1)
permitted_classifications = [
k for k, v in classification_levels.items() if v <= ceiling
]
must_conditions = [
FieldCondition(
key="classification",
match=MatchAny(any=permitted_classifications)
),
FieldCondition(key="status", match=MatchValue(value="active"))
]
if user_jurisdiction:
must_conditions.append(
FieldCondition(
key="jurisdiction",
match=MatchValue(value=user_jurisdiction)
)
)
access_filter = Filter(must=must_conditions)
# Dense vector search with integrated access filtering
dense_hits = self.client.search(
collection_name=self.collection,
query_vector=("dense", query_embedding),
query_filter=access_filter,
limit=self.config.dense_candidate_pool,
score_threshold=self.config.score_threshold,
with_payload=True
)
dense_results = [(hit.id, hit.score) for hit in dense_hits]
# Sparse BM25 retrieval with post-retrieval access filtering
tokens = re.findall(r'\b\w+\b', query.lower())
bm25_scores = self.bm25.get_scores(tokens)
sparse_ranked = sorted(enumerate(bm25_scores), key=lambda x: x[1], reverse=True)
chunk_ids = list(self.chunk_store.keys())
sparse_results = []
for corpus_idx, score in sparse_ranked:
if score <= 0 or len(sparse_results) >= self.config.sparse_candidate_pool:
break
chunk_id = chunk_ids[corpus_idx]
chunk_meta = self.chunk_store.get(chunk_id, {})
if chunk_meta.get("classification") not in permitted_classifications:
continue
if user_jurisdiction and chunk_meta.get("jurisdiction") != user_jurisdiction:
continue
if chunk_meta.get("status") != "active":
continue
sparse_results.append((chunk_id, score))
# Weighted RRF fusion
rrf_scores: Dict[str, float] = defaultdict(float)
k = self.config.rrf_k
for rank, (chunk_id, _) in enumerate(dense_results, start=1):
rrf_scores[chunk_id] += self.config.dense_weight * (1.0 / (k + rank))
for rank, (chunk_id, _) in enumerate(sparse_results, start=1):
rrf_scores[chunk_id] += self.config.sparse_weight * (1.0 / (k + rank))
fused = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
top_chunk_ids = [cid for cid, _ in fused[:self.config.final_top_k]]
# Assemble results with provenance metadata
results = []
for chunk_id in top_chunk_ids:
chunk_data = self.chunk_store.get(chunk_id, {})
results.append({
"chunk_id": chunk_id,
"rrf_score": rrf_scores[chunk_id],
**chunk_data
})
logger.info(
f"Retrieval complete: query={query[:60]!r}, "
f"dense_candidates={len(dense_results)}, "
f"sparse_candidates={len(sparse_results)}, "
f"final_results={len(results)}"
)
return results
Observabilité : Que mesurer en production
Lors de l’étape d’Observabilité : Que mesurer ?, écrivez 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 garantit l’honnêteté des modifications ultérieures du code. Considérez cette étape 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 les terminations partielles silencieuses. Mesurez le taux de rappel sur un ensemble fixe de questions avant d’ajuster les prompts. Le changement fréquent des prompts résout rarement un système de récupération insuffisant. Lors de l’étape d’Observabilité : Que mesurer ?, écrivez 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 garantit l’honnêteté des modifications ultérieures du code. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes 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.
import time
import logging
from typing import Callable, TypeVar, Any
from functools import wraps
logger = logging.getLogger(__name__)
F = TypeVar("F", bound=Callable[..., Any])
def retrieval_instrumented(func: F) -> F:
"""
Decorator that adds structured latency logging and empty-result alerting
to retrieval functions. Wrap your primary retrieve() method in production.
"""
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = None
error = None
try:
result = func(*args, **kwargs)
return result
except Exception as e:
error = str(e)
raise
finally:
elapsed_ms = (time.perf_counter() - start) * 1000
n_results = len(result) if result is not None else 0
log_payload = {
"function": func.__name__,
"latency_ms": round(elapsed_ms, 2),
"n_results": n_results,
"error": error
}
query = kwargs.get("query", args[1] if len(args) > 1 else None)
if query:
log_payload["query_prefix"] = str(query)[:80]
if error:
logger.error("retrieval_error", extra=log_payload)
elif n_results == 0:
logger.warning("retrieval_empty_result", extra=log_payload)
elif elapsed_ms > 500:
logger.warning("retrieval_high_latency", extra=log_payload)
else:
logger.info("retrieval_success", extra=log_payload)
return wrapper # type: ignore
Retour à la proposition de crédit de 12 millions d’euros
La phase de retour à l’état initial fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un enregistrement exemplaire, un cas d’échec et la note de réversion avant d’élargir le périmètre. Documentez ensemble le parcours optimal et celui 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 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é évoluent.
Au préalable du passage au réclassement : une liste de vérification pour la production
La phase « Avant de migrer » fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un exemple réussi, 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é changent.
Que vient ensuite ?
La phase « What Comes Next » 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. Traitez 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. La phase « What Comes Next » 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. Conservez les configurations 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.
Liste de contrôle opérationnelle
Pour l’étape de la liste de contrôle opérationnelle, définissez les entrées, le responsable de l’étape et les critères d’achèvement 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 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 l’environnement de démonstration à des environnements partagés.
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 dans l’indexation.
Suivez les coûts et la latence en même temps que la qualité. Une réponse légèrement moins bonne mais coûtant 10 fois moins peut être le meilleur compromis en environnement de production.
Fixez les versions des dépendances et enregistrez le résumé de l’image ayant servi à exécuter la démonstration. La reproductibilité vaut mieux que les connaissances empiriques.
Préférez des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’erreur doit indiquer une seule responsabilité et non un processus embrouillé.
Au préalable de promouvoir la pile technologique, figez les versions, conservez une transcription exemplaire pour le chemin critique, et vérifiez les étapes de rollback. Les environnements partagés nécessitent des limites de fréquence, des contrôles d’attribution et un responsable clair pour la rotation des secrets. Préférez une fiabilité banale à des démonstrations originales mais éphémères.
Note de batch pour fa68a70d815a : gardez les clés du fournisseur hors du répertoire, fixez un plafond pour les tokens par session, et stockez les transcriptions à côté des fichiers de configuration eval afin que les remplacements ultérieurs de modèles restent comparables.
Pour l’étape 0 de la note de renforcement, 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 deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute exécution partielle silencieuse.
Détail de renforcement 0/864 : mesurez le temps d’exécution, la classe de l’erreur et la consommation de tokens pour cette note, puis décidez si vous conservez le changement en vous basant sur un ensemble de questions prédéfini plutôt que sur des observations subjectives.
Lors de la première étape des notes de renforcement, 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.
Détail de renforcement 1/864 : mesurez le temps d’exécution, la catégorie de l’erreur et l’utilisation des tokens pour cette note, puis décidez si vous souhaitez conserver la modification en vous basant sur un ensemble de critères prédéfinis plutôt que sur des observations subjectives.
La deuxième étape des notes de renforcement fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un exemple idéal de fonctionnement, un cas d’échec et une 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’erreur doit indiquer une responsabilité précise plutôt qu’un processus embrouillé.
Détail de renforcement 2/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des observations anecdotiques.
Pour la phase 3 de la note de renforcement, définir les entrées, le responsable de l’étape et les critères d’achèvement 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é. Enregistrer 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 permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés.
Détail de renforcement 3/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des observations anecdotiques.
Lors de la réalisation de l’étape 4 des notes de renforcement, notez d’abord les conditions contractuelles : entrées requises, 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 livrés font partie intégrante du produit, et non d’améliorations apportées ultérieurement.
Détail de renforcement 4/864 : mesurez le temps d’exécution, la catégorie de l’erreur et l’utilisation des tokens pour cette note, puis décidez si vous conservez la modification en vous basant sur un ensemble de questions prédéfinies plutôt que sur des observations subjectives.
L’étape 5 des notes de renforcement fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Recueillez un exemple idéal de fonctionnement, un cas d’échec et une note de réversion avant d’élargir le périmètre. Considérez cette étape 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 les terminations partielles silencieuses.
Détail de renforcement 5/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Pour la phase 6 de la note de renforcement, définir les entrées, le responsable de l’étape et les critères d’achèvement 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é. Conserver 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.
Détail de renforcement 6/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Lors de l’étape 7 des notes de renforcement, notez d’abord les éléments essentiels : 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. 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é.
Détail de renforcement 7/864 : mesurez le temps d’exécution, la catégorie de l’erreur et la consommation de tokens pour cette note, puis décidez si vous souhaitez conserver la modification en vous basant sur un ensemble de critères prédéfinis plutôt que sur des observations subjectives.
L’étape 8 des notes de renforcement fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple idéal de fonctionnement, un cas d’échec et des notes 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 des surprises lors du passage de l’environnement de démonstration aux environnements partagés.
Détail de renforcement 8/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Pour l’étape 9 de la note de renforcement, définir les entrées, le responsable de l’étape et les critères d’achèvement 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é. Documenter 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 livrés font partie du produit, et non d’une mise en forme ultérieure.
Détail de renforcement 9/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Lors de l’exécution de l’étape 10 des notes de renforcement, notez d’abord les conditions contractuelles : 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. Considérez cette étape 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 les terminaisons partielles silencieuses.
Détail de renforcement 10/864 : mesurez le temps d’exécution, la classe de l’erreur et la consommation de tokens pour cette note, puis décidez si vous souhaitez conserver la modification en vous basant sur un ensemble de questions prédéfini plutôt que sur des observations subjectives.
L’étape 11 des notes de renforcement fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple idéal, un cas d’échec et la note de réversion avant d’élargir le périmètre. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets 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.
Détail de renforcement 11/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Pour l’étape 12 du processus de renforcement, définir les entrées, le responsable de l’étape et les critères d’achèvement 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érer des unités petites et testables plutôt que des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé.
Détail de renforcement 12/864 : mesurer le temps d’exécution, la classe d’erreur et la consommation de tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble de questions prédéfini plutôt que sur des anecdotes.
Lors de l’exécution de l’étape 13 des notes de renforcement, notez d’abord les éléments essentiels : 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.
Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés.
Détail 13/864 du renforcement : mesurez le temps d’exécution réel, la catégorie de l’erreur et la consommation de jetons pour cette note, puis décidez si vous souhaitez conserver la modification en vous basant sur un ensemble de critères prédéfinis plutôt que sur des observations subjectives.
L’étape 14 des notes de renforcement fonctionne le mieux lorsqu’elle est considérée comme une surface mesurable. Capturez un exemple idéal de fonctionnement, un cas d’échec et la note de réversion avant d’élargir le périmètre.
Dokumentez ensemble le parcours normal et le parcours de récupération. Les tentatives de répétition, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations apportées ultérieurement.
Détail de renforcement 14/864 : mesurer le temps d’exécution, la classe d’erreur et l’utilisation des tokens pour cette note, puis décider de conserver ou non le changement en se basant sur un ensemble fixe de questions plutôt que sur des anecdotes.