​​Construindo um fluxo de trabalho RAG usando LangGraph e Elasticsearch

O modelo de agente de recuperação LangGraph é um projeto inicial desenvolvido pela LangChain para facilitar a criação de sistemas de perguntas e respostas baseados em recuperação usando o LangGraph no LangGraph Studio. Este modelo é pré-configurado para integração perfeita com o Elasticsearch, permitindo que os desenvolvedores criem rapidamente agentes que podem indexar e recuperar documentos de forma eficiente.

Este blog se concentra na execução e personalização do modelo do agente de recuperação LangChain usando o LangGraph Studio e o LangGraph CLI. O modelo fornece uma estrutura para a criação de aplicativos de geração aumentada de recuperação (RAG), aproveitando vários backends de recuperação, como o Elasticsearch.

Orientaremos você na configuração do ambiente e na execução eficiente do modelo com o Elastic, ao mesmo tempo em que personalizamos o fluxo do agente.

Pré-requisitos

Antes de prosseguir, certifique-se de ter o seguinte instalado:

• Implantação do Elasticsearch Cloud ou implantação do Elasticsearch no local (ou crie uma avaliação gratuita de 14 dias no Elastic Cloud) - Versão 8.0.0 ou superior

• Python 3.9+

• Acesso a um provedor de LLM como Cohere (usado neste guia), OpenAI ou Anthropic/Claude

Criando o aplicativo LangGraph

1. Instale o LangGraph CLI

pip install --upgrade "langgraph-cli[inmem]"

2. Crie o aplicativo LangGraph a partir do modelo de agente de recuperação

mkdir lg-agent-demo
cd lg-agent-demo
langgraph new lg-agent-demo

Será apresentado um menu interativo que permitirá que você escolha entre uma lista de modelos disponíveis. Selecione 4 para Agente de Recuperação e 1 para Python, conforme mostrado abaixo:

• Solução de problemas: se você encontrar o erro “urllib.error.URLError: <erro urlopen [SSL: CERTIFICATE_VERIFY_FAILED] falha na verificação do certificado: não é possível obter o certificado do emissor local (_ssl.c:1000)> “

Execute o comando Instalar Certificado do Python para resolver o problema, conforme mostrado abaixo.

3. Instalar dependências

Na raiz do seu novo aplicativo LangGraph, crie um ambiente virtual e instale as dependências no modo edit para que suas alterações locais sejam usadas pelo servidor:

#For Mac
python3 -m venv lg-demo
source lg-demo/bin/activate 
pip install -e .

#For Windows
python3 -m venv lg-demo
lg-demo\Scripts\activate 
pip install -e .

Configurando o ambiente

1. Crie um arquivo .environment arquivo

O arquivo .env contém chaves de API e configurações para que o aplicativo possa se conectar ao LLM e ao provedor de recuperação escolhidos. Gere um novo arquivo .env duplicando a configuração de exemplo:

cp .env.example .env

2. Configure o arquivo .env arquivo

O arquivo .env vem com um conjunto de configurações padrão. Você pode atualizá-lo adicionando as chaves de API e os valores necessários com base na sua configuração. Quaisquer chaves que não sejam relevantes para seu caso de uso podem ser deixadas inalteradas ou removidas.

# To separate your traces from other applications
LANGSMITH_PROJECT=retrieval-agent

# LLM choice (set the API key for your selected provider):
ANTHROPIC_API_KEY=your_anthropic_api_key
FIREWORKS_API_KEY=your_fireworks_api_key
OPENAI_API_KEY=your_openai_api_key

# Retrieval provider (configure based on your chosen service):

## Elastic Cloud:
ELASTICSEARCH_URL=https://your_elastic_cloud_url
ELASTICSEARCH_API_KEY=your_elastic_api_key

## Elastic Local:
ELASTICSEARCH_URL=http://host.docker.internal:9200
ELASTICSEARCH_USER=elastic
ELASTICSEARCH_PASSWORD=changeme

## Pinecone:
PINECONE_API_KEY=your_pinecone_api_key
PINECONE_INDEX_NAME=your_pinecone_index_name

## MongoDB Atlas:
MONGODB_URI=your_mongodb_connection_string

# Cohere API key:
COHERE_API_KEY=your_cohere_api_key

• Exemplo de arquivo .env (usando Elastic Cloud e Cohere)

Abaixo está um exemplo de configuração .env para usar o Elastic Cloud como provedor de recuperação e o Cohere como LLM, conforme demonstrado neste blog:

# To separate your traces from other applications
LANGSMITH_PROJECT=retrieval-agent
#Retrieval Provider
# Elasticsearch configuration
ELASTICSEARCH_URL=elastic-url:443
ELASTICSEARCH_API_KEY=elastic_api_key
# Cohere API key
COHERE_API_KEY=cohere_api_key

Observação: embora este guia utilize o Cohere para geração de respostas e incorporações, você pode usar outros provedores de LLM, como OpenAI, Claudeou até mesmo um modelo de LLM local, dependendo do seu caso de uso. Certifique-se de que cada chave que você pretende usar esteja presente e definida corretamente no arquivo.env.

3. Atualize o arquivo de configuração - configuration.py

Depois de configurar seu arquivo .env com as chaves de API apropriadas, a próxima etapa é atualizar a configuração do modelo padrão do seu aplicativo. Atualizar a configuração garante que o sistema use os serviços e modelos que você especificou no seu arquivo .env .

Navegue até o arquivo de configuração:

 cd src/retrieval_graph

O arquivo configuration.py contém as configurações de modelo padrão usadas pelo agente de recuperação para três tarefas principais:

• Modelo de incorporação – converte documentos em representações vetoriais

• Modelo de consulta – processa a consulta do usuário em um vetor

• Modelo de resposta – gera a resposta final

Por padrão, o código usa modelos do OpenAI (por exemplo, openai/text-embedding-3-small) e do Anthropic (por exemplo, anthropic/claude-3-5-sonnet-20240620 and anthropic/claude-3-haiku-20240307). Neste blog, estamos mudando para o uso de modelos Cohere. Se você já estiver usando OpenAI ou Anthropic, nenhuma alteração será necessária.

Exemplos de alterações (usando Cohere):

Abra configuration.py e modifique os padrões do modelo conforme mostrado abaixo:

…
 embedding_model: Annotated[
       str,
       {"__template_metadata__": {"kind": "embeddings"}},
   ] = field(
       default="cohere/embed-english-v3.0",
…
response_model: Annotated[str, {"__template_metadata__": {"kind": "llm"}}] = field(
       default="cohere/command-r-08-2024",
…
query_model: Annotated[str, {"__template_metadata__": {"kind": "llm"}}] = field(
       default="cohere/command-r-08-2024",
       metadata={

Executando o agente de recuperação com a CLI do LangGraph

1. Inicie o servidor LangGraph

cd lg-agent-demo
langgraph dev

Isso iniciará o servidor LangGraph API localmente. Se isso for executado com sucesso, você deverá ver algo como:

URL da interface do usuário do Open Studio.

Há dois gráficos disponíveis:

• Gráfico de recuperação: Recupera dados do Elasticsearch e responde à consulta usando um LLM (Language-Level Model).

• Gráfico do indexador: Indexa documentos no Elasticsearch e gera embeddings usando um LLM.

2. Configurando o gráfico do indexador

• Abra o gráfico do indexador.

• Clique em Gerenciar assistentes.

  • Clique em 'Adicionar novo assistente ', insira os dados do usuário conforme especificado e, em seguida, feche a janela.

{"user_id": "101"}

3. Indexação de documentos de amostra

• Indexe os seguintes documentos de exemplo, que representam um relatório trimestral hipotético para a organização NoveTech:

[
  {    "page_content": "NoveTech Solutions Q1 2025 Report - Revenue: $120.5M, Net Profit: $18.2M, EPS: $2.15. Strong AI software launch and $50M government contract secured."
  },
  {
    "page_content": "NoveTech Solutions Business Highlights - AI-driven analytics software gained 15% market share. Expansion into Southeast Asia with two new offices. Cloud security contract secured."
  },
  {
    "page_content": "NoveTech Solutions Financial Overview - Operating expenses at $85.3M, Gross Margin 29.3%. Stock price rose from $72.5 to $78.3. Market Cap reached $5.2B."
  },
  {
    "page_content": "NoveTech Solutions Challenges - Rising supply chain costs impacting hardware production. Regulatory delays slowing European expansion. Competitive pressure in cybersecurity sector."
  },
  {
    "page_content": "NoveTech Solutions Future Outlook - Expected revenue for Q2 2025: $135M. New AI chatbot and blockchain security platform launch planned. Expansion into Latin America."
  },
  {
    "page_content": "NoveTech Solutions Market Performance - Year-over-Year growth at 12.7%. Stock price increase reflects investor confidence. Cybersecurity and AI sectors remain competitive."
  },
  {
    "page_content": "NoveTech Solutions Strategic Moves - Investing in R&D to enhance AI-driven automation. Strengthening partnerships with enterprise cloud providers. Focusing on data privacy solutions."
  },
  {
    "page_content": "NoveTech Solutions CEO Statement - 'NoveTech Solutions continues to innovate in AI and cybersecurity. Our growth strategy remains strong, and we foresee steady expansion in the coming quarters.'"
  }
]

Depois que os documentos forem indexados, você verá uma mensagem de exclusão no tópico, conforme mostrado abaixo.

4. Executando o grafo de recuperação

• Mude para o gráfico de recuperação.

• Digite a seguinte consulta de pesquisa:

What was NovaTech Solutions total revenue in Q1 2025?

O sistema retornará documentos relevantes e fornecerá uma resposta exata com base nos dados indexados.

Personalize o agente de recuperação.

Para melhorar a experiência do usuário, introduzimos uma etapa de personalização no grafo de recuperação para prever as próximas três perguntas que um usuário poderá fazer. Essa previsão se baseia em:

• Contexto dos documentos recuperados

• Interações anteriores do usuário

• Última consulta do usuário

As seguintes alterações de código são necessárias para implementar o recurso de Previsão de Consulta:

1. Atualize o arquivo graph.py

• Adicione a função predict_query :

async def predict_query(
   state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
   logger.info(f"predict_query predict_querypredict_query predict_query predict_query predict_query")  # Log the query

   configuration = Configuration.from_runnable_config(config)
   prompt = ChatPromptTemplate.from_messages(
       [
           ("system", configuration.predict_next_question_prompt),
           ("placeholder", "{messages}"),
       ]
   )
   model = load_chat_model(configuration.response_model)
   user_query = state.queries[-1] if state.queries else "No prior query available"
   logger.info(f"user_query: {user_query}")
   logger.info(f"statemessage: {state.messages}")
   #human_messages = [msg for msg in state.message if isinstance(msg, HumanMessage)]

   message_value = await prompt.ainvoke(
       {
           "messages": state.messages,
           "user_query": user_query,  # Use the most recent query as primary input
           "system_time": datetime.now(tz=timezone.utc).isoformat(),
       },
       config,
   )

   next_question = await model.ainvoke(message_value, config)
   return {"next_question": [next_question]}

• Modifique a função respond para retornar o objeto response , em vez da mensagem:

async def respond(
   state: State, *, config: RunnableConfig
) -> dict[str, list[BaseMessage]]:
   """Call the LLM powering our "agent"."""
   configuration = Configuration.from_runnable_config(config)
   # Feel free to customize the prompt, model, and other logic!
   prompt = ChatPromptTemplate.from_messages(
       [
           ("system", configuration.response_system_prompt),
           ("placeholder", "{messages}"),
       ]
   )
   model = load_chat_model(configuration.response_model)

   retrieved_docs = format_docs(state.retrieved_docs)
   message_value = await prompt.ainvoke(
       {
           "messages": state.messages,
           "retrieved_docs": retrieved_docs,
           "system_time": datetime.now(tz=timezone.utc).isoformat(),
       },
       config,
   )
   response = await model.ainvoke(message_value, config)
   # We return a list, because this will get added to the existing list
   return {"response": [response]}

• Atualizar estrutura do gráfico para adicionar novo nó e aresta para predict_query:

builder.add_node(generate_query)
builder.add_node(retrieve)
builder.add_node(respond)
builder.add_node(predict_query)
builder.add_edge("__start__", "generate_query")
builder.add_edge("generate_query", "retrieve")
builder.add_edge("retrieve", "respond")
builder.add_edge("respond", "predict_query")

2. Atualize o arquivo prompts.py

• Crie um prompt para previsão de guery em prompts.py:

PREDICT_NEXT_QUESTION_PROMPT = """Given the user query and the retrieved documents, suggest the most likely next question the user might ask.

**Context:**
- Previous Queries:
{previous_queries}

- Latest User Query: {user_query}

- Retrieved Documents:
{retrieved_docs}

**Guidelines:**
1. Do not suggest a question that has already been asked in previous queries.
2. Consider the retrieved documents when predicting the next logical question.
3. If the user's query is already fully answered, suggest a relevant follow-up question.
4. Keep the suggested question natural and conversational.
5. Suggest at least 3 question

System time: {system_time}"""

3. Atualize o arquivo configuration.py

• Adicionar predict_next_question_prompt:

predict_next_question_prompt: str = field(
       default=prompts.PREDICT_NEXT_QUESTION_PROMPT,
       metadata={"description": "The system prompt used for generating responses."},
   )

4. Atualize o arquivo state.py

• Adicione os seguintes atributos:

response: Annotated[Sequence[AnyMessage], add_messages]
next_question : Annotated[Sequence[AnyMessage], add_messages]

5. Execute novamente o grafo de recuperação.

• Digite a seguinte consulta de pesquisa novamente:

What was NovaTech Solutions total revenue in Q1 2025?

O sistema processará a entrada e preverá três perguntas relacionadas que os usuários podem fazer, conforme mostrado abaixo.

Conclusão

A integração do modelo do Retrieval Agent no LangGraph Studio e na CLI oferece vários benefícios importantes:

• Desenvolvimento acelerado: o modelo e as ferramentas de visualização simplificam a criação e a depuração de fluxos de trabalho de recuperação, reduzindo o tempo de desenvolvimento.

• Implantação perfeita: o suporte integrado para APIs e o dimensionamento automático garantem uma implantação tranquila em todos os ambientes.

• Atualizações fáceis: modificar fluxos de trabalho, adicionar novas funcionalidades e integrar nós adicionais é simples, facilitando o dimensionamento e o aprimoramento do processo de recuperação.

• Memória persistente: o sistema retém os estados e o conhecimento dos agentes, melhorando a consistência e a confiabilidade.

• Modelagem de fluxo de trabalho flexível: os desenvolvedores podem personalizar a lógica de recuperação e as regras de comunicação para casos de uso específicos.

• Interação e depuração em tempo real: a capacidade de interagir com agentes em execução permite testes eficientes e resolução de problemas.

Ao aproveitar esses recursos, as organizações podem criar sistemas de recuperação poderosos, eficientes e escaláveis que melhoram a acessibilidade dos dados e a experiência do usuário.

O código-fonte completo deste projeto está disponível no GitHub.