Chapitre 3 : Application RAG Multi-Tenant

Aperçu du projet

Construire une plateforme SaaS RAG desservant plusieurs organisations :

• Isolation complète des locataires
• Personnalisation par locataire
• Suivi de l'utilisation et facturation
• Infrastructure évolutive

Architecture

Isolation des locataires

# tenant_service.py
class TenantService:
    def __init__(self, vectorstore_factory, config_db):
        self.vectorstore_factory = vectorstore_factory
        self.config_db = config_db
    
    def get_tenant_retriever(self, tenant_id):
        """Get retriever scoped to tenant"""
        config = self.config_db.get(tenant_id)
        
        # Option 1: Namespace isolation
        return self.vectorstore.as_retriever(
            search_kwargs={
                "namespace": f"tenant_{tenant_id}",
                "k": config.get("top_k", 5)
            }
        )
        
        # Option 2: Metadata filter
        # return retriever with filter={"tenant_id": tenant_id}
    
    def get_tenant_llm(self, tenant_id):
        """Get LLM with tenant-specific config"""
        config = self.config_db.get(tenant_id)
        
        return ChatOpenAI(
            model=config.get("model", "gpt-3.5-turbo"),
            temperature=config.get("temperature", 0),
            max_tokens=config.get("max_tokens", 500)
        )

Suivi de l'utilisation

# usage_tracking.py
from datetime import datetime

class UsageTracker:
    def __init__(self, db):
        self.db = db
    
    def track_query(self, tenant_id, query, tokens_used):
        """Track API usage"""
        self.db.usage.insert({
            "tenant_id": tenant_id,
            "timestamp": datetime.utcnow(),
            "type": "query",
            "tokens": tokens_used
        })
    
    def track_indexing(self, tenant_id, documents_count, chunks_count):
        """Track indexing usage"""
        self.db.usage.insert({
            "tenant_id": tenant_id,
            "timestamp": datetime.utcnow(),
            "type": "indexing",
            "documents": documents_count,
            "chunks": chunks_count
        })
    
    def get_monthly_usage(self, tenant_id, month):
        """Get usage for billing"""
        return self.db.usage.aggregate([
            {"$match": {"tenant_id": tenant_id, "month": month}},
            {"$group": {
                "_id": "$type",
                "total_tokens": {"$sum": "$tokens"},
                "total_docs": {"$sum": "$documents"}
            }}
        ])

Intégration de la facturation

# billing.py
class BillingService:
    PRICING = {
        "queries": 0.001,      # per query
        "tokens": 0.00001,     # per token
        "storage": 0.10,       # per GB/month
    }
    
    def calculate_invoice(self, tenant_id, month):
        usage = usage_tracker.get_monthly_usage(tenant_id, month)
        
        invoice = {
            "tenant_id": tenant_id,
            "period": month,
            "line_items": []
        }
        
        for item in usage:
            if item["_id"] == "query":
                cost = item["count"] * self.PRICING["queries"]
                invoice["line_items"].append({
                    "description": f"{item['count']} queries",
                    "amount": cost
                })
        
        invoice["total"] = sum(i["amount"] for i in invoice["line_items"])
        return invoice

Personnalisation des locataires

Fonctionnalité	Options par locataire
Modèle	GPT-3.5, GPT-4, Claude
Température	0-1
Top-K	3-10
Taille des morceaux	500-2000
Prompts personnalisés	Oui
Image de marque	Logo, couleurs

Considérations de déploiement

• Kubernetes pour l'évolutivité
• Bases de données séparées par niveau
• Limitation de débit par locataire
• Surveillance de la santé par locataire
• Sauvegardes automatisées