Capítulo 3: Aplicación RAG Multi-Tenant

Descripción general del proyecto

Construya una plataforma SaaS RAG que sirva a múltiples organizaciones:

• Aislamiento completo de inquilinos
• Personalización por inquilino
• Seguimiento de uso y facturación
• Infraestructura escalable

Arquitectura

Aislamiento de inquilinos

# 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)
        )

Seguimiento de uso

# 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"}
            }}
        ])

Integración de facturación

# 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

Personalización por inquilino

Funcionalidad	Opciones por inquilino
Modelo	GPT-3.5, GPT-4, Claude
Temperatura	0-1
Top-K	3-10
Tamaño de fragmento	500-2000
Prompts personalizados	Sí
Branding	Logo, colores

Consideraciones de despliegue

• Kubernetes para escalabilidad
• Bases de datos separadas por nivel
• Limitación de tasa por inquilino
• Monitoreo de salud por inquilino
• Copias de seguridad automatizadas