DEV Community: Martin Palopoli

Cache semántico y FAQ matching: cómo reduje un 40% el costo de LLM en mi motor RAG

Martin Palopoli — Tue, 07 Apr 2026 13:10:00 +0000

Cada query RAG cuesta dinero: embedding + tokens de LLM. Implementé tres capas de optimización que reducen el costo real un 30-45%: FAQ matching (respuestas curadas a costo cero), cache semántico (pgvector con similaridad >= 0.95), y auto-generación de FAQs desde queries frecuentes sin respuesta. Todo con código de producción y un fallback que mantiene el servicio activo cuando el presupuesto de LLM se agota.

El problema: cada query cuesta dinero real

En los artículos anteriores construí un pipeline RAG con búsqueda híbrida, reranking y streaming. Funciona bien. Pero en producción:

Query típica:  embedding + búsqueda + LLM (~800 tokens) ≈ $0.0003
1,000 queries/día × 30 días = 30,000 queries/mes
30,000 × $0.0003 = ~$9/mes por tenant
50 tenants = ~$450/mes solo en LLM

El insight clave: el 35-40% de las queries son repetidas o muy similares. Cada una es dinero quemado.

Arquitectura de las 3 capas de ahorro

Query del usuario
     │
     ▼
┌─────────────┐
│  FAQ Match   │──→ Si score ≥ 0.85: respuesta curada (costo LLM = $0)
└─────┬───────┘
      │ No match
      ▼
┌─────────────────┐
│ Semantic Cache   │──→ Si similaridad ≥ 0.95: respuesta cacheada ($0)
└─────┬───────────┘
      │ Cache miss
      ▼
┌─────────────────┐
│ Budget Check     │──→ Si presupuesto agotado: fallback FAQ-only
└─────┬───────────┘
      │ Budget OK
      ▼
  Pipeline RAG completo → fire-and-forget: guardar en cache

Capa 1: FAQ Matching — la inversión más rentable

async def match_faq(
    db: AsyncSession, query: str, kb_ids: list[UUID], threshold: float = 0.85,
) -> dict | None:
    query_embedding = embedding_service.embed_query(query)
    kb_ids_str = ",".join(f"'{str(kb_id)}'" for kb_id in kb_ids)

    stmt = text(f"""
        SELECT id, question, answer, attachments,
               1 - (embedding <=> CAST(:embedding AS vector)) as score
        FROM faqs
        WHERE knowledge_base_id IN ({kb_ids_str})
          AND is_active = true AND embedding IS NOT NULL
        ORDER BY embedding <=> CAST(:embedding AS vector)
        LIMIT 1
    """)
    result = await db.execute(stmt, {"embedding": str(query_embedding)})
    row = result.mappings().first()

    if not row or float(row["score"]) < threshold:
        return None

    # Increment hit_count atomically
    await db.execute(
        update(FAQ).where(FAQ.id == row["id"]).values(hit_count=FAQ.hit_count + 1)
    )
    await db.commit()
    return {"faq_id": str(row["id"]), "question": row["question"],
            "answer": row["answer"], "score": float(row["score"])}

El modelo usa un índice HNSW sobre el embedding para que la búsqueda sea ~15ms:

class FAQ(Base):
    __tablename__ = "faqs"
    embedding = mapped_column(Vector(384), nullable=True)
    hit_count = mapped_column(Integer, default=0, server_default="0")

    __table_args__ = (
        Index("ix_faqs_embedding", "embedding",
              postgresql_using="hnsw",
              postgresql_ops={"embedding": "vector_cosine_ops"},
              postgresql_with={"m": 16, "ef_construction": 64}),
    )

Threshold tuning: de 0.75 a 0.85

Empecé con 0.75. Demasiado bajo — obtuve false positives donde "¿Cómo configuro el widget?" matcheaba con "¿Qué es un widget?". Subí a 0.85 y los falsos positivos desaparecieron:

Query	FAQ almacenada	Score	Match?
"como reseteo mi clave"	"¿Cómo cambio mi contraseña?"	0.91	Si
"aceptan visa?"	"¿Qué medios de pago aceptan?"	0.87	Si
"como configuro el widget"	"¿Qué es un widget?"	0.72	No

Cuando hay match, el pipeline completo no se ejecuta:

if faq_match:
    yield {"event": "faq_match", "data": json.dumps(faq_match)}
    # FAQ matches are free — don't increment billing counters
    fire_and_forget_log_usage(
        query_type="faq", response_tokens=0, provider=None, ...
    )

response_tokens=0 y query_type="faq" — fundamental para las métricas de ahorro.

Capa 2: Cache Semántico — pgvector como cache inteligente

El concepto

Si alguien pregunta "¿Cómo reseteo mi contraseña?" y hace 2 horas otro preguntó "¿Cómo puedo cambiar mi password?", la respuesta es la misma. El cache semántico detecta esa equivalencia comparando embeddings, no strings exactos.

El modelo almacena el embedding del query, la respuesta, las fuentes, y un hash de la configuración RAG:

class ResponseCache(Base):
    __tablename__ = "response_cache"

    query_embedding = mapped_column(Vector(384), nullable=False)
    query_text = mapped_column(String(500), nullable=False)
    response_text = mapped_column(Text, nullable=False)
    sources = mapped_column(JSONB, nullable=False, default=list)
    confidence = mapped_column(Float, nullable=False)
    knowledge_base_ids = mapped_column(ARRAY(UUID(as_uuid=True)), nullable=False)
    rag_config_hash = mapped_column(String(64), nullable=False)
    hit_count = mapped_column(Integer, default=0, server_default="0")
    expires_at = mapped_column(DateTime(timezone=True), nullable=False)

Config hash: scoping por configuración

Si el admin cambia parámetros de RAG, las respuestas cacheadas ya no son válidas:

def compute_config_hash(retrieval_config, language: str | None = None) -> str:
    key_fields = {
        "candidate_k": retrieval_config.candidate_k,
        "rerank_top_n": retrieval_config.rerank_top_n,
        "top_k": retrieval_config.top_k,
        "lambda_param": retrieval_config.lambda_param,
        "max_per_doc": retrieval_config.max_per_doc,
        "bm25_weight": retrieval_config.bm25_weight,
        "language": language or "es",
    }
    return hashlib.sha256(json.dumps(key_fields, sort_keys=True).encode()).hexdigest()

Lookup: similaridad >= 0.95

CACHE_SIMILARITY_THRESHOLD = 0.95

async def lookup_cache(db, query_embedding, kb_ids, rag_config_hash, tenant_id):
    stmt = text(f"""
        SELECT id, response_text, sources, confidence,
               1 - (query_embedding <=> CAST(:embedding AS vector)) AS similarity
        FROM response_cache
        WHERE tenant_id = :tid AND expires_at > now()
          AND rag_config_hash = :config_hash
          AND knowledge_base_ids @> {kb_array}
        ORDER BY query_embedding <=> CAST(:embedding AS vector)
        LIMIT 1
    """)
    row = result.mappings().first()
    if not row or float(row["similarity"]) < CACHE_SIMILARITY_THRESHOLD:
        return None

    await db.execute(
        text("UPDATE response_cache SET hit_count = hit_count + 1 WHERE id = :cid"),
        {"cid": str(row["id"])},
    )
    return {"response_text": row["response_text"], "sources": row["sources"],
            "confidence": float(row["confidence"])}

¿Por qué 0.95? Con 0.90 tuve cache hits incorrectos:

Query A	Query B	Similaridad	¿Misma respuesta?
"Cómo exporto a CSV"	"Cómo descargo datos en CSV"	0.97	Si
"Cómo configuro el webhook"	"Cómo pruebo el webhook"	0.92	No
"Cómo agrego usuarios"	"Cómo elimino usuarios"	0.91	No

Fire-and-forget storage

No podés bloquear el streaming SSE para guardar en cache:

def fire_and_forget_store_cache(tenant_id, query_embedding, query_text,
                                 kb_ids, rag_config_hash, response_text,
                                 sources, confidence, ttl_hours=168):
    async def _store():
        try:
            async with async_session() as db:
                await store_cache(db, tenant_id, query_embedding, ...)
        except Exception as e:
            logger.error("Failed to store cache: %s", e)

    try:
        loop = asyncio.get_running_loop()
        loop.create_task(_store())
    except RuntimeError:
        pass  # No event loop (tests) — skip

Y la condición para cachear:

if rag_config.retrieval.cache_enabled and kb_ids and not low_confidence:
    fire_and_forget_store_cache(...)

not low_confidence: no cacheamos respuestas malas.

Invalidación proactiva: el cache que no miente

async def invalidate_kb_cache(db: AsyncSession, kb_id: UUID) -> int:
    result = await db.execute(
        text("DELETE FROM response_cache WHERE CAST(:kb_id AS UUID) = ANY(knowledge_base_ids)"),
        {"kb_id": str(kb_id)},
    )
    await db.commit()
    return result.rowcount

Se llama automáticamente en cada operación que cambia contenido de una KB:

# En faq_service.py — al crear, actualizar o importar FAQs
async def create_faq(db, kb_id, data):
    faq = FAQ(knowledge_base_id=kb_id, ...)
    db.add(faq)
    await db.commit()
    await invalidate_kb_cache(db, kb_id)  # Cache muere
    return faq

Lo mismo ocurre al subir o reprocesar documentos. Regla simple: si el contenido de una KB cambió, el cache de esa KB muere.

Auto-FAQ: convertir preguntas frecuentes en FAQs reales

El sistema registra queries con baja confianza. Cuando una aparece múltiples veces, el admin puede auto-generar una FAQ:

async def generate_faq_suggestion(db, unanswered_query_id, kb_id, tenant_id):
    uq = await db.get(UnansweredQuery, unanswered_query_id)

    # Dedup: skip si ya hay sugerencia pendiente o FAQ similar (>= 0.85)
    similar_faq = await db.execute(text("""
        SELECT 1 - (embedding <=> CAST(:embedding AS vector)) AS score
        FROM faqs WHERE knowledge_base_id = :kb_id AND is_active = true
        ORDER BY embedding <=> CAST(:embedding AS vector) LIMIT 1
    """), ...)
    if faq_row and float(faq_row["score"]) >= 0.85:
        return None  # Similar FAQ already exists

    # Search KB for context, then call LLM
    sources = await search_chunks(db, uq.query_text, [kb_id], ...)
    result = await generate_playground_response(
        query=uq.query_text, sources=sources[:5],
        temperature=0.2, max_tokens=512,
        system_prompt="Responde SOLO con información del contexto. "
                      "2-4 oraciones. Si no hay info suficiente: NO_ANSWER"
    )
    if not result.get("response") or result["response"].strip() == "NO_ANSWER":
        return None

    return SuggestedFAQ(
        tenant_id=tenant_id, knowledge_base_id=kb_id,
        question=uq.query_text, generated_answer=result["response"].strip(),
        embedding=query_embedding, status="pending",
    )

Batch generation prioriza por frecuencia:

async def batch_generate_suggestions(db, tenant_id, kb_id, limit=5):
    stmt = text("""
        SELECT uq.id FROM unanswered_queries uq
        WHERE uq.tenant_id = :tid AND uq.resolved = false
          AND NOT EXISTS (
              SELECT 1 FROM suggested_faqs sf
              WHERE sf.source_query_id = uq.id AND sf.status = 'pending'
          )
        ORDER BY uq.occurrence_count DESC LIMIT :lim
    """)
    # Generate suggestions for each...

Al aprobar, se crea la FAQ real, se invalida el cache, y la query se marca como resuelta. Ciclo cerrado.

Fallback FAQ-only: servicio degradado sin corte

Plan Free: 50 queries de IA/mes. Cuando se agotan, las FAQs siguen activas:

async def is_llm_budget_exhausted(db, tenant_id) -> bool:
    """Soft check — enables FAQ-only fallback, does NOT raise."""
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return False
    usage = await get_monthly_usage(db, tenant_id)
    if plan.max_messages_month != -1 and usage["messages_month"] >= plan.max_messages_month:
        return True
    return False

En el flujo principal:

faq_only_mode = await is_llm_budget_exhausted(db, tenant_id)

# FAQ matching funciona siempre
if faq_match:
    return faq_response(faq_match)  # Gratis

# Sin match + sin presupuesto → upgrade card
if faq_only_mode:
    yield {"event": "upgrade_required", "data": json.dumps({
        "message": "Has alcanzado el límite de consultas de IA. "
                   "Las FAQs siguen disponibles sin costo."
    })}

Métricas de costo

async def _get_cost_metrics(db, tenant_id, cutoff):
    # Tokens by provider
    # ...configurable prices from settings...
    groq_price = await settings_service.get(db, "cost.groq.price_per_1k_tokens", 0.00027)

    # FAQ savings estimation
    avg_llm_tokens = total_tokens / max(llm_query_count, 1)
    estimated_faq_tokens_saved = int(faq_count * avg_llm_tokens)
    estimated_faq_cost_saved = estimated_faq_tokens_saved / 1000 * groq_price

    return {
        "by_provider": by_provider,
        "total_estimated_cost": round(total_cost, 4),
        "faq_savings": {
            "queries_without_llm": faq_count,
            "estimated_tokens_saved": estimated_faq_tokens_saved,
            "estimated_cost_saved": round(estimated_faq_cost_saved, 4),
        },
        "avg_cost_per_conversation": round(total_cost / conv_count, 6),
    }

Cache hit rate como 7ma stat card en el dashboard:

async def get_cache_stats(db, tenant_id):
    rate_result = await db.execute(text("""
        SELECT
            COUNT(*) FILTER (WHERE query_type = 'cached') AS cached_queries,
            COUNT(*) FILTER (WHERE query_type IN ('chat','widget','cached')) AS total_queries
        FROM usage_logs WHERE tenant_id = :tid
    """))
    return {"hit_rate": round(cached / total, 3), ...}

El flujo completo en el endpoint de chat

Para ver cómo encajan las 3 capas, este es el orden real en chat.py:

# 1. Budget check (soft — no lanza excepción)
faq_only_mode = await is_llm_budget_exhausted(db, tenant_id)

# 2. FAQ match (funciona siempre, incluso en faq_only_mode)
faq_match = await match_faq(db, data.content, kb_ids, threshold=0.85)
if faq_match:
    return EventSourceResponse(faq_event_generator())  # $0

# 3. Budget exhausted + no FAQ match → upgrade card
if faq_only_mode:
    yield {"event": "upgrade_required", ...}
    return

# 4. Semantic cache lookup
query_embedding = embedding_service.embed_query(effective_query)
config_hash = compute_config_hash(rag_config.retrieval, detected_lang)
cache_hit = await lookup_cache(db, query_embedding, kb_ids, config_hash, tenant_id)
if cache_hit:
    return EventSourceResponse(cache_event_generator())  # $0

# 5. Full RAG pipeline (vector + BM25 + rerank + LLM)
# ... streaming response ...

# 6. Fire-and-forget: store in cache for next time
if not low_confidence:
    fire_and_forget_store_cache(...)

Números reales

Métrica	Valor
Threshold FAQ	0.85
Threshold cache	0.95
TTL cache default	7 días
Latencia FAQ match	~15ms
Latencia cache lookup	~20ms
Latencia pipeline completo	~2-4s
% queries resueltas por FAQ	15-25%
% queries resueltas por cache	10-20%
% queries que llegan al LLM	55-75%
Ahorro estimado total	30-45%

Lecciones aprendidas

1. Las FAQs son la mejor inversión

No es sexy. Es una tabla con preguntas y respuestas. Pero cada FAQ es una respuesta perfecta servida en 15ms a costo cero, para siempre.

2. El threshold del cache debe ser muy alto

Con 0.90 tuve falsos positivos que sirvieron respuestas incorrectas. 0.95 es el mínimo seguro.

3. Fire-and-forget es obligatorio para cache storage

El primer intento fue síncrono. Si el INSERT tarda, el último token del streaming se demora. Fire-and-forget elimina esa latencia.

4. La invalidación proactiva vale la pena

Es tentador dejar que el cache expire solo (TTL). Pero el admin que sube un documento espera respuestas actualizadas inmediatamente.

5. Auto-FAQ cierra el ciclo

Sin auto-FAQ, las knowledge gaps se acumulan. Con auto-FAQ, en 2 clicks la pregunta frecuente pasa de gap a FAQ activa.

6. Fallback FAQ-only > cortar el servicio

Los usuarios free siguen haciendo preguntas que matchean FAQs. Reduce churn y da incentivo real para upgradear.

7. Los precios de LLM deben ser configurables

Hardcodear $0.00027/1K tokens es una trampa. Los precios cambian. Guardarlos en una tabla de settings permite ajustarlos sin deploy.

Conclusión

Optimizar costos en RAG es un problema de evitar trabajo innecesario. Las tres capas (FAQ, cache, auto-FAQ) atacan el mismo principio: si la respuesta ya existe, no pagues por generarla de nuevo.

Lo interesante: también mejoran la experiencia. FAQ match en 15ms vs 2-4s del pipeline. Cache hit en 20ms con la misma calidad. Y pgvector ya estaba en el stack — no necesitás Redis ni Elasticsearch.

Este es el quinto artículo de la serie. Si te sirvió, un like ayuda a que llegue a más personas. ¿Tenés preguntas sobre cache semántico o FAQ matching? Dejá un comentario.

How I Implemented End-to-End SSE Streaming: From LLM to Browser Through Nginx

Martin Palopoli — Tue, 07 Apr 2026 13:10:00 +0000

I implemented end-to-end SSE streaming for a RAG engine: an async generator in FastAPI emits custom events (token, sources, confidence, done, error, etc.), the frontend consumes them with fetch + ReadableStream, an embeddable widget in vanilla JS does the same inside a closed Shadow DOM, and Nginx needs 6 specific directives to not ruin everything. Here's the real code, the event protocol, and the lessons learned.

Why SSE and Not WebSockets

When you need streaming in a RAG app, the first question is: WebSockets or SSE.

For an LLM chat, the answer is clear: SSE.

Criterion	SSE	WebSockets
Flow direction	Server → client (unidirectional)	Bidirectional
Protocol	Standard HTTP	Custom protocol (ws://)
Reconnection	Automatic (built-in)	Manual
Proxy/CDN compat	Excellent (it's HTTP)	Problematic
Server complexity	Low (async generator)	High (state management)
Auth	Normal HTTP headers	Hack in query params or first message

A RAG chat is fundamentally unidirectional during streaming: the user sends a message (a normal POST) and the server responds with a stream of tokens. You don't need a permanent bidirectional channel.

Also, SSE travels over standard HTTP. That means it works with any reverse proxy, CDN, load balancer, and firewall without special configuration (beyond disabling buffering, which we'll cover).

With WebSockets you'd need to handle: manual reconnection, handshake authentication, connection state, heartbeats, and proxies that close idle connections. All of that is complexity that adds nothing in this use case.

The Stack

┌─────────────┐     POST /messages     ┌─────────────┐
│   Browser    │ ──────────────────────►│   Nginx     │
│  (Vue/Widget)│                        │  (reverse   │
│              │◄──── SSE stream ───────│   proxy)    │
└─────────────┘                        └──────┬──────┘
                                              │
                                       ┌──────▼──────┐
                                       │   FastAPI    │
                                       │  (uvicorn)   │
                                       │              │
                                       │  async gen   │──► Groq/OpenAI/Ollama
                                       │  → SSE       │──► PostgreSQL
                                       │              │──► Redis
                                       └─────────────┘

Backend: The Async Generator That Emits SSE

The Base: EventSourceResponse

FastAPI doesn't have native SSE support, but sse-starlette solves it in one line:

from sse_starlette.sse import EventSourceResponse

@router.post("/conversations/{conv_id}/messages")
async def send_message(conv_id: UUID, data: MessageCreate, ...):
    # ... validations, RAG search, etc.

    async def event_generator():
        async for event in _build_stream(conv_id, data, ...):
            yield event

    return EventSourceResponse(event_generator())

EventSourceResponse takes an async generator and converts it to an SSE stream with the correct format:

event: token
data: {"content": "The "}

event: token
data: {"content": "answer "}

event: token
data: {"content": "is..."}

event: done
data: {"message_id": "abc-123"}

Each yield from the generator becomes an event: + data: block separated by \n\n.

The Event Protocol

I designed a protocol with 10 event types, each with a specific JSON payload:

# SSE Protocol Events
# ─────────────────────────────────────────
# user_message_id  → Real ID of message saved in DB
# sources          → Chunks retrieved from RAG pipeline
# confidence       → Confidence score (0-1) from reranking
# token            → Text fragment from LLM
# faq_match        → Direct FAQ answer (no LLM)
# searching        → Agentic search in progress (round N)
# suggestions      → Suggested follow-up questions
# error            → Error during streaming
# upgrade_required → Plan limit reached
# content_blocked  → Content blocked by rules
# done             → Stream finished, with message_id

Why typed events and not just data:? Because the frontend needs to know what it's receiving before parsing it. A token renders differently than a source, an error is handled differently than a confidence event. Without types, the frontend would have to guess the JSON content.

The Full Generator

Here's the real generator (simplified for readability, but with the exact structure):

async def event_generator():
    # Track concurrent stream with Redis
    await increment_concurrent(str(tenant_id))
    try:
        async for event in _event_generator_inner():
            yield event
    finally:
        # ALWAYS decrement, even if client disconnects
        await decrement_concurrent(str(tenant_id))


async def _event_generator_inner():
    # 1. Real user message ID (to sync with optimistic UI)
    yield {
        "event": "user_message_id",
        "data": json.dumps({"message_id": str(user_msg.id)})
    }

    # 2. Sources retrieved from RAG pipeline
    yield {
        "event": "sources",
        "data": json.dumps({"sources": sources})
    }

    # 3. Confidence score
    yield {
        "event": "confidence",
        "data": json.dumps({"score": round(confidence, 3)})
    }

    # 4. LLM tokens (the heart of streaming)
    full_response = ""
    buffer = ""

    stream = stream_chat_response(query, sources, history, provider, ...)

    if inspect.isasyncgen(stream):
        async for token in stream:
            buffer += token
            full_response += token

            # Buffer for agentic search: don't emit if [SEARCH:] might be coming
            if buffer and '[' not in buffer[-20:]:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
                buffer = ""
        else:
            # Flush final buffer
            if buffer:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
    else:
        # Synchronous stream (Groq, OpenAI)
        for token in stream:
            buffer += token
            full_response += token
            if buffer and '[' not in buffer[-20:]:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
                buffer = ""
        else:
            if buffer:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }

    # 5. Follow-up suggestions (post-stream)
    suggestion_list = await generate_suggestions(full_response, sources, provider)
    if suggestion_list:
        yield {
            "event": "suggestions",
            "data": json.dumps({"suggestions": suggestion_list})
        }

    # 6. Save message and signal completion
    msg = await chat_service.save_message(db, conv_id, MessageRole.ASSISTANT, full_response, ...)

    yield {
        "event": "done",
        "data": json.dumps({"message_id": str(msg.id)})
    }

The Buffer Detail for Agentic Search

The system supports agentic search: the LLM can emit [SEARCH: new query] mid-response to request more information. The problem: if you emit tokens one by one, the frontend could render a partial [SEARCH: as visible text.

The solution is a buffer that holds the last 20 characters if they contain [:

if buffer and '[' not in buffer[-20:]:
    yield {"event": "token", "data": json.dumps({"content": buffer})}
    buffer = ""

If the full marker [SEARCH: ...] is detected, the stream is interrupted, a new search is executed, and generation continues with expanded context:

match = SEARCH_PATTERN.search(accumulated)
if match and search_round < MAX_SEARCH_ROUNDS:
    search_query = match.group(1).strip()
    search_round += 1

    yield {"event": "searching", "data": json.dumps({
        "query": search_query, "round": search_round
    })}

    # Execute new search with the LLM's query
    new_sources = await search_chunks(db, search_query, kb_ids, rag_config=rag_config)

    # Deduplicate sources
    existing_ids = {s["chunk_id"] for s in all_sources}
    unique_new = [s for s in new_sources if s["chunk_id"] not in existing_ids]

    if unique_new:
        yield {"event": "sources", "data": json.dumps({
            "sources": unique_new, "round": search_round
        })}

    # Continue generation with expanded context
    # (max 3 search rounds)

Short-Circuits: FAQ, Cache and Upgrade

Before the full RAG pipeline, there are several "shortcuts" that return a different EventSourceResponse:

# FAQ match → instant response without LLM
if faq_match:
    async def faq_event_generator():
        yield {"event": "user_message_id", "data": json.dumps({"message_id": str(user_msg.id)})}
        yield {"event": "faq_match", "data": json.dumps(faq_match)}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id)})}
    return EventSourceResponse(faq_event_generator())

# Cache hit → cached response, simulate streaming
if cache_hit:
    async def cache_event_generator():
        yield {"event": "sources", "data": json.dumps({"sources": cache_hit["sources"]})}
        yield {"event": "confidence", "data": json.dumps({"score": cache_hit["confidence"]})}
        # Simulate streaming word by word
        for token in cache_hit["response_text"].split(" "):
            yield {"event": "token", "data": json.dumps({"content": token + " "})}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id), "cached": True})}
    return EventSourceResponse(cache_event_generator())

# Budget exhausted → upgrade signal
if faq_only_mode:
    async def upgrade_event_generator():
        yield {"event": "upgrade_required", "data": json.dumps({
            "message": "You've reached your AI query limit for this plan."
        })}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id)})}
    return EventSourceResponse(upgrade_event_generator())

Why simulate streaming for cache hits? Because the frontend has a single rendering path. If cache returned all the text at once, it would need special logic. Simulating streaming keeps the frontend code simple.

Streaming from LLM Providers

Each provider has its own format. The abstraction layer unifies everything into generators that yield strings:

# Groq and OpenAI: SYNCHRONOUS generator (SDK compatible)
def _stream_groq(messages, temperature=0.3, max_tokens=4096):
    stream = groq_client.chat.completions.create(
        model=settings.groq_model, messages=messages,
        stream=True, temperature=temperature, max_tokens=max_tokens,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

# Ollama: ASYNC generator (NDJSON over HTTP streaming)
async def _stream_ollama(messages, temperature=0.3, max_tokens=4096):
    async with httpx.AsyncClient(timeout=httpx.Timeout(300.0)) as client:
        async with client.stream("POST", f"{settings.ollama_url}/api/chat",
                                 json={"model": settings.ollama_model,
                                       "messages": messages, "stream": True}) as response:
            async for line in response.aiter_lines():
                data = json.loads(line)
                content = data.get("message", {}).get("content", "")
                if content:
                    yield content
                if data.get("done"):
                    break

The key detail: Groq/OpenAI return synchronous generators, Ollama returns async. The SSE generator needs to handle both with inspect.isasyncgen(). It's ugly, but wrapping sync in async adds complexity and indirection that makes debugging harder.

Frontend: Consuming with fetch + ReadableStream

Why Not EventSource

The browser's EventSource API has a fatal flaw for this use case: it only supports GET. Our chat endpoint is a POST with JSON body (message content, knowledge base IDs, provider, etc.).

The alternative: fetch + ReadableStream. More code, but total control.

The SSE Parser

export async function sendMessage(
  convId: string,
  content: string,
  knowledgeBaseIds: string[],
  callbacks: StreamCallbacks,
  provider?: string,
): Promise<void> {
  const token = localStorage.getItem('access_token')

  const response = await fetch(`/api/v1/conversations/${convId}/messages`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      ...(token ? { Authorization: `Bearer ${token}` } : {}),
    },
    body: JSON.stringify({
      content,
      knowledge_base_ids: knowledgeBaseIds,
      provider,
    }),
  })

  if (!response.ok) {
    // Detect plan limit error (403)
    if (response.status === 403) {
      const errorData = await response.json()
      if (errorData.detail?.includes('limit')) {
        callbacks.onError(`__PLAN_LIMIT__:${errorData.detail}`)
        return
      }
    }
    callbacks.onError(`Error: ${response.status}`)
    return
  }

  const reader = response.body?.getReader()
  if (!reader) {
    callbacks.onError('No response stream')
    return
  }

  const decoder = new TextDecoder()
  let buffer = ''
  let receivedDone = false

  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) break

      buffer += decoder.decode(value, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop() || ''  // Last incomplete line → to buffer

      let currentEvent = ''
      for (const line of lines) {
        if (line.startsWith('event: ')) {
          currentEvent = line.slice(7).trim()
          continue
        }
        if (line.startsWith('data: ')) {
          const parsed = JSON.parse(line.slice(6))

          // Dispatch by event type
          if (currentEvent === 'token') {
            callbacks.onToken(parsed.content)
          } else if (currentEvent === 'sources') {
            callbacks.onSources(parsed.sources)
          } else if (currentEvent === 'confidence') {
            callbacks.onConfidence?.(parsed.score)
          } else if (currentEvent === 'faq_match') {
            callbacks.onFaqMatch?.(parsed)
          } else if (currentEvent === 'upgrade_required') {
            callbacks.onUpgradeRequired?.(parsed.message)
          } else if (currentEvent === 'content_blocked') {
            callbacks.onContentBlocked?.(parsed.response, parsed.type)
          } else if (currentEvent === 'error') {
            callbacks.onError(parsed.error)
          } else if (currentEvent === 'done' || 'message_id' in parsed) {
            receivedDone = true
            callbacks.onDone(parsed.message_id)
          }
          currentEvent = ''
        }
      }
    }
  } catch (err) {
    callbacks.onError(`Connection error: ${err}`)
    return
  }

  // If stream ended without "done" event, something went wrong
  if (!receivedDone) {
    callbacks.onError('Connection was interrupted before completing the response')
  }
}

The Callbacks Interface

export interface StreamCallbacks {
  onSources: (sources: Source[]) => void
  onToken: (token: string) => void
  onCodeResult: (result: CodeResult) => void
  onSearching: (query: string, round: number) => void
  onExecuting: (count: number) => void
  onSuggestions: (suggestions: string[]) => void
  onUserMessageId: (messageId: string) => void
  onFaqMatch?: (match: FAQMatch) => void
  onConfidence?: (score: number) => void
  onUpgradeRequired?: (message: string) => void
  onContentBlocked?: (response: string, blockType: string) => void
  onDone: (messageId: string) => void
  onError: (error: string) => void
}

The useChat composable connects these callbacks with Vue's reactive state:

await api.sendMessage(conversationId, content, kbIds, {
  onToken: (token) => {
    streamingContent.value += token
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.content = streamingContent.value
    }
  },
  onSources: (s) => {
    sources.value = [...sources.value, ...s]
  },
  onConfidence: (score) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.confidenceScore = score
    }
  },
  onDone: (messageId) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.id = messageId
      last.sources = sources.value
    }
    streaming.value = false
  },
  onError: (err) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.content = `Error: ${err}`
    }
    streaming.value = false
  },
})

Optimistic UI with Reconciliation

A subtle detail: when the user sends a message, we add it immediately to the UI with a temporary ID (crypto.randomUUID()). But the real ID is generated by the database. The user_message_id event syncs both:

const userMsg: Message = {
  id: crypto.randomUUID(),  // Optimistic temporary ID
  role: 'user',
  content,
  ...
}
messages.value.push(userMsg)

// When the real ID arrives from the server:
onUserMessageId: (realId) => {
  userMsg.id = realId  // Replace temporary ID with DB ID
},

This matters for feedback: when the user gives a thumbs-up to a message, it needs the real ID so the PATCH reaches the correct message.

Embeddable Widget: SSE in Vanilla JS

The widget is ~970 lines of vanilla JS running inside a closed Shadow DOM (mode: 'closed'). The SSE parsing is identical to the TypeScript frontend, but with var instead of const and no types:

var reader = res.body.getReader();
var decoder = new TextDecoder();
var buffer = '';
var assistantContent = '';

while (true) {
  var readResult = await reader.read();
  if (readResult.done) break;
  buffer += decoder.decode(readResult.value, { stream: true });

  var lines = buffer.split('\n');
  buffer = lines.pop() || '';

  var event = null;
  for (var i = 0; i < lines.length; i++) {
    var line = lines[i].trim();
    if (line.startsWith('event:')) {
      event = line.slice(6).trim();
    } else if (line.startsWith('data:')) {
      var data = JSON.parse(line.slice(5).trim());

      if (event === 'session') {
        sessionToken = data.session_token;
        localStorage.setItem(STORAGE_KEY, sessionToken);
      } else if (event === 'token') {
        assistantContent += (data.content || '');
        messages[assistantMsgIndex].content = assistantContent;
        renderMessages();  // Re-render on each token
      } else if (event === 'faq_match') {
        messages.push({ role: 'assistant', content: data.answer, faqMatch: data });
        renderMessages();
      }
    }
  }
}

Key Differences from the Main App

Auth: The widget uses X-API-Key (SHA-256 hashed) instead of JWT. No login.
First event: Instead of user_message_id, the widget receives session with a token persisted in localStorage to maintain history across visits.
Closed Shadow DOM: host.attachShadow({ mode: 'closed' }) completely isolates styles and DOM from the host page. Not even document.querySelector can access the widget from outside.

Nginx: The 6 Directives That Make It Work

This is the section that caused me the most headaches. Without the correct Nginx configuration, SSE streaming doesn't work. Tokens accumulate in Nginx's buffer and arrive all at once at the end.

location /api/ {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;

    # === The 6 SSE directives ===
    proxy_buffering off;             # 1. Don't buffer the response
    proxy_cache off;                 # 2. Don't cache
    proxy_read_timeout 300s;         # 3. Long timeout (streams last minutes)
    proxy_set_header Connection '';  # 4. Disable upstream keep-alive
    proxy_http_version 1.1;          # 5. HTTP/1.1 (required for chunked)
    chunked_transfer_encoding off;   # 6. Disable Nginx's chunked encoding
}

What Each One Does

1. proxy_buffering off — The most important one. By default, Nginx buffers the complete upstream response before sending it to the client. With SSE, each event must reach the client immediately.

2. proxy_cache off — Nginx can cache upstream responses. A cached SSE stream is... a very bad idea.

3. proxy_read_timeout 300s — The default is 60 seconds. An LLM response with agentic search (3 rounds) can take 2-3 minutes. Without this timeout, Nginx closes the connection mid-response.

4. proxy_set_header Connection '' — Clears the Connection header to prevent Nginx from maintaining the upstream connection with keep-alive, which can cause data to be held back.

5. proxy_http_version 1.1 — HTTP/1.1 is required for chunked transfer-encoding. Nginx's default for upstream is HTTP/1.0.

6. chunked_transfer_encoding off — Seems contradictory to the previous point, but this disables chunked encoding from Nginx, not from upstream. Without this, Nginx can re-chunk the response and alter event timing.

The Symptom You'll See Without These Directives

If you're missing any of them, you'll see this behavior:

The user sends a message
The "loading" spinner stays for 5-30 seconds
Suddenly, the entire response appears at once
Streaming "works" on localhost (without Nginx) but not in production

If this happens to you, it's Nginx buffering. Add the 6 directives and it resolves immediately.

Cloudflare: The Other Invisible Proxy

If you use Cloudflare as proxy (orange cloud), there's a detail: Cloudflare can also buffer responses. But in practice, Cloudflare detects SSE automatically by the Content-Type: text/event-stream and disables buffering. No special configuration needed.

Redis: Stream Concurrency

A user could open 10 tabs and launch 10 simultaneous streams. Each stream consumes LLM resources, DB connections, and memory. Concurrency control uses Redis as an atomic counter:

_CONCURRENT_TTL = 300  # 5-minute TTL as safety net

async def increment_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0  # Fail-open: without Redis, allow

    redis_key = f"concurrent:{tenant_id}"
    pipe = client.pipeline()
    pipe.incr(redis_key)
    pipe.expire(redis_key, _CONCURRENT_TTL)
    results = await pipe.execute()
    return results[0]

async def decrement_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    count = await client.decr(redis_key)
    if count <= 0:
        await client.delete(redis_key)  # Clean dead keys
        return 0
    await client.expire(redis_key, _CONCURRENT_TTL)
    return count

The try/finally Pattern

Increment and decrement are in a try/finally to guarantee the counter decrements always, even if the client disconnects mid-stream:

async def event_generator():
    await increment_concurrent(str(tenant_id))
    try:
        async for event in _event_generator_inner():
            yield event
    finally:
        await decrement_concurrent(str(tenant_id))

TTL as Safety Net

What happens if the server crashes mid-stream? The decrement never executes and the counter stays inflated. The 5-minute TTL solves this: if it's not renewed with expire, the key self-deletes. The system self-heals.

Fail-Open and Rate Limiting

Two important principles:

Fail-open: If Redis is down, the rate limiter allows everything. I prefer serving requests without rate limiting to rejecting everything because Redis went down. Redis is a guard, not a gate.
Sliding window per API key: Besides per-tenant concurrency, each widget API key has its own rate limit using Redis sorted sets. Timestamps as scores, zremrangebyscore to clean expired entries, zcard to count — all in an atomic pipeline.

Error Handling: The 3 Layers

Backend: Always emit done after error. Without this, the frontend gets stuck in "streaming" state indefinitely:

except Exception as e:
    yield {"event": "error", "data": json.dumps({"error": f"LLM error: {e}"})}
    yield {"event": "done", "data": json.dumps({"message_id": "error"})}
    return

Frontend: Detect incomplete streams. If the reader returns done: true but we never received a done event, something went wrong (Nginx timeout, backend crash):

if (!receivedDone) {
  callbacks.onError('Connection was interrupted before completing the response')
}

Pre-stream: 403 errors (plan limit) arrive as normal HTTP before the SSE stream. The frontend uses a __PLAN_LIMIT__: prefix as internal convention so the composable distinguishes plan errors from generic errors and shows different UI (upgrade card vs. error message).

Real Numbers

Metric	Value
First token latency (Groq)	~200-400ms
First token latency (Ollama local)	~1-2s
Typical total response time	~3-8s
Total time with agentic search (3 rounds)	~15-30s
FAQ match (no LLM)	~15ms
Cache hit (simulated)	~50ms
Nginx SSE overhead	<1ms per event
Concurrent streams per tenant (Free plan)	2
API key rate limit (default)	30 req/min

Lessons Learned

1. SSE > WebSockets for LLM Streaming

Don't try to shove WebSockets where SSE is sufficient. For a unidirectional flow (server → client), SSE is simpler, more proxy-compatible, and easier to debug. The browser's automatic reconnection is a bonus.

2. Always Emit "done" at the End

Without an explicit termination event, the frontend can't distinguish between "the stream ended" and "the stream was cut off". Emitting done always — even after errors — is what makes the protocol robust.

3. Nginx Buffering Is the Silent Enemy

In development without Nginx everything works perfectly. In production, tokens arrive all at once. The first time it happened, I spent 2 hours thinking the problem was in the backend. It was 6 lines of Nginx.

4. Buffer for Inline Markers

If your LLM can emit special markers ([SEARCH:], executable code, etc.), you need a buffer that holds text until you're sure there's no partial marker. Without this, the user sees artifacts like [SEAR in the chat.

5. Try/Finally with Redis Counters

Every counter that's incremented before a stream must be decremented in a finally. Not in the happy path, not in the error handler: in finally. Clients disconnect without warning, servers crash, and generators get interrupted.

6. Fail-Open for Auxiliary Services

Redis for rate limiting should be fail-open. If Redis goes down, it's better to serve without rate limiting than to reject all requests. The same applies to logging, analytics, and any non-critical service.

7. fetch + ReadableStream > EventSource

EventSource is elegant but limited (GET only, minimal headers). fetch + ReadableStream requires more code but supports POST, custom headers (JWT), and fine-grained error handling. For a real-world case, the flexibility justifies the ~30 extra lines.

What's Next

Optional WebSocket upgrade: For future bidirectional features (typing indicators, presence), evaluate a selective upgrade without replacing SSE
Backpressure: If the frontend can't render as fast as the LLM generates, implement backpressure signaling
Compression: Evaluate SSE over HTTP/2 with frame compression to reduce bandwidth for high-traffic widgets

Conclusion

SSE for LLM streaming isn't hard to implement — but it's easy to implement poorly. The pitfalls are in the details: Nginx that buffers, generators that don't clean up counters, frontends that don't detect incomplete streams, and widgets that don't handle all event types.

The custom event protocol (token, sources, confidence, done, error, etc.) is what transforms a text stream into a usable system: the frontend knows what it's receiving and can render it correctly, show confidence indicators, accumulate sources from different search rounds, and handle edge cases like plan limits or blocked content.

Most importantly: always test through Nginx. What works on localhost without a proxy is not representative of production. Those 6 Nginx directives are the difference between "real token-by-token streaming" and "all text at once after 10 seconds".

Cómo implementé streaming SSE de extremo a extremo: desde el LLM hasta el navegador pasando por Nginx

Martin Palopoli — Tue, 31 Mar 2026 13:10:00 +0000

TL;DR

Implementé streaming SSE de extremo a extremo para un motor RAG: un generador async en FastAPI emite eventos custom (token, sources, confidence, done, error, etc.), el frontend los consume con fetch + ReadableStream, un widget embebible en vanilla JS hace lo mismo dentro de un Shadow DOM cerrado, y Nginx necesita 6 directivas específicas para no arruinar todo. Comparto el código real, el protocolo de eventos, y las lecciones aprendidas.

Por qué SSE y no WebSockets

Cuando necesitás streaming en una app RAG, la primera pregunta es: WebSockets o SSE.

Para un chat con LLM, la respuesta es clara: SSE.

Criterio	SSE	WebSockets
Dirección del flujo	Servidor → cliente (unidireccional)	Bidireccional
Protocolo	HTTP estándar	Protocolo propio (ws://)
Reconexión	Automática (built-in)	Manual
Proxy/CDN compat	Excelente (es HTTP)	Problemática
Complejidad server	Baja (generador async)	Alta (state management)
Auth	Headers HTTP normales	Hack en query params o primer mensaje

Un chat RAG es fundamentalmente unidireccional durante el streaming: el usuario envía un mensaje (un POST normal) y el servidor responde con un stream de tokens. No necesitás un canal bidireccional permanente.

Además, SSE viaja sobre HTTP estándar. Eso significa que funciona con cualquier reverse proxy, CDN, load balancer, y firewall sin configuración especial (más allá de desactivar buffering, que ya veremos).

Con WebSockets tenés que manejar: reconexión manual, autenticación en el handshake, estado de la conexión, heartbeats, y proxies que cierran conexiones idle. Todo eso es complejidad que no aporta nada en este caso de uso.

El stack

┌─────────────┐     POST /messages     ┌─────────────┐
│   Browser    │ ──────────────────────►│   Nginx     │
│  (Vue/Widget)│                        │  (reverse   │
│              │◄──── SSE stream ───────│   proxy)    │
└─────────────┘                        └──────┬──────┘
                                              │
                                       ┌──────▼──────┐
                                       │   FastAPI    │
                                       │  (uvicorn)   │
                                       │              │
                                       │  async gen   │──► Groq/OpenAI/Ollama
                                       │  → SSE       │──► PostgreSQL
                                       │              │──► Redis
                                       └─────────────┘

Backend: el generador async que emite SSE

La base: EventSourceResponse

FastAPI no tiene soporte nativo para SSE, pero sse-starlette lo resuelve con una línea:

from sse_starlette.sse import EventSourceResponse

@router.post("/conversations/{conv_id}/messages")
async def send_message(conv_id: UUID, data: MessageCreate, ...):
    # ... validaciones, búsqueda RAG, etc.

    async def event_generator():
        async for event in _build_stream(conv_id, data, ...):
            yield event

    return EventSourceResponse(event_generator())

EventSourceResponse toma un generador async y lo convierte en un stream SSE con el formato correcto:

event: token
data: {"content": "La "}

event: token
data: {"content": "respuesta "}

event: token
data: {"content": "es..."}

event: done
data: {"message_id": "abc-123"}

Cada yield del generador se convierte en un bloque event: + data: separado por \n\n.

El protocolo de eventos

Diseñé un protocolo con 10 tipos de eventos, cada uno con un payload JSON específico:

# Eventos del protocolo SSE
# ─────────────────────────────────────────
# user_message_id  → ID real del mensaje guardado en DB
# sources          → Chunks recuperados del RAG pipeline
# confidence       → Score de confianza (0-1) del reranking
# token            → Fragmento de texto del LLM
# faq_match        → Respuesta FAQ directa (sin LLM)
# searching        → Búsqueda agentic en progreso (round N)
# suggestions      → Preguntas de seguimiento sugeridas
# error            → Error durante el streaming
# upgrade_required → Límite de plan alcanzado
# content_blocked  → Contenido bloqueado por reglas
# done             → Stream finalizado, con message_id

¿Por qué eventos tipados y no solo data:? Porque el frontend necesita saber qué está recibiendo antes de parsearlo. Un token se renderiza diferente a una fuente, un error se maneja diferente a un evento de confianza. Sin tipos, el frontend tendría que adivinar el contenido del JSON.

El generador completo

Así se ve el generador real (simplificado para legibilidad, pero con la estructura exacta):

async def event_generator():
    # Trackear stream concurrente con Redis
    await increment_concurrent(str(tenant_id))
    try:
        async for event in _event_generator_inner():
            yield event
    finally:
        # SIEMPRE decrementar, incluso si el cliente desconecta
        await decrement_concurrent(str(tenant_id))


async def _event_generator_inner():
    # 1. ID real del mensaje del usuario (para sincronizar con el optimistic UI)
    yield {
        "event": "user_message_id",
        "data": json.dumps({"message_id": str(user_msg.id)})
    }

    # 2. Fuentes recuperadas del pipeline RAG
    yield {
        "event": "sources",
        "data": json.dumps({"sources": sources})
    }

    # 3. Score de confianza
    yield {
        "event": "confidence",
        "data": json.dumps({"score": round(confidence, 3)})
    }

    # 4. Tokens del LLM (el corazón del streaming)
    full_response = ""
    buffer = ""

    stream = stream_chat_response(query, sources, history, provider, ...)

    if inspect.isasyncgen(stream):
        async for token in stream:
            buffer += token
            full_response += token

            # Buffer para agentic search: no emitir si puede venir [SEARCH:]
            if buffer and '[' not in buffer[-20:]:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
                buffer = ""
        else:
            # Flush del buffer final
            if buffer:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
    else:
        # Stream sincrónico (Groq, OpenAI)
        for token in stream:
            buffer += token
            full_response += token
            if buffer and '[' not in buffer[-20:]:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }
                buffer = ""
        else:
            if buffer:
                yield {
                    "event": "token",
                    "data": json.dumps({"content": buffer})
                }

    # 5. Sugerencias de seguimiento (post-stream)
    suggestion_list = await generate_suggestions(full_response, sources, provider)
    if suggestion_list:
        yield {
            "event": "suggestions",
            "data": json.dumps({"suggestions": suggestion_list})
        }

    # 6. Guardar mensaje y señalizar fin
    msg = await chat_service.save_message(db, conv_id, MessageRole.ASSISTANT, full_response, ...)

    yield {
        "event": "done",
        "data": json.dumps({"message_id": str(msg.id)})
    }

El detalle del buffer para agentic search

El sistema soporta búsqueda agentic: el LLM puede emitir [SEARCH: nueva consulta] en medio de su respuesta para pedir más información. El problema: si emitís tokens uno a uno, el frontend podría renderizar [SEARCH: parcial como texto visible.

La solución es un buffer que retiene los últimos 20 caracteres si contienen [:

if buffer and '[' not in buffer[-20:]:
    yield {"event": "token", "data": json.dumps({"content": buffer})}
    buffer = ""

Si se detecta el marcador completo [SEARCH: ...], se interrumpe el stream, se ejecuta una nueva búsqueda, y se continúa con contexto expandido:

match = SEARCH_PATTERN.search(accumulated)
if match and search_round < MAX_SEARCH_ROUNDS:
    search_query = match.group(1).strip()
    search_round += 1

    yield {"event": "searching", "data": json.dumps({
        "query": search_query, "round": search_round
    })}

    # Ejecutar nueva búsqueda con el query del LLM
    new_sources = await search_chunks(db, search_query, kb_ids, rag_config=rag_config)

    # Deduplicar fuentes
    existing_ids = {s["chunk_id"] for s in all_sources}
    unique_new = [s for s in new_sources if s["chunk_id"] not in existing_ids]

    if unique_new:
        yield {"event": "sources", "data": json.dumps({
            "sources": unique_new, "round": search_round
        })}

    # Continuar generación con contexto expandido
    # (máximo 3 rounds de búsqueda)

Short-circuits: FAQ, cache y upgrade

Antes del pipeline RAG completo, hay varios "atajos" que retornan un EventSourceResponse diferente:

# FAQ match → respuesta instantánea sin LLM
if faq_match:
    async def faq_event_generator():
        yield {"event": "user_message_id", "data": json.dumps({"message_id": str(user_msg.id)})}
        yield {"event": "faq_match", "data": json.dumps(faq_match)}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id)})}
    return EventSourceResponse(faq_event_generator())

# Cache hit → respuesta cacheada, simular streaming
if cache_hit:
    async def cache_event_generator():
        yield {"event": "sources", "data": json.dumps({"sources": cache_hit["sources"]})}
        yield {"event": "confidence", "data": json.dumps({"score": cache_hit["confidence"]})}
        # Simular streaming palabra por palabra
        for token in cache_hit["response_text"].split(" "):
            yield {"event": "token", "data": json.dumps({"content": token + " "})}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id), "cached": True})}
    return EventSourceResponse(cache_event_generator())

# Budget agotado → señal de upgrade
if faq_only_mode:
    async def upgrade_event_generator():
        yield {"event": "upgrade_required", "data": json.dumps({
            "message": "Has alcanzado el límite de consultas de IA de tu plan."
        })}
        yield {"event": "done", "data": json.dumps({"message_id": str(msg.id)})}
    return EventSourceResponse(upgrade_event_generator())

¿Por qué simular streaming en cache hits? Porque el frontend tiene una sola ruta de renderizado. Si el cache devolviera todo el texto de golpe, necesitaría lógica especial. Simular el streaming mantiene el código del frontend simple.

Streaming desde los providers LLM

Cada provider tiene su propio formato. La capa de abstracción unifica todo en generadores que yield strings:

# Groq y OpenAI: generador SINCRÓNICO (SDK compatible)
def _stream_groq(messages, temperature=0.3, max_tokens=4096):
    stream = groq_client.chat.completions.create(
        model=settings.groq_model, messages=messages,
        stream=True, temperature=temperature, max_tokens=max_tokens,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

# Ollama: generador ASYNC (NDJSON sobre HTTP streaming)
async def _stream_ollama(messages, temperature=0.3, max_tokens=4096):
    async with httpx.AsyncClient(timeout=httpx.Timeout(300.0)) as client:
        async with client.stream("POST", f"{settings.ollama_url}/api/chat",
                                 json={"model": settings.ollama_model,
                                       "messages": messages, "stream": True}) as response:
            async for line in response.aiter_lines():
                data = json.loads(line)
                content = data.get("message", {}).get("content", "")
                if content:
                    yield content
                if data.get("done"):
                    break

El detalle clave: Groq/OpenAI retornan generadores sincrónicos, Ollama retorna async. El generador SSE necesita manejar ambos con inspect.isasyncgen(). Es feo, pero wrappear sync en async agrega complejidad e indirección que dificulta el debugging.

Frontend: consumo con fetch + ReadableStream

Por qué no EventSource

La API EventSource del browser tiene un problema fatal para este caso de uso: solo soporta GET. Nuestro endpoint de chat es un POST con body JSON (contenido del mensaje, IDs de knowledge bases, provider, etc.).

La alternativa: fetch + ReadableStream. Es más código, pero da control total.

El parser SSE

export async function sendMessage(
  convId: string,
  content: string,
  knowledgeBaseIds: string[],
  callbacks: StreamCallbacks,
  provider?: string,
): Promise<void> {
  const token = localStorage.getItem('access_token')

  const response = await fetch(`/api/v1/conversations/${convId}/messages`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      ...(token ? { Authorization: `Bearer ${token}` } : {}),
    },
    body: JSON.stringify({
      content,
      knowledge_base_ids: knowledgeBaseIds,
      provider,
    }),
  })

  if (!response.ok) {
    // Detectar error de límite de plan (403)
    if (response.status === 403) {
      const errorData = await response.json()
      if (errorData.detail?.includes('Limite de')) {
        callbacks.onError(`__PLAN_LIMIT__:${errorData.detail}`)
        return
      }
    }
    callbacks.onError(`Error: ${response.status}`)
    return
  }

  const reader = response.body?.getReader()
  if (!reader) {
    callbacks.onError('No response stream')
    return
  }

  const decoder = new TextDecoder()
  let buffer = ''
  let receivedDone = false

  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) break

      buffer += decoder.decode(value, { stream: true })
      const lines = buffer.split('\n')
      buffer = lines.pop() || ''  // Última línea incompleta → al buffer

      let currentEvent = ''
      for (const line of lines) {
        if (line.startsWith('event: ')) {
          currentEvent = line.slice(7).trim()
          continue
        }
        if (line.startsWith('data: ')) {
          const parsed = JSON.parse(line.slice(6))

          // Despachar según tipo de evento
          if (currentEvent === 'token') {
            callbacks.onToken(parsed.content)
          } else if (currentEvent === 'sources') {
            callbacks.onSources(parsed.sources)
          } else if (currentEvent === 'confidence') {
            callbacks.onConfidence?.(parsed.score)
          } else if (currentEvent === 'faq_match') {
            callbacks.onFaqMatch?.(parsed)
          } else if (currentEvent === 'upgrade_required') {
            callbacks.onUpgradeRequired?.(parsed.message)
          } else if (currentEvent === 'content_blocked') {
            callbacks.onContentBlocked?.(parsed.response, parsed.type)
          } else if (currentEvent === 'error') {
            callbacks.onError(parsed.error)
          } else if (currentEvent === 'done' || 'message_id' in parsed) {
            receivedDone = true
            callbacks.onDone(parsed.message_id)
          }
          currentEvent = ''
        }
      }
    }
  } catch (err) {
    callbacks.onError(`Error de conexión: ${err}`)
    return
  }

  // Si el stream terminó sin evento "done", hubo un problema
  if (!receivedDone) {
    callbacks.onError('La conexión se interrumpió antes de completar la respuesta')
  }
}

La interfaz de callbacks

export interface StreamCallbacks {
  onSources: (sources: Source[]) => void
  onToken: (token: string) => void
  onCodeResult: (result: CodeResult) => void
  onSearching: (query: string, round: number) => void
  onExecuting: (count: number) => void
  onSuggestions: (suggestions: string[]) => void
  onUserMessageId: (messageId: string) => void
  onFaqMatch?: (match: FAQMatch) => void
  onConfidence?: (score: number) => void
  onUpgradeRequired?: (message: string) => void
  onContentBlocked?: (response: string, blockType: string) => void
  onDone: (messageId: string) => void
  onError: (error: string) => void
}

El composable useChat conecta estos callbacks con el estado reactivo de Vue:

await api.sendMessage(conversationId, content, kbIds, {
  onToken: (token) => {
    streamingContent.value += token
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.content = streamingContent.value
    }
  },
  onSources: (s) => {
    sources.value = [...sources.value, ...s]
  },
  onConfidence: (score) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.confidenceScore = score
    }
  },
  onDone: (messageId) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.id = messageId
      last.sources = sources.value
    }
    streaming.value = false
  },
  onError: (err) => {
    const last = messages.value[messages.value.length - 1]
    if (last.role === 'assistant') {
      last.content = `Error: ${err}`
    }
    streaming.value = false
  },
})

Optimistic UI con reconciliación

Un detalle sutil: cuando el usuario envía un mensaje, lo agregamos inmediatamente a la UI con un ID temporal (crypto.randomUUID()). Pero el ID real lo genera la base de datos. El evento user_message_id sincroniza ambos:

const userMsg: Message = {
  id: crypto.randomUUID(),  // ID temporal optimista
  role: 'user',
  content,
  ...
}
messages.value.push(userMsg)

// Cuando llega el ID real del server:
onUserMessageId: (realId) => {
  userMsg.id = realId  // Reemplazar ID temporal con ID de DB
},

Esto es importante para el feedback: cuando el usuario le da thumbs-up a un mensaje, necesita el ID real para que el PATCH llegue al mensaje correcto.

Widget embebible: SSE en vanilla JS

El widget es ~970 líneas de vanilla JS corriendo dentro de un Shadow DOM cerrado (mode: 'closed'). El parsing SSE es idéntico al del frontend TypeScript, pero con var en lugar de const y sin types:

var reader = res.body.getReader();
var decoder = new TextDecoder();
var buffer = '';
var assistantContent = '';

while (true) {
  var readResult = await reader.read();
  if (readResult.done) break;
  buffer += decoder.decode(readResult.value, { stream: true });

  var lines = buffer.split('\n');
  buffer = lines.pop() || '';

  var event = null;
  for (var i = 0; i < lines.length; i++) {
    var line = lines[i].trim();
    if (line.startsWith('event:')) {
      event = line.slice(6).trim();
    } else if (line.startsWith('data:')) {
      var data = JSON.parse(line.slice(5).trim());

      if (event === 'session') {
        sessionToken = data.session_token;
        localStorage.setItem(STORAGE_KEY, sessionToken);
      } else if (event === 'token') {
        assistantContent += (data.content || '');
        messages[assistantMsgIndex].content = assistantContent;
        renderMessages();  // Re-render en cada token
      } else if (event === 'faq_match') {
        messages.push({ role: 'assistant', content: data.answer, faqMatch: data });
        renderMessages();
      }
    }
  }
}

Diferencias clave con la app principal

Auth: El widget usa X-API-Key (SHA-256 hasheada) en vez de JWT. No hay login.
Primer evento: En vez de user_message_id, el widget recibe session con un token que se persiste en localStorage para mantener historial entre visitas.
Shadow DOM cerrado: host.attachShadow({ mode: 'closed' }) aísla estilos y DOM completamente de la página host. Ni document.querySelector puede acceder al widget desde afuera.

Nginx: las 6 directivas que lo hacen funcionar

Esta es la sección que más dolores de cabeza me dio. Sin la configuración correcta de Nginx, el streaming SSE no funciona. Los tokens se acumulan en el buffer de Nginx y llegan todos juntos al final.

location /api/ {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;

    # === Las 6 directivas SSE ===
    proxy_buffering off;             # 1. No buffear la respuesta
    proxy_cache off;                 # 2. No cachear
    proxy_read_timeout 300s;         # 3. Timeout largo (streams duran minutos)
    proxy_set_header Connection '';  # 4. Deshabilitar keep-alive del upstream
    proxy_http_version 1.1;          # 5. HTTP/1.1 (requerido para chunked)
    chunked_transfer_encoding off;   # 6. Evitar chunked encoding de Nginx
}

Qué hace cada una

1. proxy_buffering off — La más importante. Por default, Nginx buferea la respuesta completa del upstream antes de enviarla al cliente. Con SSE, cada evento debe llegar al cliente inmediatamente.

2. proxy_cache off — Nginx puede cachear respuestas de upstream. Un stream SSE cacheado es... una muy mala idea.

3. proxy_read_timeout 300s — El default es 60 segundos. Una respuesta de LLM con búsqueda agentic (3 rounds) puede tardar 2-3 minutos. Sin este timeout, Nginx cierra la conexión a la mitad de la respuesta.

4. proxy_set_header Connection '' — Limpia el header Connection para evitar que Nginx mantenga la conexión upstream con keep-alive, lo que puede causar que los datos se retengan.

5. proxy_http_version 1.1 — HTTP/1.1 es requerido para transfer-encoding chunked. El default de Nginx para upstream es HTTP/1.0.

6. chunked_transfer_encoding off — Parece contradictorio con el punto anterior, pero esto desactiva el chunked encoding de Nginx, no del upstream. Sin esto, Nginx puede re-chunked la respuesta y alterar el timing de los eventos.

El síntoma que vas a ver sin estas directivas

Si te falta alguna, vas a ver este comportamiento:

El usuario envía un mensaje
El spinner de "cargando" se queda 5-30 segundos
De repente, toda la respuesta aparece de golpe
El streaming "funciona" en localhost (sin Nginx) pero no en producción

Si te pasa esto, es Nginx buffereando. Agregá las 6 directivas y se resuelve inmediatamente.

Cloudflare: el otro proxy invisible

Si usás Cloudflare como proxy (nube naranja), hay un detalle: Cloudflare también puede buffear responses. Pero en la práctica, Cloudflare detecta SSE automáticamente por el Content-Type: text/event-stream y desactiva el buffering. No necesité configuración especial.

Redis: concurrencia de streams

Un usuario podría abrir 10 tabs y lanzar 10 streams simultáneos. Cada stream consume recursos del LLM, conexiones de DB, y memoria. El control de concurrencia usa Redis como contador atómico:

_CONCURRENT_TTL = 300  # 5 minutos de TTL como red de seguridad

async def increment_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0  # Fail-open: sin Redis, permitir

    redis_key = f"concurrent:{tenant_id}"
    pipe = client.pipeline()
    pipe.incr(redis_key)
    pipe.expire(redis_key, _CONCURRENT_TTL)
    results = await pipe.execute()
    return results[0]

async def decrement_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    count = await client.decr(redis_key)
    if count <= 0:
        await client.delete(redis_key)  # Limpiar keys muertos
        return 0
    await client.expire(redis_key, _CONCURRENT_TTL)
    return count

El patrón try/finally

El incremento y decremento están en un try/finally para garantizar que el contador se decrementa siempre, incluso si el cliente desconecta a mitad del stream:

async def event_generator():
    await increment_concurrent(str(tenant_id))
    try:
        async for event in _event_generator_inner():
            yield event
    finally:
        await decrement_concurrent(str(tenant_id))

TTL como red de seguridad

¿Qué pasa si el servidor crashea a mitad de un stream? El decrement nunca se ejecuta y el contador queda inflado. El TTL de 5 minutos resuelve esto: si no se renueva con expire, la key se auto-elimina. El sistema se auto-sana.

Fail-open y rate limiting

Dos principios importantes:

Fail-open: Si Redis está caído, el rate limiter permite todo. Prefiero servir requests sin rate limiting a rechazar todo porque Redis se cayó. Redis es un guardia, no una puerta.
Sliding window por API key: Además de la concurrencia por tenant, cada API key del widget tiene su propio rate limit usando sorted sets de Redis. Timestamps como scores, zremrangebyscore para limpiar expirados, zcard para contar — todo en un pipeline atómico.

Error handling: las 3 capas

Backend: Siempre emitir done después de error. Sin esto, el frontend se queda en estado "streaming" indefinidamente:

except Exception as e:
    yield {"event": "error", "data": json.dumps({"error": f"Error del LLM: {e}"})}
    yield {"event": "done", "data": json.dumps({"message_id": "error"})}
    return

Frontend: Detectar streams incompletos. Si el reader retorna done: true pero nunca recibimos un evento done, algo salió mal (timeout de Nginx, crash del backend):

if (!receivedDone) {
  callbacks.onError('La conexión se interrumpió antes de completar la respuesta')
}

Pre-stream: Los errores 403 (límite de plan) llegan como HTTP normal antes del stream SSE. El frontend usa un prefijo __PLAN_LIMIT__: como convención interna para que el composable distinga errores de plan de errores genéricos y muestre UI diferente (card de upgrade vs. mensaje de error).

Números reales

Métrica	Valor
Latencia primer token (Groq)	~200-400ms
Latencia primer token (Ollama local)	~1-2s
Tiempo total respuesta típica	~3-8s
Tiempo total con agentic search (3 rounds)	~15-30s
FAQ match (sin LLM)	~15ms
Cache hit (simulado)	~50ms
Overhead de Nginx SSE	<1ms por evento
Concurrent streams por tenant (Free plan)	2
Rate limit API key (default)	30 req/min

Lecciones aprendidas

1. SSE > WebSockets para LLM streaming

No intentés meter WebSockets donde SSE alcanza. Para un flujo unidireccional (servidor → cliente), SSE es más simple, más compatible con proxies, y más fácil de debuggear. La reconexión automática del browser es un bonus.

2. Siempre emitir "done" al final

Sin un evento de terminación explícito, el frontend no puede distinguir entre "el stream terminó" y "el stream se cortó". Emitir done siempre — incluso después de errores — es lo que hace al protocolo robusto.

3. Nginx buffering es el enemigo silencioso

En desarrollo sin Nginx todo funciona perfecto. En producción, los tokens llegan todos juntos. La primera vez que me pasó, perdí 2 horas pensando que el problema era en el backend. Eran 6 líneas de Nginx.

4. Buffer para marcadores inline

Si tu LLM puede emitir marcadores especiales ([SEARCH:], código ejecutable, etc.), necesitás un buffer que retenga texto hasta estar seguro de que no hay un marcador parcial. Sin esto, el usuario ve artefactos como [SEAR en el chat.

5. Try/finally con contadores Redis

Todo counter que se incrementa antes de un stream debe decrementarse en un finally. No en el happy path, no en el error handler: en finally. Los clientes desconectan sin aviso, los servidores crashean, y los generadores se interrumpen.

6. Fail-open en servicios auxiliares

Redis para rate limiting debe ser fail-open. Si Redis se cae, es mejor servir sin rate limiting que rechazar todos los requests. Lo mismo aplica para logging, analytics, y cualquier servicio que no sea crítico para la respuesta.

7. fetch + ReadableStream > EventSource

EventSource es elegante pero limitado (solo GET, headers mínimos). fetch + ReadableStream requiere más código pero soporta POST, headers custom (JWT), y manejo fino de errores. Para un caso real, la flexibilidad justifica las ~30 líneas extra.

Lo que sigue

WebSocket upgrade opcional: Para features bidireccionales futuras (typing indicators, presencia), evaluar un upgrade selectivo sin reemplazar SSE
Backpressure: Si el frontend no puede renderizar tan rápido como el LLM genera, implementar señalización de backpressure
Compression: Evaluar SSE sobre HTTP/2 con compresión de frames para reducir bandwidth en widgets con alto tráfico

Conclusión

SSE para streaming de LLM no es difícil de implementar — pero sí es fácil de implementar mal. Las trampas están en los detalles: Nginx que buferea, generadores que no limpian contadores, frontends que no detectan streams incompletos, y widgets que no manejan todos los tipos de evento.

El protocolo de eventos custom (token, sources, confidence, done, error, etc.) es lo que transforma un stream de texto en un sistema usable: el frontend sabe qué está recibiendo y puede renderizarlo correctamente, mostrar indicadores de confianza, acumular fuentes de diferentes rounds de búsqueda, y manejar edge cases como límites de plan o contenido bloqueado.

Lo más importante: testear siempre a través de Nginx. Lo que funciona en localhost sin proxy no es representativo de producción. Esas 6 directivas de Nginx son la diferencia entre "streaming real token por token" y "todo el texto de golpe después de 10 segundos".

Real Multi-Tenancy with FastAPI and PostgreSQL: Plans, Quotas and Data Isolation

Martin Palopoli — Tue, 31 Mar 2026 13:00:00 +0000

I implemented real multi-tenancy in a RAG SaaS engine: tenants with plans, quota enforcement with atomic counters (UPSERT), Redis rate limiting (sliding window), RBAC with 3 roles, and data isolation where every query is filtered by tenant_id. This article covers the full architecture, the pitfalls I avoided, and the key code.

Why "Adding a tenant_id" Isn't Multi-Tenancy

Most multi-tenancy tutorials stop at "add a tenant_id column and filter every query". That's 10% of the problem. In production you'll need:

Plans with real limits that can't be bypassed with concurrent requests
Atomic counters that don't lose increments under load
Rate limiting that survives server restarts
Roles that restrict actions, not just visibility
Real isolation where a bug in one service doesn't expose another tenant's data
Billing cycles that reset correctly each month

This article shows how I solved each one in a real system with async FastAPI and PostgreSQL.

The Stack

Component	Technology
Backend	Python 3.12 + FastAPI (100% async)
Database	PostgreSQL 16
ORM	SQLAlchemy async + Alembic
Cache/Rate limit	Redis 7
Auth	JWT (access 30min, refresh 7d with rotation)
Passwords	bcrypt (via passlib)

Database Design

The 4 Fundamental Tables

┌──────────────┐     ┌──────────────┐
│    plans     │     │   tenants    │
├──────────────┤     ├──────────────┤
│ id (UUID)    │◄────│ plan_id (FK) │
│ name         │     │ id (UUID)    │
│ slug         │     │ name         │
│ max_users    │     │ slug (unique)│
│ max_kbs      │     │ is_active    │
│ max_documents│     │ plan_started │
│ max_storage  │     │ created_at   │
│ max_messages │     └──────┬───────┘
│ max_tokens   │            │
│ max_api_keys │     ┌──────┴───────┐
│ price_monthly│     │    users     │
│ max_concurrent│    ├──────────────┤
└──────────────┘     │ id (UUID)    │
                     │ tenant_id(FK)│
                     │ email        │
                     │ role (enum)  │
                     │ is_active    │
                     └──────────────┘

┌───────────────────────┐
│ tenant_monthly_usage  │
├───────────────────────┤
│ id (UUID)             │
│ tenant_id (FK)        │
│ billing_period (str)  │
│ messages_count (int)  │
│ tokens_used (bigint)  │
│ UNIQUE(tenant_id,     │
│   billing_period)     │
└───────────────────────┘

Plan Model

Each plan defines all the limits the system will enforce:

class Plan(Base):
    __tablename__ = "plans"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    max_users: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_concurrent: Mapped[int] = mapped_column(Integer, nullable=False, default=2)
    max_kbs: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_documents: Mapped[int] = mapped_column(Integer, nullable=False, default=20)
    max_storage_mb: Mapped[int] = mapped_column(Integer, nullable=False, default=200)
    max_messages_month: Mapped[int] = mapped_column(Integer, nullable=False, default=500)
    max_tokens_month: Mapped[int] = mapped_column(BigInteger, nullable=False, default=200000)
    max_api_keys: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    price_monthly: Mapped[float | None] = mapped_column(Numeric(10, 2), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

Key decision: -1 means unlimited. Instead of having an is_unlimited boolean per resource, I use -1 as a convention. Simplifies all comparisons:

if plan.max_users == -1:
    return  # Unlimited, skip check

Tenant Model

class Tenant(Base):
    __tablename__ = "tenants"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    slug: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    plan_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True), ForeignKey("plans.id"), nullable=True
    )
    plan_started_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True, default=None
    )
    notification_preferences: Mapped[dict | None] = mapped_column(JSONB, nullable=True)

plan_started_at is crucial: it defines the billing cycle anchor. When a tenant upgrades from Free to a paid plan, it resets to the payment timestamp. This determines when monthly counters reset.

User Model

class UserRole(str, PyEnum):
    SUPER_ADMIN = "super_admin"
    TENANT_ADMIN = "tenant_admin"
    MEMBER = "member"


class User(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str] = mapped_column(String(255), default="")
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole, name="user_role", values_callable=lambda e: [x.value for x in e]),
        default=UserRole.MEMBER,
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    email_verified: Mapped[bool] = mapped_column(Boolean, default=False)

A user belongs to exactly one tenant. No multi-tenant users. If someone needs access to two organizations, create two accounts. This enormously simplifies the permissions model.

RBAC: 3 Roles, Zero Ambiguity

The Roles

Role	Scope	Can Do
`super_admin`	Platform	View all tenants, change plans, view global metrics
`tenant_admin`	Their tenant	CRUD KBs, documents, FAQs, invite users, view analytics
`member`	Their tenant (restricted)	Chat, view assigned KBs, feedback

Dependency Injection for Authorization

FastAPI has a dependency system that's perfect for RBAC. I created a dependency factory:

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    payload = decode_access_token(token)
    user_id = payload.get("sub")
    if not user_id:
        raise UnauthorizedException("Invalid token payload")

    user = await get_user_by_id(db, UUID(user_id))
    if not user:
        raise UnauthorizedException("User not found")
    if not user.is_active:
        raise UnauthorizedException("Account deactivated")

    # Verify tenant is active (super_admin bypasses)
    if user.role != UserRole.SUPER_ADMIN:
        tenant = await db.get(Tenant, user.tenant_id)
        if not tenant or not tenant.is_active:
            raise UnauthorizedException("Tenant deactivated")

    return user

Subtle detail: super_admin bypasses the active tenant check. If we deactivate a malicious tenant, the super_admin can still get in to investigate.

Role Restriction Factory

def require_role(*roles):
    """Dependency factory: restricts endpoint to specific roles."""
    async def _check(current_user=Depends(get_current_user)):
        if current_user.role not in roles:
            raise ForbiddenException("You don't have permission for this action")
        return current_user
    return _check


# Shortcuts
def require_super_admin():
    return require_role(UserRole.SUPER_ADMIN)

def require_tenant_admin_or_above():
    return require_role(UserRole.SUPER_ADMIN, UserRole.TENANT_ADMIN)

Usage in endpoints:

@router.get("/tenants", response_model=list[TenantResponse])
async def list_tenants(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role != UserRole.SUPER_ADMIN:
        raise ForbiddenException("Only super_admin can list all tenants")
    return await tenant_service.list_tenants(db)


@router.put("/{kb_id}/access")
async def update_kb_access(
    kb_id: UUID,
    data: KBAccessConfig,
    db: AsyncSession = Depends(require_tenant_admin_or_above()),
):
    # Only tenant_admin or super_admin reach here
    ...

Data Isolation: tenant_id Everywhere

The Pattern

Every table containing user data has a tenant_id column (direct or transitive). No exceptions.

class KnowledgeBase(Base):
    __tablename__ = "knowledge_bases"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

class ApiKey(Base):
    __tablename__ = "api_keys"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

Filtering in Every Service

Every function that lists or searches data filters by the authenticated user's tenant_id:

async def list_knowledge_bases(db: AsyncSession, user: User) -> list[dict]:
    stmt = (
        select(KnowledgeBase, func.count(Document.id).label("document_count"))
        .outerjoin(Document, Document.knowledge_base_id == KnowledgeBase.id)
    )

    if user.role == UserRole.SUPER_ADMIN:
        # Super admin sees only their own KBs + system KBs
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
    else:
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
        # For members, apply access filter
        access_filter = await get_accessible_kb_filter(user)
        if access_filter is not None:
            stmt = stmt.where(or_(KnowledgeBase.is_system == True, access_filter))

    stmt = stmt.group_by(KnowledgeBase.id).order_by(KnowledgeBase.created_at.desc())
    result = await db.execute(stmt)
    # ...

Important lesson: the super_admin does NOT see other tenants' data. They're the platform operator. They see aggregate statistics (total tenants, revenue, etc.) but can't read other tenants' conversations or KBs. This was a deliberate privacy decision.

Granular KB Access

Within the same tenant, not all members see all KBs. There's an access control system:

async def _check_kb_access(db: AsyncSession, kb: KnowledgeBase, user: User) -> None:
    has_access = await user_can_access_kb(db, user, kb)
    if not has_access:
        raise ForbiddenException("You don't have access to this knowledge base")

KBs can be:

Public within the tenant: all members can see them
Private: only members assigned via groups can see them
System: system KBs, visible to all, only modifiable by super_admin

CASCADE: Clean Deletion

All foreign keys use ondelete="CASCADE". When a tenant is deleted, everything cascades: users, KBs, documents, chunks, conversations, API keys, usage logs. A single operation:

async def delete_tenant(db: AsyncSession, tenant_id: UUID, current_user: User) -> dict:
    tenant = await get_tenant(db, tenant_id)

    # Safety checks
    if current_user.tenant_id == tenant_id:
        raise BadRequestException("You cannot delete your own tenant")
    if tenant.slug == "system-tenant":
        raise ForbiddenException("Cannot delete the system tenant")

    # 1. Cancel active subscription
    try:
        await admin_cancel_subscription(db, tenant_id, immediate=True)
    except NotFoundException:
        pass

    # 2. Clean Redis cache
    redis = await get_redis()
    if redis:
        keys = await redis.keys(f"*:{tenant_id}:*")
        if keys:
            await redis.delete(*keys)

    # 3. DELETE → CASCADE handles the rest
    await db.delete(tenant)
    await db.commit()

Plan Enforcement: Atomic Counters with UPSERT

The Problem

Imagine a Free-plan tenant has a limit of 50 messages/month. Two users send a message at the same time. If you do SELECT count → check → INSERT, you have a race condition: both read 49, both pass the check, both insert. Now there are 51.

The Solution: Atomic UPSERT

The tenant_monthly_usage table uses a UNIQUE constraint on (tenant_id, billing_period):

class TenantMonthlyUsage(Base):
    __tablename__ = "tenant_monthly_usage"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    billing_period: Mapped[str] = mapped_column(String(10), nullable=False)
    messages_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    tokens_used: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)

    __table_args__ = (
        UniqueConstraint("tenant_id", "billing_period", name="uq_tenant_billing_usage"),
    )

The increment uses INSERT ... ON CONFLICT DO UPDATE:

async def increment_message_count(db: AsyncSession, tenant_id: UUID) -> None:
    anchor = await _get_billing_anchor(db, tenant_id)
    bp = _get_billing_period_key(anchor)
    await db.execute(text("""
        INSERT INTO tenant_monthly_usage (id, tenant_id, billing_period, messages_count, tokens_used)
        VALUES (gen_random_uuid(), :tid, :bp, 1, 0)
        ON CONFLICT (tenant_id, billing_period)
        DO UPDATE SET messages_count = tenant_monthly_usage.messages_count + 1,
                      updated_at = now()
    """), {"tid": tenant_id, "bp": bp})
    await db.commit()

Why does this work? PostgreSQL executes INSERT ... ON CONFLICT DO UPDATE atomically. There's no race condition window. If two requests arrive at the same time, one will INSERT and the other will UPDATE, but the counter will always be correct.

Anti-fraud detail: counters only increment. There's no endpoint to decrement. No way for a user to manipulate their usage. The only way to "reset" is when the billing period changes (i.e., a month passes).

Billing Period Calculation

def _get_billing_period_key(anchor: datetime) -> str:
    now = datetime.now(timezone.utc)
    billing_day = anchor.day

    # Clamping: if anchor is day 31 and we're in February (28 days),
    # effective billing day is 28
    max_day = calendar.monthrange(now.year, now.month)[1]
    effective_day = min(billing_day, max_day)

    if now.day >= effective_day:
        return f"{now.year}-{now.month:02d}-{effective_day:02d}"
    else:
        # Current period started last month
        if now.month == 1:
            prev_year, prev_month = now.year - 1, 12
        else:
            prev_year, prev_month = now.year, now.month - 1
        max_day_prev = calendar.monthrange(prev_year, prev_month)[1]
        effective_day_prev = min(billing_day, max_day_prev)
        return f"{prev_year}-{prev_month:02d}-{effective_day_prev:02d}"

Why is this complex? Because months don't have the same number of days. If a tenant paid on January 31st, their next cycle doesn't start on February 31st (doesn't exist). The clamping handles this.

Limit Verification

Before executing the RAG pipeline, I verify limits:

async def check_message_limit(db: AsyncSession, tenant_id: UUID) -> None:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return
    if plan.max_messages_month == -1:
        return  # Unlimited
    usage = await get_monthly_usage(db, tenant_id)
    if usage["messages_month"] >= plan.max_messages_month:
        raise PlanLimitException(
            f"Monthly message limit reached "
            f"({plan.max_messages_month} max for {plan.name} plan)"
        )

PlanLimitException returns HTTP 403, not 429. It's a plan restriction, not rate limiting. The frontend distinguishes between "upgrade your plan" (403) and "wait a moment" (429).

Limits on All Resources

Not just messages. Every resource has its check:

# Before creating a user
await check_user_limit(db, tenant_id)

# Before creating a KB
await check_kb_limit(db, tenant_id)

# Before uploading a document
await check_document_limit(db, tenant_id)
await check_storage_limit(db, tenant_id, new_file_size_bytes=file_size)

# Before creating an API key
await check_api_key_limit(db, tenant_id)

# Before sending a message to the LLM
await check_message_limit(db, tenant_id)
await check_concurrent_limit(db, tenant_id)

The pattern is always the same:

async def check_kb_limit(db: AsyncSession, tenant_id: UUID) -> None:
    tenant = await get_tenant(db, tenant_id)
    if not tenant.plan_id:
        return
    plan = await db.get(Plan, tenant.plan_id)
    if not plan or plan.max_kbs == -1:
        return
    stmt = select(func.count(KnowledgeBase.id)).where(KnowledgeBase.tenant_id == tenant_id)
    result = await db.execute(stmt)
    current_count = result.scalar() or 0
    if current_count >= plan.max_kbs:
        raise PlanLimitException(
            f"Knowledge base limit reached ({plan.max_kbs} max)"
        )

Rate Limiting with Redis: Sliding Window

Why Redis and Not PostgreSQL

Billing period counters are in PostgreSQL because they need durability. Rate limiting is in Redis because it needs speed and automatic TTL.

If the server restarts mid-stream, I don't want a "phantom" counter in PostgreSQL blocking the tenant. Redis with TTL self-cleans.

Sliding Window with Sorted Sets

import time
import redis.asyncio as redis

async def is_allowed(key: str, max_requests: int, window_seconds: int = 60) -> bool:
    client = await get_redis()
    if client is None:
        return True  # Fail-open

    redis_key = f"ratelimit:{key}"
    now = time.time()
    cutoff = now - window_seconds

    try:
        pipe = client.pipeline()
        pipe.zremrangebyscore(redis_key, 0, cutoff)   # Clean expired
        pipe.zcard(redis_key)                          # Count current
        pipe.zadd(redis_key, {str(now): now})          # Add this request
        pipe.expire(redis_key, window_seconds + 1)     # TTL safety net
        results = await pipe.execute()

        current_count = results[1]
        if current_count >= max_requests:
            await client.zrem(redis_key, str(now))     # Rollback
            return False
        return True
    except Exception:
        return True  # Fail-open

Why sorted sets and not a simple INCR? Because I need a sliding window, not a fixed window. With INCR, if a user sends 60 requests at second 59 of the minute, and 60 more at second 1 of the next minute, 120 requests pass in 2 seconds. With sorted sets, each request has its timestamp and the window slides continuously.

Why fail-open? If Redis goes down, I prefer to let requests through (graceful degradation) than to block all users. Billing limits in PostgreSQL are still active as a safety net.

Stream Concurrency

To limit simultaneous SSE connections per tenant, I use a different pattern: INCR/DECR with safety TTL.

_CONCURRENT_TTL = 300  # 5 minutes

async def increment_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        pipe = client.pipeline()
        pipe.incr(redis_key)
        pipe.expire(redis_key, _CONCURRENT_TTL)
        results = await pipe.execute()
        return results[0]
    except Exception:
        return 0


async def decrement_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        count = await client.decr(redis_key)
        if count <= 0:
            await client.delete(redis_key)
            return 0
        await client.expire(redis_key, _CONCURRENT_TTL)
        return count
    except Exception:
        return 0

Why 5-minute TTL? If the server crashes during a stream, the DECR never executes and the counter stays inflated. The TTL self-heals: after 5 minutes without activity, the counter deletes itself.

Usage in the chat endpoint:

@router.post("/conversations/{conv_id}/messages/stream")
async def stream_chat(conv_id: UUID, data: MessageCreate, ...):
    # Check LLM budget (soft check)
    faq_only_mode = await is_llm_budget_exhausted(db, current_user.tenant_id)
    if not faq_only_mode:
        await check_concurrent_limit(db, current_user.tenant_id)

    # ... prepare pipeline ...

    async def event_generator():
        await increment_concurrent(str(current_user.tenant_id))
        try:
            async for event in _event_generator_inner():
                yield event
        finally:
            # ALWAYS decrement, even on error
            await decrement_concurrent(str(current_user.tenant_id))

    return EventSourceResponse(event_generator())

The finally is critical. Without it, any exception during streaming would leave the counter inflated.

Registration: One Tenant Per User, Automatic

When someone registers, a tenant with the Free plan is automatically created:

async def register(db: AsyncSession, data: UserRegister) -> User:
    existing = await get_user_by_email(db, data.email)
    if existing:
        raise BadRequestException("Email already registered")

    # Create tenant automatically
    tenant = await create_tenant(db, data.tenant_name, plan_slug="free")

    user = User(
        email=data.email,
        hashed_password=hash_password(data.password),
        full_name=data.full_name,
        role=UserRole.TENANT_ADMIN,  # Creator is admin
        tenant_id=tenant.id,
        email_verified=False,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

The tenant slug is auto-generated with deduplication:

def _generate_slug(name: str) -> str:
    slug = re.sub(r'[^a-z0-9]+', '-', name.lower()).strip('-')
    return slug[:80] if slug else 'tenant'


async def create_tenant(db: AsyncSession, name: str, plan_slug: str = "free") -> Tenant:
    # Find plan
    stmt = select(Plan).where(Plan.slug == plan_slug, Plan.is_active == True)
    result = await db.execute(stmt)
    plan = result.scalar_one_or_none()

    base_slug = _generate_slug(name)
    slug = base_slug

    # If exists, add random suffix
    existing = await db.execute(select(Tenant).where(Tenant.slug == slug))
    if existing.scalar_one_or_none():
        slug = f"{base_slug}-{uuid.uuid4().hex[:6]}"

    tenant = Tenant(name=name, slug=slug, plan_id=plan.id if plan else None)
    db.add(tenant)
    await db.flush()
    return tenant

JWT: tenant_id in the Token

The access token includes tenant_id and role directly in the payload:

def create_access_token(user_id: str, tenant_id: str, role: str,
                        expire_minutes: int | None = None) -> str:
    minutes = expire_minutes or settings.access_token_expire_minutes
    expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
    payload = {
        "sub": user_id,
        "tenant_id": tenant_id,
        "role": role,
        "exp": expire,
        "type": "access",
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

Why include tenant_id in the JWT? To avoid a DB query on every request just to find out which tenant the user belongs to. In get_current_user I still verify against the DB (in case the user was deactivated), but the token's tenant_id serves as a quick hint.

Refresh Token Rotation

Every time a refresh token is used, it's revoked and a new one is issued:

async def refresh_access_token(db: AsyncSession, refresh_token_value: str):
    stmt = select(RefreshToken).where(
        RefreshToken.token == refresh_token_value,
        RefreshToken.revoked == False,
    )
    result = await db.execute(stmt)
    token_record = result.scalar_one_or_none()

    if not token_record:
        raise UnauthorizedException("Invalid refresh token")

    # Revoke the old one (rotation)
    token_record.revoked = True

    user = await get_user_by_id(db, token_record.user_id)
    if not user or not user.is_active:
        await db.commit()
        raise UnauthorizedException("User not found or deactivated")

    # Issue new tokens
    new_access = create_access_token(str(user.id), str(user.tenant_id), user.role.value)
    new_refresh_value = create_refresh_token_value()
    new_refresh = RefreshToken(
        user_id=user.id,
        token=new_refresh_value,
        expires_at=datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_days),
    )
    db.add(new_refresh)
    await db.commit()

    return new_access, new_refresh_value

If someone intercepts a refresh token and uses it, the original token gets revoked. When the legitimate user tries to refresh, it will fail and they'll know there was a compromise.

Threshold Warnings: Alert Before It Breaks

I don't wait until the tenant hits 100% of their limit. I implemented warnings at 80% and 95%:

async def get_usage_warnings(db: AsyncSession, tenant_id: UUID) -> list[dict]:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return []

    usage = await get_monthly_usage(db, tenant_id)
    warnings = []

    resources = [
        ("messages", usage["messages_month"], plan.max_messages_month),
        ("tokens", usage["tokens_month"], plan.max_tokens_month),
    ]

    for resource, used, max_val in resources:
        if max_val == -1 or max_val <= 0:
            continue
        pct = round((used / max_val) * 100, 1)
        if pct >= 95:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "critical"
            })
        elif pct >= 80:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "warning"
            })

    return warnings

Warnings are sent by email to tenant admins, with deduplication to avoid spamming:

# Only send if:
# 1. Not sent for this billing period, OR
# 2. Escalating from warning → critical
if last_sent == bp and last_level == "critical":
    return  # Already sent critical for this period

# Save dedup marker
updated_prefs["_last_warning_bp"] = bp
updated_prefs["_last_warning_level"] = most_severe["level"]

FAQ-Only Fallback: Graceful Degradation

When a tenant exhausts their LLM budget, I don't block everything. FAQs keep working:

async def is_llm_budget_exhausted(db: AsyncSession, tenant_id: UUID) -> bool:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return False

    usage = await get_monthly_usage(db, tenant_id)

    if plan.max_messages_month != -1 and usage["messages_month"] >= plan.max_messages_month:
        return True
    if plan.max_tokens_month != -1 and usage["tokens_month"] >= plan.max_tokens_month:
        return True
    return False

In the chat flow, if is_llm_budget_exhausted returns True, FAQ match is attempted first. If it matches, free response. If not, an upgrade_required SSE event is sent and the frontend shows an upgrade card.

Custom Exceptions: Semantic Errors

class PlanLimitException(HTTPException):
    def __init__(self, detail: str = "Plan limit reached"):
        super().__init__(status_code=status.HTTP_403_FORBIDDEN, detail=detail)

class TooManyRequestsException(HTTPException):
    def __init__(self, detail: str = "Too many requests"):
        super().__init__(status_code=429, detail=detail)

The frontend distinguishes:

403 PlanLimit: "You've reached your plan limit. Upgrade available."
429 RateLimit: "Too many requests. Try again in a few seconds."
401 Unauthorized: "Session expired. Please log in again."

Numbers That Matter

Resource	Free	Starter	Pro
Users	3	10	50
Knowledge Bases	3	10	-1 (unlimited)
Documents	20	100	-1
Storage	200 MB	1 GB	10 GB
AI Messages/month	50	500	5000
API Keys	1	5	20
Concurrency	2	5	20

Lessons Learned

1. UPSERT > SELECT + INSERT

Never do "read count → check → insert" for usage counters. The race condition is inevitable under load. PostgreSQL's atomic UPSERT is the only correct approach. Saved me hours of debugging "why does the tenant have 51 messages when the limit is 50".

2. Fail-Open for Rate Limiting, Fail-Closed for Billing

If Redis goes down, rate limits relax (fail-open) but plan limits in PostgreSQL remain active (fail-closed). Never let a downed dependency block all your users.

3. The Super Admin Isn't an Omniscient God

My first design gave the super_admin full access to all tenants' data. Bad. The super_admin is the platform operator, not the data owner. They see aggregate metrics, not individual conversations. This simplifies compliance and builds trust.

4. `-1` for Unlimited Is Better Than a Boolean

Having max_kbs: int with -1 = unlimited is cleaner than max_kbs: int + is_kbs_unlimited: bool. One column, one check. Applied it to all plan limits.

5. Billing Period Based on Payment Date, Not Calendar Month

If a tenant paid on March 15th, their cycle runs from 3/15 to 4/14, not 3/1 to 3/31. Seems like a minor detail until a user complains that "I paid yesterday and already used half my quota".

6. TTL as Safety Net for Redis Counters

Redis concurrency counters need TTL. Without it, a server crash leaves "zombie" counters that block the tenant. With a 5-minute TTL, the system self-heals.

7. CASCADE Simplifies Deletion But You Must Clean Redis Too

PostgreSQL's ON DELETE CASCADE is wonderful for deleting everything related to a tenant. But it doesn't clean Redis. You have to do it explicitly. Found out when a deleted tenant was still counting rate limits.

What's Next

Audit log: Record who did what, when, on what resource
Fine-grained permissions: Permissions per KB and per action (not just global role)
Multi-region: Tenants assigned to specific regions for compliance

Conclusion

Real multi-tenancy isn't an extra column in the database. It's an enforcement system spanning from the data model to rate limiting, through authorization and billing. The most critical points:

Atomic counters (UPSERT) so limits are respected under concurrency
Redis for the ephemeral (rate limits, concurrency) and PostgreSQL for the durable (billing, usage)
Clear roles with dependency injection that makes authorization impossible to skip
Fail-open where it hurts little, fail-closed where it hurts a lot
Isolation by default — every query filters by tenant_id, no exceptions

If you're building a multi-tenant SaaS, invest in enforcement from day 1. Refactoring this later is orders of magnitude more painful than doing it right from the start.

If you're building something multi-tenant and hit a problem I didn't cover, drop a comment. And if this article saved you time, a like helps it reach more people.

How I Deployed a RAG Engine to Production with Docker, Nginx and DigitalOcean

Martin Palopoli — Tue, 24 Mar 2026 13:28:00 +0000

I deployed a full RAG engine (FastAPI + PostgreSQL + pgvector + Redis) on a 4GB RAM VPS for $24/month. This article covers the real deployment architecture: Docker multi-stage builds, PostgreSQL tuned for limited resources, Nginx as reverse proxy with SSE support, zero-downtime deploys with maintenance mode, automated backups and cron monitoring.

The Context

In the previous article I built a production RAG pipeline with hybrid search, cross-encoder reranking and semantic cache. Everything worked perfectly in local Docker.

The problem: getting it to production on a budget VPS without it exploding.

A RAG system isn't a typical CRUD app. It has:

Embedding models that consume ~500MB of RAM per worker
PostgreSQL with heavy extensions (pgvector + HNSW indexes)
SSE streaming that needs long-lived connections
Redis for rate limiting and cache
All of that competing for 4GB of RAM

Chosen Infrastructure

Component	Specification
VPS	DigitalOcean 4GB RAM / 2 vCPU / 80GB SSD
OS	Ubuntu 24.04 LTS
Containers	Docker Compose (5 services)
SSL	Cloudflare (proxy + origin certificates)
DNS	Cloudflare
Total cost	~$24/month

Why not Kubernetes? Because for a single VPS it's overkill. Docker Compose with restart policies and health checks covers 95% of what you need for a web service with a few thousand users.

Docker Compose: Dev vs Production

Development

# docker-compose.yml
services:
  db:
    image: pgvector/pgvector:pg16
    ports:
      - "5433:5432"
    environment:
      POSTGRES_DB: ragdb
      POSTGRES_USER: raguser
      POSTGRES_PASSWORD: localpass123
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U raguser -d ragdb"]
      interval: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  backend:
    build: ./backend
    ports:
      - "8000:8000"
    volumes:
      - ./backend/app:/app/app  # Hot reload
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  frontend:
    build: ./frontend
    ports:
      - "5173:5173"
    volumes:
      - ./frontend/src:/app/src  # Hot reload

Nothing surprising: exposed ports, volumes for hot reload, basic health checks.

Production: The Differences That Matter

# docker-compose.prod.yml
services:
  db:
    image: pgvector/pgvector:pg16
    container_name: app-db
    restart: always
    env_file: .env.production
    ports:
      - "127.0.0.1:5432:5432"  # Localhost only
    volumes:
      - pgdata:/var/lib/postgresql/data
    command: >
      postgres
        -c shared_buffers=128MB
        -c effective_cache_size=256MB
        -c max_connections=50
        -c work_mem=4MB
        -c maintenance_work_mem=64MB
        -c random_page_cost=1.1
        -c effective_io_concurrency=200
        -c wal_buffers=4MB
        -c checkpoint_completion_target=0.9
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"]
      interval: 10s
      retries: 5

  redis:
    image: redis:7-alpine
    container_name: app-redis
    restart: always
    command: redis-server --maxmemory 64mb --maxmemory-policy allkeys-lru
    ports:
      - "127.0.0.1:6379:6379"  # Localhost only
    volumes:
      - redisdata:/data

  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    container_name: app-backend
    restart: always
    env_file: .env.production
    ports:
      - "127.0.0.1:8000:8000"  # Localhost only, Nginx in front
    deploy:
      resources:
        limits:
          memory: 1536M
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile.prod  # Multi-stage with Nginx
    container_name: app-frontend
    restart: always
    ports:
      - "127.0.0.1:5173:5173"  # Localhost only
    deploy:
      resources:
        limits:
          memory: 64M

The Key Differences

1. Ports bound to localhost only

ports:
  - "127.0.0.1:8000:8000"  # ✅ Only Nginx can access
  # vs
  - "8000:8000"  # ❌ Open to the world

If you publish a port without 127.0.0.1, Docker modifies iptables and bypasses the system firewall. This is a classic mistake.

2. PostgreSQL tuned for 4GB

shared_buffers=128MB        # ~25% of RAM available for PG (~512MB)
effective_cache_size=256MB  # What the OS can cache
max_connections=50          # You don't need 100 with an async backend
work_mem=4MB               # Careful: multiplied by connection × sort ops
random_page_cost=1.1       # SSD, not spinning disk

PostgreSQL defaults assume a dedicated server with 1GB+ of RAM just for PG. On a shared VPS with 4 other services, you need to be conservative.

3. Redis with strict limits

maxmemory 64mb
maxmemory-policy allkeys-lru

Redis without maxmemory can grow indefinitely and trigger the OOM killer. With allkeys-lru, when it hits the limit, it evicts the least recently used keys instead of returning errors.

4. Memory limits on the backend

deploy:
  resources:
    limits:
      memory: 1536M

The backend with loaded embedding models uses ~800MB-1.2GB. The 1536MB limit gives it headroom without allowing a memory leak to consume the entire VPS.

Dockerfiles: Multi-Stage Builds

Backend: Pre-Download Models

# === Stage 1: Builder ===
FROM python:3.12-slim AS builder

WORKDIR /build
COPY requirements.txt .

# PyTorch CPU-only (saves ~1.5GB vs CUDA version)
RUN pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu \
    && pip install --no-cache-dir -r requirements.txt

# Pre-download embedding and cross-encoder models
RUN python -c "
from sentence_transformers import SentenceTransformer, CrossEncoder
SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
"

# === Stage 2: Runtime ===
FROM python:3.12-slim AS runtime

WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
    libfontconfig1 curl ca-certificates && rm -rf /var/lib/apt/lists/*

# Copy dependencies and pre-downloaded models
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY --from=builder /root/.cache /root/.cache

COPY . .
RUN chmod +x start.sh

EXPOSE 8000
CMD ["./start.sh"]

Why pre-download models during build? Without it, the first request after each deploy takes 30-60 seconds while models download. With pre-download, the container starts ready to serve.

Why PyTorch CPU-only? The CUDA version weighs ~2GB extra. On a VPS without GPU it's dead weight.

Frontend: Build + Static Nginx

# === Stage 1: Build ===
FROM node:20-alpine AS build

WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

# === Stage 2: Serve ===
FROM nginx:1.27-alpine

# Copy static build
COPY --from=build /app/dist /usr/share/nginx/html

# SPA config
COPY nginx.conf /etc/nginx/conf.d/default.conf

EXPOSE 5173
CMD ["nginx", "-g", "daemon off;"]

The production frontend does NOT run Vite. It's Nginx serving static files. The image goes from ~400MB (node + deps) to ~25MB (nginx alpine + dist).

Internal Frontend Nginx (SPA)

server {
    listen 5173;
    root /usr/share/nginx/html;
    index index.html;

    # Vite-hashed assets → aggressive cache
    location /assets/ {
        expires 1y;
        add_header Cache-Control "public, immutable";
    }

    # index.html → never cache (so new deploys are reflected)
    location = /index.html {
        add_header Cache-Control "no-cache";
    }

    # SPA fallback: all routes → index.html
    location / {
        try_files $uri $uri/ /index.html;
    }

    gzip on;
    gzip_types text/plain text/css application/json application/javascript;
}

Nginx: The Reverse Proxy That Connects Everything

# === Main HTTPS ===
server {
    listen 443 ssl http2;
    server_name your-domain.com;

    # Origin certificates (Cloudflare → VPS)
    ssl_certificate     /etc/ssl/certs/origin.pem;
    ssl_certificate_key /etc/ssl/private/origin.key;
    ssl_protocols TLSv1.2 TLSv1.3;

    client_max_body_size 50M;  # For document uploads

    # === Maintenance mode ===
    set $maintenance 0;
    if (-f /etc/nginx/maintenance.on) {
        set $maintenance 1;
    }

    # Health check always available (for monitoring)
    location = /api/v1/health {
        proxy_pass http://127.0.0.1:8000;
    }

    # If maintenance mode → 503
    if ($maintenance) {
        return 503;
    }

    # === API Backend ===
    location /api/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # SSE streaming: CRITICAL
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 300s;
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding off;
    }

    # === Widget JS (cacheable) ===
    location = /widget.js {
        proxy_pass http://127.0.0.1:8000;
        proxy_cache_valid 200 1h;
    }

    # === Frontend SPA ===
    location / {
        proxy_pass http://127.0.0.1:5173;
    }

    # === Gzip ===
    gzip on;
    gzip_comp_level 4;
    gzip_min_length 256;
    gzip_types text/plain text/css application/json application/javascript
               text/xml application/xml text/javascript image/svg+xml;

    # === Security headers ===
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
}

# === HTTP → HTTPS redirect ===
server {
    listen 80;
    server_name your-domain.com;
    return 301 https://$server_name$request_uri;
}

# === www → non-www ===
server {
    listen 443 ssl http2;
    server_name www.your-domain.com;
    ssl_certificate     /etc/ssl/certs/origin.pem;
    ssl_certificate_key /etc/ssl/private/origin.key;
    return 301 https://your-domain.com$request_uri;
}

# === 503 Maintenance page ===
error_page 503 @maintenance;
location @maintenance {
    default_type text/html;
    return 503 '<!DOCTYPE html>
    <html><head><meta charset="UTF-8"><title>Maintenance</title>
    <style>body{font-family:system-ui;display:flex;justify-content:center;
    align-items:center;min-height:100vh;background:#0f172a;color:#e2e8f0;
    text-align:center}h1{font-size:2rem}p{color:#94a3b8}</style></head>
    <body><div><h1>Under Maintenance</h1>
    <p>We will be back in a few minutes.</p></div></body></html>';
}

The SSE Block That Almost Broke Everything

location /api/ {
    proxy_buffering off;      # Nginx must NOT buffer the response
    proxy_cache off;          # Nor cache it
    proxy_read_timeout 300s;  # SSE can last minutes
    proxy_set_header Connection '';  # Disable upstream keep-alive
    proxy_http_version 1.1;   # HTTP/1.1 required for chunked
    chunked_transfer_encoding off;
}

Without these headers, Nginx buffers SSE tokens and sends them all at once at the end. The user sees "nothing nothing nothing... entire text at once". These 6 parameters are mandatory for real streaming through a reverse proxy.

The Deploy Script

#!/bin/bash
set -e

SERVER="user@server-ip"
SSH_KEY="~/.ssh/deploy_key"
PROJECT_DIR="/opt/my-app"
BACKUP_DIR="/opt/my-app/backups/db"
DEPLOY_MODE="${1:-full}"  # frontend | backend | full

ssh_cmd() {
    ssh -i "$SSH_KEY" "$SERVER" "$1"
}

echo "=== Deploy: $DEPLOY_MODE ==="

# 1. Push code
git push origin main

# 2. Pull on server
ssh_cmd "cd $PROJECT_DIR && git fetch origin && git reset --hard origin/main"

# 3. Database backup (backend/full only)
if [[ "$DEPLOY_MODE" != "frontend" ]]; then
    echo "Creating DB backup..."
    TIMESTAMP=$(date +%Y%m%d_%H%M%S)
    ssh_cmd "docker exec app-db pg_dump -U \$POSTGRES_USER \$POSTGRES_DB \
        | gzip > $BACKUP_DIR/backup_${TIMESTAMP}.gz"

    # Enable maintenance mode
    ssh_cmd "touch /etc/nginx/maintenance.on && nginx -s reload"
    echo "Maintenance mode: ON"
fi

# 4. Rebuild and restart
case $DEPLOY_MODE in
    frontend)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            build frontend && docker compose -f docker-compose.prod.yml \
            up -d frontend"
        ;;
    backend)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            build backend && docker compose -f docker-compose.prod.yml \
            up -d backend"
        ;;
    full)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            up -d --build"
        ;;
esac

# 5. Disable maintenance mode
if [[ "$DEPLOY_MODE" != "frontend" ]]; then
    sleep 15  # Wait for backend to load models
    ssh_cmd "rm -f /etc/nginx/maintenance.on && nginx -s reload"
    echo "Maintenance mode: OFF"
fi

# 6. Health check
echo "Checking health..."
MAX_RETRIES=30
for i in $(seq 1 $MAX_RETRIES); do
    STATUS=$(ssh_cmd "curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8000/api/v1/health")
    if [[ "$STATUS" == "200" ]]; then
        echo "Deploy successful. Health: OK"
        exit 0
    fi
    echo "Attempt $i/$MAX_RETRIES... (status: $STATUS)"
    sleep 5
done

echo "ERROR: Health check failed after $MAX_RETRIES attempts"
exit 1

Why Maintenance Mode?

The backend takes ~15 seconds to start: it runs Alembic migrations and then Uvicorn loads the embedding models. Without maintenance mode, during those 15 seconds Nginx returns 502 Bad Gateway.

With the /etc/nginx/maintenance.on file, Nginx returns a styled 503 page. The health check (/api/v1/health) is exempted so the script can verify when the backend is ready.

Backend Start Script

#!/bin/bash
set -e

echo "Running database migrations..."
alembic upgrade head

echo "Starting FastAPI server..."

# 1 worker = ~800MB with loaded models
# 2 workers = ~1.3GB (only if you have RAM to spare)
WORKERS="${UVICORN_WORKERS:-1}"

exec uvicorn app.main:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers "$WORKERS" \
    --log-level "${UVICORN_LOG_LEVEL:-info}" \
    "$@"

Why 1 worker? Each Uvicorn worker loads its own copy of the embedding models (~500MB). With 2 workers you're already at 1.3GB for backend alone. On a 4GB VPS with PostgreSQL, Redis and Nginx, 1 worker is the safe choice.

FastAPI is async, so 1 worker handles concurrency well — I/O operations (DB, LLM API, Redis) don't block the event loop.

Automated Maintenance with Cron

#!/bin/bash
# Run: crontab -e → 0 4 * * 0 /opt/my-app/scripts/maintenance.sh

LOG="/var/log/app-maintenance.log"
echo "=== Maintenance $(date) ===" >> "$LOG"

# 1. Clean orphaned Docker images (>7 days)
docker image prune -f --filter "until=168h" >> "$LOG" 2>&1

# 2. Clean stopped containers
docker container prune -f --filter "until=168h" >> "$LOG" 2>&1

# 3. Check disk usage
DISK_USAGE=$(df / --output=pcent | tail -1 | tr -dc '0-9')
if [ "$DISK_USAGE" -gt 80 ]; then
    echo "ALERT: Disk at ${DISK_USAGE}%" >> "$LOG"
fi

# 4. Check memory
MEM_USAGE=$(free | awk '/Mem:/ {printf("%.0f"), $3/$2 * 100}')
if [ "$MEM_USAGE" -gt 90 ]; then
    echo "ALERT: RAM at ${MEM_USAGE}%" >> "$LOG"
fi

# 5. Container status
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Size}}" >> "$LOG"

# 6. Backup inventory
BACKUP_COUNT=$(ls /opt/my-app/backups/db/*.gz 2>/dev/null | wc -l)
LATEST=$(ls -t /opt/my-app/backups/db/*.gz 2>/dev/null | head -1)
echo "Backups: $BACKUP_COUNT files. Latest: $LATEST" >> "$LOG"

echo "=== Done ===" >> "$LOG"

This runs every Sunday at 4am. It cleans Docker garbage, checks disk and RAM, and logs everything. Simple but effective — it saved me twice from running out of disk due to accumulated Docker images.

Cloudflare + SSL: The Configuration

Internet → Cloudflare (proxy) → VPS (Nginx 443) → Docker containers

Setup

DNS on Cloudflare: A record pointing to VPS, proxy enabled (orange cloud)
SSL on Cloudflare: "Full (strict)" — encrypts both browser→Cloudflare and Cloudflare→VPS
Origin certificate: Generated in Cloudflare Dashboard → SSL/TLS → Origin Server → Create Certificate
On the VPS: Copy the certificate and key to /etc/ssl/certs/origin.pem and /etc/ssl/private/origin.key

Why not Let's Encrypt? With Cloudflare proxy enabled, Let's Encrypt can't verify the domain via HTTP challenge (Cloudflare intercepts). Cloudflare origin certificates last 15 years and are configured in 2 minutes.

Bonus: Cloudflare as Free CDN

With the proxy enabled, Cloudflare automatically caches static assets (JS, CSS, images). The VPS only receives API requests and HTML. This significantly reduces server traffic.

RAM Distribution

Here's how it looks on a 4GB VPS:

┌──────────────────────────────────────────┐
│              4GB RAM Total               │
├──────────────────────────────────────────┤
│  OS + Docker Engine      ~400MB          │
│  PostgreSQL              ~200-400MB      │
│  Backend (1 worker)      ~800MB-1.2GB    │
│  Redis                   ≤64MB           │
│  Nginx + Frontend        ~30MB           │
│  Free / Buffer           ~1.5-2GB        │
├──────────────────────────────────────────┤
│  Swap (2GB)              emergency       │
└──────────────────────────────────────────┘

The ~1.5-2GB free are for:

OS filesystem cache (helps PostgreSQL)
Traffic spikes
Maintenance operations (backups, builds)

Tip: Always configure swap (2GB) as a safety net. Without swap, the OOM killer terminates processes without warning when RAM fills up.

fallocate -l 2G /swapfile
chmod 600 /swapfile
mkswap /swapfile
swapon /swapfile
echo '/swapfile none swap sw 0 0' >> /etc/fstab

Security Checklist

Before considering the deploy "ready":

[x] Docker ports on localhost only (127.0.0.1:port:port)
[x] Firewall active (ufw allow 22,80,443/tcp && ufw enable)
[x] SSH key-only (disable password auth in /etc/ssh/sshd_config)
[x] .env.production outside the repo (.gitignore)
[x] Secrets never in compose files (use env_file reference)
[x] DB without external port (only accessible via Docker network)
[x] Automated backups with periodic verification
[x] Health check in the deploy script
[x] Swap configured to prevent OOM kills

Real Numbers

Metric	Value
Build time (backend)	~3-4 min (first time), ~1 min (cached)
Build time (frontend)	~30s
Full deploy time	~2-3 min
Visible downtime	0s (maintenance page during rebuild)
Backend image size	~2.1GB (includes PyTorch CPU + models)
Frontend image size	~25MB (nginx + dist)
Stable RAM usage	~2-2.5GB of 4GB
Monthly cost	~$24 (VPS) + $0 (Cloudflare free)
Uptime last month	99.8% (one outage for OS update)

Lessons Learned

1. Docker + Firewall = Be Careful

Docker modifies iptables directly. If you publish a port without 127.0.0.1, ufw deny won't block it. Always use 127.0.0.1: in production and let Nginx be the only entry point.

2. Pre-Download ML Models in the Build

If models download at runtime, the first cold start takes a minute. Worse: if HuggingFace has downtime, your deploy fails. Pre-downloading in the Dockerfile eliminates both problems.

3. 1 Uvicorn Worker Is Enough (If It's Async)

The temptation is to set 4 workers "just in case". But each one loads ~500MB of models. With async FastAPI, a single worker handles hundreds of concurrent requests. Scale workers only when you have evidence that CPU is the bottleneck.

4. Maintenance Mode > Zero-Downtime Rolling Deploys

For a single-node VPS, a rolling deploy requires complex orchestration. A 503 page for 15 seconds is infinitely simpler and nobody complains — especially if the health check guarantees automatic deactivation.

5. PostgreSQL Defaults Are for Dedicated Servers

PostgreSQL defaults assume it has all the RAM to itself. On a shared VPS with 4 other services, not tuning PostgreSQL is a guaranteed OOM kill. shared_buffers, effective_cache_size and max_connections are the first parameters to adjust.

6. Cloudflare Origin Certs > Let's Encrypt

With proxy enabled, Let's Encrypt is more complicated to configure and renew. Cloudflare origin certs last 15 years, are configured once and forgotten.

What's Next

Monitoring with Prometheus + Grafana: Latency, errors, and resource usage metrics (currently logs + cron only)
Offsite backup: Copy backups to an S3/R2 bucket instead of storing them only on the same VPS
Blue-green deploys: When traffic justifies a second VPS
CI/CD with GitHub Actions: Automate the deploy script (currently a manual ./deploy.sh backend)

Conclusion

Deploying a RAG system isn't like deploying a CRUD app. Embedding models consume real RAM, SSE streaming needs specific Nginx configuration, and PostgreSQL with pgvector needs careful tuning on limited servers.

But you also don't need Kubernetes or a $200/month cluster. A $24 VPS with well-configured Docker Compose, Nginx as reverse proxy, and deploy scripts with maintenance mode + health checks is enough to serve thousands of queries per day with consistent latencies.

Most importantly: measure first, scale later. Starting with 1 worker, 4GB of RAM and basic monitoring gives you all the information you need to make infrastructure decisions based on real data, not estimates.

This is the second article in the series. The first covers the RAG pipeline architecture. If it was useful, a like helps it reach more people. Questions about any part of the deploy? Drop a comment.

Multi-tenancy real con FastAPI y PostgreSQL: planes, cuotas y aislamiento de datos

Martin Palopoli — Tue, 24 Mar 2026 13:21:00 +0000

Implementé multi-tenancy real en un motor RAG SaaS: tenants con planes, enforcement de cuotas con contadores atómicos (UPSERT), rate limiting con Redis (sliding window), RBAC con 3 roles, y aislamiento de datos donde cada query se filtra por tenant_id. En este artículo comparto la arquitectura completa, las trampas que evité, y el código clave.

Por qué "agregar un tenant_id" no es multi-tenancy

La mayoría de los tutoriales de multi-tenancy se quedan en "agrega una columna tenant_id y filtra en cada query". Eso es el 10% del problema. En producción vas a necesitar:

Planes con límites reales que no se puedan burlar con requests concurrentes
Contadores atómicos que no pierdan incrementos bajo carga
Rate limiting que sobreviva a reinicios del servidor
Roles que restrinjan acciones, no solo visibilidad
Aislamiento real donde un bug en un service no exponga datos de otro tenant
Billing cycles que se reinicien correctamente cada mes

Este artículo muestra cómo resolví cada uno en un sistema real con FastAPI async y PostgreSQL.

El stack

Componente	Tecnología
Backend	Python 3.12 + FastAPI (100% async)
Base de datos	PostgreSQL 16
ORM	SQLAlchemy async + Alembic
Cache/Rate limit	Redis 7
Auth	JWT (access 30min, refresh 7d con rotación)
Passwords	bcrypt (via passlib)

Diseño de base de datos

Las 4 tablas fundamentales

┌──────────────┐     ┌──────────────┐
│    plans     │     │   tenants    │
├──────────────┤     ├──────────────┤
│ id (UUID)    │◄────│ plan_id (FK) │
│ name         │     │ id (UUID)    │
│ slug         │     │ name         │
│ max_users    │     │ slug (unique)│
│ max_kbs      │     │ is_active    │
│ max_documents│     │ plan_started │
│ max_storage  │     │ created_at   │
│ max_messages │     └──────┬───────┘
│ max_tokens   │            │
│ max_api_keys │     ┌──────┴───────┐
│ price_monthly│     │    users     │
│ max_concurrent│    ├──────────────┤
└──────────────┘     │ id (UUID)    │
                     │ tenant_id(FK)│
                     │ email        │
                     │ role (enum)  │
                     │ is_active    │
                     └──────────────┘

┌───────────────────────┐
│ tenant_monthly_usage  │
├───────────────────────┤
│ id (UUID)             │
│ tenant_id (FK)        │
│ billing_period (str)  │
│ messages_count (int)  │
│ tokens_used (bigint)  │
│ UNIQUE(tenant_id,     │
│   billing_period)     │
└───────────────────────┘

Modelo del Plan

Cada plan define todos los límites que el sistema va a enforcar:

class Plan(Base):
    __tablename__ = "plans"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    slug: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    max_users: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_concurrent: Mapped[int] = mapped_column(Integer, nullable=False, default=2)
    max_kbs: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    max_documents: Mapped[int] = mapped_column(Integer, nullable=False, default=20)
    max_storage_mb: Mapped[int] = mapped_column(Integer, nullable=False, default=200)
    max_messages_month: Mapped[int] = mapped_column(Integer, nullable=False, default=500)
    max_tokens_month: Mapped[int] = mapped_column(BigInteger, nullable=False, default=200000)
    max_api_keys: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
    price_monthly: Mapped[float | None] = mapped_column(Numeric(10, 2), nullable=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)

Decisión clave: -1 significa ilimitado. En vez de tener un booleano is_unlimited por cada recurso, uso -1 como convención. Simplifica todas las comparaciones:

if plan.max_users == -1:
    return  # Ilimitado, no verificar

Modelo del Tenant

class Tenant(Base):
    __tablename__ = "tenants"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    slug: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    plan_id: Mapped[uuid.UUID | None] = mapped_column(
        UUID(as_uuid=True), ForeignKey("plans.id"), nullable=True
    )
    plan_started_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True, default=None
    )
    notification_preferences: Mapped[dict | None] = mapped_column(JSONB, nullable=True)

plan_started_at es crucial: define el ancla del ciclo de billing. Cuando un tenant pasa de Free a un plan pago, se resetea al momento del pago. Esto determina cuándo se reinician los contadores mensuales.

Modelo del Usuario

class UserRole(str, PyEnum):
    SUPER_ADMIN = "super_admin"
    TENANT_ADMIN = "tenant_admin"
    MEMBER = "member"


class User(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    full_name: Mapped[str] = mapped_column(String(255), default="")
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole, name="user_role", values_callable=lambda e: [x.value for x in e]),
        default=UserRole.MEMBER,
    )
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    email_verified: Mapped[bool] = mapped_column(Boolean, default=False)

Un usuario pertenece a exactamente un tenant. No hay usuarios multi-tenant. Si necesitás que alguien acceda a dos organizaciones, crea dos cuentas. Simplifica enormemente el modelo de permisos.

RBAC: 3 roles, 0 ambiguedad

Los roles

Rol	Alcance	Puede hacer
`super_admin`	Plataforma	Ver todos los tenants, cambiar planes, ver métricas globales
`tenant_admin`	Su tenant	CRUD de KBs, documentos, FAQs, invitar usuarios, ver analytics
`member`	Su tenant (restringido)	Chat, ver KBs asignadas, feedback

Dependency injection para autorización

FastAPI tiene un sistema de dependencias que es perfecto para RBAC. Creé un factory de dependencias:

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")


async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    payload = decode_access_token(token)
    user_id = payload.get("sub")
    if not user_id:
        raise UnauthorizedException("Token payload invalido")

    user = await get_user_by_id(db, UUID(user_id))
    if not user:
        raise UnauthorizedException("Usuario no encontrado")
    if not user.is_active:
        raise UnauthorizedException("Cuenta desactivada")

    # Verificar que el tenant esté activo (super_admin bypasses)
    if user.role != UserRole.SUPER_ADMIN:
        tenant = await db.get(Tenant, user.tenant_id)
        if not tenant or not tenant.is_active:
            raise UnauthorizedException("Tenant desactivado")

    return user

Detalle sutil: el super_admin bypasea la verificación de tenant activo. Si desactivamos un tenant malicioso, el super_admin todavía puede entrar a investigar.

Factory para restricción por rol

def require_role(*roles):
    """Dependency factory: restringe endpoint a roles específicos."""
    async def _check(current_user=Depends(get_current_user)):
        if current_user.role not in roles:
            raise ForbiddenException("No tienes permisos para esta accion")
        return current_user
    return _check


# Shortcuts
def require_super_admin():
    return require_role(UserRole.SUPER_ADMIN)

def require_tenant_admin_or_above():
    return require_role(UserRole.SUPER_ADMIN, UserRole.TENANT_ADMIN)

Uso en los endpoints:

@router.get("/tenants", response_model=list[TenantResponse])
async def list_tenants(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    if current_user.role != UserRole.SUPER_ADMIN:
        raise ForbiddenException("Solo super_admin puede listar todos los tenants")
    return await tenant_service.list_tenants(db)


@router.put("/{kb_id}/access")
async def update_kb_access(
    kb_id: UUID,
    data: KBAccessConfig,
    db: AsyncSession = Depends(require_tenant_admin_or_above()),
):
    # Solo tenant_admin o super_admin llegan acá
    ...

Aislamiento de datos: tenant_id en todo

El patrón

Cada tabla que contiene datos de usuario tiene una columna tenant_id (directa o transitiva). No hay excepciones.

class KnowledgeBase(Base):
    __tablename__ = "knowledge_bases"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

class ApiKey(Base):
    __tablename__ = "api_keys"
    # ...
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )

Filtrado en cada service

Cada función que lista o busca datos filtra por tenant_id del usuario autenticado:

async def list_knowledge_bases(db: AsyncSession, user: User) -> list[dict]:
    stmt = (
        select(KnowledgeBase, func.count(Document.id).label("document_count"))
        .outerjoin(Document, Document.knowledge_base_id == KnowledgeBase.id)
    )

    if user.role == UserRole.SUPER_ADMIN:
        # Super admin ve solo sus propias KBs + system KBs
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
    else:
        stmt = stmt.where(or_(
            KnowledgeBase.tenant_id == user.tenant_id,
            KnowledgeBase.is_system == True,
        ))
        # Para members, aplicar filtro de acceso
        access_filter = await get_accessible_kb_filter(user)
        if access_filter is not None:
            stmt = stmt.where(or_(KnowledgeBase.is_system == True, access_filter))

    stmt = stmt.group_by(KnowledgeBase.id).order_by(KnowledgeBase.created_at.desc())
    result = await db.execute(stmt)
    # ...

Leccion importante: el super_admin NO ve los datos de otros tenants. Es el operador de la plataforma. Ve estadisticas agregadas (total tenants, revenue, etc.) pero no puede leer las conversaciones ni KBs de otros. Esto fue una decision deliberada de privacidad.

Acceso granular a KBs

Dentro de un mismo tenant, no todos los miembros ven todas las KBs. Hay un sistema de control de acceso:

async def _check_kb_access(db: AsyncSession, kb: KnowledgeBase, user: User) -> None:
    has_access = await user_can_access_kb(db, user, kb)
    if not has_access:
        raise ForbiddenException("No tienes acceso a esta base de conocimiento")

Las KBs pueden ser:

Publicas dentro del tenant: todos los miembros las ven
Privadas: solo miembros asignados via grupos las ven
System: KBs del sistema, visibles para todos, solo modificables por super_admin

CASCADE: borrado limpio

Todas las foreign keys usan ondelete="CASCADE". Cuando se borra un tenant, todo se borra en cascada: usuarios, KBs, documentos, chunks, conversaciones, API keys, usage logs. Una sola operacion:

async def delete_tenant(db: AsyncSession, tenant_id: UUID, current_user: User) -> dict:
    tenant = await get_tenant(db, tenant_id)

    # Safety checks
    if current_user.tenant_id == tenant_id:
        raise BadRequestException("No puedes eliminar tu propio tenant")
    if tenant.slug == "system-tenant":
        raise ForbiddenException("No se puede eliminar el tenant del sistema")

    # 1. Cancelar suscripcion activa
    try:
        await admin_cancel_subscription(db, tenant_id, immediate=True)
    except NotFoundException:
        pass

    # 2. Limpiar cache Redis
    redis = await get_redis()
    if redis:
        keys = await redis.keys(f"*:{tenant_id}:*")
        if keys:
            await redis.delete(*keys)

    # 3. DELETE → CASCADE hace el resto
    await db.delete(tenant)
    await db.commit()

Plan enforcement: contadores atomicos con UPSERT

El problema

Imaginate que un tenant en plan Free tiene limite de 50 mensajes/mes. Dos usuarios mandan un mensaje al mismo tiempo. Si hacés SELECT count → check → INSERT, tenés una race condition: ambos leen 49, ambos pasan el check, ambos insertan. Ahora hay 51.

La solucion: UPSERT atomico

La tabla tenant_monthly_usage usa un constraint UNIQUE en (tenant_id, billing_period):

class TenantMonthlyUsage(Base):
    __tablename__ = "tenant_monthly_usage"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    tenant_id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False
    )
    billing_period: Mapped[str] = mapped_column(String(10), nullable=False)
    messages_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    tokens_used: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)

    __table_args__ = (
        UniqueConstraint("tenant_id", "billing_period", name="uq_tenant_billing_usage"),
    )

El incremento usa INSERT ... ON CONFLICT DO UPDATE:

async def increment_message_count(db: AsyncSession, tenant_id: UUID) -> None:
    anchor = await _get_billing_anchor(db, tenant_id)
    bp = _get_billing_period_key(anchor)
    await db.execute(text("""
        INSERT INTO tenant_monthly_usage (id, tenant_id, billing_period, messages_count, tokens_used)
        VALUES (gen_random_uuid(), :tid, :bp, 1, 0)
        ON CONFLICT (tenant_id, billing_period)
        DO UPDATE SET messages_count = tenant_monthly_usage.messages_count + 1,
                      updated_at = now()
    """), {"tid": tenant_id, "bp": bp})
    await db.commit()

¿Por qué esto funciona? PostgreSQL ejecuta el INSERT ... ON CONFLICT DO UPDATE de forma atómica. No hay ventana de race condition. Si dos requests llegan al mismo tiempo, uno hara INSERT y el otro hara UPDATE, pero el contador siempre será correcto.

Detalle anti-fraude: los contadores solo incrementan. No hay endpoint para decrementar. No hay forma de que un usuario manipule su usage. La unica forma de "resetear" es que cambie el billing period (es decir, que pase un mes).

Calculo del billing period

def _get_billing_period_key(anchor: datetime) -> str:
    now = datetime.now(timezone.utc)
    billing_day = anchor.day

    # Clamping: si el anchor es dia 31 y estamos en febrero (28 dias),
    # el billing day efectivo es 28
    max_day = calendar.monthrange(now.year, now.month)[1]
    effective_day = min(billing_day, max_day)

    if now.day >= effective_day:
        return f"{now.year}-{now.month:02d}-{effective_day:02d}"
    else:
        # El periodo actual empezo el mes pasado
        if now.month == 1:
            prev_year, prev_month = now.year - 1, 12
        else:
            prev_year, prev_month = now.year, now.month - 1
        max_day_prev = calendar.monthrange(prev_year, prev_month)[1]
        effective_day_prev = min(billing_day, max_day_prev)
        return f"{prev_year}-{prev_month:02d}-{effective_day_prev:02d}"

¿Por qué es complicado? Porque los meses no tienen la misma cantidad de dias. Si un tenant pagó el 31 de enero, su siguiente ciclo no empieza el 31 de febrero (no existe). El clamping maneja esto.

Verificacion de limites

Antes de ejecutar el pipeline RAG, verifico los limites:

async def check_message_limit(db: AsyncSession, tenant_id: UUID) -> None:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return
    if plan.max_messages_month == -1:
        return  # Ilimitado
    usage = await get_monthly_usage(db, tenant_id)
    if usage["messages_month"] >= plan.max_messages_month:
        raise PlanLimitException(
            f"Limite de mensajes mensuales alcanzado "
            f"({plan.max_messages_month} max para plan {plan.name})"
        )

PlanLimitException devuelve HTTP 403, no 429. Es una restriccion de plan, no de rate limiting. El frontend distingue entre "pagá mas" (403) y "esperá un rato" (429).

Limites en todos los recursos

No solo mensajes. Cada recurso tiene su check:

# Antes de crear un usuario
await check_user_limit(db, tenant_id)

# Antes de crear una KB
await check_kb_limit(db, tenant_id)

# Antes de subir un documento
await check_document_limit(db, tenant_id)
await check_storage_limit(db, tenant_id, new_file_size_bytes=file_size)

# Antes de crear una API key
await check_api_key_limit(db, tenant_id)

# Antes de enviar un mensaje al LLM
await check_message_limit(db, tenant_id)
await check_concurrent_limit(db, tenant_id)

El patron es siempre el mismo:

async def check_kb_limit(db: AsyncSession, tenant_id: UUID) -> None:
    tenant = await get_tenant(db, tenant_id)
    if not tenant.plan_id:
        return
    plan = await db.get(Plan, tenant.plan_id)
    if not plan or plan.max_kbs == -1:
        return
    stmt = select(func.count(KnowledgeBase.id)).where(KnowledgeBase.tenant_id == tenant_id)
    result = await db.execute(stmt)
    current_count = result.scalar() or 0
    if current_count >= plan.max_kbs:
        raise PlanLimitException(
            f"Limite de bases de conocimiento alcanzado ({plan.max_kbs} max)"
        )

Rate limiting con Redis: sliding window

Por que Redis y no PostgreSQL

Los contadores de billing period estan en PostgreSQL porque necesitan durabilidad. El rate limiting esta en Redis porque necesita velocidad y TTL automatico.

Si el servidor se reinicia a mitad de un stream, no quiero un contador "fantasma" en PostgreSQL que bloquee al tenant. Redis con TTL se limpia solo.

Sliding window con sorted sets

import time
import redis.asyncio as redis

async def is_allowed(key: str, max_requests: int, window_seconds: int = 60) -> bool:
    client = await get_redis()
    if client is None:
        return True  # Fail-open

    redis_key = f"ratelimit:{key}"
    now = time.time()
    cutoff = now - window_seconds

    try:
        pipe = client.pipeline()
        pipe.zremrangebyscore(redis_key, 0, cutoff)   # Limpiar expirados
        pipe.zcard(redis_key)                          # Contar actuales
        pipe.zadd(redis_key, {str(now): now})          # Agregar este request
        pipe.expire(redis_key, window_seconds + 1)     # TTL safety net
        results = await pipe.execute()

        current_count = results[1]
        if current_count >= max_requests:
            await client.zrem(redis_key, str(now))     # Rollback
            return False
        return True
    except Exception:
        return True  # Fail-open

¿Por qué sorted sets y no un simple INCR? Porque necesito una ventana deslizante, no una ventana fija. Con INCR, si un usuario manda 60 requests en el segundo 59 del minuto, y 60 mas en el segundo 1 del siguiente minuto, pasaría 120 requests en 2 segundos. Con sorted sets, cada request tiene su timestamp y la ventana se desliza continuamente.

¿Por qué fail-open? Si Redis se cae, prefiero dejar pasar requests (degradacion graciosa) que bloquear a todos los usuarios. Los limites de billing en PostgreSQL siguen activos como red de seguridad.

Concurrencia de streams

Para limitar conexiones SSE simultaneas por tenant, uso un patron diferente: INCR/DECR con TTL de seguridad.

_CONCURRENT_TTL = 300  # 5 minutos

async def increment_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        pipe = client.pipeline()
        pipe.incr(redis_key)
        pipe.expire(redis_key, _CONCURRENT_TTL)
        results = await pipe.execute()
        return results[0]
    except Exception:
        return 0


async def decrement_concurrent(tenant_id: str) -> int:
    client = await get_redis()
    if client is None:
        return 0

    redis_key = f"concurrent:{tenant_id}"
    try:
        count = await client.decr(redis_key)
        if count <= 0:
            await client.delete(redis_key)
            return 0
        await client.expire(redis_key, _CONCURRENT_TTL)
        return count
    except Exception:
        return 0

¿Por qué TTL de 5 minutos? Si el servidor crashea durante un stream, el DECR nunca se ejecuta y el contador queda inflado. El TTL auto-cura esta situacion: despues de 5 minutos sin actividad, el contador se borra solo.

El uso en el endpoint de chat:

@router.post("/conversations/{conv_id}/messages/stream")
async def stream_chat(conv_id: UUID, data: MessageCreate, ...):
    # Verificar presupuesto LLM (soft check)
    faq_only_mode = await is_llm_budget_exhausted(db, current_user.tenant_id)
    if not faq_only_mode:
        await check_concurrent_limit(db, current_user.tenant_id)

    # ... preparar pipeline ...

    async def event_generator():
        await increment_concurrent(str(current_user.tenant_id))
        try:
            async for event in _event_generator_inner():
                yield event
        finally:
            # SIEMPRE decrementa, incluso si hay error
            await decrement_concurrent(str(current_user.tenant_id))

    return EventSourceResponse(event_generator())

El finally es critico. Sin el, cualquier excepcion durante el streaming dejaria el contador inflado.

Registro: un tenant por usuario, automatico

Cuando alguien se registra, se crea automaticamente un tenant con plan Free:

async def register(db: AsyncSession, data: UserRegister) -> User:
    existing = await get_user_by_email(db, data.email)
    if existing:
        raise BadRequestException("El email ya esta registrado")

    # Crear tenant automaticamente
    tenant = await create_tenant(db, data.tenant_name, plan_slug="free")

    user = User(
        email=data.email,
        hashed_password=hash_password(data.password),
        full_name=data.full_name,
        role=UserRole.TENANT_ADMIN,  # El creador es admin
        tenant_id=tenant.id,
        email_verified=False,
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

El slug del tenant se genera automaticamente con deduplicacion:

def _generate_slug(name: str) -> str:
    slug = re.sub(r'[^a-z0-9]+', '-', name.lower()).strip('-')
    return slug[:80] if slug else 'tenant'


async def create_tenant(db: AsyncSession, name: str, plan_slug: str = "free") -> Tenant:
    # Buscar plan
    stmt = select(Plan).where(Plan.slug == plan_slug, Plan.is_active == True)
    result = await db.execute(stmt)
    plan = result.scalar_one_or_none()

    base_slug = _generate_slug(name)
    slug = base_slug

    # Si ya existe, agregar sufijo aleatorio
    existing = await db.execute(select(Tenant).where(Tenant.slug == slug))
    if existing.scalar_one_or_none():
        slug = f"{base_slug}-{uuid.uuid4().hex[:6]}"

    tenant = Tenant(name=name, slug=slug, plan_id=plan.id if plan else None)
    db.add(tenant)
    await db.flush()
    return tenant

JWT: tenant_id en el token

El access token incluye tenant_id y role directamente en el payload:

def create_access_token(user_id: str, tenant_id: str, role: str,
                        expire_minutes: int | None = None) -> str:
    minutes = expire_minutes or settings.access_token_expire_minutes
    expire = datetime.now(timezone.utc) + timedelta(minutes=minutes)
    payload = {
        "sub": user_id,
        "tenant_id": tenant_id,
        "role": role,
        "exp": expire,
        "type": "access",
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

¿Por qué incluir tenant_id en el JWT? Para no tener que hacer una query a la DB en cada request solo para saber de qué tenant es el usuario. En get_current_user igual verifico contra la DB (por si el usuario fue desactivado), pero el tenant_id del token sirve como hint rápido.

Refresh token rotation

Cada vez que se usa un refresh token, se revoca y se emite uno nuevo:

async def refresh_access_token(db: AsyncSession, refresh_token_value: str):
    stmt = select(RefreshToken).where(
        RefreshToken.token == refresh_token_value,
        RefreshToken.revoked == False,
    )
    result = await db.execute(stmt)
    token_record = result.scalar_one_or_none()

    if not token_record:
        raise UnauthorizedException("Refresh token invalido")

    # Revocar el viejo (rotacion)
    token_record.revoked = True

    user = await get_user_by_id(db, token_record.user_id)
    if not user or not user.is_active:
        await db.commit()
        raise UnauthorizedException("Usuario no encontrado o desactivado")

    # Emitir nuevos tokens
    new_access = create_access_token(str(user.id), str(user.tenant_id), user.role.value)
    new_refresh_value = create_refresh_token_value()
    new_refresh = RefreshToken(
        user_id=user.id,
        token=new_refresh_value,
        expires_at=datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_days),
    )
    db.add(new_refresh)
    await db.commit()

    return new_access, new_refresh_value

Si alguien intercepta un refresh token y lo usa, el token original queda revocado. Cuando el usuario legítimo intente refrescar, va a fallar y sabrá que hubo un compromiso.

Threshold warnings: avisar antes de que explote

No espero a que el tenant llegue al 100% del límite para avisarle. Implementé warnings en 80% y 95%:

async def get_usage_warnings(db: AsyncSession, tenant_id: UUID) -> list[dict]:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return []

    usage = await get_monthly_usage(db, tenant_id)
    warnings = []

    resources = [
        ("messages", usage["messages_month"], plan.max_messages_month),
        ("tokens", usage["tokens_month"], plan.max_tokens_month),
    ]

    for resource, used, max_val in resources:
        if max_val == -1 or max_val <= 0:
            continue
        pct = round((used / max_val) * 100, 1)
        if pct >= 95:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "critical"
            })
        elif pct >= 80:
            warnings.append({
                "resource": resource, "used": used, "max": max_val,
                "pct": pct, "level": "warning"
            })

    return warnings

Los warnings se envian por email a los admins del tenant, con deduplicacion para no spammear:

# Solo enviar si:
# 1. No se envió para este billing period, O
# 2. Se está escalando de warning → critical
if last_sent == bp and last_level == "critical":
    return  # Ya envié critical para este periodo

# Guardar marker de dedup
updated_prefs["_last_warning_bp"] = bp
updated_prefs["_last_warning_level"] = most_severe["level"]

FAQ-only fallback: degradacion graciosa

Cuando un tenant agota su presupuesto de LLM, no bloqueo todo. Las FAQs siguen funcionando:

async def is_llm_budget_exhausted(db: AsyncSession, tenant_id: UUID) -> bool:
    plan, tenant = await _get_plan_and_tenant(db, tenant_id)
    if not plan:
        return False

    usage = await get_monthly_usage(db, tenant_id)

    if plan.max_messages_month != -1 and usage["messages_month"] >= plan.max_messages_month:
        return True
    if plan.max_tokens_month != -1 and usage["tokens_month"] >= plan.max_tokens_month:
        return True
    return False

En el chat, si is_llm_budget_exhausted devuelve True, se intenta FAQ match primero. Si matchea, respuesta gratuita. Si no, se envia un evento SSE upgrade_required y el frontend muestra un card de upgrade.

Custom exceptions: errores semánticos

class PlanLimitException(HTTPException):
    def __init__(self, detail: str = "Plan limit reached"):
        super().__init__(status_code=status.HTTP_403_FORBIDDEN, detail=detail)

class TooManyRequestsException(HTTPException):
    def __init__(self, detail: str = "Too many requests"):
        super().__init__(status_code=429, detail=detail)

El frontend distingue:

403 PlanLimit: "Llegaste al limite de tu plan. Upgrade disponible."
429 RateLimit: "Demasiados requests. Reintentá en unos segundos."
401 Unauthorized: "Sesion expirada. Volvé a iniciar sesion."

Numeros que importan

Recurso	Free	Starter	Pro
Usuarios	3	10	50
Knowledge Bases	3	10	-1 (ilimitado)
Documentos	20	100	-1
Storage	200 MB	1 GB	10 GB
Mensajes IA/mes	50	500	5000
API Keys	1	5	20
Concurrencia	2	5	20

Lecciones aprendidas

1. UPSERT > SELECT + INSERT

Nunca hagas "read count → check → insert" para contadores de uso. La race condition es inevitable bajo carga. El UPSERT atomico de PostgreSQL es la unica forma correcta. Me ahorré horas de debugging de "por qué el tenant tiene 51 mensajes si el limite es 50".

2. Fail-open para rate limiting, fail-closed para billing

Si Redis se cae, los rate limits se relajan (fail-open) pero los limites de plan en PostgreSQL siguen activos (fail-closed). Nunca dejes que una dependencia caida bloquee a todos tus usuarios.

3. El super_admin no es un dios omnisciente

Mi primer diseño daba acceso total al super_admin sobre los datos de todos los tenants. Mal. El super_admin es el operador de la plataforma, no el dueño de los datos. Ve metricas agregadas, no conversaciones individuales. Esto simplifica compliance y genera confianza.

4. `-1` para ilimitado es mejor que un booleano

Tener max_kbs: int con -1 = ilimitado es mas limpio que max_kbs: int + is_kbs_unlimited: bool. Una sola columna, un solo check. Lo apliqué a todos los limites de plan.

5. Billing period basado en fecha de pago, no en mes calendario

Si un tenant pagó el 15 de marzo, su ciclo va del 15/3 al 14/4, no del 1/3 al 31/3. Parece un detalle menor hasta que un usuario se queja de que "pagué ayer y ya usé la mitad de mi cuota".

6. TTL como safety net para contadores en Redis

Los contadores de concurrencia en Redis necesitan TTL. Sin el, un crash del servidor deja contadores "zombie" que bloquean al tenant. Con TTL de 5 minutos, el sistema se auto-cura.

7. CASCADE simplifica el borrado pero hay que limpiar Redis tambien

El ON DELETE CASCADE de PostgreSQL es maravilloso para borrar todo lo relacionado a un tenant. Pero no limpia Redis. Hay que hacerlo explicitamente. Me enteré cuando un tenant borrado seguia contando rate limits.

Lo que sigue

Audit log: registrar quién hizo qué, cuándo, sobre qué recurso
Fine-grained permissions: permisos por KB y por accion (no solo por rol global)
Multi-region: tenants asignados a regiones especificas para compliance

Conclusion

Multi-tenancy real no es una columna extra en la base de datos. Es un sistema de enforcement que abarca desde el modelo de datos hasta el rate limiting, pasando por la autorizacion y el billing. Los puntos mas criticos:

Contadores atomicos (UPSERT) para que los limites se respeten bajo concurrencia
Redis para lo efimero (rate limits, concurrencia) y PostgreSQL para lo durable (billing, usage)
Roles claros con dependency injection que haga imposible saltear la autorizacion
Fail-open donde duele poco, fail-closed donde duele mucho
Aislamiento por defecto — cada query filtra por tenant_id, sin excepciones

Si estas construyendo un SaaS multi-tenant, invertí en el enforcement desde el dia 1. Refactorear esto despues es ordenes de magnitud mas doloroso que hacerlo bien desde el principio.

Si estas construyendo algo multi-tenant y te chocaste con un problema que no cubrí, dejá un comentario. Y si este articulo te ahorro tiempo, un like ayuda a que llegue a mas personas.

Cómo desplegué un motor RAG en producción con Docker, Nginx y DigitalOcean

Martin Palopoli — Tue, 17 Mar 2026 12:28:14 +0000

Desplegué un motor RAG completo (FastAPI + PostgreSQL + pgvector + Redis) en un VPS de 4GB de RAM por $24/mes. En este artículo comparto la arquitectura de deploy real: Docker multi-stage builds, PostgreSQL tuneado para recursos limitados, Nginx como reverse proxy con soporte SSE, zero-downtime deploys con maintenance mode, backups automáticos y monitoreo con cron.

El contexto

En el artículo anterior construí un pipeline RAG de producción con búsqueda híbrida, cross-encoder reranking y cache semántico. Todo funcionaba perfecto en Docker local.

El problema: pasarlo a producción en un VPS económico sin que explote.

Un sistema RAG no es un CRUD típico. Tiene:

Modelos de embeddings que consumen ~500MB de RAM por worker
PostgreSQL con extensiones pesadas (pgvector + HNSW indexes)
Streaming SSE que necesita conexiones long-lived
Redis para rate limiting y cache
Todo eso compitiendo por 4GB de RAM

Infraestructura elegida

Componente	Especificación
VPS	DigitalOcean 4GB RAM / 2 vCPU / 80GB SSD
OS	Ubuntu 24.04 LTS
Containers	Docker Compose (5 servicios)
SSL	Cloudflare (proxy + certificados origin)
DNS	Cloudflare
Costo total	~$24/mes

¿Por qué no Kubernetes? Porque para un solo VPS es overkill. Docker Compose con restart policies y health checks cubre el 95% de lo que necesitás para un servicio web con pocos miles de usuarios.

Docker Compose: dev vs producción

Desarrollo

# docker-compose.yml
services:
  db:
    image: pgvector/pgvector:pg16
    ports:
      - "5433:5432"
    environment:
      POSTGRES_DB: ragdb
      POSTGRES_USER: raguser
      POSTGRES_PASSWORD: localpass123
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U raguser -d ragdb"]
      interval: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  backend:
    build: ./backend
    ports:
      - "8000:8000"
    volumes:
      - ./backend/app:/app/app  # Hot reload
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  frontend:
    build: ./frontend
    ports:
      - "5173:5173"
    volumes:
      - ./frontend/src:/app/src  # Hot reload

Nada sorprendente: puertos expuestos, volúmenes para hot reload, health checks básicos.

Producción: las diferencias que importan

# docker-compose.prod.yml
services:
  db:
    image: pgvector/pgvector:pg16
    container_name: app-db
    restart: always
    env_file: .env.production
    ports:
      - "127.0.0.1:5432:5432"  # Solo localhost
    volumes:
      - pgdata:/var/lib/postgresql/data
    command: >
      postgres
        -c shared_buffers=128MB
        -c effective_cache_size=256MB
        -c max_connections=50
        -c work_mem=4MB
        -c maintenance_work_mem=64MB
        -c random_page_cost=1.1
        -c effective_io_concurrency=200
        -c wal_buffers=4MB
        -c checkpoint_completion_target=0.9
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"]
      interval: 10s
      retries: 5

  redis:
    image: redis:7-alpine
    container_name: app-redis
    restart: always
    command: redis-server --maxmemory 64mb --maxmemory-policy allkeys-lru
    ports:
      - "127.0.0.1:6379:6379"  # Solo localhost
    volumes:
      - redisdata:/data

  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
    container_name: app-backend
    restart: always
    env_file: .env.production
    ports:
      - "127.0.0.1:8000:8000"  # Solo localhost, Nginx al frente
    deploy:
      resources:
        limits:
          memory: 1536M
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile.prod  # Multi-stage con Nginx
    container_name: app-frontend
    restart: always
    ports:
      - "127.0.0.1:5173:5173"  # Solo localhost
    deploy:
      resources:
        limits:
          memory: 64M

Las diferencias clave

1. Puertos solo en localhost

ports:
  - "127.0.0.1:8000:8000"  # ✅ Solo Nginx puede acceder
  # vs
  - "8000:8000"  # ❌ Abierto al mundo

Si publicás el puerto sin 127.0.0.1, Docker modifica iptables y bypassea el firewall del sistema. Es un error clásico.

2. PostgreSQL tuneado para 4GB

shared_buffers=128MB        # 25% de la RAM disponible para PG (~512MB)
effective_cache_size=256MB  # Lo que el OS puede cachear
max_connections=50          # No necesitás 100 con un backend async
work_mem=4MB               # Cuidado: se multiplica por conexión × sort ops
random_page_cost=1.1       # SSD, no disco rotacional

Los defaults de PostgreSQL asumen un servidor dedicado con 1GB+ de RAM solo para PG. En un VPS compartido con 4 servicios más, necesitás ser conservador.

3. Redis con límite estricto

maxmemory 64mb
maxmemory-policy allkeys-lru

Redis sin maxmemory puede crecer indefinidamente y matar el OOM killer. Con allkeys-lru, cuando llega al límite, elimina las keys menos usadas en vez de devolver errores.

4. Memory limits en el backend

deploy:
  resources:
    limits:
      memory: 1536M

El backend con modelos de embeddings cargados usa ~800MB-1.2GB. El límite de 1536MB le da margen sin permitir que un memory leak se coma todo el VPS.

Dockerfiles: multi-stage builds

Backend: pre-descargar modelos

# === Stage 1: Builder ===
FROM python:3.12-slim AS builder

WORKDIR /build
COPY requirements.txt .

# PyTorch CPU-only (ahorro ~1.5GB vs versión con CUDA)
RUN pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu \
    && pip install --no-cache-dir -r requirements.txt

# Pre-descargar modelos de embeddings y cross-encoder
RUN python -c "
from sentence_transformers import SentenceTransformer, CrossEncoder
SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
"

# === Stage 2: Runtime ===
FROM python:3.12-slim AS runtime

WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
    libfontconfig1 curl ca-certificates && rm -rf /var/lib/apt/lists/*

# Copiar dependencias y modelos pre-descargados
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY --from=builder /root/.cache /root/.cache

COPY . .
RUN chmod +x start.sh

EXPOSE 8000
CMD ["./start.sh"]

¿Por qué pre-descargar modelos en build? Si no lo hacés, el primer request después de cada deploy va a tardar 30-60 segundos mientras se descargan los modelos. Con pre-descarga, el contenedor arranca listo para servir.

¿Por qué PyTorch CPU-only? La versión con CUDA pesa ~2GB extra. En un VPS sin GPU es peso muerto.

Frontend: build + Nginx estático

# === Stage 1: Build ===
FROM node:20-alpine AS build

WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

# === Stage 2: Serve ===
FROM nginx:1.27-alpine

# Copiar build estático
COPY --from=build /app/dist /usr/share/nginx/html

# Config para SPA
COPY nginx.conf /etc/nginx/conf.d/default.conf

EXPOSE 5173
CMD ["nginx", "-g", "daemon off;"]

El frontend en producción NO corre Vite. Es Nginx sirviendo archivos estáticos. La imagen pasa de ~400MB (node + deps) a ~25MB (nginx alpine + dist).

Nginx interno del frontend (SPA)

server {
    listen 5173;
    root /usr/share/nginx/html;
    index index.html;

    # Assets con hash de Vite → cache agresivo
    location /assets/ {
        expires 1y;
        add_header Cache-Control "public, immutable";
    }

    # index.html → nunca cachear (para que nuevos deploys se reflejen)
    location = /index.html {
        add_header Cache-Control "no-cache";
    }

    # SPA fallback: toda ruta → index.html
    location / {
        try_files $uri $uri/ /index.html;
    }

    gzip on;
    gzip_types text/plain text/css application/json application/javascript;
}

Nginx: el reverse proxy que lo conecta todo

# === HTTPS principal ===
server {
    listen 443 ssl http2;
    server_name tu-dominio.com;

    # Certificados origin (Cloudflare → VPS)
    ssl_certificate     /etc/ssl/certs/origin.pem;
    ssl_certificate_key /etc/ssl/private/origin.key;
    ssl_protocols TLSv1.2 TLSv1.3;

    client_max_body_size 50M;  # Para upload de documentos

    # === Maintenance mode ===
    set $maintenance 0;
    if (-f /etc/nginx/maintenance.on) {
        set $maintenance 1;
    }

    # Health check siempre disponible (para monitoreo)
    location = /api/v1/health {
        proxy_pass http://127.0.0.1:8000;
    }

    # Si maintenance mode → 503
    if ($maintenance) {
        return 503;
    }

    # === API Backend ===
    location /api/ {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # SSE streaming: CRÍTICO
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 300s;
        proxy_set_header Connection '';
        proxy_http_version 1.1;
        chunked_transfer_encoding off;
    }

    # === Widget JS (cacheable) ===
    location = /widget.js {
        proxy_pass http://127.0.0.1:8000;
        proxy_cache_valid 200 1h;
    }

    # === Frontend SPA ===
    location / {
        proxy_pass http://127.0.0.1:5173;
    }

    # === Gzip ===
    gzip on;
    gzip_comp_level 4;
    gzip_min_length 256;
    gzip_types text/plain text/css application/json application/javascript
               text/xml application/xml text/javascript image/svg+xml;

    # === Security headers ===
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
}

# === HTTP → HTTPS redirect ===
server {
    listen 80;
    server_name tu-dominio.com;
    return 301 https://$server_name$request_uri;
}

# === www → non-www ===
server {
    listen 443 ssl http2;
    server_name www.tu-dominio.com;
    ssl_certificate     /etc/ssl/certs/origin.pem;
    ssl_certificate_key /etc/ssl/private/origin.key;
    return 301 https://tu-dominio.com$request_uri;
}

# === Página 503 de mantenimiento ===
error_page 503 @maintenance;
location @maintenance {
    default_type text/html;
    return 503 '<!DOCTYPE html>
    <html><head><meta charset="UTF-8"><title>Mantenimiento</title>
    <style>body{font-family:system-ui;display:flex;justify-content:center;
    align-items:center;min-height:100vh;background:#0f172a;color:#e2e8f0;
    text-align:center}h1{font-size:2rem}p{color:#94a3b8}</style></head>
    <body><div><h1>En mantenimiento</h1>
    <p>Volvemos en unos minutos.</p></div></body></html>';
}

El bloque SSE que casi me rompe todo

location /api/ {
    proxy_buffering off;      # Nginx NO debe buffear la respuesta
    proxy_cache off;          # Ni cachearla
    proxy_read_timeout 300s;  # SSE puede durar minutos
    proxy_set_header Connection '';  # Deshabilitar keep-alive del upstream
    proxy_http_version 1.1;   # HTTP/1.1 requerido para chunked
    chunked_transfer_encoding off;
}

Sin estos headers, Nginx buferea los tokens del SSE y los envía todos juntos al final. El usuario ve "nada nada nada... texto completo de golpe". Estos 6 parámetros son obligatorios para streaming real a través de un reverse proxy.

El script de deploy

#!/bin/bash
set -e

SERVER="usuario@ip-del-servidor"
SSH_KEY="~/.ssh/deploy_key"
PROJECT_DIR="/opt/mi-app"
BACKUP_DIR="/opt/mi-app/backups/db"
DEPLOY_MODE="${1:-full}"  # frontend | backend | full

ssh_cmd() {
    ssh -i "$SSH_KEY" "$SERVER" "$1"
}

echo "=== Deploy: $DEPLOY_MODE ==="

# 1. Push código
git push origin main

# 2. Pull en servidor
ssh_cmd "cd $PROJECT_DIR && git fetch origin && git reset --hard origin/main"

# 3. Backup de base de datos (solo backend/full)
if [[ "$DEPLOY_MODE" != "frontend" ]]; then
    echo "Creando backup de DB..."
    TIMESTAMP=$(date +%Y%m%d_%H%M%S)
    ssh_cmd "docker exec app-db pg_dump -U \$POSTGRES_USER \$POSTGRES_DB \
        | gzip > $BACKUP_DIR/backup_${TIMESTAMP}.gz"

    # Activar maintenance mode
    ssh_cmd "touch /etc/nginx/maintenance.on && nginx -s reload"
    echo "Maintenance mode: ON"
fi

# 4. Rebuild y restart
case $DEPLOY_MODE in
    frontend)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            build frontend && docker compose -f docker-compose.prod.yml \
            up -d frontend"
        ;;
    backend)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            build backend && docker compose -f docker-compose.prod.yml \
            up -d backend"
        ;;
    full)
        ssh_cmd "cd $PROJECT_DIR && docker compose -f docker-compose.prod.yml \
            up -d --build"
        ;;
esac

# 5. Desactivar maintenance mode
if [[ "$DEPLOY_MODE" != "frontend" ]]; then
    sleep 15  # Esperar a que el backend cargue modelos
    ssh_cmd "rm -f /etc/nginx/maintenance.on && nginx -s reload"
    echo "Maintenance mode: OFF"
fi

# 6. Health check
echo "Verificando salud..."
MAX_RETRIES=30
for i in $(seq 1 $MAX_RETRIES); do
    STATUS=$(ssh_cmd "curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8000/api/v1/health")
    if [[ "$STATUS" == "200" ]]; then
        echo "Deploy exitoso. Health: OK"
        exit 0
    fi
    echo "Intento $i/$MAX_RETRIES... (status: $STATUS)"
    sleep 5
done

echo "ERROR: Health check falló después de $MAX_RETRIES intentos"
exit 1

¿Por qué maintenance mode?

El backend tarda ~15 segundos en arrancar: corre migraciones de Alembic y luego Uvicorn carga los modelos de embeddings. Sin maintenance mode, durante esos 15 segundos Nginx devuelve 502 Bad Gateway.

Con el archivo /etc/nginx/maintenance.on, Nginx devuelve una página 503 estilizada. El health check (/api/v1/health) queda exento para que el script pueda verificar cuándo el backend está listo.

Start script del backend

#!/bin/bash
set -e

echo "Running database migrations..."
alembic upgrade head

echo "Starting FastAPI server..."

# 1 worker = ~800MB con modelos cargados
# 2 workers = ~1.3GB (solo si tenés RAM disponible)
WORKERS="${UVICORN_WORKERS:-1}"

exec uvicorn app.main:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers "$WORKERS" \
    --log-level "${UVICORN_LOG_LEVEL:-info}" \
    "$@"

¿Por qué 1 worker? Cada worker de Uvicorn carga su propia copia de los modelos de embeddings (~500MB). Con 2 workers ya estás en 1.3GB solo de backend. En un VPS de 4GB con PostgreSQL, Redis y Nginx, 1 worker es lo seguro.

FastAPI es async, así que 1 worker maneja bien la concurrencia — las operaciones de I/O (DB, LLM API, Redis) no bloquean el event loop.

Mantenimiento automático con Cron

#!/bin/bash
# Ejecutar: crontab -e → 0 4 * * 0 /opt/mi-app/scripts/maintenance.sh

LOG="/var/log/app-maintenance.log"
echo "=== Mantenimiento $(date) ===" >> "$LOG"

# 1. Limpiar imágenes Docker huérfanas (>7 días)
docker image prune -f --filter "until=168h" >> "$LOG" 2>&1

# 2. Limpiar contenedores parados
docker container prune -f --filter "until=168h" >> "$LOG" 2>&1

# 3. Verificar uso de disco
DISK_USAGE=$(df / --output=pcent | tail -1 | tr -dc '0-9')
if [ "$DISK_USAGE" -gt 80 ]; then
    echo "ALERTA: Disco al ${DISK_USAGE}%" >> "$LOG"
fi

# 4. Verificar memoria
MEM_USAGE=$(free | awk '/Mem:/ {printf("%.0f"), $3/$2 * 100}')
if [ "$MEM_USAGE" -gt 90 ]; then
    echo "ALERTA: RAM al ${MEM_USAGE}%" >> "$LOG"
fi

# 5. Estado de contenedores
docker ps --format "table {{.Names}}\t{{.Status}}\t{{.Size}}" >> "$LOG"

# 6. Inventario de backups
BACKUP_COUNT=$(ls /opt/mi-app/backups/db/*.gz 2>/dev/null | wc -l)
LATEST=$(ls -t /opt/mi-app/backups/db/*.gz 2>/dev/null | head -1)
echo "Backups: $BACKUP_COUNT archivos. Último: $LATEST" >> "$LOG"

echo "=== Fin ===" >> "$LOG"

Esto corre cada domingo a las 4am. Limpia basura de Docker, verifica disco y RAM, y loguea todo. Simple pero efectivo — me salvó dos veces de quedarme sin disco por imágenes Docker acumuladas.

Cloudflare + SSL: la configuración

Internet → Cloudflare (proxy) → VPS (Nginx 443) → Docker containers

Setup

DNS en Cloudflare: A record apuntando al VPS, proxy habilitado (nube naranja)
SSL en Cloudflare: "Full (strict)" — encripta tanto el tramo browser→Cloudflare como Cloudflare→VPS
Certificado origin: Generado en Cloudflare Dashboard → SSL/TLS → Origin Server → Create Certificate
En el VPS: Copiar el certificado y key a /etc/ssl/certs/origin.pem y /etc/ssl/private/origin.key

¿Por qué no Let's Encrypt? Con Cloudflare proxy activado, Let's Encrypt no puede verificar el dominio por HTTP challenge (Cloudflare intercepta). Los certificados origin de Cloudflare duran 15 años y se configuran en 2 minutos.

Bonus: Cloudflare como CDN gratis

Con el proxy activado, Cloudflare cachea automáticamente assets estáticos (JS, CSS, imágenes). El VPS solo recibe requests de API y HTML. Esto reduce significativamente el tráfico al servidor.

Distribución de RAM

Así queda la distribución en un VPS de 4GB:

┌──────────────────────────────────────────┐
│              4GB RAM Total               │
├──────────────────────────────────────────┤
│  OS + Docker Engine      ~400MB          │
│  PostgreSQL              ~200-400MB      │
│  Backend (1 worker)      ~800MB-1.2GB    │
│  Redis                   ≤64MB           │
│  Nginx + Frontend        ~30MB           │
│  Libre / Buffer          ~1.5-2GB        │
├──────────────────────────────────────────┤
│  Swap (2GB)              emergencia      │
└──────────────────────────────────────────┘

Los ~1.5-2GB libres son para:

Cache de filesystem del OS (ayuda a PostgreSQL)
Picos de tráfico
Operaciones de mantenimiento (backups, builds)

Tip: Siempre configurá swap (2GB) como red de seguridad. Sin swap, el OOM killer mata procesos sin aviso cuando la RAM se llena.

fallocate -l 2G /swapfile
chmod 600 /swapfile
mkswap /swapfile
swapon /swapfile
echo '/swapfile none swap sw 0 0' >> /etc/fstab

Checklist de seguridad

Antes de considerar el deploy "listo":

[x] Puertos Docker solo en localhost (127.0.0.1:puerto:puerto)
[x] Firewall activo (ufw allow 22,80,443/tcp && ufw enable)
[x] SSH solo por key (deshabilitar password auth en /etc/ssh/sshd_config)
[x] .env.production fuera del repo (.gitignore)
[x] Secretos nunca en compose files (usar env_file reference)
[x] DB sin puerto externo (solo accesible via Docker network)
[x] Backups automatizados con verificación periódica
[x] Health check en el script de deploy
[x] Swap configurado para evitar OOM kills

Números reales

Métrica	Valor
Tiempo de build (backend)	~3-4 min (primera vez), ~1 min (cache)
Tiempo de build (frontend)	~30s
Tiempo de deploy completo	~2-3 min
Downtime visible	0s (maintenance page durante rebuild)
Tamaño imagen backend	~2.1GB (incluye PyTorch CPU + modelos)
Tamaño imagen frontend	~25MB (nginx + dist)
RAM en uso estable	~2-2.5GB de 4GB
Costo mensual	~$24 (VPS) + $0 (Cloudflare free)
Uptime último mes	99.8% (una caída por actualización de OS)

Lecciones aprendidas

1. Docker + firewall = cuidado

Docker modifica iptables directamente. Si publicás un puerto sin 127.0.0.1, ufw deny no lo bloquea. Siempre usá 127.0.0.1: en producción y dejá que Nginx sea el único punto de entrada.

2. Pre-descargar modelos de ML en el build

Si los modelos se descargan en runtime, el primer cold start tarda un minuto. Peor: si HuggingFace tiene downtime, tu deploy falla. Pre-descargar en el Dockerfile elimina ambos problemas.

3. 1 Uvicorn worker es suficiente (si es async)

La tentación es poner 4 workers "por las dudas". Pero cada uno carga ~500MB de modelos. Con FastAPI async, un solo worker maneja cientos de requests concurrentes. Escalá en workers solo cuando tengas evidencia de que el CPU es el cuello de botella.

4. Maintenance mode > zero-downtime rolling deploys

Para un VPS single-node, un rolling deploy requiere orquestación compleja. Una página 503 por 15 segundos es infinitamente más simple y nadie se queja — especialmente si el health check garantiza que se desactiva automáticamente.

5. PostgreSQL defaults son para servidores dedicados

Los defaults de PostgreSQL asumen que tiene toda la RAM para sí. En un VPS compartido con otros 4 servicios, no tunear PostgreSQL es garantía de OOM kills. Los parámetros shared_buffers, effective_cache_size y max_connections son los primeros que hay que ajustar.

6. Cloudflare origin certs > Let's Encrypt

Con proxy activado, Let's Encrypt es más complicado de configurar y renovar. Los origin certs de Cloudflare duran 15 años, se configuran una vez y te olvidás.

Lo que sigue

Monitoring con Prometheus + Grafana: Métricas de latencia, errores, y uso de recursos (actualmente solo logs + cron)
Backup offsite: Copiar backups a un bucket S3/R2 en vez de guardarlos solo en el mismo VPS
Blue-green deploys: Cuando el tráfico justifique un segundo VPS
CI/CD con GitHub Actions: Automatizar el deploy script (actualmente es un ./deploy.sh backend manual)

Conclusión

Desplegar un sistema RAG no es como desplegar un CRUD. Los modelos de embeddings consumen RAM real, el streaming SSE necesita configuración específica en Nginx, y PostgreSQL con pgvector necesita tuning cuidadoso en servidores limitados.

Pero tampoco necesitás Kubernetes ni un cluster de $200/mes. Un VPS de $24 con Docker Compose bien configurado, Nginx como reverse proxy, y scripts de deploy con maintenance mode + health checks es suficiente para servir miles de queries al día con latencias consistentes.

Lo más importante: medí primero, escalá después. Empezar con 1 worker, 4GB de RAM y monitoreo básico te da toda la información que necesitás para tomar decisiones de infraestructura basadas en datos reales, no en estimaciones.

How I Built a Production RAG Pipeline with FastAPI, pgvector and Cross-Encoder Reranking

Martin Palopoli — Tue, 17 Mar 2026 11:00:00 +0000

I built a production RAG engine that combines hybrid search (pgvector + BM25), cross-encoder reranking, MMR diversity, semantic caching and automatic language detection — all on top of async FastAPI and PostgreSQL. This article covers the real architecture, technical decisions, and key code.

Why Another RAG Article

Most RAG tutorials stop at "embed → cosine search → prompt". That works for a demo, but in production you'll run into:

Queries in Spanish that match English chunks (or vice versa)
The top 10 most similar chunks come from the same document
Semantic search fails with proper nouns or exact codes
Repeated answers burn tokens unnecessarily
No way to know if the answer is actually reliable

This article shows how I solved each of these problems in a real multi-tenant system.

The Stack

Component	Technology
Backend	Python 3.12 + FastAPI (100% async)
Database	PostgreSQL 16 + pgvector + tsvector
Embeddings	`paraphrase-multilingual-MiniLM-L12-v2` (384d)
Reranker	`cross-encoder/ms-marco-MiniLM-L-6-v2`
LLM	Groq (`llama-3.3-70b-versatile`)
Cache	Redis + PostgreSQL (semantic cache)
Streaming	Server-Sent Events (SSE)

Pipeline Architecture

The pipeline has 3 fixed stages with pre and post filters:

User query
     │
     ▼
┌─────────────┐
│  FAQ Match   │──→ If score ≥ 0.75: instant answer (LLM cost = 0)
└─────┬───────┘
      │ No match
      ▼
┌─────────────┐
│ Cache Check  │──→ If similarity ≥ 0.95: cached response
└─────┬───────┘
      │ Cache miss
      ▼
┌─────────────────────────────────────┐
│  STAGE 1: Hybrid Search            │
│  pgvector (semantic) + BM25 (lexical)│
│  Fusion with Reciprocal Rank Fusion │
└─────────────┬───────────────────────┘
              │
              ▼
┌─────────────────────────────────────┐
│  STAGE 2: Cross-Encoder Reranking  │
│  Re-evaluates actual relevance     │
│  Computes confidence score         │
└─────────────┬───────────────────────┘
              │
              ▼
┌─────────────────────────────────────┐
│  STAGE 3: MMR (Diversity)          │
│  Balances relevance vs variety     │
│  Max 3 chunks per document         │
└─────────────┬───────────────────────┘
              │
              ▼
         LLM + SSE streaming

Let's look at each part in detail.

Stage 1: Hybrid Search (Vector + BM25)

The Problem with Vector-Only Search

Embedding-based search is excellent at capturing semantic meaning, but fails with:

Proper nouns: "What did John Smith say?"
Codes or IDs: "HTTP 403 error"
Exact technical terms the embedding model doesn't know well

The Solution: Hybrid Search

I run two searches in parallel and merge the results:

async def search_chunks(self, query: str, kb_ids: list[int],
                        embedding: list[float], config: RetrievalConfig):
    # Vector search (pgvector HNSW, cosine distance)
    vector_results = await self._vector_search(
        embedding, kb_ids, config.candidate_k
    )

    # Lexical search (PostgreSQL tsvector + ts_rank_cd)
    bm25_results = await self._bm25_search(
        query, kb_ids, config.candidate_k
    )

    # Fusion with Reciprocal Rank Fusion
    merged = self._rrf_merge(vector_results, bm25_results, config.bm25_weight)

    return merged

Reciprocal Rank Fusion (RRF)

RRF is elegant in its simplicity. Instead of comparing scores from different systems (which have different scales), it uses ranking positions:

def _rrf_merge(self, vector_results, bm25_results, bm25_weight=0.3):
    k = 60  # Smoothing constant
    vector_weight = 1.0 - bm25_weight
    scores = {}

    for rank, chunk in enumerate(vector_results):
        scores[chunk.id] = vector_weight / (k + rank + 1)

    for rank, chunk in enumerate(bm25_results):
        scores[chunk.id] = scores.get(chunk.id, 0) + bm25_weight / (k + rank + 1)

    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

Why k=60? It prevents the #1 chunk from having a disproportionately high score relative to #2. With k=60, the difference between position 1 and 2 is minimal, making the fusion more stable.

Why 70/30? In my tests, semantic search is more robust for most queries. The 30% BM25 rescues the cases where you need exact lexical matching. This is configurable per Knowledge Base.

BM25 with Fallback

An important detail: BM25 search first tries an AND query, and if it finds no results, falls back to OR:

# First: AND (all terms must be present)
ts_query = func.plainto_tsquery('spanish', query)

# If no results: OR (any term)
words = [w for w in query.split() if w.lower() not in STOP_WORDS]
or_query = " | ".join(words)
ts_query = func.to_tsquery('spanish', or_query)

Stage 2: Cross-Encoder Reranking

Why Embeddings Aren't Enough

Bi-encoder embeddings (like MiniLM) encode query and document separately. They're fast but miss subtle interactions between query and chunk words.

A cross-encoder processes query + chunk together as a single input, capturing cross-attention relationships. It's slower but significantly more accurate.

Implementation

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(self, query: str, candidates: list[dict]) -> list[dict]:
    pairs = [(query, c["content"]) for c in candidates]
    scores = self.reranker.predict(pairs)

    for candidate, score in zip(candidates, scores):
        candidate["rerank_score"] = float(score)

    # Sort by rerank_score descending
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)

We reduce from ~60 candidates (RRF output) to the top 15 after reranking.

Confidence Score: How Reliable Is the Answer?

Using cross-encoder scores, I compute a confidence score that indicates how confident we are that the retrieved chunks are relevant:

from math import exp

def compute_confidence(sources: list[dict]) -> float:
    scores = sorted([s["rerank_score"] for s in sources], reverse=True)

    max_score = scores[0]
    top3_avg = sum(scores[:3]) / min(len(scores), 3)
    median = scores[len(scores) // 2]

    # Blend: 40% best score, 30% top-3 average, 30% median
    blended = 0.4 * max_score + 0.3 * top3_avg + 0.3 * median

    # Sigmoid → [0, 1]
    return 1.0 / (1.0 + exp(-blended))

Why a blend and not just the max? If the best chunk scores 0.9 but the rest are at -0.5, the answer is probably weak — a single good chunk doesn't compensate for bad context. The blend captures the overall quality of the retrieved context.

The confidence score is sent to the frontend, which uses it to:

Display visual confidence indicators
Decide whether to suggest alternative FAQs
Trigger automatic escalation when below threshold

Stage 3: MMR (Maximum Marginal Relevance)

The Redundancy Problem

Without MMR, the top 10 most relevant chunks could come from the same document or be consecutive paragraphs saying essentially the same thing. That wastes LLM context.

The Formula

MMR(chunk) = λ × relevance(chunk) - (1-λ) × max_similarity(chunk, already_selected) - document_penalty

def _mmr_select(self, candidates, query_embedding, config):
    selected = []
    remaining = list(candidates)
    lambda_param = config.lambda_param  # Default: 0.6
    doc_counts = {}

    for _ in range(config.top_k):  # Default: 10
        best_score = -float('inf')
        best_idx = 0

        for i, candidate in enumerate(remaining):
            # Relevance to query
            relevance = cosine_similarity(query_embedding, candidate["embedding"])

            # Max similarity with already selected chunks
            max_sim = max(
                (cosine_similarity(candidate["embedding"], s["embedding"])
                 for s in selected),
                default=0
            )

            # Penalty for repeated document
            doc_name = candidate["document_name"]
            penalty = doc_counts.get(doc_name, 0) * 0.1

            # Hard cap: max 3 chunks per document
            if doc_counts.get(doc_name, 0) >= config.max_per_doc:
                continue

            score = lambda_param * relevance - (1 - lambda_param) * max_sim - penalty

            if score > best_score:
                best_score = score
                best_idx = i

        chosen = remaining.pop(best_idx)
        selected.append(chosen)
        doc_counts[chosen["document_name"]] = doc_counts.get(chosen["document_name"], 0) + 1

    return selected

λ = 0.6 means 60% relevance, 40% diversity. It's the sweet spot I found: enough diversity to avoid repetition, but without sacrificing truly relevant chunks.

Semantic Cache: Don't Repeat Work

The Concept

If someone asks "How do I reset my password?" and 2 hours ago another user asked "How can I change my password?", the answer is the same. Why run the entire pipeline again?

Implementation

async def lookup_cache(self, query_embedding, tenant_id, kb_ids, config_hash):
    result = await db.execute(
        select(ResponseCache)
        .where(
            ResponseCache.tenant_id == tenant_id,
            ResponseCache.rag_config_hash == config_hash,
            ResponseCache.expires_at > func.now(),
        )
        .order_by(
            ResponseCache.query_embedding.cosine_distance(query_embedding)
        )
        .limit(1)
    )
    cache_entry = result.scalar_one_or_none()

    if cache_entry:
        similarity = 1 - cosine_distance(query_embedding, cache_entry.query_embedding)
        if similarity >= 0.95:
            return cache_entry  # Cache hit

    return None  # Cache miss → run full pipeline

Key Decisions

Threshold 0.95: Very high on purpose. I'd rather have a cache miss than serve an incorrect answer.
Config hash: Cache is invalidated if RAG parameters change. Changing top_k from 10 to 5 produces different answers.
Fire-and-forget store: After generating the LLM response, I save to cache without blocking the stream.
TTL 7 days: Configurable per KB. Documents that change often can use a shorter TTL.
Proactive invalidation: When a document is uploaded or reprocessed, the cache for that KB is invalidated.

# Store (doesn't block the response)
async def store_cache(self, query_embedding, response, sources, confidence, ...):
    if confidence < config.min_confidence:
        return  # Don't cache low-confidence answers

    cache_entry = ResponseCache(
        query_embedding=query_embedding,
        response_text=response,
        sources=sources,
        confidence=confidence,
        rag_config_hash=config_hash,
        expires_at=datetime.utcnow() + timedelta(hours=config.cache_ttl_hours),
    )
    db.add(cache_entry)
    await db.commit()

FAQ Match: The Most Valuable Shortcut

Before the entire pipeline, I check if there's a FAQ whose embedding is similar enough to the query:

async def match_faq(self, query_embedding, kb_ids, threshold=0.75):
    result = await db.execute(
        select(FAQ.id, FAQ.question, FAQ.answer,
               (1 - FAQ.embedding.cosine_distance(query_embedding)).label("score"))
        .where(FAQ.knowledge_base_id.in_(kb_ids), FAQ.is_active == True)
        .order_by(FAQ.embedding.cosine_distance(query_embedding))
        .limit(1)
    )
    faq = result.first()

    if faq and faq.score >= threshold:
        # Increment hit_count atomically
        await db.execute(
            update(FAQ).where(FAQ.id == faq.id)
            .values(hit_count=FAQ.hit_count + 1)
        )
        return faq

    return None

Why is this so valuable?

Zero cost: No LLM call, no token consumption
Minimal latency: A single SQL query with HNSW index
Curated answers: FAQs are written by humans, always correct
Always available: Even when the user has exhausted their monthly LLM query quota, FAQs keep working

Language Detection Without External APIs

The system supports Spanish, English, and Portuguese. Instead of calling a detection API (extra latency + cost), I use heuristic detection:

EN_WORDS = {"how", "what", "the", "is", "can", "do", "where", "when", "why", ...}  # 60+
PT_WORDS = {"como", "onde", "você", "qual", "para", "não", "uma", ...}  # 45+
ES_WORDS = {"qué", "cómo", "dónde", "ayuda", "puedo", "quiero", ...}  # 40+

def detect_language(query: str) -> str:
    words = set(query.lower().split())
    scores = {
        "en": len(words & EN_WORDS),
        "pt": len(words & PT_WORDS),
        "es": len(words & ES_WORDS),
    }
    best = max(scores, key=scores.get)
    return best if scores[best] >= 2 else "es"  # Default: Spanish

Is it perfect? No. But it has zero latency, no external dependencies, and for short queries of 5-15 words it works surprisingly well. The detected language is injected into the LLM's system prompt so it responds in the same language.

Smart Routing: Automatically Choosing the Right KB

When a tenant has multiple Knowledge Bases (e.g., "Sales", "Technical Support", "HR"), the user shouldn't have to manually choose where to search.

KB Centroids

For each KB, I compute the centroid (average of all chunk embeddings):

SELECT AVG(embedding)::text AS centroid FROM chunks WHERE knowledge_base_id = :kb_id

Similarity-Based Routing

async def route_query(self, query_embedding, available_kbs):
    scores = []
    for kb in available_kbs:
        centroid = await self._get_centroid(kb.id)  # Redis cache 24h
        similarity = cosine_similarity(query_embedding, centroid)
        if similarity >= 0.15:  # Minimum threshold
            scores.append((kb, similarity))

    # Top 3 most relevant KBs
    scores.sort(key=lambda x: x[1], reverse=True)
    return [kb for kb, _ in scores[:3]]

If no KB exceeds the threshold, search across all of them (safe fallback).

Numbers That Matter

Metric	Value
Initial candidates per branch	30 (vector) + 30 (BM25)
After RRF merge	~40-60 unique
After reranking	Top 15
After MMR	Top 10 (max 3 per doc)
Embedding dimensions	384
FAQ threshold	0.75
Cache threshold	0.95
Cache TTL	7 days
FAQ match latency	~15ms
Full pipeline latency	~2-4s
Savings per cache hit	~90% latency, 100% LLM tokens

Lessons Learned

1. The Cross-Encoder Is Worth It

The quality difference between using embeddings alone vs. embeddings + cross-encoder reranking is enormous. Bi-encoder embeddings are good for recall (finding candidates), but the cross-encoder is much better for precision (ranking by actual relevance).

2. BM25 Isn't Dead

For queries with proper nouns, error codes, or specific technical terms, BM25 regularly outperforms vector search. The hybrid combination is strictly better than either one alone.

3. MMR Is More Important Than It Seems

Without MMR diversity, the LLM receives redundant context and produces answers that orbit around a single point. With MMR, answers are more complete and cover different angles of the topic.

4. Semantic Cache Requires a High Threshold

At 0.90 I had problems with incorrect answers being served from cache. 0.95 is the sweet spot: enough to cache obvious reformulations, but not so low as to match queries that actually need different answers.

5. FAQs Are Your Best Investment

Every FAQ you add is a perfect answer served in 15ms with zero LLM cost. I implemented auto-generation of suggested FAQs from frequent unanswered queries, which the admin can approve with a single click.

What's Next

Evaluation framework: Automatic quality metrics (faithfulness, relevance) with evaluation datasets per KB
Chunk contextualization: Using the LLM to add document context to chunks before indexing
Late interaction models (ColBERT): Better than cross-encoder for high volumes

Conclusion

A production RAG pipeline is not "embed + search + prompt". It's a multi-stage system where each component solves a specific weakness of the previous one. Hybrid search covers semantic and lexical gaps, the cross-encoder ranks by actual relevance, MMR ensures diversity, cache eliminates redundant work, and FAQs shortcut when the answer already exists.

Most importantly: everything is configurable per Knowledge Base. There's no universal set of parameters — each knowledge domain has its own characteristics and needs different tuning.

If you're working on production RAG, drop a comment with your biggest challenge. And if this article was useful, a like helps it reach more people.

Cómo construí un pipeline RAG de producción con FastAPI, pgvector y cross-encoder reranking

Martin Palopoli — Thu, 12 Mar 2026 16:50:00 +0000

Construí un motor RAG de producción que combina búsqueda híbrida (pgvector + BM25), reranking con cross-encoder, diversidad MMR, cache semántico y detección automática de idioma — todo sobre FastAPI async y PostgreSQL. En este artículo comparto la arquitectura real, las decisiones técnicas, y el código clave.

Por qué otro artículo sobre RAG

La mayoría de los tutoriales RAG se quedan en "embed → cosine search → prompt". Eso funciona para una demo, pero en producción te vas a encontrar con:

Queries en español que matchean chunks en inglés (o viceversa)
Los 10 chunks más similares vienen del mismo documento
La búsqueda semántica falla con nombres propios o códigos exactos
Respuestas repetidas consumen tokens innecesariamente
No tenés forma de saber si la respuesta es confiable o no

Este artículo muestra cómo resolví cada uno de esos problemas en un sistema real multi-tenant.

El stack

Componente	Tecnología
Backend	Python 3.12 + FastAPI (100% async)
Base de datos	PostgreSQL 16 + pgvector + tsvector
Embeddings	`paraphrase-multilingual-MiniLM-L12-v2` (384d)
Reranker	`cross-encoder/ms-marco-MiniLM-L-6-v2`
LLM	Groq (`llama-3.3-70b-versatile`)
Cache	Redis + PostgreSQL (cache semántico)
Streaming	Server-Sent Events (SSE)

Arquitectura del pipeline

El pipeline tiene 3 etapas fijas con filtros pre y post:

Query del usuario
     │
     ▼
┌─────────────┐
│  FAQ Match   │──→ Si score ≥ 0.75: respuesta instantánea (costo LLM = 0)
└─────┬───────┘
      │ No match
      ▼
┌─────────────┐
│ Cache Check  │──→ Si similaridad ≥ 0.95: respuesta cacheada
└─────┬───────┘
      │ Cache miss
      ▼
┌─────────────────────────────────────┐
│  ETAPA 1: Búsqueda Híbrida         │
│  pgvector (semántica) + BM25 (léxica)│
│  Fusión con Reciprocal Rank Fusion  │
└─────────────┬───────────────────────┘
              │
              ▼
┌─────────────────────────────────────┐
│  ETAPA 2: Cross-Encoder Reranking   │
│  Re-evalúa relevancia real          │
│  Calcula confidence score           │
└─────────────┬───────────────────────┘
              │
              ▼
┌─────────────────────────────────────┐
│  ETAPA 3: MMR (Diversidad)          │
│  Balancea relevancia vs variedad    │
│  Máx 3 chunks por documento         │
└─────────────┬───────────────────────┘
              │
              ▼
         LLM + SSE streaming

Veamos cada parte en detalle.

Etapa 1: Búsqueda Híbrida (Vector + BM25)

El problema con solo búsqueda vectorial

La búsqueda por embeddings es excelente para capturar significado semántico, pero falla con:

Nombres propios: "¿Qué dijo Juan Pérez?"
Códigos o IDs: "Error HTTP 403"
Términos técnicos exactos que el modelo de embeddings no conoce bien

La solución: búsqueda híbrida

Ejecuto dos búsquedas en paralelo y fusiono los resultados:

async def search_chunks(self, query: str, kb_ids: list[int],
                        embedding: list[float], config: RetrievalConfig):
    # Búsqueda vectorial (pgvector HNSW, distancia coseno)
    vector_results = await self._vector_search(
        embedding, kb_ids, config.candidate_k
    )

    # Búsqueda léxica (PostgreSQL tsvector + ts_rank_cd)
    bm25_results = await self._bm25_search(
        query, kb_ids, config.candidate_k
    )

    # Fusión con Reciprocal Rank Fusion
    merged = self._rrf_merge(vector_results, bm25_results, config.bm25_weight)

    return merged

Reciprocal Rank Fusion (RRF)

RRF es elegante en su simplicidad. En vez de comparar scores de sistemas diferentes (que tienen escalas distintas), usa posiciones en el ranking:

def _rrf_merge(self, vector_results, bm25_results, bm25_weight=0.3):
    k = 60  # Constante de suavizado
    vector_weight = 1.0 - bm25_weight
    scores = {}

    for rank, chunk in enumerate(vector_results):
        scores[chunk.id] = vector_weight / (k + rank + 1)

    for rank, chunk in enumerate(bm25_results):
        scores[chunk.id] = scores.get(chunk.id, 0) + bm25_weight / (k + rank + 1)

    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

¿Por qué k=60? Evita que el chunk #1 tenga un score desproporcionadamente alto respecto al #2. Con k=60, la diferencia entre posición 1 y 2 es mínima, lo que hace la fusión más estable.

¿Por qué 70/30? En mis pruebas, la búsqueda semántica es más robusta para la mayoría de queries. El 30% de BM25 rescata los casos donde necesitás match léxico exacto. Esto es configurable por Knowledge Base.

BM25 con fallback

Un detalle importante: la búsqueda BM25 primero intenta un AND query, y si no encuentra resultados, hace fallback a OR:

# Primero: AND (todos los términos deben estar presentes)
ts_query = func.plainto_tsquery('spanish', query)

# Si no hay resultados: OR (cualquier término)
words = [w for w in query.split() if w.lower() not in STOP_WORDS]
or_query = " | ".join(words)
ts_query = func.to_tsquery('spanish', or_query)

Etapa 2: Cross-Encoder Reranking

Por qué no alcanza con embeddings

Los embeddings bi-encoder (como MiniLM) codifican query y documento por separado. Son rápidos pero pierden interacciones sutiles entre las palabras del query y del chunk.

Un cross-encoder procesa query + chunk juntos como un solo input, capturando relaciones cruzadas. Es más lento pero significativamente más preciso.

Implementación

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(self, query: str, candidates: list[dict]) -> list[dict]:
    pairs = [(query, c["content"]) for c in candidates]
    scores = self.reranker.predict(pairs)

    for candidate, score in zip(candidates, scores):
        candidate["rerank_score"] = float(score)

    # Ordenar por rerank_score descendente
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)

Reducimos de ~60 candidatos (resultado de RRF) a los top 15 después del reranking.

Confidence Score: ¿qué tan confiable es la respuesta?

Con los scores del cross-encoder, calculo un confidence score que indica qué tan seguro estoy de que los chunks recuperados son relevantes:

from math import exp

def compute_confidence(sources: list[dict]) -> float:
    scores = sorted([s["rerank_score"] for s in sources], reverse=True)

    max_score = scores[0]
    top3_avg = sum(scores[:3]) / min(len(scores), 3)
    median = scores[len(scores) // 2]

    # Blend: 40% mejor score, 30% promedio top-3, 30% mediana
    blended = 0.4 * max_score + 0.3 * top3_avg + 0.3 * median

    # Sigmoid → [0, 1]
    return 1.0 / (1.0 + exp(-blended))

¿Por qué un blend y no solo el max? Si el mejor chunk tiene score 0.9 pero el resto está en -0.5, la respuesta probablemente es débil — un solo chunk bueno no compensa contexto malo. El blend captura la calidad general del contexto recuperado.

El confidence score se envía al frontend, que lo usa para:

Mostrar indicadores visuales de confianza
Decidir si sugerir FAQs alternativas
Trigger de escalación automática cuando está por debajo del umbral

Etapa 3: MMR (Maximum Marginal Relevance)

El problema de la redundancia

Sin MMR, los 10 chunks más relevantes podrían venir del mismo documento o ser párrafos consecutivos que dicen prácticamente lo mismo. Eso desperdicia contexto del LLM.

La fórmula

MMR(chunk) = λ × relevancia(chunk) - (1-λ) × max_similaridad(chunk, ya_seleccionados) - penalidad_documento

def _mmr_select(self, candidates, query_embedding, config):
    selected = []
    remaining = list(candidates)
    lambda_param = config.lambda_param  # Default: 0.6
    doc_counts = {}

    for _ in range(config.top_k):  # Default: 10
        best_score = -float('inf')
        best_idx = 0

        for i, candidate in enumerate(remaining):
            # Relevancia vs query
            relevance = cosine_similarity(query_embedding, candidate["embedding"])

            # Máxima similaridad con chunks ya seleccionados
            max_sim = max(
                (cosine_similarity(candidate["embedding"], s["embedding"])
                 for s in selected),
                default=0
            )

            # Penalidad por documento repetido
            doc_name = candidate["document_name"]
            penalty = doc_counts.get(doc_name, 0) * 0.1

            # Hard cap: máximo 3 chunks por documento
            if doc_counts.get(doc_name, 0) >= config.max_per_doc:
                continue

            score = lambda_param * relevance - (1 - lambda_param) * max_sim - penalty

            if score > best_score:
                best_score = score
                best_idx = i

        chosen = remaining.pop(best_idx)
        selected.append(chosen)
        doc_counts[chosen["document_name"]] = doc_counts.get(chosen["document_name"], 0) + 1

    return selected

λ = 0.6 significa 60% relevancia, 40% diversidad. Es el sweet spot que encontré: suficiente diversidad para no repetir, pero sin sacrificar chunks realmente relevantes.

Cache Semántico: no repitas trabajo

El concepto

Si alguien pregunta "¿Cómo reseteo mi contraseña?" y hace 2 horas otro usuario preguntó "¿Cómo puedo cambiar mi password?", la respuesta es la misma. ¿Para qué ejecutar todo el pipeline de nuevo?

Implementación

async def lookup_cache(self, query_embedding, tenant_id, kb_ids, config_hash):
    result = await db.execute(
        select(ResponseCache)
        .where(
            ResponseCache.tenant_id == tenant_id,
            ResponseCache.rag_config_hash == config_hash,
            ResponseCache.expires_at > func.now(),
        )
        .order_by(
            ResponseCache.query_embedding.cosine_distance(query_embedding)
        )
        .limit(1)
    )
    cache_entry = result.scalar_one_or_none()

    if cache_entry:
        similarity = 1 - cosine_distance(query_embedding, cache_entry.query_embedding)
        if similarity >= 0.95:
            return cache_entry  # Cache hit

    return None  # Cache miss → ejecutar pipeline completo

Decisiones clave

Threshold 0.95: Muy alto a propósito. Prefiero un cache miss a servir una respuesta incorrecta.
Config hash: El cache se invalida si cambian los parámetros de RAG. Cambiar top_k de 10 a 5 produce respuestas diferentes.
Fire-and-forget store: Después de generar la respuesta del LLM, guardo en cache sin bloquear el streaming.
TTL 7 días: Configurable por KB. Documentos que cambian seguido pueden usar TTL más corto.
Invalidación proactiva: Cuando se sube o reprocesa un documento, se invalida el cache de esa KB.

# Almacenar (no bloquea el response)
async def store_cache(self, query_embedding, response, sources, confidence, ...):
    if confidence < config.min_confidence:
        return  # No cachear respuestas de baja confianza

    cache_entry = ResponseCache(
        query_embedding=query_embedding,
        response_text=response,
        sources=sources,
        confidence=confidence,
        rag_config_hash=config_hash,
        expires_at=datetime.utcnow() + timedelta(hours=config.cache_ttl_hours),
    )
    db.add(cache_entry)
    await db.commit()

FAQ Match: la shortcut más valiosa

Antes de todo el pipeline, verifico si existe una FAQ cuyo embedding sea suficientemente similar al query:

async def match_faq(self, query_embedding, kb_ids, threshold=0.75):
    result = await db.execute(
        select(FAQ.id, FAQ.question, FAQ.answer,
               (1 - FAQ.embedding.cosine_distance(query_embedding)).label("score"))
        .where(FAQ.knowledge_base_id.in_(kb_ids), FAQ.is_active == True)
        .order_by(FAQ.embedding.cosine_distance(query_embedding))
        .limit(1)
    )
    faq = result.first()

    if faq and faq.score >= threshold:
        # Incrementar hit_count atómicamente
        await db.execute(
            update(FAQ).where(FAQ.id == faq.id)
            .values(hit_count=FAQ.hit_count + 1)
        )
        return faq

    return None

¿Por qué es tan valioso?

Costo cero: No llama al LLM, no consume tokens
Latencia mínima: Una sola query SQL con índice HNSW
Respuestas curadas: Las FAQs son redactadas por humanos, siempre correctas
Siempre disponible: Incluso cuando el usuario agotó su cuota mensual de queries LLM, las FAQs siguen funcionando

Detección de idioma sin APIs externas

Mi sistema soporta español, inglés y portugués. En vez de llamar a una API de detección (latencia extra + costo), uso detección heurística:

EN_WORDS = {"how", "what", "the", "is", "can", "do", "where", "when", "why", ...}  # 60+
PT_WORDS = {"como", "onde", "você", "qual", "para", "não", "uma", ...}  # 45+
ES_WORDS = {"qué", "cómo", "dónde", "ayuda", "puedo", "quiero", ...}  # 40+

def detect_language(query: str) -> str:
    words = set(query.lower().split())
    scores = {
        "en": len(words & EN_WORDS),
        "pt": len(words & PT_WORDS),
        "es": len(words & ES_WORDS),
    }
    best = max(scores, key=scores.get)
    return best if scores[best] >= 2 else "es"  # Default: español

¿Es perfecto? No. Pero tiene latencia cero, no depende de servicios externos, y para queries cortas de 5-15 palabras funciona sorprendentemente bien. El idioma detectado se inyecta en el system prompt del LLM para que responda en el mismo idioma.

Smart Routing: elegir la KB correcta automáticamente

Cuando un tenant tiene múltiples Knowledge Bases (ej: "Ventas", "Soporte Técnico", "RRHH"), el usuario no debería tener que elegir manualmente dónde buscar.

Centroides de KB

Para cada KB, calculo el centroide (promedio de todos los embeddings de sus chunks):

SELECT AVG(embedding)::text AS centroid FROM chunks WHERE knowledge_base_id = :kb_id

Routing por similaridad

async def route_query(self, query_embedding, available_kbs):
    scores = []
    for kb in available_kbs:
        centroid = await self._get_centroid(kb.id)  # Redis cache 24h
        similarity = cosine_similarity(query_embedding, centroid)
        if similarity >= 0.15:  # Umbral mínimo
            scores.append((kb, similarity))

    # Top 3 KBs más relevantes
    scores.sort(key=lambda x: x[1], reverse=True)
    return [kb for kb, _ in scores[:3]]

Si ninguna KB supera el umbral, se busca en todas (fallback seguro).

Números que importan

Métrica	Valor
Candidatos iniciales por rama	30 (vector) + 30 (BM25)
Después de RRF merge	~40-60 únicos
Después de reranking	Top 15
Después de MMR	Top 10 (máx 3 por doc)
Dimensiones de embeddings	384
Threshold FAQ	0.75
Threshold cache	0.95
TTL cache	7 días
Latencia FAQ match	~15ms
Latencia pipeline completo	~2-4s
Ahorro por cache hit	~90% latencia, 100% tokens LLM

Lecciones aprendidas

1. El cross-encoder vale la pena

La diferencia de calidad entre usar solo embeddings vs. embeddings + cross-encoder reranking es enorme. Los embeddings bi-encoder son buenos para recall (encontrar candidatos), pero el cross-encoder es mucho mejor para precision (ordenar por relevancia real).

2. BM25 no murió

Para queries con nombres propios, códigos de error, o términos técnicos específicos, BM25 regularmente supera a la búsqueda vectorial. La combinación híbrida es estrictamente mejor que cualquiera por separado.

3. MMR es más importante de lo que parece

Sin diversidad MMR, el LLM recibe contexto redundante y produce respuestas que orbitan alrededor de un solo punto. Con MMR, las respuestas son más completas y cubren diferentes ángulos del tema.

4. Cache semántico requiere threshold alto

Con 0.90 tuve problemas de respuestas incorrectas servidas del cache. 0.95 es el sweet spot: suficiente para cachear reformulaciones obvias, pero no tan bajo como para matchear queries que realmente necesitan respuestas diferentes.

5. Las FAQs son tu mejor inversión

Cada FAQ que agregás es una respuesta perfecta que se sirve en 15ms sin costo de LLM. Implementé auto-generación de FAQs sugeridas a partir de queries frecuentes sin respuesta, que el admin puede aprobar con un click.

Lo que sigue

Evaluation framework: métricas automáticas de calidad (faithfulness, relevance) con datasets de evaluación por KB
Chunk contextualization: usar el LLM para agregar contexto del documento al chunk antes de indexar
Late interaction models (ColBERT): mejor que cross-encoder para volúmenes altos

Conclusión

Un pipeline RAG de producción no es "embed + search + prompt". Es un sistema de múltiples etapas donde cada componente resuelve una debilidad específica del anterior. La búsqueda híbrida cubre gaps semánticos y léxicos, el cross-encoder ordena por relevancia real, MMR garantiza diversidad, el cache elimina trabajo redundante, y las FAQs cortan camino cuando la respuesta ya existe.

Lo más importante: todo es configurable por Knowledge Base. No existe un set de parámetros universal — cada dominio de conocimiento tiene sus propias características y necesita tuning diferente.

Si te interesa el tema de RAG en producción, dejá un comentario con tu mayor desafío. Y si este artículo te fue útil, un like ayuda a que llegue a más personas.