使用 LangChain 和 Elasticsearch 开发代理 RAG 助手
了解如何使用 LangChain 和 Elasticsearch 构建一个代理式抹布新闻助手，通过自适应路由回答有关文章的查询。
本博文将深入探讨代理 RAG 工作流，解释其主要特点和常见设计模式。它通过一个使用 Elasticsearch 作为向量存储和 LangChain 构建代理 RAG 框架的实践示例，进一步演示了如何实施这些工作流程。最后，文章简要讨论了与设计和实施此类架构相关的最佳实践和挑战。您可以使用此Jupyter 笔记本创建一个简单的代理 RAG 管道。
代理 RAG 简介
检索增强生成（RAG）已成为基于 LLM 的应用的基石，它使模型能够根据用户查询检索相关上下文，从而提供最佳答案。RAG 系统通过从应用程序接口或数据存储中获取外部信息，而不是局限于预先训练的 LLM 知识，从而提高了 LLM 响应的准确性和上下文。另一方面，人工智能代理可自主运行，为实现指定目标做出决策并采取行动。
代理 RAG 是一个将检索增强生成和代理推理的优势结合在一起的框架。它将 RAG 集成到代理的决策过程中，使系统能够动态地选择数据源，改进查询以获得更好的上下文检索，生成更准确的响应，并应用反馈循环来不断提高输出质量。
代理 RAG 的主要特点
代理 RAG 框架标志着传统 RAG 系统的重大进步。它不再遵循固定的检索流程，而是利用能够实时规划、执行和优化结果的动态代理。
让我们来看看代理 RAG 管道的一些主要特点：
动态决策：Agentic RAG 使用推理机制来理解用户的意图，并将每个查询路由到最相关的数据源，从而生成准确且能感知上下文的响应。
全面的查询分析：Agentic RAG 深入分析用户查询，包括子问题及其总体意图。它能评估查询的复杂性，并动态选择最相关的数据源来检索信息，确保准确和完整的响应。
多阶段协作：该框架通过专业代理网络实现多阶段协作。每个代理处理更大目标中的特定部分，依次或同时工作，以实现协调一致的结果。
自我评估机制：代理式 RAG 管道利用自我反思来评估检索到的文档和生成的回复。它可以检查检索到的信息是否完全符合查询要求，然后审查输出信息的准确性、完整性和事实一致性。
与外部工具集成：该工作流程可与外部应用程序接口、数据库和实时信息源交互，纳入最新信息并动态适应不断变化的数据。
代理 RAG 的工作流程模式
工作流模式定义了代理人工智能如何以可靠、高效的方式构建、管理和协调基于 LLM 的应用程序。一些框架和平台，如 LangChain 、 LangGraph 、 CrewAI 和 LlamaIndex ，可用于实现这些代理工作流。
顺序检索链：顺序工作流将复杂的任务划分为简单、有序的步骤。每一步都会改进下一步的输入，从而取得更好的结果。例如，在创建客户档案时，一名代理可能会从客户关系管理中获取基本信息，另一名代理可能会从交易数据库中检索购买历史记录，最后一名代理可能会将这些信息结合起来，生成一份完整的客户档案，用于推荐或报告。
路由检索链：在这种工作流程模式中，路由器代理分析输入，并将其导向最合适的流程或数据源。当存在多个不同的数据源且重叠程度极低时，这种方法尤为有效。例如，在客户服务系统中，路由器代理会对收到的请求（如技术问题、退款或投诉）进行分类，并将其路由到相应的部门进行有效处理。
并行检索链：在这种工作流程模式中，多个独立的子任务同时执行，然后将其输出汇总，生成最终响应。这种方法大大缩短了处理时间，提高了工作流程效率。例如，在客户服务并行工作流程中，一名代理检索过去的类似请求，另一名则查阅相关的知识库文章。然后，聚合器将这些输出合并起来，生成一份综合决议。
Orchestrator 工作链：这种工作流程与并行化有相似之处，因为它利用了独立的子任务。然而，一个关键的区别在于集成了一个协调代理。该代理负责分析用户查询，在运行期间将查询动态地划分为子任务，并确定制定准确回复所需的适当流程或工具。
从零开始建立代理 RAG 管道
为了说明代理 RAG 的原理，让我们使用 LangChain 和 Elasticsearch 设计一个工作流程。该工作流程采用基于路由的架构，多个代理协作分析查询、检索相关信息、评估结果并生成一致的回复。您可以参考这个Jupyter 笔记本来学习这个示例。
工作流程从路由器代理开始，路由器代理分析用户的查询，选择最佳检索方法，即vectorstore 、websearch 或composite 方法。矢量存储处理传统的基于 RAG 的文档检索，网络搜索获取未存储在矢量存储中的最新信息，而复合方法则在需要来自多个来源的信息时将两者结合起来。
如果文件被认为合适，摘要代理就会生成清晰且与上下文相符的回复。但是，如果文档不足或不相关，查询重写代理就会重新制定查询，以改进搜索。修改后的查询会重新启动路由过程，使系统能够改进搜索并提高最终输出结果。
准备工作
该工作流程依靠以下核心组件来有效执行示例：
Python 3.10
Azure OpenAI
Elasticsearch
LangChain
在继续之前，系统会提示您配置本例所需的以下环境变量。
AZURE_OPENAI_ENDPOINT="Add your azure openai endpoint"
AZURE_OPENAI_KEY="Add your azure openai key"
AZURE_OPENAI_DEPLOYMENT="gpt-4.1"
AZURE_OPENAI_API_VERSION="Add your azure openai api version"
ES_ENDPOINT = "Add your Elasticsearch ENDPOINT"
ES_API_KEY = "Add your Elasticsearch API KEY"数据来源
本工作流程使用 AG 新闻数据集的一个子集进行说明。数据集包含不同类别的新闻文章，如国际、体育、商业和科学/技术。
dataset = load_dataset("ag_news", split="train[:1000]")
docs = [
Document(
page_content=sample["text"],
metadata={"category": sample["label"]}
)
for sample in dataset
]从langchain_elasticsearch 开始使用ElasticsearchStore 模块作为我们的向量存储。在检索方面，我们采用 Elastic 专有的嵌入模型ELSER，实施 SparseVectorStrategy。在启动向量存储之前，必须确认 ELSER 模型已正确安装并部署到 Elasticsearch 环境中。
elastic_vectorstore = ElasticsearchStore.from_documents(
docs,
es_url=ES_ENDPOINT,
es_api_key=ES_API_KEY,
index_name=index_name,
strategy=SparseVectorStrategy(model_id=".elser_model_2"),
)
elastic_vectorstore.client.indices.refresh(index=index_name)网络搜索功能是利用 LangChain 社区工具中的DuckDuckGoSearchRun实现的，它能让系统高效地从网上检索实时信息。您还可以考虑使用其他搜索 API，它们可能会提供更相关的结果。之所以选择该工具，是因为它无需 API 密钥即可进行搜索。
duckduckgo = DuckDuckGoSearchRun(description= "A custom DuckDuckGo search tool for finding latest news stories.", verbose=True)
def websearch_retriever(query):
results = duckduckgo.run(f"{query}")
return results复合检索器专为需要结合多种来源的查询而设计。它通过同时检索网络上的实时数据和查询矢量存储中的历史新闻，提供全面、准确的响应。
def composite_retriever(query):
related_docs = vectorstore_retriever(query)
related_docs += websearch_retriever(query)
return related_docs设置代理
下一步，将定义 LLM 代理，以便在该工作流程中提供推理和决策能力。我们将创建的 LLM 链包括router_chain,grade_docs_chain,rewrite_query_chain, 和summary_chain 。
路由器代理使用 LLM 助手，在运行时为给定查询确定最合适的数据源。分级代理对检索到的文档进行相关性评估。如果文件被认为是相关的，它们就会被传递给摘要代理，以生成摘要。否则，重写查询代理会重新制定查询，并将其发送回路由过程，进行另一次检索尝试。您可以在笔记本的 LLM chains 部分找到所有代理的说明。
class RouteQuery(BaseModel):
datasource: Literal["vectorstore", "websearch", "composite"] = Field(
...,
description="Choose to route the query to web search, vectorstore or composite."
)
router_prompt = ChatPromptTemplate.from_template("""You are an assistant that decides the best data source for questions based on news articles.
Choose one of the following options:
- 'vectorstore': for general, background, or historical news articles.
- 'websearch': for recent discoveries, 'latest', 'current', or '2025' type queries.
- 'composite': when the question needs both historical and current knowledge on news articles.
Question: {query}
Return one word: 'vectorstore', 'websearch', or 'composite'.
""")
router_structured = llm.with_structured_output(RouteQuery)
router_chain: RunnableSequence = router_prompt | router_structuredllm.with_structured_output 约束模型的输出，使其遵循RouteQuery 类下 BaseModel 定义的预定义模式，确保结果的一致性。第二行通过连接router_prompt 和router_structured 来组成RunnableSequence ，形成一个流水线，在这个流水线中，语言模型对输入提示进行处理，产生结构化的、符合模式的结果。
定义图形节点
这部分包括定义图形的状态，这些状态代表系统不同组件之间流动的数据。对这些状态的明确说明可确保工作流程中的每个节点都知道自己可以访问和更新哪些信息。
class RAGState(TypedDict):
query: str
docs: List[Document]
router: str
summary: str
self_reflection: bool
retry_count: int = 0一旦定义了状态，下一步就是定义图的节点。节点就像图中的功能单元，可对数据执行特定操作。我们的管道中有 7 个不同的节点。
def router(state: RAGState):
router = router_chain.invoke({'query': state["query"]})
logger.info(f"Router selected the datasource: {router.datasource}")
logger.info(f"User query: {state['query']}")
return {"router": router.datasource}
def vectorstore(state: RAGState):
return {"docs": vectorstore_retriever(state["query"])}
def websearch(state: RAGState):
return {"docs": websearch_retriever(state["query"])}
def composite(state: RAGState):
return {"docs": composite_retriever(state["query"])}
def self_reflection(state: RAGState):
evaluation = grade_docs_chain.invoke(
{"query": state["query"], "docs": state["docs"]}
)
if evaluation.binary_score:
logger.info(f"Self-reflection passed -- binary_score={evaluation.binary_score}")
else:
logger.info(f"Self-reflection failed -- binary_score={evaluation.binary_score}")
return {
"self_reflection": evaluation.binary_score,
}
def query_rewriter(state: RAGState):
retry_count = state.get("retry_count", 0) + 1
new_query = rewrite_query_chain.invoke({"query": state["query"]})
logger.info(f"Query rewritten: {new_query}, retry_count: {retry_count}")
return {
"query": new_query,
"retry_count": retry_count,
}
def summarize(state: RAGState):
summary = summarize_chain.run(
query=state["query"],
docs=state["docs"],
)
return {"summary": summary}query_rewriter 节点在工作流程中有两个作用。首先，当自我反思代理评估的文档被认为不充分或不相关时，它会使用rewrite_query_chain 重写用户查询，以改进检索。其次，它还可以作为一个计数器，跟踪查询被重写的次数。
每次调用节点时，都会递增存储在工作流状态中的retry_count 。这种机制可防止工作流程进入无限循环。如果retry_count 超过预定义的阈值，系统就会退回到错误状态、默认响应或您选择的任何其他预定义条件。
编制图表
最后一步是定义图的边，并在编译前添加必要的条件。每个图都必须从指定的起始节点开始，作为工作流程的入口点。图中的边代表节点之间的数据流，有两种类型：
直边：它们定义了从一个节点到另一个节点的直接、无条件的流动。每当第一个节点完成任务后，工作流程就会自动沿直线进入下一个节点。
条件边：这些边允许工作流根据节点的当前状态或计算结果进行分支。下一个节点根据评估结果、路由决定或重试次数等条件动态选择。
graph.add_edge(START, "router")
def after_router(state: RAGState):
route = state.get("router", None)
if route == "vectorstore":
return "vectorstore"
elif route == "websearch":
return "websearch"
else:
return "composite"
def after_self_reflection(state: RAGState):
if state["self_reflection"]:
return "summarize"
return "query_rewriter"
def after_query_rewriter(state: RAGState):
while state['retry_count'] <= 3:
return "router"
raise RuntimeError("Maximum retries (3) reached -- evaluation failed.")
graph.add_conditional_edges(
"router",
after_router,
{
"vectorstore": "vectorstore",
"websearch": "websearch",
"composite": "composite"
}
)
graph.add_edge("vectorstore", "self_reflection")
graph.add_edge("websearch", "self_reflection")
graph.add_edge("composite", "self_reflection")
graph.add_conditional_edges(
"self_reflection",
after_self_reflection,
{
"summarize": "summarize",
"query_rewriter": "query_rewriter"
}
)
graph.add_conditional_edges("query_rewriter", after_query_rewriter, {"router": "router"})
graph.add_edge("summarize", END)
agent=graph.compile()这样，第一个代理 RAG 管道就准备就绪，可以使用编译后的代理进行测试了。
result = agent.invoke({"query": query1})
logger.info(f"\nFinal Summary:\n: {result['summary']}")测试代理 RAG 管道
现在，我们将使用以下三种不同类型的查询对该管道进行测试。请注意，结果可能各不相同，下面的例子只是说明了一种可能的结果。
query1="What are the latest AI models released this month?"
query2="What technological innovations are discussed in Sci/Tech news?"
query3="Compare a Sci/Tech article from the dataset with a current web article about AI trends."对于第一次查询，路由器选择websearch 作为数据源。如输出所示，该查询未通过自我反省评估，随后被重定向到查询重写阶段。
INFO | __main__:router:11 - Router selected the datasource: websearch
INFO | __main__:router:12 - User query: What are the latest AI models released this month?
Latest Singapore news, including the city state's relationships with Malaysia and Mahathir, China and Xi Jinping, and the rest of Southeast Asia. 3 days ago · The latest military news, insights and analysis from China. All the latest news, opinions and analysis on Hong Kong, China, Asia and around the world Latest news, in-depth features and opinion on Malaysia, covering politics, economy, society and the Asean member-nation's relationships with China, Singapore, and other Southeast Asian ... Oct 12, 2025 · Brics (an acronym for Brazil, Russia, India, China and South Africa) refers to an association of 10 leading emerging markets. The other member states are Egypt, Ethiopia, ...
INFO | __main__:self_reflection:31 - Self-reflection failed -- binary_score=False
INFO | __main__:query_rewriter:40 - Query rewritten: query='Which AI models have been officially released in June 2024?', retry_count: 1
INFO | __main__:router:11 - Router selected the datasource: websearch
INFO | __main__:router:12 - User query: query='Which AI models have been officially released in June 2024?'
Dream Machine is a text-to-video model created by Luma Labs and launched in June 2024 . It generates video output based on user prompts or still images. Dream Machine has been noted for its ability to realistically capture motion... Released in June 2023. In June 2024 , Baidu announced Ernie 4.0 Turbo. In April 2025, Ernie 4.5 Turbo and X1 Turbo were released . These models are optimized for faster response times and lower operational costs.[28][29]. The meaning of QUERY is question, inquiry. How to use query in a sentence. Synonym Discussion of Query. QUERY definition: 1. a question, often expressing doubt about something or looking for an answer from an authority.... Learn more. Query definition: a question; an inquiry.. See examples of QUERY used in a sentence.
INFO | __main__:self_reflection:29 - Self-reflection passed -- binary_score=True
INFO | __main__:<module>:2 -
Final Summary:
: In June 2024, two AI models were officially released: Dream Machine, a text-to-video model launched by Luma Labs, and Ernie 4.0 Turbo, announced by Baidu, which is optimized for faster response times and lower operational costs.接下来，我们以第二个查询为例，对使用vectorstore 检索的示例进行研究。
INFO | __main__:router:11 - Router selected the datasource: vectorstore
INFO | __main__:router:12 - User query: What technological innovations are discussed in Sci/Tech news?
INFO | __main__:self_reflection:29 - Self-reflection passed -- binary_score=True
INFO | __main__:<module>:2 -
Final Summary:
: Recent Sci/Tech news highlights several technological innovations: NASA is collaborating with Silicon Valley firms to build a powerful Linux-based supercomputer to support theoretical research and shuttle engineering; new chromatin transfer techniques have enabled the cloning of cats; cybersecurity advancements are being discussed in relation to protecting personal technology; Princeton University scientists assert that existing technologies can be used immediately to stabilize global warming; and a set of GameBoy micro-games has been recognized for innovation in game design.最后的查询被导向复合检索，它同时利用了矢量存储和网络搜索。
INFO | __main__:router:11 - Router selected the datasource: composite
INFO | __main__:router:12 - User query: Compare a Sci/Tech article from the dataset with a current web article about AI trends.
Atlas currently only available on macOS, built on Chromium with planned features like ad-blocking still in development. OpenAI's Atlas browser launched with bold promises of AI -powered web browsing, but early real-world testing reveals a different story. Career-long data are updated to end-of-2024 and single recent year data pertain to citations received during calendar year 2024. The selection is based on the top 100,000 scientists by c-score (with and without self-citations) or a percentile rank of 2% or above in the sub-field. In this article I list 45 AI tools across 21 different categories. After exploring all the available options in each category, I've carefully selected the best tools based on my personal experience. Reading a complex technical article ? Simply highlight confusing terminology and ask "what's this?" to receive instant explanations. compare browsers. Comparison showing traditional browser navigation versus OpenAI Atlas AI -powered workflows. After putting Gemini, ChatGPT, Grok, and DeepSeek through rigorous testing in October 2025, it's clear that there isn't one AI that reigns supreme across all categories.
INFO | __main__:self_reflection:29 - Self-reflection passed -- binary_score=True
INFO | __main__:<module>:2 -
Final Summary:
: A Sci/Tech article from the dataset highlights NASA's development of robust artificial intelligence software for planetary rovers, aiming to make them more self-reliant and capable of decision-making during missions. In contrast, a current web article about AI trends focuses on the proliferation of AI-powered tools across various categories, including browsers like OpenAI Atlas, and compares leading models such as Gemini, ChatGPT, Grok, and DeepSeek, noting that no single AI currently excels in all areas. While the NASA article emphasizes specialized AI applications for autonomous robotics in space exploration, the current trends article showcases the broadening impact of AI across consumer and professional technologies, with ongoing competition and rapid innovation among major AI platforms.在上述工作流程中，代理 RAG 可以在检索用户查询的信息时智能地确定使用哪个数据源，从而提高响应的准确性和相关性。您可以创建更多示例来测试代理，并查看输出结果是否产生了任何有趣的结果。
构建代理 RAG 工作流程的最佳实践
既然我们已经了解了代理 RAG 的工作原理，那么让我们来看看构建这些工作流程的一些最佳实践。遵循这些准则将有助于保持系统的效率和易于维护。
做好后备准备：提前规划后备策略，以应对工作流程中任何步骤出现故障的情况。这可能包括返回默认答案、触发错误状态或使用替代工具。这可确保系统从容应对故障，而不会破坏整体工作流程。
实施全面的日志记录：尝试在工作流程的每个阶段实施日志记录，如重试、生成输出、路由选择和查询重写。这些日志有助于提高透明度，方便调试，并有助于随着时间的推移完善提示、代理行为和检索策略。
选择合适的工作流程模式：检查您的使用案例，选择最适合您需求的工作流程模式。使用顺序工作流进行逐步推理，使用并行工作流处理独立数据源，使用协调器-工作器模式处理多工具或复杂查询。
纳入评估战略：在工作流程的不同阶段纳入评估机制。这可以包括自我反思代理、对检索到的文件进行分级或自动质量检查。评估有助于验证检索到的文件是否相关、响应是否准确，以及复杂查询的所有部分是否都得到了处理。
挑战
虽然代理 RAG 系统在适应性、精确性和动态推理方面具有显著优势，但它们在设计和实施阶段也面临着一些必须解决的挑战。一些主要挑战包括
复杂的工作流程：随着代理和决策点的增加，整个工作流程会变得越来越复杂。这可能导致运行时出现错误或故障的几率增加。在可能的情况下，消除多余的代理和不必要的决策点，优先简化工作流程。
可扩展性：要扩展代理 RAG 系统以处理大型数据集和高查询量，可能具有挑战性。采用高效的索引、缓存和分布式处理策略，以保持大规模性能。
协调和计算开销：使用多个代理执行工作流需要高级协调。这包括谨慎的调度、依赖管理和代理协调，以防止出现瓶颈和冲突，所有这些都会增加整个系统的复杂性。
评估的复杂性：对这些工作流程进行评估本身就存在挑战，因为每个阶段都需要不同的评估策略。例如，RAG 阶段必须对检索文件的相关性和完整性进行评估，而生成的摘要则需要检查其质量和准确性。同样，查询重写的有效性也需要一个单独的评估逻辑，以确定重写后的查询是否改善了检索结果。
结论
在这篇博文中，我们介绍了代理 RAG 的概念，并强调了它如何通过结合代理人工智能的自主能力来增强传统的 RAG 框架。我们探索了代理 RAG 的核心功能，并通过一个实践案例演示了这些功能，即使用 Elasticsearch 作为向量存储和 LangChain 创建代理框架来构建一个新闻助手。
此外，我们还讨论了在设计和实施代理 RAG 管道时需要考虑的最佳实践和主要挑战。这些见解旨在指导开发人员创建稳健、可扩展和高效的代理系统，将检索、推理和决策有效地结合起来。
未来发展
我们建立的工作流程非常简单，为改进和实验留下了足够的空间。我们可以通过尝试各种嵌入模型和改进检索策略来加强这一点。此外，集成一个重新排序代理来确定检索文件的优先次序也是有益的。另一个探索领域涉及为代理框架制定评估战略，特别是确定适用于不同类型框架的通用和可重复使用的方法。最后，在大型和更复杂的数据集上试验这些框架。
与此同时，如果您也有类似的实验，欢迎与我们分享！欢迎提供反馈意见，或通过我们的社区 Slack 频道或论坛与我们联系。
