Capítulo 3: Aplicação RAG Multi-Tenant

Visão Geral do Projeto

Construa uma plataforma SaaS RAG atendendo múltiplas organizações:

• Isolamento completo de locatários
• Customização por locatário
• Rastreamento de uso e cobrança
• Infraestrutura escalável

Arquitetura

Isolamento de Locatários

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

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

Integração de Cobrança

# 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

Customização por Locatário

Recurso	Opções por Locatário
Modelo	GPT-3.5, GPT-4, Claude
Temperatura	0-1
Top-K	3-10
Tamanho do chunk	500-2000
Prompts customizados	Sim
Branding	Logo, cores

Considerações de Deploy

• Kubernetes para escalabilidade
• Bancos de dados separados por nível
• Rate limiting por locatário
• Monitoramento de saúde por locatário
• Backups automatizados