SharePoint + RAG : Exploiter vos documents Microsoft 365

SharePoint est le coffre-fort documentaire de millions d'entreprises. Procedures, contrats, presentations, rapports : tout finit dans SharePoint. Mais cette richesse devient rapidement un frein. Retrouver un document precis dans des gigaoctets de fichiers tient du parcours du combattant. La recherche native ne comprend pas les questions en langage naturel et se limite aux mots-cles exacts.

Un assistant RAG connecte a SharePoint change la donne. Il comprend le sens de vos questions, explore vos documents en profondeur, et synthetise des reponses avec les sources pertinentes. Ce guide vous accompagne dans cette integration critique pour les environnements Microsoft 365.

Pourquoi SharePoint pose probleme pour la recherche

Les frustrations quotidiennes

Les utilisateurs SharePoint connaissent ces situations :

• Recherche par mots-cles limitee : "Comment demander un conge" ne trouve pas "procedure d'absence"
• Resultats noyes : Des centaines de documents pour une requete simple
• Versions multiples : Lequel est le bon document parmi les 5 versions ?
• Pas de synthese : Il faut ouvrir et lire chaque fichier
• Formats heterogenes : PDF, Word, Excel, PowerPoint... difficiles a parcourir

Statistiques revelateurs

Recherche SharePoint vs RAG

Architecture SharePoint + RAG

L'integration repose sur Microsoft Graph API pour l'extraction et une pipeline RAG pour l'indexation et l'interrogation.

┌─────────────────────────────────────────────────────────────────────────┐
│                     Architecture 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  │                                                             │
│   ┌──────┴───────┐     ┌─────────────┐     ┌──────────────┐           │
│   │   Question   │────▶│  Retrieval  │────▶│   Reranker   │           │
│   │  utilisateur │     │   Top-30    │     │    Top-5     │           │
│   └──────────────┘     └─────────────┘     └──────┬───────┘           │
│                                                    │                   │
│                        ┌──────────────┐    ┌──────┴───────┐           │
│                        │   Reponse    │◀───│     LLM      │           │
│                        │  + Sources   │    │  GPT-4/Claude│           │
│                        └──────────────┘    └──────────────┘           │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

Flux de donnees

1. Authentification : OAuth2 avec Microsoft Entra ID (Azure AD)
2. Extraction : Microsoft Graph API recupere les fichiers des sites/drives
3. Parsing : Conversion des formats Office en texte exploitable
4. Chunking : Decoupage en segments de 512 tokens
5. Embedding : Vectorisation avec BGE-M3 (multilingue)
6. Indexation : Stockage dans Qdrant avec metadonnees riches
7. Retrieval : Recherche semantique sur les questions
8. Generation : LLM synthetise avec citations

Connecteur SharePoint complet

Voici l'implementation de reference utilisant Microsoft Graph :

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

# Parsers pour les formats Office
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
    ):
        """
        Initialise le connecteur SharePoint via Microsoft Graph.

        Args:
            tenant_id: ID du tenant Azure AD
            client_id: ID de l'application enregistree
            client_secret: Secret de l'application
            site_url: URL du site SharePoint (optionnel)
        """
        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:
        """Obtient un token d'acces via 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"Erreur auth: {result.get('error_description')}")

    def _graph_request(self, endpoint: str, params: dict = None) -> dict:
        """Execute une requete Graph API."""
        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]:
        """Recupere tous les sites SharePoint accessibles."""
        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]:
        """Recupere les bibliotheques de documents d'un site."""
        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]:
        """
        Recupere tous les fichiers d'une bibliotheque de maniere recursive.
        """
        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 = ""
    ):
        """Parcours recursif des dossiers."""
        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:
                # C'est un dossier, on descend
                self._traverse_folder(
                    site_id, drive_id, item['id'], files, item_path
                )
            elif 'file' in item:
                # C'est un fichier
                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:
        """Telecharge le contenu d'un fichier."""
        response = requests.get(download_url)
        response.raise_for_status()
        return response.content

    def extract_document_content(self, file_info: dict) -> Optional[dict]:
        """
        Extrait le contenu textuel d'un document.

        Supporte : 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)

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

            else:
                return None  # Format non supporte

            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**Chemin**: {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"Erreur extraction {file_info['name']}: {e}")
            return None

    def _parse_pdf(self, content: bytes) -> str:
        """Extrait le texte d'un 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:
        """Extrait le texte d'un DOCX."""
        doc = Document(BytesIO(content))
        paragraphs = []

        for para in doc.paragraphs:
            if para.text.strip():
                # Detecter les titres
                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)

        # Extraire les tableaux
        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:
        """Convertit un tableau Word en 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:
        """Extrait le contenu d'un Excel."""
        workbook = openpyxl.load_workbook(BytesIO(content), data_only=True)
        sheets_content = []

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

            # Convertir en tableau markdown
            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:
        """Extrait le texte d'un PowerPoint."""
        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):
    """Extension pour indexer plusieurs sites."""

    def get_all_documents(
        self,
        site_ids: List[str] = None,
        exclude_paths: List[str] = None,
        file_types: List[str] = None
    ) -> List[dict]:
        """
        Indexe les documents de plusieurs sites.

        Args:
            site_ids: Sites a indexer (None = tous)
            exclude_paths: Chemins a exclure (ex: ['Archive/', 'Old/'])
            file_types: Extensions a inclure (ex: ['.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"Indexation 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:
                    # Filtrer par chemin
                    if exclude_paths:
                        if any(file_info['path'].startswith(p) for p in exclude_paths):
                            continue

                    # Filtrer par type
                    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

Gestion des permissions

SharePoint a des permissions complexes. Le RAG doit les respecter :

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

    def get_item_permissions(self, site_id: str, item_id: str) -> dict:
        """Recupere les permissions d'un element."""
        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:
        """Verifie si un utilisateur peut acceder a un document."""
        for perm in doc_permissions:
            if perm['type'] == 'user' and perm['email'] == user_email:
                return True
            # Pour les groupes, il faudrait verifier l'appartenance
        return False

Synchronisation incrementale

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):
        """Synchronise uniquement les fichiers modifies."""
        # Utiliser delta query pour les changements
        endpoint = f"/sites/{site_id}/drives/{drive_id}/root/delta"

        if self.last_sync_token:
            # Ajouter le token de changement
            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'):
                # Fichier supprime
                self.indexer.delete_document(f"sharepoint_{item['id']}")
                deleted += 1
            elif 'file' in item:
                # Fichier modifie ou ajoute
                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

        # Sauvegarder le token pour la prochaine sync
        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} mis a jour, {deleted} supprimes")

Cas d'usage concrets

1. Recherche de contrats

Question : "Quelles sont les conditions de resiliation du contrat avec Fournisseur X ?"

Reponse RAG :

D'apres le contrat-cadre signe le 15/03/2024 :

• Preavis de resiliation : 3 mois
• Conditions : lettre recommandee avec AR
• Penalites : 10% du montant restant si resiliation anticipee
• Exception : resiliation immediate en cas de manquement grave

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

2. Procedures internes

Question : "Comment faire une note de frais ?"

Reponse RAG :

Procedure note de frais :

1. Remplir le formulaire NDF-2024.xlsx
2. Joindre les justificatifs scannes
3. Soumettre via le portail RH
4. Validation manager sous 5 jours

Plafonds : 50 EUR/repas, 150 EUR/nuit hotel

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

3. Documentation technique

Question : "Comment configurer le SSO avec Azure AD ?"

Reponse RAG :

Configuration SSO Azure AD :

1. Creer une Enterprise Application dans Azure
2. Configurer SAML avec les endpoints fournis
3. Mapper les attributs utilisateur
4. Tester avec un compte de test

Voir le guide detaille pour les screenshots.

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

Bonnes pratiques

Structure SharePoint pour le RAG

Optimiser l'indexation

• Exclure les dossiers temporaires et archives
• Prioriser les formats textuels (PDF, Word)
• Limiter la taille des fichiers (< 50 MB)
• Mettre a jour quotidiennement via delta sync

Ressources complementaires

• Base de connaissances entreprise - Guide pilier
• Notion + RAG - Pour les equipes sur Notion
• Confluence + RAG - Pour Atlassian
• Introduction au RAG - Les fondamentaux

Connectez SharePoint avec Ailog

Liberez le potentiel de vos documents Microsoft 365. Ailog offre :

• Connecteur Microsoft Graph : Integration native SharePoint et OneDrive
• Parsing multi-formats : PDF, Word, Excel, PowerPoint
• Respect des permissions : Acces conditionnel par utilisateur
• Sync incrementale : Mise a jour automatique des documents
• Hebergement France : Conformite RGPD native

Testez Ailog gratuitement et rendez vos documents SharePoint interrogeables par IA en 20 minutes.