DRAG avec KNEE : découpage dynamique pour un RAG performant
DRAG avec KNEE est un algorithme d'augmentation de génération par récupération dynamique qui utilise une méthode de suppression du point de genou pour optimiser la sélection du contexte. Il construit des arbres vectoriels hiérarchiques dans Qdrant et exploite une recherche hybride avec fusion du rang réciproque (RRF) afin de récupérer de manière adaptative le contexte pertinent — résolvant ainsi les limites des approches statiques basées sur top_k, comme le débordement de contexte ou la perte d'information.
Construction d'index hiérarchiques
Les documents sont divisés en pages. Pour chaque page, un modèle linguistique (LLM) génère un résumé concis et des termes clés via une sortie structurée — ces éléments deviennent des nœuds feuilles de l'arbre.
Les feuilles sont regroupées en nœuds parents (3 à 5 par groupe). Chaque parent hérite des mots-clés enfants et génère son propre résumé en agrégeant le contenu des enfants. Ce processus se répète jusqu’à atteindre le nœud racine (parent_id = -1).
Dans Qdrant, les données sont stockées sous forme de liste plate de points avec des métadonnées pour le filtrage. Cela permet une manipulation flexible de la structure hiérarchique via les champs de charge utile (payload).
Recherche hybride et RRF
La comparaison des points combine des embeddings denses et creux grâce à la fusion du rang réciproque :
$$RRFscore(d \in D) = \sum_{r \in R} \frac{1}{k + r(d)}$$
où $D$ est l'ensemble des documents, $R$ représente les classements (denses, creux), et $k=60$.
Algorithmes de recherche
Recherche par branche
Trouve les racines de haut niveau, puis compare récursivement les parents aux enfants. Si les enfants obtiennent un meilleur score, l’algorithme descend plus profondément ; sinon, il s’arrête sur le parent.
def parent_vs_children(query: str, parent: ScoredPoint) -> list[ScoredPoint]:
file_name = parent.payload["file_name"]
parent_id = parent.payload["id"]
child_ids = parent.payload["child_ids"]
all_ids = [parent_id] + child_ids
if len(child_ids) == 0:
console.print(
f"Parent with id {parent_id} is leave!", style="bold underline violet"
)
return []
console.print()
console.print("#" * 20, style="red")
console.print(
f"Children (ids: {child_ids}) [underline]VS[/underline] Parent (id: {parent_id}) [ FILE: {file_name} ]\nFor query: {query[:20]}...",
style="bold violet",
)
with RAGalicClient() as client: # artifacts from a previous life :)
dense_model: str = client.client.embedding_model_name # type: ignore
sparse_model: str = client.client.sparse_embedding_model_name # type: ignore
children_and_parent_filter = Filter(
must=FieldCondition(key="id", match=MatchAny(any=all_ids))
)
sorted_points = client.client.query_points(
collection_name="ragalic",
prefetch=[
Prefetch(
query=Document(text=query, model=dense_model),
using="dense",
limit=len(all_ids),
filter=children_and_parent_filter,
),
Prefetch(
query=Document(text=query, model=sparse_model),
using="sparse",
limit=len(all_ids),
filter=children_and_parent_filter,
),
],
query=FusionQuery(fusion=Fusion.RRF),
query_filter=children_and_parent_filter,
limit=len(all_ids),
).points
children_better_then_parent = []
for point in sorted_points:
if point.payload["id"] == parent_id:
break
children_better_then_parent.append(point)
ids = [child.payload["id"] for child in children_better_then_parent]
console.print(
f"Child ids which is better than parent: {ids if ids else 'N/A'}",
style="italic purple",
)
console.print("#" * 20, style="red")
console.print()
return children_better_then_parent
def branch_search(query: str, num_roots: int = 3) -> list[Chunk]:
roots = find_roots(query=query, num_to_find=num_roots)
final_points = []
points_to_process = roots
while len(points_to_process) > 0:
new_point_to_process = []
for point in points_to_process:
new_points = parent_vs_children(query=query, parent=point)
if len(new_points) == 0:
console.print(
f"--- NEW FINAL POINT (id: {point.payload['id']} | file: {point.payload['file_name']} | pages: {point.payload['page_start']} - {point.payload['page_end']}) ---",
style="bold green",
)
final_points.append(point)
else:
new_point_to_process.extend(new_points)
points_to_process = new_point_to_process
console.print(
f"--- WAS FOUND {len(final_points)} POINTS FOR QUERY: '{query[:20]}...' ---",
style="bold green on white",
)
return prepare_chunks(final_points)Utilisation :
from rag_lib.search import branch_search
results = branch_search(query="your_query", num_roots=3)
for chunk in results:
print(f"{chunk.file_name} (pages {chunk.page_start}-{chunk.page_end}):
{chunk.text[:200]}...")Avantages : Haute couverture. Inconvénients : Peut inclure des branches non pertinentes.
Recherche par faisceau (fixe)
- Prend la génération courante des parents et de leurs enfants.
- Les gagnants parmi les parents éliminent leurs enfants ; les gagnants parmi les enfants éliminent leurs parents.
- Les survivants s'affrontent pour une largeur de faisceau fixe.
Réduit le bruit mais risque d’éliminer du contenu pertinent ou d’inclure des éléments hors sujet.
Faisceau adaptable avec point de genou
Largeur de faisceau dynamique utilisant la méthode du point de genou. Identifie le point d'inflexion sur la courbe RRF (convexité vers le bas).
Suppression dynamique du bruit : à chaque itération, appliquer d’abord la suppression par point de genou, puis générer le prochain faisceau. Élimine environ 62 % des documents non pertinents.
Points clés
- Arbres hiérarchiques dans Qdrant avec mots-clés hérités.
- Recherche hybride + RRF pour comparer efficacement les nœuds.
- Suppression par point de genou pour un découpage dynamique basé sur la pertinence.
- Recherche par branche pour une exhaustivité, faisceau adaptable pour une efficacité.
- Sortie structurée du LLM pour minimiser les hallucinations.
— Editorial Team
Aucun commentaire pour le moment.