Сокращение расхода токенов в RAG на 67% с помощью мультивекторов Qdrant и изоляции предложений

Проблема токенов стала серьезной заботой для команд, использующих технологии генерации с усилением через поиск информации. Крупные компании ищут способы сократить расходы, и многие команды обнаруживают, что неоптимизированный процесс работы с большими языковыми моделями может обойтись дороже, чем время, которое он должен был сэкономить. Проверка юридических документов является хорошей площадкой для тестирования, поскольку документы длинные, ответы краткие, а объем большой.

Страница, которая никому не была нужна

Представьте адвоката на третьей неделе рассмотрения дела по спору о контракте, который должен просмотреть множество соглашений о предоставлении услуг и корпоративных документов. Вопрос конкретен: до какой суммы может достигать лимит компенсации в случае утверждения третьей стороны о нарушении прав интеллектуальной собственности? Ответ на него содержится в одном предложении где-то в сорокалистовом контракте.

Стандартная система RAG возвращает три лучших фрагмента по 500–1 000 токенов каждый; в основном это изложения, определения, условия оплаты и положения о применимом законодательстве, которые не имеют никакого отношения к претензиям в области интеллектуальной собственности. Единственное полезное предложение находится посередине. Это раздражает читателя, а в масштабах — стоит дорого: при аудите тысяч документов в рамках судебного разбирательства поставщик LLM получает оплату за обработку огромного количества избыточного шаблонного текста.

Возникают два вопроса: насколько большая часть полученного текста действительно необходима? И можно ли выделить именно это предложение без использования дополнительного сервиса для переранжирования?

Ничто в этой проблеме связано непосредственно с законодательством. Любой набор длинных документов в шаблонном формате, таких как страховые полисы, контракты с поставщиками или руководства по соблюдению норм, имеет одинаковую структуру: краткий ответ, окружённый множеством страниц текста, которые при каждом поиске включаются в запрос. Расходы растут пропорционально размеру полученных фрагментов, а не размеру ответа, поэтому важно определить, какая часть каждого фрагмента попадёт к модели.

Почему происходит переранжирование в векторной базе данных

Типичная двухэтапная схема включает быстрый плотный поиск для отбора примерно 100 лучших кандидатов, затем второй сетевой запрос к внешнему инструменту переранжирования, такому как Cohere, SageMaker или пользовательский контейнер, для их сортировки. Этот подход работает, но имеет три недостатка.

• Проводятся два сетевых запроса двум разным поставщикам.
• Если инструмент переранжирования выходит из строя, поиск терпит неудачу.
• Этот проект предусматривает хранение векторов и переупорядочивание запросов.

Qdrant предлагает альтернативу. Начиная с версии 1.10 он поддерживает нативные многовекторные поля с компаратором MAX_SIM, который обеспечивает работу по принципу позднего взаимодействия в стиле ColBERT. Сравнение токенов выполняется непосредственно в базе данных в рамках одного запроса, благодаря чему достигается точность позднего взаимодействия без необходимости дополнительных операций. В сочетании с изоляцией предложений такая конфигурация позволяет сократить затраты на обработку токенов на 67%.

Настройки памяти дополняют этот профиль.

• Бинарная квантизация в плотном индексе. Плотные векторы сокращаются до одного бита на измерение, и эти небольшие векторы остаются в ОЗУ (always_ram=True), что обеспечивает очень быстрое предзагрузчивание кандидатов.
• Матрицы ColBERT на диске. Тяжелые матрицы токенов хранятся на диске (on_disk=True) и загружаются только для набора лучших кандидатов — примерно 50–100 фрагментов — во время пересчёта оценок, что позволяет снизить нагрузку на оперативную память.

Поток обработки

Процесс состоит из шести шагов, включающих один запрос к базе данных и не использующих внешние API для переурочивания кандидатов.

1. Разделите документы на разделы так, чтобы ни один фрагмент не разрезал юридическую клаузулу пополам.
2. Зашифруйте каждый фрагмент дважды: один раз в виде стандартного плотного вектора для быстрой обработки, а второй раз — в виде многовекторной матрицы ColBERT для пересчёта оценок на уровне токенов.
3. Загрузите заранее лучшие N кандидатов с использованием двоично квантизированных плотных векторов.
4. Пересчитайте оценки этих кандидатов с помощью алгоритма ColBERT MAX_SIM внутри Qdrant.
5. Возьмите лучшие фрагменты и оцените их отдельные предложения локально по сравнению с вектором запроса ColBERT.
6. Отправьте в ЯЗЫК-модель только соответствующие предложения.

Цена для эталонного теста исчислялась в размере 3,00 доллара за миллион входных токенов, что соответствует стандартной коммерческой модели среднего уровня.

Схема сбора данных

Приведённый ниже скрипт конфигурации настраивает стандартные плотные эмбеддинги и многовекторные модели ColBERT рядом друг с другом, причём модели FastEmbed загружаются локально на процессоре.

from qdrant_client import QdrantClient, models
# ADDED: Load FastEmbed models locally on CPU
from fastembed import TextEmbedding, LateInteractionTextEmbedding
COLLECTION_NAME = "legal_discovery"
DENSE_DIM = 384  # BAAI/bge-small-en-v1.5
COLBERT_DIM = 128  # colbert-ir/colbertv2.0
# ADDED: Instantiate the vector models
dense_model = TextEmbedding("BAAI/bge-small-en-v1.5")
colbert_model = LateInteractionTextEmbedding("colbert-ir/colbertv2.0")
client = QdrantClient("<http://localhost:6333>")
client.create_collection(
    collection_name=COLLECTION_NAME,
    vectors_config={
        "dense": models.VectorParams(
            size=DENSE_DIM,
            distance=models.Distance.COSINE,
            quantization_config=models.BinaryQuantization(
                binary=models.BinaryQuantizationConfig(always_ram=True),
            ),
        ),
        "colbert": models.VectorParams(
            size=COLBERT_DIM,
            distance=models.Distance.COSINE,
            multivector_config=models.MultiVectorConfig(
                comparator=models.MultiVectorComparator.MAX_SIM
            ),
            on_disk=True,
            hnsw_config=models.HnswConfigDiff(m=0),
        ),
    },
)

Среди параметров векторов ColBERT выделяется одна настройка: конфигурация HNSW с параметром m, установленным в 0. Поскольку ColBERT используется только для пересчёта оценок предварительно отфильтрованного набора данных, для него не требуется граф HNSW. Игнорирование графа позволяет существенно сократить время создания индекса и объём хранилища.

Одна операция для предзагрузки и пересчёта оценок

Универсальный API запросов Qdrant позволяет выполнять предзагрузку на первом этапе и пересчёт оценок на втором этапе в виде одной операции. Не требуется формирование запроса, парсинг ID и повторный обращение к механизму переоценки; весь процесс осуществляется в рамках одной операции обмена данными с базой данных.

# ADDED: Generate query embeddings (ColBERT uses query_embed to add prefix padding)
dense_query = next(dense_model.query_embed(query)).tolist()
colbert_query = next(colbert_model.query_embed(query)).tolist()
# Run the two-stage query in one network round-trip
results = client.query_points(
    collection_name=COLLECTION_NAME,
    prefetch=models.Prefetch(
        query=dense_query,
        using="dense",
        limit=prefetch_limit,
        params=models.SearchParams(
            quantization=models.QuantizationSearchParams(rescore=False),
        ),
    ),
    query=colbert_query,
    using="colbert",
    limit=top_k,
    with_payload=True,
)

Заслуживает внимания флаг rescore=False внутри блока prefetch. При бинарной квантовке Qdrant обычно загружает кандидатов с квантизированными векторами, а затем загружает векторы с полной точностью типа float32 для их повторной оценки. Поскольку этап ColBERT и так выполняет более точную оценку, отключение процедуры повторной оценки с полной точностью позволяет избежать чтения векторов с полной точностью с диска и снижает нагрузку на базу данных.

import re
import numpy as np
# ADDED: Basic sentence splitter regex
SENTENCE_SPLIT = re.compile(r"(?<=[.;])\s+(?=[A-Z])")
def max_sim(query_vecs: np.ndarray, doc_vecs: np.ndarray) -> float:
    # Compute token-to-token similarity matrix
    sims = query_vecs @ doc_vecs.T  # (num_query_tokens, num_doc_tokens)
    # Sum the maximum similarity scores along the document axis
    return float(sims.max(axis=1).sum())
def isolate_sentences(chunk_text: str, query_vecs: np.ndarray, colbert_model, top_n: int = 1):
    # ADDED: Split chunk text into candidate sentences
    sentences = [s.strip() for s in SENTENCE_SPLIT.split(chunk_text) if len(s.strip()) > 15]
    if not sentences:
        return [(chunk_text, 0.0)]
    # Embed each sentence locally using ColBERT
    sentence_vecs = list(colbert_model.embed(sentences))
    scored = [(sentences[i], max_sim(query_vecs, sentence_vecs[i])) for i in range(len(sentences))]
    scored.sort(key=lambda pair: pair[1], reverse=True)
    return scored[:top_n]

def build_optimized_prompt(query: str, chunk_texts: list[str], colbert_model) -> str:
    query_vecs = next(colbert_model.query_embed(query))
    context_parts = []
for i, text in enumerate(chunk_texts):
        top_sentences = isolate_sentences(text, query_vecs, colbert_model, top_n=1)
        isolated_text = " ".join(s for s, _ in top_sentences)
        context_parts.append(f"[Source Chunk {i+1}]: {isolated_text}")
    context_str = "\n\n".join(context_parts)
    return f"Context:\n{context_str}\n\nQuestion: {query}\nAnswer:"