Como Usar RAG com LLMs: Busca Semântica na Prática

Como Usar RAG com LLMs: Implementação Prática e Casos de Uso

RAG (Retrieval Augmented Generation) é a arquitetura que resolve o problema mais frustrante dos LLMs: alucinações e conhecimento desatualizado. Neste guia você vai implementar um sistema RAG funcional do zero, aprender os padrões avançados e entender quando (e quando não) usar RAG.

---

O Problema Central que RAG Resolve

Imagine que você tem 10.000 documentos internos da empresa — políticas, manuais, contratos — e quer que um LLM responda perguntas sobre eles. As opções sem RAG são péssimas:

1. Enviar tudo no contexto: impossível — 10.000 docs não cabem em contexto nenhum
2. Fine-tuning: caro, lento, e os dados ficam "congelados" no modelo
3. Confiar na memória do modelo: ele vai inventar respostas plausíveis mas erradas

RAG resolve isso: busca só os trechos relevantes em tempo real e os injeta no contexto. O modelo responde baseado em evidência concreta, não em memória.

---

Arquitetura Completa de um Sistema RAG

[Documentos] → [Chunking] → [Embeddings] → [Vector Store]
                                                    ↓
[Pergunta do usuário] → [Embedding da pergunta] → [Busca por similaridade]
                                                    ↓
                                            [Top-K chunks relevantes]
                                                    ↓
                                            [LLM + contexto] → [Resposta]

---

Implementação Passo a Passo

Instalação

pip install langchain langchain-openai langchain-community \
            chromadb pypdf sentence-transformers python-dotenv

Passo 1: Carregar e Dividir Documentos

from langchain_community.document_loaders import (
    PyPDFLoader, 
    TextLoader,
    DirectoryLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

def carregar_documentos(caminho: str):
    """Carrega PDFs ou .txt de uma pasta"""
    loader = DirectoryLoader(
        caminho,
        glob="**/*.pdf",
        loader_cls=PyPDFLoader,
        show_progress=True
    )
    return loader.load()

def dividir_chunks(documentos, chunk_size=1000, overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        add_start_index=True
    )
    chunks = splitter.split_documents(documentos)
    print(f"Total de chunks: {len(chunks)}")
    return chunks

# Uso
docs = carregar_documentos("./documentos")
chunks = dividir_chunks(docs)

Passo 2: Criar Vector Store com Embeddings

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

def criar_vectorstore(chunks, caminho_persistencia="./chroma_db"):
    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small"  # mais barato e bom o suficiente
    )
    
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=caminho_persistencia,
        collection_metadata={"hnsw:space": "cosine"}
    )
    
    print(f"Vector store: {vectorstore._collection.count()} vetores indexados")
    return vectorstore

# Carregar vector store existente (não reindexar tudo)
def carregar_vectorstore_existente(caminho="./chroma_db"):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    return Chroma(
        persist_directory=caminho,
        embedding_function=embeddings
    )

Passo 3: Retriever com MMR (Anti-Redundância)

def criar_retriever(vectorstore, k=4, fetch_k=20):
    """
    MMR (Maximal Marginal Relevance) evita recuperar 4 chunks
    que dizem exatamente a mesma coisa. Balanceia relevância + diversidade.
    
    lambda_mult: 0 = máxima diversidade, 1 = máxima relevância
    """
    return vectorstore.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": k,
            "fetch_k": fetch_k,
            "lambda_mult": 0.7
        }
    )

Passo 4: Chain Completa

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser

PROMPT_RAG = """Você é um assistente que responde perguntas baseando-se 
EXCLUSIVAMENTE no contexto fornecido abaixo. 

Se a informação não estiver no contexto, responda: "Não encontrei informação 
sobre isso na base de conhecimento."

Nunca invente informações. Nunca use conhecimento fora do contexto.

Contexto:
{context}

---
Pergunta: {question}
Resposta:"""

def formatar_docs(docs) -> str:
    return "\n\n---\n\n".join([
        f"[Fonte: {doc.metadata.get('source', 'N/A')}, Página: {doc.metadata.get('page', 'N/A')}]\n{doc.page_content}"
        for doc in docs
    ])

def criar_rag_chain(vectorstore):
    retriever = criar_retriever(vectorstore)
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    prompt = ChatPromptTemplate.from_template(PROMPT_RAG)
    
    chain = (
        {"context": retriever | formatar_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain, retriever

# Uso completo
vectorstore = criar_vectorstore(chunks)
chain, retriever = criar_rag_chain(vectorstore)

pergunta = "Quantos dias de férias o colaborador tem direito?"
resposta = chain.invoke(pergunta)
print(resposta)

---

RAG com Claude (Anthropic)

from langchain_anthropic import ChatAnthropic

def criar_rag_chain_claude(vectorstore):
    retriever = criar_retriever(vectorstore)
    
    # Claude 3.5 Haiku é mais barato para RAG em volume
    llm = ChatAnthropic(
        model="claude-3-5-haiku-20241022",
        temperature=0,
        max_tokens=1500
    )
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Responda baseando-se EXCLUSIVAMENTE no contexto. 
         Se não estiver no contexto, diga que não encontrou a informação.
         
         Contexto:
         {context}"""),
        ("human", "{question}")
    ])
    
    chain = (
        {"context": retriever | formatar_docs, "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    
    return chain

---

Casos de Uso Reais

1. Chatbot de Documentação Técnica

class ChatbotDocumentacao:
    def __init__(self, pasta_docs: str):
        docs = carregar_documentos(pasta_docs)
        chunks = dividir_chunks(docs, chunk_size=800, overlap=150)
        vectorstore = criar_vectorstore(chunks)
        self.chain, self.retriever = criar_rag_chain(vectorstore)
        self.historico = []
    
    def perguntar(self, pergunta: str) -> dict:
        # Buscar chunks relevantes para citar as fontes
        docs_relevantes = self.retriever.get_relevant_documents(pergunta)
        fontes = list(set([
            doc.metadata.get('source', 'Desconhecido') 
            for doc in docs_relevantes
        ]))
        
        resposta = self.chain.invoke(pergunta)
        
        return {
            "resposta": resposta,
            "fontes": fontes,
            "chunks_usados": len(docs_relevantes)
        }

# Uso
bot = ChatbotDocumentacao("./manuais")
resultado = bot.perguntar("Como solicitar reembolso de despesas?")
print(resultado["resposta"])
print(f"Fontes: {resultado['fontes']}")

2. Análise de Contratos Jurídicos

def analisar_contrato(caminho_pdf: str, perguntas: list[str]) -> dict:
    loader = PyPDFLoader(caminho_pdf)
    docs = loader.load()
    chunks = dividir_chunks(docs, chunk_size=1500, overlap=300)
    
    vectorstore = Chroma.from_documents(
        chunks, 
        OpenAIEmbeddings(model="text-embedding-3-small"),
        collection_name=f"contrato_{hash(caminho_pdf)}"
    )
    
    chain, _ = criar_rag_chain(vectorstore)
    
    resultados = {}
    for pergunta in perguntas:
        resultados[pergunta] = chain.invoke(pergunta)
    
    return resultados

perguntas_contrato = [
    "Qual é o prazo de vigência do contrato?",
    "Quais são as condições de rescisão antecipada?",
    "Há cláusula de não-concorrência? Se sim, quais são os termos?",
    "Qual é a multa por descumprimento?"
]

analise = analisar_contrato("contrato_servicos.pdf", perguntas_contrato)
for pergunta, resposta in analise.items():
    print(f"\n**{pergunta}**\n{resposta}")

---

Estratégias Para Melhorar a Qualidade do RAG

Problema 1: Retriever não encontra o chunk certo

Solução: Query Expansion

from langchain_openai import ChatOpenAI

def expandir_query(pergunta: str) -> list[str]:
    """Gera variações da pergunta para aumentar o recall"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    
    prompt = f"""Gere 3 variações desta pergunta para buscar em documentos.
    Cada variação deve usar palavras-chave diferentes mas manter o mesmo significado.
    Retorne apenas as 3 variações, uma por linha.
    
    Pergunta: {pergunta}"""
    
    response = llm.invoke(prompt)
    variacoes = response.content.strip().split("\n")
    return [pergunta] + [v.strip() for v in variacoes if v.strip()]

def retriever_expandido(vectorstore, pergunta: str, k=4):
    variacoes = expandir_query(pergunta)
    
    todos_chunks = []
    vistos = set()
    
    for variacao in variacoes:
        docs = vectorstore.similarity_search(variacao, k=k)
        for doc in docs:
            chave = doc.page_content[:100]
            if chave not in vistos:
                vistos.add(chave)
                todos_chunks.append(doc)
    
    return todos_chunks[:k]  # retorna apenas os melhores k únicos

Problema 2: Chunks com contexto incompleto

Solução: Parent-Child Chunking

Guarda chunks pequenos para busca (melhor precisão), mas retorna o chunk pai maior para contexto:

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Splitter para os filhos (buscados)
filho_splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=50)

# Splitter para os pais (retornados)
pai_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

docstore = InMemoryStore()  # ou usar Redis para persistência

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=filho_splitter,
    parent_splitter=pai_splitter,
)

retriever.add_documents(documentos)

---

Quando NÃO Usar RAG

---

Avaliação: Como Medir a Qualidade do Seu RAG

# Métricas RAGAS (instalar: pip install ragas)
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall
from datasets import Dataset

def avaliar_rag(chain, retriever, pares_qa):
    """
    pares_qa: lista de dicts com 'question' e 'ground_truth'
    """
    dados = {"question": [], "answer": [], "contexts": [], "ground_truth": []}
    
    for par in pares_qa:
        pergunta = par["question"]
        docs = retriever.get_relevant_documents(pergunta)
        resposta = chain.invoke(pergunta)
        
        dados["question"].append(pergunta)
        dados["answer"].append(resposta)
        dados["contexts"].append([d.page_content for d in docs])
        dados["ground_truth"].append(par["ground_truth"])
    
    dataset = Dataset.from_dict(dados)
    resultado = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_recall])
    
    return resultado

---

Perguntas Frequentes

Qual vector store usar para produção?

Para produção em escala: Pinecone (gerenciado, SLA) ou Weaviate. Para desenvolvimento: Chroma (local, fácil). Para alto volume sem custo: Qdrant self-hosted.

Qual o melhor tamanho de chunk?

Comece com 800 chars e 150 de overlap. Ajuste medindo as métricas RAGAS com seus dados reais. Não há resposta universal.

RAG funciona bem com tabelas em PDF?

Mal. PDFs mal formatados perdem a estrutura de tabelas. Use camelot ou tabula para extrair tabelas de PDF antes de indexar.

Como indexar documentos novos sem reprocessar tudo?

Use hash do conteúdo para detectar mudanças e indexe apenas os documentos novos ou modificados.