Notion + RAG: Ihr Unternehmens-Wiki verbinden

Notion hat sich für tausende Unternehmen als Referenz-Wiki etabliert. Seine Flexibilität, die intuitive Oberfläche und die Kollaborationsfunktionen machen es zu einem unverzichtbaren Werkzeug, um das Wissen eines Teams zu zentralisieren. Doch mit wachsendem Workspace entsteht ein Problem: Informationen wiederzufinden wird zum Albtraum. Bei Hunderten von Seiten, Unterseiten und Datenbanken verbringen selbst erfahrene Nutzer wertvolle Minuten damit, etwas zu suchen, von dem sie wissen, dass es irgendwo existiert.

Ein mit Notion verbundener RAG-Chatbot verwandelt diese Dokumentenmasse in einen intelligenten Assistenten. Statt zu navigieren, stellen Sie eine Frage in natürlicher Sprache und erhalten eine zusammengefasste, mit Quellen belegte und kontextualisierte Antwort. Dieser Leitfaden zeigt Ihnen Schritt für Schritt, wie Sie diese Integration umsetzen.

Warum Notion mit RAG verbinden?

Die Grenzen der nativen Notion-Suche

Die integrierte Suche von Notion ist zwar nützlich, weist jedoch für große Organisationen deutliche Einschränkungen auf:

Was RAG bringt

Der RAG-Ansatz (Retrieval-Augmented Generation) löst diese Einschränkungen, indem er semantische Suche mit Sprachgenerierung kombiniert:

• Semantische Suche: Findet Informationen auch bei abweichender Formulierung. „Wie beantrage ich Überstundenausgleich" findet auch „Verfahren zur Beantragung freier Tage"
• Intelligente Zusammenfassung: Antwortet direkt, ohne durch 5 Seiten navigieren zu müssen
• Multi-Page-Aggregation: Kombiniert Informationen aus mehreren Quellen zu einer vollständigen Antwort
• Konversationelles Gedächtnis: Jede Frage profitiert vom Kontext vorheriger Austausche
• Zitierte Quellen: Jede Aussage verweist auf die Ursprungsseite

Architektur Notion + RAG

Die Integration folgt einer dreischichtigen Architektur, die Extraktion, Indexierung und Abfrage trennt:

┌─────────────────────────────────────────────────────────────────────────┐
│                        Architecture Notion + RAG                         │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   EXTRAKTION                   INDEXIERUNG                ABFRAGE       │
│   ┌──────────────┐            ┌──────────────┐          ┌────────────┐ │
│   │  Notion API  │───────────▶│   Chunking   │─────────▶│   Qdrant   │ │
│   │              │            │              │          │            │ │
│   │  - Seiten    │            │  - Abschnitte│          │  Vektoren  │ │
│   │  - DBs       │            │  - 500 Token │          │            │ │
│   │  - Blöcke    │            │  - Overlap   │          └──────┬─────┘ │
│   └──────────────┘            └──────────────┘                 │       │
│                                      │                         │       │
│                               ┌──────┴──────┐                  │       │
│                               │  Embeddings │                  │       │
│                               │   BGE-M3    │                  │       │
│                               └─────────────┘                  │       │
│                                                                │       │
│   CHATBOT                                                      │       │
│   ┌──────────────┐     ┌─────────────┐     ┌──────────────┐   │       │
│   │    Frage     │────▶│  Retrieval  │◀────│   Reranker   │◀──┘       │
│   │   Nutzer     │     │   Top-20    │     │   Top-5      │           │
│   └──────────────┘     └─────────────┘     └──────┬───────┘           │
│                                                    │                   │
│                                             ┌──────┴──────┐            │
│                                             │     LLM     │            │
│                                             │   Antwort   │            │
│                                             └─────────────┘            │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Kernkomponenten

1. Extraktion: Der Notion-Connector nutzt die offizielle API, um Seiten und Datenbanken abzurufen
2. Chunking: Lange Dokumente werden in Segmente von 500 Token mit Überlappung zerlegt
3. Embeddings: Jeder Chunk wird in einen semantischen Vektor umgewandelt (BGE-M3 für Mehrsprachigkeit)
4. Vektordatenbank: Qdrant speichert und indexiert die Vektoren für schnelle Suche
5. Reranking: Ein zweites Modell ordnet die Ergebnisse nach Relevanz neu
6. Generierung: Das LLM erstellt eine Antwort aus den relevanten Chunks

Vollständiger Notion-Connector

Hier eine Referenzimplementierung zum Extrahieren von Notion-Inhalten:

DEVELOPERpython
from notion_client import Client
from datetime import datetime
import hashlib

class NotionConnector:
    def __init__(self, token: str):
        """Initialisiert den Connector mit dem Integrations-Token."""
        self.client = Client(auth=token)
        self.processed_ids = set()

    def get_all_pages(self, filter_by_parent: str = None) -> list:
        """
        Ruft alle für die Integration zugänglichen Seiten ab.

        Args:
            filter_by_parent: ID der übergeordneten Seite zum Filtern (optional)

        Returns:
            Liste von für RAG formatierten Dokumenten
        """
        pages = []
        has_more = True
        cursor = None

        while has_more:
            results = self.client.search(
                filter={"property": "object", "value": "page"},
                start_cursor=cursor,
                page_size=100
            )

            for page in results['results']:
                # Duplikate vermeiden
                if page['id'] in self.processed_ids:
                    continue

                # Nach Parent filtern, falls angegeben
                if filter_by_parent:
                    parent = page.get('parent', {})
                    if parent.get('page_id') != filter_by_parent:
                        continue

                doc = self._format_page(page)
                if doc and len(doc['content']) > 50:  # Leere Seiten ignorieren
                    pages.append(doc)
                    self.processed_ids.add(page['id'])

            has_more = results['has_more']
            cursor = results.get('next_cursor')

        return pages

    def _format_page(self, page: dict) -> dict:
        """Formatiert eine Notion-Seite als RAG-Dokument."""
        title = self._extract_title(page)
        content = self._extract_content(page['id'])

        # Hash generieren, um Änderungen zu erkennen
        content_hash = hashlib.md5(content.encode()).hexdigest()

        return {
            "id": f"notion_{page['id']}",
            "title": title,
            "content": f"# {title}\n\n{content}",
            "metadata": {
                "source": "notion",
                "source_type": "wiki",
                "page_id": page['id'],
                "url": page.get('url', ''),
                "last_edited": page['last_edited_time'],
                "created_time": page['created_time'],
                "content_hash": content_hash,
                "parent_type": page.get('parent', {}).get('type'),
                "icon": self._extract_icon(page)
            }
        }

    def _extract_title(self, page: dict) -> str:
        """Extrahiert den Titel einer Seite."""
        props = page.get('properties', {})

        # In den Eigenschaften 'title' oder 'Name' suchen
        for key in ['title', 'Title', 'Name', 'name']:
            if key in props and props[key].get('title'):
                title_parts = props[key]['title']
                return ''.join([t['plain_text'] for t in title_parts])

        return "Ohne Titel"

    def _extract_content(self, page_id: str) -> str:
        """Extrahiert den vollständigen Textinhalt einer Seite."""
        content_parts = []

        def process_blocks(block_id: str, depth: int = 0):
            """Rekursiv zur Verarbeitung verschachtelter Blöcke."""
            if depth > 5:  # Begrenzung der Tiefe
                return

            blocks = self.client.blocks.children.list(block_id=block_id)

            for block in blocks['results']:
                text = self._block_to_text(block, depth)
                if text:
                    content_parts.append(text)

                # Kinder verarbeiten, falls der Block welche hat
                if block.get('has_children'):
                    process_blocks(block['id'], depth + 1)

        process_blocks(page_id)
        return "\n\n".join(content_parts)

    def _block_to_text(self, block: dict, depth: int = 0) -> str:
        """Konvertiert einen Notion-Block in Markdown."""
        block_type = block['type']
        indent = "  " * depth

        handlers = {
            'paragraph': lambda b: self._rich_text(b['paragraph']['rich_text']),
            'heading_1': lambda b: f"# {self._rich_text(b['heading_1']['rich_text'])}",
            'heading_2': lambda b: f"## {self._rich_text(b['heading_2']['rich_text'])}",
            'heading_3': lambda b: f"### {self._rich_text(b['heading_3']['rich_text'])}",
            'bulleted_list_item': lambda b: f"{indent}- {self._rich_text(b['bulleted_list_item']['rich_text'])}",
            'numbered_list_item': lambda b: f"{indent}1. {self._rich_text(b['numbered_list_item']['rich_text'])}",
            'to_do': lambda b: f"{indent}- [{'x' if b['to_do']['checked'] else ' '}] {self._rich_text(b['to_do']['rich_text'])}",
            'toggle': lambda b: f"{indent}> {self._rich_text(b['toggle']['rich_text'])}",
            'quote': lambda b: f"> {self._rich_text(b['quote']['rich_text'])}",
            'callout': lambda b: f"> {b['callout'].get('icon', {}).get('emoji', '')} {self._rich_text(b['callout']['rich_text'])}",
            'code': lambda b: f"```{b['code']['language']}\n{self._rich_text(b['code']['rich_text'])}\n```",
            'divider': lambda b: "---",
            'table_row': lambda b: self._table_row_to_text(b),
        }

        handler = handlers.get(block_type)
        return handler(block) if handler else ""

    def _rich_text(self, rich_text: list) -> str:
        """Konvertiert Notion-Rich-Text in Text mit Markdown-Formatierung."""
        parts = []
        for rt in rich_text:
            text = rt['plain_text']
            annotations = rt.get('annotations', {})

            if annotations.get('bold'):
                text = f"**{text}**"
            if annotations.get('italic'):
                text = f"*{text}*"
            if annotations.get('code'):
                text = f"`{text}`"
            if rt.get('href'):
                text = f"[{text}]({rt['href']})"

            parts.append(text)

        return ''.join(parts)

    def _table_row_to_text(self, block: dict) -> str:
        """Konvertiert eine Tabellenzeile."""
        cells = block['table_row']['cells']
        row = [self._rich_text(cell) for cell in cells]
        return "| " + " | ".join(row) + " |"

    def _extract_icon(self, page: dict) -> str:
        """Extrahiert das Icon der Seite."""
        icon = page.get('icon', {})
        if icon.get('type') == 'emoji':
            return icon.get('emoji', '')
        return ''


class NotionDatabaseConnector(NotionConnector):
    """Erweiterung zum Extrahieren von Notion-Datenbanken."""

    def get_database_entries(self, database_id: str) -> list:
        """
        Ruft alle Einträge einer Datenbank ab.

        Jeder Eintrag wird zu einem Dokument mit seinen Eigenschaften
        als strukturierten Metadaten.
        """
        entries = []
        has_more = True
        cursor = None

        while has_more:
            results = self.client.databases.query(
                database_id=database_id,
                start_cursor=cursor,
                page_size=100
            )

            for entry in results['results']:
                doc = self._format_database_entry(entry, database_id)
                if doc:
                    entries.append(doc)

            has_more = results['has_more']
            cursor = results.get('next_cursor')

        return entries

    def _format_database_entry(self, entry: dict, db_id: str) -> dict:
        """Formatiert einen Datenbankeintrag."""
        props = entry.get('properties', {})

        # Alle Eigenschaften als strukturierten Text extrahieren
        prop_texts = []
        metadata_props = {}

        for name, prop in props.items():
            value = self._extract_property_value(prop)
            if value:
                prop_texts.append(f"**{name}**: {value}")
                metadata_props[name] = value

        title = metadata_props.get('Name', metadata_props.get('Titre', 'Eintrag'))
        content = "\n".join(prop_texts)

        # Falls vorhanden, Seiteninhalt hinzufügen
        page_content = self._extract_content(entry['id'])
        if page_content:
            content += f"\n\n{page_content}"

        return {
            "id": f"notion_db_{entry['id']}",
            "title": title,
            "content": f"# {title}\n\n{content}",
            "metadata": {
                "source": "notion",
                "source_type": "database",
                "database_id": db_id,
                "entry_id": entry['id'],
                "url": entry.get('url', ''),
                "last_edited": entry['last_edited_time'],
                **metadata_props
            }
        }

    def _extract_property_value(self, prop: dict) -> str:
        """Extrahiert den Wert einer Notion-Eigenschaft."""
        prop_type = prop.get('type')

        extractors = {
            'title': lambda p: self._rich_text(p.get('title', [])),
            'rich_text': lambda p: self._rich_text(p.get('rich_text', [])),
            'number': lambda p: str(p.get('number', '')),
            'select': lambda p: p.get('select', {}).get('name', '') if p.get('select') else '',
            'multi_select': lambda p: ', '.join([s['name'] for s in p.get('multi_select', [])]),
            'date': lambda p: p.get('date', {}).get('start', '') if p.get('date') else '',
            'checkbox': lambda p: 'Ja' if p.get('checkbox') else 'Nein',
            'url': lambda p: p.get('url', ''),
            'email': lambda p: p.get('email', ''),
            'phone_number': lambda p: p.get('phone_number', ''),
            'status': lambda p: p.get('status', {}).get('name', '') if p.get('status') else '',
        }

        extractor = extractors.get(prop_type)
        return extractor(prop) if extractor else ''

Intelligente Synchronisierung

Die Synchronisierung kann je nach Bedarf auf verschiedene Arten ausgelöst werden:

Synchronisierung per Polling

DEVELOPERpython
from datetime import datetime, timedelta

class NotionSyncManager:
    def __init__(self, connector: NotionConnector, indexer):
        self.connector = connector
        self.indexer = indexer
        self.last_sync = None

    def sync_incremental(self):
        """
        Inkrementelle Synchronisierung: verarbeitet nur Seiten, die seit
        der letzten Synchronisierung geändert wurden.
        """
        pages = self.connector.get_all_pages()

        updated = []
        for page in pages:
            last_edited = datetime.fromisoformat(
                page['metadata']['last_edited'].replace('Z', '+00:00')
            )

            if self.last_sync is None or last_edited > self.last_sync:
                updated.append(page)

        if updated:
            self.indexer.upsert_documents(updated)
            print(f"{len(updated)} Seiten synchronisiert")

        self.last_sync = datetime.now()

    def sync_full(self):
        """Vollständige Synchronisierung: indexiert alles neu."""
        pages = self.connector.get_all_pages()
        self.indexer.replace_all(pages)
        self.last_sync = datetime.now()
        print(f"{len(pages)} Seiten indexiert")

Echtzeit-Synchronisierung

Für eine Echtzeit-Synchronisierung nutzen Sie die Notion-Webhooks (über die API verfügbar) oder einen Worker, der regelmäßig mit feiner Granularität pollt:

DEVELOPERpython
import schedule
import time

def start_sync_worker(sync_manager: NotionSyncManager):
    """Startet den Synchronisierungs-Worker."""
    # Inkrementelle Synchronisierung alle 5 Minuten
    schedule.every(5).minutes.do(sync_manager.sync_incremental)

    # Tägliche Vollsynchronisierung (Aufräumen)
    schedule.every().day.at("03:00").do(sync_manager.sync_full)

    while True:
        schedule.run_pending()
        time.sleep(60)

Optimierter System-Prompt für Notion

Der System-Prompt ist entscheidend, um qualitativ hochwertige Antworten zu erhalten. Hier eine für Unternehmens-Wikis optimierte Version:

DEVELOPERpython
NOTION_KB_SYSTEM_PROMPT = """Du bist der Wissensdatenbank-Assistent von {company_name}.
Du hilfst Mitarbeitenden dabei, Informationen schnell in unserem Notion-Wiki zu finden.

## Deine Aufgabe
- Fragen ausschließlich auf Basis des Wiki-Inhalts beantworten
- Quellen systematisch mit dem Seitentitel angeben
- Bei Fragen, die über das Wiki hinausgehen, an die richtige Person/das richtige Team verweisen

## Strikte Regeln
1. Niemals Informationen erfinden, die nicht im Wiki stehen
2. Wenn du dir nicht sicher bist, sage es klar
3. Aktuellste Informationen bevorzugen
4. Detailgrad an die Frage anpassen (Zusammenfassung vs. Detail)

## Antwortformat
- Beginne mit einer direkten Antwort auf die Frage
- Füge relevante Details hinzu
- Schließe mit „Quelle: [Seitentitel]" für jede Information

## Verfügbarer Kontext
{context}

## Gesprächsverlauf
{chat_history}
"""

def build_notion_prompt(company: str, docs: list, history: list) -> str:
    """Erstellt den Prompt mit dem Kontext."""
    context_parts = []
    for doc in docs:
        context_parts.append(f"### {doc['title']}\n{doc['content'][:1000]}")

    history_parts = []
    for msg in history[-5:]:  # Letzte 5 Nachrichten
        role = "Nutzer" if msg['role'] == 'user' else "Assistent"
        history_parts.append(f"{role}: {msg['content']}")

    return NOTION_KB_SYSTEM_PROMPT.format(
        company_name=company,
        context="\n\n---\n\n".join(context_parts),
        chat_history="\n".join(history_parts) if history_parts else "Keiner"
    )

Konkrete Anwendungsfälle

1. Onboarding neuer Mitarbeitender

Frage: „Wie richte ich meine Entwicklungsumgebung ein?"

RAG-Antwort:

Um deine Dev-Umgebung einzurichten, folge diesen Schritten:

1. Klone das Haupt-Repo: git clone [email protected]:company/main.git
2. Installiere die Abhängigkeiten: npm install
3. Kopiere die Datei .env.example nach .env
4. Fordere die Zugangsdaten beim DevOps-Team über Slack an (#devops-support)

Quelle: [Guide Setup Dev] | [Onboarding Tech]

2. Häufige HR-Fragen

Frage: „Wie beantrage ich Urlaub?"

RAG-Antwort:

Um Urlaub zu beantragen:

1. Gehe zu Payfit > Abwesenheiten > Neuer Antrag
2. Wähle den Typ (Erholungsurlaub, Überstundenausgleich, unbezahlt)
3. Gib die Daten an und bestätige

Genehmigungsfrist: 48 Stunden durch deine Führungskraft. Bei Dringlichkeit wende dich an HR.

Quelle: [Verfahren Urlaub und Abwesenheiten]

3. Technische Dokumentation

Frage: „Wie sieht die Architektur des Zahlungsdienstes aus?"

RAG-Antwort:

Der Zahlungsdienst folgt einer event-getriebenen Architektur:

• API Gateway: Authentifizierung und Routing
• Payment Service: Orchestrierung der Transaktionen
• Stripe Adapter: Stripe-Integration
• Event Bus: Kafka für Benachrichtigungen

Vollständiges Diagramm auf der zugehörigen Seite verfügbar.

Quelle: [Architektur Payment Service] | [Technische Diagramme]

Best Practices

Notion für RAG strukturieren

Berechtigungen verwalten

Der RAG erbt die Berechtigungen der Notion-Integration. Für eine granulare Kontrolle:

1. Eine dedizierte Integration für den RAG erstellen
2. Nur öffentliche Seiten mit der Integration teilen
3. Zugriffe bei Multi-Tenant-Umgebungen pro Workspace verwalten

Qualität überwachen

• Fragen ohne Antwort verfolgen
• Nutzerfeedback sammeln
• Am häufigsten zitierte Seiten identifizieren
• Veraltete Inhalte erkennen

Weiterführende Ressourcen

• Unternehmens-Wissensdatenbank - Vollständiger Pillar-Guide
• Confluence + RAG - Für Atlassian-Umgebungen
• SharePoint + RAG - Für Microsoft 365
• Einführung in RAG - Die Grundlagen

Verbinden Sie Notion mit Ailog

Verwandeln Sie Ihr Notion-Wiki in einen intelligenten Assistenten, ohne eine Zeile Code zu schreiben. Ailog vereinfacht die Integration:

• Nativer Notion-Connector: Automatische Synchronisierung in wenigen Klicks
• Semantische Suche: Finden Sie Informationen mit Ihren eigenen Worten, nicht denen des Wikis
• Multi-Workspace: Verwalten Sie mehrere Notion-Bereiche in einer einzigen Oberfläche
• Zugriffskontrolle: Respektieren Sie die Berechtigungen Ihrer Organisation
• Hosting in Frankreich: Daten auf französischen Servern, native DSGVO-Konformität

Testen Sie Ailog kostenlos und stellen Sie Ihren Notion-Assistenten in 10 Minuten bereit.