使用 LangGraph 和 Elasticsearch 构建强大的 RAG 工作流

作者：来自 Elastic  Neha Saini

在这篇博客中，我们将向你展示如何配置和自定义 LangGraph Retrieval Agent 模板与 Elasticsearch，以构建一个强大的 RAG 工作流，实现高效的数据检索和由 AI 驱动的响应。

Elasticsearch 原生集成了行业领先的生成式 AI 工具和提供商。查看我们关于超越 RAG 基础知识或构建可投入生产的应用程序 Elastic Vector Database 的网络研讨会。

要为你的用例构建最佳搜索解决方案，现在就开始免费云试用，或在本地机器上尝试 Elastic。

---

LangGraph Retrieval Agent Template 是由 LangChain 开发的入门项目，旨在通过 LangGraph Studio 使用 LangGraph 构建基于检索的问题解答系统。该模板预先配置，可与 Elasticsearch 无缝集成，使开发者能够快速构建能够高效索引和检索文档的智能体。

本博客重点介绍如何使用 LangGraph Studio 和 LangGraph CLI 运行并自定义 LangChain Retrieval Agent Template。该模板为构建检索增强生成（RAG）应用提供了框架，并支持多种检索后端，如 Elasticsearch。

我们将带你逐步完成设置、配置环境，以及如何高效结合 Elastic 执行该模板，并自定义智能体流程。

先决条件

在继续之前，请确保你已安装以下内容：

• Elasticsearch Cloud 部署或本地 Elasticsearch 部署（也可以在 Elastic Cloud 上创建一个 14 天免费试用）——版本需为 8.0.0 或更高

• Python 3.9 及以上版本

• 一个 LLM 提供商的访问权限，如 Cohere（本指南使用）、 OpenAI 或 Anthropic / Claude

创建 LangGraph 应用

1. 安装 LangGraph CLI

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

2. 从 retrieval-agent-template 创建 LangGraph 应用

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

你会看到一个交互式菜单，可以从中选择可用的模板列表。选择 4 表示 Retrieval Agent，选择 1 表示 Python，如下所示：

故障排除：如果遇到错误 “urllib.error.URLError: <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1000)>”

请运行 Python 的安装证书命令来解决此问题，如下所示。

3. 安装依赖项

在新建的 LangGraph 应用根目录下，创建一个虚拟环境并以 edit 模式安装依赖项，这样你的本地更改将被服务器使用：

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

设置环境

1. 创建一个 .env 文件

.env 文件保存 API 密钥和配置，以便应用可以连接到你选择的 LLM 和检索提供者。通过复制示例配置生成一个新的 .env 文件：

cp .env.example .env

2. 配置 .env 文件

.env 文件自带一组默认配置。你可以根据你的设置添加必要的 API 密钥和值来更新它。对于与你的使用场景无关的键，可以保持不变或者删除。

# 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

Elastic Cloud 和 Cohere 示例 .env 文件

下面是一个示例 .env 配置，使用 Elastic Cloud 作为检索提供者，Cohere 作为 LLM，正如本博客中演示的：

# 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

注意：虽然本指南使用 Cohere 进行响应生成和向量生成，但你也可以根据自己的用例选择其他 LLM 提供商，比如 OpenAI、Claude，或者本地的 LLM 模型。请确保你要使用的每个 key 都已经在 .env 文件中正确设置。

3. 更新配置文件 - configuration.py

在 .env 文件中设置好相应的 API key 后，下一步是更新应用程序的默认模型配置。更新配置可以确保系统使用你在 .env 文件中指定的服务和模型。

导航到配置文件：

 cd src/retrieval_graph

configuration.py 文件包含检索代理执行以下三项主要任务时使用的默认模型设置：

• 嵌入模型（Embedding model）—— 将文档转换为向量表示

• 查询模型（Query model）—— 将用户查询处理为向量

• 响应模型（Response model）—— 生成最终回答

默认情况下，代码使用的是 OpenAI（如 openai/text-embedding-3-small）和 Anthropic（如 anthropic/claude-3-5-sonnet-20240620 和 anthropic/claude-3-haiku-20240307）提供的模型。

在本博客中，我们切换为使用 Cohere 模型。如果你已经在使用 OpenAI 或 Anthropic，则无需更改。

使用 Cohere 的示例更改：

打开 configuration.py 并按如下方式修改默认模型设置：

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

使用 LangGraph CLI 运行检索代理

1. 启动 LangGraph 服务器

cd lg-agent-demo
langgraph dev

这将在本地启动 LangGraph API 服务器。如果运行成功，你应该会看到类似下面的内容：

打开 Studio UI 链接。

这里有两个可用的图表：

• Retrieval Graph：从 Elasticsearch 检索数据并使用 LLM 响应查询。

• Indexer Graph：将文档索引到 Elasticsearch 并使用 LLM 生成向量。

2. 配置 Indexer graph

• 打开 Indexer graph。
• 点击管理助手 (Manage Assistants)。
  • 点击 “添加新助手” (Add New Assistant)，输入指定的用户详情，然后关闭窗口。

{"user_id": "101"}

3. 索引示例文档

索引以下示例文档，这些文档代表组织 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.'"}
]

一旦文档被索引，你将在线程中看到一条删除消息，如下所示。

4. 运行检索图（retrieval graph）

切换到检索图。 输入以下搜索查询：

What was NovaTech Solutions total revenue in Q1 2025?

系统将返回相关文档，并根据已索引的数据提供准确的答案。

自定义检索代理

为了增强用户体验，我们在检索图中引入了一个自定义步骤，用于预测用户可能提出的下三个问题。这个预测基于以下内容：

• 从检索到的文档中的上下文

• 之前的用户交互

• 上一个用户查询

实现查询预测功能需要以下代码更改：

1. 更新 graph.py

添加 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 queryconfiguration = 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]}

修改 respond 函数以返回 response 对象，而不是消息：

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 listreturn {"response": [response]}

更新图结构以添加新的节点和边用于 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. 更新 prompts.py

为查询预测（Query Prediction）编写 prompt：

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 questionSystem time: {system_time}"""

3. 更新 configuration.py

添加 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. 新 state.py

添加以下属性：

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

5. 重新运行 Retrieval Graph

再次输入以下搜索查询：

What was NovaTech Solutions total revenue in Q1 2025?

系统将处理输入并预测用户可能会问的三个相关问题，如下所示。

结论

在 LangGraph Studio 和 CLI 中集成 Retrieval Agent 模板提供了几个关键优势：

• 加速开发：模板和可视化工具简化了检索工作流的创建和调试，减少了开发时间。
• 无缝部署：内置对 API 和自动扩展的支持，确保在不同环境中的顺利部署。
• 简易更新：修改工作流、添加新功能和集成额外节点变得简单，便于扩展和增强检索过程。
• 持久化记忆：系统保留代理的状态和知识，提升一致性和可靠性。
• 灵活的工作流建模：开发人员可以根据特定用例自定义检索逻辑和通信规则。
• 实时交互和调试：能够与运行中的代理进行交互，有效进行测试和问题解决。

通过利用这些功能，组织可以构建强大、高效且可扩展的检索系统，增强数据访问性和用户体验。

该项目的完整源代码可在 GitHub 上找到。

原文：Build a powerful RAG workflow using LangGraph and Elasticsearch - Elasticsearch Labs