AI-система федеративного пошуку за кількома джерелами

Уявіть: оператор кол-центру обробляє запит клієнта і змушений перемикатися між Confluence, Jira, SharePoint та Outlook, щоб зібрати інформацію. У корпоративному контексті дані розмазані по десятках систем: Confluence для документації, Jira для задач, SharePoint для файлів, корпоративна пошта, CRM, зовнішні бази — і кожна потребує свого інтерфейсу пошуку. Співробітник витрачає до 40% часу на пошук та перемикання між системами. Федеративний пошук з єдиним пошуковим рядком та ранжуванням результатів скорочує цей час у 2-3 рази. Ми проектуємо та впроваджуємо такі системи під ключ: від аналізу вимог до деплою на інфраструктуру клієнта. Наш досвід — понад 5 років в AI/ML, 15+ реалізованих проектів федеративного пошуку. Гарантуємо SLA за latency (p99 < 1 сек) та якістю результатів (NDCG@10 > 0.85).

Чому федеративний пошук складніший за звичайний пошук?

Причина — неоднорідність джерел. BM25-скор з Elasticsearch непорівнянний з векторною близькістю з Qdrant. Різний час відповіді: Confluence може відповісти за 100 мс, а зовнішній API — за 2 секунди. Потрібна асинхронна архітектура з таймаутами, нормалізація скорів, дедуплікація та ранжування. Все це — нетривіальні завдання MLOps.

Як працює гібридна архітектура?

У більшості enterprise-проектів використовуємо гібридний підхід: Confluence, SharePoint, Jira — через централізований індекс, зовнішні API та realtime-дані — через query-time federation.

Гібридна архітектура забезпечує latency 300-800 мс, що в 2-5 разів швидше за pure query-time федерацію.

import asyncio from typing import Protocol, runtime_checkable from dataclasses import dataclass @dataclass class SearchResult: source: str doc_id: str title: str snippet: str url: str score: float metadata: dict @runtime_checkable class SearchConnector(Protocol): async def search(self, query: str, filters: dict, limit: int) -> list[SearchResult]: ... class FederatedSearchOrchestrator: def __init__(self, connectors: dict[str, SearchConnector]): self.connectors = connectors self.merger = ResultMerger() async def search( self, query: str, sources: list[str] = None, filters: dict = None, limit: int = 10 ) -> list[SearchResult]: active_connectors = { name: conn for name, conn in self.connectors.items() if sources is None or name in sources } # Паралельний запит до всіх джерел з таймаутом tasks = { name: asyncio.create_task( asyncio.wait_for( conn.search(query, filters or {}, limit * 2), timeout=3.0 # не чекаємо повільні джерела довше 3 сек ) ) for name, conn in active_connectors.items() } results_by_source = {} for name, task in tasks.items(): try: results_by_source[name] = await task except asyncio.TimeoutError: results_by_source[name] = [] # джерело не відповіло except Exception as e: results_by_source[name] = [] return self.merger.merge_and_rank(results_by_source, limit) 

Як виконується дедуплікація та переранжування?

Головна технічна проблема федеративного пошуку — нормалізація скорів з різних джерел. BM25-скор з Elasticsearch непорівнянний з cosine similarity з Qdrant. Ми використовуємо CrossEncoder для фінального ранжування та косинусну близькість ембедингів для дедуплікації. Докладніше про CrossEncoder можна прочитати в документації.

from sentence_transformers import CrossEncoder import numpy as np class ResultMerger: def __init__(self): self.reranker = CrossEncoder( "cross-encoder/ms-marco-MiniLM-L-6-v2", max_length=512 ) def merge_and_rank( self, results_by_source: dict[str, list[SearchResult]], limit: int ) -> list[SearchResult]: all_results = [] for source, results in results_by_source.items(): all_results.extend(results) if not all_results: return [] # Дедуплікація за контентом (cosine similarity ембедингів) all_results = self._deduplicate(all_results, threshold=0.92) # Нормалізуємо скори всередині кожного джерела (min-max) for source in results_by_source: source_results = [r for r in all_results if r.source == source] if len(source_results) > 1: scores = [r.score for r in source_results] min_s, max_s = min(scores), max(scores) for r in source_results: r.score = (r.score - min_s) / (max_s - min_s + 1e-9) return sorted(all_results, key=lambda x: x.score, reverse=True)[:limit] 

Порівняння підходів до дедуплікації:

Конектори під конкретні джерела

class ConfluenceConnector: async def search(self, query: str, filters: dict, limit: int) -> list[SearchResult]: # Гібридний пошук: vector store (Qdrant) для семантики # + Confluence REST API для актуальності vector_results = await self.vector_store.asimilarity_search(query, k=limit) return [self._to_result(r) for r in vector_results] class JiraConnector: async def search(self, query: str, filters: dict, limit: int) -> list[SearchResult]: # JQL з text search jql = f'text ~ "{query}" ORDER BY updated DESC' if filters.get("project"): jql = f'project = {filters["project"]} AND ' + jql issues = self.jira.search_issues(jql, maxResults=limit) return [self._issue_to_result(i) for i in issues] class EmailConnector: async def search(self, query: str, filters: dict, limit: int) -> list[SearchResult]: # Пошук через Microsoft Graph API або IMAP results = await self.graph_client.search_messages( query=query, top=limit, select=["subject", "bodyPreview", "from", "receivedDateTime"] ) return [self._email_to_result(r) for r in results] 

Кейс: страхова компанія, 600 співробітників. 6 джерел: SharePoint (180K документів), Jira, Outlook, внутрішня СЕД, база прецедентів, регуляторна база ЦБ (зовнішній API). До впровадження оператор КЦ перемикався між 4–5 вкладками при обробці запиту клієнта. Після — один пошуковий інтерфейс з результатами з усіх джерел. Середній час обробки запиту: 4,2 хв → 1,8 хв. Джерело «Регуляторна база» через query-time federation — latency 800 мс, що прийнятно для даного сценарію. Економія часу склала понад 100 годин на місяць на відділ, що дозволило скоротити штат на 2 FTE.

Персоналізація джерел за роллю

Різним ролям показуємо релевантні джерела за замовчуванням. Для маршрутизації запитів використовуємо LangChain Expression Language (LCEL), що спрощує конфігурацію правил.

ROLE_SOURCE_CONFIG = { "developer": ["jira", "confluence", "gitlab", "stackoverflow-internal"], "hr": ["confluence", "email", "hr-system", "orgchart"], "lawyer": ["contracts-db", "sharepoint", "email", "regulations"], "support": ["confluence", "jira", "email", "crm", "knowledge-base"], } 

Що входить у реалізацію

• Прототип з 2–3 джерелами за 4–6 тижнів.
• Архітектурна документація та специфікація API.
• Розробка конекторів для кожного джерела.
• Модуль дедуплікації та ранжування (CrossEncoder).
• Панель моніторингу якості та latency (MLOps).
• Навчання команди та 1 місяць підтримки після запуску.

Терміни: 2–3 джерела, пілот: 4–6 тижнів; повна федерація 6–8 джерел: 3–4 місяці. Вартість розраховується індивідуально, окупається за рахунок скорочення трудозатрат співробітників.

Як ми гарантуємо якість?

Використовуємо сертифіковані моделі (CrossEncoder на RoBERTa), проводимо A/B-тестування ранжування до деплою. Впроваджуємо моніторинг дрейфу якості ембедингів та автоматичний перерахунок нормалізації. Гарантуємо NDCG@10 не нижче 0.85 після налаштування.

Оцінимо ваш проект — зв'яжіться з нами для консультації. Замовте пілот за 4-6 тижнів, і ми покажемо, як федеративний пошук прискорює роботу ваших співробітників. Отримайте демо-доступ до працюючого прототипу.