Реализация Multi-Index RAG (объединение нескольких источников)

Реализация Multi-Index RAG (объединение нескольких источников)

Multi-Index RAG — архитектура, при которой поиск выполняется по нескольким раздельным индексам (векторным хранилищам или коллекциям), а результаты объединяются в единый контекст для LLM. Это необходимо при работе с разнородными источниками данных, требующими разных стратегий индексации, или при изоляции данных по доменам.

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

Разные типы данных: структурированные FAQ (короткие ответы) и длинные регламенты требуют разных размеров чанков и стратегий retrieval.

Разные домены: юридическая документация, техническая документация, продуктовые описания — семантические пространства слабо пересекаются, раздельные индексы дают более точный retrieval.

Разные источники: Confluence, SharePoint, Notion, GitHub — каждый требует своего парсера и имеет специфические метаданные.

Изоляция по безопасности: данные разных отделов хранятся в разных индексах с контролем доступа.

Архитектура Multi-Index RAG

from typing import Optional
from langchain_community.vectorstores import Qdrant
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.schema import Document
import asyncio

class MultiIndexRAG:
    def __init__(self, embeddings, llm):
        self.embeddings = embeddings
        self.llm = llm
        self.indexes: dict[str, Qdrant] = {}
        self.router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

    def add_index(self, name: str, collection: str, description: str):
        """Регистрируем индекс с описанием для роутера"""
        self.indexes[name] = {
            "retriever": Qdrant.from_existing_collection(
                embeddings=self.embeddings,
                collection_name=collection,
                url="http://localhost:6333",
            ).as_retriever(search_kwargs={"k": 5}),
            "description": description,
        }

    def route_query(self, query: str) -> list[str]:
        """LLM-роутер определяет релевантные индексы"""
        index_descriptions = "\n".join([
            f"- {name}: {info['description']}"
            for name, info in self.indexes.items()
        ])

        response = self.router_llm.invoke(f"""
Определи, в каких из следующих индексов нужно искать для ответа на запрос.
Верни JSON-список имён индексов.

Доступные индексы:
{index_descriptions}

Запрос: {query}

Ответ (JSON список):""")

        import json
        try:
            return json.loads(response.content)
        except:
            return list(self.indexes.keys())  # Fallback: все индексы

    async def _search_index(self, index_name: str, query: str) -> tuple[str, list]:
        """Асинхронный поиск в одном индексе"""
        retriever = self.indexes[index_name]["retriever"]
        docs = await asyncio.to_thread(retriever.invoke, query)
        return index_name, docs

    async def retrieve(self, query: str) -> dict[str, list]:
        """Параллельный поиск по релевантным индексам"""
        relevant_indexes = self.route_query(query)

        tasks = [
            self._search_index(idx, query)
            for idx in relevant_indexes
            if idx in self.indexes
        ]

        results = await asyncio.gather(*tasks)
        return dict(results)

    def build_context(self, search_results: dict[str, list]) -> str:
        """Собираем контекст из нескольких индексов"""
        context_parts = []
        for index_name, docs in search_results.items():
            if docs:
                context_parts.append(f"## Источник: {index_name}\n")
                for doc in docs:
                    context_parts.append(f"- {doc.page_content}\n")

        return "\n".join(context_parts)

Настройка индексов для корпоративной базы знаний

rag = MultiIndexRAG(
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
)

rag.add_index(
    name="legal",
    collection="legal_contracts",
    description="Договоры, соглашения, юридические заключения",
)
rag.add_index(
    name="hr",
    collection="hr_policies",
    description="Политики HR: отпуска, командировки, найм, увольнение",
)
rag.add_index(
    name="it",
    collection="it_procedures",
    description="IT-процедуры: доступы, оборудование, информационная безопасность",
)
rag.add_index(
    name="finance",
    collection="finance_regulations",
    description="Финансовые регламенты: бюджет, закупки, авансовые отчёты",
)
rag.add_index(
    name="faq",
    collection="general_faq",
    description="Общие часто задаваемые вопросы сотрудников",
)

Reranking объединённых результатов

После сбора результатов из нескольких индексов важно перемешать и переранжировать:

from flashrank import Ranker, RerankRequest

ranker = Ranker(model_name="ms-marco-MiniLM-L-12-v2")

def rerank_multi_index_results(
    query: str,
    search_results: dict[str, list[Document]],
    top_n: int = 6,
) -> list[Document]:
    """Объединяет и переранжирует результаты из разных индексов"""

    # Собираем все документы
    all_docs = []
    for docs in search_results.values():
        all_docs.extend(docs)

    if not all_docs:
        return []

    # Reranking
    passages = [{"id": i, "text": doc.page_content} for i, doc in enumerate(all_docs)]
    rerank_req = RerankRequest(query=query, passages=passages)
    ranked = ranker.rerank(rerank_req)

    return [all_docs[r["id"]] for r in ranked[:top_n]]

Практический кейс: корпоративный ассистент из 5 источников

Источники: Confluence (5200 страниц), SharePoint (3800 документов), JIRA (экспорт задач), GitHub (wiki, README), внутренняя CRM-документация.

Проблема монолитного индекса: разные типы контента имеют разные оптимальные размеры чанков. README из GitHub оптимально индексировать по-функционально (блоки кода + описание), Confluence-страницы — по разделам, CRM-документацию — по ответам.

Конфигурация Multi-Index:

• 5 раздельных коллекций в Qdrant
• LLM-роутер на GPT-4o-mini (~15мс overhead)
• Параллельный поиск (async) сокращает latency с 5×T до 1.2×T

Результаты:

• Context Recall: 0.71 (монолитный) → 0.88 (multi-index)
• Precision@5: 0.74 → 0.86
• Latency P95: 1.2с → 1.5с (параллельный поиск vs последовательный +250мс)
• Routing accuracy (правильный набор индексов): 91%

Failure cases: 9% запросов попадают в неправильный набор индексов — преимущественно кросс-доменные вопросы. Решение: при низком router confidence запускать поиск по всем индексам с threshold отсечки по score.

Федеративный поиск с access control

def retrieve_with_permissions(
    query: str,
    user_id: str,
    permission_service,
) -> dict[str, list]:
    """Поиск только по разрешённым для пользователя индексам"""
    allowed_indexes = permission_service.get_allowed_indexes(user_id)
    relevant_indexes = [
        idx for idx in route_query(query)
        if idx in allowed_indexes
    ]
    return {idx: search(idx, query) for idx in relevant_indexes}

Сроки

• Проектирование Multi-Index архитектуры: 1 неделя
• Разработка ingestion пайплайнов (5 источников): 3–4 недели
• LLM-роутер и интеграция: 1 неделя
• Reranking и оценка: 1 неделя
• Итого: 6–8 недель