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 недель, и мы покажем, как федеративный поиск ускоряет работу ваших сотрудников. Получите демо-доступ к работающему прототипу.