概述#

LlamaIndex（原名 GPT Index）是一个开源的数据框架，专门用于构建大语言模型（LLM）应用。它解决了 LLM 的一个核心局限性：LLM 在训练后就无法访问私有数据。LlamaIndex 通过检索增强生成（RAG）技术，将用户的私有数据与 LLM 的生成能力无缝连接。与langchain相比，langchian是编排工具流工作流等，而llamaindex专注于数据格式和数据索引。

LlamaIndex 在 LLM 应用开发栈中处于框架层，介于基础模型层和应用层之间。其内置了多种RAG必须的使用的范式，如：文本切分(Chunking)、向量化(Embedding)、检索(Retrieval)和生成(Generation)。

功能#

集成了300+的数据连接器：

多种索引策略：

• VectorStoreIndex: 向量存储索引（最常用，支持语义检索）
• TreeIndex: 树形索引（适合层级式文档）
• ListIndex: 列表索引（顺序检索）
• KeywordTableIndex: 关键词表索引（精确匹配）
• KnowledgeGraphIndex: 知识图谱索引（关系推理）

高级检索能力:

• 向量检索: 基于语义相似度的 top-k 检索
• 混合检索: 向量 + 关键词的融合检索
• 自动合并检索: RecursiveRetriever 自动合并相似节点
• 路由检索: RouterRetriever 根据查询类型选择不同的检索器
• 重排序: 检索后使用 Cohere/Rerank 等模型优化结果

多样的response合成模式：

LlamaIndex 主要用于构建以下类型的应用：

Embedding 调用流程#

流程说明：

1. Settings 触发 embed_nodes() ，在索引构建时被调用
2. embed_nodes 收集需要 embedding 的 nodes，筛选出尚未计算向量的节点
3. 调用 OpenAIEmbedding.get_text_embedding_batch() ，批量生成向量
4. 按 embed_batch_size 分批处理，避免单次请求过多文本
5. 对每个 batch 创建重试装饰器，处理网络错误和 API 限流
6. 调用 openai.embeddings.create() API，发送文本到 OpenAI 服务
7. API 返回 embeddings，List[List[float]] 格式的向量列表
8. 返回 Dict[node_id: embedding]，建立节点 ID 到向量的映射
9. 设置 node.embedding 属性，将向量附加到每个节点对象
10. 返回更新后的 nodes，完成向量化过程

核心意义：将文本转换为高维向量表示，使语义相似的文本在向量空间中距离相近，支撑后续的向量检索。

索引(Ingest + Structure)#

索引分为两阶段：阶段一负责数据加载与切分，阶段二负责索引构建与存储。

索引构建示例#

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI

# 1. 配置 Settings
Settings.embed_model = OpenAIEmbedding()
Settings.llm = OpenAI(model="gpt-4")

# 2. 加载文档
documents = SimpleDirectoryReader("./data").load_data()
# Shape: List[Document]

# 3. 创建索引
index = VectorStoreIndex.from_documents(documents)
# 内部: Documents -> Nodes -> Embeddings -> VectorStore + DocStore

# 4. 持久化
index.storage_context.persist("./storage")

阶段一：数据加载与切分#

阶段一流程说明：

1. 用户调用 from_documents(documents) 启动索引构建，传入原始文档列表

2. BaseIndex 创建 StorageContext，初始化 docstore 和 vector_store 用于存储

3. run_transformations(documents) 调用 NodeParser，将 Documents 转换为 Nodes

4. SentenceSplitter 执行文档切分：

   • split_text_metadata_aware(): 根据 metadata 长度调整 chunk_size
   • _split(): 按 separator（段落、句子）分段
   • _merge(): 处理 chunk_overlap，确保上下文连贯
   • build_nodes_from_splits(): 创建 TextNode 对象，建立关系链

5. 返回 List[TextNode]，包含切分后的文本节点，如下：

# 数据结构示例
[
    TextNode(
        id_="node_1",
        text="这是文档的第一段内容...",
        embedding=None,  # 尚未计算向量
        metadata={
            "file_name": "report.pdf",
            "page": 1,
            "chunk_index": 0
        },
        relationships={
            NodeRelationship.SOURCE: "doc_123",
            NodeRelationship.NEXT: "node_2"
        }
    ),
    TextNode(
        id_="node_2",
        text="这是文档的第二段内容...",
        # ... 其他字段
    ),
    # ... 更多节点
]

阶段二：索引构建与存储#

阶段二流程说明：

1. VectorStoreIndex 初始化，接收 List[TextNode]，此时 embedding 全部为 None

2. build_index_from_nodes() 启动索引构建流程

3. embed_nodes() 收集并计算 embeddings：

   • 筛选 embedding == None 的节点
   • 提取文本内容调用 embed_model.get_text_embedding_batch()
   • 返回 Dict[node_id: List[float]] 映射
   • 为每个节点设置 embedding 属性

4. vector_store.add() 存储向量：

   • 将 (node_id, embedding) 对存入向量数据库
   • 用于后续的相似度搜索

5. docstore.add_documents() 存储完整节点：

   • 将完整的 Node 对象（含 text、metadata）存入文档存储
   • 保持向量索引和文档内容的一致性

6. 创建 IndexDict 并保存到 index_store，记录索引元数据

7. 返回构建完成的 VectorStoreIndex 给用户

查询(Retrieve + Synthesize)#

查询流程分为两个阶段：阶段一负责检索相关节点，阶段二负责生成最终答案。

查询示例#

# 1. 创建查询引擎
query_engine = index.as_query_engine(
    response_mode="compact",  # 使用 COMPACT 模式
    similarity_top_k=5,
)

# 2. 执行查询
response = query_engine.query("什么是 RAG？")
# 内部: Query -> Embedding -> VectorSearch -> Synthesize -> Response

# 3. 获取结果
print(response.response)
print([node.node.metadata for node in response.source_nodes])

阶段一：向量检索 (Retrieve)#

阶段一流程说明：

1. 用户调用 query_engine.query() 发起查询请求，传入问题文本
2. query_engine 创建 QueryBundle，封装查询字符串和可选的向量表示
3. retriever.retrieve() 调用 VectorIndexRetriever 执行向量检索
4. embed_model 为查询生成 embedding（如果尚未计算），将文本转换为向量空间
5. 构建 VectorStoreQuery，包含查询向量、top-k 参数、过滤条件
6. vector_store.query() 计算余弦相似度，在向量空间中找到最相关的 top-k 节点
7. 返回 VectorStoreQueryResult，包含 node_id 列表和相似度分数
8. docstore.get_nodes() 获取完整节点内容，根据 node_id 补充完整文本和元数据
9. 创建 NodeWithScore 对象，将节点和相似度分数封装在一起
10. 返回 List[NodeWithScore]，作为检索结果传递给后续的合成阶段

核心意义：通过向量相似度在向量空间中找到与用户问题语义最相关的内容。

阶段二：答案合成 (Synthesize)#

阶段二流程说明：

1. query_engine 调用 response_synthesizer.synthesize() ，传入 QueryBundle 和检索到的节点

2. BaseSynthesizer 提取节点内容，从 NodeWithScore 中提取 text 字段

3. get_response() 遍历每个 text_chunk，对每个检索到的文本块进行处理

4. 首次调用使用 text_qa_template：

   • partial_format(query_str) 填充查询字符串
   • prompt_helper.repack() 优化 chunk 大小
   • llm.predict(template, context_str) 生成初始答案

5. 后续调用使用 refine_template：

   • partial_format(query_str, existing_answer) 填充查询和已有答案
   • 检查 available_chunk_size 确保不超限
   • llm.predict(refine_template) 生成优化后的答案

6. 合成最终答案，将所有 chunk 的处理结果合并

7. 构建 Response 对象，包含答案文本、引用的源节点和元数据

8. 返回 Response 给用户

核心意义：通过检索增强生成（RAG），结合向量检索找到相关上下文，再通过 LLM 生成准确、有依据的答案。

Response Mode 的执行差异#

在查询的Synthesize阶段中，最重要的就是把get_response() 遍历每个 text_chunk的时候，按照什么方式来调用LLM，这里有四种模式的response合成，其差异在概述里提到过。

REFINE 模式#

流程说明：

1. get_response() 遍历 text_chunks，对每个检索到的文本块进行处理
2. 第一个 chunk 调用首次处理逻辑，使用 text_qa_template 生成初始答案，首次调用的template和后续调用的template不一样
3. text_qa_template.partial_format() 填充查询字符串，准备 QA 模板
4. prompt_helper.repack() 优化 chunk 大小，确保不超过 context window
5. llm.predict() 生成初始答案，基于第一个 chunk 的内容
6. 后续 chunk 调用优化逻辑，使用 refine_template 逐步改进答案
7. refine_template.partial_format() 填充现有答案，将已有答案作为上下文
8. 检查 available_chunk_size，确保模板和上下文能放入 context window
9. llm.predict() 生成优化后的答案，结合当前 chunk 和之前的答案
10. 返回最终更新后的 response

COMPACT模式#

流程说明：

1. get_response() 调用 makecompact_text_chunks() ，执行智能合并逻辑
2. 分别格式化两种模板，text_qa_template 和 refine_template 都需要考虑
3. get_biggest_prompt() 选择最大的模板，确保合并后的 chunks 能放入任何一个模板
4. prompt_helper.repack() 合并小 chunks，将多个小文本块重新打包成大块
5. 返回数量减少的 new_texts，减少后续 LLM 调用次数
6. 调用父类 Refine.get_response() ，在合并后的大 chunks 上执行迭代优化

核心意义：通过预先合并减少 LLM 调用次数，在保证答案质量的同时提升性能，是默认推荐的模式。

TREE_SUMMARIZE 模式#

流程说明：

1. get_response() 调用 prompt_helper.repack() ，将原始 chunks 重新打包
2. 返回多个大 chunks，例如 10 个原始 chunks 变成 10 个大 chunks
3. Level 0: 并行调用 llm.predict() ，对每个大 chunk 生成摘要，可并行执行
4. 收集 10 个 summaries，作为下一层的输入
5. Level 1: 递归调用 get_response() ，处理 summaries 列表
6. 再次 repack() ，将 10 个 summaries 合并为 3 个更大的 chunks
7. 并行调用 llm.predict() ，生成 3 个二级 summaries
8. Level 2: 再次递归，处理 3 个 summaries
9. 达到 base case，len(chunks) == 1 时执行最终调用
10. llm.predict() 生成 final_answer

核心意义：树形递归结构，每层内可并行处理，适合大量节点的汇总场景，平衡了速度和质量。

SIMPLE_SUMMARIZE 模式#

流程说明：

1. get_response() 调用 join 操作，使用 "\n" 连接所有 text_chunks
2. 生成 single_text_chunk，合并所有检索到的文本
3. prompt_helper.truncate() 检查并截断，如果超过 context window 则截断
4. 返回 truncated_chunk，确保能放入 LLM 的上下文窗口
5. llm.predict() 单次调用生成 response，一次性处理所有上下文

核心意义：最简单直接的方式，单次 LLM 调用，但超长上下文会被截断，可能丢失信息。

LLM 调用流程#

流程说明：

1. Settings 触发 LLM.predict() ，response_synthesizer 发起 LLM 调用
2. LLM.predict 调用 getmessages() ，将 prompt 转换为消息格式
3. prompt.format_messages() 填充模板变量，调用 mapall_vars 处理参数
4. 填充 template 变量，将 kwargs 中的值替换到模板占位符
5. 返回 List[ChatMessage]，格式化后的消息列表
6. LLM.predict 调用 self.chat(messages) ，执行聊天 API 调用
7. @llm_chat_callback 装饰器包装调用，添加事件追踪和监控
8. callback_manager.on_event_start() 触发开始事件，记录 LLM 调用开始
9. 调用 openai.chat.completions.create() API，发送消息到 OpenAI 服务
10. API 返回 ChatCompletion，包含生成的响应和元数据
11. callback_manager.on_event_end() 触发结束事件，记录 LLM 调用结束
12. 提取 message.content，从 ChatResponse 中获取生成的文本
13. 返回 str 格式的响应，给调用方

核心意义：统一封装 LLM 调用接口，支持多种 LLM 提供商，通过 callback 机制实现可观测性，为 RAG 系统提供生成能力。

第三章：关键数据结构#

3.1 Document#

class Document(Node):
    """文档对象，表示一个完整的数据源（如一个文件）"""

    id_: str                               # 唯一标识符
    text_resource: MediaResource           # 文本内容和元数据
    metadata: Dict[str, Any]               # 文件路径、创建时间等
    excluded_embed_metadata_keys: List[str]  # Embedding 时排除的元数据
    excluded_llm_metadata_keys: List[str]     # LLM 时排除的元数据
    embedding: Optional[List[float]]       # 延迟计算的向量

示例:

Document(
    id_="doc_123",
    text_resource=MediaResource(text="这是一篇文档的内容..."),
    metadata={"file_name": "report.pdf", "page": 1}
)

3.2 TextNode#

class TextNode(BaseNode):
    """文本节点，表示文档切分后的一个语义单元"""

    id_: str                               # 节点唯一标识
    text: str                              # 节点文本内容
    embedding: Optional[List[float]]       # 向量表示
    metadata: Dict[str, Any]               # 继承自 Document
    relationships: Dict[NodeRelationship, RelatedNodeInfo]
    start_char_idx: Optional[int]          # 在原文档中的起始位置
    end_char_idx: Optional[int]            # 在原文档中的结束位置

Relationships 类型:

• SOURCE: 指向原 Document
• PREVIOUS: 指向前一个节点
• NEXT: 指向后一个节点
• PARENT: 指向父节点

3.3 NodeWithScore#

class NodeWithScore(BaseComponent):
    """带分数的节点，用于检索结果"""

    node: BaseNode                         # 节点对象
    score: Optional[float]                 # 相似度分数或其他评分

3.4 QueryBundle#

class QueryBundle(DataClassJsonMixin):
    """查询对象，封装查询字符串和其向量表示"""

    query_str: str                         # 查询字符串
    embedding: Optional[List[float]]       # 延迟计算的查询向量
    custom_embedding_strs: Optional[List[str]]  # 自定义嵌入字符串

3.5 Response#

class Response:
    """查询响应对象"""

    response: str                          # 生成的答案
    source_nodes: List[NodeWithScore]      # 引用的节点
    metadata: Optional[Dict[str, Any]]     # 额外元数据

3.6 IndexStruct#

class IndexDict(IndexStruct):
    """向量索引结构元数据"""

    index_id: str                          # 索引ID
    summary: Optional[str]                 # 索引摘要
    nodes_dict: Dict[str, IndexNode]       # 节点映射
    doc_id_dict: Dict[str, List[str]]      # 文档到节点的映射

第四章：源码文件映射#