Chapitre 2 : Base de connaissances d'entreprise

Aperçu du projet

Construisez une base de connaissances d'entreprise avec :

• Plusieurs sources de données (Confluence, Notion, docs)
• Authentification des utilisateurs
• Contrôle d'accès basé sur le département
• Boucles de rétroaction

Architecture

Service de synchronisation des données

# sync_service.py
from datetime import datetime
import schedule

class KnowledgeSync:
    def __init__(self, vectorstore, metadata_db):
        self.vectorstore = vectorstore
        self.metadata_db = metadata_db
    
    def sync_confluence(self):
        """Synchroniser depuis Confluence"""
        from atlassian import Confluence
        
        confluence = Confluence(url=CONF_URL, token=CONF_TOKEN)
        spaces = confluence.get_all_spaces()
        
        for space in spaces:
            pages = confluence.get_all_pages_from_space(space['key'])
            for page in pages:
                self._process_page(page, source="confluence")
    
    def sync_notion(self):
        """Synchroniser depuis Notion"""
        from notion_client import Client
        
        notion = Client(auth=NOTION_TOKEN)
        # Récupérer et traiter les pages Notion
    
    def _process_page(self, page, source):
        # Vérifier si mis à jour depuis la dernière synchronisation
        if self._needs_update(page, source):
            chunks = self._chunk_content(page)
            self._upsert_vectors(chunks, page.id)
            self._update_metadata(page, source)

# Planifier les synchronisations
schedule.every(1).hour.do(sync.sync_confluence)
schedule.every(1).hour.do(sync.sync_notion)

Contrôle d'accès

# access_control.py
from enum import Enum

class Department(Enum):
    ENGINEERING = "engineering"
    SALES = "sales"
    HR = "hr"
    ALL = "all"

def get_accessible_docs(user):
    """Obtenir les IDs des documents accessibles par l'utilisateur"""
    user_dept = user.department
    user_role = user.role
    
    # L'administrateur voit tout
    if user_role == "admin":
        return None  # Pas de filtre
    
    # Département + documents publics
    allowed = metadata_db.query(
        f"department IN ('{user_dept}', 'all')"
    )
    
    return [doc.id for doc in allowed]

def search_with_access(query, user):
    """Recherche avec contrôle d'accès"""
    allowed_docs = get_accessible_docs(user)
    
    filter_dict = {}
    if allowed_docs:
        filter_dict["doc_id"] = {"$in": allowed_docs}
    
    return vectorstore.similarity_search(
        query,
        filter=filter_dict
    )

Boucle de rétroaction

# feedback.py
from pydantic import BaseModel

class Feedback(BaseModel):
    query_id: str
    helpful: bool
    comment: str = None

def record_feedback(feedback: Feedback):
    """Enregistrer les commentaires des utilisateurs"""
    db.feedback.insert({
        "query_id": feedback.query_id,
        "helpful": feedback.helpful,
        "comment": feedback.comment,
        "timestamp": datetime.utcnow()
    })

def analyze_feedback():
    """Identifier les domaines d'amélioration"""
    # Les requêtes mal notées indiquent des problèmes de récupération/génération
    poor_queries = db.feedback.find({"helpful": False})
    
    for query in poor_queries:
        # Consigner pour examen
        # Potentiellement réentraîner ou ajuster le pipeline
        pass

Caractéristiques clés

Fonctionnalité	Implémentation
Multi-source	Service de synchronisation
Auth	JWT + RBAC
Contrôle d'accès	Filtrage des métadonnées
Versioning	Métadonnées des documents
Rétroaction	Évaluation + commentaires