一般最简单的RAG 做法是加载文档，构建 LlamaIndex
VectorStoreIndex
，把
QueryEngine
包装成
@tool
，再交给 LangGraph Agent。这个方案本身没有问题，在它对应的场景里也足够可靠。
但是问题在于“RAG” 现在已经不是单一架构的名字。生产系统里常见的是一组差异很大的检索架构，各自处理不同类型的问题。选错架构损失的不只是性能；系统还可能稳定地返回看似有把握、实际上错误的答案，而且这种错误会随着规模一起放大。
下面把这组架构放在一起梳理：看到一个检索问题时，能判断它更适合哪一种架构，并知道如何用 LangGraph 和 LlamaIndex 落地。
顺序从简单到复杂：
这 5 种架构并不互斥，它们是一层层往上叠的能力：问题越复杂，需要加上的层就越多。实际生产系统通常至少会组合其中两种。
这 5 种架构在解决同一个底层问题
怎样在模型需要知识的时候，把训练阶段没有见过的内容，以适合推理的形式送到它面前？
训练数据有截止时间，公司的内部文档、产品规格，以及模型冻结之后出现的内容都不在其中。把这些数据拿去 Fine-tuning，成本高、周期长；文档一变，模型也不会跟着自动更新。
RAG 选择在查询时解决知识更新。相关内容先被检索出来，再作为上下文放进 Prompt。模型不必记住它，只要在作答时能读到即可。
差别就落在“怎么检索”上。下面 5 种架构给出了 5 种不同回答，各自对应一类不同的检索难题。
架构 1：Naive RAG
Naive RAG 是最基础的一层，内部政策机器人、FAQ 助手、文档搜索、入职辅助这类需求，用它往往就够了。“naive” 在这里不等于不成熟；这套模式已经相当稳定，也长期用于大规模生产环境。
它的 Pipeline 一共 5 步，几乎可以直接对应到 LlamaIndex 已经提供的组件：
阶段 1–4：索引时间（只运行一次）阶段 5：查询时间（每个用户问题都运行）
索引时间和查询时间必须分开看。Embedding 的成本在索引阶段一次性支付；用户真正发起查询时，主要只剩一次相似度搜索和一次 LLM 调用。
用户问题和文档分块会进入同一个向量空间。查询发生时，系统返回几何距离上最接近问题向量的那些分块，前提是“向量空间中的语义相似”大体等价于“内容相关”。语料干净、事实性强时，这个前提通常成立，所以 Naive RAG 才能在不少简单场景里表现得很稳。
代码
# ============================================================
# NAIVE RAG——完整模板
# ============================================================
# ── 模块 1：导入与配置 ───────────────────────────────────
importos
fromtypingimportLiteral
fromlangchain_openaiimportChatOpenAI
fromlangchain_core.messagesimportHumanMessage, SystemMessage
fromlangchain_core.toolsimporttool
fromlanggraph.graphimportStateGraph, MessagesState, START, END
fromlanggraph.prebuiltimportToolNode
fromlanggraph.checkpoint.memoryimportMemorySaver
fromllama_index.coreimport (
Settings, SimpleDirectoryReader, VectorStoreIndex,
StorageContext, load_index_from_storage
)
fromllama_index.llms.openaiimportOpenAIasLlamaOpenAI
fromllama_index.embeddings.openaiimportOpenAIEmbedding
# LangGraph 的推理模型
llm=ChatOpenAI(model="gpt-4o", temperature=0)
# LlamaIndex 的内部模型——与上面的模型分开
# 这两个模型配置不会共享状态或对话历史
Settings.llm=LlamaOpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model=OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size=512
Settings.chunk_overlap=50
# ── 模块 2：状态 ──────────────────────────────────────────
classState(MessagesState):
pass # 继承 messages 列表；如有需要可继续扩展
# ── 模块 3：知识库（构建一次、持久化、重新加载） ───────
PERSIST_DIR="./storage/naive"
ifos.path.exists(PERSIST_DIR):
storage_context=StorageContext.from_defaults(persist_dir=PERSIST_DIR)
index=load_index_from_storage(storage_context)
else:
documents=SimpleDirectoryReader("./data").load_data()
index=VectorStoreIndex.from_documents(documents, show_progress=True)
index.storage_context.persist(persist_dir=PERSIST_DIR)
query_engine=index.as_query_engine(similarity_top_k=3)
# ── 模块 3（续）：工具 ──────────────────────────────────
@tool
defsearch_knowledge_base(query: str) ->str:
"""Search internal company documents for policies, product specs,
and procedures. Use this for any question requiring domain-specific
knowledge rather than general world knowledge."""
response=query_engine.query(query)
returnstr(response)
tools= [search_knowledge_base]
llm_with_tools=llm.bind_tools(tools)
tool_node=ToolNode(tools)
# ── 模块 4：节点 ──────────────────────────────────────────
defagent_node(state: State) ->dict:
system_prompt=SystemMessage(content=(
"You are a helpful assistant with access to an internal knowledge base. "
"Use search_knowledge_base for company-specific questions. "
"Answer general questions directly."
))
response=llm_with_tools.invoke([system_prompt] +state["messages"])
return {"messages": [response]}
# ── 模块 5：路由 ──────────────────────────────────────────
defshould_continue(state: State) ->Literal["tools", "__end__"]:
last=state["messages"][-1]
ifhasattr(last, "tool_calls") andlast.tool_calls:
return"tools"
return"__end__"
# ── 模块 6：图装配 ─────────────────────────────────────────
builder=StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue,
{"tools": "tools", "__end__": END})
builder.add_edge("tools", "agent")
graph=builder.compile(checkpointer=MemorySaver())
Naive RAG 会在哪些地方出问题
一类是术语不一致。用户问“一级客户的 SLA 是什么？”，文档写的却是“Gold-tier 客户保证 99.9% 的正常运行时间承诺”。SLA、tier-1、Gold-tier 彼此相关，却不是同一个词。只靠向量相似度，目标分块可能排不到足够靠前的位置，答案也就跟着漏掉。
另一类是关系型问题，例如“如果 Supplier X 下线，我们的哪些产品会受到影响？”。答案散落在多份文档里，需要沿着关系链逐步连接。单个分块无法回答，而 Naive RAG 只是把不同文档里的分块取回来，本身没有机制把这些关系串起来。
Hybrid RAG 和 Graph RAG 分别处理这两个问题。
架构 2：Hybrid RAG
生产实践里，语义相似度和精确关键词匹配更像两路互补信号。只用其中一路，都会留下明显盲区。Hybrid RAG 就是把两路检索放到同一条链路里。
稠密检索（Dense Retrieval，也就是向量搜索）擅长找语义等价内容。即使文档写的是 “Gold-tier 正常运行时间承诺”，查询 “SLA” 也有机会命中。碰到专有名词、产品代码、医学术语、法律引用这类必须精确匹配的内容，它的优势会迅速变小。
稀疏检索（Sparse Retrieval，也就是 BM25/关键词搜索）刚好补上这一块。搜索 “SKU-4829”，它会直接命中 “SKU-4829”；代价是它不理解语义等价，查询 “SLA” 时并不会自动联想到 “正常运行时间保证”。
Hybrid RAG 会并行跑两路检索，把候选结果合并，再交给重排序器（reranker）得到最终排名。
真正决定结果质量的是重排序器。Embedding 模型对查询和分块分别编码；交叉编码器（cross-encoder）则把查询与每个候选分块放在一起读取，直接判断二者的相关性。速度确实比相似度搜索慢，但它只处理已经收窄过的候选集，延迟通常还能接受。
代码
# ============================================================
# HYBRID RAG——LLAMAINDEX 实现
# pip install llama-index-retrievers-bm25 rank-bm25
# ============================================================
fromllama_index.coreimportSimpleDirectoryReader, VectorStoreIndex
fromllama_index.core.retrieversimportVectorIndexRetriever
fromllama_index.retrievers.bm25importBM25Retriever
fromllama_index.core.retrieversimportQueryFusionRetriever
fromllama_index.core.query_engineimportRetrieverQueryEngine
fromlangchain_core.toolsimporttool
# 和前面一样构建文档存储与索引
documents=SimpleDirectoryReader("./data").load_data()
index=VectorStoreIndex.from_documents(documents)
# ── 稠密检索器（向量相似度） ────────────────────────────
vector_retriever=VectorIndexRetriever(
index=index,
similarity_top_k=5, # 在重排序前获取更多候选项
)
# ── 稀疏检索器（BM25 关键词匹配） ────────────────────────
# BM25Retriever 直接基于索引中的节点构建
bm25_retriever=BM25Retriever.from_defaults(
index=index,
similarity_top_k=5,
)
# ── 融合：用倒数排名融合合并两个检索器 ──────────────────
# QueryFusionRetriever 是 LlamaIndex 内置的混合检索组合器
# mode="reciprocal_rerank" 实现 RRF 算法：
# 它可以在不做分数校准的情况下合并多个排序列表
hybrid_retriever=QueryFusionRetriever(
retrievers=[vector_retriever, bm25_retriever],
similarity_top_k=3, # 融合后的最终 top-k
num_queries=1, # 该模式下不做查询扩展
mode="reciprocal_rerank", # 融合算法
use_async=True, # 并行运行两个检索器
)
# 在混合检索器之上构建 QueryEngine
hybrid_query_engine=RetrieverQueryEngine.from_args(
retriever=hybrid_retriever,
)
# ── LANGGRAPH 工具 ─────────────────────────────────────────
@tool
defsearch_hybrid(query: str) ->str:
"""Search internal knowledge base using both semantic similarity
and keyword matching. More precise than pure vector search,
especially for technical terms, product codes, and exact names."""
response=hybrid_query_engine.query(query)
returnstr(response)
# LangGraph 的图结构与 Naive RAG 完全相同。
# 这里只替换工具。模块 4 及其后的所有内容
# 保持完全不变。
Hybrid RAG 的分块不宜照搬 Naive RAG，因为混合检索往往更适合较小的分块。原因和 BM25 有关：同一个高频词出现在小分块里，相关性信号比落在一大段文本里更清楚。引入 BM25 后，可以先用 256 个 Token、30 个 Token 重叠作为起点，再根据召回结果调整。
什么时候使用 Hybrid RAG
文档只要兼有自由文本和结构化术语，Hybrid RAG 就很值得优先考虑。法律文档里的引用格式、医疗记录中的药品名称与剂量代码、金融分析里的股票代码和合同条款标识符、技术文档里的错误码、API 方法名和版本号，都要求语义检索与精确匹配一起工作。
架构 3：Graph RAG
Graph RAG 改变的是对“文档”的建模方式。Naive RAG 和 Hybrid RAG 看到的是文本块，检索目标也是与查询最相关的文本块；Graph RAG 看到的是实体和关系，查询过程更接近沿着网络寻找一条路径。
例如：如果我们弃用旧版认证模块，哪些企业客户会受到影响？
普通检索会尝试寻找共同出现“企业客户”和“认证模块”的分块，可能确实能找到一些。但这个问题真正需要的是沿着关系链继续走下去：
答案并不藏在某一个分块里，而是出现在知识图谱的结构中。多跳推理、依赖关系追踪，以及需要把语料不同位置的事实连接起来的问题，都是 Graph RAG 的典型用法。
Graph RAG 如何构建索引
索引阶段先做实体抽取，而不是直接把所有分块拿去做 Embedding。系统从文档里识别产品、人物、组织、概念等命名实体，再抽取“依赖于”“是……的客户”“由……撰写”“被……取代”这类关系，把它们变成知识图谱中的节点和边。随后用图聚类算法组织出彼此关联紧密的社区，每个社区由 LLM 生成摘要。查询进来后，先匹配社区摘要，再沿图遍历到相关实体。
代码
# ============================================================
# GRAPH RAG——LLAMAINDEX 属性图索引
# pip install llama-index-core
# ============================================================
fromllama_index.coreimportSimpleDirectoryReader, PropertyGraphIndex
fromllama_index.core.indices.property_graphimport (
ImplicitPathExtractor,
SimpleLLMPathExtractor,
)
fromlangchain_core.toolsimporttool
# 加载文档
documents=SimpleDirectoryReader("./data").load_data()
# ── 构建属性图索引 ──────────────────────────────────────
# LlamaIndex 的 PropertyGraphIndex 负责完整的抽取 Pipeline。
# SimpleLLMPathExtractor 使用 LLM 抽取（主语、关系、宾语）
# 三元组；这些三元组会成为图中的边。
# ImplicitPathExtractor 使用快速启发式方法（成本更低、精度也更低）。
index=PropertyGraphIndex.from_documents(
documents,
kg_extractors=[
# 基于 LLM 的抽取：质量更高，成本也更高
SimpleLLMPathExtractor(
llm=Settings.llm,
max_paths_per_chunk=10,
),
# 启发式抽取：快速的备用方案
ImplicitPathExtractor(),
],
show_progress=True,
)
# ── 创建图感知检索器 ───────────────────────────────────
# 这个检索器遍历图，而不是搜索扁平向量。
kg_retriever=index.as_retriever(
include_text=True, # 为每个实体附带周围文本
retriever_mode="hybrid", # 在图上组合关键词与 Embedding
similarity_top_k=3,
)
fromllama_index.core.query_engineimportRetrieverQueryEngine
graph_query_engine=RetrieverQueryEngine.from_args(retriever=kg_retriever)
# ── LANGGRAPH 工具 ─────────────────────────────────────────
@tool
defsearch_knowledge_graph(query: str) ->str:
"""Search the knowledge graph for questions involving relationships
between entities - dependencies, organizational hierarchies, impact
analysis, and multi-hop reasoning across connected information.
Use this when the answer requires tracing a chain of relationships
rather than finding a single relevant document."""
response=graph_query_engine.query(query)
returnstr(response)
# ── 与 NAIVE RAG 组合：双工具 AGENT ────────────────────
# 实践中经常会把 Graph RAG 和 Naive RAG 组合使用。
# Agent 的 LLM 决定哪一个工具更适合当前查询。
@tool
defsearch_knowledge_base(query: str) ->str:
"""Search for factual information in company documents.
Best for direct questions with answers in a single document."""
response=query_engine.query(query) # Naive 查询引擎
returnstr(response)
# 把两个工具都交给 LangGraph Agent。
# 它会根据问题类型路由到合适的工具。
tools= [search_knowledge_base, search_knowledge_graph]
llm_with_tools=llm.bind_tools(tools)
tool_node=ToolNode(tools)
# 图装配保持不变，只修改工具列表。
Graph RAG 的实体抽取会对语料中的每个分块调用一次 LLM。文档集一大，索引阶段就可能出现数千次调用。慢和贵是这套结构的固有成本，因为它用一次性的索引开销换来更丰富的数据结构。Graph RAG 索引不应该在每次启动时重建。
什么时候使用 Graph RAG
问题是否需要沿关系链往下追。合规和风险分析要找“哪些流程会受到法规 X 的影响”，供应链系统要回答“哪些产品依赖这个供应商”，组织知识要追踪“谁负责什么，以及责任链如何连接”，软件依赖映射则会问“删除模块 Y 后哪里会坏”。这些都属于 Graph RAG 的范围。
架构 4：Advanced RAG
Advanced RAG 不是另一套完全独立的检索架构，更像是在现有检索基础上加的一组校正机制。Naive RAG 通常接受第一次检索结果；Advanced RAG 会继续改写、筛选和验证，直到拿到更可用的上下文。
检索前：查询改写与 HyDE
用户自然表达出来的问题，经常不是最适合搜索的查询。“我们企业客户的退货政策到底怎么规定？” 但 “企业客户退货与退款政策流程” 更利于检索。查询改写（Query Rewriting）先让 LLM 把原问题变成更适合索引的搜索语句，再真正执行查询。
HyDE（Hypothetical Document Embedding）换了一个角度。它先让 LLM 写出一份可能回答该问题的假设文档，再对这份文档做 Embedding，并用它去搜索。直觉在于：答案形态的文本在向量空间里，往往比问题形态的文本更接近真正的答案文本。
查询分解
多步骤问题通常不是一次检索能处理的。Advanced RAG 会先拆问题再分别检索。
“比较企业客户和零售客户的退款政策，并总结主要差异” 表面上是一句话，实际包含 3 个动作：找企业客户政策、找零售客户政策、比较两者。查询分解（Query Decomposition）把这些动作变成独立子查询，分别检索，最后再合并结果进行综合。
检索后：重排序与 CRAG
重排序和 Hybrid RAG 里使用的是同一个思路。即使底层只有稠密检索，也可以把 cross-encoder 接到检索结果后面，对候选分块再排一次。
CRAG（Corrective RAG）多加了一层判断。检索结束后，系统先评估当前分块是否真的足以回答问题；如果答案是否定的，就退回其他来源，例如 Web 搜索或更宽泛的索引，而不是拿着质量不够的上下文继续让 LLM 硬答。
代码
# ============================================================
# ADVANCED RAG——多技术实现
# ============================================================
fromllama_index.coreimportSimpleDirectoryReader, VectorStoreIndex
fromllama_index.core.query_engineimport (
SubQuestionQueryEngine, # 处理查询分解
RetrieverQueryEngine,
)
fromllama_index.core.toolsimportQueryEngineTool, ToolMetadata
fromllama_index.core.postprocessorimportSentenceTransformerRerank
fromlangchain_core.toolsimporttool
documents=SimpleDirectoryReader("./data").load_data()
index=VectorStoreIndex.from_documents(documents)
# ── 重排序器：检索后的 cross-encoder 评分 ───────────────
# 获取 8 个候选项，重排序后保留 3 个
# 这是 Advanced RAG 中影响最大的一项单独改进
reranker=SentenceTransformerRerank(
model="cross-encoder/ms-marco-MiniLM-L-2-v2",
top_n=3,
)
# ── 带重排序的检索器 ───────────────────────────────────
base_retriever=index.as_retriever(similarity_top_k=8)
reranked_engine=RetrieverQueryEngine.from_args(
retriever=base_retriever,
node_postprocessors=[reranker], # 检索后应用
)
# ── 使用 SubQuestionQueryEngine 做查询分解 ─────────────
# 把基础引擎包装成分解器可以调用的“工具”
engine_tools= [
QueryEngineTool(
query_engine=reranked_engine,
metadata=ToolMetadata(
name="company_knowledge_base",
description=(
"Searches company documents for policies, procedures, "
"and product information."
),
),
)
]
# SubQuestionQueryEngine 会把复杂问题拆成
# 多个子问题，分别交给可用工具执行，
# 再根据所有子答案综合出最终答案
decomposed_engine=SubQuestionQueryEngine.from_defaults(
query_engine_tools=engine_tools,
use_async=True, # 条件允许时并行运行子查询
)
# ── HYDE：Hypothetical Document Embedding ──────────────
fromllama_index.core.indices.query.query_transform.baseimport (
HyDEQueryTransform,
)
fromllama_index.core.query_engineimportTransformQueryEngine
hyde_transform=HyDEQueryTransform(include_original=True)
hyde_engine=TransformQueryEngine(
query_engine=reranked_engine,
query_transform=hyde_transform,
)
# ── LANGGRAPH 工具：不同模式使用不同引擎 ───────────────
@tool
defsearch_with_decomposition(query: str) ->str:
"""Search for answers to complex questions that may require
combining information from multiple documents. Automatically
breaks the question into sub-questions and merges the results.
Best for comparison questions, multi-part questions, and anything
requiring synthesis across different topics."""
response=decomposed_engine.query(query)
returnstr(response)
@tool
defsearch_with_hyde(query: str) ->str:
"""Search the knowledge base using hypothetical document embedding.
More effective than standard search for abstract or exploratory
questions where the exact terminology in the answer differs from
the terminology in the question."""
response=hyde_engine.query(query)
returnstr(response)
# 现在 LangGraph Agent 拥有专门的检索工具
# 它的 LLM 会判断每个问题需要哪一种检索策略。
tools= [search_with_decomposition, search_with_hyde]
小技巧：不要把所有技巧一次堆上去。更合理的顺序，是先针对当前最明显的失败模式加一项改进，并测量召回率变化。通常重排序最值得先做，查询分解排在后面；概念型、抽象型语料再考虑 HyDE。复杂度应该随着可观测的问题逐步增加。
架构 5：Agentic RAG
Agentic RAG 把前面的变化再往前推了一步：不只改变检索方式，还把“该怎么检索”的决定交给 Agent。
前 4 种架构都可以看成 Pipeline，执行顺序预先确定，每次查询沿着同一条路径走。Agentic RAG 则是一段循环。LLM Agent 先决定搜什么，再评估结果够不够；如果不够，就改查询继续搜，直到可以回答，或者判断现有信息无法得到答案。
这类循环正好对应 LangGraph 的结构：Agent 节点负责决策，工具节点执行检索，条件边决定下一步。
使用多个检索工具的 Agentic RAG
把前面几种检索策略都注册成工具后，Agentic RAG 的价值才更完整。每个子问题到底走向量检索、混合检索、图遍历，还是查询分解，可以由 Agent 的 LLM 在运行时判断。
# ============================================================
# AGENTIC RAG——完整多工具模板
# 这是完整的生产脚手架：全部 5 种架构
# 都提供给同一个 Agent，由它动态选择。
# ============================================================
# ── 模块 1：导入与配置 ───────────────────────────────────
importos
fromtypingimportLiteral, Annotated
fromlangchain_openaiimportChatOpenAI
fromlangchain_core.messagesimportHumanMessage, SystemMessage, BaseMessage
fromlangchain_core.toolsimporttool
fromlanggraph.graphimportStateGraph, MessagesState, START, END
fromlanggraph.prebuiltimportToolNode
fromlanggraph.checkpoint.memoryimportMemorySaver
fromllama_index.coreimport (
Settings, SimpleDirectoryReader, VectorStoreIndex,
StorageContext, load_index_from_storage, PropertyGraphIndex
)
fromllama_index.core.indices.property_graphimportSimpleLLMPathExtractor
fromllama_index.core.postprocessorimportSentenceTransformerRerank
fromllama_index.core.query_engineimportRetrieverQueryEngine, SubQuestionQueryEngine
fromllama_index.core.toolsimportQueryEngineTool, ToolMetadata
fromllama_index.llms.openaiimportOpenAIasLlamaOpenAI
fromllama_index.embeddings.openaiimportOpenAIEmbedding
fromllama_index.retrievers.bm25importBM25Retriever
fromllama_index.core.retrieversimportQueryFusionRetriever, VectorIndexRetriever
# Agent 的主要推理模型（LangGraph）
llm=ChatOpenAI(model="gpt-4o", temperature=0)
# LlamaIndex 内部配置
Settings.llm=LlamaOpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model=OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size=512
Settings.chunk_overlap=50
# ── 模块 2：状态 ──────────────────────────────────────────
classAgentState(MessagesState):
# 如需可观测性，可加入检索元数据
retrieval_count: int=0 # 记录执行了多少次检索调用
# ── 模块 3：知识库（全部在启动时构建） ────────────────
VECTOR_DIR="./storage/vector"
GRAPH_DIR ="./storage/graph"
documents =SimpleDirectoryReader("./data").load_data()
# 向量索引（用于 Naive、Hybrid、Advanced）
ifos.path.exists(VECTOR_DIR):
ctx=StorageContext.from_defaults(persist_dir=VECTOR_DIR)
vector_index=load_index_from_storage(ctx)
else:
vector_index=VectorStoreIndex.from_documents(documents, show_progress=True)
vector_index.storage_context.persist(persist_dir=VECTOR_DIR)
# 属性图索引（用于 Graph RAG）
ifos.path.exists(GRAPH_DIR):
ctx=StorageContext.from_defaults(persist_dir=GRAPH_DIR)
graph_index=load_index_from_storage(ctx)
else:
graph_index=PropertyGraphIndex.from_documents(
documents,
kg_extractors=[SimpleLLMPathExtractor(llm=Settings.llm)],
show_progress=True,
)
graph_index.storage_context.persist(persist_dir=GRAPH_DIR)
# ── 设置各个独立引擎 ──────────────────────────────────
reranker=SentenceTransformerRerank(
model="cross-encoder/ms-marco-MiniLM-L-2-v2", top_n=3
)
# 工具 1：Naive / 向量搜索
vector_engine=vector_index.as_query_engine(
similarity_top_k=3,
node_postprocessors=[reranker],
)
# 工具 2：Hybrid（向量 + BM25）
hybrid_retriever=QueryFusionRetriever(
retrievers=[
VectorIndexRetriever(index=vector_index, similarity_top_k=5),
BM25Retriever.from_defaults(index=vector_index, similarity_top_k=5),
],
similarity_top_k=3,
mode="reciprocal_rerank",
use_async=True,
)
hybrid_engine=RetrieverQueryEngine.from_args(
retriever=hybrid_retriever,
node_postprocessors=[reranker],
)
# 工具 3：图遍历
graph_engine=graph_index.as_query_engine(
include_text=True,
retriever_mode="hybrid",
similarity_top_k=3,
)
# 工具 4：Decomposed（用于复杂的多部分问题）
decomposed_engine=SubQuestionQueryEngine.from_defaults(
query_engine_tools=[
QueryEngineTool(
query_engine=vector_engine,
metadata=ToolMetadata(
name="docs",
description="Company documents and policies."
)
)
],
use_async=True,
)
# ── 模块 3（续）：全部工具 ────────────────────────────
@tool
defsearch_documents(query: str) ->str:
"""Search company documents by semantic meaning. Best for
conceptual questions where the exact wording in the answer
may differ from the question."""
returnstr(vector_engine.query(query))
@tool
defsearch_exact_terms(query: str) ->str:
"""Search using both keyword and semantic matching. Best when
the query contains specific terminology, product codes, names,
or exact phrases that must appear in the result."""
returnstr(hybrid_engine.query(query))
@tool
defsearch_relationships(query: str) ->str:
"""Search the knowledge graph for questions about how things
connect: dependencies, impact chains, organizational links,
and multi-hop reasoning. Use when the answer requires tracing
a relationship across multiple entities."""
returnstr(graph_engine.query(query))
@tool
defsearch_complex_question(query: str) ->str:
"""For multi-part questions requiring synthesis across several
topics. Automatically decomposes the question into sub-queries,
retrieves each independently, and combines the results."""
returnstr(decomposed_engine.query(query))
tools= [
search_documents,
search_exact_terms,
search_relationships,
search_complex_question,
]
llm_with_tools=llm.bind_tools(tools)
tool_node=ToolNode(tools)
# ── 模块 4：节点 ──────────────────────────────────────────
defagent_node(state: AgentState) ->dict:
system_prompt=SystemMessage(content=(
"You are a precise research assistant with access to four retrieval tools:\n\n"
"1. search_documents - semantic search over company documents\n"
"2. search_exact_terms - hybrid semantic + keyword search\n"
"3. search_relationships - graph traversal for relationship questions\n"
"4. search_complex_question - decomposed retrieval for multi-part questions\n\n"
"Think step by step. Use the tool that best fits the question type. "
"You may call multiple tools if a question has multiple parts. "
"Only answer when you have retrieved sufficient evidence."
))
response=llm_with_tools.invoke([system_prompt] +state["messages"])
return {
"messages": [response],
"retrieval_count": state.get("retrieval_count", 0),
}
# ── 模块 5：路由 ──────────────────────────────────────────
defshould_continue(state: AgentState) ->Literal["tools", "__end__"]:
last=state["messages"][-1]
ifhasattr(last, "tool_calls") andlast.tool_calls:
return"tools"
return"__end__"
# ── 模块 6：图装配 ─────────────────────────────────────────
builder=StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue,
{"tools": "tools", "__end__": END})
builder.add_edge("tools", "agent")
graph=builder.compile(checkpointer=MemorySaver())
# ── 模块 7：入口 ──────────────────────────────────────
if__name__=="__main__":
config= {"configurable": {"thread_id": "agentic-session-001"}}
print("Agentic RAG ready. Ask anything.\n")
whileTrue:
user_text=input("You: ").strip()
ifnotuser_textoruser_text.lower() in ("exit", "quit"):
break
response=graph.invoke(
{"messages": [HumanMessage(content=user_text)]},
config=config,
)
print(f"\nAgent: {response['messages'][-1].content}\n")
它和前面几种架构最关键的差别是自我纠错。Pipeline 不知道自己拿错了结果；Agent 可以在后续推理中发现这一点。
第一次检索质量较弱时，Agent 可以换一组搜索词重新查询。回答过程中发现新的依赖关系，也可以追加检索；问题本身有歧义时，还可以先要求澄清，再决定搜什么。固定 Pipeline 很难自然处理这些分支。
合规、法律、金融、医疗这类高风险场景更容易体现这种能力的价值。置信度阈值和人在回路（human-in-the-loop）检查点，都可以直接放进 Agent 图里。
不过Agentic RAG 更慢。单工具 Pipeline 通常是 200 到 500 毫秒；Agent 如果在回答前连续做 3 次检索，可能要 8 到 12 秒。实时界面的主交互路径很难忽略这段等待。生产上常见的处理方式有两种：把中间步骤流式返回，让用户看到系统正在做什么；或者异步执行 Agentic 检索，提前为可能出现的后续问题准备答案。
汇总：怎么选
这 5 种架构并没有统一的“最好”。更有用的问题是：当前检索失败在什么地方。
这些架构天然可以组合。项目不需要在几个方案里做单选，常见做法是从简单层开始，根据问题逐步叠加。
LangGraph 里的实现很直接：工具列表就是这套架构栈的接口。Agentic RAG Agent 拿到 Naive、Hybrid、Graph 和 Decomposed 工具后，就形成完整的 5 层架构栈；第 5 层 Agent 会在每一轮里选择并编排第 1 到第 4 层的能力。
总结
这 5 种架构不是完成同一件事的 5 种做法，Naive、Hybrid、Graph、Advanced、Agentic 分别补上不同层面的检索能力，复杂度也随之逐步增加。
Naive RAG 快、便宜，适合多数文档问答；Hybrid RAG 对专业术语更稳，通常可以作为生产默认方案；Graph RAG 处理关系链；Advanced RAG 用在召回和上下文质量不够的时候；Agentic RAG 则把多轮检索、自我纠错和更开放的推理交给 Agent。
把这些能力组合起来后，LlamaIndex 负责数据层，LangGraph 负责编排层；这 5 种模式基本覆盖了基于检索构建生产 AI 应用时最常见的需求。
两个框架之间的接缝是一个由
@tool
装饰的函数。
作者：Bessie Delight Kekeli