Filtragem Dinâmica em RAG

Quando construímos assistentes de IA com LangChain, especialmente para tarefas como análise de documentos, um dos maiores desafios é gerenciar o contexto. Como garantimos que nossa IA busque informações apenas nos documentos relevantes para uma tarefa específica, e não em toda a base de dados?

O Desafio: Um Retriever para Múltiplos Contextos

Vamos imaginar um cenário comum: temos um único Vector Store com documentos de vários processos diferentes. Cada documento possui um metadado que o identifica, como process_id.

Se criarmos um retriever padrão, ele buscará em todos os documentos. O que queremos é que nosso agente, ao receber uma tarefa para o "processo A", use uma ferramenta que busque apenas nos documentos com process_id: "A". Na próxima execução, para o "processo B", a mesma ferramenta deve filtrar por process_id: "B".

Como criamos essa ferramenta dinâmica?

Desacoplando o Estado da Ferramenta com um Nó Customizado

Nossa primeira intuição poderia ser tentar "injetar" o estado diretamente na assinatura da ferramenta. No entanto, a forma mais robusta e explícita no LangGraph é separar as responsabilidades:

A Ferramenta: Deve ser simples. Ela faz uma única coisa: busca documentos recebendo uma query e um process_id como identificador do "processo", parâmetros normais. Ela não precisa saber de onde o process_id veio.

1. O Nó do Grafo (Tool Node): Este é o conector. É um nó customizado que lê o estado do grafo (onde o process_id da execução atual está armazenado) e o passa como argumento para a ferramenta no momento da chamada.

Vamos a prática.

Passo 1: A Ferramenta Pura

Primeiro, definimos nossa ferramenta com @tool. Note que ela é uma função Python simples, sem dependências mágicas do estado. Isso a torna fácil de testar e reutilizar.

from langchain_core.documents import Document
from langchain_core.tools import tool
from langchain_community.vectorstores.base import VectorStore

# Assumimos que 'store' é o seu VectorStore já populado
store: VectorStore 

@tool
def busca_semantica_por_processo(query: str, process_id: str) -> list[Document]:
 """
 Realiza uma busca semântica em documentos, filtrando por um 'process_id' específico.
 Use esta ferramenta para encontrar informações dentro de um processo conhecido.
 
 Args:
 query (str): O texto para a busca de informação relevante.
 process_id (str): O identificador único do processo para filtrar a busca.
 """
 print(f"🔧 EXECUTANDO FERRAMENTA: Buscando por '{query}' no processo '{process_id}'")
 
 # Cria um retriever na hora, aplicando o filtro dinâmico
 process_retriever = store.as_retriever(
 search_type="mmr", # ou "similarity", etc.
 search_kwargs={'filter': {'process_id': process_id}}
 )
 
 result = process_retriever.invoke(query)
 
 if not result:
 print(f"📢 Nenhum documento encontrado para a busca.")
 return []
 
 print(f"🎯 Encontrados {len(result)} documentos.")
 return result

Passo 2: Grafo de contexto

Agora, vamos usar o ToolNode padrão do LangGraph, para criarmos um grafo de chamada de funções que da as instrue a llm a usar a ferramenta de recuperação de contexto. Porem como ela ira saber qual processo buscar?

def get_context_graph(
 llm: BaseChatModel = get_llm()
):

 llm_with_tools = llm.bind_tools([busca_semantica_por_processo])

 sys_msg = SystemMessage(...)

 
 def assistant(state):
 print("🤖 assistente acionado")
 print(state)
 return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

 # Build graph
 builder = StateGraph(SummarizeEntityState)
 builder.add_node("assistant", assistant)
 builder.add_node("tools", ToolNode(tools))

 builder.add_edge(START, "assistant")
 builder.add_conditional_edges(
 "assistant",
 # If the latest message (result) from assistant is a tool call -> tools_condition routes to tools
 # If the latest message (result) from assistant is a not a tool call -> tools_condition routes to END
 tools_condition,
 )
 builder.add_edge("tools", "assistant")

 # Compile graph
 context_graph = builder.compile()
 return context_graph

Passo 3: O grande truque

Agora podemos usar o nosso grafo de recuperação de contexto como um simples no. Para usa-lo só precisamos passar na primeira posição de message do stado do grafo um prompt contento os identificadores únicos dos processos que ele será capaz de recuperar o contexto.

def retriever_context_node(state: SummarizeEntityState) -> SummarizeEntityState:
 ...
 state["messages"] = [...Prompt com os id]
 return state