RAG em Produção: Arquitetura, Chunking e Avaliação

RAG (Retrieval-Augmented Generation) é hoje uma das técnicas mais usadas em aplicações LLM de produção. Mas a diferença entre um RAG que funciona em demo e um que funciona em produção é enorme. Neste artigo, compartilho o que aprendi implementando RAG para análise de contratos no Detran-RJ.

A Arquitetura Completa

Um sistema RAG de produção tem três fases distintas:

1. Indexação (Offline)

Documentos → Extração de Texto → Chunking → Embedding → Vector Store

Documentos → Extração de Texto → Chunking → Embedding → Vector Store

2. Recuperação (Online)

Query → Embedding → Busca Vetorial → Re-ranking → Contexto

Query → Embedding → Busca Vetorial → Re-ranking → Contexto

3. Geração (Online)

Contexto + Query → Prompt Engineering → LLM → Resposta

Contexto + Query → Prompt Engineering → LLM → Resposta

Chunking: O Passo Mais Subestimado

A maioria dos tutoriais usa chunking simples por tamanho fixo. Em produção, isso é um erro grave. Uso três estratégias diferentes dependendo do tipo de documento:

Chunking Semântico para Contratos

python

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Para documentos legais: chunking semântico
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # Mais conservador = chunks maiores
)

# Para documentos técnicos: hierárquico
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Para documentos legais: chunking semântico
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,  # Mais conservador = chunks maiores
)

# Para documentos técnicos: hierárquico
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)

Metadata Enriquecida

Cada chunk deve carregar metadata rica para filtros e re-ranking:

python

def create_chunk_with_metadata(text: str, doc_metadata: dict, chunk_index: int) -> dict:
    return {
        "page_content": text,
        "metadata": {
            "doc_id": doc_metadata["id"],
            "doc_type": doc_metadata["type"],  # "contrato", "edital", "lei"
            "section": extract_section(text),
            "chunk_index": chunk_index,
            "created_at": doc_metadata["created_at"],
            "keywords": extract_keywords(text),  # TF-IDF ou KeyBERT
        }
    }

def create_chunk_with_metadata(text: str, doc_metadata: dict, chunk_index: int) -> dict:
    return {
        "page_content": text,
        "metadata": {
            "doc_id": doc_metadata["id"],
            "doc_type": doc_metadata["type"],  # "contrato", "edital", "lei"
            "section": extract_section(text),
            "chunk_index": chunk_index,
            "created_at": doc_metadata["created_at"],
            "keywords": extract_keywords(text),  # TF-IDF ou KeyBERT
        }
    }

PGvector com Supabase

Para produção, uso PGvector via Supabase. A vantagem é ter SQL completo junto com busca vetorial:

sql

-- Criação da tabela
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    content TEXT NOT NULL,
    embedding VECTOR(1536),  -- OpenAI ada-002
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Índice HNSW para performance
CREATE INDEX ON documents 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Função de busca híbrida (vetorial + full-text)
CREATE OR REPLACE FUNCTION hybrid_search(
    query_embedding VECTOR(1536),
    query_text TEXT,
    match_count INT DEFAULT 10,
    filter_metadata JSONB DEFAULT '{}'
)
RETURNS TABLE (id UUID, content TEXT, similarity FLOAT, metadata JSONB)
LANGUAGE plpgsql AS $$
BEGIN
    RETURN QUERY
    SELECT 
        d.id,
        d.content,
        1 - (d.embedding <=> query_embedding) AS similarity,
        d.metadata
    FROM documents d
    WHERE 
        d.metadata @> filter_metadata
        AND (
            1 - (d.embedding <=> query_embedding) > 0.7
            OR to_tsvector('portuguese', d.content) @@ plainto_tsquery('portuguese', query_text)
        )
    ORDER BY similarity DESC
    LIMIT match_count;
END;
$$;

-- Criação da tabela
CREATE TABLE documents (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    content TEXT NOT NULL,
    embedding VECTOR(1536),  -- OpenAI ada-002
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Índice HNSW para performance
CREATE INDEX ON documents 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Função de busca híbrida (vetorial + full-text)
CREATE OR REPLACE FUNCTION hybrid_search(
    query_embedding VECTOR(1536),
    query_text TEXT,
    match_count INT DEFAULT 10,
    filter_metadata JSONB DEFAULT '{}'
)
RETURNS TABLE (id UUID, content TEXT, similarity FLOAT, metadata JSONB)
LANGUAGE plpgsql AS $$
BEGIN
    RETURN QUERY
    SELECT 
        d.id,
        d.content,
        1 - (d.embedding <=> query_embedding) AS similarity,
        d.metadata
    FROM documents d
    WHERE 
        d.metadata @> filter_metadata
        AND (
            1 - (d.embedding <=> query_embedding) > 0.7
            OR to_tsvector('portuguese', d.content) @@ plainto_tsquery('portuguese', query_text)
        )
    ORDER BY similarity DESC
    LIMIT match_count;
END;
$$;

Re-ranking com Cross-Encoder

A busca vetorial retorna candidatos, mas o re-ranking com cross-encoder melhora muito a precisão:

python

from sentence_transformers import CrossEncoder

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

def rerank_results(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
    pairs = [(query, doc["content"]) for doc in candidates]
    scores = reranker.predict(pairs)
    
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    return [doc for doc, score in ranked[:top_k] if score > 0.3]

from sentence_transformers import CrossEncoder

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

def rerank_results(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
    pairs = [(query, doc["content"]) for doc in candidates]
    scores = reranker.predict(pairs)
    
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    return [doc for doc, score in ranked[:top_k] if score > 0.3]

Avaliação com LangSmith + RAGAS

Avaliar RAG é tão importante quanto construí-lo. Uso RAGAS para métricas automáticas:

python

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Dataset de avaliação
eval_data = {
    "question": ["Qual é o prazo de vigência do contrato?", ...],
    "answer": [rag_pipeline(q) for q in questions],
    "contexts": [retrieve_contexts(q) for q in questions],
    "ground_truth": ["O contrato tem vigência de 12 meses...", ...],
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.89
# answer_relevancy: 0.92
# context_precision: 0.85
# context_recall: 0.78

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Dataset de avaliação
eval_data = {
    "question": ["Qual é o prazo de vigência do contrato?", ...],
    "answer": [rag_pipeline(q) for q in questions],
    "contexts": [retrieve_contexts(q) for q in questions],
    "ground_truth": ["O contrato tem vigência de 12 meses...", ...],
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.89
# answer_relevancy: 0.92
# context_precision: 0.85
# context_recall: 0.78

Erros Comuns que Cometi

1. Chunk size muito pequeno: chunks de 200 tokens perdem contexto. Use 800-1200 para documentos legais.

2. Sem overlap: sem overlap entre chunks, perguntas sobre informações na fronteira entre dois chunks falham.

3. Embedding model errado: para português, use modelos multilíngues como multilingual-e5-large ou fine-tune um modelo em seu domínio.

4. Ignorar re-ranking: a busca vetorial por cosseno não é perfeita. Re-ranking com cross-encoder pode melhorar a precisão em 15-20%.

5. Sem avaliação contínua: RAG degrada com o tempo à medida que os documentos mudam. Monitore métricas semanalmente.

Conclusão

RAG em produção é muito mais do que vectorstore.similarity_search(query). A diferença entre um sistema que impressiona em demo e um que entrega valor real está nos detalhes: chunking semântico, metadata rica, busca híbrida, re-ranking e avaliação contínua.

Moises Costa é AI Engineer no Detran-RJ. Veja mais projetos no GitHub.

RAG em Produção: Arquitetura, Chunking e Avaliação

RAG em Produção: Arquitetura, Chunking e Avaliação

A Arquitetura Completa

1. Indexação (Offline)

2. Recuperação (Online)

3. Geração (Online)

Chunking: O Passo Mais Subestimado

Chunking Semântico para Contratos

Metadata Enriquecida

PGvector com Supabase

Re-ranking com Cross-Encoder

Avaliação com LangSmith + RAGAS

Erros Comuns que Cometi

Conclusão

References & Resources

See the concepts in action

Related Articles

Orquestração Agêntica com LangGraph: Do Zero ao Deploy

System Prompts e Role Prompts: Engenharia de Prompts Avançada

Full Stack AI: React + FastAPI + LangChain em Produção