​​Construcción de un flujo de trabajo RAG usando LangGraph y Elasticsearch

La plantilla del agente de recuperación de LangGraph es un proyecto inicial desarrollado por LangChain para facilitar la creación de sistemas de respuesta a preguntas basados en la recuperación empleando LangGraph en LangGraph Studio. Esta plantilla está preconfigurada para integrar perfectamente con Elasticsearch, lo que permite a los desarrolladores crear rápidamente agentes que puedan indexar y recuperar documentos de manera eficiente.

Este blog se centra en la ejecución y personalización de la plantilla del agente de recuperación de LangChain mediante LangGraph Studio y LangGraph CLI. La plantilla proporciona un marco para crear aplicaciones de generación aumentada de recuperación (RAG), aprovechando varios backends de recuperación como Elasticsearch.

Te guiaremos a través de la instalación, la configuración del entorno y la ejecución de la plantilla de manera eficiente con Elastic mientras personalizas el flujo del agente.

Prerrequisitos

Antes de continuar, cerciorar de tener instalado lo siguiente:

• Despliegue de Elasticsearch Cloud o despliegue de Elasticsearch local (o crea una prueba gratis de 14 días en Elastic Cloud) - Versión 8.0.0 o superior

• Python 3.9+

• Acceso a un proveedor de LLM como Cohere (empleado en esta guía), OpenAI o Anthropic/Claude

Creación de la aplicación LangGraph

1. Instalar la CLI de LangGraph

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

2. Crear la aplicación LangGraph a partir de retrieval-agent-template

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

Se le presentará un menú interactivo que le permitirá elegir entre una lista de plantillas disponibles. Seleccione 4 para el Agente de recuperación y 1 para Python, como se muestra a continuación:

• Solución de problemas: Si encuentra el error "urllib.error.URLError: error <urlopen [SSL: CERTIFICATE_VERIFY_FAILED] error de verificación del certificado: no se puede obtener el certificado del emisor local (_ssl.c:1000)> “

Ejecute el comando Instalar certificado de Python para resolver el problema, como se muestra a continuación.

3. Instalar dependencias

En la raíz de su nueva aplicación LangGraph, cree un entorno virtual e instale las dependencias en modo edit para que el servidor emplee sus cambios locales:

#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 .

Configuración del entorno

1. Crear un entorno .. archivo

El archivo .env contiene claves y configuraciones de API para que la aplicación pueda conectarse al proveedor de recuperación y LLM elegido. Genere un nuevo archivo .env duplicando la configuración de ejemplo:

cp .env.example .env

2. Configurar el .env archivo

El archivo .env viene con un conjunto de configuraciones predeterminadas. Puede actualizarlo agregando las claves y los valores de API necesarios según su configuración. Las claves que no sean relevantes para tu caso de uso se pueden dejar sin cambios o quitar.

# 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

• Ejemplo .env archivo (con Elastic Cloud y Cohere)

A continuación, se muestra un ejemplo de configuración .env para usar Elastic Cloud como proveedor de recuperación y Cohere como LLM, como se muestra en este 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

Nota: Si bien esta guía usa Cohere tanto para la generación de respuestas como para las incrustaciones, puede usar otros proveedores de LLM como OpenAI, Claudeo incluso un modelo de LLM local, según su caso de uso. Cerciorar de que cada tecla que desea emplear esté presente y configurada correctamente en el archivo.env.

3. Actualizar archivo de configuración -configuration.py

Luego de configurar tu archivo .env con las claves de API adecuadas, el siguiente paso es actualizar la configuración del modelo predeterminado de tu aplicación. La actualización de la configuración garantiza que el sistema use los servicios y modelos que especificó en el archivo .env .

Vaya al archivo de configuración:

 cd src/retrieval_graph

El archivo configuration.py contiene la configuración predeterminada del modelo empleada por el agente de recuperación para tres tareas principales:

• Modelo de incrustación : convierte documentos en representaciones vectoriales

• Modelo de consulta : procesa la consulta del usuario en un vector

• Modelo de respuesta : genera la respuesta final

De forma predeterminada, el código emplea modelos de OpenAI (por ejemplo, openai/text-embedding-3-small) y Anthropic (por ejemplo, anthropic/claude-3-5-sonnet-20240620 and anthropic/claude-3-haiku-20240307). En este blog, estamos cambiando al uso de modelos Cohere. Si ya está empleando OpenAI o Anthropic, no se necesitan cambios.

Ejemplos de cambios (usando Cohere):

Abra configuration.py y modifique los valores predeterminados del modelo como se muestra a continuación:

…
 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={

Ejecutando el agente de recuperación con la CLI de LangGraph

1. Inicie el servidor LangGraph

cd lg-agent-demo
langgraph dev

Esto iniciará el servidor de la API de LangGraph localmente. Si esto se ejecuta correctamente, debería ver algo como:

Abra la URL de la interfaz de usuario de Studio.

Hay dos gráficos disponibles:

• Gráfico de recuperación: Recupera datos de Elasticsearch y responde a la consulta usando un LLM.

• Gráfico indexador: Indexa documentos en Elasticsearch y genera incrustaciones usando un LLM.

2. Configuración del grafo indexador

• Abre el gráfico del indexador.

• Haz clic en gestionar asistentes.

  • Haz clic en 'Agregar nuevo asistente', introduce los datos del usuario según lo especificado y luego cierra la ventana.

{"user_id": "101"}

3. Indexación de documentos de muestra

• Indexe los siguientes documentos de muestra, que representan un reporte trimestral hipotético para la organización 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.'"
  }
]

Una vez indexados los documentos, verá un mensaje de eliminación en el hilo, como se muestra a continuación.

4. Ejecutar el grafo de recuperación

• Cambia al gráfico de recuperación.

• Introduzca la siguiente consulta de búsqueda:

What was NovaTech Solutions total revenue in Q1 2025?

El sistema devolverá los documentos relevantes y proporcionará una respuesta exacta basada en los datos indexados.

Personalizar el agente de recuperación

Para mejorar la experiencia del usuario, introducimos un paso de personalización en el gráfico de recuperación para predecir las siguientes tres preguntas que un usuario podría hacer. Esta predicción se basa en:

• Contexto de los documentos recuperados

• Interacciones anteriores de los usuarios

• Última consulta de usuario

Se requieren los siguientes cambios de código para implementar la función de predicción de consultas:

1. Actualización graph.py

• Agregue predict_query función:

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 respond función para devolver response Object , en lugar de message:

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]}

• Actualice la estructura del gráfico para agregar un nuevo nodo y borde 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. Actualización prompts.py

• Prompt de creación para predicción de guery en 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. Actualización configuration.py

• Agregar 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. Actualización state.py

• Agregue los siguientes atributos:

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

5. Volver a ejecutar el grafo de recuperación

• Vuelva a introducir la siguiente consulta de búsqueda:

What was NovaTech Solutions total revenue in Q1 2025?

El sistema procesará la entrada y predecirá tres preguntas relacionadas que los usuarios podrían hacer, como se muestra a continuación.

Conclusión

La integración de la plantilla del agente de recuperación dentro de LangGraph Studio y CLI proporciona varios beneficios clave:

• Desarrollo acelerado: las herramientas de plantilla y visualización agilizan la creación y depuración de flujos de trabajo de recuperación, lo que reduce el tiempo de desarrollo.

• Implementación perfecta: la compatibilidad integrada con las API y el escalado automático garantiza una implementación fluida en todos los entornos.

• Actualizaciones fáciles: Modificar los flujos de trabajo, agregar nuevas funcionalidades e integrar nodos adicionales es simple, lo que facilita escalar y mejorar el proceso de recuperación.

• Memoria persistente: el sistema conserva los estados y el conocimiento de los agentes, lo que mejora la coherencia y la confiabilidad.

• Modelado de flujo de trabajo flexible: los desarrolladores pueden personalizar la lógica de recuperación y las reglas de comunicación para casos de uso específicos.

• Interacción y depuración en tiempo real: la capacidad de interactuar con los agentes en ejecución permite realizar pruebas y resolver problemas de manera eficiente.

Al aprovechar estas características, las organizaciones pueden crear sistemas de recuperación poderosas, eficientes y escalables que mejoren la accesibilidad de los datos y la experiencia del usuario.

El código fuente completo de este proyecto está disponible en GitHub.