LlamaIndex 深入

Index / Retriever / Query Engine / Ingestion Pipeline、Agent Runner / Workflow 引擎

LlamaIndex 概览

LlamaIndex（原名 GPT Index）是一个专注于构建 RAG（检索增强生成）系统的数据框架，旨在弥合私有数据与大型语言模型之间的鸿沟。与 LangChain 偏重 Agent 和 Chain 编排、覆盖范围广泛的定位不同，LlamaIndex 的核心关注点始终围绕"数据"这一中心——如何高效地索引、检索和消费数据。

维度	LlamaIndex	LangChain
设计哲学	数据为中心，索引驱动	链/代理为中心，编排驱动
核心抽象	Index / Retriever / Query Engine	Chain / Agent / Tool
检索优化	内置丰富检索策略（向量、关键词、树、混合）	需通过第三方集成实现
数据管线	原生 Ingestion Pipeline 支持	依赖 LangGraph 或自定义流程
学习曲线	概念清晰、API 一致，上手较快	抽象层级多，生态庞大，学习曲线陡

LlamaIndex 的核心设计理念可以概括为三点：

索引即入口：一切从 Index 开始，Index 是数据组织和检索的单元，不同类型的 Index 适配不同的数据形态和查询需求。
组件可插拔：Retriever、QueryEngine、ResponseSynthesizer 等组件均遵循统一接口，用户可以自由组合和替换。
管道化数据流：从原始文档到可查询索引，再到多轮 Agent 交互，每一阶段都可被显式建模为可组合的数据流。

Index 索引体系

Index 是 LlamaIndex 中组织文档数据的核心结构。不同类型的 Index 采用不同的方式对文档进行编码和组织，从而适配不同的检索场景。

Index 类型对比

Index 类型	原理	优势	劣势	适用场景
VectorStoreIndex	将文档分块后通过 Embedding 模型转为向量，存入向量数据库	语义检索能力强，支持模糊匹配	需要 Embedding 模型和向量数据库，冷启动开销大	大规模文档库、开放域问答
SummaryIndex	为每个文档块生成摘要，创建摘要到块索引的映射	结构简单，无需向量化	仅支持基于摘要的关键词匹配，语义能力弱	小规模文档、快速原型
TreeIndex	以树形结构组织文档：叶节点为原始块，父节点为子节点摘要，递归构建	支持从粗到精的层级检索，适合长文档	构建成本随深度指数增长；对短文档优势不明显	长文档问答、教科书式检索
KeywordTableIndex	从每个文档块中提取关键词，构建关键词到文档块的多对多映射表	实现简单、检索速度快	依赖关键词提取质量，无法处理语义近义词	关键词明确的查询、FAQ 系统

VectorStoreIndex 深入

VectorStoreIndex 是生产环境中最常用的 Index 类型。其核心流程如下：

text

原始文档 → 文本分块（NodeParser） → 向量化（Embedding） → 存入 VectorStore

LlamaIndex 支持多种后端向量存储，包括：

内存向量存储（SimpleVectorStore）：适用于开发和小规模数据
Qdrant / Pinecone / Milvus / Weaviate：生产级向量数据库
PostgreSQL + pgvector：复用关系型数据库基础设施

SummaryIndex 深入

SummaryIndex 不依赖向量化，而是为每个文档块维护一个顺序列表。查询时，它会遍历所有块的摘要或内容，筛选出与查询匹配的块。适用于文档量小但需要高精度匹配的场景。

TreeIndex 深入

TreeIndex 递归地构建一棵摘要树：

将原始文档切分为叶节点块
对相邻的叶节点块生成摘要，形成父节点
对父节点继续生成摘要，直到树根
查询时从根节点开始，逐层向下选择最匹配的子节点，直到叶节点

这种"从粗到精"的检索策略特别适合需要理解文档整体结构的场景。

KeywordTableIndex 深入

通过 TF-IDF 或基于 LLM 的关键词提取器从每个文档块中提取关键词，建立关键词到文档块的倒排索引。查询时将用户问题也提取关键词，匹配倒排索引获取相关文档块。

Retriever 检索器

Retriever 负责根据用户查询从 Index 中检索最相关的文档节点。所有 Retriever 均继承自 BaseRetriever 接口。

BaseRetriever 核心接口

python

class BaseRetriever:
    def retrieve(self, query: str) -> List[NodeWithScore]:
        """根据查询检索相关节点，返回带权重的节点列表。"""
        ...

核心返回值 NodeWithScore 包含两个字段：

node：文档节点（包含文本内容、元数据等）
score：相关性得分（越高越相关，具体含义因检索器类型而异）

主要 Retriever 实现

Retriever	对应 Index	工作机制
VectorIndexRetriever	VectorStoreIndex	将查询向量化，在向量空间中执行 ANN（近似最近邻）搜索
SummaryRetriever	SummaryIndex	遍历所有节点，基于关键词匹配过滤
TreeIndexRetriever	TreeIndex	从树根逐层向下遍历，每层选择最相关的子节点
KeywordTableRetriever	KeywordTableIndex	从查询中提取关键词，匹配倒排索引
BM25Retriever	独立使用（非 Index 绑定）	基于 BM25 算法的稀疏检索，适合精确关键词匹配

混合检索

混合检索是 LlamaIndex 的一个重要特性。它将向量检索（密集检索、语义匹配）与关键词检索（稀疏检索、精确匹配）结合，兼顾召回率与精确度。

python

from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.retrievers import VectorIndexRetriever

# 同时使用向量检索和 BM25 检索
from llama_index.retrievers.bm25 import BM25Retriever

vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=5)
bm25_retriever = BM25Retriever.from_defaults(index=vector_index, similarity_top_k=5)

# QueryFusionRetriever 支持融合多个检索器的结果
hybrid_retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    similarity_top_k=10,
    num_queries=1,  # 为每个检索器生成多个查询变体
    mode="reciprocal_rerank",  # 使用倒数排名融合算法
)

Query Engine 查询引擎

Query Engine 是 LlamaIndex 面向用户的顶层接口，封装了检索和生成的完整流程。

RetrieverQueryEngine

RetrieverQueryEngine 是最基本的查询引擎实现，它将一个 Retriever 与一个 ResponseSynthesizer 组合起来：

text

用户查询 → Retriever（检索相关节点） → ResponseSynthesizer（合成回答） → 最终回答

ResponseSynthesizer 支持多种合成模式：

模式	说明	适用场景
`compact`（默认）	将所有检索到的节点文本合并后一次性送入 LLM	通用场景，节点数适中
`refine`	逐节点迭代：首节点生成初始回答，后续节点依次精炼	需要逐步修正的场景
`tree_summarize`	将节点按树形结构逐层汇总	节点数过多的场景
`simple`	将所有节点拼接成单一上下文	节点文本较短时
`no_text`	仅检索，不生成回答	纯检索场景
`accumulate`	每个节点独立生成回答后合并	需要多视角回答的场景
`compact_accumulate`	先 compact 再 accumulate 的组合模式	大规模文档分片处理

自定义 Query Engine

通过继承 CustomQueryEngine，可以灵活组合检索与生成的流程：

python

from llama_index.core.query_engine import CustomQueryEngine
from llama_index.core.retrievers import BaseRetriever
from llama_index.core.response_synthesizers import BaseSynthesizer
from llama_index.core.llms import LLM

class MyCustomQueryEngine(CustomQueryEngine):
    """自定义查询引擎：先检索，再对结果排序，最后生成回答。"""
    retriever: BaseRetriever
    synthesizer: BaseSynthesizer
    llm: LLM

    def custom_query(self, query_str: str) -> str:
        # 1. 检索
        nodes = self.retriever.retrieve(query_str)

        # 2. 自定义后处理：按得分过滤
        nodes = [n for n in nodes if n.score > 0.5]

        # 3. 合成回答
        return self.synthesizer.synthesize(query_str, nodes)

Ingestion Pipeline 数据管线

Ingestion Pipeline 提供了从原始文档到可查询索引的端到端处理流程，将文档加载、解析、转换和索引构建串联为可复用的管道。

流水线组成

text

文档加载（Readers） → 文档解析（Parsers） → 文本转换（Transformations） → 索引写入（Indexing）
                         ↓
                   节点提取与分块

各阶段详解

1. 文档加载（Readers）

LlamaIndex 内置了丰富的文档阅读器，覆盖多种数据源：

python

from llama_index.core import SimpleDirectoryReader
from llama_index.readers.file import PDFReader, MarkdownReader, DocxReader

# 加载目录下所有文件
documents = SimpleDirectoryReader(
    input_dir="./data",
    file_extractor={
        ".pdf": PDFReader(),
        ".md": MarkdownReader(),
        ".docx": DocxReader(),
    }
).load_data()

2. 文档解析（Parsers）

原始文档被加载后，需要解析为结构化的 Document 对象。解析器负责提取文档中的文本、表格、图片等元素，并保留结构信息（如标题层级、段落顺序）。

3. 文本转换（Transformations）

转换阶段是整个数据管线的核心，涉及以下操作：

python

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.extractors import TitleExtractor

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=128),
        TitleExtractor(),
        OpenAIEmbedding(model="text-embedding-3-small"),
    ]
)

关键 Transform 类型：

Transform	作用	典型参数
`SentenceSplitter`	按句子边界递归分块	chunk_size=512, chunk_overlap=128
`TokenTextSplitter`	按 token 数固定分块	chunk_size=1024
`TitleExtractor`	为每个节点提取标题	-
`QuestionsAnsweredExtractor`	为节点生成预期问题	-
`Embedding` 模型	将节点文本转为向量	如 text-embedding-3-small

4. 索引写入（Indexing）

转换后的节点最终被写入 Index，完成从原始文档到可查询索引的转化。

python

# 完整的数据管线示例
from llama_index.core import VectorStoreIndex

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=128),
        OpenAIEmbedding(),
    ]
)

nodes = pipeline.run(documents=documents)
index = VectorStoreIndex(nodes)

Agent Runner / Workflow 引擎

LlamaIndex 从 v0.10 开始引入 Workflow 引擎，替代了早期版本中较为简陋的 Agent 机制，提供了事件驱动的图结构编排能力。

AgentRunner 与 AgentWorker

AgentRunner 是 Agent 系统的入口，负责管理对话状态、调用 AgentWorker 执行任务，并在多个工具之间进行路由。AgentWorker 是实际执行任务的单元，每次收到任务后执行以下步骤：

text

AgentRunner 接收用户查询
    → 将查询包装为 Task
    → 调用 AgentWorker.run_step()
    → AgentWorker 分析任务，选择合适的工具
    → 执行工具，返回中间结果
    → 判断是否继续（如需要调用更多工具）
    → 最终返回完整回答

python

from llama_index.core.agent import AgentRunner
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# 创建工具
query_tool = QueryEngineTool(
    query_engine=my_query_engine,
    metadata=ToolMetadata(
        name="data_query",
        description="用于查询知识库中的文档数据",
    ),
)

# 创建 AgentRunner
agent = AgentRunner.from_llm(
    llm=llm,
    tools=[query_tool],
    verbose=True,
)

# 执行查询
response = agent.chat("请查询上季度的销售数据")

Workflow 事件驱动图

Workflow 引擎是 LlamaIndex 在 v0.10 引入的核心新特性，它允许用户将有状态的应用程序建模为事件驱动的图结构。

核心概念

概念	说明
Step	Workflow 中的基本计算单元，接收事件并产生新事件
Event	在 Step 之间传递的数据单元，包含任意负载
Context	跨 Step 共享的状态上下文，支持读写操作
Workflow	一个由 Step 和 Event 连接而成的有向图

Step 定义示例

python

from llama_index.core.workflow import (
    Workflow,
    step,
    Context,
    StartEvent,
    StopEvent,
)

# 定义事件类型
class RetrieverEvent(Event):
    """检索完成事件，携带检索到的节点。"""
    nodes: list

class GenerateEvent(Event):
    """生成完成事件，携带最终回答。"""
    response: str

# 定义 Workflow
class RAGWorkflow(Workflow):
    @step
    async def retrieve(
        self, ctx: Context, ev: StartEvent
    ) -> RetrieverEvent:
        """检索步骤：接收查询并检索相关节点。"""
        query = ev.query
        nodes = self.retriever.retrieve(query)
        await ctx.set("query", query)
        return RetrieverEvent(nodes=nodes)

    @step
    async def generate(
        self, ctx: Context, ev: RetrieverEvent
    ) -> GenerateEvent:
        """生成步骤：基于检索节点生成回答。"""
        query = await ctx.get("query")
        nodes = ev.nodes
        response = self.llm.complete(
            f"基于以下信息回答：{nodes}\n问题：{query}"
        )
        return GenerateEvent(response=response.text)

    @step
    async def finalize(
        self, ctx: Context, ev: GenerateEvent
    ) -> StopEvent:
        """终步骤：返回最终回答。"""
        return StopEvent(result=ev.response)

# 运行 Workflow
async def run():
    workflow = RAGWorkflow()
    result = await workflow.run(query="什么是 LlamaIndex？")
    print(result)

Context 的跨步骤状态传递

Context 对象在 Workflow 的所有 Step 之间共享，提供类型安全的键值存储：

python

@step
async def step_a(self, ctx: Context, ev: StartEvent) -> EventA:
    await ctx.set("user_name", "Alice")
    await ctx.set("history", ["第一轮交互"])
    return EventA(...)

@step
async def step_b(self, ctx: Context, ev: EventA) -> EventB:
    name = await ctx.get("user_name")  # "Alice"
    history = await ctx.get("history", default=[])  # 提供默认值
    ...

Workflow 的高级特性

并行 Step：多个 Step 可以同时监听同一事件，Workflow 引擎自动并行调度。
条件分支：Step 可以根据事件类型或内容选择性地触发不同下游 Step。
循环与重试：Step 可以重新触发自身或上游 Step，实现重试逻辑。
超时控制：Workflow 运行时可设置全局超时，避免无限挂起。

完整代码示例

以下是一个综合示例，展示从构建索引到 Workflow 查询的完整流程：

python

import asyncio
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.workflow import (
    Workflow,
    step, Context, StartEvent, StopEvent, Event,
)
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import CompactAndRefine
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置全局设置
Settings.llm = OpenAI(model="gpt-4o-mini")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 步骤一：构建数据管线
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=128),
        OpenAIEmbedding(),
    ]
)

# 加载文档
documents = SimpleDirectoryReader("./data").load_data()
nodes = pipeline.run(documents=documents)

# 构建索引
index = VectorStoreIndex(nodes)

# 步骤二：构建查询引擎
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=5,
)
synthesizer = CompactAndRefine(llm=Settings.llm)
query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=synthesizer,
)

# 步骤三：使用 Workflow 编排多步查询
class AdvancedQueryWorkflow(Workflow):
    @step
    async def retrieve_docs(
        self, ctx: Context, ev: StartEvent
    ) -> Event:
        query = ev.get("query")
        nodes = retriever.retrieve(query)
        await ctx.set("nodes", nodes)
        await ctx.set("query", query)
        return Event(payload={"type": "retrieved"})

    @step
    async def synthesize(
        self, ctx: Context, ev: Event
    ) -> StopEvent:
        query = await ctx.get("query")
        nodes = await ctx.get("nodes")
        response = query_engine.synthesize(query, nodes)
        return StopEvent(result=str(response))

# 执行 Workflow
async def main():
    workflow = AdvancedQueryWorkflow(timeout=60)
    result = await workflow.run(query="LlamaIndex 的核心特性有哪些？")
    print(f"回答：{result}")

asyncio.run(main())

总结

LlamaIndex 以其清晰的分层架构和以数据为中心的设计理念，在 RAG 应用开发中占据独特位置。其核心优势在于：

索引多样性：覆盖向量、关键词、摘要、树结构等多种索引方式，灵活应对不同数据形态。
检索策略丰富：从单一检索器到混合融合检索，再到可自定义的后处理逻辑。
查询引擎即用性：RetrieverQueryEngine 封装了检索与生成的标准流程，开箱即用。
数据管线规范化：IngestionPipeline 将文档处理流程标准化、可复现。
Workflow 引擎：事件驱动的图结构编排，适合构建复杂的多步 RAG 和 Agent 应用。

对于追求数据质量和检索效果的 RAG 项目，LlamaIndex 是比通用编排框架更具针对性的选择。

LlamaIndex 深入 ​

LlamaIndex 概览 ​

Index 索引体系 ​

Index 类型对比 ​

VectorStoreIndex 深入 ​

SummaryIndex 深入 ​

TreeIndex 深入 ​

KeywordTableIndex 深入 ​

Retriever 检索器 ​

BaseRetriever 核心接口 ​

主要 Retriever 实现 ​

混合检索 ​

Query Engine 查询引擎 ​

RetrieverQueryEngine ​

自定义 Query Engine ​

Ingestion Pipeline 数据管线 ​

流水线组成 ​

各阶段详解 ​

1. 文档加载（Readers） ​

2. 文档解析（Parsers） ​

3. 文本转换（Transformations） ​

4. 索引写入（Indexing） ​

Agent Runner / Workflow 引擎 ​

AgentRunner 与 AgentWorker ​

Workflow 事件驱动图 ​

核心概念 ​

Step 定义示例 ​

Context 的跨步骤状态传递 ​

Workflow 的高级特性 ​

完整代码示例 ​

总结 ​

LlamaIndex 深入

LlamaIndex 概览

Index 索引体系

Index 类型对比

VectorStoreIndex 深入

SummaryIndex 深入

TreeIndex 深入

KeywordTableIndex 深入

Retriever 检索器

BaseRetriever 核心接口

主要 Retriever 实现

混合检索

Query Engine 查询引擎

RetrieverQueryEngine

自定义 Query Engine

Ingestion Pipeline 数据管线

流水线组成

各阶段详解

1. 文档加载（Readers）

2. 文档解析（Parsers）

3. 文本转换（Transformations）

4. 索引写入（Indexing）

Agent Runner / Workflow 引擎

AgentRunner 与 AgentWorker

Workflow 事件驱动图

核心概念

Step 定义示例

Context 的跨步骤状态传递

Workflow 的高级特性

完整代码示例

总结