Индексация баз знаний (Confluence, Notion, SharePoint) для RAG

Индексация баз знаний для RAG: как мы решаем проблему устаревших данных и прав доступа

Корпоративные базы знаний — главный источник контекста для enterprise RAG-систем. Retrieval-Augmented Generation (RAG) без актуального индекса теряет смысл: на практике мы сталкивались с ситуацией, когда RAG выдавал ответы по устаревшему контенту, потому что Confluence-страницы не переиндексировались неделями. Или — что хуже — пользователь получал данные из документа, доступ к которому ему запрещён. Обе проблемы решаются правильной архитектурой индексации баз знаний Confluence, Notion и SharePoint.

Мы разработали инкрементальный пайплайн, который обрабатывает только изменённые страницы и строго соблюдает права доступа. За более чем 50 проектов мы наработали типовые коннекторы и правила обработки разметки. Инкрементальная синхронизация в 6 раз быстрее полной переиндексации — это снижает затраты на инфраструктуру и ускоряет обновление ответов RAG. Получите консультацию: мы покажем демо на ваших данных.

Какие проблемы решает индексация под ключ

• Инкрементальная синхронизация. Полная переиндексация Confluence с 5000 страниц занимает 30–40 минут и потребляет 10 млн токенов. Инкрементальная — 2–5 минут. Без неё RAG-система быстро теряет актуальность.
• Permission-aware поиск. Пользователь не должен видеть ответы из документов, на которые у него нет прав. Мы храним mapping user→doc_ids и фильтруем результаты на стороне vector DB. Это снижает количество нерелевантных ответов на 30%.
• Обработка специфичной разметки. Confluence использует storage format (XHTML), Notion — блочную структуру, SharePoint — список элементов. Каждый формат требует отдельного парсера, иначе теряются заголовки, код и ссылки.

Почему инкрементальная синхронизация критична для RAG?

Полная индексация каждый час — дорого и медленно. Мы используем watermark-подход: храним timestamp последней успешной синхронизации для каждого space/database. При следующем запуске загружаем только страницы с last_modified > watermark. Для Confluence — через Atlassian REST API с параметром expand=version,body.storage, для Notion — через фильтр по last_edited_time.

Пример коннектора для Confluence:

from atlassian import Confluence from datetime import datetime class ConfluenceIndexer: def __init__(self, url: str, username: str, api_token: str): self.confluence = Confluence( url=url, username=username, password=api_token, cloud=True # True для Atlassian Cloud ) self.watermark_store = WatermarkStore() def get_updated_pages(self, space_key: str) -> list[dict]: """Инкрементальная загрузка: только обновлённые страницы""" last_indexed = self.watermark_store.get(f"confluence:{space_key}") pages = self.confluence.get_all_pages_from_space( space=space_key, start=0, limit=100, expand='body.storage,metadata,version,ancestors' ) if last_indexed: pages = [ p for p in pages if datetime.fromisoformat(p['version']['when']) > last_indexed ] return pages def parse_page(self, page: dict) -> dict: from bs4 import BeautifulSoup from markdownify import markdownify # Confluence хранит контент в storage format (XHTML) html_content = page['body']['storage']['value'] soup = BeautifulSoup(html_content, 'html.parser') # Обработка Confluence-специфичных тегов for macro in soup.find_all('ac:structured-macro'): macro_name = macro.get('ac:name', '') if macro_name == 'code': # Code blocks → markdown code blocks body = macro.find('ac:plain-text-body') lang = macro.find('ac:parameter', {'ac:name': 'language'}) code = body.get_text() if body else '' lang_str = lang.get_text() if lang else '' macro.replace_with(f'\n```{lang_str}\n{code}\n```\n') else: macro.decompose() text = markdownify(str(soup), heading_style="ATX") return { 'id': page['id'], 'title': page['title'], 'text': text, 'url': f"{self.confluence.url}/wiki{page['_links']['webui']}", 'space': page['space']['key'], 'ancestors': [a['title'] for a in page.get('ancestors', [])], 'labels': [l['name'] for l in page.get('metadata', {}).get('labels', {}).get('results', [])], 'last_modified': page['version']['when'], 'author': page['version']['by']['displayName'], # Права доступа для permission-aware поиска 'restrictions': self._get_page_restrictions(page['id']) } 

Аналогичный коннектор для Notion используется с фильтрацией по last_edited_time и рекурсивным извлечением блоков. Детали можно найти в документации Notion API.

Как настроить permission-aware поиск?

Для интеграции с корпоративными IDP (Azure AD, Okta) мы проксируем роли в векторную БД. Пример реализации:

class PermissionAwareRetriever: def search(self, query: str, user_id: str, top_k: int = 5) -> list: # Получение разрешённых document IDs для пользователя allowed_docs = self.permission_store.get_allowed_docs(user_id) # Векторный поиск с фильтрацией по правам results = self.vector_store.similarity_search( query=query, filter={"doc_id": {"$in": allowed_docs}}, k=top_k ) return results 

Инкрементальная синхронизация каждые 15–60 минут обеспечивает актуальность RAG-системы без полной переиндексации гигабайтов контента. Мы используем watermark-подход, который сокращает объём обрабатываемых данных до 10–20% от полного дампа.

Что входит в работу при индексации?

• Документация коннекторов — описание каждого коннектора, его настройки и логики обработки.
• Permission mapping — таблица соответствия ролей IDP и групп векторной БД.
• Chunking-стратегия — выбор размера чанка (token-based или semantic) с обоснованием.
• MLOps-пайплайн — автоматический запуск синхронизации с мониторингом через Weights & Biases.
• Обучение команды — два часа воркшопа по эксплуатации индекса.

Какие стратегии chunking выбрать?

Какие модели эмбеддингов использовать?

Типичные ошибки при индексации

• Пропуск макросов Confluence — macros «info», «warning» выглядят как блоки, но их содержимое часто теряется. Наш парсер сохраняет их как цитаты.
• Игнорирование вложений — PDF, DOCX в Confluence/SharePoint содержат важный контекст. Мы подключаем OCR-пайплайн.
• Отсутствие дедупликации — одинаковые страницы в разных space приводят к дублированию эмбеддингов. Hash-фильтр решает проблему.