Nota : développement en cours
Contexte et Prérequis
Contexte :
– un serveur privé,
– le langage Python,
– les bibliothèques LlamaIndex et HugginFace,
– un LLM capable de fonctions (ex. : Mistral medium ou large),
– un modèle d’embeddings compatible (ex. : Sentence-BERT, OpenAI embeddings, BAAI Bge small etc.).
Prérequis :
Un index de documents construit avec nos outils : Ingestion : le RAG Manager, Le Web Service IngestWSG.
La construction de l’outil RetrieverTool
# retriever_tool.py
from typing import List, Dict, Any
from llama_index import Document
class RetrieverTool:
def __init__(self, retriever, embedder=None, hybrid_fn=None):
self.retriever = retriever
self.embedder = embedder
self.hybrid_fn = hybrid_fn
def run(self, query: str, k: int = 5, metadata_filter: Dict[str,Any] = None, hybrid: bool = False) -> List[Dict]:
# Optionnel : hybrid search combine lexical + vector
if hybrid and self.hybrid_fn:
hits = self.hybrid_fn(query, k=k, metadata_filter=metadata_filter)
else:
# LlamaIndex retriever: get top-k documents (API may be retrieve or get_relevant_documents)
hits = self.retriever.retrieve(query, search_kwargs={"k": k, "filter": metadata_filter})
results = []
for doc in hits:
text = getattr(doc, "text", str(doc))
meta = getattr(doc, "metadata", {})
# Optionnel : tronquer / extraire passage le plus pertinent
snippet = text[:1000]
results.append({"snippet": snippet, "metadata": meta, "source_id": meta.get("source_id")})
return resultsFiltrage par métadonnées
Le filtrage par métadonnées consiste à appliquer des contraintes booléennes ou de plage avant ou après la recherche vectorielle pour ne considérer que les documents pertinents au contexte métier. Exemples de métadonnées : source, langue, date, client_id, mention de confidentialité etc.
En pratique on peut :
• Pré‑filtrer l’ensemble indexé (ex. ) puis exécuter la recherche vectorielle sur ce sous‑ensemble ;
• Post‑filtrer les résultats vectoriels (réduire ou réordonner) si le moteur vectoriel ne supporte pas de filtre natif.
Avantage clé : réduit le bruit et évite que des documents hors‑contexte soient ramenés au LLM, ce qui diminue les hallucinations et le coût des appels LLM.
Patterns d’implémentation
• Filtres simples : dictionnaire transmis au retriever (beaucoup de vector stores supportent un paramètre filter).
• Filtres complexes : expressions booléennes ou fonctions de scoring métier (ex. boost si client_id correspond).
• Sécurité : appliquer les règles d’accès (ACL) en pré‑filtre pour éviter toute fuite d’information.
Bonnes pratiques :
• limiter le nombre de documents renvoyés (k = 3–10),
• renvoyer des snippets + métadonnées pour que l’agent puisse raisonner.
Hybrid Search
Le concept hybrid search désigne la combinaison d’une recherche lexicale (BM25/keyword) et d’une recherche sémantique (embeddings) pour améliorer précision et rappel.
Le hybrid search combine une recherche lexicale (BM25 ou TF‑IDF) et une recherche vectorielle (embeddings) pour tirer parti des forces complémentaires :
– BM25 excelle sur les correspondances exactes et mots‑clés,
- les embeddings capturent le sens et les synonymes.
Le flux typique consiste à exécuter BM25 et vector search séparément, fusionner les listes et reranker (ou reranker uniquement les top N).
Stratégies de fusion et pondération
– Score normalisé : normaliser scores BM25 et vectoriels (min‑max) puis combiner : `score = α * score_vector + (1-α) * score_bm25`. Ajuster `α` selon requête (courte → plus lexical).
– Ensemble / cascade : BM25 pour récupérer candidates (large), puis rerank par similarité vectorielle. Ou inversement si on veut donner la priorité au sémantique.
– Reranking par LLM : prendre top‑N fusionnés et demander au LLM de reranker ou d’extraire les passages clés.
Conseils pratiques et pièges à éviter
– Mesurer : A/B test avec métriques de rappel/precision et qualité des réponses LLM.
– Coût : hybrid augmente les opérations (deux recherches) — cache les embeddings de requêtes et les résultats fréquents.
– Normalisation : tokenisation cohérente entre index lexical et embeddings.
– Sécurité et filtres : toujours appliquer ACL en pré‑filtre.
Exemple minimal (pseudocode)
bm25_hits = bm25.search(query, k=50)
vec_hits = vector_store.search(query_embedding, k=50, filter=meta_filter)
candidates = merge_and_dedup(bm25_hits, vec_hits)
for c in candidates: c.score = alpha*norm(c.vec_score)+(1-alpha)*norm(c.bm25_score)
top = sorted(candidates, key=lambda x: x.score, reverse=True)[:k]exemple concret pour FAISS + BM25 (code prêt à brancher sur l’index) :
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
# documents: list of dicts {'id':int,'text':str,'meta':{...}}
docs = load_documents()
# BM25
tokenized = [d['text'].split() for d in docs]
bm25 = BM25Okapi(tokenized)
# Embeddings + FAISS
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
embs = model.encode([d['text'] for d in docs]).astype('float32')
index = faiss.IndexFlatL2(embs.shape[1])
index.add(embs)
id_map = {i: docs[i]['id'] for i in range(len(docs))}
meta_map = {d['id']: d['meta'] for d in docs}
def hybrid_search(query, k=10, meta_filter=None, alpha=0.6):
# BM25 candidates
bm25_scores = bm25.get_scores(query.split())
bm25_top = np.argsort(bm25_scores)[-200:][::-1]
# Vector candidates (apply meta_filter by masking ids)
q_emb = model.encode([query]).astype('float32')
D, I = index.search(q_emb, 200)
vec_candidates = I[0]
# Merge and dedupe
candidates = []
seen = set()
for idx in list(bm25_top) + list(vec_candidates):
doc_id = id_map[idx]
if meta_filter and not meta_matches(meta_map[doc_id], meta_filter):
continue
if doc_id in seen: continue
seen.add(doc_id)
candidates.append({
'doc_id': doc_id,
'bm25_score': float(bm25_scores[idx]),
'vec_score': float(1.0 / (1.0 + D[0][list(vec_candidates).index(idx)]) ) if idx in vec_candidates else 0.0
})
# Normalize and combine
bm = np.array([c['bm25_score'] for c in candidates])
ve = np.array([c['vec_score'] for c in candidates])
if len(bm)>0:
bm_n = (bm - bm.min())/(bm.ptp()+1e-9)
ve_n = (ve - ve.min())/(ve.ptp()+1e-9)
for i,c in enumerate(candidates):
c['score'] = alpha*ve_n[i] + (1-alpha)*bm_n[i]
topk = sorted(candidates, key=lambda x: x['score'], reverse=True)[:k]
return topk
RSS 2.0