Реализация Multi-Query RAG для повышения качества извлечения

Реализация Multi-Query RAG для повышения качества извлечения

Multi-Query RAG — техника улучшения retrieval, при которой исходный запрос автоматически перефразируется несколькими способами, каждый вариант запускается в поиске, а результаты объединяются. Это снижает зависимость качества ответа от конкретной формулировки запроса и повышает полноту извлечения.

Проблема, которую решает Multi-Query

Векторные embedding-модели чувствительны к формулировке запроса. Один и тот же вопрос, заданный по-разному, может давать разные top-K результаты:

• «Как оформить отпуск?» → находит статьи про заявления
• «Процедура получения ежегодного отпуска» → находит раздел регламента
• «Правила предоставления отпускных дней» → находит политику HR

Multi-Query объединяет все три и получает более полный контекст.

Реализация с LangChain

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Qdrant.from_existing_collection(
    embeddings=embeddings,
    collection_name="knowledge_base",
    url="http://localhost:6333",
)

retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm,
    include_original=True,  # Включает оригинальный запрос
)

# Использование
docs = retriever.invoke("каков порядок согласования крупной сделки")
# Внутри LangChain генерирует 3 перефразирования + оригинал,
# ищет по каждому и дедуплицирует результаты

Кастомный Multi-Query с контролем промпта

Стандартный промпт LangChain можно заменить специализированным:

from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import BaseOutputParser

class LineListOutputParser(BaseOutputParser):
    """Парсит список вопросов из ответа LLM"""
    def parse(self, text: str) -> list[str]:
        lines = text.strip().split("\n")
        return [line.strip().lstrip("123456789.-) ") for line in lines if line.strip()]

MULTI_QUERY_PROMPT = PromptTemplate(
    input_variables=["question"],
    template="""Ты — AI-ассистент по поиску документов. Твоя задача — сгенерировать
5 различных вариантов следующего вопроса для улучшения поиска в векторной базе.

Правила:
- Используй синонимы и альтернативные формулировки
- Один вариант — более конкретный, один — более общий
- Сохраняй смысл оригинального вопроса
- Каждый вопрос с новой строки, без нумерации

Оригинальный вопрос: {question}

Варианты:"""
)

custom_retriever = MultiQueryRetriever(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    llm_chain=MULTI_QUERY_PROMPT | llm | LineListOutputParser(),
    include_original=True,
)

Parallel Multi-Query с дедупликацией

Для уменьшения latency запускаем поиск по всем вариантам параллельно:

import asyncio
from openai import AsyncOpenAI

async def multi_query_search(
    original_query: str,
    vectorstore,
    n_variants: int = 4,
    top_k_per_query: int = 5,
) -> list[str]:
    """Параллельный multi-query retrieval"""

    async_client = AsyncOpenAI()

    # Генерируем варианты запроса
    response = await async_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Сгенерируй {n_variants} перефразирования вопроса:\n{original_query}\nОдин вопрос на строку."
        }],
        temperature=0.5,
    )
    variants = response.choices[0].message.content.strip().split("\n")
    all_queries = [original_query] + variants[:n_variants]

    # Параллельный поиск
    search_tasks = [
        asyncio.to_thread(vectorstore.similarity_search, q, k=top_k_per_query)
        for q in all_queries
    ]
    results_per_query = await asyncio.gather(*search_tasks)

    # Дедупликация по content
    seen_texts = set()
    unique_docs = []
    for docs in results_per_query:
        for doc in docs:
            text_hash = hash(doc.page_content[:100])
            if text_hash not in seen_texts:
                seen_texts.add(text_hash)
                unique_docs.append(doc)

    return unique_docs

Практический кейс: влияние на recall

Датасет: корпоративная база знаний юридической компании (3200 документов).

Тестовый набор: 200 запросов, для каждого размечены все релевантные документы.

Multi-query поднимает recall с 0.61 до 0.84 (+38%) при умеренном росте latency (×2.4). После reranker precision также восстанавливается до 0.81.

Когда Multi-Query не нужен

• Очень высокий latency requirement (<200мс)
• Запросы уже хорошо структурированы пользователями
• Датасет небольшой (<5000 документов) — single query уже даёт высокий recall

Сроки

• Реализация Multi-Query Retriever: 2–3 дня
• Подбор промпта и числа вариантов: 2–3 дня
• Тестирование на датасете: 2–3 дня
• Итого: 1 неделя