Індексація баз знань для RAG: синхронізація Confluence, Notion, SharePoint

Індексація баз знань для 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-фільтр вирішує проблему.