Introdução
A Geração Aumentada por Recuperação começou como uma ideia simples: em vez de depender exclusivamente da memória paramétrica de um modelo de linguagem, recuperar documentos relevantes de uma base de conhecimento externa e incluí-los no contexto do prompt. Este padrão de RAG ingênuo, consistindo em embed-recuperar-gerar, funcionou surpreendentemente bem para resposta a perguntas simples sobre coleções de documentos estruturados. Mas quando as equipes empurraram o RAG para produção para casos de uso mais complexos, suas limitações ficaram dolorosamente claras. Consultas que exigiam raciocínio através de múltiplos documentos, perguntas ambíguas que necessitavam de esclarecimento e bases de conhecimento com tipos de conteúdo heterogêneos expuseram a fragilidade do pipeline de recuperar-e-gerar.
O RAG agêntico representa uma mudança fundamental em como arquitetamos sistemas de recuperação. Em vez de um pipeline fixo onde cada etapa alimenta linearmente a próxima, o RAG agêntico dá ao modelo de linguagem a capacidade de planejar sua estratégia de recuperação, avaliar a qualidade dos resultados recuperados, reformular consultas quando a recuperação inicial falha e decidir quando tem informação suficiente para gerar uma resposta final. O modelo se torna um participante ativo no processo de recuperação em vez de um consumidor passivo do contexto recuperado.
Este guia cobre os padrões de arquitetura, frameworks de implementação, métodos de avaliação e considerações de produção para construir sistemas RAG agênticos. Baseamo-nos em implantações do mundo real em bases de conhecimento empresariais, sistemas de suporte ao cliente e plataformas de documentação técnica onde esses padrões foram validados em escala.
Do RAG Ingênuo ao RAG Agêntico: O Que Mudou
O RAG ingênuo segue um pipeline determinístico de três etapas: a consulta do usuário é embedded, uma busca de similaridade vetorial recupera os top-k fragmentos de documentos mais similares, e esses fragmentos são concatenados em um prompt que o LLM usa para gerar uma resposta. Este pipeline tem três fraquezas fundamentais que o RAG agêntico aborda.
Primeiro, o RAG ingênuo assume que a consulta do usuário já está bem formulada para recuperação. Na prática, as consultas dos usuários são frequentemente vagas, multifacetadas ou usam terminologia que não corresponde ao vocabulário dos documentos indexados. Um usuário perguntando sobre falhas de implantação pode precisar de documentos sobre tratamento de erros, configuração de infraestrutura e pipelines de CI/CD, mas uma única busca vetorial pode retornar apenas uma dessas facetas.
Segundo, o RAG ingênuo não tem portão de qualidade. Se os documentos recuperados são irrelevantes, desatualizados ou insuficientes, o pipeline prossegue mesmo assim e o LLM gera uma resposta a partir de contexto deficiente. Não há mecanismo para o sistema reconhecer uma falha de recuperação e tentar novamente.
Terceiro, o RAG ingênuo trata todas as consultas de forma idêntica. Uma pergunta de busca factual, uma pergunta analítica complexa e uma pergunta que requer síntese de múltiplas fontes passam todas pelo mesmo pipeline de recuperar-gerar. O RAG agêntico introduz lógica condicional que seleciona diferentes estratégias de recuperação e geração baseando-se nas características da consulta.
A transição do RAG ingênuo para o agêntico envolve adicionar três capacidades: planejamento de consultas (decompor consultas complexas em subconsultas), avaliação de recuperação (avaliar se o contexto recuperado é suficiente) e refinamento iterativo (reformular consultas e re-recuperar quando os resultados são insuficientes). Essas capacidades transformam um pipeline estático em um sistema dinâmico e autocorretivo.
Padrões de Arquitetura Fundamentais
Os sistemas RAG agênticos são construídos a partir de um pequeno número de padrões componíveis. Compreender esses padrões permite que você projete sistemas sob medida para seu caso de uso específico.
Roteamento
O padrão agêntico mais simples roteia consultas para diferentes backends de recuperação baseando-se na classificação da consulta. Um agente roteador analisa a consulta recebida e a direciona para a fonte de conhecimento mais apropriada:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
router_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user query into one of these categories:
- technical_docs: Questions about API usage, configuration, or code
- policy: Questions about company policies, procedures, or compliance
- support: Questions about troubleshooting or known issues
- general: General questions that don't fit other categories
Respond with ONLY the category name."""),
("human", "{query}")
])
router_chain = router_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
O roteamento é valioso quando sua base de conhecimento abrange múltiplos domínios com diferentes estratégias de indexação. A documentação técnica pode estar indexada com chunking consciente de código e embeddings, enquanto documentos de políticas usam chunking semântico com filtros de metadados.
Decomposição de Consultas
Perguntas complexas frequentemente requerem informações de múltiplos fragmentos de documentos que não apareceriam juntos em uma única recuperação. A decomposição de consultas divide uma consulta complexa em subconsultas independentes, recupera para cada uma e sintetiza os resultados:
decomposition_prompt = ChatPromptTemplate.from_messages([
("system", """Break the following complex question into 2-4 simpler
sub-questions that, when answered together, provide a complete answer
to the original question. Return as a JSON array of strings."""),
("human", "{query}")
])
# Example: "How does our API rate limiting compare to competitors
# and what are customers saying about it?"
# Decomposes to:
# ["What are our current API rate limiting policies?",
# "What rate limiting do our main competitors use?",
# "What customer feedback have we received about rate limiting?"]
Autocorreção
A autocorreção é o padrão que mais distingue o RAG agêntico do RAG ingênuo. Após a recuperação, uma etapa de avaliação examina se os documentos recuperados são relevantes e suficientes. Se não forem, o sistema reformula a consulta e tenta novamente:
grading_prompt = ChatPromptTemplate.from_messages([
("system", """You are a retrieval quality grader. Given a user question
and a set of retrieved documents, determine:
1. Are the documents relevant to the question? (yes/no)
2. Do the documents contain sufficient information to answer? (yes/no)
3. If insufficient, suggest a reformulated query.
Respond as JSON with keys: relevant, sufficient, reformulated_query"""),
("human", "Question: {query}\n\nDocuments: {documents}")
])
Este padrão tipicamente permite 2-3 tentativas de recuperação antes de recorrer a uma mensagem de falha elegante. As consultas reformuladas frequentemente mudam de busca por similaridade semântica para busca baseada em palavras-chave, ampliam o escopo ou visam campos de metadados específicos.
LangGraph para Orquestração de Agentes com Estado
O LangGraph emergiu como o framework padrão para construir sistemas RAG agênticos porque fornece gerenciamento de estado explícito, roteamento condicional e suporte a ciclos que as abstrações mais simples baseadas em cadeias do LangChain não conseguem expressar.
Um sistema RAG agêntico baseado em LangGraph é definido como um grafo de estados onde os nós representam etapas de processamento e as arestas representam transições entre etapas. Arestas condicionais permitem que o grafo se ramifique baseando-se em resultados intermediários:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from operator import add
class AgentState(TypedDict):
query: str
sub_queries: List[str]
documents: List[dict]
generation: str
retry_count: int
retrieval_grade: str
def route_query(state: AgentState) -> AgentState:
"""Classify and route the incoming query."""
query = state["query"]
classification = router_chain.invoke({"query": query})
state["route"] = classification
return state
def retrieve(state: AgentState) -> AgentState:
"""Retrieve documents based on query and route."""
query = state["query"]
docs = retriever.invoke(query)
state["documents"] = docs
return state
def grade_documents(state: AgentState) -> AgentState:
"""Grade retrieved documents for relevance and sufficiency."""
grade = grading_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["retrieval_grade"] = grade["sufficient"]
state["retry_count"] = state.get("retry_count", 0) + 1
if not grade["sufficient"] and grade.get("reformulated_query"):
state["query"] = grade["reformulated_query"]
return state
def generate(state: AgentState) -> AgentState:
"""Generate final answer from retrieved context."""
answer = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["generation"] = answer
return state
def should_retry(state: AgentState) -> str:
"""Decide whether to retry retrieval or proceed to generation."""
if state["retrieval_grade"] == "yes":
return "generate"
if state["retry_count"] >= 3:
return "generate" # Give up and generate with what we have
return "retrieve"
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("route", route_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.set_entry_point("route")
workflow.add_edge("route", "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"retrieve": "retrieve",
"generate": "generate"
})
workflow.add_edge("generate", END)
app = workflow.compile()
A estrutura do grafo torna o fluxo de controle explícito e depurável. Você pode visualizar o grafo, rastrear execuções através de cada nó e entender exatamente por que o sistema tomou um caminho particular para qualquer consulta. Essa transparência é crítica para sistemas de produção onde você precisa diagnosticar falhas e explicar o comportamento.
O LangGraph também suporta padrões mais avançados como recuperação paralela através de múltiplos índices, checkpoints humano-no-loop onde o agente pausa para confirmação do usuário, e estado persistente que sobrevive entre conversações.
Estratégias de Recuperação
A camada de recuperação em um sistema RAG agêntico é tipicamente mais sofisticada do que uma única busca em vector store. Os sistemas de produção combinam múltiplas estratégias de recuperação.
Busca Híbrida
A busca híbrida combina a recuperação vetorial densa com a recuperação esparsa por palavras-chave, proporcionando a compreensão semântica dos embeddings com a precisão do matching de palavras-chave BM25:
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="documents",
embedding_function=embedding_model,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
Re-ranking
Após a recuperação inicial, um re-ranker de codificador cruzado pontua cada documento contra a consulta com precisão muito superior à similaridade de codificador bidirecional. Isso é computacionalmente caro mas melhora drasticamente a precisão:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
reranker = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever,
)
Recuperação Multi-Índice
As bases de conhecimento de produção frequentemente abrangem múltiplos índices com esquemas diferentes, modelos de embeddings e tipos de documentos. Um sistema agêntico pode consultar múltiplos índices em paralelo e mesclar resultados:
async def multi_index_retrieve(query: str, indexes: list) -> list:
"""Retrieve from multiple indexes in parallel."""
import asyncio
async def retrieve_from_index(index, query):
return await index.aretrieve(query)
tasks = [retrieve_from_index(idx, query) for idx in indexes]
results = await asyncio.gather(*tasks)
# Flatten and deduplicate
all_docs = []
seen_ids = set()
for result_set in results:
for doc in result_set:
if doc.metadata["id"] not in seen_ids:
all_docs.append(doc)
seen_ids.add(doc.metadata["id"])
return all_docs
Avaliação com RAGAS e DeepEval
Avaliar sistemas RAG agênticos requer medir múltiplas dimensões de qualidade. RAGAS e DeepEval são os dois frameworks de avaliação mais amplamente adotados, cada um com pontos fortes distintos.
O RAGAS fornece um conjunto de métricas sem referência que avaliam a qualidade do RAG sem exigir respostas de verdade base para cada consulta de teste:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"answer": generated_answers,
"contexts": retrieved_contexts,
"ground_truth": reference_answers,
})
results = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
],
)
print(results)
A fidelidade mede se a resposta gerada é suportada pelo contexto recuperado, detectando alucinações. A relevância da resposta mede se a resposta aborda a pergunta. A precisão do contexto mede se os documentos recuperados são relevantes, e o recall do contexto mede se a recuperação capturou todas as informações necessárias.
O DeepEval estende a avaliação com métricas adicionais particularmente relevantes para sistemas agênticos:
from deepeval import evaluate
from deepeval.metrics import (
FaithfulnessMetric,
AnswerRelevancyMetric,
ContextualRelevancyMetric,
HallucinationMetric,
)
from deepeval.test_case import LLMTestCase
test_case = LLMTestCase(
input="How do I configure rate limiting?",
actual_output=generated_answer,
expected_output=reference_answer,
retrieval_context=retrieved_chunks,
)
metrics = [
FaithfulnessMetric(threshold=0.8),
AnswerRelevancyMetric(threshold=0.7),
ContextualRelevancyMetric(threshold=0.7),
HallucinationMetric(threshold=0.5),
]
evaluate(test_cases=[test_case], metrics=metrics)
Para RAG agêntico especificamente, você também deve medir a eficiência de recuperação: quantas rodadas de recuperação o sistema precisa em média, qual porcentagem de consultas requer reformulação e como a qualidade da resposta se degrada ao longo das tentativas de retry. Essas métricas são específicas do loop agêntico e não são cobertas pelos frameworks padrão de avaliação RAG.
Padrões de Produção
Mover o RAG agêntico do protótipo para produção requer abordar preocupações de confiabilidade, segurança e operacionais que não surgem durante o desenvolvimento.
Guardrails
Todo sistema RAG de produção precisa de guardrails que impeçam o LLM de gerar respostas prejudiciais, fora de tópico ou sem suporte factual:
from guardrails import Guard
from guardrails.hub import ToxicLanguage, CompetitorCheck
guard = Guard().use_many(
ToxicLanguage(on_fail="exception"),
CompetitorCheck(
competitors=["competitor_a", "competitor_b"],
on_fail="fix"
),
)
raw_response = generation_chain.invoke({"query": query, "documents": docs})
validated_response = guard.validate(raw_response)
Cadeias de Fallback
Quando o loop agêntico não consegue encontrar contexto suficiente após o máximo de retries, o sistema precisa de degradação elegante em vez de gerar uma resposta não confiável:
def generate_with_fallback(state: AgentState) -> AgentState:
if state["retrieval_grade"] != "yes" and state["retry_count"] >= 3:
state["generation"] = (
"I wasn't able to find sufficient information in our knowledge base "
"to fully answer your question. Here's what I found:\n\n"
f"{partial_answer_from_context(state['documents'])}\n\n"
"For a complete answer, I'd recommend contacting the support team."
)
state["confidence"] = "low"
else:
state["generation"] = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["confidence"] = "high"
return state
Humano-no-Loop
Para aplicações de alto risco, o LangGraph suporta pontos de interrupção onde o agente pausa a execução e aguarda aprovação humana:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Add interrupt before generation for sensitive queries
workflow.add_node("human_review", lambda state: state)
workflow.add_conditional_edges(
"grade",
lambda state: "human_review" if state.get("sensitive") else "generate",
{"human_review": "human_review", "generate": "generate"}
)
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
Observabilidade com LangSmith e Phoenix
A observabilidade é inegociável para sistemas agênticos em produção. A natureza não determinística do fluxo de controle dirigido por LLM significa que você não pode prever ou testar todos os caminhos de execução possíveis. Você precisa de rastreamento abrangente para entender o que seu sistema está fazendo em produção.
O LangSmith fornece rastreamento de ponta a ponta para aplicações LangGraph:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag-production"
# All LangGraph invocations are automatically traced
result = app.invoke({"query": "How do I configure rate limiting?"})
Cada trace mostra o caminho de execução completo através do grafo: quais nós foram visitados, quais foram as entradas e saídas do LLM em cada etapa, resultados de recuperação, decisões de avaliação e contagens de tokens. Isso é essencial para depurar casos onde o agente toma um caminho inesperado ou gera uma resposta deficiente.
O Arize Phoenix fornece uma alternativa de código aberto com foco no monitoramento de qualidade de recuperação:
import phoenix as px
from phoenix.trace.langchain import LangChainInstrumentor
px.launch_app()
LangChainInstrumentor().instrument()
# Traces are now collected in Phoenix
result = app.invoke({"query": user_query})
O Phoenix é particularmente valioso para monitorar a qualidade de recuperação ao longo do tempo. Ele rastreia o drift de embeddings, distribuições de relevância de recuperação e pode alertar quando a qualidade de recuperação se degrada, o que frequentemente indica que a base de conhecimento se desviou da distribuição de treinamento do modelo de embeddings e precisa de reindexação.
Métricas-chave para monitorar em produção:
# Custom metrics to track
metrics = {
"avg_retrieval_rounds": [],
"reformulation_rate": [],
"fallback_rate": [],
"latency_p50_ms": [],
"latency_p99_ms": [],
"token_cost_per_query": [],
"faithfulness_score": [],
}
Otimização de Custo e Latência
Os sistemas RAG agênticos são inerentemente mais caros e lentos do que o RAG ingênuo porque fazem múltiplas chamadas ao LLM por consulta. Otimizar custo e latência sem sacrificar qualidade requer decisões arquitetônicas cuidadosas.
Use modelos menores para roteamento e avaliação. O roteador e o avaliador de documentos não precisam da capacidade completa de raciocínio de um modelo de fronteira. GPT-4o-mini ou Claude 3.5 Haiku podem lidar com essas tarefas a uma fração do custo e latência:
# Use a small, fast model for routing and grading
routing_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
grading_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Use a larger model only for final generation
generation_model = ChatOpenAI(model="gpt-4o", temperature=0.1)
Faça cache dos resultados de recuperação agressivamente. Muitas consultas de usuários são variações da mesma pergunta subjacente. Um cache semântico que combina consultas dentro de um limiar de similaridade pode eliminar recuperações e chamadas LLM redundantes:
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=embedding_model,
score_threshold=0.95,
))
Defina timeouts agressivos em cada etapa para prevenir latência descontrolada. Uma única chamada LLM lenta não deve causar o timeout de toda a requisição:
from asyncio import wait_for, TimeoutError
async def retrieve_with_timeout(state, timeout_seconds=5):
try:
return await wait_for(retrieve(state), timeout=timeout_seconds)
except TimeoutError:
state["documents"] = []
state["retrieval_grade"] = "no"
return state
Casos de Estudo Reais e Anti-Padrões
O padrão de falha mais comum em implantações de RAG agêntico é o loop de retry infinito. Sem um limite rígido nas tentativas de recuperação e comportamento de fallback claro, o agente pode ciclar por reformulações indefinidamente, queimando tokens e latência. Sempre defina limites de retry explícitos e meça sua taxa de reformulação. Se mais de 20% das consultas requerem reformulação, o problema provavelmente está na sua camada de recuperação (chunking deficiente, modelo de embeddings errado, índice desatualizado) e não na lógica do agente.
Outro anti-padrão frequente é o over-routing. As equipes às vezes constroem grafos de roteamento complexos com dezenas de índices especializados quando um único retriever híbrido bem projetado teria melhor desempenho. Comece com a arquitetura mais simples que atenda seus requisitos e adicione complexidade apenas quando tiver métricas mostrando que abordagens mais simples são insuficientes.
O enchimento de janela de contexto é um terceiro anti-padrão. Recuperar 20 fragmentos de documentos e amontoá-los todos no prompt desperdiça tokens e pode realmente degradar a qualidade da resposta porque o LLM tem dificuldade em identificar a informação relevante em um mar de contexto marginalmente relacionado. O re-ranking para um pequeno número de fragmentos de alta qualidade (tipicamente 3-5) supera consistentemente janelas de contexto maiores em benchmarks de avaliação.
Finalmente, cuidado com a otimização excessiva dirigida por avaliação. Equipes que otimizam exclusivamente para pontuações RAGAS podem construir sistemas que pontuam bem em benchmarks mas produzem respostas excessivamente cautelosas e evasivas que os usuários acham inúteis. Equilibre a avaliação quantitativa com a revisão qualitativa de conversas reais de usuários.
O Futuro: Sistemas RAG Multi-Agente
A próxima fronteira no RAG agêntico são os sistemas multi-agente onde agentes especializados colaboram em tarefas complexas de recuperação e síntese. Em vez de um único agente lidando com todo o pipeline de recuperação-geração, você implanta uma equipe de agentes com diferentes especializações.
Um agente de pesquisa lida com a decomposição de consultas e recuperação multi-hop através de grandes coleções de documentos. Um agente de verificação de fatos verifica as afirmações geradas contra os documentos fonte. Um agente de síntese combina as descobertas de múltiplas subconsultas em uma resposta coerente e bem estruturada. Um agente editor revisa a saída final quanto à clareza, precisão e tom.
from langgraph.graph import StateGraph
# Multi-agent RAG architecture
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("planner", planner_agent)
multi_agent.add_node("researcher", researcher_agent)
multi_agent.add_node("fact_checker", fact_check_agent)
multi_agent.add_node("synthesizer", synthesis_agent)
multi_agent.set_entry_point("planner")
multi_agent.add_edge("planner", "researcher")
multi_agent.add_edge("researcher", "fact_checker")
multi_agent.add_conditional_edges(
"fact_checker",
lambda s: "researcher" if s["needs_more_research"] else "synthesizer",
)
multi_agent.add_edge("synthesizer", END)
Esta arquitetura é mais cara por consulta mas lida com questões de pesquisa complexas com as quais sistemas de agente único têm dificuldade. O desafio de design chave é definir interfaces claras entre agentes e garantir que a sobrecarga de comunicação inter-agente não anule os benefícios da especialização.
O ecossistema de ferramentas está convergindo para suportar padrões multi-agente nativamente. A composição de subgrafos do LangGraph, a camada de orquestração de agentes do LlamaIndex e o CrewAI fornecem todos primitivas para construir sistemas RAG multi-agente. À medida que esses frameworks amadurecem e os custos continuam caindo, o RAG multi-agente se tornará prático para uma gama mais ampla de aplicações de produção. Os princípios cobertos neste guia — gerenciamento de estado explícito, avaliação de qualidade de recuperação, fallbacks elegantes e observabilidade abrangente — permanecem essenciais independentemente de você implantar um único agente ou uma equipe de agentes especializados.