Реалізація 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.2s → 1.5s (паралельний vs послідовний +250мс)
• Routing accuracy (правильний набір індексів): 91%

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

Федеративний пошук з контролем доступу

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 тижнів