RAG — Retrieval-Augmented Generation

1 O que é RAG?

A arquitetura que conecta LLMs aos seus documentos em tempo real.

Por que RAG é necessário?

Cenário: Nimbus Cloud

Vamos construir um RAG para a Nimbus Cloud, empresa fictícia de software SaaS. Ela tem:

2 O pipeline completo

Duas fases independentes: indexação (offline) e consulta (online).

Fase 1 — Indexação (roda uma vez por documento)

Fase 2 — Consulta (roda a cada pergunta)

Exemplo real: pergunta sobre timeout

# Pergunta do usuário
"Por que minha sincronização falha após 30 segundos?"

# 1. Embedding da pergunta
→ [0.19, -0.68, 0.51, ...]  # 768 dimensões

# 2. Busca vetorial retorna top-5 chunks
1. Ticket #4521 (score 0.94) # "timeout 30s endpoint /v2/sync"
2. Docs API v3.2 (score 0.88) # "timeout default: 30s"
3. Runbook RB-018 (score 0.81) # "aumentar NIMBUS_TIMEOUT"

# 3. Prompt montado
SYSTEM: Responda com base no contexto. Cite a fonte.
CONTEXTO: [Ticket #4521] ... [Docs API] ... [Runbook]
PERGUNTA: Por que minha sincronização falha após 30 segundos?

# 4. LLM responde
→ "O timeout padrão do endpoint /v2/sync é 30s (Docs API v3.2).
   Para arquivos >500MB, aumente a variável NIMBUS_TIMEOUT
   para 120s — solução aplicada no ticket #4521."

3 Parsing/ETL — limpando a sujeira

Antes do chunking, vem a parte mais subestimada: transformar documentos brutos em texto limpo.

Os 4 vilões do parsing

Exemplo: extraindo um ticket do Zendesk

import requests
from unstructured.partition.html import partition_html

def fetch_ticket(ticket_id):
    # 1. Busca ticket via API
    r = requests.get(f"https://nimbus.zendesk.com/api/v2/tickets/{ticket_id}.json")
    data = r.json()["ticket"]

    # 2. Extrai texto limpo do HTML dos comentários
    elements = partition_html(text=data["description"])
    clean_text = "\n".join([str(e) for e in elements])

    # 3. Enriquece com metadados (crucial para busca!)
    return {
        "text": clean_text,
        "metadata": {
            "source": "zendesk",
            "ticket_id": ticket_id,
            "priority": data["priority"],
            "status": data["status"],
            "created_at": data["created_at"],
            "tags": data["tags"]
        }
    }

Ferramentas do mercado

4 Chunking — cortando em pedaços

Dividir para conquistar — mas onde cortar faz toda diferença.

🎮 Simulador de Chunking

Estratégia por tipo de documento

Enriquecendo chunks com metadados

Prefixe cada chunk com informações do documento antes de indexar:

[Ticket #4521 | Acme Corp | Prioridade: Alta | API v3.2 | 2026-06-20]
Cliente reporta falhas intermitentes ao sincronizar arquivos
maiores que 500MB. Logs mostram timeout no endpoint /v2/sync
após exatamente 30 segundos.
Solução: aumentar NIMBUS_TIMEOUT para 120s.

5 Embeddings — o GPS do significado

Como um modelo comprime o significado de um texto em um vetor.

Na prática: scores de similaridade

Modelos populares em 2026

6 A matemática por trás

Cosine similarity, HNSW e por que a busca é rápida mesmo com milhões de vetores.

Cosine Similarity — a régua do RAG

import numpy as np

def cosine_similarity(a, b):
    """Calcula similaridade cosseno entre dois vetores."""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Exemplo: pergunta vs. chunk
pergunta = np.array([0.19, -0.68, 0.51, 0.09])
chunk_1  = np.array([0.23, -0.71, 0.45, 0.08])  # timeout sync
chunk_2  = np.array([-0.82, 0.34, -0.19, 0.91]) # receita bolo

print(cosine_similarity(pergunta, chunk_1))  # 0.997 ✅
print(cosine_similarity(pergunta, chunk_2))  # -0.42 ❌

O problema: escala

Se você tem 1 milhão de chunks, comparar a pergunta com cada um (força bruta) custa 1 milhão de cosine similarities. Em vetores de 768 dimensões, isso é lento.

A solução: índices aproximados (ANN)

7 Busca — vetorial, lexical e híbrida

Vetores não bastam — especialmente quando há códigos, números e IDs.

🎮 Simulador de Busca Híbrida

Reranking — o refinamento final

Após recuperar os top-K chunks (ex: 20), um segundo modelo (cross-encoder) reordena os resultados avaliando a pergunta e cada chunk juntos. É mais caro, mas muito mais preciso.

# Pipeline completo de busca
1. Busca híbrida → top-20 chunks (rápido, aproximado)
2. Cross-encoder reranker → reordena os 20 (lento, preciso)
3. Pega top-5 → envia ao LLM como contexto

8 Estratégias avançadas

Quando o chunking básico não é suficiente.

🪆 Parent-Child Retrieval (Small-to-Big)

Ideia: indexar chunks pequenos (precisão na busca) mas retornar o chunk pai (contexto completo para o LLM).

🧠 HyDE (Hypothetical Document Embeddings)

Problema: perguntas curtas geram embeddings "pobres". Solução: pedir ao LLM para gerar uma resposta fictícia antes de buscar.

🔗 GraphRAG

🤖 Agentic RAG

9 Engenharia de Prompt para RAG

O prompt é o "briefing" do estagiário. Mal feito, tudo desmorona.

Template de prompt para RAG

# SYSTEM
Você é o "Nimbus Assistant", ajudante de suporte técnico da Nimbus Cloud.

REGRAS:
1. Responda APENAS com base no CONTEXTO fornecido.
2. Se a informação não estiver no contexto, responda exatamente:
   "Não encontrei essa informação na base de conhecimento."
3. Cite a fonte de cada afirmação no formato [Fonte: XXX].
4. Se houver múltiplas fontes, liste todas.
5. Seja conciso e técnico.

CONTEXTO:
{contexto_formatado}

PERGUNTA DO USUÁRIO:
{pergunta}

Formatando o contexto com citações

def format_context(chunks):
    parts = []
    for i, c in enumerate(chunks, 1):
        src = c.metadata.get("ticket_id") or c.metadata.get("doc")
        parts.append(f"[Fonte {i}: {src}]\n{c.text}")
    return "\n\n---\n\n".join(parts)

10 Observabilidade e MLOps

Colocar em produção é fácil. Manter funcionando é que são elas.

🧠 Semantic Cache

📊 Tracing — rastreie tudo

Ferramentas: Langfuse (open-source), Arize Phoenix, LangSmith, Helicone.

📈 Avaliação contínua com RAGAS

11 Arquitetura e Deploy

Como empacotar tudo isso em algo que rode em produção.

Arquitetura recomendada

API mínima em FastAPI

from fastapi import FastAPI
from pydantic import BaseModel
import chromadb
import ollama

app = FastAPI()
db = chromadb.HttpClient(host="localhost", port=8000)
collection = db.get_collection("nimbus_docs")

@app.post("/ask")
async def ask(q: Query):
    # 1. Embed da pergunta
    emb = ollama.embed(model="nomic-embed-text", input=q.question)

    # 2. Busca vetorial
    results = collection.query(
        query_embeddings=[emb["embeddings"[0]]],
        n_results=q.top_k
    )

    # 3. Monta prompt
    contexto = format_context(results)
    prompt = SYSTEM_PROMPT.format(contexto=contexto, pergunta=q.question)

    # 4. Gera resposta
    resp = ollama.chat(
        model="llama3.1:8b",
        messages=[{"role": "user", "content": prompt}]
    )

    return {
        "answer": resp["message"]["content"],
        "sources": results["metadatas"][0]
    }

🐳 Docker Compose para subir tudo

version: '3.8'
services:
  api:
    build: .
    ports: ["8080:8080"]
    depends_on: [chromadb, ollama]

  chromadb:
    image: chromadb/chroma:latest
    ports: ["8000:8000"]
    volumes: ["./chroma_data:/chroma/chroma"]

  ollama:
    image: ollama/ollama:latest
    ports: ["11434:11434"]

  langfuse:
    image: langfuse/langfuse:2
    ports: ["3000:3000"]

🎯 Quiz — teste seu conhecimento

Clique em uma alternativa para ver se acertou.

→ O que vem a seguir?

Agora que você construiu um RAG completo, vamos dar ao seu sistema a capacidade de agir no mundo real.

Conceitos que vamos construir aqui