AI-система корпоративного пошуку

AI Enterprise Search: корпоративный интеллектуальный поиск

В компании из 500 человек знания хранятся в 7–12 системах одновременно: Confluence, SharePoint, Jira, корпоративная почта, 1C, внутренние CRM, файловые сервера. Сотрудник тратит в среднем 2,5 часа в день на поиск информации — это не метафора, это данные McKinsey. Ключевая проблема не в том, что данные есть, а в том, что найти их через обычный keyword-поиск невозможно: документ называется «Регламент_v3_final_FINAL2», а запрос — «как оформить командировку».

AI Enterprise Search решает это через семантическое понимание запроса и гибридный поиск по всем источникам одновременно.

Архитектура системы

[Источники данных]
Confluence, SharePoint, Jira,
Email, 1C, CRM, FileServer
         ↓
[Indexing Pipeline]
  Коннекторы → Парсинг → Chunking
  → Embedding (E5-large / BGE-M3)
  → Vector DB (Qdrant) + BM25 index
         ↓
[Search Engine]
  Query Understanding (LLM)
  → Hybrid Search (векторный + keyword)
  → Re-ranking (CrossEncoder)
  → Answer Generation (GPT-4o / Claude)
         ↓
[Интерфейс]
  Web UI, Slack-бот, API

Гибридный поиск: почему только vector недостаточно

Чистый векторный поиск хорошо работает на семантически похожих запросах, но плохо — на точных совпадениях: артикулах, именах, датах. Запрос «договор с ООО "Ромашка" от 12.03.2024» даст лучший результат через BM25, а «порядок действий при инциденте с утечкой данных» — через векторный поиск.

from qdrant_client import QdrantClient
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer, CrossEncoder
import numpy as np

class HybridSearchEngine:
    def __init__(self):
        self.dense_model = SentenceTransformer("intfloat/multilingual-e5-large")
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
        self.qdrant = QdrantClient(url="http://localhost:6333")
        self.bm25_index = None
        self.doc_store = {}

    def search(self, query: str, top_k: int = 20, final_k: int = 5) -> list[dict]:
        # Dense search
        query_emb = self.dense_model.encode(
            f"query: {query}",  # E5-формат
            normalize_embeddings=True
        )
        dense_results = self.qdrant.search(
            collection_name="enterprise_docs",
            query_vector=query_emb.tolist(),
            limit=top_k
        )

        # Sparse search (BM25)
        bm25_scores = self.bm25_index.get_scores(query.lower().split())
        top_bm25_ids = np.argsort(bm25_scores)[-top_k:][::-1]

        # Merge результатов (RRF - Reciprocal Rank Fusion)
        merged = self._reciprocal_rank_fusion(dense_results, top_bm25_ids)

        # Cross-encoder reranking
        pairs = [(query, self.doc_store[doc_id]["text"]) for doc_id in merged[:top_k]]
        rerank_scores = self.reranker.predict(pairs)
        reranked = sorted(
            zip(merged[:top_k], rerank_scores),
            key=lambda x: x[1],
            reverse=True
        )

        return [self.doc_store[doc_id] for doc_id, _ in reranked[:final_k]]

    def _reciprocal_rank_fusion(self, dense: list, sparse: list, k: int = 60) -> list:
        scores = {}
        for rank, item in enumerate(dense):
            doc_id = item.id
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        for rank, doc_id in enumerate(sparse):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)
        return sorted(scores, key=scores.get, reverse=True)

Понимание запроса и генерация ответа

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class EnterpriseSearchAssistant:
    def __init__(self, search_engine: HybridSearchEngine):
        self.search = search_engine
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)

    def answer(self, query: str, user_context: dict) -> dict:
        # Расширение запроса для лучшего поиска
        expanded_query = self._expand_query(query, user_context)

        # Поиск по всем источникам
        results = self.search.search(expanded_query, final_k=7)

        if not results:
            return {"answer": "Документы по вашему запросу не найдены.", "sources": []}

        # Генерация ответа с цитатами
        context = "\n\n".join([
            f"[{i+1}] {r['title']} ({r['source']})\n{r['text'][:500]}"
            for i, r in enumerate(results)
        ])

        prompt = f"""Ответь на вопрос сотрудника на основе документов компании.
Используй только предоставленные документы. Цитируй источники [1], [2] и т.д.
Если ответ неполный — скажи об этом.

Вопрос: {query}
Документы:
{context}"""

        answer = self.llm.invoke(prompt).content
        return {
            "answer": answer,
            "sources": [{"id": i+1, "title": r["title"], "url": r["url"]}
                        for i, r in enumerate(results)]
        }

Индексация источников

Для каждого источника — отдельный коннектор с инкрементальным обновлением:

Кейс из практики: производственный холдинг, 1200 сотрудников. Индексировали 340 000 документов из Confluence, SharePoint и самописной СЭД. Время загрузки индекса — 18 часов (однократно). Регулярное обновление — дельта за 30 минут, ~200 документов, занимает 4 минуты. Средняя точность топ-3 ответов (оценка HR-командой по выборке 500 запросов): 78% релевантных результатов против 41% у прежнего Elasticsearch keyword-поиска.

Access Control: безопасность на уровне поиска

Критическая деталь enterprise-поиска — пользователь должен видеть только те документы, к которым у него есть доступ. Реализуется через metadata-фильтры в Qdrant:

# При поиске фильтруем по правам пользователя
results = qdrant.search(
    collection_name="enterprise_docs",
    query_vector=query_emb,
    query_filter=Filter(
        must=[
            FieldCondition(
                key="allowed_groups",
                match=MatchAny(any=user_groups)
            )
        ]
    ),
    limit=20
)

Сроки

• Пилот (1–2 источника, ~50 000 документов): 4–6 недель
• Полная система с 5–8 источниками и UI: 3–4 месяца
• Настройка ACL и доступов: +2–3 недели