LlamaIndex 实战——索引构建、查询引擎、多文档联合检索深度指南
大约 9 分钟
LlamaIndex 实战——索引构建、查询引擎、多文档联合检索深度指南
适读人群：想用 Python 构建知识检索系统的工程师 | 阅读时长：约18分钟 | 核心价值：掌握 LlamaIndex 核心机制，搭建比 LangChain RAG 更灵活的多文档检索方案
小周是个做了五年后端的工程师，去年开始做公司的 AI 文档助手项目。他最开始用 LangChain 的 RAG，跑通了基本流程，但随着文档量增加，发现了一个棘手的问题：公司有产品手册、技术文档、FAQ、合同模板几十种文档，每种文档的结构完全不同，有的是 PDF，有的是 Excel，有的是层级明显的 Markdown。用同一套分割策略处理所有文档，效果参差不齐。
他找我的那天语气很无奈："老张，我把 LangChain 的文档从头到尾看了一遍，感觉文档类型越多，代码就越乱，什么手动分割、自定义 Loader，搞得很复杂。有没有更专门处理文档的框架？"
有。就是 LlamaIndex。
如果说 LangChain 是"做 AI 应用的瑞士军刀"，那 LlamaIndex 就是"专门为文档数据设计的精密仪器"。它对文档结构有原生的理解，索引构建、查询引擎、多文档路由都比 LangChain 做得更细腻。
不是说 LlamaIndex 一定比 LangChain 好，而是当你的核心需求是文档检索和问答时，LlamaIndex 是更合适的工具。
今天我们从实战角度把 LlamaIndex 的核心用法讲透。
一、LlamaIndex 的核心概念
在写代码之前，先建立几个关键概念：
- Document：原始文档，可以是文件、网页、数据库记录
- Node：文档被分割后的最小单元，保留与父文档的关系
- Index：从 Node 构建的索引结构，支持多种类型（向量索引、关键词索引、树形索引等）
- QueryEngine：查询引擎，接收用户问题，返回答案
- RetrieverQueryEngine：可以自定义检索策略的查询引擎
LlamaIndex 最大的优势是这些组件之间的接口非常干净，组合起来很自然。
二、环境安装与基础配置
# 安装核心包
# pip install llama-index llama-index-embeddings-openai llama-index-llms-openai
# pip install llama-index-readers-file pypdf
import os
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Settings,
StorageContext,
load_index_from_storage
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# 全局配置（推荐在项目入口统一设置）
Settings.llm = OpenAI(
model="gpt-3.5-turbo",
temperature=0.1,
api_key=os.environ.get("OPENAI_API_KEY")
)
Settings.embed_model = OpenAIEmbedding(
model="text-embedding-3-small",
api_key=os.environ.get("OPENAI_API_KEY")
)
Settings.chunk_size = 512 # 每个 Node 的最大 Token 数
Settings.chunk_overlap = 64 # Node 之间的重叠 Token三、文档加载与索引构建
3.1 从目录批量加载文档
from llama_index.core import SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter, MarkdownNodeParser
from llama_index.core.ingestion import IngestionPipeline
def load_and_index_documents(docs_dir: str, persist_dir: str = "./storage") -> VectorStoreIndex:
"""
加载目录中的所有文档并构建向量索引
支持持久化存储，避免重复向量化
"""
# 检查是否已有持久化索引
if os.path.exists(persist_dir) and os.listdir(persist_dir):
print("加载已有索引...")
storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
index = load_index_from_storage(storage_context)
return index
# 加载文档
print("首次构建索引，正在加载文档...")
reader = SimpleDirectoryReader(
input_dir=docs_dir,
recursive=True, # 递归读取子目录
required_exts=[".pdf", ".md", ".txt", ".docx"],
filename_as_id=True # 用文件名作为文档 ID，方便后续更新
)
documents = reader.load_data()
print(f"加载了 {len(documents)} 个文档")
# 构建索引
index = VectorStoreIndex.from_documents(
documents,
show_progress=True
)
# 持久化保存
index.storage_context.persist(persist_dir=persist_dir)
print(f"索引已保存到 {persist_dir}")
return index
# 使用
index = load_and_index_documents("/data/company_docs")3.2 精细化的 Node 解析管道
对于结构化程度不同的文档，需要差异化处理：
from llama_index.core.node_parser import (
SentenceSplitter,
MarkdownNodeParser,
CodeSplitter,
SemanticSplitterNodeParser
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.schema import Document
def build_ingestion_pipeline():
"""
构建文档处理管道
对不同类型的文档使用不同的分割策略
"""
# 通用文本分割器
text_splitter = SentenceSplitter(
chunk_size=512,
chunk_overlap=64,
separator=" ",
paragraph_separator="\n\n\n",
)
return text_splitter
def process_markdown_documents(md_docs: list) -> list:
"""对 Markdown 文档按标题层级分割，保留结构信息"""
parser = MarkdownNodeParser()
nodes = parser.get_nodes_from_documents(md_docs)
# 每个 Node 会自动包含其所在的 Heading 层级
for node in nodes[:3]:
print(f"Node 标题路径：{node.metadata.get('header_path', '无')}")
print(f"内容预览：{node.text[:100]}...")
return nodes
def process_code_documents(code_docs: list) -> list:
"""对代码文档按函数/类边界分割"""
parser = CodeSplitter(
language="python",
chunk_lines=40, # 每块最多40行
chunk_lines_overlap=5, # 5行重叠
max_chars=1500
)
return parser.get_nodes_from_documents(code_docs)四、查询引擎深度使用
4.1 基础查询引擎
# 构建基础查询引擎
query_engine = index.as_query_engine(
similarity_top_k=5, # 检索最相关的5个 Node
response_mode="compact", # 紧凑模式：合并多个 Node 后回答
# 其他 response_mode：
# "refine"：逐个 Node 迭代完善答案（慢但准）
# "tree_summarize"：树形汇总（适合长文档摘要）
# "no_text"：只返回检索结果，不生成答案
)
# 基础查询
response = query_engine.query("产品的退款政策是什么？")
print("回答：", response)
print("\n引用来源：")
for node in response.source_nodes:
print(f" 文件：{node.metadata.get('file_name', 'unknown')}")
print(f" 相关度：{node.score:.3f}")
print(f" 内容：{node.text[:150]}...")4.2 踩坑实录一：response_mode 选错导致回答质量差
现象：用户问一个跨越多个章节的综合性问题（比如"产品有哪些核心优势"），AI 只回答了检索到的第一个 Node 的内容，漏掉了其他维度。
原因：默认的 compact 模式会把所有 Node 合并成一段文本，如果模型 context 不够，后面的内容会被截断。
解法：
# 对综合性问题用 tree_summarize
query_engine_comprehensive = index.as_query_engine(
similarity_top_k=8,
response_mode="tree_summarize", # 树形汇总，能处理更多 Node
summary_template="""
请基于以下信息，综合归纳回答问题。
信息来源：{context_str}
问题：{query_str}
综合回答：
"""
)
# 对需要精准引用的问题用 refine
query_engine_precise = index.as_query_engine(
similarity_top_k=5,
response_mode="refine", # 慢但准，每个 Node 都会参与完善答案
)五、多文档联合检索
这是 LlamaIndex 真正强于 LangChain 的地方。
5.1 SubQuestionQueryEngine——拆分复杂问题
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.query_engine import SubQuestionQueryEngine
# 假设你有三个不同主题的文档库
product_index = VectorStoreIndex.from_documents(product_docs)
tech_index = VectorStoreIndex.from_documents(tech_docs)
faq_index = VectorStoreIndex.from_documents(faq_docs)
# 把每个索引包装成 Tool，并给清晰的描述
tools = [
QueryEngineTool(
query_engine=product_index.as_query_engine(similarity_top_k=3),
metadata=ToolMetadata(
name="product_manual",
description="包含产品功能介绍、使用说明、规格参数等产品手册信息"
)
),
QueryEngineTool(
query_engine=tech_index.as_query_engine(similarity_top_k=3),
metadata=ToolMetadata(
name="tech_docs",
description="包含 API 文档、集成指南、技术架构等开发者文档"
)
),
QueryEngineTool(
query_engine=faq_index.as_query_engine(similarity_top_k=3),
metadata=ToolMetadata(
name="faq",
description="包含常见问题解答、故障排查、账号管理等支持文档"
)
)
]
# SubQuestionQueryEngine 会自动：
# 1. 把复杂问题拆分为多个子问题
# 2. 路由每个子问题到最合适的文档库
# 3. 综合所有子问题的答案给出最终回答
sub_question_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=tools,
verbose=True
)
# 测试：这个问题涉及多个文档库
response = sub_question_engine.query(
"这个产品有哪些核心功能，API 集成文档在哪里，常见的集成问题如何解决？"
)
print("最终回答：", response)5.2 RouterQueryEngine——智能路由
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector, LLMMultiSelector
# 单选路由：每次查询只选一个最合适的索引
router_engine = RouterQueryEngine(
selector=LLMSingleSelector.from_defaults(),
query_engine_tools=tools,
verbose=True
)
# 多选路由：可以同时查询多个索引（适合综合性问题）
multi_router_engine = RouterQueryEngine(
selector=LLMMultiSelector.from_defaults(),
query_engine_tools=tools,
verbose=True
)
# 测试路由效果
print("单选路由结果：")
r1 = router_engine.query("如何调用产品的 REST API？") # 应该路由到 tech_docs
print(r1)
print("\n多选路由结果：")
r2 = multi_router_engine.query("产品价格是多少，有没有集成示例代码？") # 应路由到 product + tech
print(r2)5.3 踩坑实录二：SubQuestion 拆分不合理
现象：SubQuestionQueryEngine 把一个简单问题拆分成了7-8个子问题，花了20多秒，成本也大幅增加。
原因：LLM 在拆分问题时过于激进，把一个本来可以直接回答的问题拆得很细。
解法：
from llama_index.core.question_gen import LLMQuestionGenerator
from llama_index.core.prompts import PromptTemplate
# 自定义子问题生成提示词，限制拆分数量
CUSTOM_SUB_QUESTION_PROMPT = PromptTemplate(
"""你是一个查询分析助手。分析以下问题，拆分成最少必要的子问题（通常1-3个，不超过4个）。
可用的信息源：
{tools_str}
原始问题：{query_str}
请只在问题真正需要从多个信息源获取信息时才拆分，否则直接给出1个子问题。
输出格式（每行一个子问题）：
"""
)
question_gen = LLMQuestionGenerator.from_defaults(
prompt_template_str=CUSTOM_SUB_QUESTION_PROMPT.template
)
sub_question_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=tools,
question_gen=question_gen,
verbose=True,
use_async=True # 异步并行执行子问题，加快速度
)六、高级特性：知识图谱索引
LlamaIndex 除了向量索引，还支持知识图谱索引，特别适合实体关系明确的文档：
from llama_index.core import KnowledgeGraphIndex
from llama_index.core.graph_stores import SimpleGraphStore
# 构建知识图谱索引
# 适用于：人物关系、产品依赖、技术架构等有明确关系的文档
graph_store = SimpleGraphStore()
storage_context = StorageContext.from_defaults(graph_store=graph_store)
kg_index = KnowledgeGraphIndex.from_documents(
documents,
storage_context=storage_context,
max_triplets_per_chunk=5, # 每个文本块最多提取5个三元组
include_embeddings=True, # 同时构建向量索引，支持混合查询
show_progress=True
)
# 知识图谱查询引擎
kg_query_engine = kg_index.as_query_engine(
include_text=True,
response_mode="tree_summarize",
embedding_mode="hybrid", # 混合知识图谱和向量检索
similarity_top_k=3
)
response = kg_query_engine.query("张三负责哪些项目，与哪些团队有协作关系？")
print(response)七、踩坑实录三：多索引时 Node ID 冲突
现象：把多个 SimpleDirectoryReader 加载的文档合并构建索引时，偶尔出现 Node ID already exists 警告，并且部分文档的内容被覆盖。
原因：LlamaIndex 默认用文档内容的哈希值生成 Node ID，不同目录下如果有内容完全相同的文件（比如多个版本的同一份模板），ID 会冲突。
解法：
from llama_index.core.schema import TextNode
import uuid
def load_with_unique_ids(dirs: list) -> list:
"""从多个目录加载文档，确保 Node ID 唯一"""
all_documents = []
for dir_path in dirs:
reader = SimpleDirectoryReader(
input_dir=dir_path,
recursive=True,
filename_as_id=True
)
docs = reader.load_data()
# 强制使用唯一 ID，加上目录前缀
for doc in docs:
doc.doc_id = f"{os.path.basename(dir_path)}_{doc.doc_id}_{uuid.uuid4().hex[:8]}"
doc.metadata["source_dir"] = dir_path
all_documents.extend(docs)
return all_documents
# 使用
documents = load_with_unique_ids([
"/data/product_v1",
"/data/product_v2",
"/data/faq"
])八、性能对比与选型建议
基于我在实际项目中的测试数据（10万文档块，gpt-3.5-turbo）：
| 查询引擎类型 | 平均延迟 | Token 消耗 | 适用场景 |
|---|---|---|---|
| 基础 VectorStoreIndex | 2.1s | ~1200 | 单一主题，问题明确 |
| SubQuestionQueryEngine | 8.5s | ~4500 | 复杂多文档问题 |
| RouterQueryEngine | 3.2s | ~1800 | 多主题，单次查询 |
| KnowledgeGraphIndex | 4.0s | ~2000 | 实体关系查询 |
选型建议：
- 文档少、主题单一 → 直接用
VectorStoreIndex - 文档多、主题分散 →
RouterQueryEngine - 问题复杂、需要综合多源 →
SubQuestionQueryEngine（注意成本） - 有明确实体关系 →
KnowledgeGraphIndex
LlamaIndex vs LangChain：两者各有侧重，文档密集型应用首选 LlamaIndex，工作流编排和 Agent 首选 LangChain。项目中也可以组合使用——LlamaIndex 做索引和检索，LangChain 做 Agent 调度。