前言：为什么有了 LangChain 还需要 LlamaIndex？

在构建 LLM（大语言模型）应用时，开发者最常面临的两个挑战是：

  1. 如何让模型按既定逻辑办事？ (Agent/Chain 编排)
  2. 如何让模型准确理解私有数据？ (RAG/Data 检索)

LangChain 也是一把瑞士军刀，它起初侧重于解决第一个问题（编排），虽然现在也有检索功能，但其核心基因在于“通过链（Chain）将组件连接起来”。

LlamaIndex (原 GPT Index) 则专注于解决第二个问题（数据）。它是一个数据框架，专门用于将自定义数据源连接到大语言模型。

LangChain 可以自己搞，就像你可以用 Python 原生代码写一个 Web 服务器，但你通常会用 Django 或 FastAPI。

LlamaIndex 的核心优势在于：

  1. 数据处理的颗粒度：LlamaIndex 提供了极其精细的数据分块（Chunking）、节点（Node）管理和元数据提取策略，这是 LangChain 的通用 Retriever 难以比拟的。
  2. 索引结构的丰富性：LangChain 主要是向量检索（Vector Store）。而 LlamaIndex 支持树状索引 (Tree Index)、关键词表索引 (Keyword Table Index)、知识图谱索引 (Knowledge Graph Index) 以及组合索引 (Composable Indices)。这对于复杂的查询（如“总结全书”或“对比两个文档的差异”）至关重要。
  3. 检索策略的优化：LlamaIndex 拥有更高级的检索后处理（Post-processing）和重排序（Re-ranking）机制，能显著提高 RAG（检索增强生成）的准确率。

第一部分：LangChain vs LlamaIndex —— 深度对比

1.1 定位差异

特性 LangChain LlamaIndex
核心隐喻 管道/胶水 (Glue) 图书馆管理员 (Librarian)
主要关注点 交互流程、Agent 行为、多组件编排 数据摄入、数据索引、数据检索策略
最强项 打造全能 Agent、工具调用、记忆管理 处理海量、杂乱、非结构化数据的 RAG 优化
数据结构 主要是 Document 对象，相对扁平 主要是 Nodes (节点) 和 Indices (索引)，支持层级关系

1.2 哪些问题 LangChain “搞不定” (或很难搞)？

虽然 LangChain 几乎是图灵完备的，但在以下场景中，使用 LlamaIndex 是降维打击：

  1. 跨文档的复杂推理：

    • 场景：你有 5 个 PDF，需要问“对比这 5 个文件中提到的财务策略有何不同？”
    • LangChain：通常只是把所有相关片段找出来塞给 LLM，容易超出 Context Window 或造成幻觉。
    • LlamaIndex：可以使用 SubQuestionQueryEngine 把大问题拆解为针对每个文档的小问题，或者使用 Tree Index 进行层级摘要，最终汇总答案。

  2. 结构化与非结构化混合查询：

    • 场景：想在 SQL 数据库和 PDF 文档中同时查找信息。
    • LlamaIndex：拥有原生的 Text-to-SQL 和 RouterQueryEngine，可以智能判断该去查数据库还是查向量库，甚至同时查并合并结果。

  3. 索引更新成本：

    • 场景：文档库每天新增 100 个文件。
    • LlamaIndex：索引结构设计之初就考虑了增量更新（Insertion），维护成本更低。

第二部分：LlamaIndex 完整功能清单

在深入学习之前，我们先全面了解 LlamaIndex 的能力矩阵。LlamaIndex 不仅仅是一个 RAG 库，它是一个完整的数据到 LLM 的桥梁生态系统。

2.1 核心功能模块

📦 数据连接器 (Data Connectors)

LlamaIndex 支持 100+ 种数据源，远超 LangChain：

数据源类型 连接器示例 用途
文件系统 SimpleDirectoryReader, MarkdownReader, PDFReader 读取本地文件
数据库 PostgresReader, MongoReader, SQLDatabaseReader 连接关系型/非关系型数据库
云存储 S3Reader, GCSReader, AzureBlobStorageReader 读取云存储文件
API 数据 NotionReader, SlackReader, DiscordReader 读取第三方服务数据
网页 BeautifulSoupWebReader, TrafilaturaWebReader 爬取网页内容
代码仓库 GithubRepositoryReader 读取 GitHub 代码
结构化数据 PandasCSVReader, ExcelReader 处理表格数据
音视频 YoutubeTranscriptReader, WhisperTranscriber 处理多媒体内容
🔍 索引类型 (Index Types)

这是 LlamaIndex 的核心竞争力，LangChain 主要只有向量检索：

索引类型 适用场景 优势
VectorStoreIndex 语义相似度检索 快速、准确，适合大多数 RAG 场景
TreeIndex 层级摘要、文档总结 自底向上构建摘要树，适合"总结全书"类查询
KeywordTableIndex 关键词精确匹配 不依赖 Embedding，适合结构化查询
KnowledgeGraphIndex 实体关系查询 提取实体和关系，支持"谁和谁有关系"类问题
DocumentSummaryIndex 多文档摘要检索 预先为每个文档生成摘要，检索时先查摘要
ComposableGraphIndex 组合多种索引 将多个索引组合，实现复杂查询策略
PandasIndex 表格数据查询 直接查询 DataFrame，支持自然语言转 SQL
🛠️ 查询引擎 (Query Engines)

不同的查询引擎解决不同的查询需求：

查询引擎 功能 示例场景
RetrieverQueryEngine 检索+生成 标准 RAG 流程
RouterQueryEngine 智能路由 根据问题选择不同的数据源
SubQuestionQueryEngine 子问题分解 “对比5个文档的差异” → 拆解为5个子问题
MultiStepQueryEngine 多步推理 需要多轮检索和推理的复杂问题
TransformQueryEngine 查询转换 在检索前改写查询，提高召回率
RetrieverQueryEngine + ReRanking 重排序 检索后对结果重新排序，提高准确率
✂️ 切块方式 (Node Parsers)

不同的切块策略决定了 LLM 能看到什么样的上下文：

切块方式 (Node Parser) 核心原理 适用场景
SentenceSplitter (默认) 句子边界保持
尽量在句号、换行符处切断，保持句子完整性。		 通用场景
大多数普通文本、文章、PDF 转文本的首选。
TokenTextSplitter 硬切分
严格按照 Token 数量强制切断，不考虑语法结构。		 严格限制窗口
对上下文长度极其敏感，或文本本身无标点符号。
MarkdownNodeParser 结构化切分
根据 Markdown 标题层级 (#, ##) 切分，保留父级标题作为元数据。		 技术文档
README、开发手册、Notion 导出笔记。
JSONNodeParser 层级保留
解析 JSON 结构，保持 Key-Value 的层级关系不被打散。		 结构化数据
API 返回结果、NoSQL 数据库导出内容。
SemanticSplitterNodeParser 语义聚合
利用 Embedding 模型判断前后句相似度，在话题转变的地方切分。		 长文/论文
确保每一块都在讲同一个完整的主题，避免语义断裂。
HierarchicalNodeParser 父子节点 (Small-to-Big)
生成“父节点”和对应的“子节点”。索引子节点，返回父节点。		 复杂长文档
配合 AutoMergingRetriever，既要精准匹配细节，又要看大段上下文。
SentenceWindowNodeParser 句子窗口
切成单句，但元数据里附带前后 N 句的内容。		 高精度问答
配合 MetadataReplacementPostProcessor，极大减少幻觉。
🧩 节点与文档处理 (Node Processing)

LlamaIndex 的节点系统比 LangChain 的 Document 更强大：

  • Node 元数据：每个节点可以携带丰富的元数据（文件名、页码、创建时间等）
  • 节点关系：自动维护 Previous/Next 关系，保留上下文
  • 自定义解析器：支持代码、Markdown、HTML 等结构化解析
  • 后处理：去重、过滤、合并等操作
🔄 数据更新与维护 (Data Management)
  • 增量更新：只更新变化的文档，无需重建整个索引
  • 版本控制：支持索引版本管理
  • 数据清理：自动清理过期或无效数据
🎯 评估与优化 (Evaluation)
  • 检索评估：评估检索质量（Hit Rate, MRR 等）
  • 生成评估：评估生成质量（相关性、准确性）
  • 端到端评估：评估整个 RAG 流程的效果
🤖 Agent 集成
  • ReAct Agent：基于 LlamaIndex 的检索能力构建 Agent
  • OpenAI Agent：与 OpenAI Function Calling 集成
  • LangChain Agent：作为 LangChain Agent 的工具

第三部分：LlamaIndex 入门 (Python 实现)

我们将基于 LlamaIndex v0.10+ (引入了 llama-index-core 命名空间) 和 Python 3.10+ 进行演示。

3.1 环境准备

版本说明

本文基于以下版本编写：

  • LlamaIndex: v0.10+ (最新稳定版)
  • Python: 3.10+
  • LangChain: 1.0+ (如需要集成)
安装步骤
# 1. 安装核心库（最新版本）
pip install llama-index

# 2. 安装 LLM 集成（选择你需要的）
pip install llama-index-llms-openai          # OpenAI
pip install llama-index-llms-anthropic       # Anthropic Claude
pip install llama-index-llms-ollama          # 本地 Ollama
pip install llama-index-llms-azure-openai    # Azure OpenAI

# 3. 安装 Embedding 集成
pip install llama-index-embeddings-openai    # OpenAI Embeddings
pip install llama-index-embeddings-huggingface  # HuggingFace
pip install llama-index-embeddings-voyageai  # Voyage AI

# 4. 安装向量数据库集成（选择你需要的）
pip install llama-index-vector-stores-chroma  # Chroma
pip install llama-index-vector-stores-pinecone  # Pinecone
pip install llama-index-vector-stores-qdrant  # Qdrant
pip install llama-index-vector-stores-weaviate  # Weaviate
pip install llama-index-vector-stores-milvus  # Milvus

# 5. 安装数据连接器（按需安装）
pip install llama-index-readers-file         # 文件读取
pip install llama-index-readers-pdf          # PDF
pip install llama-index-readers-web          # 网页
pip install llama-index-readers-database     # 数据库
pip install llama-index-readers-notion       # Notion
pip install llama-index-readers-github       # GitHub
pip install llama-index-readers-youtube      # YouTube

# 6. 安装评估工具（可选）
pip install llama-index-evaluation

# 7. 如果需要与 LangChain 集成
pip install langchain langchain-openai
验证安装
import llama_index
print(llama_index.__version__)  # 应该显示 0.10.x 或更高版本

设置 API Key：

import os
from llama_index.core import Settings

# 方式1：环境变量
os.environ["OPENAI_API_KEY"] = "sk-..."

# 方式2：直接在 Settings 中设置（推荐）
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 1024
Settings.chunk_overlap = 20

3.2 Hello World：最简单的 RAG

假设你有一个文件夹 data，里面放着你的 knowledge.txt。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

# 1. 加载数据 (Data Ingestion)
# SimpleDirectoryReader 极其强大，支持 PDF, Markdown, Word 等多种格式
documents = SimpleDirectoryReader("data").load_data()

# 2. 建立索引 (Indexing)
# 这步会自动进行分块(Chunking)、Embedding 并存入内存向量库
index = VectorStoreIndex.from_documents(documents)

# 3. 创建查询引擎 (Query Engine)
query_engine = index.as_query_engine()

# 4. 提问
response = query_engine.query("这篇文档的核心观点是什么？")
print(response)
print(f"\n来源节点: {response.source_nodes}")

3.3 进阶示例：带持久化的 RAG

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext, load_index_from_storage
import os

# 检查是否已有索引
persist_dir = "./storage"
if not os.path.exists(persist_dir):
    # 首次运行：创建索引
    documents = SimpleDirectoryReader("data").load_data()
    index = VectorStoreIndex.from_documents(documents)
    # 持久化
    index.storage_context.persist(persist_dir=persist_dir)
    print("索引已创建并保存")
else:
    # 后续运行：加载索引
    storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
    index = load_index_from_storage(storage_context)
    print("索引已加载")

# 查询
query_engine = index.as_query_engine()
response = query_engine.query("文档中提到了哪些关键技术？")
print(response)

3.4 使用外部向量数据库（Chroma）

import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext

# 初始化 Chroma 客户端
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("knowledge_base")

# 创建向量存储
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 加载文档并建立索引
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(
    documents, 
    storage_context=storage_context
)

# 查询
query_engine = index.as_query_engine()
response = query_engine.query("查询问题")
print(response)

第四部分：LlamaIndex 切块方式详解

切块（Chunking）是 RAG 系统的核心环节，直接影响检索质量。LlamaIndex 提供了丰富的切块策略，比 LangChain 更加灵活和强大。

4.1 为什么切块很重要？

切块质量直接影响：

  • ✅ 检索准确率：切块太小会丢失上下文，太大会导致噪声
  • ✅ 召回率：合理的切块能提高相关内容的召回
  • ✅ 生成质量：合适的块大小能提供足够的上下文给 LLM

LlamaIndex vs LangChain 切块对比：

特性 LangChain LlamaIndex
切块策略 主要是文本分割 多种专业切块器
上下文保留 基础支持 自动维护节点关系
元数据 有限 丰富的元数据支持
代码切块 需要手动处理 原生支持代码切块
语义切块 不支持 支持语义感知切块

4.2 基础切块器：SentenceSplitter

最常用的切块方式，按句子分割。

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import Settings, SimpleDirectoryReader, VectorStoreIndex

# 创建句子分割器
sentence_splitter = SentenceSplitter(
    chunk_size=512,        # 每个块的最大字符数
    chunk_overlap=50,       # 块之间的重叠字符数（保留上下文）
    separator=" ",          # 分隔符
    paragraph_separator="\n\n",  # 段落分隔符
    secondary_chunking_regex="[^,.;。]+[,.;。]?",  # 二级切分正则
)

# 应用到全局设置
Settings.node_parser = sentence_splitter

# 或者只用于特定索引
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    transformations=[sentence_splitter]  # 指定切块器
)

# 查看切块结果
nodes = sentence_splitter.get_nodes_from_documents(documents)
print(f"共切分为 {len(nodes)} 个节点")
for i, node in enumerate(nodes[:3]):  # 查看前3个节点
    print(f"\n节点 {i+1}:")
    print(f"文本长度: {len(node.text)}")
    print(f"文本预览: {node.text[:100]}...")
    print(f"元数据: {node.metadata}")

参数说明：

  • chunk_size: 建议 256-1024，根据文档类型调整
  • chunk_overlap: 建议 10-20% 的 chunk_size，保留上下文连贯性
  • separator: 默认空格，可根据语言调整

4.3 Token 切块器：TokenTextSplitter

按 Token 数量切块，更精确控制块大小。

from llama_index.core.node_parser import TokenTextSplitter

# Token 切块器（适合需要精确控制 Token 数的场景）
token_splitter = TokenTextSplitter(
    chunk_size=1024,        # 每个块的最大 Token 数
    chunk_overlap=100,      # 重叠 Token 数
    separator=" ",          # 分隔符
    backup_separators=["\n", "\n\n", "。", "！", "？"],  # 备用分隔符
)

Settings.node_parser = token_splitter

# 使用
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

# 查看 Token 统计
nodes = token_splitter.get_nodes_from_documents(documents)
for node in nodes[:3]:
    # 估算 Token 数（实际需要调用 LLM 的 tokenizer）
    print(f"节点文本长度: {len(node.text)} 字符")

适用场景：

  • 需要精确控制输入 LLM 的 Token 数
  • 使用按 Token 计费的 API（如 OpenAI）

4.4 代码切块器：CodeSplitter

专门用于代码文件的切块，保持代码结构完整。

from llama_index.core.node_parser import CodeSplitter

# Python 代码切块
python_splitter = CodeSplitter(
    language="python",
    max_chars=2000,          # 每个块的最大字符数
    chunk_lines=40,         # 每个块的最大行数
    chunk_lines_overlap=5,  # 重叠行数
)

# JavaScript 代码切块
js_splitter = CodeSplitter(
    language="javascript",
    max_chars=2000,
)

# TypeScript 代码切块
ts_splitter = CodeSplitter(
    language="typescript",
    max_chars=2000,
)

# 使用
from llama_index.readers.github import GithubRepositoryReader

loader = GithubRepositoryReader(
    github_token="your_token",
    owner="owner",
    repo="repo",
    filter_file_extensions=([".py"], GithubRepositoryReader.FilterType.INCLUDE)
)

documents = loader.load_data(branch="main")

# 为代码文档使用代码切块器
Settings.node_parser = python_splitter
index = VectorStoreIndex.from_documents(documents)

支持的编程语言：

  • Python, JavaScript, TypeScript, Java, C++, Go, Rust, PHP, Ruby, Swift, Kotlin 等

4.5 语义切块器：SemanticSplitterNodeParser

基于语义相似度切块，比固定大小切块更智能。

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

# 需要 Embedding 模型
embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 创建语义切块器
semantic_splitter = SemanticSplitterNodeParser(
    buffer_size=1,                          # 缓冲区大小
    breakpoint_percentile_threshold=95,     # 切分阈值（百分位）
    embed_model=embed_model,
)

Settings.node_parser = semantic_splitter

# 使用
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

# 语义切块会在语义变化大的地方切分，保持语义完整性
nodes = semantic_splitter.get_nodes_from_documents(documents)
print(f"语义切块结果: {len(nodes)} 个节点")

优势：

  • ✅ 在语义边界切分，保持语义完整性
  • ✅ 自动适应不同长度的段落
  • ✅ 减少跨主题的块

适用场景：

  • 长文档（书籍、报告）
  • 多主题文档
  • 需要保持语义连贯性的场景

4.6 Markdown 切块器：MarkdownNodeParser

专门处理 Markdown 文档，按标题层级切块。

from llama_index.core.node_parser import MarkdownNodeParser

# Markdown 切块器
markdown_splitter = MarkdownNodeParser()

# 使用
from llama_index.readers.file import MarkdownReader

loader = MarkdownReader()
documents = loader.load_data(file_path="documentation.md")

# 按 Markdown 结构切块
nodes = markdown_splitter.get_nodes_from_documents(documents)

# 查看切块结果（会保留标题层级信息）
for node in nodes[:3]:
    print(f"标题: {node.metadata.get('heading', 'N/A')}")
    print(f"内容: {node.text[:100]}...")

特点：

  • 按标题（H1, H2, H3…）切分
  • 保留标题层级信息在元数据中
  • 适合技术文档、Wiki 等

4.7 HTML 切块器：HTMLNodeParser

处理 HTML 文档，按标签结构切块。

from llama_index.core.node_parser import HTMLNodeParser

# HTML 切块器
html_splitter = HTMLNodeParser(
    tags=["p", "div", "section", "article"],  # 要提取的标签
    ignore_tags=["script", "style", "nav"],    # 忽略的标签
)

# 使用
from llama_index.readers.web import BeautifulSoupWebReader

loader = BeautifulSoupWebReader()
documents = loader.load_data(urls=["https://example.com"])

nodes = html_splitter.get_nodes_from_documents(documents)

4.8 自定义切块器

当内置切块器不满足需求时，可以自定义。

from llama_index.core.node_parser import BaseNodeParser
from llama_index.core.schema import Document, NodeWithScore, TextNode
from typing import List

class CustomParagraphSplitter(BaseNodeParser):
    """按段落切块的自定义切块器"""
    
    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def _get_nodes_from_documents(
        self,
        documents: List[Document],
        show_progress: bool = False,
    ) -> List[TextNode]:
        """将文档切分为节点"""
        nodes = []
        
        for doc in documents:
            # 按双换行符分割段落
            paragraphs = doc.text.split("\n\n")
            
            current_chunk = ""
            for para in paragraphs:
                # 如果当前块加上新段落不超过大小限制
                if len(current_chunk) + len(para) <= self.chunk_size:
                    current_chunk += "\n\n" + para if current_chunk else para
                else:
                    # 保存当前块
                    if current_chunk:
                        node = TextNode(
                            text=current_chunk,
                            metadata=doc.metadata.copy()
                        )
                        nodes.append(node)
                    
                    # 开始新块（带重叠）
                    overlap_text = current_chunk[-self.chunk_overlap:] if current_chunk else ""
                    current_chunk = overlap_text + "\n\n" + para
            
            # 保存最后一个块
            if current_chunk:
                node = TextNode(
                    text=current_chunk,
                    metadata=doc.metadata.copy()
                )
                nodes.append(node)
        
        return nodes

# 使用自定义切块器
custom_splitter = CustomParagraphSplitter(chunk_size=1000, chunk_overlap=100)
Settings.node_parser = custom_splitter

documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

4.9 切块策略对比与选择

切块器 适用场景 优势 劣势
SentenceSplitter 通用文本、自然语言 简单、快速、保留句子完整性 可能切分跨段落
TokenTextSplitter 需要精确 Token 控制 精确控制 Token 数 可能切分句子
CodeSplitter 代码文件 保持代码结构、语法完整 只适用于代码
SemanticSplitter 长文档、多主题文档 语义感知、智能切分 需要 Embedding，速度较慢
MarkdownNodeParser Markdown 文档 保留文档结构 只适用于 Markdown
HTMLNodeParser 网页内容 保留 HTML 结构 只适用于 HTML

4.10 切块最佳实践

实践1：根据文档类型选择切块器
from llama_index.core.node_parser import (
    SentenceSplitter,
    CodeSplitter,
    MarkdownNodeParser,
    SemanticSplitterNodeParser
)

# 根据文件类型动态选择切块器
def get_node_parser(file_path: str):
    if file_path.endswith(('.py', '.js', '.ts', '.java')):
        return CodeSplitter(language="python", max_chars=2000)
    elif file_path.endswith('.md'):
        return MarkdownNodeParser()
    elif file_path.endswith(('.html', '.htm')):
        from llama_index.core.node_parser import HTMLNodeParser
        return HTMLNodeParser()
    else:
        return SentenceSplitter(chunk_size=512, chunk_overlap=50)

# 使用
documents = SimpleDirectoryReader("data").load_data()
for doc in documents:
    file_path = doc.metadata.get("file_path", "")
    parser = get_node_parser(file_path)
    nodes = parser.get_nodes_from_documents([doc])
    # 处理节点...
实践2：调整切块大小
# 短文档（邮件、聊天记录）
short_doc_splitter = SentenceSplitter(
    chunk_size=256,
    chunk_overlap=25
)

# 中等文档（文章、报告）
medium_doc_splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50
)

# 长文档（书籍、长报告）
long_doc_splitter = SentenceSplitter(
    chunk_size=1024,
    chunk_overlap=100
)

# 超长文档（使用语义切块）
from llama_index.core.node_parser import SemanticSplitterNodeParser
very_long_doc_splitter = SemanticSplitterNodeParser(
    buffer_size=2,
    breakpoint_percentile_threshold=90
)
实践3：保留上下文关系
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import Settings

# 设置重叠以保留上下文
splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50,  # 10% 重叠
)

Settings.node_parser = splitter

# LlamaIndex 会自动维护节点之间的 Previous/Next 关系
documents = SimpleDirectoryReader("data").load_data()
index = VectorStoreIndex.from_documents(documents)

# 查询时可以获取相邻节点
retriever = index.as_retriever(similarity_top_k=3)
nodes = retriever.retrieve("查询问题")

# 获取相邻节点以提供更多上下文
for node in nodes:
    if hasattr(node, 'prev_node') and node.prev_node:
        print(f"前一个节点: {node.prev_node.text[:50]}...")
    if hasattr(node, 'next_node') and node.next_node:
        print(f"后一个节点: {node.next_node.text[:50]}...")
实践4：添加元数据到节点
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import Document

# 创建带元数据的文档
documents = [
    Document(
        text="文档内容...",
        metadata={
            "file_name": "report.pdf",
            "page_number": 1,
            "section": "introduction",
            "author": "张三",
            "created_at": "2024-01-01"
        }
    )
]

# 切块时元数据会自动继承到节点
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = splitter.get_nodes_from_documents(documents)

# 每个节点都包含元数据
for node in nodes:
    print(f"文件: {node.metadata.get('file_name')}")
    print(f"页码: {node.metadata.get('page_number')}")
    print(f"文本: {node.text[:100]}...")
实践5：切块质量检查
def analyze_chunks(nodes):
    """分析切块质量"""
    chunk_sizes = [len(node.text) for node in nodes]
    
    print(f"总节点数: {len(nodes)}")
    print(f"平均块大小: {sum(chunk_sizes) / len(chunk_sizes):.0f} 字符")
    print(f"最小块: {min(chunk_sizes)} 字符")
    print(f"最大块: {max(chunk_sizes)} 字符")
    print(f"块大小标准差: {__import__('statistics').stdev(chunk_sizes):.0f}")
    
    # 检查是否有过小的块
    small_chunks = [s for s in chunk_sizes if s < 50]
    if small_chunks:
        print(f"警告: 有 {len(small_chunks)} 个过小的块（<50字符）")
    
    # 检查是否有过大的块
    large_chunks = [s for s in chunk_sizes if s > 2000]
    if large_chunks:
        print(f"警告: 有 {len(large_chunks)} 个过大的块（>2000字符）")

# 使用
documents = SimpleDirectoryReader("data").load_data()
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = splitter.get_nodes_from_documents(documents)
analyze_chunks(nodes)

4.11 切块参数调优指南

# 参数调优示例
def optimize_chunking(documents, target_avg_size=512):
    """根据文档特征自动优化切块参数"""
    
    # 分析文档
    total_chars = sum(len(doc.text) for doc in documents)
    avg_doc_size = total_chars / len(documents)
    
    # 根据文档大小调整参数
    if avg_doc_size < 1000:
        # 短文档：小块、小重叠
        chunk_size = 256
        chunk_overlap = 25
    elif avg_doc_size < 10000:
        # 中等文档：标准参数
        chunk_size = 512
        chunk_overlap = 50
    else:
        # 长文档：大块、大重叠，或使用语义切块
        chunk_size = 1024
        chunk_overlap = 100
    
    splitter = SentenceSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    
    nodes = splitter.get_nodes_from_documents(documents)
    actual_avg = sum(len(n.text) for n in nodes) / len(nodes)
    
    print(f"目标平均大小: {target_avg_size}")
    print(f"实际平均大小: {actual_avg:.0f}")
    print(f"参数: chunk_size={chunk_size}, overlap={chunk_overlap}")
    
    return splitter

# 使用
documents = SimpleDirectoryReader("data").load_data()
optimized_splitter = optimize_chunking(documents)
Settings.node_parser = optimized_splitter

第五部分：进阶 —— LlamaIndex 的核心魔法

要精通 LlamaIndex，必须理解它区别于 LangChain 的三个核心概念：Nodes (节点)、Indices (索引) 和 Router (路由)。

5.1 节点关系与上下文管理

LlamaIndex 的核心优势之一是自动维护节点之间的关系，保留上下文连贯性。

LlamaIndex 允许我们将文档解析为带有丰富元数据（Metadata）的节点。

from llama_index.core.node_parser import SentenceSplitter, TokenTextSplitter, CodeSplitter
from llama_index.core import Settings, Document

# 示例1：句子分割器（适合自然语言）
sentence_splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=20,
)

# 示例2：Token 分割器（适合代码）
token_splitter = TokenTextSplitter(
    chunk_size=1024,
    chunk_overlap=50,
)

# 示例3：代码分割器（专门处理代码）
code_splitter = CodeSplitter(
    language="python",
    max_chars=2000,
)

# 全局设置
Settings.text_splitter = sentence_splitter

# 创建带元数据的文档
documents = [
    Document(
        text="这是文档内容...",
        metadata={
            "file_name": "report.pdf",
            "page_number": 1,
            "author": "张三",
            "created_at": "2024-01-01"
        }
    )
]

# 建立索引，每个 Node 都会携带元数据
index = VectorStoreIndex.from_documents(documents)

# 查询时可以过滤元数据
retriever = index.as_retriever(
    filters=MetadataFilters(
        filters=[
            ExactMatchFilter(key="author", value="张三")
        ]
    )
)

5.2 持久化存储 (Storage Context)

默认索引在内存中，生产环境需要存到磁盘或向量数据库（如 Chroma, Qdrant）。

from llama_index.core import StorageContext, load_index_from_storage

# 保存
index.storage_context.persist(persist_dir="./storage")

# 读取
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

5.3 路由模式 (Router Engine) —— LlamaIndex 的杀手锏

这是 LangChain 用户最应该学习的功能。假设你有一份关于"苹果公司财报"的文档，和一份关于"乔布斯传记"的文档。

  • 问"2023年营收多少" -> 应该只查财报。
  • 问"乔布斯的童年" -> 应该只查传记。
  • 问"乔布斯对苹果营收的影响" -> 应该两边都查。

LlamaIndex 可以自动选择工具：

from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector, LLMMultiSelector

# 假设 index_finance 和 index_bio 已经建立好

finance_tool = QueryEngineTool(
    query_engine=index_finance.as_query_engine(),
    metadata=ToolMetadata(
        name="finance_data",
        description="用于查询具体的财务数据、营收、利润等信息"
    )
)

bio_tool = QueryEngineTool(
    query_engine=index_bio.as_query_engine(),
    metadata=ToolMetadata(
        name="biography_data",
        description="用于查询乔布斯的生平、轶事和个人生活"
    )
)

# 方式1：单选路由（选择一个最相关的）
query_engine_single = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[finance_tool, bio_tool]
)

# 方式2：多选路由（选择多个相关的，然后合并结果）
query_engine_multi = RouterQueryEngine(
    selector=LLMMultiSelector.from_defaults(),
    query_engine_tools=[finance_tool, bio_tool]
)

# 智能路由
response = query_engine_single.query("乔布斯在建立苹果初期赚了多少钱？")
print(response)

5.4 Tree Index：层级摘要索引

Tree Index 是 LlamaIndex 独有的强大功能，适合"总结全书"类查询。

from llama_index.core import TreeIndex, SimpleDirectoryReader

# 加载文档
documents = SimpleDirectoryReader("data").load_data()

# 创建树状索引
# Tree Index 会自底向上构建摘要树
tree_index = TreeIndex.from_documents(documents)

# 查询引擎
query_engine = tree_index.as_query_engine(
    child_branch_factor=2,  # 每个节点查询2个子节点
    similarity_top_k=3      # 返回最相似的3个结果
)

# 适合的查询类型
response = query_engine.query("总结这份文档的核心内容")
print(response)

5.5 Keyword Table Index：关键词索引

不依赖 Embedding，适合精确匹配和结构化查询。

from llama_index.core import KeywordTableIndex, SimpleDirectoryReader

documents = SimpleDirectoryReader("data").load_data()

# 创建关键词表索引
keyword_index = KeywordTableIndex.from_documents(documents)

# 查询
query_engine = keyword_index.as_query_engine()
response = query_engine.query("文档中提到了哪些技术栈？")
print(response)

5.6 Knowledge Graph Index：知识图谱索引

提取实体和关系，构建知识图谱，适合关系查询。

from llama_index.core import KnowledgeGraphIndex, SimpleDirectoryReader
from llama_index.core.graph_stores import SimpleGraphStore

# 创建图存储
graph_store = SimpleGraphStore()

# 加载文档
documents = SimpleDirectoryReader("data").load_data()

# 创建知识图谱索引
kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    max_triplets_per_chunk=10,  # 每个块最多提取10个三元组
    graph_store=graph_store
)

# 查询
query_engine = kg_index.as_query_engine()
response = query_engine.query("张三和谁有关系？")
print(response)

# 也可以直接查询图谱
g = kg_index.get_networkx_graph()
# 可视化或分析图谱

5.7 SubQuestionQueryEngine：子问题分解引擎

这是解决"跨文档复杂推理"的核心工具。

from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# 假设有多个文档索引
index1 = VectorStoreIndex.from_documents(docs1)
index2 = VectorStoreIndex.from_documents(docs2)
index3 = VectorStoreIndex.from_documents(docs3)

# 创建工具
tools = [
    QueryEngineTool(
        query_engine=index1.as_query_engine(),
        metadata=ToolMetadata(name="文档1", description="关于财务策略的文档")
    ),
    QueryEngineTool(
        query_engine=index2.as_query_engine(),
        metadata=ToolMetadata(name="文档2", description="关于市场分析的文档")
    ),
    QueryEngineTool(
        query_engine=index3.as_query_engine(),
        metadata=ToolMetadata(name="文档3", description="关于技术架构的文档")
    ),
]

# 创建子问题查询引擎
query_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools
)

# 复杂查询：会自动拆解为多个子问题
response = query_engine.query(
    "对比这三个文档中提到的策略有何不同？"
)
print(response)

5.8 多步查询引擎 (MultiStepQueryEngine)

需要多轮检索和推理的复杂问题。

from llama_index.core.query_engine import MultiStepQueryEngine
from llama_index.core.query_engine.router_query_engine import RouterQueryEngine

# 创建多步查询引擎
multi_step_engine = MultiStepQueryEngine(
    query_engine=query_engine,  # 基础查询引擎
    num_outputs=3,              # 每步输出3个结果
    early_stopping=True         # 提前停止
)

response = multi_step_engine.query("深入分析这个问题的多个方面")
print(response)

5.9 重排序 (Re-ranking) 提高准确率

from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.core import VectorStoreIndex

# 创建重排序器
reranker = SentenceTransformerRerank(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2",
    top_n=3  # 只保留前3个最相关的结果
)

# 创建带重排序的查询引擎
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(
    similarity_top_k=10,  # 先检索10个
    node_postprocessors=[reranker]  # 然后重排序到3个
)

response = query_engine.query("查询问题")
print(response)

第六部分：数据连接器实战示例

LlamaIndex 的强大之处在于它支持 100+ 种数据源。以下是常见场景的示例。

6.1 读取 PDF 文档

from llama_index.readers.pdf import PDFReader
from llama_index.core import VectorStoreIndex

# 读取 PDF
loader = PDFReader()
documents = loader.load_data(file_path="report.pdf")

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("PDF 中的关键信息是什么？")

6.2 读取网页内容

from llama_index.readers.web import BeautifulSoupWebReader
from llama_index.core import VectorStoreIndex

# 读取网页
loader = BeautifulSoupWebReader()
documents = loader.load_data(urls=["https://example.com/article"])

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("这篇文章讲了什么？")

6.3 连接数据库（PostgreSQL）

from llama_index.readers.database import DatabaseReader
from llama_index.core import VectorStoreIndex
import psycopg2

# 连接数据库
connection_string = "postgresql://user:password@localhost/dbname"
db = DatabaseReader(connection_string=connection_string)

# 查询数据
documents = db.load_data(
    query="SELECT * FROM articles WHERE category = 'tech'",
    max_rows=1000
)

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("这些文章的共同主题是什么？")

6.4 读取 Notion 页面

from llama_index.readers.notion import NotionPageReader
from llama_index.core import VectorStoreIndex

# 需要 Notion Integration Token
integration_token = "your_notion_token"
reader = NotionPageReader(integration_token=integration_token)

# 读取页面
page_ids = ["page-id-1", "page-id-2"]
documents = reader.load_data(page_ids=page_ids)

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("Notion 页面中的关键信息")

6.5 读取 GitHub 代码仓库

from llama_index.readers.github import GithubRepositoryReader
from llama_index.core import VectorStoreIndex

# 需要 GitHub Token
github_token = "your_github_token"
owner = "owner_name"
repo = "repo_name"

# 读取代码
loader = GithubRepositoryReader(
    github_token=github_token,
    owner=owner,
    repo=repo,
    filter_file_extensions=([".py", ".js", ".ts"], GithubRepositoryReader.FilterType.INCLUDE),
    verbose=True
)

documents = loader.load_data(branch="main")

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("这个项目的核心功能是什么？")

6.6 读取 YouTube 视频字幕

from llama_index.readers.youtube import YoutubeTranscriptReader
from llama_index.core import VectorStoreIndex

# 读取 YouTube 视频字幕
loader = YoutubeTranscriptReader()
documents = loader.load_data(ytlinks=["https://www.youtube.com/watch?v=VIDEO_ID"])

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("这个视频的主要内容是什么？")

6.7 读取 S3 存储的文件

from llama_index.readers.s3 import S3Reader
from llama_index.core import VectorStoreIndex

# 配置 S3
s3_reader = S3Reader(
    bucket="my-bucket",
    aws_access_id="your_access_id",
    aws_access_secret="your_access_secret"
)

# 读取文件
documents = s3_reader.load_data(key="documents/")

# 建立索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

第六部分：LlamaIndex 的高级应用场景

6.1 结构化数据提取 (Extraction)

不仅仅是对话，LlamaIndex 擅长从非结构化文本中提取结构化数据。

from llama_index.core.program import LLMTextCompletionProgram
from llama_index.llms.openai import OpenAI
from pydantic import BaseModel, Field
from typing import List

# 定义数据结构
class Product(BaseModel):
    name: str = Field(description="产品名称")
    price: float = Field(description="价格")
    category: str = Field(description="类别")

class ProductList(BaseModel):
    products: List[Product] = Field(description="产品列表")

# 创建提取程序
llm = OpenAI(model="gpt-4")
program = LLMTextCompletionProgram.from_defaults(
    output_cls=ProductList,
    prompt_template_str="从以下文本中提取所有产品信息：\n{text}",
    llm=llm
)

# 执行提取
text = """
我们有以下产品：
1. iPhone 15 Pro，价格 8999 元，属于手机类别
2. MacBook Pro，价格 12999 元，属于电脑类别
3. AirPods Pro，价格 1899 元，属于耳机类别
"""

output = program(text=text)
print(output.products)
# 输出: [Product(name='iPhone 15 Pro', price=8999.0, category='手机'), ...]

6.2 递归检索 (Recursive Retrieval)

这是解决表格和图片检索的关键。

from llama_index.core import VectorStoreIndex, Document
from llama_index.core.node_parser import SimpleNodeParser

# 创建包含表格摘要的文档
documents = [
    Document(
        text="这是一份财务报告。",
        metadata={"type": "summary"}
    ),
    Document(
        text="详细财务数据表格：收入1000万，支出600万，利润400万。",
        metadata={"type": "detail", "parent_id": "summary"}
    )
]

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

# 递归检索：先检索到摘要，然后自动获取详细信息
query_engine = index.as_query_engine(
    similarity_top_k=2,
    # 自动处理父子关系
)

response = query_engine.query("财务报告中的利润是多少？")
print(response)

6.3 Text-to-SQL：自然语言查询数据库

from llama_index.core import SQLDatabase
from llama_index.core.query_engine import NLSQLTableQueryEngine
from sqlalchemy import create_engine

# 连接数据库
engine = create_engine("sqlite:///example.db")
sql_database = SQLDatabase(engine)

# 创建 Text-to-SQL 查询引擎
query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    tables=["users", "orders"]  # 指定可查询的表
)

# 自然语言查询
response = query_engine.query("查询最近一个月订单金额超过1000的用户")
print(response)
print(f"执行的 SQL: {response.metadata['sql_query']}")

6.4 混合检索：向量 + 关键词

from llama_index.core import VectorStoreIndex, KeywordTableIndex
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever, KeywordTableSimpleRetriever

# 创建向量索引和关键词索引
vector_index = VectorStoreIndex.from_documents(documents)
keyword_index = KeywordTableIndex.from_documents(documents)

# 创建混合检索器
vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=3)
keyword_retriever = KeywordTableSimpleRetriever(index=keyword_index)

# 组合检索器（需要自定义实现）
class HybridRetriever:
    def __init__(self, vector_retriever, keyword_retriever):
        self.vector_retriever = vector_retriever
        self.keyword_retriever = keyword_retriever
    
    def retrieve(self, query_str):
        vector_nodes = self.vector_retriever.retrieve(query_str)
        keyword_nodes = self.keyword_retriever.retrieve(query_str)
        # 合并并去重
        all_nodes = {node.node_id: node for node in vector_nodes + keyword_nodes}
        return list(all_nodes.values())

hybrid_retriever = HybridRetriever(vector_retriever, keyword_retriever)
query_engine = RetrieverQueryEngine.from_args(retriever=hybrid_retriever)

response = query_engine.query("查询问题")

6.5 流式响应

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(streaming=True)

# 流式输出
response = query_engine.query("长问题...")
response.print_response_stream()

6.6 与 LangChain 协同工作

你不需要二选一。最佳实践是：用 LangChain 做 Agent 骨架，用 LlamaIndex 做超级检索工具。

from langchain.agents import initialize_agent, Tool
from langchain_openai import ChatOpenAI
from llama_index.core import VectorStoreIndex

# 1. 构建 LlamaIndex 引擎
llama_index = VectorStoreIndex.from_documents(docs)
llama_engine = llama_index.as_query_engine()

# 2. 包装成 LangChain Tool
tools = [
    Tool(
        name="LlamaIndex_Knowledge_Base",
        func=lambda q: str(llama_engine.query(q)),
        description="当用户询问关于私有知识库内容时使用此工具"
    )
]

# 3. 在 LangChain Agent 中使用
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent="zero-shot-react-description",
    verbose=True
)

result = agent.run("查询知识库中的信息，然后总结")

6.7 LlamaIndex Agent

LlamaIndex 也支持构建自己的 Agent。

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.llms.openai import OpenAI

# 创建工具
query_engine = index.as_query_engine()
tools = [
    QueryEngineTool(
        query_engine=query_engine,
        metadata=ToolMetadata(
            name="知识库",
            description="用于查询公司内部知识库"
        )
    )
]

# 创建 Agent
llm = OpenAI(model="gpt-4")
agent = ReActAgent.from_tools(tools, llm=llm, verbose=True)

# 使用 Agent
response = agent.chat("帮我查询知识库，然后总结关键信息")
print(response)

第八部分：评估与优化

8.1 检索评估

评估检索质量是优化 RAG 系统的关键。

from llama_index.core.evaluation import (
    RetrieverEvaluator,
    generate_question_context_pairs,
    EmbeddingFaithfulnessEvaluator,
    EmbeddingRelevancyEvaluator
)
from llama_index.core import VectorStoreIndex

# 创建索引
index = VectorStoreIndex.from_documents(documents)
retriever = index.as_retriever(similarity_top_k=5)

# 生成评估数据集
eval_questions = [
    "文档中提到了哪些技术？",
    "核心观点是什么？",
    "有哪些关键数据？"
]

# 评估检索器
evaluator = RetrieverEvaluator.from_metric_names(
    ["mrr", "hit_rate"]  # Mean Reciprocal Rank, Hit Rate
)

results = []
for question in eval_questions:
    result = evaluator.evaluate(question, retriever)
    results.append(result)
    print(f"问题: {question}")
    print(f"MRR: {result.mrr}, Hit Rate: {result.hit_rate}")

# 计算平均指标
avg_mrr = sum(r.mrr for r in results) / len(results)
avg_hit_rate = sum(r.hit_rate for r in results) / len(results)
print(f"\n平均 MRR: {avg_mrr:.2f}, 平均 Hit Rate: {avg_hit_rate:.2f}")

8.2 生成评估

评估生成质量（相关性、准确性）。

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    CorrectnessEvaluator
)

# 创建评估器
faithfulness_evaluator = FaithfulnessEvaluator()
relevancy_evaluator = RelevancyEvaluator()

# 评估生成结果
query = "文档的核心观点是什么？"
response = query_engine.query(query)

# 评估忠实度（是否基于检索内容）
faithfulness_result = faithfulness_evaluator.evaluate_response(
    query=query,
    response=response,
    contexts=response.source_nodes
)
print(f"忠实度分数: {faithfulness_result.score}")

# 评估相关性（回答是否相关）
relevancy_result = relevancy_evaluator.evaluate_response(
    query=query,
    response=response
)
print(f"相关性分数: {relevancy_result.score}")

8.3 端到端评估

from llama_index.core.evaluation import DatasetGenerator, QueryResponseDataset

# 生成评估数据集
dataset_generator = DatasetGenerator.from_documents(documents)
eval_dataset = dataset_generator.generate_questions_from_nodes(num=10)

# 保存数据集
eval_dataset.save_json("eval_dataset.json")

# 加载数据集
eval_dataset = QueryResponseDataset.from_json("eval_dataset.json")

# 批量评估
from llama_index.core.evaluation import BatchEvalRunner

runner = BatchEvalRunner(
    evaluators={
        "faithfulness": faithfulness_evaluator,
        "relevancy": relevancy_evaluator
    },
    workers=4  # 并行评估
)

results = runner.evaluate_dataset(
    dataset=eval_dataset,
    query_engine=query_engine
)

print(results)

8.4 优化策略

优化1：调整切块策略
from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser
)

# 方式1：基于语义的分块（更智能）
semantic_splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95
)

# 方式2：自定义分块
class CustomNodeParser:
    def get_nodes_from_documents(self, documents):
        # 自定义逻辑
        nodes = []
        for doc in documents:
            # 按段落分割
            paragraphs = doc.text.split("\n\n")
            for para in paragraphs:
                nodes.append(Node(text=para, metadata=doc.metadata))
        return nodes

Settings.node_parser = semantic_splitter
优化2：调整检索参数
# 增加检索数量，然后重排序
query_engine = index.as_query_engine(
    similarity_top_k=20,  # 先检索20个
    node_postprocessors=[
        SentenceTransformerRerank(top_n=5)  # 重排序到5个
    ]
)
优化3：使用更好的 Embedding 模型
from llama_index.embeddings.openai import OpenAIEmbedding

# 使用更大的模型
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-large"  # 更大的模型，更好的效果
)

第九部分：实际项目案例

案例1：企业知识库问答系统

"""
场景：企业内部有大量技术文档、产品手册、会议记录等，
需要构建一个智能问答系统。
"""
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
import chromadb

# 1. 数据准备：读取多个目录的文档
tech_docs = SimpleDirectoryReader("data/tech").load_data()
product_docs = SimpleDirectoryReader("data/product").load_data()
meeting_docs = SimpleDirectoryReader("data/meetings").load_data()

# 2. 为不同类别建立索引
chroma_client = chromadb.PersistentClient(path="./chroma_db")

tech_collection = chroma_client.get_or_create_collection("tech")
tech_vector_store = ChromaVectorStore(chroma_collection=tech_collection)
tech_index = VectorStoreIndex.from_documents(
    tech_docs,
    storage_context=StorageContext.from_defaults(vector_store=tech_vector_store)
)

product_collection = chroma_client.get_or_create_collection("product")
product_vector_store = ChromaVectorStore(chroma_collection=product_collection)
product_index = VectorStoreIndex.from_documents(
    product_docs,
    storage_context=StorageContext.from_defaults(vector_store=product_vector_store)
)

# 3. 创建路由查询引擎
tech_tool = QueryEngineTool(
    query_engine=tech_index.as_query_engine(),
    metadata=ToolMetadata(
        name="技术文档",
        description="用于查询技术文档、API文档、开发指南等"
    )
)

product_tool = QueryEngineTool(
    query_engine=product_index.as_query_engine(),
    metadata=ToolMetadata(
        name="产品文档",
        description="用于查询产品手册、用户指南、功能介绍等"
    )
)

router_engine = RouterQueryEngine.from_defaults(
    query_engine_tools=[tech_tool, product_tool]
)

# 4. 使用
response = router_engine.query("如何配置API密钥？")
print(response)

案例2：多文档对比分析系统

"""
场景：需要对比多个文档的差异，比如对比不同版本的合同、
不同公司的财报等。
"""
from llama_index.core import VectorStoreIndex
from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# 为每个文档建立索引
doc1_index = VectorStoreIndex.from_documents(doc1)
doc2_index = VectorStoreIndex.from_documents(doc2)
doc3_index = VectorStoreIndex.from_documents(doc3)

# 创建工具
tools = [
    QueryEngineTool(
        query_engine=doc1_index.as_query_engine(),
        metadata=ToolMetadata(name="文档1", description="2023年财报")
    ),
    QueryEngineTool(
        query_engine=doc2_index.as_query_engine(),
        metadata=ToolMetadata(name="文档2", description="2024年财报")
    ),
    QueryEngineTool(
        query_engine=doc3_index.as_query_engine(),
        metadata=ToolMetadata(name="文档3", description="行业报告")
    ),
]

# 创建子问题查询引擎
comparison_engine = SubQuestionQueryEngine.from_defaults(
    query_engine_tools=tools
)

# 对比查询
response = comparison_engine.query(
    "对比这三份文档中关于营收增长策略的差异"
)
print(response)

案例3：代码库智能问答

"""
场景：为大型代码库构建智能问答系统，帮助开发者快速理解代码。
"""
from llama_index.readers.github import GithubRepositoryReader
from llama_index.core import VectorStoreIndex
from llama_index.core.node_parser import CodeSplitter

# 读取代码
loader = GithubRepositoryReader(
    github_token="your_token",
    owner="owner",
    repo="repo",
    filter_file_extensions=([".py", ".js", ".ts"], GithubRepositoryReader.FilterType.INCLUDE)
)

documents = loader.load_data(branch="main")

# 使用代码分割器
Settings.node_parser = CodeSplitter(language="python", max_chars=2000)

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

# 查询
query_engine = index.as_query_engine()
response = query_engine.query("这个项目是如何处理用户认证的？")
print(response)

案例4：法律文档智能检索

"""
场景：律师事务所需要快速检索法律条文、案例等。
"""
from llama_index.core import VectorStoreIndex, TreeIndex
from llama_index.core.query_engine import RouterQueryEngine

# 法律条文需要精确匹配，使用关键词索引
from llama_index.core import KeywordTableIndex

law_index = KeywordTableIndex.from_documents(law_documents)

# 案例需要语义理解，使用向量索引
case_index = VectorStoreIndex.from_documents(case_documents)

# 创建路由引擎
law_tool = QueryEngineTool(
    query_engine=law_index.as_query_engine(),
    metadata=ToolMetadata(
        name="法律条文",
        description="用于精确查询法律条文、法规等"
    )
)

case_tool = QueryEngineTool(
    query_engine=case_index.as_query_engine(),
    metadata=ToolMetadata(
        name="案例",
        description="用于查询相关案例、判例等"
    )
)

router = RouterQueryEngine.from_defaults(
    query_engine_tools=[law_tool, case_tool]
)

response = router.query("关于劳动法加班费的规定是什么？")
print(response)

案例5：实时文档更新系统

"""
场景：文档库每天新增大量文档，需要增量更新索引。
"""
from llama_index.core import VectorStoreIndex, Document
from llama_index.core.storage.storage_context import StorageContext

# 加载已有索引
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

# 读取新文档
new_documents = SimpleDirectoryReader("data/new").load_data()

# 增量插入（只添加新文档，不重建整个索引）
for doc in new_documents:
    index.insert(doc)

# 保存更新后的索引
index.storage_context.persist(persist_dir="./storage")

# 查询（包含新旧文档）
query_engine = index.as_query_engine()
response = query_engine.query("查询问题")

第十部分：LlamaIndex 应用场景总结

10.1 企业级应用场景

场景 核心需求 LlamaIndex 优势
企业知识库 多源数据、智能检索 Router Engine、多索引组合
客服系统 快速准确回答 向量检索 + 重排序
代码助手 理解代码库 代码分割器、结构化解析
法律检索 精确匹配 + 语义理解 关键词索引 + 向量索引
财务分析 多文档对比 SubQuestion Engine
文档总结 层级摘要 Tree Index
数据提取 结构化输出 Extraction Program

10.2 技术场景

场景 技术方案
大规模文档库 向量数据库（Chroma/Pinecone）+ 持久化
实时更新 增量插入 + 版本控制
多模态数据 图片/音频连接器 + 递归检索
混合查询 向量检索 + 关键词检索 + SQL 查询
复杂推理 SubQuestion Engine + MultiStep Engine

第十一部分：总结与建议

11.1 什么时候该用 LlamaIndex？

强烈推荐使用 LlamaIndex 的场景：

  1. ✅ 数据密集型应用：你的应用核心是处理大量、多样的数据
  2. ✅ 复杂查询需求：需要跨文档推理、对比分析、层级摘要
  3. ✅ 多数据源整合：需要从数据库、文件、API 等多种来源整合数据
  4. ✅ 检索质量要求高：LangChain 的 Retriever 无法满足准确率要求
  5. ✅ 需要增量更新：文档库频繁更新，需要高效的增量索引
  6. ✅ 结构化数据提取：需要从非结构化文本中提取结构化信息

可以考虑 LangChain 的场景：

  1. ⚠️ 简单 RAG：只需要基本的向量检索，不需要复杂索引
  2. ⚠️ Agent 为主：应用核心是 Agent 编排，检索只是辅助功能
  3. ⚠️ 快速原型：需要快速搭建原型，LangChain 上手更快

11.2 常见问题与解决方案

Q1: LlamaIndex 和 LangChain 可以一起用吗？
A: 完全可以！最佳实践是用 LangChain 做 Agent 编排，用 LlamaIndex 做检索工具。

Q2: 索引太大，内存不够怎么办？
A: 使用外部向量数据库（Chroma、Pinecone、Qdrant）进行持久化存储。

Q3: 检索结果不准确怎么办？
A:

  • 调整分块策略（使用语义分割）
  • 增加检索数量 + 重排序
  • 使用混合检索（向量 + 关键词）
  • 优化 Embedding 模型

Q4: 如何提高查询速度？
A:

  • 使用更快的向量数据库（如 Qdrant）
  • 减少 similarity_top_k
  • 使用缓存机制
  • 异步查询

Q5: 如何处理多语言文档？
A: 使用支持多语言的 Embedding 模型（如 multilingual-e5）。

Q6: 导入错误：ModuleNotFoundError
A:

# 错误：from llama_index import VectorStoreIndex
# 正确（v0.10+）：
from llama_index.core import VectorStoreIndex

Q9: 向量维度不匹配
A:

# 确保 Embedding 模型和向量数据库的维度一致
# OpenAI text-embedding-3-small: 1536 维
# OpenAI text-embedding-3-large: 3072 维

11.4 最佳实践

实践1：合理设置分块大小
# 对于长文档（如书籍）
Settings.chunk_size = 1024
Settings.chunk_overlap = 100

# 对于短文档（如邮件、聊天记录）
Settings.chunk_size = 256
Settings.chunk_overlap = 50

# 对于代码
from llama_index.core.node_parser import CodeSplitter
Settings.node_parser = CodeSplitter(language="python", max_chars=2000)
实践2：使用元数据过滤
from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter

# 创建带元数据的文档
documents = [
    Document(
        text="内容...",
        metadata={"department": "tech", "year": 2024}
    )
]

# 查询时过滤
retriever = index.as_retriever(
    filters=MetadataFilters(
        filters=[
            ExactMatchFilter(key="department", value="tech"),
            ExactMatchFilter(key="year", value=2024)
        ]
    )
)
实践3：错误处理和重试
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def query_with_retry(query_engine, query):
    try:
        response = query_engine.query(query)
        return response
    except Exception as e:
        print(f"查询失败: {e}")
        raise

# 使用
response = query_with_retry(query_engine, "查询问题")
实践4：日志记录
import logging
from llama_index.core import Settings

# 启用日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("llama_index")

# 记录查询
def logged_query(query_engine, query):
    logger.info(f"查询: {query}")
    response = query_engine.query(query)
    logger.info(f"响应: {response}")
    return response
实践5：性能监控
import time
from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler

# 启用调试回调
debug_handler = LlamaDebugHandler()
callback_manager = CallbackManager([debug_handler])

Settings.callback_manager = callback_manager

# 查询并监控
start_time = time.time()
response = query_engine.query("查询问题")
end_time = time.time()

print(f"查询耗时: {end_time - start_time:.2f}秒")
print(f"检索到的节点数: {len(response.source_nodes)}")
实践6：缓存机制
from llama_index.core.cache import SimpleCache
from llama_index.core import Settings

# 启用缓存
cache = SimpleCache()
Settings.cache = cache

# 相同查询会直接返回缓存结果
response1 = query_engine.query("查询问题")  # 第一次，会执行
response2 = query_engine.query("查询问题")  # 第二次，从缓存返回

结语

LlamaIndex 不仅仅是一个库，它代表了一种**“以数据为中心” (Data-Centric) 的 AI 开发理念**。在 AI 应用日益复杂的今天，掌握它将是你构建企业级知识库应用的杀手锏。

核心要点回顾：

  1. LlamaIndex 专注于数据：它是数据到 LLM 的桥梁，而 LangChain 专注于流程编排
  2. 丰富的索引类型：不仅仅是向量检索，还有树状、关键词、知识图谱等
  3. 智能查询引擎：Router、SubQuestion、MultiStep 等引擎解决复杂查询问题
  4. 强大的数据连接：支持 100+ 种数据源
  5. 生产级特性：持久化、增量更新、评估优化等
# 人工智能 # python
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

智算之核：深度解构昇腾CANN架构的演进逻辑与开发者生态新图景

avatar 2048 AI社区
cover

AutoGen智能体开发：autogen_ext.agents.web_surfer包

avatar 2048 AI社区
cover

9 个研究生思想汇报 AI 工具，千笔推荐降重查重率优化

avatar 2048 AI社区