RAG completo com Supabase pgvector + DeepSeek via Tokia: do zero ao produção (Python + Next.js)

RAG é o "Hello World" de aplicação séria de LLM. Esse post mostra a versão completa de produção — não a versão simplificada que todo mundo posta. Tem trade-offs reais, código testado e custos honestos.

O que é RAG (em 30 segundos)

LLMs têm conhecimento limitado ao corpus de treino. Pra perguntas sobre seus documentos, você precisa de RAG:

Documentos → Embeddings → Vector DB → Query similar → LLM com contexto → Resposta

Use cases comuns:

• Chatbot "pergunte sobre nossa documentação"
• Search semântico em base de conhecimento interna
• Q&A em PDFs de contratos
• Suporte automatizado com base no FAQ histórico

Stack escolhida (e por quê)

| Camada | Escolha | Por quê | |---|---|---| | Embeddings | text-embedding-3-small (OpenAI via Tokia) | $0.02/M tokens, 1536 dims, melhor qualidade PT-BR | | Vector DB | Supabase pgvector | Postgres você já conhece, sem novo serviço, JOIN com tabelas existentes | | LLM geração | DeepSeek V3 (via Tokia) | $0.27/M input, igual GPT-4o-mini mas 1/15 do preço | | Backend | Python FastAPI | Ecosystem ML maduro + tipos com Pydantic | | Frontend | Next.js 16 | Streaming SSE nativo |

Alternativas consideradas e descartadas:

• Pinecone: caro ($70+/mês), 1 serviço extra pra manter, vale só

  10M vetores

• Weaviate: poderoso mas complexo demais pra MVP
• Qdrant: rápido mas ecosystem menor
• OpenAI embeddings v3-large: 3072 dims = pgvector lento, custo 4x sem ganho real pra PT-BR

Schema Postgres com pgvector

-- Habilita extensão
CREATE EXTENSION IF NOT EXISTS vector;

-- Tabela de documentos fonte
CREATE TABLE documents (
  id BIGSERIAL PRIMARY KEY,
  title TEXT NOT NULL,
  source_url TEXT,
  content TEXT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Chunks com embeddings
CREATE TABLE chunks (
  id BIGSERIAL PRIMARY KEY,
  document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
  content TEXT NOT NULL,
  embedding vector(1536),  -- text-embedding-3-small = 1536 dims
  chunk_index INT NOT NULL,
  token_count INT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Index HNSW pra busca rápida (mais rápido que IVFFlat pra menos de 1M vetores)
CREATE INDEX idx_chunks_embedding
  ON chunks USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

-- Index pra filtros
CREATE INDEX idx_chunks_document ON chunks(document_id);

Etapa 1 — Chunking (a etapa que ninguém pensa)

Dividir documento em pedaços de tamanho certo é arte. Vou direto ao que funciona:

from typing import List
import tiktoken

enc = tiktoken.encoding_for_model("gpt-4")

def chunk_text(
    text: str,
    chunk_size: int = 500,   # tokens, não caracteres
    overlap: int = 50,
) -> List[str]:
    """
    Chunk com overlap. Tokens, não caracteres, pra controle preciso.
    Overlap garante que ideia que cruza fronteira de chunk não se perde.
    """
    tokens = enc.encode(text)
    chunks = []
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        chunks.append(enc.decode(chunk_tokens))
    return chunks


def chunk_by_section(text: str) -> List[str]:
    """
    Estratégia alternativa: por seção (## ou ###).
    Funciona melhor pra documentação técnica/Markdown.
    """
    import re
    sections = re.split(r"\n##+\s", text)
    return [s.strip() for s in sections if len(s.strip()) > 50]

Quando usar qual:

• Token-based + overlap: contratos longos, PDFs gerais, transcrição
• Section-based: documentação Markdown, FAQs estruturados
• Semântico (mais avançado): combina ambos — chunk por seção, mas divide seções > 800 tokens. Use langchain.text_splitter.RecursiveCharacterTextSplitter ou implementação própria

Etapa 2 — Gerar embeddings via Tokia

import os
import asyncio
from openai import AsyncOpenAI
from typing import List

client = AsyncOpenAI(
    api_key=os.environ["TOKIA_KEY"],
    base_url="https://api.usetokia.com/v1",
)

async def embed_batch(texts: List[str]) -> List[List[float]]:
    """
    Embeddings em batch (mais barato que 1-por-1).
    OpenAI/Tokia aceitam até ~2k textos por request.
    """
    if not texts:
        return []
    resp = await client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in resp.data]


async def index_document(doc_id: int, title: str, content: str, conn):
    chunks = chunk_text(content, chunk_size=500, overlap=50)
    embeddings = await embed_batch(chunks)
    rows = [
        (doc_id, chunk, emb, idx, len(enc.encode(chunk)))
        for idx, (chunk, emb) in enumerate(zip(chunks, embeddings))
    ]
    await conn.executemany(
        """INSERT INTO chunks (document_id, content, embedding, chunk_index, token_count)
           VALUES ($1, $2, $3, $4, $5)""",
        rows,
    )

Etapa 3 — Query (a parte que importa)

async def retrieve(query: str, top_k: int = 5, conn=None) -> List[dict]:
    # 1. Embedding da query (mesma família do que foi indexado)
    query_emb = (await embed_batch([query]))[0]

    # 2. Busca cosine similarity
    rows = await conn.fetch(
        """
        SELECT c.id, c.content, c.document_id, d.title,
               1 - (c.embedding <=> $1::vector) AS similarity
        FROM chunks c
        JOIN documents d ON d.id = c.document_id
        ORDER BY c.embedding <=> $1::vector
        LIMIT $2
        """,
        query_emb,
        top_k,
    )
    return [dict(row) for row in rows]


SYSTEM_PROMPT = """Você é assistente da Empresa Tal. Responda APENAS
com base no contexto fornecido. Se a resposta não está no contexto,
diga "Não tenho essa informação na minha base — pode reformular ou
pedir contato com humano?". NÃO invente.

Cite a fonte (título do documento) ao fim da resposta entre [colchetes]."""


async def rag_query(question: str, conn) -> str:
    chunks = await retrieve(question, top_k=5, conn=conn)

    # Threshold mínimo de similarity (evita responder com contexto irrelevante)
    relevant = [c for c in chunks if c["similarity"] > 0.7]
    if not relevant:
        return "Não tenho essa informação na minha base. Pode reformular?"

    context = "\n\n---\n\n".join(
        f"FONTE: {c['title']}\n{c['content']}" for c in relevant
    )

    completion = await client.chat.completions.create(
        model="deepseek-v3",
        temperature=0.2,
        max_tokens=500,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": f"CONTEXTO:\n{context}\n\nPERGUNTA: {question}",
            },
        ],
    )
    return completion.choices[0].message.content

Etapa 4 — Streaming pro frontend (UX importa)

Backend FastAPI:

from fastapi.responses import StreamingResponse

@app.post("/api/rag/stream")
async def rag_stream(req: QueryRequest):
    async def event_generator():
        chunks = await retrieve(req.question, top_k=5)
        relevant = [c for c in chunks if c["similarity"] > 0.7]
        if not relevant:
            yield "data: " + json.dumps({"type": "no_context"}) + "\n\n"
            return

        # Manda fontes primeiro pra UI mostrar enquanto LLM responde
        yield "data: " + json.dumps({
            "type": "sources",
            "sources": [
                {"title": c["title"], "similarity": c["similarity"]}
                for c in relevant
            ],
        }) + "\n\n"

        context = build_context(relevant)
        stream = await client.chat.completions.create(
            model="deepseek-v3",
            stream=True,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": f"CONTEXTO:\n{context}\n\nPERGUNTA: {req.question}"},
            ],
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            if delta:
                yield "data: " + json.dumps({"type": "delta", "content": delta}) + "\n\n"
        yield "data: " + json.dumps({"type": "done"}) + "\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

Custo real por query

Para o caso de uso "FAQ corporativo 200 docs ~50 chunks cada":

| Operação | Custo | |---|---| | Indexação inicial (10k chunks × ~400 tokens) | $0.08 USD (R$ 0,68) | | Storage Supabase | ~0,5 GB → free tier | | Query embedding (~30 tokens) | $0.000001 USD | | Query retrieve (Postgres index) | ~5ms, $0 | | LLM resposta (~3k context + 500 output) | $0.001 USD (R$ 0,01) | | Total por query | ~R$ 0,01 |

10.000 queries/mês = R$ 100/mês total (embeddings de query são desprezíveis vs LLM).

Os 6 trade-offs que ninguém fala

1. Chunks ruins = respostas ruins

90% dos problemas em RAG são culpa do chunking, não do LLM. Dedique tempo iterando. Visualize: pegue 10 queries reais, veja os chunks recuperados, identifique se faz sentido.

2. Similaridade alta ≠ resposta certa

Cosine similarity 0.85 pode ser tópico relacionado mas não a resposta. Solução: reranking com modelo cross-encoder ou um LLM intermediário ("dado essa pergunta e esses 5 candidatos, qual mais responde?").

3. RAG não substitui base estruturada

Pergunta "qual o preço do plano X?" → resposta tem que ser exata. RAG pode alucinar valor errado se o chunk tiver ambiguidade. Pra dado estruturado (preço, datas, contatos), faça lookup direto em tabela e injete no contexto, não busque por embedding.

4. Atualização é dolorida

Documento muda → você precisa re-indexar. Vale fazer cron diário re-embedding dos docs modificados (UPDATE chunks WHERE document_id IN (SELECT id FROM documents WHERE updated_at > now() - interval '1 day')).

5. Contexto grande ≠ resposta melhor

Tentação: passar top-20 chunks pra LLM. Resultado: confusão, "lost in the middle" (LLM ignora chunks no meio do contexto). Use top-3 a top-5 e re-rank antes.

6. Avaliação é fundamental e ninguém faz

Crie eval set de 50-100 (pergunta, resposta esperada) e rode regressão toda vez que muda prompt/modelo. Métrica simples: % de respostas onde LLM-evaluator (outro LLM) considera "factualmente correta vs fonte".

Anti-pattern: usar RAG quando não precisa

Não use RAG se:

• Base é < 50 documentos curtos: cabe no system prompt. Just stuff it. Mais barato e simples.
• Pergunta é estruturada (preço, prazo, contato): SQL direto.
• Pergunta é matemática/lógica: function calling melhor.
• Documentos mudam toda hora: complexidade de re-index não vale. Considere search tradicional (full-text + BM25).

Próximo passo

1. Cria conta Supabase free tier + habilita pgvector extension
2. Cria conta Tokia no /dashboard + key com modelos text-embedding-3-small + deepseek-v3
3. Indexa 10 documentos teste + roda 5 queries reais
4. Mede no /dashboard/usage quanto custou e ajusta chunk size

Quando essa stack não couber (você ultrapassou 1M vetores ou precisa sub-50ms latency consistente), aí sim considera Pinecone/Weaviate.

Links

• Stack completa de IA pro dev brasileiro
• DeepSeek V3 vs Gemini Flash: comparativo BRL
• Webhooks outbound + retry com backoff