1.查询 #

现在你已经加载了数据、构建了索引并存储了该索引，接下来就可以进入LLM应用最重要的部分：查询。

在最简单的情况下，查询就是向大语言模型发起的一次提示调用：可以是一个问题并得到答案，也可以是摘要请求，或是一条更复杂的指令。

更复杂的查询可能涉及多次/链式的提示+LLM调用，甚至跨多个组件的推理循环。

2.入门指南 #

所有查询的基础是 QueryEngine。获取 QueryEngine 的最简单方法是让您的 index 为您创建一个：

# 导入llama_index核心库中的VectorStoreIndex和Document类
from llama_index.core import VectorStoreIndex, Document

# 构建示例文档内容的列表
sample_texts = [
    "LlamaIndex是一个强大的数据框架，用于构建LLM应用。它提供了多种索引类型和查询功能。",
    "查询引擎是LlamaIndex的核心组件，负责处理用户查询并返回相关答案。",
    "RAG（检索增强生成）技术结合了检索和生成，能够提供基于文档的准确回答。",
    "向量索引通过嵌入向量实现语义搜索，能够理解查询的深层含义。",
]

# 将每条文本内容封装为Document对象，便于后续索引处理
documents = [Document(text=text) for text in sample_texts]

# 基于文档对象列表创建向量索引
index = VectorStoreIndex.from_documents(documents)

# 通过索引对象获取查询引擎
query_engine = index.as_query_engine()

# 使用查询引擎执行问题检索，自动返回相关答案
response = query_engine.query("什么是LlamaIndex？")

# 打印查询结果到控制台
print(f"查询结果: {response}")

LlamaIndex查询的核心流程：

1. 文档准备：创建包含示例文本的Document对象
2. 索引构建：将文档转换为可检索的向量索引
3. 查询引擎创建：从索引创建查询引擎
4. 执行查询：使用自然语言查询获取答案

3.查询阶段 #

查询操作不仅仅是表面看到的那样。整个查询过程包含三个核心阶段：

1. 检索：从数据中查找并返回与查询最相关的文档。最常见的检索类型是"top-k"语义检索，但还存在许多其他检索策略。
2. 后处理：对检索到的 Node 结果进行可选的重新排序、转换或过滤，例如要求它们具备特定的元数据或关键词。
3. 响应合成：将查询、最相关的数据及提示组合在一起，发送给LLM以生成最终响应。

提示：你可以了解更多关于如何将元数据附加到文档和节点的信息。

4.自定义查询阶段 #

LlamaIndex 提供了低层级的组合API，让你能精细控制查询过程。

例如，可以自定义检索器的 top_k 数值，并添加后处理步骤，要求检索到的节点必须达到最低相似度分数才能被包含。这样在有相关结果时能提供更多数据，但若没有相关内容则可能返回空结果。

# 导入必要的库
from llama_index.core import VectorStoreIndex, Document, get_response_synthesizer
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor

# 创建示例文档
sample_texts = [
    "LlamaIndex提供了多种检索策略，包括向量检索、关键词检索等。",
    "相似度阈值可以过滤掉相关性较低的检索结果，提高回答质量。",
    "自定义查询引擎允许开发者精确控制检索和响应过程。",
    "后处理器可以对检索结果进行进一步筛选和优化。",
]

# 将文本转换为Document对象
documents = [Document(text=text) for text in sample_texts]

# 构建向量索引
# 索引将文档转换为可检索的向量表示
index = VectorStoreIndex.from_documents(documents)

# 配置检索器
# VectorIndexRetriever是专门用于向量索引的检索器
# similarity_top_k参数控制返回的最相关节点数量
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=10,  # 返回前10个最相关的节点
)

# 配置响应合成器
# 响应合成器负责将检索到的节点信息整合成最终答案
response_synthesizer = get_response_synthesizer()

# 配置相似度后处理器
# 只保留相似度分数超过0.7的节点，过滤掉相关性较低的结果
similarity_postprocessor = SimilarityPostprocessor(similarity_cutoff=0.7)

# 组装自定义查询引擎
# 将检索器、响应合成器和后处理器组合成完整的查询引擎
query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=response_synthesizer,
    node_postprocessors=[similarity_postprocessor],  # 应用后处理器
)

# 执行查询
# 查询引擎会按照配置的流程处理查询
response = query_engine.query("如何提高检索结果的质量？")

# 打印查询结果
print(f"查询结果: {response}")

# 查看检索到的节点信息
print(f"\n检索到的节点数量: {len(response.source_nodes)}")
for i, node in enumerate(response.source_nodes):
    print(f"节点 {i+1}: {node.text[:100]}...")

自定义查询流程：

1. 检索器配置：设置返回的节点数量和相似度阈值
2. 响应合成器：配置如何整合检索到的信息
3. 后处理器：添加相似度过滤，提高结果质量
4. 结果分析：查看检索到的节点数量和内容

你还可以通过实现相应接口，添加自己的检索、响应合成和整体查询逻辑。要查看已实现组件的完整列表及支持的配置，请访问API参考文档。

4.1 配置检索器 #

检索器负责从索引中查找与查询最相关的文档节点。LlamaIndex提供了多种检索器类型：

# 导入VectorStoreIndex和Document类，用于向量存储和文档表示
from llama_index.core import VectorStoreIndex, Document

# 导入VectorIndexRetriever类，用于向量检索
from llama_index.core.retrievers import VectorIndexRetriever

# 定义一个包含示例文本的列表
sample_texts = [
    "向量检索器使用嵌入向量进行语义搜索。",
    "关键词检索器基于精确的关键词匹配。",
    "混合检索器结合多种检索策略。",
]

# 将每个示例文本包装成Document对象，生成文档列表
documents = [Document(text=text) for text in sample_texts]
# 通过文档列表创建向量存储索引
index = VectorStoreIndex.from_documents(documents)

# 配置向量索引检索器
# similarity_top_k参数指定返回最相关的前5个节点
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=5,  # 返回前5个最相关节点
)

# 基于检索器创建查询引擎
# 注意：不需要显式传递retriever参数，因为index.as_query_engine()会使用索引的默认检索器
query_engine = index.as_query_engine()

# 执行查询，查询内容为"什么是向量检索？"
response = query_engine.query("什么是向量检索？")
# 打印查询结果
print(f"查询结果: {response}")

检索器配置示例

1. 设置检索参数：控制返回的节点数量
2. 应用检索器：将检索器集成到查询引擎中
3. 优化检索效果：通过参数调整提高检索质量

更多检索器类型和用法请参考检索器模块指南。

4.2 配置节点后处理器 #

LlamaIndex 支持多种高级 Node 过滤与增强技术，可进一步提升检索结果的相关性，减少LLM调用次数和成本，或提升响应质量。

常见的节点后处理器包括：

• KeywordNodePostprocessor：根据节点筛选 required_keywords 和 exclude_keywords
• SimilarityPostprocessor：通过设定相似度分数阈值筛选节点（仅支持基于嵌入的检索器）
• PrevNextNodePostprocessor：基于节点关系增强检索内容

完整列表请见Node 后处理器参考。

配置示例：

# 导入VectorStoreIndex和Document类，用于向量存储和文档表示
from llama_index.core import VectorStoreIndex, Document

# 导入向量检索器
from llama_index.core.retrievers import VectorIndexRetriever

# 导入检索查询引擎
from llama_index.core.query_engine import RetrieverQueryEngine

# 导入关键词和相似度后处理器
from llama_index.core.postprocessor import (
    KeywordNodePostprocessor,
    SimilarityPostprocessor,
)

# 构建示例文本列表
sample_texts = [
    "LlamaIndex是一个强大的数据框架，用于构建LLM应用。",
    "关键词 后处理器 可以根据特定词汇过滤检索结果。",
    "相似度 后处理器 可以过滤掉相关性较低的节点。",
    "后处理器 可以显著提高查询结果的 质量 和相关性。",
]

# 将每个文本包装成Document对象
documents = [Document(text=text) for text in sample_texts]
# 基于文档列表创建向量索引
index = VectorStoreIndex.from_documents(documents)

# 配置向量检索器，设置返回最相似的前10个结果
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=10,
)

# 配置相似度后处理器，只保留相似度分数大于0.3的节点
similarity_postprocessor = SimilarityPostprocessor(similarity_cutoff=0.3)
# 配置关键词后处理器，要求包含“后处理器”且不包含“质量”
keyword_postprocessor = KeywordNodePostprocessor(
    required_keywords=["后处理器"],
    exclude_keywords=["质量"],
)

# 创建检索查询引擎，使用关键词和相似度后处理器
query_engine = RetrieverQueryEngine.from_args(
    retriever, node_postprocessors=[keyword_postprocessor, similarity_postprocessor]
)

# 执行查询，查询内容为“后处理器有什么作用？”
response = query_engine.query("后处理器有什么作用？")
# 打印查询结果
print(f"查询结果: {response}")

后处理器配置示例：

1. 关键词过滤：使用特定词汇筛选节点
2. 相似度过滤：基于相似度分数过滤结果
3. 多处理器组合：同时应用多个后处理器
4. 结果验证：检查过滤后的节点数量

4.3 配置响应合成 #

检索器获取相关节点后，BaseSynthesizer 负责整合信息并合成最终响应。

# 导入VectorStoreIndex和Document类，用于向量存储和文档表示
from llama_index.core import VectorStoreIndex, Document

# 导入VectorIndexRetriever类，用于向量检索
from llama_index.core.retrievers import VectorIndexRetriever

# 导入RetrieverQueryEngine类，用于基于检索器的查询引擎
from llama_index.core.query_engine import RetrieverQueryEngine

# 定义一个包含示例文本的列表
sample_texts = [
    "响应合成器负责将检索到的信息整合成最终答案。",
    "不同的响应模式适用于不同的应用场景。",
    "default模式适合需要详细回答的场景。",
    "compact模式适合处理大量文本信息。",
    "tree_summarize模式适合生成文档摘要。",
]

# 将每个示例文本包装成Document对象，组成文档列表
documents = [Document(text=text) for text in sample_texts]
# 通过文档列表创建向量存储索引
index = VectorStoreIndex.from_documents(documents)

# 创建向量检索器，设置检索时返回最相似的5条结果
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=5,
)

# 定义要演示的三种响应模式
response_modes = ["refine", "compact", "tree_summarize"]

# 遍历每种响应模式进行演示
for mode in response_modes:
    # 打印当前使用的响应模式
    print(f"\n=== 使用 {mode} 响应模式 ===")

    # 创建基于当前检索器和响应模式的查询引擎
    query_engine = RetrieverQueryEngine.from_args(retriever, response_mode=mode)

    # 执行查询，问题为“响应合成器有什么作用？”
    response = query_engine.query("响应合成器有什么作用？")
    # 打印查询结果
    print(f"查询结果: {response}")

响应合成配置示例展示了如何：

1. 选择响应模式：根据应用需求选择合适的模式
2. 比较不同模式：测试不同响应模式的效果
3. 优化响应质量：通过模式选择提高回答质量

目前支持的响应模式包括：

• refine： 使用LLM对检索到的节点进行细化，适合需要详细回答的场景
• compact： 尽可能多地压缩提示信息，适合大文本量
• tree_summarize： 递归构建树结构，适合摘要生成
• no_text： 仅运行检索器，不实际发送给LLM，便于调试
• accumulate：对每个文本块单独运行相同查询，返回拼接结果

5.高级查询示例 #

5.1 多步骤查询 #

# 导入VectorStoreIndex和Document类，用于向量存储和文档表示
from llama_index.core import VectorStoreIndex, Document

# 导入SubQuestionQueryEngine类，用于多步骤子问题查询
from llama_index.core.query_engine import SubQuestionQueryEngine

# 导入QueryEngineTool类，用于将查询引擎封装为工具
from llama_index.core.tools import QueryEngineTool

# 定义一个包含示例文本的列表
sample_texts = [
    "LlamaIndex支持多种查询策略，包括简单查询和复杂查询。",
    "多步骤查询可以将复杂问题分解为多个子问题。",
    "SubQuestionQueryEngine能够自动分解复杂查询并整合结果。",
    "查询工具可以将不同的查询引擎组合成更强大的系统。",
]

# 将每个示例文本包装成Document对象，组成文档列表
documents = [Document(text=text) for text in sample_texts]
# 通过文档列表创建向量存储索引
index = VectorStoreIndex.from_documents(documents)

# 创建基础查询引擎，用于后续封装
base_query_engine = index.as_query_engine()

# 使用QueryEngineTool将基础查询引擎封装为工具，便于组合和调用
query_tool = QueryEngineTool.from_defaults(
    query_engine=base_query_engine, description="用于回答关于LlamaIndex的问题"
)

# 创建SubQuestionQueryEngine，支持将复杂查询自动分解为多个子问题并整合结果
# verbose=True表示输出详细的查询过程
sub_question_query_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=[query_tool], verbose=True
)

# 执行复杂查询，自动分解为子问题并整合答案
response = sub_question_query_engine.query(
    "LlamaIndex有哪些查询策略？它们各自有什么特点？"
)
# 打印复杂查询的最终结果
print(f"复杂查询结果: {response}")

多步骤查询示例展示了如何：

1. 工具化查询引擎：将查询引擎包装成可组合的工具
2. 自动问题分解：将复杂问题分解为多个子问题
3. 结果整合：自动整合多个子问题的答案

5.2 条件查询 #

# 导入VectorStoreIndex和Document类，用于向量存储和文档表示
from llama_index.core import VectorStoreIndex, Document

# 导入RouterQueryEngine类，用于实现查询路由
from llama_index.core.query_engine import RouterQueryEngine

# 导入QueryEngineTool类，用于将查询引擎封装为工具
from llama_index.core.tools import QueryEngineTool

# 定义与LlamaIndex相关的文本文档
llamaindex_texts = ["LlamaIndex是一个强大的数据框架。", "LlamaIndex支持多种索引类型。"]

# 定义与RAG相关的文本文档
rag_texts = ["RAG是检索增强生成的缩写。", "RAG结合了检索和生成技术。"]

# 将LlamaIndex相关文本包装为Document对象
llamaindex_docs = [Document(text=text) for text in llamaindex_texts]
# 将RAG相关文本包装为Document对象
rag_docs = [Document(text=text) for text in rag_texts]

# 基于LlamaIndex文档创建向量索引
llamaindex_index = VectorStoreIndex.from_documents(llamaindex_docs)
# 基于RAG文档创建向量索引
rag_index = VectorStoreIndex.from_documents(rag_docs)

# 由LlamaIndex索引创建查询引擎
llamaindex_query_engine = llamaindex_index.as_query_engine()
# 由RAG索引创建查询引擎
rag_query_engine = rag_index.as_query_engine()

# 将LlamaIndex查询引擎封装为查询工具
llamaindex_tool = QueryEngineTool.from_defaults(
    query_engine=llamaindex_query_engine, description="用于回答关于LlamaIndex的问题"
)

# 将RAG查询引擎封装为查询工具
rag_tool = QueryEngineTool.from_defaults(
    query_engine=rag_query_engine, description="用于回答关于RAG技术的问题"
)

# 创建路由器查询引擎，能够根据问题内容自动选择合适的查询工具
router_query_engine = RouterQueryEngine.from_defaults(
    query_engine_tools=[llamaindex_tool, rag_tool], verbose=True
)

# 定义要测试的查询列表
queries = ["什么是LlamaIndex？", "RAG技术有什么优势？", "LlamaIndex和RAG有什么关系？"]

# 遍历每个查询，使用路由器查询引擎自动选择合适的工具进行回答
for query in queries:
    print(f"\n查询: {query}")
    response = router_query_engine.query(query)
    print(f"回答: {response}")

条件查询示例展示了如何：

1. 多索引管理：为不同主题创建独立的索引
2. 智能路由：根据查询内容自动选择最合适的查询引擎
3. 工具组合：将多个查询工具组合成统一的接口

6.查询性能优化 #

6.1 缓存查询结果 #

# 导入VectorStoreIndex和Document类，用于向量存储和文档处理
from llama_index.core import VectorStoreIndex, Document

# 定义一个包含示例文本的列表
sample_texts = [
    "查询缓存可以显著提高重复查询的响应速度。",
    "缓存机制可以避免重复的检索和生成过程。",
    "SimpleCache提供了基本的缓存功能。",
]

# 将每个示例文本包装成Document对象，生成文档列表
documents = [Document(text=text) for text in sample_texts]

# 通过文档列表创建向量索引
index = VectorStoreIndex.from_documents(documents)

# 创建一个简单的内存缓存（字典类型）
cache = {}

# 创建一个查询引擎
query_engine = index.as_query_engine()

# 第一次执行查询（此时会进行实际计算）
print("第一次查询（会计算）:")
response1 = query_engine.query("什么是查询缓存？")
print(f"结果: {response1}")

# 将第一次查询的结果存入缓存
cache["什么是查询缓存？"] = response1
print(f"缓存大小: {len(cache)}")

# 第二次执行相同的查询（此时会命中缓存）
print("\n第二次查询（使用缓存）:")
if "什么是查询缓存？" in cache:
    # 如果缓存中有结果，则直接从缓存获取
    response2 = cache["什么是查询缓存？"]
    print("从缓存中获取结果")
else:
    # 如果缓存中没有，则重新计算
    response2 = query_engine.query("什么是查询缓存？")
    print("重新计算")
print(f"结果: {response2}")

# 判断两次查询结果是否一致，验证缓存是否生效
print(f"\n缓存命中: {response1 == response2}")
print(f"缓存内容: {list(cache.keys())}")

缓存优化示例展示了如何：

1. 配置缓存：使用SimpleCache提供缓存功能
2. 缓存键设置：使用查询字符串作为缓存键
3. 性能对比：比较缓存前后的查询性能

6.2 批量查询 #

# 导入llama_index的向量索引和文档类
from llama_index.core import VectorStoreIndex, Document

# 导入异步IO库
import asyncio

# 定义示例文本列表
sample_texts = [
    "批量查询可以同时处理多个查询请求。",
    "异步查询可以提高并发性能。",
    "批量处理适合处理大量查询任务。",
]

# 将每个示例文本包装为Document对象，生成文档列表
documents = [Document(text=text) for text in sample_texts]

# 基于文档列表创建向量索引
index = VectorStoreIndex.from_documents(documents)

# 创建异步查询引擎
async_query_engine = index.as_query_engine()


# 定义批量异步查询函数
async def batch_query(queries):
    """批量执行查询"""
    # 为每个查询创建异步任务
    tasks = [async_query_engine.aquery(query) for query in queries]
    # 并发执行所有查询任务并收集结果
    responses = await asyncio.gather(*tasks)
    # 返回所有查询结果
    return responses


# 定义要批量查询的问题列表
queries = ["什么是批量查询？", "异步查询有什么优势？", "批量处理适合什么场景？"]

# 执行批量查询并获取结果
print("执行批量查询...")
responses = asyncio.run(batch_query(queries))

# 遍历并输出每个查询及其对应的答案
for i, (query, response) in enumerate(zip(queries, responses)):
    print(f"\n查询 {i+1}: {query}")
    print(f"回答: {response}")

批量查询示例展示了如何：

1. 异步查询：使用asyncio实现并发查询
2. 批量处理：同时处理多个查询请求
3. 性能提升：通过并发提高查询效率

7.结构化输出 #

你可能希望确保输出结构化。请参阅查询引擎 + Pydantic 输出了解如何从query engine类中提取Pydantic对象。

更多内容请见结构化输出指南。

8.创建自定义查询工作流 #

如果你想设计更复杂的查询流程，可以通过多种模块组合自定义查询工作流，涵盖从prompts/LLMs/output parsers到retrievers、response synthesizers，乃至你自己的定制组件。

详细内容请见工作流程指南。

9.总结 #

通过灵活配置查询引擎、检索器、后处理器和响应合成器，LlamaIndex 能够满足各种复杂的RAG应用查询需求。

9.1 主要特性： #

1. 简单查询：使用 index.as_query_engine() 快速创建查询引擎
2. 自定义检索：配置检索器的参数和策略
3. 后处理过滤：使用各种后处理器优化检索结果
4. 响应模式：选择适合的响应合成模式
5. 高级功能：支持多步骤查询、条件查询和批量查询
6. 性能优化：提供缓存和异步查询功能

9.2 最佳实践： #

• 根据应用场景选择合适的响应模式
• 使用后处理器提高查询结果质量
• 对重复查询启用缓存机制
• 对大量查询使用批量处理
• 根据数据特点选择合适的检索策略

10.安装指南 #

在运行上述代码之前，您需要安装必要的包：

# 安装LlamaIndex核心包
pip install llama-index

# 安装异步支持（用于批量查询）
pip install asyncio

10.1 包说明： #

• llama-index：LlamaIndex的核心包，包含所有基本的查询功能
• asyncio：Python的异步编程支持，用于批量查询

10.2 环境变量设置： #

如果您使用OpenAI的LLM，需要设置API密钥：

import os

# 设置OpenAI API密钥
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

请将 your-api-key-here 替换为您的实际OpenAI API密钥。