Tudo que você precisa saber sobre Retrieval-Augmented Generation: arquitetura, chunking, embeddings, vector databases, retrieval strategies, avaliação e como evitar os erros que derrubam sistemas em produção

Recursos selecionados para complementar sua leitura
AI Architect
Pós-graduado em arquitetura de software e soluções. Conecto profundidade técnica com resultados de negócio para entregar produtos que as pessoas realmente usam. Também mentoro desenvolvedores e criadores em programas ao vivo, podcasts e iniciativas de comunidade focadas em tecnologia inclusiva.
Checklist de 47 pontos para encontrar bugs, riscos de segurança e problemas de performance antes do lançamento.
Continue explorando tópicos similares

A comprehensive guide to spec-driven development workflows with AI coding assistants, featuring real-world Next.js examples and Claude Code commands.

Aqui está a verdade desconfortável: seu assistente de código com IA é burro. Não porque o modelo é fraco. Claude Sonnet 4.5 e GPT-4o são genuinamente brilhantes. O problema é que eles estão vendados.

Andrej Karpathy propõe um padrão onde LLMs constroem e mantêm um wiki persistente — em vez de redescobrir conhecimento do zero a cada consulta. Uma mudança fundamental na forma como usamos IA para gerenciar informação.
Templates testados em produção, usados por desenvolvedores. Economize semanas de setup no seu próximo projeto.
Consultorias modulares para founders e CTOs fracionados. Você recebe diagnóstico acionável e acompanhamento direto comigo.
2 vagas para consultorias no Q2
Toda semana alguém em algum time de engenharia descobre que conectar um modelo de linguagem a uma base de documentos é mais fácil do que parecia. Algumas linhas de Python, um modelo de embedding, uma chamada ao OpenAI e pronto — um chatbot que "sabe" sobre os documentos da empresa.
Duas semanas depois, o mesmo time descobre que o sistema responde errado com frequência, alucina informações que não estão nos documentos, ignora contexto relevante que está lá, e degrada conforme a base de dados cresce.
Esse é o arco de quase todos os projetos de RAG que vejo falhando. Não porque a tecnologia é ruim. Porque o time tratou um problema de engenharia como um problema de integração.
RAG não é uma feature. É uma disciplina.
Retrieval-Augmented Generation — o nome completo — é uma arquitetura que combina recuperação de informação com geração de texto. Em vez de esperar que o modelo memorize todo o conhecimento no treinamento, você ensina ele a buscar evidências antes de responder. É simples em conceito. É complexo em execução.
Este guia tem um objetivo: transformar você de alguém que usa RAG em alguém que entende RAG. Isso significa cobrir cada camada do pipeline com a profundidade que ela merece — incluindo os trade-offs que a maioria dos tutoriais ignora, os erros que só aparecem em produção, e as decisões de arquitetura que determinam se o sistema vai funcionar ou vai frustrar.
Vou cobrir:
Vou ser direto quando houver controvérsia. Vou citar pesquisa quando houver evidência. E vou nomear os problemas que costumam ser minimizados.
Modelos de linguagem grandes são treinados em corpora massivos. GPT-4 foi treinado com dados até um certo ponto no tempo. Claude foi treinado com dados até outro ponto. Nenhum deles sabe sobre o documento interno que sua empresa publicou ontem, sobre a decisão de arquitetura que foi tomada na reunião da semana passada, ou sobre o incidente de produção que aconteceu esta manhã.
Mais do que isso: mesmo informações que existiam durante o treinamento podem ser "esquecidas" ou distorcidas pelo modelo. A densidade de representação de um tópico no corpus de treinamento afeta a confiança e precisão das respostas. Um modelo treinado em trilhões de tokens ainda pode responder incorretamente sobre um assunto específico simplesmente porque aquele assunto estava sub-representado nos dados.
Fine-tuning resolve parte do problema — você pode especializar um modelo em um domínio. Mas fine-tuning tem limitações severas:
RAG resolve o problema de forma diferente. Em vez de ensinar o modelo a memorizar fatos, você ensina ele a buscar evidências no momento da resposta. A pergunta do usuário aciona uma busca em uma base de conhecimento, os documentos mais relevantes são recuperados, e o modelo usa esses documentos como contexto para construir a resposta.
Isso muda a natureza do sistema:
| Característica | Modelo puro | RAG |
|---|---|---|
| Conhecimento atualizado | Limitado à data de treino | Atualiza conforme a base |
| Verificabilidade | Opaca | Citável |
| Custo de atualização | Alto (re-treino) | Baixo (atualiza documentos) |
| Precisão em domínio específico | Variável | Alta, com boa base |
| Escala de conhecimento | Limitada pelo contexto | Ilimitada (base externa) |
O artigo fundacional de RAG foi publicado pelo Facebook AI Research (agora Meta AI) em 2020: "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks", de Lewis et al. O paper propunha uma arquitetura end-to-end que combinava um retriever denso (Dense Passage Retrieval, ou DPR) com um gerador seq2seq (BART), treinados conjuntamente.
O insight central do paper era elegante: em vez de forçar o modelo a memorizar todo o conhecimento no parâmetro, você separa conhecimento paramétrico (o que o modelo aprendeu no treinamento) de conhecimento não-paramétrico (o que está na base de documentos recuperável em tempo de inferência).
O paper de 2020 era focado em benchmarks de QA como Natural Questions e TriviaQA. Mas os princípios são os mesmos que hoje guiam sistemas RAG em produção — com a diferença de que os modelos de linguagem melhoraram dramaticamente, os modelos de embedding ficaram muito mais precisos, e a infraestrutura de vector search ficou muito mais acessível.
Entre 2020 e 2026, RAG passou de técnica de pesquisa para padrão de engenharia. Algumas datas importantes:
Hoje RAG é uma commodity de infraestrutura. Qualquer empresa pode ter um sistema básico funcionando em dias. A diferença entre times está em entender as camadas com profundidade suficiente para fazer as escolhas certas.
Todo sistema RAG tem quatro componentes fundamentais. Entendê-los separadamente é o primeiro passo para tomar decisões de arquitetura corretas.
Seu repositório de documentos. Pode ser:
A qualidade da base de conhecimento é o fator mais subestimado na qualidade de um sistema RAG. Lixo entra, lixo sai — mas de forma muito mais opaca do que em sistemas tradicionais, porque o modelo pode combinar pedaços de lixo de formas que parecem plausíveis.
O processo que transforma documentos brutos em representações pesquisáveis. Inclui:
O componente que, dada uma pergunta, encontra os pedaços de documento mais relevantes. É aqui que a maior parte da engenharia acontece. Estratégias incluem:
O modelo de linguagem que usa os documentos recuperados para construir a resposta. A qualidade do prompt, a forma de estruturar o contexto, e as instruções sobre como usar (e quando não usar) os documentos são críticas aqui.
Agora vamos descer um nível e olhar para cada etapa com precisão técnica.
Antes de qualquer coisa, você precisa transformar documentos em texto limpo. Parece trivial. Não é.
PDFs são um pesadelo. PDFs são essencialmente instruções de impressão, não estruturas semânticas. Um PDF pode ter:
Bibliotecas como PyPDF2 têm qualidade inconsistente com PDFs complexos. pdfplumber e pymupdf (fitz) são geralmente superiores. Para PDFs com layouts complexos, ferramentas como unstructured.io usam modelos de visão para entender layout antes de extrair texto.
Word/DOCX são mais fáceis de processar, mas ainda têm nuances — estilos de cabeçalho que definem hierarquia semântica, tabelas embutidas, comentários.
HTML/Markdown são os formatos mais amigáveis para RAG porque preservam estrutura semântica.
# Exemplo com unstructured.io - para documentos complexos
from unstructured.partition.pdf import partition_pdf
# Estratégia "hi_res" usa modelo de visão para layout
elements = partition_pdf(
filename="documento-complexo.pdf",
strategy="hi_res",
infer_table_structure=True,
extract_images_in_pdf=True
)
# Elementos já têm tipo semântico: NarrativeText, Table, Title, etc.
for element in elements:
print(type(element).__name__, ":", str(element)[:100])
A decisão sobre qual parser usar é crítica e frequentemente ignorada. Um sistema RAG que usa PyPDF2 para extrair tabelas vai indexar lixo. O modelo vai usar esse lixo para responder perguntas. As respostas vão parecer erradas "misteriosamente".
Texto extraído raramente está limpo. O que geralmente precisa de tratamento:
Uma armadilha comum é tratar limpeza como um passo de pre-processamento genérico. Na prática, a estratégia de limpeza depende do tipo de documento. Documentação técnica com exemplos de código precisa de tratamento diferente de emails corporativos.
Esta é a camada que mais afeta a qualidade de recuperação e que mais é negligenciada. Discutiremos em profundidade em seção dedicada.
Cada chunk é transformado em um vetor numérico de alta dimensão por um modelo de embedding. O vetor captura o significado semântico do texto de forma que textos similares em significado ficam próximos no espaço vetorial.
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-en-v1.5")
chunks = [
"RAG combina recuperação de informação com geração de texto.",
"Transformers revolucionaram o processamento de linguagem natural.",
"Pizza é um prato italiano popularmente consumido no mundo inteiro.",
]
# Dimensão depende do modelo — bge-large tem 1024 dimensões
embeddings = model.encode(chunks, normalize_embeddings=True)
print(f"Shape: {embeddings.shape}") # (3, 1024)
Os vetores são persistidos junto com os metadados e o texto original. A estrutura de metadados é crucial para filtragem posterior.
# Exemplo com Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
client = QdrantClient(url="http://localhost:6333")
# Criação da collection com dimensão e métrica
client.create_collection(
collection_name="knowledge_base",
vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)
# Inserção dos pontos com metadados
points = [
PointStruct(
id=i,
vector=embeddings[i].tolist(),
payload={
"text": chunks[i],
"source": "documento-1.pdf",
"page": 3,
"section": "Introdução",
"created_at": "2026-04-14",
}
)
for i in range(len(chunks))
]
client.upsert(collection_name="knowledge_base", points=points)
Quando o usuário faz uma pergunta, ela é convertida em embedding com o mesmo modelo usado na indexação. O vector database retorna os k chunks mais similares.
def retrieve(query: str, k: int = 5) -> list[dict]:
# Mesma normalização usada na indexação
query_embedding = model.encode([query], normalize_embeddings=True)[0]
results = client.search(
collection_name="knowledge_base",
query_vector=query_embedding.tolist(),
limit=k,
with_payload=True,
)
return [
{
"text": r.payload["text"],
"source": r.payload["source"],
"score": r.score,
}
for r in results
]
Os chunks recuperados são inseridos no prompt e o modelo gera a resposta.
import anthropic
def generate_response(query: str, context_docs: list[dict]) -> str:
client = anthropic.Anthropic()
# Construção do contexto de forma estruturada
context = "\n\n---\n\n".join([
f"[Fonte: {doc['source']} | Relevância: {doc['score']:.3f}]\n{doc['text']}"
for doc in context_docs
])
message = client.messages.create(
model="claude-opus-4-6",
max_tokens=2048,
messages=[
{
"role": "user",
"content": f"""Você é um assistente especializado. Use APENAS as informações fornecidas no contexto para responder à pergunta.
Se a informação não estiver no contexto, diga explicitamente "Não encontrei essa informação nos documentos disponíveis."
<contexto>
{context}
</contexto>
<pergunta>
{query}
</pergunta>
Responda de forma precisa e cite as fontes quando relevante."""
}
]
)
return message.content[0].text
Este é o pipeline mínimo. Funciona em demo. Em produção, cada uma dessas etapas tem profundidade que vamos explorar.
Se você pudesse otimizar apenas uma coisa em um sistema RAG, seria o chunking.
A razão é simples: o retriever só pode recuperar o que foi indexado. Se os chunks não capturarem as unidades de significado corretas, o melhor modelo de embedding do mundo não vai recuperar o contexto relevante.
Um chunk é um pedaço de texto que será indexado como unidade. Quando o retriever retorna resultados, retorna chunks inteiros.
O tamanho do chunk é um trade-off fundamental:
| Chunks pequenos | Chunks grandes |
|---|---|
| Alta precisão de recuperação | Baixa precisão de recuperação |
| Mais contexto perdido por chunk | Mais contexto por chunk |
| Mais pontos para indexar | Menos pontos para indexar |
| Menor custo de embedding | Maior custo de embedding |
| Pergunta pode não ser respondida por um único chunk | Pergunta geralmente coberta por um chunk |
Não existe tamanho "correto". Existe o tamanho correto para o tipo de pergunta e o tipo de documento.
Documentação técnica com explicações longas → chunks maiores (512-1024 tokens) FAQ com perguntas e respostas diretas → chunks menores (100-256 tokens) Emails ou mensagens → chunks por email/mensagem completo Código → chunks por função ou classe
A abordagem mais simples. Divide o texto em chunks de N tokens, com overlap M.
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512, # tokens por chunk
chunk_overlap=64, # overlap entre chunks adjacentes
length_function=len,
separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_text(documento)
O RecursiveCharacterTextSplitter do LangChain é inteligente: tenta dividir em separadores "naturais" (parágrafos, depois frases, depois espaços) antes de cortar no meio.
Vantagens: simples, previsível, funciona razoavelmente bem como baseline.
Desvantagens: ignora estrutura semântica. Uma ideia que ocupa dois parágrafos pode ser cortada ao meio. Uma frase pode ser separada do parágrafo que a contextualiza.
O overlap mitiga parcialmente o problema — os últimos M tokens do chunk anterior aparecem no início do próximo, reduzindo a chance de cortar contexto crítico. Mas é um paliativo, não uma solução.
Em vez de dividir por tamanho, divide por similaridade semântica. A ideia: calcular embeddings para cada sentença, medir a distância semântica entre sentenças consecutivas, e dividir quando a distância supera um threshold.
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
splitter = SemanticChunker(
OpenAIEmbeddings(),
breakpoint_threshold_type="percentile", # ou "standard_deviation", "interquartile"
breakpoint_threshold_amount=95, # percentil para considerar quebra
)
chunks = splitter.create_documents([documento])
Vantagens: chunks têm coerência semântica real. Uma explicação de um conceito tende a ficar no mesmo chunk.
Desvantagens: mais caro (precisa gerar embedding para cada sentença durante indexação). Resultado é menos previsível em tamanho. Pode criar chunks muito longos para tópicos que se estendem por muitos parágrafos.
Usa a estrutura do documento para guiar o chunking. Para Markdown e HTML, headers definem seções naturais. Para código, funções e classes são unidades naturais.
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers = [
("#", "Header 1"),
("##", "Header 2"),
("###", "Header 3"),
]
md_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers,
strip_headers=False
)
# Cada chunk mantém contexto de qual seção pertence
chunks = md_splitter.split_text(markdown_doc)
# chunk.metadata = {"Header 1": "RAG", "Header 2": "Chunking"}
Este approach é poderoso porque o chunk herda metadados de contexto. Um chunk que vem de H1: RAG > H2: Chunking > H3: Fixed-size já carrega essa hierarquia como filtro para retrieval.
Para código, o ast module do Python permite chunking por função/classe:
import ast
def chunk_python_code(source_code: str) -> list[dict]:
tree = ast.parse(source_code)
chunks = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
start_line = node.lineno - 1
end_line = node.end_lineno
chunk_text = "\n".join(source_code.split("\n")[start_line:end_line])
chunks.append({
"text": chunk_text,
"type": type(node).__name__,
"name": node.name,
"start_line": start_line + 1,
"end_line": end_line,
})
return chunks
Técnica proposta no paper "Dense X Retrieval: What Retrieval Granularity Should We Use?" (Chen et al., 2023). Em vez de dividir por tamanho ou estrutura, você usa um LLM para extrair proposições atômicas do texto — afirmações self-contained que capturam uma ideia completa.
# Exemplo conceitual de proposition chunking
def extract_propositions(text: str) -> list[str]:
response = client.messages.create(
model="claude-haiku-4-5-20251001", # haiku para tarefa mecânica
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""Extraia proposições atômicas e auto-contidas do texto abaixo.
Cada proposição deve ser uma afirmação completa que pode ser entendida sem contexto adicional.
Texto:
{text}
Retorne uma proposição por linha."""
}]
)
return response.content[0].text.strip().split("\n")
O custo é alto (uma chamada LLM por chunk durante indexação), mas a qualidade de recuperação aumenta significativamente. O paper reporta melhoras de 10-30% em benchmarks de QA.
Uma das técnicas mais eficazes em produção. A ideia: indexar chunks pequenos para precisão de recuperação, mas retornar chunks grandes para dar contexto ao modelo.
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter
# Splitter para chunks pequenos (para indexação/busca)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
# Splitter para chunks grandes (para contexto de geração)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
# Store para documentos pai
store = InMemoryStore()
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=child_splitter,
parent_splitter=parent_splitter,
)
# Indexação: cria chunks pequenos linkados a chunks grandes
retriever.add_documents(docs)
# Retrieval: busca pelo chunk pequeno, retorna o chunk pai
results = retriever.get_relevant_documents(query)
O resultado prático: a busca é precisa (chunks pequenos capturam intenção específica) mas o contexto entregue ao modelo é rico (chunks grandes têm mais informação). Esta é geralmente a abordagem com melhor relação custo-benefício depois que você superou o baseline.
Uma heurística útil vem do LlamaIndex: pense em dois tipos de contexto separadamente:
Isso leva ao conceito de contextual retrieval (Anthropic, 2024): ao indexar, você gera um snippet de contexto de 2-3 frases que posiciona o chunk dentro do documento maior, e o indexa junto com o chunk. Isso melhora precisão de busca sem aumentar o tamanho do chunk de geração.
def add_context_to_chunk(chunk: str, document: str) -> str:
"""Usa LLM barato para gerar contexto posicional para o chunk"""
context = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[{
"role": "user",
"content": f"""<document>
{document}
</document>
<chunk>
{chunk}
</chunk>
Escreva 2-3 frases situando este chunk no documento completo.
Responda APENAS com o contexto, sem explicações."""
}]
).content[0].text
return f"{context}\n\n{chunk}"
No benchmark da Anthropic, contextual retrieval reduziu falhas de recuperação em 49% combinado com BM25 híbrido.
Um modelo de embedding transforma texto em um vetor de números reais em espaço de alta dimensão. A propriedade fundamental: textos semanticamente similares ficam próximos no espaço vetorial, medido por cosine similarity ou dot product.
Modelos de embedding são transformers treinados com objetivos específicos de similaridade. Os mais comuns usam:
Contrastive learning (SBERT, E5): pares de frases semanticamente similares são aproximadas, pares dissimilares são afastadas no espaço vetorial.
Masked Language Modeling + fine-tuning: modelos como BERT são pré-treinados em MLM e depois fine-tunados em tarefas de similaridade semântica ou pares de perguntas/documentos.
A dimensão do vetor varia:
Mais dimensões ≠ melhor qualidade necessariamente. Mas geralmente modelos maiores têm qualidade maior.
O MTEB (Massive Text Embedding Benchmark) é o benchmark mais abrangente para modelos de embedding. Em abril de 2026, os líderes para retrieval em inglês são:
| Modelo | Dimensão | Domínio | Nota MTEB (Retrieval) | Custo |
|---|---|---|---|---|
| text-embedding-3-large (OpenAI) | 3072 | Geral | ~56 | API paga |
| voyage-3-large (Voyage AI) | 1024 | Geral | ~57 | API paga |
| BAAI/bge-m3 | 1024 | Multi-língua | ~54 | Open source |
| intfloat/multilingual-e5-large | 1024 | Multi-língua | ~52 | Open source |
| BAAI/bge-large-en-v1.5 | 1024 | EN | ~54 | Open source |
| sentence-transformers/all-MiniLM-L6-v2 | 384 | Geral | ~41 | Open source |
Para português, os melhores modelos bilíngues/multilíngues são:
BAAI/bge-m3 — excelente multilíngue, forte em PT-BRintfloat/multilingual-e5-large — sólido para PT-BRrufimelo/bert-large-portuguese-cased-sts — especializado em PT, mas menor base de treinoUma propriedade importante frequentemente ignorada: para retrieval, queries e documentos têm naturezas diferentes. Uma pergunta ("O que é RAG?") tem estrutura diferente de uma afirmação ("RAG é uma arquitetura que combina...").
Modelos de embedding modernos são treinados com instruções para separar esses casos:
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-en-v1.5")
# Para indexação de documentos
doc_embedding = model.encode(
"Represent this document for retrieval: " + texto_documento
)
# Para queries de usuário
query_embedding = model.encode(
"Represent this sentence for retrieval: " + pergunta_usuario
)
O E5 e BGE são treinados explicitamente para isso. Usar instruções corretas pode melhorar retrieval accuracy em 5-10%.
| Critério | Local (open source) | API |
|---|---|---|
| Custo por token | Zero após hardware | $0.00002-0.00013 por 1K tokens |
| Latência | Depende do hardware | 50-200ms + rede |
| Privacidade | Total | Dados saem da infraestrutura |
| Manutenção | Você opera | Gerenciado |
| Qualidade | Modelos grandes são competitivos | Ligeiramente melhor no topo |
| Dimensão | Fixa | Configurável (OpenAI) |
Para dados sensíveis (documentos médicos, jurídicos, financeiros), modelos locais são frequentemente mandatórios. Para a maioria dos casos de uso de produto, APIs são mais simples de operar.
Um problema prático subestimado: se você muda o modelo de embedding, todos os vetores existentes ficam incompatíveis. Você precisa re-indexar toda a base.
Isso tem implicações para o design do sistema:
Para bases grandes (milhões de documentos), re-indexação pode ser cara. Alguns times mantêm versões paralelas da index durante migração.
Um vector database é uma infraestrutura projetada para armazenar, indexar e buscar vetores de alta dimensão com performance adequada a produção.
Busca exata de vizinhos mais próximos (exact nearest neighbor, KNN) é intratável em alta dimensão — em uma base com 10 milhões de vetores de 1024 dimensões, comparar o query vector contra todos os pontos seria proibitivo.
Vector databases usam Approximate Nearest Neighbor (ANN) algorithms — tradeoff entre velocidade e recall perfeito:
HNSW (Hierarchical Navigable Small World): O algoritmo mais usado em produção. Constrói um grafo hierárquico onde cada nó conecta a seus vizinhos mais próximos em diferentes escalas. Busca navega pelo grafo do nível mais alto (poucos nós, vizinhos distantes) até o mais baixo (muitos nós, vizinhos próximos).
Parâmetros críticos do HNSW:
ef_construction: qualidade do grafo na indexação (maior = melhor qualidade, mais lento para indexar)M: número de conexões por nó (maior = mais memória, geralmente melhor qualidade)ef_search: tamanho do espaço de busca em tempo de query (maior = mais preciso, mais lento)IVF (Inverted File Index): Divide o espaço vetorial em clusters (via k-means). Na busca, apenas os clusters mais próximos são pesquisados. Usado pelo Faiss; bom para bases muito grandes.
ScaNN (Google): Algoritmo de busca vetorial com anisotropic quantization. Excelente tradeoff qualidade/velocidade, mas mais complexo de operar.
| Feature | Pinecone | Weaviate | Qdrant | Chroma | pgvector |
|---|---|---|---|---|---|
| Tipo | Managed SaaS | Open + Cloud | Open + Cloud | Open source | PostgreSQL ext |
| Melhor para | Escala, simplicidade | Schema rico, multi-modal | Performance, self-hosted | Prototipagem | Stack Postgres |
| Metafiltering | Bom | Excelente | Excelente | Limitado | SQL completo |
| Multi-tenancy | Namespaces | Nativo | Collections | Collections | Schemas |
| Escalabilidade | Auto | Manual/Kubernetes | Manual/K8s | Limitada | PostgreSQL |
| Hybrid search | Sim | Sim (BM25 nativo) | Sim | Não nativo | pgvector + FTS |
| Custo (self-hosted) | N/A | Gratuito | Gratuito | Gratuito | Gratuito |
| Custo (managed) | $70+/mês | $25+/mês | $25+/mês | N/A | Aurora, Neon |
Pinecone é a escolha pragmática para equipes que não querem operar infraestrutura. Serverless tier com 1M vetores grátis, escala automática. A desvantagem é lock-in e custo em escala.
Qdrant é a escolha para self-hosted de alta performance. Escrito em Rust, excelente performance, payloads JSON ricos, filtros eficientes. Se você vai rodar em Kubernetes, Qdrant é forte.
Weaviate tem a melhor integração de hybrid search nativa e schema rico. Boa para casos onde você precisa de queries complexas combinando semântica e estrutura.
pgvector é a escolha para times com investimento em Postgres que querem minimizar operações. Menos performático que databases especializados em alta escala, mas SQL completo e backups/monitoring familiares.
Chroma é excelente para protótipos mas tem limitações em produção — sem suporte robusto a clustering, multi-tenancy limitado, e performance deteriora com bases maiores.
Um ponto crítico: o retriever precisa filtrar por metadados para ser útil em aplicações reais.
Exemplos de filtros essenciais:
tenant_id: multi-tenancy (usuário A só vê documentos do usuário A)document_type: buscar só em contratos, não em emailscreated_after: informação mais recente que data Xlanguage: documentos em português apenasdepartment: documentos do departamento financeiro# Filtro em Qdrant
from qdrant_client.models import Filter, FieldCondition, MatchValue, Range
results = client.search(
collection_name="knowledge_base",
query_vector=query_embedding,
query_filter=Filter(
must=[
FieldCondition(key="tenant_id", match=MatchValue(value="empresa-abc")),
FieldCondition(key="document_type", match=MatchValue(value="contrato")),
],
should=[
FieldCondition(key="language", match=MatchValue(value="pt")),
]
),
limit=10,
)
A eficiência da filtragem varia muito entre databases. Alguns aplicam filtros pós-busca (busca k*10 e depois filtra), outros integram filtros no grafo HNSW (mais eficiente). Para bases com filtros seletivos (tenant isolation), a diferença é significativa.
A busca por vizinhos mais próximos por similaridade cosine é o baseline. É boa. Não é suficiente.
O que discutimos até agora. Transforma query em embedding, busca vetores similares. Captura sinônimos, paráfrases, e conceitos relacionados que não compartilham vocabulário com a query.
Ponto forte: "O que é processamento de linguagem natural?" encontra documentos que falam em NLP, PLN, text mining — mesmo sem usar essas palavras exatas.
Ponto fraco: poor recall para termos específicos, nomes próprios, números. Uma busca por "GPT-4" pode não encontrar documentos que literalmente mencionam "GPT-4" se o modelo de embedding generaliza demais.
BM25 (Best Match 25) é o algoritmo padrão para busca por keyword. É o backend de buscas como Elasticsearch e OpenSearch. Calcula um score baseado em frequência do termo na query (TF), frequência inversa no corpus (IDF), e tamanho do documento.
Ponto forte: precisão alta para termos exatos, nomes próprios, identificadores técnicos. "CVE-2024-1234" vai ser encontrado exatamente.
Ponto fraco: não entende semântica. "NLP" e "processamento de linguagem natural" são termos completamente diferentes para BM25.
Combina scores de busca semântica e keyword usando Reciprocal Rank Fusion (RRF) ou interpolação linear.
RRF é mais robusto porque normaliza os ranks e não os scores (que têm escalas diferentes):
def reciprocal_rank_fusion(
results_list: list[list[dict]],
k: int = 60
) -> list[dict]:
"""
Combina múltiplas listas de resultados rankeados usando RRF.
k=60 é o valor default do paper original de Cormack et al., 2009
"""
scores = {}
for results in results_list:
for rank, result in enumerate(results):
doc_id = result["id"]
if doc_id not in scores:
scores[doc_id] = {"score": 0, "doc": result}
# RRF score: 1 / (k + rank)
scores[doc_id]["score"] += 1.0 / (k + rank + 1)
# Ordena por score RRF descendente
sorted_results = sorted(
scores.values(),
key=lambda x: x["score"],
reverse=True
)
return [r["doc"] for r in sorted_results]
# Uso
semantic_results = semantic_search(query, k=20)
keyword_results = bm25_search(query, k=20)
final_results = reciprocal_rank_fusion([semantic_results, keyword_results])[:10]
Dados empíricos: O paper "Bridging the Gap Between Sparse and Dense Retrieval" (2021) mostra que hybrid search geralmente supera tanto BM25 quanto dense retrieval isoladamente em 5-20% em benchmarks de QA.
Weaviate tem hybrid search nativo. Para outros databases, você precisa rodar BM25 separadamente (Elasticsearch/OpenSearch/BM25 em memória com rank_bm25) e combinar.
Uma técnica simples mas eficaz: gerar múltiplas versões da query antes de buscar.
def generate_query_variants(query: str) -> list[str]:
"""Gera variantes da query para aumentar recall"""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=512,
messages=[{
"role": "user",
"content": f"""Gere 3 variações da seguinte pergunta, mantendo o significado mas usando formulações diferentes.
Uma variação mais específica, uma mais geral, e uma reformulada.
Pergunta original: {query}
Retorne apenas as 3 perguntas, uma por linha, sem numeração."""
}]
).content[0].text
variants = [query] + response.strip().split("\n")
return variants[:4] # original + 3 variações
# Busca com todas as variantes, combina resultados
all_results = []
for variant in generate_query_variants(query):
results = semantic_search(variant, k=5)
all_results.append(results)
final_results = reciprocal_rank_fusion(all_results)
O ganho de recall é real — especialmente para queries ambíguas ou mal formuladas.
Um problema clássico: os top-k resultados por similaridade podem ser muito similares entre si. Se o chunk mais relevante fala sobre "implementação de autenticação JWT", os próximos 4 podem falar sobre o mesmo tópico com palavras ligeiramente diferentes — mas o que o usuário precisa pode estar no 8º resultado sobre um aspecto diferente.
MMR resolve esse problema balanceando relevância com diversidade:
import numpy as np
def mmr_rerank(
query_embedding: np.ndarray,
candidate_embeddings: list[np.ndarray],
candidates: list[dict],
k: int = 5,
lambda_param: float = 0.5,
) -> list[dict]:
"""
Maximal Marginal Relevance.
lambda=1: só relevância. lambda=0: só diversidade.
"""
selected = []
remaining = list(range(len(candidates)))
while len(selected) < k and remaining:
# Relevância: similaridade com query
relevance_scores = {
i: np.dot(query_embedding, candidate_embeddings[i])
for i in remaining
}
if not selected:
# Primeiro resultado: puramente por relevância
best = max(remaining, key=lambda i: relevance_scores[i])
else:
# Próximos: balanço entre relevância e diversidade
selected_embeddings = [candidate_embeddings[i] for i in selected]
def mmr_score(i):
rel = relevance_scores[i]
# Máxima similaridade com documentos já selecionados
max_sim = max(
np.dot(candidate_embeddings[i], sel_emb)
for sel_emb in selected_embeddings
)
return lambda_param * rel - (1 - lambda_param) * max_sim
best = max(remaining, key=mmr_score)
selected.append(best)
remaining.remove(best)
return [candidates[i] for i in selected]
MMR é especialmente valioso quando o documento fonte tem seções repetitivas ou quando a base tem documentos muito similares entre si.
Query expansion adiciona termos relacionados à query antes de buscar. Abordagem clássica de IR.
HyDE (Hypothetical Document Embeddings) é mais elegante: em vez de buscar com a pergunta diretamente, você usa um LLM para gerar uma resposta hipotética e busca por documentos similares a essa resposta. A intuição: embeddings de respostas tendem a ser mais similares a embeddings de documentos do que embeddings de perguntas.
def hyde_retrieve(query: str, k: int = 5) -> list[dict]:
# Gera resposta hipotética
hypothetical = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=512,
messages=[{
"role": "user",
"content": f"""Escreva um parágrafo que seria uma boa resposta para a pergunta abaixo.
Seja específico e técnico. Você pode inventar detalhes plausíveis.
Pergunta: {query}
Parágrafo de resposta:"""
}]
).content[0].text
# Gera embedding da resposta hipotética (não da pergunta)
hyp_embedding = model.encode([hypothetical], normalize_embeddings=True)[0]
# Busca por documentos similares à resposta hipotética
return semantic_search_by_embedding(hyp_embedding, k=k)
O paper original (Gao et al., 2022) mostra ganhos de 10-20% em benchmarks de retrieval. O tradeoff é latência adicional e custo de uma chamada LLM extra.
Retrieval eficiente requer algoritmos rápidos — vector search com HNSW é O(log n). Mas modelos de relevância mais precisos (cross-encoders) são O(n) na quantidade de documentos comparados.
A solução padrão em IR é um pipeline de dois estágios:
Modelos de embedding são bi-encoders: codificam query e documento separadamente e comparam os vetores. Rápido, mas a comparação é em espaço de embedding — não vê a relação direta entre query e documento.
Cross-encoders processam query e documento juntos, permitindo interação cruzada entre tokens de ambos. Muito mais preciso, muito mais lento.
from sentence_transformers import CrossEncoder
# Modelo especializado em reranking
cross_encoder = CrossEncoder("BAAI/bge-reranker-large")
def rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
# Prepara pares (query, documento) para o cross-encoder
pairs = [(query, doc["text"]) for doc in candidates]
# Score de relevância para cada par
scores = cross_encoder.predict(pairs)
# Ordena por score e retorna top-k
ranked = sorted(
zip(scores, candidates),
key=lambda x: x[0],
reverse=True
)
return [doc for _, doc in ranked[:top_k]]
# Pipeline completo
candidates = semantic_search(query, k=50) # Busca ampla
reranked = rerank(query, candidates, top_k=10) # Refina
Modelos de reranking recomendados:
BAAI/bge-reranker-large: open source, excelente qualidadecross-encoder/ms-marco-MiniLM-L-12-v2: mais rápido, menor qualidadePara casos onde precisão é crítica e latência/custo não são a maior preocupação, LLMs podem fazer reranking com qualidade superior.
def llm_rerank(query: str, candidates: list[dict], top_k: int = 5) -> list[dict]:
# Formata candidatos para o LLM
formatted = "\n\n".join([
f"[Documento {i+1}]\n{doc['text']}"
for i, doc in enumerate(candidates)
])
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=256,
messages=[{
"role": "user",
"content": f"""Pergunta: {query}
Abaixo estão {len(candidates)} documentos. Identifique os {top_k} mais relevantes para a pergunta.
Retorne apenas os números, separados por vírgula, do mais relevante para o menos relevante.
{formatted}
Documentos mais relevantes (por número):"""
}]
).content[0].text
# Parse dos números
indices = [int(x.strip()) - 1 for x in response.split(",")][:top_k]
return [candidates[i] for i in indices if i < len(candidates)]
A solução managed mais adotada em produção:
import cohere
co = cohere.Client(api_key="...")
def cohere_rerank(query: str, documents: list[str], top_n: int = 5):
results = co.rerank(
query=query,
documents=documents,
top_n=top_n,
model="rerank-multilingual-v3.0" # Suporte a PT-BR
)
return results.results # Ordenados por relevance_score
Você tem uma query, tem os documentos mais relevantes. Agora precisa gerar uma resposta boa. Essa fase tem mais nuances do que parece.
A forma como você organiza o contexto afeta dramaticamente a qualidade da resposta. Alguns princípios baseados em pesquisa e prática:
1. Posição importa: o problema da "perda no meio" (Lost in the Middle)
O paper "Lost in the Middle: How Language Models Use Long Contexts" (Liu et al., 2023) demonstrou experimentalmente que LLMs têm performance degradada para informações no meio de contextos longos. Performance é melhor no início e no fim.
Implicação prática: coloque os documentos mais relevantes primeiro e/ou último no prompt. O documento mais relevante deve estar na posição 1.
def build_prompt(query: str, docs: list[dict]) -> str:
# Docs já vêm reranked, mais relevante primeiro
context_parts = []
for i, doc in enumerate(docs):
context_parts.append(
f"[Documento {i+1} | Fonte: {doc['source']} | Relevância: {doc['score']:.3f}]\n"
f"{doc['text']}"
)
context = "\n\n---\n\n".join(context_parts)
return f"""Você é um assistente especializado. Responda à pergunta usando APENAS as informações dos documentos fornecidos.
INSTRUÇÕES:
- Se a informação estiver nos documentos, responda com precisão e cite a fonte ([Documento N])
- Se a informação NÃO estiver nos documentos, diga: "Não encontrei essa informação nos documentos disponíveis"
- Nunca invente ou suponha informações além do que está nos documentos
- Se houver informações contraditórias entre documentos, mencione a contradição
<documentos>
{context}
</documentos>
<pergunta>
{query}
</pergunta>
Resposta:"""
2. Instruções de grounding claras
LLMs tendem a alucinam mesmo quando têm contexto. Instruções explícitas sobre o que fazer quando a informação não está presente são essenciais.
3. Citações estruturadas
Peça ao modelo para citar explicitamente. Isso:
4. Chain of thought para perguntas complexas
Para perguntas que requerem síntese de múltiplos documentos:
system_prompt = """Você é um analista especializado. Para responder perguntas:
1. Liste as evidências relevantes de cada documento
2. Identifique pontos de convergência e contradição
3. Sintetize uma resposta baseada nas evidências
4. Cite os documentos usados
Nunca responda sem evidência explícita dos documentos."""
LLMs têm limites de contexto. Com muitos documentos longos, você pode exceder esses limites ou degradar qualidade.
Estratégias:
def fit_context_to_window(
docs: list[dict],
max_tokens: int = 3000,
tokenizer=None
) -> list[dict]:
"""Trunca e seleciona documentos para caber no budget de contexto"""
selected = []
current_tokens = 0
for doc in docs:
doc_tokens = len(tokenizer.encode(doc["text"]))
if current_tokens + doc_tokens <= max_tokens:
selected.append(doc)
current_tokens += doc_tokens
else:
# Tenta incluir versão truncada se houver espaço
remaining = max_tokens - current_tokens
if remaining > 100: # Mínimo de tokens para ser útil
truncated_text = tokenizer.decode(
tokenizer.encode(doc["text"])[:remaining]
)
selected.append({**doc, "text": truncated_text + "...[truncado]"})
break
return selected
Alucinação em RAG é diferente de alucinação em modelos puros. Em modelos puros, o modelo inventa informação porque não tem contexto. Em RAG, o modelo pode:
Técnicas para reduzir:
A literatura identifica uma progressão de sofisticação em sistemas RAG.
O pipeline que descrevemos até agora: indexa, busca, gera. Um ciclo único, sem iteração.
Query → Embedding → Vector Search → Context → LLM → Response
Funciona para casos simples. Falha em:
Adiciona técnicas de pré-retrieval e pós-retrieval para melhorar qualidade:
Pré-retrieval:
Pós-retrieval:
Query
↓ Query Rewriting
↓ Multi-query Generation
Retrieval (múltiplas queries)
↓ Fusion (RRF)
Re-ranking
↓ Context Compression
LLM → Response
Trata cada componente como um módulo trocável. A arquitetura permite:
class ModularRAGPipeline:
def __init__(self):
self.retrievers = {
"semantic": SemanticRetriever(),
"keyword": BM25Retriever(),
"sql": SQLRetriever(), # Para dados estruturados
"graph": GraphRetriever(), # Para dados relacionais
}
self.reranker = CrossEncoderReranker()
self.compressor = ContextCompressor()
self.router = QueryRouter()
def retrieve(self, query: str) -> list[dict]:
# Router decide qual(is) retriever(s) usar
strategy = self.router.route(query)
results = []
for retriever_name in strategy.retrievers:
retriever = self.retrievers[retriever_name]
results.append(retriever.retrieve(query, k=strategy.k_per_retriever))
# Funde resultados
fused = reciprocal_rank_fusion(results)
# Rerank e comprime
reranked = self.reranker.rerank(query, fused)
compressed = self.compressor.compress(query, reranked)
return compressed
Modular RAG é o padrão para sistemas de produção maduros. LlamaIndex implementa este paradigma com sua abstrações de QueryEngine e RouterQueryEngine.
A fronteira atual. O sistema não faz um único ciclo de retrieval-generation, mas usa um agente com ferramentas para iterar.
O agente pode:
# Exemplo com framework de agentes
tools = [
Tool(
name="search_documents",
description="Busca na base de conhecimento. Use quando precisar de informação específica.",
func=search_documents
),
Tool(
name="search_web",
description="Busca na web para informações recentes não na base.",
func=search_web
),
Tool(
name="sql_query",
description="Executa queries em dados estruturados.",
func=execute_sql
),
Tool(
name="calculator",
description="Para cálculos numéricos.",
func=calculate
),
]
agent = initialize_agent(
tools=tools,
llm=claude_llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
response = agent.run(complex_query)
Agentic RAG é poderoso mas tem riscos:
A pesquisa em RAG avançou rapidamente. Aqui estão as técnicas mais relevantes além do pipeline básico.
Paper: "Corrective Retrieval Augmented Generation" (Yan et al., 2024)
CRAG adiciona um evaluator que avalia a qualidade dos documentos recuperados e decide o que fazer:
class CRAGPipeline:
def __init__(self, retriever, web_searcher, evaluator_model):
self.retriever = retriever
self.web_searcher = web_searcher
self.evaluator = evaluator_model
def retrieve_and_correct(self, query: str) -> list[dict]:
# Retrieval inicial
docs = self.retriever.retrieve(query, k=5)
# Avaliação da relevância
relevance_score = self.evaluator.evaluate(query, docs)
if relevance_score >= 0.7:
# Alta relevância: usa documentos
return docs
elif relevance_score >= 0.3:
# Relevância média: augmenta com web
web_docs = self.web_searcher.search(query, k=3)
return docs + web_docs
else:
# Baixa relevância: usa só web
return self.web_searcher.search(query, k=5)
CRAG é especialmente útil quando a base de conhecimento não cobre todos os tópicos.
Paper: "Self-RAG: Learning to Retrieve, Generate, and Critique" (Asai et al., 2023)
Self-RAG treina o LLM para:
O modelo usa tokens especiais de controle: [Retrieve], [IsREL], [IsSUP], [IsUse].
Na prática, Self-RAG requer fine-tuning do modelo base, o que aumenta o custo de adoção. Mas o conceito de auto-avaliação é amplamente aplicável mesmo sem fine-tuning.
Microsoft Research publicou o GraphRAG em 2024, com o paper "From Local to Global: A Graph RAG Approach to Query-Focused Summarization".
A ideia: para perguntas que requerem síntese de informação dispersa em muitos documentos (perguntas "globais"), busca semântica tradicional falha — não existe um único chunk que contém a resposta, a resposta emerge da conexão entre múltiplos chunks.
GraphRAG constrói um knowledge graph a partir dos documentos:
# Conceitual — implementação completa usa Microsoft GraphRAG library
from graphrag import GraphRAGPipeline
pipeline = GraphRAGPipeline(
llm=claude_client,
embedding_model=embedding_model,
)
# Fase de indexação - constrói knowledge graph
pipeline.index(documents)
# Query local: usa subgrafo relevante
local_response = pipeline.query(
"Quais são os requisitos de autenticação?",
mode="local"
)
# Query global: usa sumários de comunidades
global_response = pipeline.query(
"Quais são os temas principais da documentação?",
mode="global"
)
GraphRAG tem custo de indexação significativamente maior que RAG tradicional (muitas chamadas LLM para extrair entidades e relações), mas performa muito melhor em perguntas que requerem síntese global.
Com modelos multimodais, RAG pode indexar e recuperar não apenas texto, mas imagens, tabelas, diagramas, e vídeos.
Abordagens:
Para documentos técnicos com muitos diagramas e figuras, multimodal RAG pode capturar informação que seria perdida em indexação texto-only.
Combina RAG com query expansion e fusion de múltiplos rankings:
def rag_fusion(query: str, k: int = 5) -> list[dict]:
# Gera múltiplas sub-queries
sub_queries = generate_sub_queries(query, n=4)
# Retrieve para cada sub-query
all_results = [retrieve(q, k=10) for q in [query] + sub_queries]
# Funde com RRF
fused = reciprocal_rank_fusion(all_results)
return fused[:k]
Saber que seu sistema RAG funciona em demo é diferente de saber que funciona em produção. Avaliação rigorosa é o que separa sistemas que melhoram iterativamente de sistemas que degradam silenciosamente.
RAGAS (RAG Assessment) é o framework open source mais adotado para avaliação de sistemas RAG. Avalia quatro dimensões:
1. Faithfulness (Fidelidade) Mede se a resposta gerada é suportada pelo contexto recuperado. Uma resposta fiel não contém afirmações que contradizem ou vão além dos documentos fornecidos.
Faithfulness = (afirmações na resposta suportadas pelo contexto) / (total de afirmações na resposta)
2. Answer Relevance (Relevância da Resposta) Mede se a resposta é relevante para a pergunta. Uma resposta irrelevante ou incompleta tem baixo answer relevance.
Técnica: gera perguntas hipotéticas a partir da resposta e mede similaridade com a pergunta original.
3. Context Recall (Recall do Contexto) Mede se o contexto recuperado contém toda a informação necessária para responder a pergunta.
Requer ground truth (resposta correta de referência).
4. Context Precision (Precisão do Contexto) Mede se o contexto recuperado é relevante — penaliza documentos recuperados que não contribuem para a resposta.
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_recall,
context_precision,
)
from datasets import Dataset
# Cria dataset de avaliação
eval_data = {
"question": ["O que é RAG?", "Como funciona chunking?"],
"answer": [generated_answers[0], generated_answers[1]],
"contexts": [retrieved_contexts[0], retrieved_contexts[1]],
"ground_truth": [reference_answers[0], reference_answers[1]],
}
dataset = Dataset.from_dict(eval_data)
result = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
)
print(result)
# Output: {"faithfulness": 0.85, "answer_relevancy": 0.92, ...}
Além do RAGAS, métricas customizadas para seu domínio:
Hallucination rate: Percentual de respostas com informação inventada
def check_hallucination(response: str, context: str) -> bool:
"""Usa LLM como juiz para verificar alucinação"""
verdict = judge_model.evaluate(
f"A resposta abaixo contém informação não presente no contexto?\n\nContexto: {context}\n\nResposta: {response}\n\nResponda: sim ou não"
)
return "sim" in verdict.lower()
Citation accuracy: Quando o modelo cita fontes, quão acuradas são as citações?
Answer completeness: A resposta cobre todos os aspectos da pergunta?
Retrieval hit rate: Quão frequentemente o documento correto está nos top-k resultados?
Avaliação sem um test set de qualidade é avaliação sem sentido. Como construir:
Golden dataset manual: pares pergunta-resposta criados por especialistas do domínio. Mais trabalhoso, mais confiável.
LLM-generated synthetic data: use um LLM para gerar perguntas a partir dos documentos existentes.
def generate_eval_questions(document: str, n: int = 5) -> list[dict]:
"""Gera pares pergunta-resposta para avaliação"""
response = client.messages.create(
model="claude-opus-4-6", # Opus para maior qualidade na geração
max_tokens=2048,
messages=[{
"role": "user",
"content": f"""A partir do documento abaixo, gere {n} perguntas de avaliação.
Para cada pergunta, inclua a resposta correta baseada no documento.
Inclua variedade:
- Perguntas factuais diretas
- Perguntas que requerem síntese de múltiplas partes
- Perguntas sobre detalhes técnicos
- Perguntas que testam compreensão de conceitos
Documento:
{document}
Retorne em JSON:
[{{"pergunta": "...", "resposta": "...", "tipo": "..."}}]"""
}]
).content[0].text
return json.loads(response)
Avaliação não é um evento único — é um processo contínuo.
Implementações práticas:
Esta é a pergunta que mais gera confusão. A resposta curta: eles resolvem problemas diferentes e frequentemente se complementam.
| Situação | RAG | Fine-tuning |
|---|---|---|
| Conhecimento privado/proprietário | ✅ | ❌ (risco de memorizar dados sensíveis) |
| Conhecimento que muda frequentemente | ✅ | ❌ (re-treino frequente é caro) |
| Resposta deve citar fontes | ✅ | ❌ (modelos não citam o treino) |
| Comportamento específico desejado | ❌ | ✅ |
| Formato de saída customizado | ❌ | ✅ |
| Redução de custo por token | ❌ | ✅ (modelos menores) |
| Conhecimento denso que cabe no treino | ❌ | ✅ |
| Base de conhecimento pequena (<100 docs) | Pode ser overkill | ✅ |
A combinação mais poderosa: fine-tuning para comportamento + RAG para conhecimento.
Fine-tune o modelo para:
Adicione RAG para:
Não as do pitch deck. As que aparecem em produção.
1. Atualização de conhecimento sem re-treino Atualize a base de documentos e o sistema imediatamente tem acesso ao novo conhecimento. Crítico para domínios que mudam — legislação, documentação de produtos, políticas internas.
2. Verificabilidade e auditoria Cada resposta pode ser rastreada aos documentos fonte. Em domínios regulados (saúde, financeiro, jurídico), isso não é opcional — é requisito.
3. Redução de alucinação LLMs em modo puro alucinam com frequência em domínios específicos. Com RAG e grounding adequado, a taxa de alucinação cai significativamente.
4. Controle granular sobre o conhecimento Você decide quais documentos o sistema acessa. Pode isolar por tenant, por departamento, por nível de acesso. Fine-tuning não oferece esse nível de controle.
5. Escalabilidade do conhecimento A base de documentos pode crescer para bilhões de tokens sem afetar o modelo base. O custo escala com a base de dados e o retrieval, não com re-treino.
6. Transparência É possível logar exatamente quais documentos foram usados para cada resposta. Debugar uma resposta errada é investigar o retrieval, não uma caixa-preta.
1. Qualidade depende da base de conhecimento Lixo entra, lixo sai — mas de forma insidiosa. Documentos desatualizados, inconsistentes, ou mal estruturados degradam silenciosamente o sistema. A qualidade da curadoria da base de conhecimento é um ativo subestimado.
2. Retrieval pode falhar O sistema não funciona se o retriever não encontra os documentos certos. Perguntas mal formuladas, domínio fora da base, queries ambíguas — todos resultam em recuperação incorreta. O modelo então gera com contexto errado ou insuficiente.
3. Latência adicional Um pipeline RAG completo (embedding da query + busca + re-ranking + geração) tem latência maior que uma chamada direta ao LLM. Em aplicações interativas, isso importa.
4. Complexidade operacional Vector database para operar. Pipeline de indexação para manter. Modelos de embedding para versionar. Métricas de retrieval para monitorar. É uma stack de engenharia real, não um add-on.
5. O "lost in the middle" problem Mesmo com retrieval perfeito, se o contexto é muito longo, o modelo pode ignorar informação relevante no meio. O balance entre volume de contexto e qualidade de atenção é um problema não resolvido.
6. Context poisoning Se a base de conhecimento contém informação deliberadamente incorreta (por erro ou ataque), o modelo vai usar essa informação. RAG expande a superfície de ataque.
7. Custo de indexação Para bases grandes com documentos frequentemente atualizados, o custo de re-indexação (embedding de todos os chunks) pode ser significativo.
8. Multi-hop reasoning limitado Perguntas que requerem encadeamento de múltiplos fatos ("Quem fundou a empresa que adquiriu o produto X em 2023?") exigem múltiplos retrieval steps ou GraphRAG — o naive RAG falha consistentemente.
Estes são os problemas que aparecem repetidamente em sistemas RAG reais.
Sintomas: O modelo responde com conhecimento geral do treinamento mesmo quando os documentos contêm a resposta correta.
Causas e soluções:
Sintomas: Os top-k resultados têm baixa relevância para a query.
Diagnóstico:
# Logar scores de retrieval para análise
def retrieval_diagnostics(query: str):
results = retrieve(query, k=10)
print(f"Query: {query}")
print(f"Top-10 resultados:")
for i, r in enumerate(results):
print(f" {i+1}. Score: {r['score']:.4f} | {r['text'][:100]}...")
Causas e soluções:
Sintomas: A resposta contém informação não presente nos documentos recuperados.
Causas e soluções:
Sintomas: Sistema funciona bem com 10k documentos, degrada com 1M.
Causas e soluções:
ef_search e M para manter qualidade (mas aumenta latência)Sintomas: A mesma query retorna respostas diferentes em chamadas subsequentes.
Causas e soluções:
seed quando disponívelSintomas: Usuário A vê documentos que deveriam ser restritos ao usuário B.
Solução: Filtros de metadados no retrieval + validação no application layer.
def secure_retrieve(
query: str,
tenant_id: str,
user_permissions: list[str],
k: int = 5
) -> list[dict]:
return client.search(
collection_name="knowledge_base",
query_vector=embed(query),
query_filter=Filter(
must=[
FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id)),
FieldCondition(
key="access_level",
match=MatchAny(any=user_permissions)
),
]
),
limit=k,
)
Nunca confie apenas no prompt para segurança de acesso. Filtros devem ser aplicados no nível do vector database.
# Stack mínima com LangChain
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
vectordb = Chroma.from_documents(
documents=chunks,
embedding=OpenAIEmbeddings(model="text-embedding-3-small")
)
qa_chain = RetrievalQA.from_chain_type(
llm=ChatOpenAI(model="gpt-4o-mini"),
retriever=vectordb.as_retriever(search_kwargs={"k": 5})
)
answer = qa_chain.invoke({"query": question})
Um sistema RAG sem observabilidade é uma caixa-preta que vai falhar misteriosamente.
Métricas essenciais para monitorar:
import time
from dataclasses import dataclass
@dataclass
class RAGTrace:
query: str
retrieval_time_ms: float
generation_time_ms: float
num_docs_retrieved: int
avg_retrieval_score: float
response: str
tenant_id: str
session_id: str
def traced_rag(query: str, tenant_id: str) -> tuple[str, RAGTrace]:
start = time.time()
docs = retrieve(query, tenant_id=tenant_id)
retrieval_time = (time.time() - start) * 1000
gen_start = time.time()
response = generate(query, docs)
gen_time = (time.time() - gen_start) * 1000
trace = RAGTrace(
query=query,
retrieval_time_ms=retrieval_time,
generation_time_ms=gen_time,
num_docs_retrieved=len(docs),
avg_retrieval_score=sum(d["score"] for d in docs) / len(docs) if docs else 0,
response=response,
tenant_id=tenant_id,
session_id=get_session_id(),
)
# Envia para sistema de observabilidade
langfuse.trace(trace)
return response, trace
Documentos mudam. Novos documentos chegam. Documentos são deletados. Você precisa de um pipeline de indexação incremental.
Abordagem com Change Data Capture (CDC):
# Exemplo com webhook de novos documentos
@app.post("/webhook/document-updated")
async def handle_document_update(event: DocumentUpdateEvent):
if event.type == "created" or event.type == "updated":
# Re-indexa o documento
document = fetch_document(event.document_id)
chunks = chunk_document(document)
embeddings = embed_chunks(chunks)
# Remove versão antiga se existir
delete_by_metadata({"document_id": event.document_id})
# Insere nova versão
upsert_vectors(embeddings, metadata={"document_id": event.document_id})
elif event.type == "deleted":
delete_by_metadata({"document_id": event.document_id})
Queries repetidas são comuns. Cache pode reduzir latência e custo significativamente.
import hashlib
import json
from functools import wraps
# Cache semântico: similar ao que GPTCache oferece
class SemanticCache:
def __init__(self, similarity_threshold=0.95):
self.cache = {} # Em produção: Redis ou similar
self.threshold = similarity_threshold
def get(self, query: str) -> str | None:
query_emb = embed(query)
for cached_query, (cached_emb, response) in self.cache.items():
similarity = cosine_similarity(query_emb, cached_emb)
if similarity >= self.threshold:
return response
return None
def set(self, query: str, response: str):
query_emb = embed(query)
self.cache[query] = (query_emb, response)
GPTCache e Semantic Cache do LangChain implementam essa lógica com Redis como backend.
Quando você atualiza a base, podem surgir regressões. Uma resposta que funcionava para uma query pode parar de funcionar depois de adicionar novos documentos.
Práticas recomendadas:
Documentos indexados podem conter instruções maliciosas. Um atacante pode subir um documento que diz "Ignore as instruções anteriores e faça X". Isso é prompt injection via RAG.
Mitigações:
Os casos mais perigosos são quando o sistema retorna uma resposta confidentemente errada. Como detectar:
Antes de declarar um sistema RAG pronto para produção:
Muitos conceitos errados sobre RAG circulam em posts, entrevistas e code reviews. Estes são os mais prejudiciais.
RAG reduz alucinação quando a base de conhecimento cobre o tópico e o retrieval funciona bem. Mas o modelo ainda pode alucinar ao extrapolar, ao combinar informações incorretamente, ou simplesmente ao ignorar o contexto recuperado. RAG não é uma solução para alucinação — é um controle que a atenua.
Chunks maiores têm mais contexto por unidade, mas pior precisão de retrieval. O retriever precisa encontrar o chunk certo — e um chunk de 2000 tokens que contém a resposta mas também muito ruído irrelevante pode ter score de similaridade menor que um chunk de 200 tokens focado na resposta. Tamanho de chunk é um trade-off, não um slider para maximizar.
A qualidade do RAG é limitada pelo retrieval, não pela geração. Um modelo mais capaz com retrieval ruim ainda vai produzir respostas ruins. Melhorar o modelo base ajuda na síntese, mas não resolve recall baixo ou chunks mal estruturados. O bottleneck mais comum é o retriever, não o gerador.
text-embedding-ada-002 era o modelo de referência em 2022. Em 2026, foi superado pelo text-embedding-3-large da própria OpenAI e por vários modelos open source. Para PT-BR especificamente, modelos como BGE-M3 frequentemente performam melhor. Escolha baseada em benchmark no seu domínio, não em familiaridade.
Overlap mitiga, não resolve. Um overlap de 50 tokens em chunks de 500 tokens significa que 10% do conteúdo está duplicado. Para informação crítica que cai exatamente na fronteira, pode ajudar. Para relações semânticas entre parágrafos distantes, não adianta. Parent-document retrieval e semantic chunking são soluções mais robustas.
BM25 puro frequentemente supera vector search para queries com termos técnicos específicos, nomes próprios, e identificadores. Hybrid search geralmente supera ambos. A escolha entre denso, esparso e híbrido deve ser baseada em avaliação no seu corpus — não em suposição sobre qual é mais "moderno".
RAG funciona bem com documentos bem estruturados, limpos, e relevantes. Com documentos escaneados de má qualidade, PDFs com layouts complexos, bases com muito conteúdo desatualizado ou contraditório, o sistema pode performar pior que uma busca simples por keyword. A curadoria da base é tão importante quanto a arquitetura.
O "Lost in the Middle" problem é real. Aumentar o contexto além de um ponto reduz atenção a informações no meio. O ótimo geralmente está entre 3-10 documentos dependendo do modelo e da query. Mais não é melhor — relevância e posição importam mais que volume.
RAGAS mede faithfulness, relevância, recall e precisão em um test set sintético. Um score alto no test set não garante comportamento correto em distribuições reais de queries, especialmente para queries adversariais, fora do domínio, ou que requerem raciocínio multi-hop. Avaliação em produção com usuários reais é insubstituível.
O pipeline básico é simples. Fazer funcionar bem em produção — com retrieval confiável, chunking adequado ao domínio, re-ranking, observabilidade, segurança, indexação incremental, e avaliação contínua — é um projeto de engenharia real que pode levar semanas a meses para atingir produção confiável.
RAG se tornou uma das técnicas mais importantes na engenharia de sistemas de IA. Mas a popularidade criou um problema: muitos sistemas chamados de RAG são apenas demos com a arquitetura correta e a engenharia errada.
A diferença entre um sistema RAG que funciona em demo e um que funciona em produção está em cada uma das camadas que cobrimos: a qualidade do parsing de documentos, a estratégia de chunking, a escolha e uso correto dos modelos de embedding, a configuração do vector database, a estratégia de retrieval (frequentemente híbrida), o re-ranking, o prompt engineering, a avaliação sistemática, e a operação com observabilidade.
Nenhuma dessas camadas é difícil isoladamente. A dificuldade está em entender como elas interagem e onde estão os gargalos no seu sistema específico.
Revisando times que construíram sistemas RAG, os padrões de falha mais comuns são:
Falha 1: Otimizar a camada errada. O time passa semanas escolhendo o melhor modelo de embedding enquanto o problema real é que o chunking está cortando sentenças no meio. Meça antes de otimizar.
Falha 2: Não ter baseline. Sem uma métrica inicial (retrieval hit rate, RAGAS score, usuário satisfaction), você não sabe se está melhorando. O primeiro passo depois de ter um sistema funcionando é medir onde ele está.
Falha 3: Confundir "funciona no meu exemplo" com "funciona em produção". Um exemplo funcionando bem é um anedota, não uma avaliação. Você precisa de um test set representativo com pelo menos 50-100 exemplos antes de confiar nas métricas.
Falha 4: Ignorar a qualidade dos dados. A base de conhecimento é o ativo mais importante do sistema RAG. Um sistema tecnicamente sofisticado com dados de baixa qualidade vai performar pior que um sistema simples com dados limpos e bem estruturados.
Falha 5: Adicionar complexidade prematuramente. Multi-agent RAG, GraphRAG, Self-RAG — todas são técnicas válidas para problemas específicos. Mas a maioria dos times não precisa delas. Comece simples, meça, adicione complexidade quando as métricas mostrarem que você atingiu o teto da abordagem atual.
O caminho que funciona:
1. Defina o job-to-be-done específico
↓
2. Construa o pipeline mais simples que poderia funcionar
↓
3. Crie um test set com exemplos reais
↓
4. Meça: retrieval hit rate, RAGAS, user satisfaction
↓
5. Identifique o bottleneck (retrieval? chunking? geração?)
↓
6. Aplique a técnica mais simples que resolve o bottleneck
↓
7. Meça novamente — confirme melhora
↓
8. Repita
Esse ciclo parece óbvio. Poucos times o seguem. A maioria pula para a etapa 6 antes de fazer as etapas 3-5.
Quando RAG funciona bem, o impacto é real:
Esses resultados existem em produção hoje. Não são promessas de roadmap.
A diferença entre os times que chegam lá e os que ficam no demo é o rigor de engenharia em cada camada.
Se eu precisasse comprimir este artigo em uma frase: RAG funciona quando você trata cada camada como decisão de engenharia com trade-offs reais, não como configuração padrão.
Comece com o pipeline mais simples que resolve seu problema. Meça com rigor. Itere na camada que mais afeta a métrica que mais importa para seus usuários. Não adicione complexidade sem evidência de que ela melhora a qualidade.
Esse é o ciclo que leva de demo a produção confiável.
Os dois frameworks mais adotados para RAG têm filosofias diferentes. Entender a diferença evita migração cara mais tarde.
LangChain pensa em termos de chains — sequências de operações que transformam inputs em outputs. É flexível para compor pipelines complexos, mas tem uma curva de aprendizado acentuada e as abstrações às vezes atrapalham o debugging.
Pontos fortes:
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_anthropic import ChatAnthropic
from langchain_openai import OpenAIEmbeddings
from langchain_qdrant import QdrantVectorStore
# Setup
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = QdrantVectorStore.from_existing_collection(
embedding=embeddings,
collection_name="knowledge_base",
url="http://localhost:6333",
)
retriever = vectorstore.as_retriever(
search_type="mmr", # Maximum Marginal Relevance
search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5}
)
# Prompt template
prompt = ChatPromptTemplate.from_template("""Responda à pergunta usando apenas o contexto fornecido.
Se a informação não estiver no contexto, diga que não encontrou.
Contexto:
{context}
Pergunta: {question}
Resposta:""")
# Formatação do contexto
def format_docs(docs):
return "\n\n---\n\n".join(
f"[Fonte: {doc.metadata.get('source', 'desconhecida')}]\n{doc.page_content}"
for doc in docs
)
# Pipeline LCEL — composição declarativa
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| ChatAnthropic(model="claude-sonnet-4-6")
| StrOutputParser()
)
# Uso com streaming
for chunk in rag_chain.stream("O que é RAG?"):
print(chunk, end="", flush=True)
LCEL com múltiplos retrievers:
from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain.retrievers import BM25Retriever, EnsembleRetriever
# Retriever BM25 para keyword search
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5
# Ensemble combina semântico + BM25 com pesos
ensemble_retriever = EnsembleRetriever(
retrievers=[retriever, bm25_retriever],
weights=[0.6, 0.4],
)
# Pipeline com ensemble
hybrid_rag_chain = (
{"context": ensemble_retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| ChatAnthropic(model="claude-sonnet-4-6")
| StrOutputParser()
)
Conversational RAG com histórico:
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.messages import HumanMessage, AIMessage
# Prompt para reformular a query com contexto do histórico
contextualize_q_prompt = ChatPromptTemplate.from_messages([
("system", """Dada uma conversa e uma pergunta de follow-up, reformule a pergunta
de forma independente que possa ser entendida sem o histórico.
Não responda à pergunta — apenas reformule se necessário."""),
("placeholder", "{chat_history}"),
("human", "{input}"),
])
history_aware_retriever = create_history_aware_retriever(
llm=ChatAnthropic(model="claude-haiku-4-5-20251001"),
retriever=retriever,
prompt=contextualize_q_prompt,
)
# Chain final com histórico
qa_prompt = ChatPromptTemplate.from_messages([
("system", "Responda usando apenas o contexto:\n\n{context}"),
("placeholder", "{chat_history}"),
("human", "{input}"),
])
question_answer_chain = create_stuff_documents_chain(
llm=ChatAnthropic(model="claude-sonnet-4-6"),
prompt=qa_prompt
)
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)
# Uso com histórico
chat_history = []
def chat(user_message: str) -> str:
response = rag_chain.invoke({
"input": user_message,
"chat_history": chat_history,
})
chat_history.extend([
HumanMessage(content=user_message),
AIMessage(content=response["answer"]),
])
return response["answer"]
LlamaIndex pensa em termos de índices — estruturas de dados otimizadas para diferentes tipos de busca. É mais opinativo, mais fácil para casos de uso de QA sobre documentos, e tem abstrações melhores para RAG.
Pontos fortes:
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
)
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.llms.anthropic import Anthropic
from llama_index.vector_stores.qdrant import QdrantVectorStore
import qdrant_client
# Configuração global
Settings.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-large-en-v1.5")
Settings.llm = Anthropic(model="claude-sonnet-4-6")
# Semantic chunking nativo
splitter = SemanticSplitterNodeParser(
buffer_size=1,
breakpoint_percentile_threshold=95,
embed_model=Settings.embed_model,
)
# Carregamento e indexação
documents = SimpleDirectoryReader("./docs").load_data()
nodes = splitter.get_nodes_from_documents(documents)
# Vector store (Qdrant)
client = qdrant_client.QdrantClient(url="http://localhost:6333")
vector_store = QdrantVectorStore(client=client, collection_name="my_docs")
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(nodes, storage_context=storage_context)
# Query engine simples
query_engine = index.as_query_engine(
similarity_top_k=5,
response_mode="compact", # refine, compact, tree_summarize
)
response = query_engine.query("O que é RAG e como funciona?")
print(response.response)
print("\nFontes:")
for node in response.source_nodes:
print(f" - {node.metadata.get('file_name')} (score: {node.score:.3f})")
RAG avançado com LlamaIndex — Sub-question decomposition:
from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool
# Múltiplos índices especializados
docs_engine = index.as_query_engine(similarity_top_k=3)
code_engine = code_index.as_query_engine(similarity_top_k=3)
# Wrap em tools
query_engine_tools = [
QueryEngineTool.from_defaults(
query_engine=docs_engine,
name="documentacao",
description="Documentação técnica sobre RAG, LLMs e NLP",
),
QueryEngineTool.from_defaults(
query_engine=code_engine,
name="exemplos_codigo",
description="Exemplos de código e implementações",
),
]
# Sub-question engine: decompõe pergunta complexa em sub-perguntas
sq_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=query_engine_tools,
use_async=True,
)
# Para "Compare implementações de RAG em LangChain vs LlamaIndex com exemplos"
# O engine automaticamente cria sub-perguntas para cada tool
response = await sq_engine.aquery(
"Compare RAG em LangChain vs LlamaIndex com exemplos de código"
)
| Critério | LangChain | LlamaIndex |
|---|---|---|
| QA sobre documentos | Bom | Excelente |
| Chatbots conversacionais | Excelente | Bom |
| Pipelines customizados | Excelente | Bom |
| Múltiplos tipos de dado | Bom | Excelente |
| Curva de aprendizado | Íngreme | Moderada |
| Ecosystem/integrações | Maior | Menor |
| Production readiness | Bom | Bom |
Muitos times usam os dois: LlamaIndex para construir e consultar índices, LangChain para orquestrar o pipeline maior que inclui o índice.
RAG pode ser caro. Cada query pode envolver: embedding da query, busca vetorial, re-ranking com cross-encoder, e chamada ao LLM. Em escala, isso soma.
Exemplo de cálculo para um sistema de produção típico:
Query embedding: $0.00002 (text-embedding-3-small, 50 tokens)
Vector search: ~$0.000 (self-hosted Qdrant)
Reranking: $0.001 (Cohere Rerank, 10 candidatos × ~100 tokens)
LLM generation: $0.006 (Claude Haiku, 1K input + 500 output)
─────────────────────────────────
Total por query: ~$0.007
Em 100k queries/mês: $700/mês
Em 1M queries/mês: $7.000/mês
Esses números variam muito com escolhas de stack. Self-hosting embedding e LLM reduz drasticamente o custo variável.
Queries repetem. Em bases de usuários grandes, 20-40% das queries são semanticamente similares a queries anteriores.
import redis
import hashlib
import numpy as np
from typing import Optional
class RAGCache:
def __init__(self, redis_client, similarity_threshold: float = 0.95, ttl: int = 3600):
self.redis = redis_client
self.threshold = similarity_threshold
self.ttl = ttl
def _get_key(self, query_embedding: np.ndarray) -> str:
"""Hash do embedding para lookup exato"""
return hashlib.sha256(query_embedding.tobytes()).hexdigest()[:16]
def get_exact(self, query: str) -> Optional[str]:
"""Cache por query string exata"""
key = f"exact:{hashlib.sha256(query.encode()).hexdigest()}"
cached = self.redis.get(key)
return cached.decode() if cached else None
def set(self, query: str, response: str, embedding: np.ndarray):
# Cache exato
exact_key = f"exact:{hashlib.sha256(query.encode()).hexdigest()}"
self.redis.setex(exact_key, self.ttl, response)
# Cache semântico: armazena embedding + response para busca por similaridade
# Em produção: use Redis com módulo RediSearch ou Faiss separado
emb_key = f"emb:{self._get_key(embedding)}"
self.redis.setex(emb_key, self.ttl, response)
# Impacto típico: 20-40% de cache hit rate reduz custo proporcional
Nem toda query precisa do modelo mais caro.
def classify_query_complexity(query: str) -> str:
"""Classifica complexidade da query para routing"""
# Heurísticas simples (pode ser substituído por um classificador)
# Perguntas simples: lookup factual direto
simple_patterns = [
r"o que é",
r"quando foi",
r"qual é o",
r"quem é",
]
# Perguntas complexas: análise, comparação, síntese
complex_patterns = [
r"compare",
r"analise",
r"por que",
r"como funciona internamente",
r"quais são as implicações",
]
import re
query_lower = query.lower()
for pattern in complex_patterns:
if re.search(pattern, query_lower):
return "complex"
for pattern in simple_patterns:
if re.search(pattern, query_lower):
return "simple"
return "medium"
def route_to_model(complexity: str) -> str:
return {
"simple": "claude-haiku-4-5-20251001",
"medium": "claude-sonnet-4-6",
"complex": "claude-opus-4-6",
}[complexity]
Para queries de seguimento em conversas, frequentemente o contexto já está disponível:
def decide_retrieval_needed(query: str, chat_history: list[dict]) -> bool:
"""Verifica se retrieval é necessário ou se histórico é suficiente"""
if not chat_history:
return True
# LLM decide se precisa buscar mais contexto
decision = cheap_llm.complete(f"""Histórico da conversa:
{format_history(chat_history)}
Nova pergunta: {query}
O histórico acima contém informação suficiente para responder?
Responda apenas: sim ou não.""")
return "não" in decision.lower()
Indexar documentos individualmente é ineficiente. Batch processing reduz custo de embedding em até 50%:
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("BAAI/bge-large-en-v1.5")
def batch_embed(texts: list[str], batch_size: int = 256) -> list[np.ndarray]:
"""Embedding em batches para eficiência máxima"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# encode() em batch é muito mais eficiente que loop
embeddings = model.encode(
batch,
batch_size=batch_size,
normalize_embeddings=True,
show_progress_bar=False,
device="cuda" # GPU quando disponível
)
all_embeddings.extend(embeddings)
return all_embeddings
# Para APIs como OpenAI, use async + rate limiting
import asyncio
import aiohttp
async def async_embed_batch(texts: list[str], semaphore_size: int = 10) -> list[list[float]]:
semaphore = asyncio.Semaphore(semaphore_size)
async def embed_single(session, text):
async with semaphore:
async with session.post(
"https://api.openai.com/v1/embeddings",
headers={"Authorization": f"Bearer {api_key}"},
json={"model": "text-embedding-3-large", "input": text}
) as response:
data = await response.json()
return data["data"][0]["embedding"]
async with aiohttp.ClientSession() as session:
tasks = [embed_single(session, text) for text in texts]
return await asyncio.gather(*tasks)
A arquitetura geral precisa de adaptações para domínios específicos.
Código tem estrutura diferente de texto. Estratégias específicas:
1. Chunking por estrutura sintática:
import ast
import tree_sitter # Para múltiplas linguagens
def chunk_by_function(code: str, language: str = "python") -> list[dict]:
"""Chunking que respeita limites de funções e classes"""
if language == "python":
tree = ast.parse(code)
chunks = []
for node in ast.walk(tree):
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
chunk_lines = code.split("\n")[node.lineno-1:node.end_lineno]
docstring = ast.get_docstring(node)
chunks.append({
"text": "\n".join(chunk_lines),
"name": node.name,
"type": type(node).__name__,
"docstring": docstring,
"start_line": node.lineno,
"end_line": node.end_lineno,
# Para busca: indexa docstring separadamente
"searchable_text": f"{node.name}: {docstring or ''}",
})
return chunks
2. Indexação dual: código + docstring: Indexe o código e sua documentação separadamente. Busca por docstring para encontrar a função certa; recupera o código completo.
3. Modelos de embedding especializados para código:
microsoft/codebert-base: treinado em pares código-docstringSalesforce/codet5p-110m-embedding: excelente para Python, Java, JStext-embedding-3-large da OpenAI: surpreendentemente bom para códigoCaracterísticas especiais:
Estratégia:
def chunk_legal_document(text: str, law_id: str) -> list[dict]:
import re
# Detecta artigos por padrão "Art. X" ou "Artigo X"
article_pattern = r"(?:Art\.|Artigo)\s+(\d+[°º]?\.?)(.+?)(?=(?:Art\.|Artigo)\s+\d|$)"
chunks = []
for match in re.finditer(article_pattern, text, re.DOTALL | re.IGNORECASE):
article_num = match.group(1)
content = match.group(2).strip()
chunks.append({
"text": f"Art. {article_num} {content}",
"law_id": law_id,
"article_number": article_num,
"type": "article",
# Para busca: texto limpo sem numeração de parágrafo
"search_text": re.sub(r"§\s*\d+[°º]?\s*", "", content),
})
return chunks
Características:
Padrões específicos:
class CustomerSupportRAG:
def __init__(self):
self.retriever = HybridRetriever()
self.reranker = CrossEncoderReranker()
self.fast_llm = AnthropicClient(model="claude-haiku-4-5-20251001")
self.confident_threshold = 0.85
def answer(self, query: str, customer_context: dict) -> dict:
# Enriquece query com contexto do cliente
enriched_query = self._enrich_query(query, customer_context)
docs = self.retriever.retrieve(enriched_query, k=20)
docs = self.reranker.rerank(enriched_query, docs, top_k=5)
# Resposta apenas se confiante
if docs[0]["score"] < self.confident_threshold:
return {
"answer": None,
"should_escalate": True,
"reason": "Sem informação suficiente para resposta confiável",
}
response = self.fast_llm.generate(
query=query,
context=docs,
system="Você é um agente de suporte. Responda de forma direta e empática. "
"Se não souber com certeza, diga e ofereça escalation."
)
return {
"answer": response,
"sources": [d["source"] for d in docs],
"should_escalate": False,
}
def _enrich_query(self, query: str, customer_context: dict) -> str:
"""Adiciona contexto do cliente para melhor retrieval"""
plan = customer_context.get("plan", "")
issue_type = customer_context.get("recent_issues", [])
if plan:
query += f" [plano: {plan}]"
if issue_type:
query += f" [contexto: {', '.join(issue_type[-2:])}]"
return query
O domínio financeiro tem requisitos únicos: números devem ser exatos, contexto temporal é crítico, cálculos precisam ser verificáveis.
Desafios específicos:
Soluções:
def extract_financial_table(table_element) -> str:
"""Converte tabela financeira em representação semântica para RAG"""
# Em vez de texto confuso, gera frases estruturadas
rows = parse_table(table_element)
statements = []
for row in rows:
metric = row.get("metric")
period = row.get("period")
value = row.get("value")
unit = row.get("unit", "")
if metric and period and value:
statements.append(
f"Em {period}, {metric} foi de {value}{unit}."
)
return " ".join(statements)
# Chunking especial para relatórios financeiros
def chunk_financial_report(document, ticker: str, period: str) -> list[dict]:
chunks = []
for section in document.sections:
if section.type == "table":
# Converte tabela em linguagem natural
text = extract_financial_table(section)
else:
text = section.text
chunks.append({
"text": text,
"ticker": ticker,
"period": period,
"section": section.name,
"type": section.type,
# Metadados temporais explícitos para filtragem
"year": period[:4],
"quarter": period[4:] if len(period) > 4 else None,
})
return chunks
RAG não existe em isolamento — frequentemente é uma ferramenta dentro de um sistema de agentes maior.
No contexto de Claude Code e frameworks de agentes, RAG se torna uma tool chamável:
from anthropic import Anthropic
client = Anthropic()
# Definição da ferramenta RAG para uso em agentes
rag_tool = {
"name": "search_knowledge_base",
"description": """Busca na base de conhecimento interna.
Use quando precisar de informações específicas sobre documentos, políticas, procedimentos ou histórico.
Retorna os trechos mais relevantes com suas fontes.""",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "A pergunta ou tópico para buscar"
},
"filters": {
"type": "object",
"description": "Filtros opcionais: document_type, date_range, department",
"properties": {
"document_type": {"type": "string"},
"department": {"type": "string"},
"date_after": {"type": "string", "format": "date"},
}
},
"k": {
"type": "integer",
"description": "Número de resultados (default: 5)",
"default": 5,
}
},
"required": ["query"]
}
}
def process_tool_call(tool_name: str, tool_input: dict) -> str:
if tool_name == "search_knowledge_base":
results = rag_system.retrieve(
query=tool_input["query"],
filters=tool_input.get("filters", {}),
k=tool_input.get("k", 5),
)
if not results:
return "Nenhum documento relevante encontrado para esta query."
formatted = []
for i, r in enumerate(results):
formatted.append(
f"**Documento {i+1}** (fonte: {r['source']}, relevância: {r['score']:.3f})\n"
f"{r['text']}"
)
return "\n\n---\n\n".join(formatted)
# Loop de agente com RAG como tool
def run_rag_agent(user_query: str) -> str:
messages = [{"role": "user", "content": user_query}]
while True:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
tools=[rag_tool],
messages=messages,
)
if response.stop_reason == "end_turn":
# Extrai texto final
return next(
block.text for block in response.content
if hasattr(block, "text")
)
if response.stop_reason == "tool_use":
# Processa tool calls
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = process_tool_call(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
# Continua o loop com resultado das tools
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
Para perguntas que requerem múltiplos steps de retrieval:
Pergunta: "Qual é a política atual de férias e como ela mudou desde 2023?"
Step 1: Busca "política de férias vigente" → Documento com política atual
Step 2: Busca "política de férias 2023 versão anterior" → Documento histórico
Step 3: Compara e responde
O agente decide naturalmente quantos retrievals fazer com base no resultado de cada busca.
Para agentes de longa duração, combine RAG (conhecimento externo) com memória de conversação (conhecimento do usuário específico):
class AgentWithRAGAndMemory:
def __init__(self):
self.rag = RAGSystem()
self.memory = VectorMemory() # Armazena fatos sobre o usuário
self.conversation_history = []
def chat(self, user_id: str, message: str) -> str:
# Recupera memórias relevantes do usuário
user_memories = self.memory.recall(
query=message,
user_id=user_id,
k=3
)
# Contexto inicial do prompt
context_parts = []
if user_memories:
context_parts.append(
"Contexto sobre o usuário:\n" +
"\n".join(f"- {m}" for m in user_memories)
)
# RAG na base de conhecimento
docs = self.rag.retrieve(message, k=5)
if docs:
formatted = format_docs(docs)
context_parts.append(f"Documentos relevantes:\n{formatted}")
full_context = "\n\n".join(context_parts)
# Geração
response = generate_with_context(
message=message,
context=full_context,
history=self.conversation_history[-10:], # Últimas 10 mensagens
)
# Atualiza histórico e memória
self.conversation_history.append({"role": "user", "content": message})
self.conversation_history.append({"role": "assistant", "content": response})
# Extrai e armazena novos fatos sobre o usuário
new_facts = extract_user_facts(message, response)
for fact in new_facts:
self.memory.store(fact, user_id=user_id)
return response
Com modelos que aceitam 1M+ tokens de contexto (Gemini 1.5 Pro, Claude 3.5 com Project Knowledge), surge a pergunta: para que RAG se posso colocar tudo no contexto?
A resposta curta: para bases pequenas (< 1M tokens), long context pode ser suficiente. Para bases grandes, RAG continua necessário. Mas a fronteira está se movendo.
Comparação empírica (baseada em pesquisa de 2024):
| Tamanho da base | RAG | Long context | Vencedor |
|---|---|---|---|
| < 100K tokens | Overhead desnecessário | Simples e eficaz | Long context |
| 100K - 1M tokens | Boa opção | Possível mas caro | Empate |
| 1M - 10M tokens | Necessário | Inviável/caro | RAG |
| > 10M tokens | Necessário | Impossível | RAG |
O custo é o fator decisivo: processar 1M tokens no contexto a cada query custa ~$0.70 com Claude (input tokens). Com RAG e retrieval eficiente, você processa 5-20K tokens de contexto → ~$0.01 por query.
Para 100k queries/dia: Long context = $70.000/dia vs RAG = $1.000/dia.
A tendência mais promissora é treinar o modelo para usar RAG melhor:
Com GPT-4V, Claude 3+ e Gemini sendo natively multimodal, RAG está se expandindo para:
Para organizações com dados em múltiplos silos com restrições de compliance:
Lewis, P., et al. (2020). Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks. Facebook AI Research / NIPS 2020. https://arxiv.org/abs/2005.11401
Gao, L., et al. (2022). Precise Zero-Shot Dense Retrieval without Relevance Labels (HyDE). https://arxiv.org/abs/2212.10496
Liu, N. F., et al. (2023). Lost in the Middle: How Language Models Use Long Contexts. https://arxiv.org/abs/2307.03172
Asai, A., et al. (2023). Self-RAG: Learning to Retrieve, Generate, and Critique. https://arxiv.org/abs/2310.11511
Yan, S., et al. (2024). Corrective Retrieval Augmented Generation (CRAG). https://arxiv.org/abs/2401.15884
Edge, D., et al. (2024). From Local to Global: A Graph RAG Approach to Query-Focused Summarization. Microsoft Research. https://arxiv.org/abs/2404.16130
Chen, J., et al. (2023). Dense X Retrieval: What Retrieval Granularity Should We Use? https://arxiv.org/abs/2312.06648
Anthropic. (2024). Introducing Contextual Retrieval. https://www.anthropic.com/news/contextual-retrieval
MTEB Leaderboard. Massive Text Embedding Benchmark. https://huggingface.co/spaces/mteb/leaderboard
Es, S., et al. (2023). RAGAS: Automated Evaluation of Retrieval Augmented Generation. https://arxiv.org/abs/2309.15217
Muennighoff, N., et al. (2022). MTEB: Massive Text Embedding Benchmark. https://arxiv.org/abs/2210.07316
Cormack, G. V., et al. (2009). Reciprocal Rank Fusion outperforms Condorcet and individual Rank Learning Methods. SIGIR 2009.
LangChain Documentation. https://python.langchain.com/docs/
LlamaIndex Documentation. https://docs.llamaindex.ai/
Qdrant Documentation. https://qdrant.tech/documentation/
Weaviate Documentation. https://weaviate.io/developers/weaviate
Pinecone Documentation. https://docs.pinecone.io/
Voyager AI. (2023). Voyage Embedding Models. https://www.voyageai.com/
BAAI. (2023). BGE M3-Embedding: Multi-Lingual, Multi-Functionality, Multi-Granularity Text Embeddings Through Self-Knowledge Distillation. https://arxiv.org/abs/2309.07597
Cohere. (2023). Rerank — Improving Search Quality. https://cohere.com/rerank