4.2 — Pipeline de Indexação a Resposta: RAG e Agentes no Produto (RAG + Agents)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Projetar o pipeline completo de RAG: ingestão → chunking → embedding → busca → geração

Escolher a estratégia de chunking correta para documentos jurídicos/técnicos

Implementar RAG com pgvector em vez de vector DB externo

Integrar busca semântica como ferramenta em um agente Claude

Avaliar a qualidade do RAG com métricas antes do deploy

Por que isso importa

RAG resolve o problema central de LLMs em produtos: eles não conhecem seus documentos. A diferença entre RAG bom e RAG ruim não está no LLM — está na qualidade do chunking e na relevância dos chunks recuperados. Um chunk de 2000 tokens que inclui texto irrelevante piora mais a resposta do que não ter RAG.

Conceitos Fundamentais

Pipeline de RAG em 5 Etapas

INGESTÃO:

  - Ler PDF/DOCX/TXT

  - Extrair texto (pdfplumber para PDFs com tabelas, pypdf para simples)

  - Normalizar: remover headers/footers repetidos, quebras de linha extras



CHUNKING:

  - Dividir em chunks de tamanho controlado

  - Estratégia por domínio (ver abaixo)

  - Overlap para não perder contexto nas bordas



EMBEDDING:

  - Converter cada chunk em vetor numérico

  - OpenAI text-embedding-3-small: 1536 dimensões

  - Armazenar vetor + metadados (arquivo, página, posição)



BUSCA (RETRIEVE):

  - Converter query do usuário em vetor

  - Buscar top-K chunks mais similares (cosine similarity)

  - Filtrar por metadados se necessário (ex: apenas do contrato X)



GERAÇÃO:

  - Montar contexto: chunks relevantes + histórico da conversa

  - Chamar Claude com instrução: "Responda baseado APENAS no contexto fornecido"

  - Citar fonte (arquivo + página) na resposta

Estratégias de Chunking

# Chunking fixo (simples, OK para texto contínuo):

def chunking_fixo(texto: str, tamanho: int = 800, overlap: int = 100) -> list[str]:

    chunks = []

    inicio = 0

    while inicio < len(texto):

        fim = min(inicio + tamanho, len(texto))

        chunks.append(texto[inicio:fim])

        inicio += tamanho - overlap  # overlap para não perder contexto

    return chunks



# Chunking por parágrafo (melhor para documentos jurídicos):

def chunking_paragrafo(texto: str, max_chars: int = 1000) -> list[str]:

    paragrafos = texto.split('\n\n')

    chunks = []

    atual = ""

    for p in paragrafos:

        if len(atual) + len(p) < max_chars:

            atual += p + "\n\n"

        else:

            if atual:

                chunks.append(atual.strip())

            atual = p + "\n\n"

    if atual:

        chunks.append(atual.strip())

    return chunks



# Chunking hierárquico (melhor para contratos com cláusulas):

def chunking_clausulas(texto: str) -> list[dict]:

    """Identifica cláusulas por padrão 'Cláusula X' ou 'Art. X'"""

    import re

    padrao = r'(?:Cláusula|CLÁUSULA|Art\.|Artigo)\s+\w+'

    partes = re.split(padrao, texto)

    titulos = re.findall(padrao, texto)

    return [

        {"titulo": t, "conteudo": c.strip()}

        for t, c in zip(titulos, partes[1:])

        if c.strip()

    ]

RAG com pgvector (sem vector DB externo)

-- Setup (uma vez):

CREATE EXTENSION IF NOT EXISTS vector;



CREATE TABLE chunks (

    id SERIAL PRIMARY KEY,

    arquivo VARCHAR(255),

    pagina INTEGER,

    conteudo TEXT,

    embedding vector(1536),  -- dimensões do text-embedding-3-small

    created_at TIMESTAMP DEFAULT NOW()

);



-- Índice para busca eficiente:

CREATE INDEX ON chunks USING ivfflat (embedding vector_cosine_ops)

WITH (lists = 100);  -- ajustar lists para ~sqrt(n_chunks)

# Inserir chunk:

def inserir_chunk(conn, arquivo, pagina, conteudo, embedding):

    conn.execute(

        "INSERT INTO chunks (arquivo, pagina, conteudo, embedding) VALUES (%s, %s, %s, %s)",

        (arquivo, pagina, conteudo, embedding)  # embedding é lista de floats

    )



# Buscar top-K chunks similares:

def buscar_similar(conn, query_embedding, k=5, filtro_arquivo=None):

    sql = """

        SELECT arquivo, pagina, conteudo,

               1 - (embedding <=> %s::vector) as similaridade

        FROM chunks

        WHERE (%s IS NULL OR arquivo = %s)

        ORDER BY embedding <=> %s::vector

        LIMIT %s

    """

    return conn.execute(sql, (query_embedding, filtro_arquivo, filtro_arquivo, query_embedding, k))

Aprofundamento Técnico

RAG como Ferramenta do Agente

// Integrar busca como tool do Claude:

const tools = [{

  name: "buscar_documentos",

  description: "Busca trechos relevantes dos documentos indexados para responder a pergunta do usuário",

  input_schema: {

    type: "object",

    properties: {

      query: { type: "string", description: "Termos de busca" },

      n_resultados: { type: "number", description: "Número de trechos (default: 5)" },

    },

    required: ["query"],

  },

}]



// Implementação:

async function executarBusca(query: string, n = 5): Promise<string> {

  const embedding = await openai.embeddings.create({

    model: "text-embedding-3-small",

    input: query,

  })

  const chunks = await buscarSimilar(embedding.data[0].embedding, n)

  return chunks.map(c => `[${c.arquivo} p.${c.pagina}]\n${c.conteudo}`).join("\n\n---\n\n")

}

Avaliando Qualidade do RAG

def avaliar_rag(pares_pergunta_resposta: list[dict]) -> dict:

    """

    pares: [{"pergunta": str, "resposta_esperada": str}]

    Avalia se os chunks recuperados contêm a resposta esperada.

    """

    hits = 0

    for par in pares_pergunta_resposta:

        chunks = buscar_chunks(par["pergunta"], k=5)

        conteudo_chunks = " ".join(c["conteudo"] for c in chunks)

        # Verificar se a resposta esperada aparece nos chunks (recall aproximado)

        palavras_resposta = par["resposta_esperada"].lower().split()[:5]

        if any(p in conteudo_chunks.lower() for p in palavras_resposta):

            hits += 1

    

    recall = hits / len(pares_pergunta_resposta)

    return {"recall_chunks": recall, "n_testado": len(pares_pergunta_resposta)}

Exemplos Anotados

Exemplo 1: Contrato Jurídico com Chunking Hierárquico

contrato = """

Cláusula 1 — Objeto

O presente contrato tem por objeto a prestação de serviços de consultoria...



Cláusula 5 — Rescisão

Qualquer das partes poderá rescindir o presente contrato mediante aviso prévio

de 30 (trinta) dias...

"""



chunks = chunking_clausulas(contrato)

# [

#   {"titulo": "Cláusula 1", "conteudo": "O presente contrato tem por objeto..."},

#   {"titulo": "Cláusula 5", "conteudo": "Qualquer das partes poderá..."},

# ]



# Query: "Qual o prazo de aviso prévio para rescisão?"

# → chunk da Cláusula 5 tem score alto → LLM responde "30 dias" com fonte correta

Padrões e Armadilhas

Padrões

Padrão 1: Chunking por domínio, não tamanho fixo

Texto legal → cláusulas/artigos

Documentação técnica → seções/subseções

Artigos científicos → parágrafos

Texto geral → chunking fixo com overlap

Padrão 2: Sempre incluir metadados no chunk

# Sempre armazenar: arquivo, página, posição

# Para citar fonte: "Cláusula 5 do Contrato X (p. 3)"

Armadilhas

⚠️ Armadilha 1: Chunks muito grandes

Chunk de 3000 tokens → LLM recebe texto irrelevante → resposta piora

Chunk ideal: 200-800 tokens para documentos técnicos/jurídicos

⚠️ Armadilha 2: Sem overlap entre chunks

Informação na borda entre chunk N e N+1 → nunca recuperada

Overlap de 10-20% (50-150 tokens) garante continuidade

⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

Esta unidade guia a Entrega 1 (RAG CLI). Use exercicios.md e o README em entrega-1-rag-cli/ para: 1. Implementar pipeline completo: ingestão → chunking → embedding → pgvector → busca → Claude 2. Testar com 5 perguntas sobre um documento real 3. Calcular recall com o avaliador acima

Agora você está pronto para o lab.