PDIQ: Как мы построили корпоративного ИИ-ассистента на AWS, который знает всё

Проблема: 5000 страниц документации и никто ничего не помнит

PDI Technologies — компания в сфере топливного и коммерческого транспорта. У них есть всё: Confluence с процессами, SharePoint с договорами, внутренние базы знаний, Slack-архивы. И есть проблема: новый сотрудник тратит недели на поиск информации. Старый сотрудник не помнит, как настроить тот редкий кейс, который делал три года назад. Инженеры поддержки листают десятки страниц, чтобы ответить на вопрос клиента.

Типичная enterprise-ситуация: данные разбросаны по 15 источникам, доступ контролируется разными политиками, форматы — от PDF до HTML-страниц Confluence. И все хотят одного: задать вопрос на естественном языке и получить точный ответ с ссылкой на источник.

Решение: не просто RAG, а многоуровневая система знаний

Мы построили PDIQ — внутреннего ассистента, который агрегирует знания из всех источников. Ключевое отличие от типичных RAG-систем: PDIQ понимает контекст сотрудника (отдел, роль, проект) и фильтрует ответы на основе прав доступа. И да, он даёт ответы с цитатами и ссылками на исходные документы.

Архитектурно мы отказались от монолита. Вместо этого создали pipeline из независимых сервисов:

1. Ingestion Pipeline — забирает данные из источников (Confluence API, SharePoint, S3 с PDF)
2. Processing Layer — чистит, разбивает на чанки, генерирует эмбеддинги
3. Vector Store — хранит векторы + метаданные (источник, права доступа, дата обновления)
4. Query Engine — обрабатывает вопросы, ищет релевантные чанки, генерирует ответы
5. Access Control Layer — проверяет права пользователя на каждый чанк

Шаг за шагом: как мы собирали систему на AWS

1 Выбор инструментов: почему не OpenAI и не Pinecone

Enterprise-требования диктуют свои правила: данные не должны уходить за периметр компании, нужен детальный контроль над инфраструктурой, SLA на уровне 99.9%. Поэтому наш стек выглядит так:

Ключевой момент: мы используем Bedrock, а не напрямую Anthropic API. Почему? PrivateLink для доступа к моделям без выхода в интернет, шифрование данных на rest и in transit, интеграция с CloudTrail для аудита всех запросов. В enterprise это не "приятные плюшки", а обязательные требования.

2 Ingestion: как забирать данные из Confluence без боли

Confluence API — это отдельный вид ада. Пагинация, rate limiting, странная структура страниц с макросами. Наш пайплайн обработки выглядит так:

# Упрощенная версия коннектора для Confluence
import boto3
from atlassian import Confluence
from langchain.text_splitter import RecursiveCharacterTextSplitter

def confluence_ingest_handler(event, context):
    # 1. Аутентификация через Secrets Manager
    secret = boto3.client('secretsmanager').get_secret_value(
        SecretId='confluence-credentials'
    )
    creds = json.loads(secret['SecretString'])
    
    # 2. Забираем страницы по пространствам
    confluence = Confluence(
        url=creds['url'],
        username=creds['username'],
        password=creds['api_token'],
        cloud=True
    )
    
    # 3. Рекурсивный обход с кэшированием в S3
    spaces = confluence.get_all_spaces()['results']
    for space in spaces:
        pages = confluence.get_all_pages_from_space(
            space['key'],
            expand='body.storage,version'
        )
        
        # 4. Очистка HTML и извлечение текста
        for page in pages:
            clean_text = html_to_text(page['body']['storage']['value'])
            
            # 5. Умное разделение на чанки с перекрытием
            splitter = RecursiveCharacterTextSplitter(
                chunk_size=1000,
                chunk_overlap=200,
                separators=['\n\n', '\n', '. ', ' ', '']
            )
            chunks = splitter.split_text(clean_text)
            
            # 6. Сохранение в S3 для дальнейшей обработки
            s3_key = f"raw/confluence/{space['key']}/{page['id']}.json"
            s3_client.put_object(
                Bucket='pdiq-data-raw',
                Key=s3_key,
                Body=json.dumps({
                    'chunks': chunks,
                    'metadata': {
                        'source': 'confluence',
                        'page_id': page['id'],
                        'space': space['key'],
                        'url': page['_links']['webui'],
                        'last_modified': page['version']['when'],
                        'access_groups': extract_access_groups(page)
                    }
                })
            )

Важные нюансы:

• Мы кэшируем raw-данные в S3 перед обработкой. Если пайплайн эмбеддингов упадет, не придется заново ходить в Confluence
• Метаданные включают группы доступа из Confluence permissions. Это критично для фильтрации ответов
• Используем recursive splitter, но с кастомными разделителями под структуру Confluence (заголовки, таблицы, код)

3 Гибридный поиск: когда векторов недостаточно

Только векторный поиск дает 60-70% релевантности на enterprise-данных. Почему? Термины из специфичных доменов ("топливный модуль T7-B"), акронимы ("PDQ" может значить и продукт, и процесс), точные совпадения номеров версий.

Наш query pipeline:

1. Query Understanding — анализируем вопрос, извлекаем ключевые сущности (номера продуктов, версии, имена людей)
2. Hybrid Search — параллельно выполняем:
   • Векторный поиск по эмбеддингам вопроса
   • Полнотекстовый поиск по индексу OpenSearch
   • Поиск по метаданным (если есть точные совпадения)
3. Reranking — используем cross-encoder (например, BGE-reranker) для переранжирования результатов
4. Access Filtering — фильтруем чанки, к которым у пользователя нет доступа
5. Context Building — собираем топ-N чанков в контекст для LLM

4 Контроль доступа: самая сложная часть enterprise RAG

В PDI Technologies у инженера из отдела разработки нет доступа к финансовым отчетам. У менеджера по продажам — к исходному коду. Наша система должна это учитывать.

Решение: двухуровневая проверка:

def check_access(user_groups, chunk_metadata):
    """Проверяет доступ пользователя к чанку"""
    # 1. Проверка по группам из метаданных
    chunk_groups = chunk_metadata.get('access_groups', [])
    
    # Если группы не указаны — доступ открыт
    if not chunk_groups:
        return True
    
    # 2. Пересечение групп пользователя и групп чанка
    user_group_set = set(user_groups)
    chunk_group_set = set(chunk_groups)
    
    if user_group_set.intersection(chunk_group_set):
        return True
    
    # 3. Fallback: проверка через LDAP/Active Directory
    return check_ldap_access(user_id, chunk_metadata['source_id'])

Но есть нюанс: что если в ответе нужно использовать чанки из разных источников с разными правами? Мы реализовали "least privilege context" — собираем контекст только из чанков, доступных пользователю. Если доступных чанков недостаточно для ответа, система говорит: "У вас нет доступа к части информации, необходимой для ответа".

Ошибки, которые мы совершили (чтобы вы их не повторили)

Мониторинг и метрики: как понять, что система работает

RAG-система без мониторинга — черный ящик. Мы отслеживаем:

• Recall@K — сколько релевантных документов в топ-K результатах
• Ответы без источников — если LLM генерирует ответ "из головы", а не из контекста
• Latency — 95-й и 99-й перцентили времени ответа
• Cache hit rate — эффективность кэширования эмбеддингов
• User feedback — кнопки "thumbs up/down" после каждого ответа

Самый важный инсайт: 40% вопросов повторяются. "Как настроить интеграцию с SAP?", "Где найти шаблон отчета?", "Какая процедура отпуска?" Кэширование ответов на уровне эмбеддингов и готовых ответов снизило нагрузку на Bedrock на 65%.

Что дальше? Agentic RAG и автономные ассистенты

Текущая система PDIQ отвечает на вопросы. Следующий шаг — выполнение действий. Не "как создать Jira-тикет", а "создай Jira-тикет на баг в интеграции с SAP с вот таким описанием".

Мы экспериментируем с мультиагентными подходами, где разные агенты специализируются на разных задачах: один ищет информацию, другой анализирует, третий выполняет действия через API.

Ключевой вызов здесь — безопасность. Агент с доступом к Jira API может создать сотни тикетов. Нужны guardrails, лимиты на действия, human-in-the-loop для критичных операций.

Главный урок из проекта PDIQ: успешная enterprise RAG-система — это на 30% машинное обучение и на 70% инженерия. Инфраструктура, мониторинг, безопасность, контроль доступа. Модели меняются каждый год, а требования к надежности и безопасности остаются.

И последнее: начинайте с пилота. Не пытайтесь сразу индексировать все 5000 страниц Confluence. Возьмите один отдел, один источник данных, сделайте работающий прототип за 2-4 недели. Получите feedback. Итеративно улучшайте. Так вы избежите ситуации, когда через полгода разработки понимаете, что система не решает реальные проблемы сотрудников.