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

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

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

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 интеграция

from notion_client import Client

class NotionIndexer:
    def __init__(self, token: str):
        self.notion = Client(auth=token)

    def get_database_pages(self, database_id: str,
                            last_sync: datetime = None) -> list:
        filter_params = {}
        if last_sync:
            filter_params = {
                "filter": {
                    "timestamp": "last_edited_time",
                    "last_edited_time": {"after": last_sync.isoformat()}
                }
            }

        results = self.notion.databases.query(
            database_id=database_id,
            **filter_params
        )
        return results.get('results', [])

    def extract_page_content(self, page_id: str) -> str:
        """Рекурсивное извлечение блоков страницы"""
        blocks = self.notion.blocks.children.list(block_id=page_id)
        return self._blocks_to_text(blocks.get('results', []))

    def _blocks_to_text(self, blocks: list) -> str:
        text_parts = []
        for block in blocks:
            block_type = block['type']
            if block_type in ['paragraph', 'heading_1', 'heading_2', 'heading_3']:
                rich_text = block[block_type].get('rich_text', [])
                text = ''.join([rt['plain_text'] for rt in rich_text])
                if block_type.startswith('heading'):
                    level = int(block_type[-1])
                    text = '#' * level + ' ' + text
                text_parts.append(text)
            elif block_type == 'bulleted_list_item':
                rich_text = block[block_type].get('rich_text', [])
                text = '- ' + ''.join([rt['plain_text'] for rt in rich_text])
                text_parts.append(text)
            elif block_type == 'code':
                lang = block[block_type].get('language', '')
                code = ''.join([rt['plain_text'] for rt in block[block_type]['rich_text']])
                text_parts.append(f'```{lang}\n{code}\n```')

        return '\n\n'.join(text_parts)

Permission-Aware поиск

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-системы без полной переиндексации гигабайтов контента.