LlamaIndex 学习笔记

git地址

https://github.com/zxliucn/LlamaIndex-Learning

目录

1. 文档加载 (DataLoader)
2. 节点解析器 (NodeParser)
3. 嵌入模型 (Embedding Model)
4. 索引与存储 (Index)
5. 检索器 (Retriever)
6. 索引、向量数据与数据库的关系

一、文档加载 (DataLoader)

文档加载是 LlamaIndex 构建 RAG 应用的第一步，也是最基础的一步。本文将全面介绍 LlamaIndex 中文档加载的各种方式、核心概念以及高级自定义技巧。

1.1 核心概念：Document

在 LlamaIndex 中，Document（文档）是对原始数据源的标准化封装，是整个 RAG 流程的数据入口。它不仅仅指传统的文本文档，而是涵盖所有可被加载的结构化/非结构化数据的抽象载体。

Document 的核心属性

Document 的关键特性

• 多源适配性：LlamaIndex 提供 100+ 内置 Data Loader，可将不同数据源统一为 Document 对象
• 不可变性：原始 Document 加载后内容不会被修改（分割操作会生成新的 Node）
• 编码保障：加载中文文档时需通过 encoding="utf-8" 指定编码

1.2 文档加载的三种方式

方式一：使用 SimpleDirectoryReader（最常用）

SimpleDirectoryReader 是最简单易用的加载器，可以将指定目录中的文件批量加载为 Document 对象：

from llama_index.core import SimpleDirectoryReader

# 基本用法：加载目录下所有文件
documents = SimpleDirectoryReader("./data").load_data()

# 指定编码（处理中文文档）
documents = SimpleDirectoryReader("./data", encoding="utf-8").load_data()

# 查看加载结果
print(f"加载文档数量：{len(documents)}")
print(f"第一个文档内容长度：{len(documents[0].text)}字符")
print(f"文档元数据：{documents[0].metadata}")

支持的文件格式：Markdown、PDF、Word 文档、PowerPoint 演示文稿、图像、音频和视频等。

方式二：使用 LlamaHub 的数据加载器

LlamaHub 是 LlamaIndex 的数据连接器注册表，提供了数百种连接器，支持从各种数据源加载数据：

from llama_index.core import download_loader
from llama_index.readers.database import DatabaseReader

# 示例：从 SQL 数据库加载数据
reader = DatabaseReader(
    scheme="postgresql",
    host="localhost",
    port=5432,
    user="myuser",
    password="mypassword",
    dbname="mydb",
)

query = "SELECT * FROM users"
documents = reader.load_data(query=query)

方式三：手动构造 Document

from llama_index.core import Document

# 从文本列表创建
text_list = ["文本1内容", "文本2内容", "文本3内容"]
documents = [Document(text=t) for t in text_list]

# 创建单个文档
document = Document(text="这是文档的原始文本内容")

# 快速创建示例文档（用于原型开发）
example_doc = Document.example()

1.3 元数据（Metadata）管理

元数据是 Document 的重要组成部分，用于记录文档的上下文信息，对回答溯源和多文档管理至关重要。

设置元数据

方式一：在构造函数中设置

document = Document(
    text="文档内容",
    metadata={
        "filename": "important_doc.pdf",
        "category": "技术文档",
        "author": "张三",
        "creation_date": "2024-01-15"
    }
)

方式二：创建后设置

document = Document(text="文档内容")
document.metadata = {"filename": "doc.txt", "category": "指南"}

方式三：使用 SimpleDirectoryReader 自动设置

from llama_index.core import SimpleDirectoryReader

# 定义元数据生成函数
filename_fn = lambda filename: {"file_name": filename, "source": "local"}

documents = SimpleDirectoryReader(
    "./data", 
    file_metadata=filename_fn
).load_data()

自定义文档 ID

# 自动将文件路径作为 ID
documents = SimpleDirectoryReader("./data", filename_as_id=True).load_data()

# 手动设置 ID
document.doc_id = "my_custom_document_id"

高级元数据自定义

from llama_index.core import Document
from llama_index.core.schema import MetadataMode

document = Document(
    text="这是一个需要处理的文档",
    metadata={
        "file_name": "secret_document.txt",
        "category": "finance",
        "author": "LlamaIndex"
    },
    excluded_llm_metadata_keys=["file_name"],
    excluded_embed_metadata_keys=["file_name"],
)

# 查看 LLM 实际看到的内容
print(document.get_content(metadata_mode=MetadataMode.LLM))

# 查看 Embedding 模型实际看到的内容
print(document.get_content(metadata_mode=MetadataMode.EMBED))

自定义元数据格式

document = Document(
    text="文档内容",
    metadata={"key1": "value1", "key2": "value2"},
    metadata_seperator=" | ",           # 键值对之间的分隔符
    metadata_template="{key}: {value}",  # 每个键值对的格式
    text_template="元数据：{metadata_str}\n---\n内容：{content}"
)

1.4 批量加载与高级用法

批量加载目录

from llama_index.core import SimpleDirectoryReader

documents = SimpleDirectoryReader(
    input_dir="./docs",
    recursive=True,           # 递归读取子目录
    required_exts=[".pdf", ".txt"],  # 只加载特定扩展名
    exclude=["temp", "draft"]         # 排除特定目录
).load_data()

使用 DocStore 管理多个索引

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, SummaryIndex
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core import StorageContext
from llama_index.core.node_parser import SentenceSplitter

documents = SimpleDirectoryReader("./data").load_data()
nodes = SentenceSplitter().get_nodes_from_documents(documents)

docstore = SimpleDocumentStore()
docstore.add_documents(nodes)
storage_context = StorageContext.from_defaults(docstore=docstore)

# 多个索引共享同一个文档存储
vector_index = VectorStoreIndex(nodes, storage_context=storage_context)
summary_index = SummaryIndex(nodes, storage_context=storage_context)

1.5 完整示例：从加载到索引

from llama_index.core import (
    SimpleDirectoryReader, 
    VectorStoreIndex,
    Settings
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI

# 1. 配置 LLM
Settings.llm = OpenAI(model="gpt-3.5-turbo", temperature=0)

# 2. 加载文档
documents = SimpleDirectoryReader(
    "./data",
    encoding="utf-8",
    filename_as_id=True
).load_data()

# 3. 设置文本分割
Settings.chunk_size = 512
Settings.chunk_overlap = 20

# 4. 构建索引
index = VectorStoreIndex.from_documents(documents)

# 5. 查询测试
query_engine = index.as_query_engine()
response = query_engine.query("你的问题在这里")
print(response)

1.6 常见问题与最佳实践

二、节点解析器 (NodeParser)

2.1 核心概念：Document vs Node

• Document：原始文档对象，可以是一整个文件或一段文本
• Node：文档切分后的"块"，包含文本内容、元数据以及节点间的关系信息（如前后节点、父节点引用）

from llama_index.core.schema import TextNode, NodeRelationship, RelatedNodeInfo

node1 = TextNode(text="这是第一个文本块", id_="node_1")
node2 = TextNode(text="这是第二个文本块", id_="node_2")

node1.relationships[NodeRelationship.NEXT] = RelatedNodeInfo(node_id=node2.node_id)
node2.relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(node_id=node1.node_id)

2.2 什么数据需要切分？什么时候不需要？

需要切分的数据类型

不需要切分的数据类型

半结构化数据：混合处理策略

# 表格部分：保留原结构，不走切分
# 文本部分：切分后建向量索引
from llama_index.core.node_parser import SentenceSplitter

text_nodes = SentenceSplitter().get_nodes_from_documents(text_docs)
vector_index = VectorStoreIndex(text_nodes)

sql_query_engine = NLSQLTableQueryEngine(sql_database)

核心原则

2.3 自动切分 vs 手动切分

VectorStoreIndex 的自动切分

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader("./data/").load_data()
index = VectorStoreIndex.from_documents(documents)  # 自动完成切分！

控制自动切分：

from llama_index.core import Settings
from llama_index.core.node_parser import SentenceSplitter

# 方式一：通过全局 Settings 配置
Settings.text_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
index = VectorStoreIndex.from_documents(documents)

# 方式二：通过 transformations 参数
text_splitter = SentenceSplitter(chunk_size=500, chunk_overlap=50)
index = VectorStoreIndex.from_documents(documents, transformations=[text_splitter])

手动切分的优势

渐进式优化建议

# 阶段一：快速验证（自动切分）
index = VectorStoreIndex.from_documents(documents)

# 阶段二：发现问题后切换手动切分
parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)

# 阶段三：手动增强元数据
for i, node in enumerate(nodes):
    node.metadata["chunk_index"] = i
    node.metadata["source_file"] = documents[0].metadata.get("file_name")
    if "summary" in node.text.lower():
        node.metadata["is_summary"] = True

index = VectorStoreIndex(nodes)

2.4 最常用的切分方式：SentenceSplitter

from llama_index.core.node_parser import SentenceSplitter

# 创建切分器
parser = SentenceSplitter(chunk_size=512, chunk_overlap=20)

# 执行切分
nodes = parser.get_nodes_from_documents(documents)

# 全局配置
from llama_index.core import Settings
Settings.chunk_size = 512
Settings.chunk_overlap = 20

2.5 进阶切分器

MarkdownNodeParser - 保留 Markdown 结构

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
nodes = parser.get_nodes_from_documents(documents)

CodeSplitter - 代码专用切分

from llama_index.core.node_parser import CodeSplitter

parser = CodeSplitter(
    language="python",
    chunk_lines=40,
    chunk_lines_overlap=15,
    max_chars=1500,
)
nodes = parser.get_nodes_from_documents(documents)

SemanticSplitterNodeParser - 语义切分

from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding

embed_model = OpenAIEmbedding()

parser = SemanticSplitterNodeParser(
    embed_model=embed_model,
    buffer_size=1,
    breakpoint_percentile_threshold=95,
)
nodes = parser.get_nodes_from_documents(documents)

2.6 使用 IngestionPipeline

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=20),
        TitleExtractor(),
    ]
)
nodes = pipeline.run(documents=documents, show_progress=True)

2.7 第三方切分器

DashScopeJsonNodeParser（阿里通义）

from llama_index.node_parser.dashscope import DashScopeJsonNodeParser

node_parser = DashScopeJsonNodeParser(
    chunk_size=100, 
    overlap_size=0,
    separator=" |,|，|。|？|！|\n|\?|\!",
    language="cn"
)

2.8 切分策略总结

💡 核心提示：当你发现检索结果不够好时，手动切分往往是第一个需要优化的地方。

三、嵌入模型 (Embedding Model)

Embedding 模型将文本转换为向量表示，直接影响检索质量。

3.1 安装与基础配置

pip install llama-index-embeddings-openai        # OpenAI
pip install llama-index-embeddings-huggingface    # Hugging Face
pip install llama-index-embeddings-ollama         # Ollama
pip install BCEmbedding                            # 中文推荐

全局配置

from llama_index.core import Settings
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.embed_model = OpenAIEmbedding()

3.2 各主要来源 Embedding 模型

OpenAI Embedding

from llama_index.embeddings.openai import OpenAIEmbedding

embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",  # 或 text-embedding-3-large
    embed_batch_size=10,
)

Hugging Face 本地模型

from llama_index.embeddings.huggingface import HuggingFaceEmbedding

# 轻量级
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")

# 高性能
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-large-en-v1.5")

Ollama 本地部署

from llama_index.embeddings.ollama import OllamaEmbedding

embed_model = OllamaEmbedding(
    model_name="nomic-embed-text",
    base_url="http://localhost:11434",
    query_instruction="Represent the question for retrieving supporting documents:",
    text_instruction="Represent the document for retrieval:",
)

ONNX 加速

from llama_index.embeddings.huggingface_optimum import OptimumEmbedding

OptimumEmbedding.create_and_save_optimum_model("BAAI/bge-small-en-v1.5", "./bge_onnx")
Settings.embed_model = OptimumEmbedding(folder_name="./bge_onnx")

3.3 中文 Embedding 模型推荐

重点推荐：BCEmbedding

from llama_index.embeddings.huggingface import HuggingFaceEmbedding

embed_model = HuggingFaceEmbedding(
    model_name="maidalun1020/bce-embedding-base_v1",
    max_length=512,
    embed_batch_size=32,
    device="cuda"
)

配合 Reranker（强烈推荐）：

from BCEmbedding.tools.llama_index import BCERerank

reranker = BCERerank(
    model="maidalun1020/bce-reranker-base_v1",
    top_n=5,
    device="cuda"
)
query_engine = index.as_query_engine(node_postprocessors=[reranker])

中文场景决策树

纯中文 + 本地部署 → M3E-Turbo
纯中文 + 高精度 → xiaobu-embedding-v2
中英双语 + RAG → bce-embedding-base_v1（首选）
多语言/长文档 → BGE-M3
图像+文本混合 → Qwen3-VL-2B

3.4 模型选择与性能评测

评测指标

• Hit Rate（命中率）：正确答案出现在 top-k 检索结果中的查询比例
• MRR（平均倒数排名）：第一个相关文档排名倒数的平均值

各模型性能对比

关键结论：Reranker 几乎总是能显著提升效果。

3.5 完整代码示例

基础 RAG Pipeline

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.node_parser import SentenceSplitter

Settings.embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-large-en-v1.5")
Settings.text_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)

documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents, show_progress=True)
index.storage_context.persist(persist_dir="./storage")

query_engine = index.as_query_engine()
response = query_engine.query("你的问题")

中文 RAG 完整示例

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, Settings
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.node_parser import SentenceSplitter
from BCEmbedding.tools.llama_index import BCERerank

Settings.embed_model = HuggingFaceEmbedding(
    model_name="maidalun1020/bce-embedding-base_v1", device="cuda"
)
Settings.text_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)

documents = SimpleDirectoryReader("./chinese_docs").load_data()
index = VectorStoreIndex.from_documents(documents, show_progress=True)

reranker = BCERerank(model="maidalun1020/bce-reranker-base_v1", top_n=3, device="cuda")
query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[reranker],
)
response = query_engine.query("《长安的荔枝》的主角是谁？")

四、索引与存储 (Index)

4.1 索引概述

索引（Index） 是核心数据结构，将原始 Document 对象组织起来，以便 LLM 高效查询。构建索引包括将文档解析、分块为 Node 对象。

4.2 核心索引类型

4.3 存储机制

核心存储组件

方案一：本地磁盘持久化

from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage

# 持久化
index.storage_context.persist(persist_dir="./my_storage")

# 加载
if os.path.exists("./my_storage"):
    storage_context = StorageContext.from_defaults(persist_dir="./my_storage")
    index = load_index_from_storage(storage_context)
else:
    index = VectorStoreIndex.from_documents(documents)
    index.storage_context.persist(persist_dir="./my_storage")

方案二：向量数据库存储（生产环境首选）

import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore

db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("my_documents")

vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex.from_documents(documents, storage_context=storage_context)

# 重启后直接加载（无需重新嵌入）
loaded_index = VectorStoreIndex.from_vector_store(vector_store)

动态更新索引

new_documents = SimpleDirectoryReader("new_data").load_data()
for doc in new_documents:
    index.insert(doc)

4.4 Agent 记忆存储

• 短期记忆：SQLite 存储，维护当前会话上下文

• 长期记忆：三种记忆块：

  1. StaticMemoryBlock：静态信息（姓名、偏好）
  2. FactExtractionMemoryBlock：LLM 驱动自动提取事实
  3. VectorMemoryBlock：历史对话嵌入向量数据库

4.5 最佳实践

五、检索器 (Retriever)

5.1 Retriever 是什么？

Retriever（检索器） 负责根据用户查询获取最相关的上下文内容，是连接索引和查询引擎的核心桥梁。

用户查询 → Retriever → Response Synthesizer → 返回响应

5.2 快速上手

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)

# 获取检索器
retriever = index.as_retriever(similarity_top_k=5)

# 执行检索
nodes = retriever.retrieve("Who is Paul Graham?")

5.3 索引与 Retriever 的关系

5.4 高阶检索技术

递归检索器（Recursive Retriever）

解耦检索与合成，通过文档代理实现复杂操作：

from llama_index.core.schema import IndexNode
from llama_index.agent.openai import OpenAIAgent

# 为每个文档构建代理
for wiki_title in wiki_titles:
    vector_index = VectorStoreIndex.from_documents(docs)
    summary_index = SummaryIndex.from_documents(docs)
    
    tools = [
        QueryEngineTool(query_engine=vector_index.as_query_engine(),
                        metadata=ToolMetadata(name="vector_tool", description="搜索具体事实")),
        QueryEngineTool(query_engine=summary_index.as_query_engine(),
                        metadata=ToolMetadata(name="summary_tool", description="总结内容")),
    ]
    agent = OpenAIAgent.from_tools(tools)

# 构建顶层检索器，查询会自动路由到正确的文档代理

混合检索

结合关键词匹配（BM25）和语义向量搜索：

class HybridRetriever(BaseRetriever):
    def __init__(self, vector_retriever, keyword_retriever, mode="AND"):
        self.vector_retriever = vector_retriever
        self.keyword_retriever = keyword_retriever
        self.mode = mode
    
    def _retrieve(self, query_bundle):
        vector_nodes = self.vector_retriever.retrieve(query_bundle)
        keyword_nodes = self.keyword_retriever.retrieve(query_bundle)
        
        vector_ids = {n.node.node_id for n in vector_nodes}
        keyword_ids = {n.node.node_id for n in keyword_nodes}
        
        combined = {n.node.node_id: n for n in vector_nodes}
        combined.update({n.node.node_id: n for n in keyword_nodes})
        
        if self.mode == "AND":
            result_ids = vector_ids.intersection(keyword_ids)
        else:
            result_ids = vector_ids.union(keyword_ids)
        
        return [combined[rid] for rid in result_ids]

检索器路由（Retriever Router）

通过向量检索动态选择最相关的查询引擎：

from llama_index.core.query_engine import ToolRetrieverRouterQueryEngine

router_query_engine = ToolRetrieverRouterQueryEngine(obj_index.as_retriever())
response = router_query_engine.query("What did the author do at Y Combinator?")

5.5 RetrieverQueryEngine

from llama_index.core.query_engine import RetrieverQueryEngine

query_engine = RetrieverQueryEngine.from_args(
    retriever=my_retriever,
    response_mode="compact",
    node_postprocessors=[reranker],
)

支持的后处理组件

5.6 性能优化

场景选型

5.7 设计哲学：渐进式复杂度

简单：index.as_retriever() → 一行代码
  ↓
进阶：配置 similarity_top_k、retriever_mode
  ↓
高级：自定义 Retriever 类、混合检索
  ↓
专业：递归检索、检索器路由、多 Agent 协作

六、索引、向量数据与数据库的关系

6.1 概念澄清

关键理解：

• 向量是数据本身
• 索引是数据结构（帮助快速检索）
• 数据库是载体（持久化存储向量+索引+元数据）

6.2 LlamaIndex 的三层存储架构

from llama_index.core import StorageContext

storage_context = StorageContext.from_defaults(
    docstore=...      # 文档存储：存原始文本和节点
    vector_store=...  # 向量存储：存向量嵌入
    index_store=...   # 索引存储：存索引元数据
)

6.3 不同存储方式对比

本地文件系统（SimpleVectorStore）

./storage/
├── docstore.json        # 所有原始文本
├── vector_store.json    # 所有向量数据（JSON格式，效率低）
└── index_store.json     # 索引元数据

• 适合小规模数据（<1万条文档）

专业向量数据库（Chroma）

chroma_db/
├── chroma.sqlite3       # 元数据、文档ID
└── data_level0.bin      # HNSW索引 + 向量数据

• 向量、索引、元数据分离但关联存储

6.4 数据流向

构建索引阶段：
  文档 → 文本分块 + 嵌入模型 → VectorStore(存向量+NodeID) + DocStore(存文本+NodeID)

查询阶段：
  查询文本 → 查询向量 → VectorStore(相似度搜索) → 返回NodeID列表
    → DocStore(根据NodeID获取原始文本) → 返回结果

6.5 常见误区

6.6 选择决策指南

总结

最佳实践：

• 小项目：用 LlamaIndex 本地文件系统
• 生产环境：用专业向量数据库（Chroma/Milvus/Qdrant）
• 超大规模：向量存对象存储，索引存 Redis 缓存