Pipeline RAG Completo: do Chunking à Resposta Certa

Lede: RAG é uma cadeia — quebra em qualquer elo, o resultado é alucinação. Esse artigo cobre o pipeline inteiro: pré-processamento, chunking strategies, embeddings, vector store, hybrid search via RRF, re-ranking, augmentation, geração e avaliação. Tudo com exemplos vendor-neutral em Python. Funciona pra qualquer stack: Pinecone, Qdrant, Weaviate, pgvector, ChromaDB, Milvus, FAISS ou hnswlib.

---

1. Contexto / Introdução

• RAG (Retrieval-Augmented Generation) = LLM + retrieval de documentos relevantes injetados no prompt antes da geração.
• O retrieval decide ~80% da qualidade. Chunking ruim = retrieval ruim = LLM alucina.
• Pipeline canônico: input → preprocess → chunk → embed → store → retrieve → augment → generate → evaluate.
• Pré-requisitos: Python 3.10+, numpy, e qualquer vector store (pgvector, ChromaDB, Qdrant, FAISS, etc).
• Trade-off central: latência × qualidade × custo.

---

2. Anatomia de um pipeline RAG

INPUT (query)
   │
   ▼
┌─────────────────────────────────────────┐
│ 1. PREPROCESS                           │  normalização, limpeza, query rewriting
└────────────────────┬────────────────────┘
                     ▼
┌─────────────────────────────────────────┐
│ 2. RETRIEVAL                            │  o coração do RAG
│  ┌────────────────────────────────────┐ │
│  │ 2a. BM25 (léxico)                  │ │
│  │ 2b. k-NN (vetorial)                │ │
│  │ 2c. RRF / fusão híbrida            │ │
│  │ 2d. Re-ranking (cross-encoder)     │ │
│  └────────────────────────────────────┘ │
└────────────────────┬────────────────────┘
                     ▼
┌─────────────────────────────────────────┐
│ 3. AUGMENTATION                         │  injeta contexto no prompt
└────────────────────┬────────────────────┘
                     ▼
┌─────────────────────────────────────────┐
│ 4. GENERATION                           │  LLM gera resposta
└────────────────────┬────────────────────┘
                     ▼
OUTPUT (resposta)

Cada bloco é melhorável independentemente. Mas o gargalo real está no retrieval — e dentro dele, no chunking.

---

3. Chunking — onde a mágica começa (e a dor também)

3.1 Por que chunking importa

LLMs têm janela de contexto finita (4k–200k tokens). Embeddings também têm limite de tokens (ex: 8192 para text-embedding-3-small). Logo, você precisa quebrar documentos em pedaços menores antes de embedar.

Chunking ruim = semântica diluída:

❌ Chunk gigante (5000 tokens)
"O capítulo 1 fala sobre X... [5000 tokens] ...o capítulo 50 fala sobre Z."
   → Embedding: "média" de 5000 tokens
   → Retrieval: traz o capítulo 50 quando você busca "capítulo 1"
   → LLM: recebe contexto irrelevante + alucina

Chunking bom = unidade semântica autocontida:

✅ Chunk focado (300 tokens)
"O capítulo 1 introduz o conceito de X, definido como..."
   → Embedding: vetor denso do conceito X
   → Retrieval: traz esse chunk quando você busca "X"
   → LLM: recebe contexto relevante + responde certo

3.2 Estratégias de chunking (do mais simples ao mais sofisticado)

a) Fixed-size com overlap

Quebra em N tokens com overlap K. É o mais simples e funciona surpreendentemente bem pra texto uniforme.

def fixed_chunk(text, chunk_size=500, overlap=50):
    tokens = text.split()
    chunks = []
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk = " ".join(tokens[i:i + chunk_size])
        chunks.append(chunk)
    return chunks

Quando usar: texto uniforme (logs, código, transcrições), prototipagem rápida.

b) Recursive character splitter

Tenta dividir por separadores hierárquicos: \n\n → \n → . → → char-by-char. Pega o maior separador que produza chunks dentro do limite.

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_text(document)

Quando usar: markdown, HTML, texto geral. É o default recomendado pra 80% dos casos.

c) Sentence-based

Quebra por sentença usando NLTK ou spaCy. Mantém fronteiras semânticas naturais.

import nltk
nltk.download("punkt")
from nltk.tokenize import sent_tokenize

def sentence_chunk(text, max_sentences=10):
    sentences = sent_tokenize(text)
    chunks, current = [], []
    for sent in sentences:
        current.append(sent)
        if len(current) >= max_sentences:
            chunks.append(" ".join(current))
            current = []
    if current:
        chunks.append(" ".join(current))
    return chunks

Quando usar: texto narrativo, jurídico, documentação técnica onde cada sentença é uma unidade lógica.

d) Semantic chunking (embedding-based)

Quebra o texto em sentenças, embeda cada uma, e agrupa sentenças adjacentes cuja similaridade semântica for alta. Quando a similaridade cai, vira um novo chunk.

from sentence_transformers import SentenceTransformer
import numpy as np

def semantic_chunk(text, threshold=0.5):
    sentences = sent_tokenize(text)
    model = SentenceTransformer("nomic-ai/nomic-embed-text-v1.5")
    embs = model.encode(sentences)

    chunks, current = [], [sentences[0]]
    for i in range(1, len(sentences)):
        sim = np.dot(embs[i], embs[i-1]) / (
            np.linalg.norm(embs[i]) * np.linalg.norm(embs[i-1])
        )
        if sim < threshold:
            chunks.append(" ".join(current))
            current = [sentences[i]]
        else:
            current.append(sentences[i])
    if current:
        chunks.append(" ".join(current))
    return chunks

Quando usar: documentos longos com mudança de tópico (papers, livros, transcrições). Mais caro (1 embedding por sentença), mas captura melhor a estrutura.

e) Document-aware (markdown, código, HTML)

Cada formato tem sua estrutura. Markdown já tem seções, código já tem funções, HTML já tem tags.

def markdown_chunk(md_text):
    """Divide por H1/H2/H3 e mantém heading no chunk."""
    chunks, current, current_level = [], [], 0
    for line in md_text.split("\n"):
        if line.startswith("# "):
            if current:
                chunks.append("\n".join(current))
            current, current_level = [line], 1
        elif line.startswith("## ") and current_level <= 2:
            if current:
                chunks.append("\n".join(current))
            current, current_level = [line], 2
        else:
            current.append(line)
    if current:
        chunks.append("\n".join(current))
    return chunks

Quando usar: docs técnicas, READMEs, código fonte (com tree-sitter por linguagem).

3.3 Tamanho ótimo de chunk

Regra empírica (LangChain/LlamaIndex benchmarks):

Tamanho	Caso de uso	Recall típico
100–200 tokens	FAQ, snippets curtos	~70%
300–500 tokens	Default RAG, melhor trade-off	~85%
500–1000 tokens	Documentos longos, contexto rico	~80%
1000+ tokens	Resumos, contexto narrativo	<70%

Teste você mesmo: rode recall@k no seu dataset de eval com 3–4 tamanhos diferentes antes de cravar.

3.4 Overlap strategies

Overlap de 10–20% do chunk_size captura informação que cruza fronteiras:

chunk_size = 500, overlap = 50 (10%):

[AAAAAABBBBBCCCCCCDDDDDDEEEEEE]   ← texto original
[AAAAAABBBBB]                      ← chunk 1
           [BBBBBCCCCCC]            ← chunk 2 (overlap de 5 tokens com chunk 1)
                    [CCCCCCDDDDDD]  ← chunk 3

Cuidado: overlap muito alto (>30%) introduz redundância no índice (mais memória, retrieval poluído).

3.5 Metadata extraction (filtros pós-retrieval)

Antes de embedar, extraia metadata que vira filtro de busca:

def extract_metadata(text, source):
    return {
        "source": source,
        "char_count": len(text),
        "token_count": len(text.split()),
        "language": detect_language(text),  # spaCy/langdetect
        "section": extract_heading(text),   # primeiro H1/H2
        "entities": extract_entities(text), # spaCy NER
        "created_at": datetime.now().isoformat(),
    }

Por que isso importa: depois do retrieval, você pode filtrar WHERE language = 'pt' AND section = 'API Reference' antes de mandar pro LLM.

3.6 Pre-processamento (limpeza)

Antes de chunkar, normalize:

import re, unicodedata

def clean(text):
    # Normaliza unicode (NFKC: "café" = "cafe\u0301")
    text = unicodedata.normalize("NFKC", text)
    # Remove HTML/Markdown residual
    text = re.sub(r"<[^>]+>", "", text)
    text = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", text)  # [text](url) → text
    # Normaliza whitespace
    text = re.sub(r"\s+", " ", text).strip()
    return text

Cuidado: nunca remova stopwords antes de embedar — embeddings modernos (nomic-embed-text, E5, BGE) esperam linguagem natural completa. Stopword removal degrada recall.

---

4. Embedding — texto vira vetor

Embedding model transforma texto em vetor denso (384–3072 dimensões). Textos similares ficam perto no espaço.

from sentence_transformers import SentenceTransformer
model = SentenceTransformer("nomic-ai/nomic-embed-text-v1.5")  # 768d, open-source

chunk = "RRF funde rankings de BM25 e k-NN pela fórmula 1/(k+rank) com k=60."
emb = model.encode(chunk)
print(emb.shape)  # (768,)

Comparativo de modelos:

Modelo	Dimensões	Acesso	Custo	Uso
text-embedding-3-small	1536	API	~$0.02/1M tok	Baseline API, melhor custo-benefício
text-embedding-3-large	3072	API	~$0.13/1M tok	Máxima qualidade OpenAI
nomic-embed-text v1.5	768	Local	grátis	~95% da qualidade OpenAI
BGE-large-en-v1.5	1024	Local	grátis	Top-tier em MTEB inglês
Cohere embed-v3	1024	API	varies	Multilingue forte
Voyage-3	1024	API	varies	Especializado em RAG

---

5. Vector store — onde os embeddings vivem

O vector store indexa embeddings pra k-NN eficiente em escala:

Camada	Stack	Quando
In-process	FAISS, hnswlib, Annoy	<10M vetores, latência mínima
DB nativa	pgvector (Postgres), sqlite-vec	Já tem DB, quer evitar novo serviço
Self-hosted	Qdrant, Milvus, Weaviate, ChromaDB	Controle total, escala horizontal
Gerenciado	Pinecone, Weaviate Cloud, Vertex Vector Search	Zero-ops, escala automática

Exemplo pgvector (Postgres nativo):

CREATE EXTENSION vector;
CREATE TABLE chunks (
  id BIGSERIAL PRIMARY KEY,
  doc_id BIGINT NOT NULL,
  content TEXT NOT NULL,
  embedding vector(768),
  metadata JSONB
);
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);

-- Buscar top-5 por similaridade do cosseno
SELECT id, content FROM chunks
ORDER BY embedding <=> $1  -- $1 = vetor de consulta
LIMIT 5;

---

6. Retrieval híbrido: Full-Text Search (BM25) + k-NN + RRF

Full-Text Search (FTS) é a busca léxica clássica: tokeniza o texto, monta um índice invertido (termo → lista de documentos), e ranqueia por relevância de keywords. O algoritmo padrão da indústria é o BM25 (Best Matching 25) — usado em Elasticsearch, OpenSearch, SQLite FTS5, PostgreSQL tsvector, Solr, Lucene e Vespa.

Por que FTS ainda importa no mundo dos embeddings? Termos técnicos raros (siglas, IDs, nomes próprios como "PyTorch 2.4", git rebase, psycopg.SQL, códigos de erro ECONNRESET) costumam ter embedding fraco porque o modelo nunca viu muitas variações deles. FTS/BM25 captura essas queries exatas com 100% de precisão.

A busca híbrida roda FTS (BM25) e k-NN (vetorial) em paralelo, depois funde os rankings por Reciprocal Rank Fusion (RRF):

RRF_Score(d) = w_vec / (k + rank_vec(d)) + w_fts / (k + rank_fts(d))
# k = 60 (constante de suavização)
# w_vec, w_fts = pesos por modalidade

Quem aparece bem em ambos rankings sobe rápido. Quem aparece só num ainda entra, com score menor.

def rrf_fusion(vec_results, bm25_results, k=60, w_vec=1.0, w_fts=1.0):
    scores = {}
    for rank, (doc_id, _) in enumerate(vec_results):
        scores[doc_id] = scores.get(doc_id, 0) + w_vec / (k + rank + 1)
    for rank, (doc_id, _) in enumerate(bm25_results):
        scores[doc_id] = scores.get(doc_id, 0) + w_fts / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

Stacks com hybrid nativo: Elasticsearch 8+, OpenSearch 2+, Weaviate, Vespa, Qdrant.

---

7. Re-ranking — o salto de qualidade final

Retrieval barato (BM25 + k-NN) ranqueia top-100. Re-ranker caro (cross-encoder) pega esses 100 e retorna top-5 com muito mais precisão.

from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query, candidates, top_k=5):
    pairs = [(query, c["content"]) for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [c for c, _ in ranked[:top_k]]

Ganho típico: +10–20% em recall@5 e precision@5. Custo: ~50ms por query pra top-100.

---

8. Quantização — 4× a 32× menos RAM

Embeddings float32 custam 3.072 bytes/vetor (768d). Em escala, isso estoura RAM. Soluções:

Formato	Bytes/vetor	Redução	Fidelidade	Quando
float32	3.072	—	100%	Baseline, <100k vetores
float16	1.536	50%	~99.9%	GPU inference
int8 (PQ)	768	75%	~95%	Custo médio
Binary	96	96.9%	~80%	Recall-heavy, re-rank depois
Product Quantization 4-bit	~256-512	~85%	~96-98%	Melhor trade-off em produção

Product Quantization (PQ) é o algoritmo mais usado em produção (FAISS, Vespa, Milvus). Divide o vetor em sub-vetores e quantiza cada um independentemente via k-means. Treinado uma vez no corpus, o índice inteiro cabe em <30% do espaço original com perda <5% de recall.

Trade-off honesto: a compressão não é grátis — cada query precisa descomprimir/desquantizar antes do k-NN. Mas como bytes reduzidos cabem inteiros no cache L2/L3, o ganho de I/O geralmente supera o custo de desquantização em bases com >50k vetores.

---

9. Augmentation + Geração

Com os top-5 chunks rankeados, monta o prompt:

def build_prompt(query, retrieved_chunks):
    context = "\n\n---\n\n".join(c["content"] for c in retrieved_chunks)
    return f"""Você é um assistente técnico. Use o contexto abaixo pra responder.
Se a resposta não estiver no contexto, diga "não sei com base no contexto fornecido".
Sempre cite a fonte entre colchetes, ex: [chunk_id=42].

Contexto:
{context}

Pergunta: {query}

Resposta:"""

def rag(query, llm_call, retriever, reranker, k_initial=20, k_final=5):
    candidates = retriever.search(query, k=k_initial)
    top = reranker(query, candidates, top_k=k_final)
    prompt = build_prompt(query, top)
    return llm_call(prompt)

---

10. Avaliação — sem isso, é achismo

Métricas padrão de RAG (use o RAGAS):

Métrica	O que mede	Como calcular
context_precision	Quão relevante é o contexto recuperado	LLM-judge: top-k contém resposta?
context_recall	Cobriu toda info necessária?	LLM-judge vs ground truth
faithfulness	LLM "inventou" algo fora do contexto?	LLM-judge: afirmações suportadas?
answer_relevancy	Resposta é relevante à query?	Embedding similarity query↔answer
answer_correctness	Resposta bate com ground truth?	LLM-judge vs ground truth

Dataset mínimo de eval: 50–100 pares (pergunta, resposta esperada, contexto esperado). Sem isso, otimize às cegas.

---

11. Otimizações de produção

• Cache de embeddings: hash do texto → embedding. Mesma string nunca re-embedada.
• Batch processing: model.encode(list, batch_size=64) é 5–10× mais rápido.
• Filtros de metadata: restrinja antes do k-NN (category, date, user_id). Reduz drasticamente o espaço de busca.
• Hybrid weight tuning: comece w_vec = w_fts = 1.0 e meça no eval.
• Quantização: Product Quantization 4-bit economiza ~85% de RAM com perda <5% de recall.
• Re-ranking: top-100 do retrieval barato → cross-encoder → top-5 pro LLM. +10–20% qualidade.
• Async embedding: use asyncio + aiohttp se tiver muitas queries simultâneas.

---

12. Quando usar / Quando NÃO usar

Quando usar RAG com pesquisa semântica + vetorial:

• Base com >500 documentos / chunks
• Queries em linguagem natural
• Sinônimos e paráfrases importam ("cancelar" vs "desistir" vs "abortar")
• Domínio técnico com termos parecidos (Python vs python, JS vs JavaScript)

Quando NÃO usar:

• Base pequena (<100 docs) — grep + LLM de contexto grande resolve
• Queries puramente léxicas (busca por ID exato, log de erro) — BM25 sozinho é mais barato
• Latência <50ms obrigatória — pipeline completo (retrieve + rerank + LLM) é lento
• Sem orçamento pra embedding — API cobra, local tem custo de GPU/CPU

---

13. TL;DR — cheatsheet rápida

Etapa	Opção default	Quando trocar
Chunking	Recursive (500 tokens, 50 overlap)	Docs longas → semantic chunking
Embedding API	OpenAI text-embedding-3-small	Privacidade → nomic-embed-text local
Embedding local	nomic-embed-text v1.5 (768d)	Qualidade máxima → BGE-large
Vector store	pgvector (Postgres)	>10M vetores → Qdrant/Milvus
Hybrid search	RRF (k=60, w=1:1)	Stack unificada → Weaviate/ES
Re-ranker	cross-encoder/ms-marco-MiniLM	Top-tier → Cohere Rerank 3
Quantização	Product Quantization (PQ) int8/4-bit	Sem latência → float32
LLM	GPT-4o-mini / Claude Sonnet	Custo fixo → Llama local
Eval framework	RAGAS	Tracing → LangSmith/Phoenix

Próximos passos:

• Definir estratégia de chunking baseado no tipo de doc (recursive é default)
• Escolher embedding (API vs local) baseado em custo/privacidade
• Implementar cache de embeddings
• Adicionar BM25 + RRF pro retrieval híbrido
• Adicionar re-ranker cross-encoder
• Criar dataset de eval (50–100 pares pergunta/resposta)
• Medir recall@5 antes e depois de cada otimização

---

Referências

• 📚 Reciprocal Rank Fusion (Cormack et al., 2009) — paper clássico do RRF
• 📚 The Probabilistic Relevance Framework: BM25 and Beyond (Robertson & Zaragoza, 2009) — paper de referência do BM25/FTS
• 📚 Product Quantization (Jégou et al., 2011) — paper original do PQ
• 📚 sentence-transformers — lib de embeddings local
• 📚 FAISS (Meta) — referência de ANN in-process
• 📚 pgvector — extensão Postgres pra vetores
• 📚 RAGAS — framework de avaliação de RAG
• 📚 HNSW paper (Malkov & Yashunin, 2018) — algoritmo de grafo pra ANN
• 📚 Chunking Strategies for LLM Applications (Pinecone) — guia de chunking
• 📚 LangChain Text Splitters — implementações de referência
• 🔧 LangChain RAG tutorial — implementação completa
• 🔧 OpenAI Embeddings Guide — embeddings como serviço
• 📖 Pinecone Learning Center — conceitos de vector search explicados