Capítulo 2: Base de Conocimiento Empresarial

Descripción General del Proyecto

Construya una base de conocimiento empresarial con:

• Múltiples fuentes de datos (Confluence, Notion, documentación)
• Autenticación de usuarios
• Control de acceso basado en departamentos
• Ciclos de retroalimentación

Arquitectura

Servicio de Sincronización de Datos

# 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):
        """Sync from 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):
        """Sync from Notion"""
        from notion_client import Client
        
        notion = Client(auth=NOTION_TOKEN)
        # Fetch and process Notion pages
    
    def _process_page(self, page, source):
        # Check if updated since last sync
        if self._needs_update(page, source):
            chunks = self._chunk_content(page)
            self._upsert_vectors(chunks, page.id)
            self._update_metadata(page, source)

# Schedule syncs
schedule.every(1).hour.do(sync.sync_confluence)
schedule.every(1).hour.do(sync.sync_notion)

Control de Acceso

# access_control.py
from enum import Enum

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

def get_accessible_docs(user):
    """Get document IDs user can access"""
    user_dept = user.department
    user_role = user.role
    
    # Admin sees everything
    if user_role == "admin":
        return None  # No filter
    
    # Department + public docs
    allowed = metadata_db.query(
        f"department IN ('{user_dept}', 'all')"
    )
    
    return [doc.id for doc in allowed]

def search_with_access(query, user):
    """Search with access control"""
    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
    )

Ciclo de Retroalimentación

# feedback.py
from pydantic import BaseModel

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

def record_feedback(feedback: Feedback):
    """Store user feedback"""
    db.feedback.insert({
        "query_id": feedback.query_id,
        "helpful": feedback.helpful,
        "comment": feedback.comment,
        "timestamp": datetime.utcnow()
    })

def analyze_feedback():
    """Identify improvement areas"""
    # Low-rated queries indicate retrieval/generation issues
    poor_queries = db.feedback.find({"helpful": False})
    
    for query in poor_queries:
        # Log for review
        # Potentially retrain or adjust pipeline
        pass

Funcionalidades Clave

Funcionalidad	Implementación
Multi-fuente	Servicio de sincronización
Autenticación	JWT + RBAC
Control de acceso	Filtrado por metadatos
Versionado	Metadatos de documentos
Retroalimentación	Calificación + comentarios