DRAG with KNEE: как реализовать динамический RAG с интеллектуальной обрезкой контекста на Python

Сломанный конвейер: почему обычный RAG отдаёт вам мусор

Вы настраиваете RAG-систему. Всё по учебнику: чанкируете документы, считаете эмбеддинги через text-embedding-3-large, кладёте в Qdrant. Запрос — и векторный поиск возвращает топ-10 фрагментов. Вы смотрите на них и понимаете: только два действительно релевантны. Остальные восемь — шум, который съедает контекстное окно LLM и портит ответ.

Проблема не в том, что поиск плохой. Проблема в том, что он статичный. Фиксированное количество чанков — это костыль. Иногда нужно три фрагмента, иногда — пятнадцать. Как определить оптимальное количество? Методом тыка? Нет, математикой.

DRAG with KNEE: динамический поиск с хирургической точностью

DRAG (Dynamic Retrieval Augmented Generation) with KNEE — алгоритм, который определяет, сколько чанков действительно нужно для ответа. Не больше, не меньше. Он строит иерархическое векторное дерево, находит точку "колена" на графике сходства и отсекает всё лишнее.

"Колено" (knee point) — это момент, когда добавление следующего чанка даёт мизерный прирост релевантности. Дальше идёт пологий хвост — шум. Алгоритм находит этот излом и обрезает.

1 Установка и подготовка: что нужно в 2026 году

Библиотеки устаревают быстро. Вот актуальный набор на март 2026:

pip install qdrant-client==1.9.0  # Последняя стабильная на 2026
torch==2.3.0  # Обязательно с поддержкой новых GPU
sentence-transformers==3.2.0  # Или используйте API OpenAI
experiment  # Нет, это не библиотека, это ваш настрой

Для эмбеддингов берите что-то из нового: text-embedding-3-large от OpenAI (если не боитесь API-стоимости) или EmbeddingGemma для локального запуска. Последняя хорошо показывает себя на мультиязычных данных.

2 Строим иерархическое векторное дерево

Обычный RAG работает с плоским списком чанков. DRAG строит дерево: родительские узлы — это обобщения (суммаризации или центроиды эмбеддингов), дочерние — конкретные фрагменты.

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import numpy as np
from typing import List, Tuple

class HierarchicalTree:
    def __init__(self, vectors: List[List[float]], texts: List[str], max_children: int = 5):
        self.vectors = vectors
        self.texts = texts
        self.max_children = max_children  # Сколько чанков на узел
        self.tree = self._build_tree(vectors, texts)
    
    def _build_tree(self, vectors, texts):
        # 1. Кластеризуем векторы (упрощённо — по сходству)
        # 2. Для каждого кластера считаем центроид — это родительский узел
        # 3. Сохраняем исходные векторы как дочерние
        # В реальности используют HNSW или подобные структуры
        tree_nodes = []
        for i in range(0, len(vectors), self.max_children):
            chunk = vectors[i:i + self.max_children]
            centroid = np.mean(chunk, axis=0).tolist()  # Родитель
            node = {
                'centroid': centroid,
                'children': list(range(i, min(i + self.max_children, len(vectors))))
            }
            tree_nodes.append(node)
        return tree_nodes

Дерево нужно хранить там же, где и векторы. Qdrant поддерживает вложенные структуры с 2025 года. Или можно хранить отдельно в графе знаний — как в статье про иерархический граф.

3 Алгоритм поиска с анализом "колена"

Сначала ищем по родительским узлам (центроидам), получаем предварительный набор. Потом спускаемся к детям и ранжируем все дочерние чанки по сходству с запросом.

def find_knee_point(similarities: List[float], threshold: float = 0.1) -> int:
    """
    Находит точку колена на графике убывающих сходств.
    similarities — отсортированный по убыванию список косинусных сходств.
    threshold — минимальный перепад для определения колена (экспериментально).
    """
    if len(similarities) < 2:
        return len(similarities)
    
    # Считаем разности между соседними значениями
    diffs = [similarities[i] - similarities[i+1] for i in range(len(similarities)-1)]
    
    # Ищем первый разрыв, превышающий порог
    for i, diff in enumerate(diffs):
        if diff > threshold:
            return i + 1  # Колено после i-го элемента
    
    # Если явного колена нет, возвращаем всё
    return len(similarities)

# Пример использования
similarities = [0.95, 0.93, 0.92, 0.91, 0.65, 0.63, 0.62, 0.61]
knee_index = find_knee_point(similarities, threshold=0.2)
print(f"Берём первые {knee_index} чанков")  # Вернёт 4

4 Слияние результатов через RRF (Reciprocal Rank Fusion)

Когда у вас несколько источников (например, векторный поиск и ключевые слова), нужно сливать ранги. RRF — простой и эффективный метод.

def reciprocal_rank_fusion(rankings: List[List[str]], k: int = 60):
    """
    rankings — список ранжированных списков (каждый список — это ID чанков).
    k — константа для сглаживания (обычно 60).
    Возвращает объединённый ранжированный список ID.
    """
    scores = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            if doc_id not in scores:
                scores[doc_id] = 0
            scores[doc_id] += 1 / (rank + k)
    
    # Сортируем по убыванию score
    fused_ranking = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_id for doc_id, score in fused_ranking]

# Пример: два метода поиска дали разные результаты
vector_results = ["chunk_123", "chunk_456", "chunk_789"]
keyword_results = ["chunk_456", "chunk_123", "chunk_999"]
final = reciprocal_rank_fusion([vector_results, keyword_results])
print(final)  # ['chunk_123', 'chunk_456', 'chunk_789', 'chunk_999']

RRF хорош тем, что не требует калибровки scores от разных моделей. Работает на рангах. В 2026 году появились гибридные методы, но RRF остаётся эталоном для быстрого прототипирования.

Где ломается DRAG with KNEE: нюансы, которые никто не рассказывает

Алгоритм чувствителен к однородности данных. Если все чанки имеют примерно одинаковое сходство с запросом (плоское распределение), колено не найдётся. Система либо возьмёт всё, либо обрежет наугад.

Что делать:

• Добавляйте второй критерий — минимальное количество чанков. Если колено не найдено, берите top-N (где N — safe default, например 3).
• Комбинируйте с графами знаний. Для сложных запросов, где нужны связи между понятиями, используйте обратный обход графа.
• Настраивайте порог динамически — можно обучать простую модель на исторических запросах, чтобы предсказывать оптимальный порог для каждого типа запроса.

Сколько это сэкономит? Цифры на 2026 год

На тестовом датасете из 10 000 запросов к технической документации:

Экономия токенов напрямую конвертируется в деньги, если используете платные LLM вроде GPT-4o или Claude 3.5. Для локальных моделей — это возможность обрабатывать более длинные контексты без деградации.

Интеграция в пайплайн: как не переписать всё с нуля

Вы уже используете LangChain или LlamaIndex? DRAG with KNEE можно встроить как кастомный ретривер. Пример для LangChain 0.2.0+:

from langchain.retrievers import BaseRetriever
from langchain.schema import Document

class DragWithKneeRetriever(BaseRetriever):
    def __init__(self, qdrant_client, tree, embedder):
        self.client = qdrant_client
        self.tree = tree
        self.embedder = embedder
    
    def get_relevant_documents(self, query: str) -> List[Document]:
        # 1. Эмбеддинг запроса
        query_vector = self.embedder.embed_query(query)
        
        # 2. Поиск по центроидам (родительский уровень)
        centroid_results = self.client.search(
            collection_name="centroids",
            query_vector=query_vector,
            limit=10
        )
        
        # 3. Сбор всех дочерних чанков
        all_chunk_ids = []
        for centroid in centroid_results:
            all_chunk_ids.extend(self.tree.get_children(centroid.id))
        
        # 4. Поиск по этим чанкам
        chunk_results = self.client.search(
            collection_name="chunks",
            query_vector=query_vector,
            limit=len(all_chunk_ids),
            with_payload=True
        )
        
        # 5. Сортировка по сходству и поиск колена
        similarities = [result.score for result in chunk_results]
        knee_index = find_knee_point(similarities)
        
        # 6. Возврат обрезанного списка
        documents = []
        for result in chunk_results[:knee_index]:
            doc = Document(
                page_content=result.payload["text"],
                metadata={"source": result.payload.get("source", "")}
            )
            documents.append(doc)
        return documents

Если работаете с видео, смотрите локальный RAG для видео — там своя специфика чанкинга.

Ошибки, которые вы совершите (и как их избежать)

• Игнорирование распределения сходств. Перед внедрением постройте гистограмму сходств для 100-200 запросов. Увидите, есть ли вообще ярко выраженное колено в ваших данных.
• Слепая вера в одно колено. На сложных запросах может быть два колена — одно отсекает явный шум, второе — умеренно релевантные чанки. Нужно брать первое (более крутое).
• Забыть про минимальный контекст. Запрос "Кто такой Джон Доу?" может дать один высокорелевантный чанк. Но для вопроса "Опишите процесс разработки ядра Linux" даже после колена нужно 10-15 фрагментов. Добавляйте логику минимального/максимального лимита.
• Использовать евклидово расстояние вместо косинусного. Для текстовых эмбеддингов косинусное сходство работает стабильнее. В Qdrant с 2025 года добавили оптимизированные метрики для sparse эмбеддингов — пробуйте их.

DRAG with KNEE не серебряная пуля. Но это шаг от тупого топ-K к интеллектуальному поиску. В 2026 году такие алгоритмы становятся must-have для продакшена.

Что дальше? Комбинируйте этот подход с графами знаний для юридических документов или с мультимодальными эмбеддингами Gemini 2. Контекст будет точным как лазер.

А если хотите быстро протестировать идею без тонн кода — попробуйте Qdrant Cloud (партнёрская ссылка). Там есть готовые примеры гибридного поиска, можно адаптировать под DRAG. Но учтите, облако — это dependency. Для локального прототипа ставьте Qdrant в Docker и не думайте о лимитах.

Подписаться на канал