SharePoint + RAG: Ihre Microsoft-365-Dokumente nutzen

SharePoint ist der Dokumententresor von Millionen von Unternehmen. Verfahren, Verträge, Präsentationen, Berichte: Alles landet in SharePoint. Doch dieser Reichtum wird schnell zum Hindernis. Ein bestimmtes Dokument in Gigabytes an Dateien wiederzufinden, gleicht einem Hindernislauf. Die native Suche versteht keine Fragen in natürlicher Sprache und beschränkt sich auf exakte Stichwörter.

Ein mit SharePoint verbundener RAG-Assistent ändert alles. Er versteht den Sinn Ihrer Fragen, durchsucht Ihre Dokumente gründlich und erstellt Antworten mit den relevanten Quellen. Dieser Leitfaden begleitet Sie bei dieser kritischen Integration für Microsoft-365-Umgebungen.

Warum SharePoint ein Problem für die Suche darstellt

Tägliche Frustrationen

SharePoint-Nutzer kennen diese Situationen:

• Eingeschränkte Stichwortsuche: "Wie beantrage ich Urlaub" findet nicht "Abwesenheitsverfahren"
• Überflutete Ergebnisse: Hunderte Dokumente für eine einfache Anfrage
• Mehrere Versionen: Welches ist das richtige Dokument unter den 5 Versionen?
• Keine Synthese: Man muss jede Datei öffnen und lesen
• Heterogene Formate: PDF, Word, Excel, PowerPoint... schwer zu durchsuchen

Aufschlussreiche Statistiken

SharePoint-Suche vs. RAG

Architektur SharePoint + RAG

Die Integration basiert auf der Microsoft Graph API für die Extraktion und einer RAG-Pipeline für die Indexierung und Abfrage.

┌─────────────────────────────────────────────────────────────────────────┐
│                     Architektur SharePoint + RAG                         │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│   MICROSOFT 365                EXTRACTION                  PROCESSING   │
│   ┌──────────────┐            ┌──────────────┐          ┌────────────┐ │
│   │  SharePoint  │───────────▶│ Graph API    │─────────▶│  Parsing   │ │
│   │              │            │              │          │            │ │
│   │  - Sites     │            │  /sites      │          │  PDF→Text  │ │
│   │  - Libraries │            │  /drives     │          │  DOCX→MD   │ │
│   │  - Documents │            │  /items      │          │  XLSX→CSV  │ │
│   └──────────────┘            └──────────────┘          └──────┬─────┘ │
│                                                                │       │
│   ┌──────────────┐                                      ┌──────┴─────┐ │
│   │   OneDrive   │──────────────────────────────────────│  Chunking  │ │
│   │   Teams      │                                      │  512 tok   │ │
│   └──────────────┘                                      └──────┬─────┘ │
│                                                                │       │
│                                                         ┌──────┴─────┐ │
│                                                         │ Embeddings │ │
│   VECTOR DB                                             │  BGE-M3    │ │
│   ┌──────────────┐                                      └──────┬─────┘ │
│   │    Qdrant    │◀────────────────────────────────────────────┘       │
│   │              │                                                     │
│   │  HNSW Index  │                                                     │
│   └──────┬───────┘                                                     │
│          │                                                             │
│   QUERY  │                                                             │
│   ┌──────┴───────┐     ┌─────────────┐     ┌──────────────┐           │
│   │    Frage     │────▶│  Retrieval  │────▶│   Reranker   │           │
│   │  Benutzer    │     │   Top-30    │     │    Top-5     │           │
│   └──────────────┘     └─────────────┘     └──────┬───────┘           │
│                                                    │                   │
│                        ┌──────────────┐    ┌──────┴───────┐           │
│                        │   Antwort    │◀───│     LLM      │           │
│                        │  + Sources   │    │  GPT-4/Claude│           │
│                        └──────────────┘    └──────────────┘           │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Datenfluss

1. Authentifizierung: OAuth2 mit Microsoft Entra ID (Azure AD)
2. Extraktion: Microsoft Graph API ruft die Dateien der Sites/Drives ab
3. Parsing: Konvertierung der Office-Formate in verwertbaren Text
4. Chunking: Aufteilung in Segmente von 512 Tokens
5. Embedding: Vektorisierung mit BGE-M3 (mehrsprachig)
6. Indexierung: Speicherung in Qdrant mit umfangreichen Metadaten
7. Retrieval: Semantische Suche auf den Fragen
8. Generierung: LLM synthetisiert mit Zitaten

Vollständiger SharePoint-Connector

Hier ist die Referenzimplementierung mit Microsoft Graph:

DEVELOPERpython
import msal
import requests
from typing import Optional, List
import hashlib
from io import BytesIO

# Parser für die Office-Formate
from pypdf import PdfReader
from docx import Document
import openpyxl
from pptx import Presentation

class SharePointConnector:
    def __init__(
        self,
        tenant_id: str,
        client_id: str,
        client_secret: str,
        site_url: str = None
    ):
        """
        Initialisiert den SharePoint-Connector über Microsoft Graph.

        Args:
            tenant_id: ID des Azure AD-Tenants
            client_id: ID der registrierten Anwendung
            client_secret: Secret der Anwendung
            site_url: URL der SharePoint-Site (optional)
        """
        self.tenant_id = tenant_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.site_url = site_url
        self.graph_url = "https://graph.microsoft.com/v1.0"
        self._token = None

    def _get_token(self) -> str:
        """Holt ein Zugriffstoken über MSAL."""
        if self._token:
            return self._token

        app = msal.ConfidentialClientApplication(
            self.client_id,
            authority=f"https://login.microsoftonline.com/{self.tenant_id}",
            client_credential=self.client_secret
        )

        result = app.acquire_token_for_client(
            scopes=["https://graph.microsoft.com/.default"]
        )

        if "access_token" in result:
            self._token = result["access_token"]
            return self._token
        else:
            raise Exception(f"Auth-Fehler: {result.get('error_description')}")

    def _graph_request(self, endpoint: str, params: dict = None) -> dict:
        """Führt eine Graph-API-Anfrage aus."""
        headers = {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json"
        }

        response = requests.get(
            f"{self.graph_url}{endpoint}",
            headers=headers,
            params=params
        )
        response.raise_for_status()
        return response.json()

    def get_sites(self) -> List[dict]:
        """Ruft alle zugänglichen SharePoint-Sites ab."""
        sites = []
        endpoint = "/sites?search=*"

        while endpoint:
            result = self._graph_request(endpoint)
            sites.extend(result.get('value', []))
            endpoint = result.get('@odata.nextLink', '').replace(self.graph_url, '')

        return [
            {
                'id': site['id'],
                'name': site['displayName'],
                'url': site['webUrl'],
                'description': site.get('description', '')
            }
            for site in sites
        ]

    def get_document_libraries(self, site_id: str) -> List[dict]:
        """Ruft die Dokumentbibliotheken einer Site ab."""
        result = self._graph_request(f"/sites/{site_id}/drives")

        return [
            {
                'id': drive['id'],
                'name': drive['name'],
                'type': drive.get('driveType'),
                'quota': drive.get('quota', {})
            }
            for drive in result.get('value', [])
        ]

    def get_all_files(self, site_id: str, drive_id: str) -> List[dict]:
        """
        Ruft rekursiv alle Dateien einer Bibliothek ab.
        """
        files = []
        self._traverse_folder(site_id, drive_id, "root", files)
        return files

    def _traverse_folder(
        self,
        site_id: str,
        drive_id: str,
        folder_id: str,
        files: list,
        path: str = ""
    ):
        """Rekursives Durchlaufen der Ordner."""
        endpoint = f"/sites/{site_id}/drives/{drive_id}/items/{folder_id}/children"
        result = self._graph_request(endpoint)

        for item in result.get('value', []):
            item_path = f"{path}/{item['name']}" if path else item['name']

            if 'folder' in item:
                # Es ist ein Ordner, wir steigen ab
                self._traverse_folder(
                    site_id, drive_id, item['id'], files, item_path
                )
            elif 'file' in item:
                # Es ist eine Datei
                file_info = {
                    'id': item['id'],
                    'name': item['name'],
                    'path': item_path,
                    'size': item.get('size', 0),
                    'mime_type': item['file'].get('mimeType'),
                    'created': item.get('createdDateTime'),
                    'modified': item.get('lastModifiedDateTime'),
                    'created_by': item.get('createdBy', {}).get('user', {}).get('displayName'),
                    'modified_by': item.get('lastModifiedBy', {}).get('user', {}).get('displayName'),
                    'download_url': item.get('@microsoft.graph.downloadUrl'),
                    'web_url': item.get('webUrl')
                }
                files.append(file_info)

    def download_file(self, download_url: str) -> bytes:
        """Lädt den Inhalt einer Datei herunter."""
        response = requests.get(download_url)
        response.raise_for_status()
        return response.content

    def extract_document_content(self, file_info: dict) -> Optional[dict]:
        """
        Extrahiert den Textinhalt eines Dokuments.

        Unterstützt: PDF, DOCX, XLSX, PPTX, TXT, MD
        """
        if not file_info.get('download_url'):
            return None

        mime_type = file_info.get('mime_type', '')
        name = file_info['name'].lower()

        try:
            content_bytes = self.download_file(file_info['download_url'])
            text_content = ""

            # PDF
            if mime_type == 'application/pdf' or name.endswith('.pdf'):
                text_content = self._parse_pdf(content_bytes)

            # Word
            elif 'wordprocessingml' in mime_type or name.endswith('.docx'):
                text_content = self._parse_docx(content_bytes)

            # Excel
            elif 'spreadsheetml' in mime_type or name.endswith('.xlsx'):
                text_content = self._parse_xlsx(content_bytes)

            # PowerPoint
            elif 'presentationml' in mime_type or name.endswith('.pptx'):
                text_content = self._parse_pptx(content_bytes)

            # Klartext
            elif mime_type.startswith('text/') or name.endswith(('.txt', '.md', '.csv')):
                text_content = content_bytes.decode('utf-8', errors='ignore')

            else:
                return None  # Nicht unterstütztes Format

            if not text_content or len(text_content) < 50:
                return None

            content_hash = hashlib.md5(text_content.encode()).hexdigest()

            return {
                'id': f"sharepoint_{file_info['id']}",
                'title': file_info['name'],
                'content': f"# {file_info['name']}\n\n**Pfad**: {file_info['path']}\n\n{text_content}",
                'metadata': {
                    'source': 'sharepoint',
                    'source_type': 'document',
                    'file_id': file_info['id'],
                    'file_name': file_info['name'],
                    'file_path': file_info['path'],
                    'file_size': file_info['size'],
                    'mime_type': file_info['mime_type'],
                    'url': file_info['web_url'],
                    'created': file_info['created'],
                    'modified': file_info['modified'],
                    'created_by': file_info['created_by'],
                    'modified_by': file_info['modified_by'],
                    'content_hash': content_hash
                }
            }

        except Exception as e:
            print(f"Extraktionsfehler {file_info['name']}: {e}")
            return None

    def _parse_pdf(self, content: bytes) -> str:
        """Extrahiert den Text aus einem PDF."""
        reader = PdfReader(BytesIO(content))
        text_parts = []

        for page in reader.pages:
            text = page.extract_text()
            if text:
                text_parts.append(text)

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

    def _parse_docx(self, content: bytes) -> str:
        """Extrahiert den Text aus einem DOCX."""
        doc = Document(BytesIO(content))
        paragraphs = []

        for para in doc.paragraphs:
            if para.text.strip():
                # Überschriften erkennen
                if para.style.name.startswith('Heading'):
                    level = int(para.style.name[-1]) if para.style.name[-1].isdigit() else 1
                    paragraphs.append(f"{'#' * level} {para.text}")
                else:
                    paragraphs.append(para.text)

        # Tabellen extrahieren
        for table in doc.tables:
            table_text = self._docx_table_to_markdown(table)
            paragraphs.append(table_text)

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

    def _docx_table_to_markdown(self, table) -> str:
        """Konvertiert eine Word-Tabelle in Markdown."""
        rows = []
        for i, row in enumerate(table.rows):
            cells = [cell.text.replace('|', '\\|') for cell in row.cells]
            rows.append('| ' + ' | '.join(cells) + ' |')
            if i == 0:
                rows.append('| ' + ' | '.join(['---'] * len(cells)) + ' |')
        return '\n'.join(rows)

    def _parse_xlsx(self, content: bytes) -> str:
        """Extrahiert den Inhalt einer Excel-Datei."""
        workbook = openpyxl.load_workbook(BytesIO(content), data_only=True)
        sheets_content = []

        for sheet_name in workbook.sheetnames:
            sheet = workbook[sheet_name]
            sheet_text = [f"## Blatt: {sheet_name}"]

            # In Markdown-Tabelle umwandeln
            rows = []
            for i, row in enumerate(sheet.iter_rows(values_only=True)):
                if any(cell is not None for cell in row):
                    cells = [str(cell) if cell else '' for cell in row]
                    rows.append('| ' + ' | '.join(cells) + ' |')
                    if i == 0:
                        rows.append('| ' + ' | '.join(['---'] * len(cells)) + ' |')

            sheet_text.append('\n'.join(rows))
            sheets_content.append('\n'.join(sheet_text))

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

    def _parse_pptx(self, content: bytes) -> str:
        """Extrahiert den Text aus einer PowerPoint-Datei."""
        prs = Presentation(BytesIO(content))
        slides_content = []

        for i, slide in enumerate(prs.slides, 1):
            slide_text = [f"## Slide {i}"]

            for shape in slide.shapes:
                if hasattr(shape, 'text') and shape.text.strip():
                    slide_text.append(shape.text)

            slides_content.append('\n\n'.join(slide_text))

        return '\n\n---\n\n'.join(slides_content)


class SharePointMultiSiteConnector(SharePointConnector):
    """Erweiterung zur Indexierung mehrerer Sites."""

    def get_all_documents(
        self,
        site_ids: List[str] = None,
        exclude_paths: List[str] = None,
        file_types: List[str] = None
    ) -> List[dict]:
        """
        Indexiert die Dokumente mehrerer Sites.

        Args:
            site_ids: Zu indexierende Sites (None = alle)
            exclude_paths: Auszuschließende Pfade (z. B. ['Archive/', 'Old/'])
            file_types: Einzuschließende Dateiendungen (z. B. ['.pdf', '.docx'])
        """
        all_docs = []

        sites = self.get_sites()
        if site_ids:
            sites = [s for s in sites if s['id'] in site_ids]

        for site in sites:
            print(f"Indexierung Site: {site['name']}")

            libraries = self.get_document_libraries(site['id'])

            for library in libraries:
                files = self.get_all_files(site['id'], library['id'])

                for file_info in files:
                    # Nach Pfad filtern
                    if exclude_paths:
                        if any(file_info['path'].startswith(p) for p in exclude_paths):
                            continue

                    # Nach Typ filtern
                    if file_types:
                        if not any(file_info['name'].lower().endswith(t) for t in file_types):
                            continue

                    doc = self.extract_document_content(file_info)
                    if doc:
                        doc['metadata']['site_name'] = site['name']
                        doc['metadata']['library_name'] = library['name']
                        all_docs.append(doc)

        return all_docs

Berechtigungsverwaltung

SharePoint hat komplexe Berechtigungen. Der RAG muss sie respektieren:

DEVELOPERpython
class SharePointPermissionManager:
    def __init__(self, connector: SharePointConnector):
        self.connector = connector

    def get_item_permissions(self, site_id: str, item_id: str) -> dict:
        """Ruft die Berechtigungen eines Elements ab."""
        endpoint = f"/sites/{site_id}/drive/items/{item_id}/permissions"
        result = self.connector._graph_request(endpoint)

        permissions = []
        for perm in result.get('value', []):
            if 'grantedToV2' in perm:
                granted = perm['grantedToV2']
                if 'user' in granted:
                    permissions.append({
                        'type': 'user',
                        'id': granted['user'].get('id'),
                        'email': granted['user'].get('email'),
                        'roles': perm.get('roles', [])
                    })
                elif 'group' in granted:
                    permissions.append({
                        'type': 'group',
                        'id': granted['group'].get('id'),
                        'name': granted['group'].get('displayName'),
                        'roles': perm.get('roles', [])
                    })

        return permissions

    def user_can_access(self, user_email: str, doc_permissions: list) -> bool:
        """Prüft, ob ein Benutzer auf ein Dokument zugreifen kann."""
        for perm in doc_permissions:
            if perm['type'] == 'user' and perm['email'] == user_email:
                return True
            # Bei Gruppen müsste die Mitgliedschaft geprüft werden
        return False

Inkrementelle Synchronisation

DEVELOPERpython
from datetime import datetime, timedelta

class SharePointSyncManager:
    def __init__(self, connector: SharePointConnector, indexer):
        self.connector = connector
        self.indexer = indexer
        self.last_sync = None
        self.last_sync_token = None

    def sync_incremental(self, site_id: str, drive_id: str):
        """Synchronisiert nur die geänderten Dateien."""
        # Delta-Query für Änderungen verwenden
        endpoint = f"/sites/{site_id}/drives/{drive_id}/root/delta"

        if self.last_sync_token:
            # Änderungstoken hinzufügen
            endpoint += f"?token={self.last_sync_token}"

        result = self.connector._graph_request(endpoint)
        changes = result.get('value', [])

        updated = 0
        deleted = 0

        for item in changes:
            if item.get('deleted'):
                # Datei gelöscht
                self.indexer.delete_document(f"sharepoint_{item['id']}")
                deleted += 1
            elif 'file' in item:
                # Datei geändert oder hinzugefügt
                file_info = {
                    'id': item['id'],
                    'name': item['name'],
                    'path': item.get('parentReference', {}).get('path', ''),
                    'size': item.get('size', 0),
                    'mime_type': item['file'].get('mimeType'),
                    'modified': item.get('lastModifiedDateTime'),
                    'download_url': item.get('@microsoft.graph.downloadUrl'),
                    'web_url': item.get('webUrl')
                }
                doc = self.connector.extract_document_content(file_info)
                if doc:
                    self.indexer.upsert_document(doc)
                    updated += 1

        # Token für die nächste Sync speichern
        delta_link = result.get('@odata.deltaLink', '')
        if 'token=' in delta_link:
            self.last_sync_token = delta_link.split('token=')[-1]

        self.last_sync = datetime.now()
        print(f"Sync: {updated} aktualisiert, {deleted} gelöscht")

Konkrete Anwendungsfälle

1. Vertragssuche

Frage: "Wie lauten die Kündigungsbedingungen des Vertrags mit Lieferant X?"

RAG-Antwort:

Laut dem am 15.03.2024 unterzeichneten Rahmenvertrag:

• Kündigungsfrist: 3 Monate
• Bedingungen: Einschreiben mit Rückschein
• Vertragsstrafen: 10% des Restbetrags bei vorzeitiger Kündigung
• Ausnahme: sofortige Kündigung bei schwerwiegender Pflichtverletzung

Quelle: [Contrats > Fournisseur X > Contrat-Cadre-2024.pdf] (Seite 12-13)

2. Interne Verfahren

Frage: "Wie erstelle ich eine Spesenabrechnung?"

RAG-Antwort:

Ablauf der Spesenabrechnung:

1. Formular NDF-2024.xlsx ausfüllen
2. Gescannte Belege beifügen
3. Über das HR-Portal einreichen
4. Genehmigung durch den Manager innerhalb von 5 Tagen

Obergrenzen: 50 EUR/Mahlzeit, 150 EUR/Hotelübernachtung

Quelle: [RH > Procedures > Note-de-frais-procedure.docx]

3. Technische Dokumentation

Frage: "Wie konfiguriere ich SSO mit Azure AD?"

RAG-Antwort:

Konfiguration SSO Azure AD:

1. Eine Enterprise Application in Azure erstellen
2. SAML mit den bereitgestellten Endpunkten konfigurieren
3. Benutzerattribute zuordnen
4. Mit einem Testkonto testen

Siehe den ausführlichen Leitfaden für Screenshots.

Quelle: [IT > Documentation > SSO-Azure-AD-Setup.pptx]

Best Practices

SharePoint-Struktur für RAG

Indexierung optimieren

• Temporäre Ordner und Archive ausschließen
• Textuelle Formate priorisieren (PDF, Word)
• Dateigröße begrenzen (< 50 MB)
• Täglich über Delta-Sync aktualisieren

Weiterführende Ressourcen

• Unternehmens-Wissensdatenbank - Säulenleitfaden
• Notion + RAG - Für Teams auf Notion
• Confluence + RAG - Für Atlassian
• Einführung in RAG - Die Grundlagen

Verbinden Sie SharePoint mit Ailog

Entfesseln Sie das Potenzial Ihrer Microsoft-365-Dokumente. Ailog bietet:

• Microsoft Graph-Connector: Native SharePoint- und OneDrive-Integration
• Multi-Format-Parsing: PDF, Word, Excel, PowerPoint
• Berechtigungsrespektierung: Bedingter Zugriff pro Benutzer
• Inkrementelle Synchronisation: Automatische Dokumentenaktualisierung
• Hosting in Frankreich: Native DSGVO-Konformität

Ailog kostenlos testen und machen Sie Ihre SharePoint-Dokumente in 20 Minuten per KI durchsuchbar.