深入理解 LlamaIndex（v0.14.22）

基于 LlamaIndex v0.14.22 编写 — 最新稳定版本（2026年5月发布）

第一章：LlamaIndex 概述

1.1 什么是 LlamaIndex？

LlamaIndex 是一个开源数据框架，专门用于构建基于大语言模型（LLM）的 AI 应用。它提供了从数据接入、索引构建到检索增强生成（RAG）和 Agent 开发的完整工具链。

核心定位：为 LLM 应用程序提供数据骨架。

生态全景

LlamaIndex 生态系统
│
├── LlamaIndex OSS (开源框架)        ← 本教程核心
│   ├── llama-index-core            # 核心库
│   └── 300+ 集成包 (LlamaHub)      # 各种模型、向量库、工具
│
├── LlamaParse                      # 企业级文档解析/OCR
├── LlamaAgents                     # 多 Agent 编排系统
├── LlamaHub                        # 数据连接器和工具市场
├── llama_deploy                    # 部署微服务
└── create-llama                    # 项目脚手架 CLI

版本体系（v0.14.x 关键特性）

版本	日期	核心变更
0.10.0	2024-05	核心重构，命名空间分割
0.11.0	2024-09	Workflow 引擎引入
0.12.0	2025-01	AgentWorkflow 正式发布
0.14.0	2026-03	多模态合成、内存管理增强
0.14.22	2026-05	最新稳定版，修复多项核心问题

1.2 两种安装方式

# 方式一：全家桶安装（推荐新手）
pip install llama-index

# 方式二：核心 + 按需安装（推荐进阶开发者）
pip install llama-index-core
pip install llama-index-llms-openai      # OpenAI LLM
pip install llama-index-embeddings-openai  # OpenAI Embedding
pip install llama-index-vector-stores-chroma  # ChromaDB

为什么分拆？ v0.10 之后，LlamaIndex 采用了命名空间化架构。llama-index-core 包含核心抽象，各集成以独立包发布。这样你可以只安装你需要的依赖，减少包的体积。

导入规范

# 核心模块从 llama_index.core 导入
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.llms import LLM
from llama_index.core.agent import AgentRunner

# 集成模块从 llama_index.xxx 直接导入
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore

第二章：核心概念与架构

2.1 LlamaIndex 数据流水线

         ┌──────────┐     ┌───────────┐     ┌──────────┐     ┌───────────┐
         │ 数据加载   │ ──▶ │ 数据转换   │ ──▶ │ 索引构建  │ ──▶ │  存储持久化 │
         │ Loading   │     │ Transform │     │ Indexing │     │  Storing  │
         └──────────┘     └───────────┘     └──────────┘     └───────────┘
                                                                    │
                                                                    ▼
┌──────────┐     ┌──────────┐     ┌──────────┐     ┌───────────┐
│  用户查询  │ ──▶ │  检索器   │ ──▶ │  后处理   │ ──▶ │  查询引擎  │ ──▶ 回答
│  Query   │     │ Retrieve │     │ Post-    │     │  Query   │
│          │     │          │     │ process  │     │  Engine  │
└──────────┘     └──────────┘     └──────────┘     └──────────┘

核心抽象层

组件	职责	关键类
Document	数据源中的原始文档	`Document`
Node	文档切分后的最小语义单元	`TextNode`, `ImageNode`
Index	节点集合的结构化表示	`VectorStoreIndex`, `SummaryIndex`
Retriever	根据查询返回相关节点	`VectorIndexRetriever`, `BM25Retriever`
QueryEngine	检索 + 合成的完整查询管道	`RetrieverQueryEngine`
Postprocessor	检索结果的后处理/重排序	`SimilarityPostprocessor`, `SentenceTransformerRerank`
ResponseSynthesizer	将检索结果送入 LLM 生成回答	`ResponseSynthesizer`

2.2 Document 与 Node

Document（文档）

Document 是数据进入 LlamaIndex 的原始形式。每个 Document 包含：

from llama_index.core import Document

doc = Document(
    text="""LlamaIndex 是一个数据框架，旨在帮助构建基于 LLM 的应用。
它提供了数据连接器、索引构建和查询接口等功能。""",
    metadata={
        "source": "llama_index_intro.md",
        "author": "Jerry Liu",
        "page": 1,
        "category": "文档"
    },
    excluded_embed_metadata_keys=["page"],
    excluded_llm_metadata_keys=["page"],
)

关键属性：

text / id_：文本内容和唯一标识
metadata：元数据字典（可被检索和合成阶段使用）
excluded_embed_metadata_keys：嵌入时排除的元数据字段
excluded_llm_metadata_keys：LLM 合成时排除的元数据字段

Node（节点）

Node 是索引的最小单元。Document 经过 NodeParser 切分后产生若干 Node。

from llama_index.core.node_parser import SentenceSplitter

parser = SentenceSplitter(
    chunk_size=512,        # 每块大小（字符数或 token 数）
    chunk_overlap=128,     # 块间重叠字符数
    separator=" ",         # 分隔符
    paragraph_separator="\n\n",
    secondary_chunking_regex="[^,.;。]+[,.;。]?"  # 次选切分规则
)

nodes = parser.get_nodes_from_documents([doc])
# nodes[0] 是 TextNode，包含 text, embedding, metadata

节点关系

Node 之间可以建立关系，形成文档结构图：

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

node1 = TextNode(text="第一章：概述")
node2 = TextNode(text="第二章：安装")

node1.relationships[NodeRelationship.NEXT] = RelatedNodeInfo(
    node_id=node2.node_id, metadata={"section": "2"}
)
node2.relationships[NodeRelationship.PREVIOUS] = RelatedNodeInfo(
    node_id=node1.node_id, metadata={"section": "1"}
)

支持的节点关系类型：PARENT、CHILD、NEXT、PREVIOUS、SOURCE。

2.3 索引（Index）

Index 是 Node 的集合加上特定的数据结构。不同类型的 Index 适用于不同的检索场景。

VectorStoreIndex（向量索引）

最常用的索引类型。将 Node 嵌入为向量，存储在向量数据库中。

from llama_index.core import VectorStoreIndex
from llama_index.core import SimpleDirectoryReader

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

# 构建索引（自动完成：切分 → 嵌入 → 存储）
index = VectorStoreIndex.from_documents(
    documents,
    embed_model="default",  # 默认使用 OpenAI text-embedding-ada-002
    transformations=[SentenceSplitter(chunk_size=512)]
)

# 保存到磁盘
index.storage_context.persist("./storage")

# 从磁盘加载
from llama_index.core import StorageContext, load_index_from_storage
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

向量索引的内部结构

VectorStoreIndex
│
├── index_struct (IndexDict)
│   └── nodes_dict: {node_id: node_metadata}
│
├── vector_store (向量数据库)
│   └── 存储 node_id → embedding 映射
│
├── docstore (文档存储)
│   └── 存储 node_id → TextNode 完整内容
│
├── index_store (索引存储)
│   └── 存储 IndexStruct 序列化数据
│
└── transformations (转换管道)
    └── 存储构建索引时使用的转换函数

其他索引类型

# 摘要索引 - 适用于小数据集，整体检索后合成
from llama_index.core import SummaryIndex
summary_index = SummaryIndex.from_documents(documents)

# 树状索引 - 构建层次化摘要树
from llama_index.core import TreeIndex
tree_index = TreeIndex.from_documents(documents)

# 关键词索引 - 基于关键词匹配
from llama_index.core import KeywordTableIndex
keyword_index = KeywordTableIndex.from_documents(documents)

索引类型	检索方式	适用场景
VectorStoreIndex	语义相似度搜索	大多数 RAG 场景
SummaryIndex	遍历所有节点	小型数据集
TreeIndex	层次化摘要遍历	长文档理解
KeywordTableIndex	关键词提取 + 匹配	结构化查询
PropertyGraphIndex	知识图谱遍历	多跳推理

2.4 检索器（Retriever）

检索器负责根据查询找到相关节点。

# 默认向量检索器
retriever = index.as_retriever(
    similarity_top_k=3,  # 返回 top-3
)

# 手动构造
from llama_index.core.retrievers import VectorIndexRetriever
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=5,
    vector_store_query_mode="default",  # "default", "sparse", "hybrid"
    alpha=0.5,  # hybrid 模式下向量/稀疏权重
)

# 执行检索
nodes = retriever.retrieve("LlamaIndex 的核心功能是什么？")
for node in nodes:
    print(f"Score: {node.score:.4f} | {node.text[:80]}...")

多种检索策略

# BM25 混合检索（关键词 + 语义）
from llama_index.core.retrievers import BM25Retriever

bm25_retriever = BM25Retriever.from_defaults(
    index=index,
    similarity_top_k=5,
)
# 与 VectorRetriever 结果融合

# 自动检索（Auto-Retrieval）- 根据元数据过滤自动构建查询
from llama_index.core.retrievers import AutoRetriever
from llama_index.core.vector_stores.types import MetadataInfo, VectorStoreInfo

vector_store_info = VectorStoreInfo(
    content_info="公司文档",
    metadata_info=[
        MetadataInfo(name="category", type="str", description="文档分类"),
        MetadataInfo(name="date", type="int", description="日期"),
    ]
)

auto_retriever = AutoRetriever(
    vector_store_info=vector_store_info,
    retriever=retriever
)
# 自动从自然语言查询中提取过滤条件

2.5 后处理器（Postprocessor）

后处理器在检索后、合成前对结果进行过滤、重排序。

from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    SentenceTransformerRerank,
    MetadataReplacementPostProcessor,
    LongContextReorder,
)

# 1. 相似度阈值过滤
similarity_filter = SimilarityPostprocessor(similarity_cutoff=0.7)

# 2. 重排序（Rerank）
reranker = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3",
    top_k=3
)

# 3. 元数据替换（将检索到的节点中的某些元数据替换到文本中）
metadata_replacer = MetadataReplacementPostProcessor(
    target_metadata_key="source"
)

# 4. 长上下文重排序（将最相关的放在开头和结尾）
reorder = LongContextReorder()

# 组合使用
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import CompactAndRefine

query_engine = RetrieverQueryEngine(
    retriever=retriever,
    response_synthesizer=CompactAndRefine(),
    node_postprocessors=[
        similarity_filter,
        reranker,
        metadata_replacer,
        reorder
    ]
)

Rerank 为什么重要？

原始检索 Top-10:
[0.82, 0.79, 0.75, 0.72, 0.68, 0.65, 0.61, 0.58, 0.55, 0.52]

Rerank 后 Top-5:
[0.93, 0.91, 0.88, 0.85, 0.83]
↑ 相关性显著提升

Rerank 模型（如 Cohere Rerank、BGE Reranker）直接计算查询与文档的相关性，比向量检索的余弦相似度更准确。

2.6 查询引擎（QueryEngine）

查询引擎是 Retriever + ResponseSynthesizer 的组合，封装了完整的"检索→合成"流程。

ResponseSynthesizer 的 4 种模式

from llama_index.core.response_synthesizers import (
    Refine,
    CompactAndRefine,
    TreeSummarize,
    SimpleSummarize,
    NoText,
    Accumulate,
    CompactAccumulate
)

# 1. Refine（精炼模式）- 逐个节点迭代增强
# 首节点 → 回答；后续节点 → 精炼回答
synthesizer = Refine(
    llm=llm,
    streaming=True
)

# 2. CompactAndRefine（压缩精炼）- 默认模式
# 将尽可能多的节点塞入 LLM 上下文，逐次精炼
synthesizer = CompactAndRefine()

# 3. TreeSummarize（树状摘要）
# 递归地将节点分组摘要，适合大量节点
synthesizer = TreeSummarize()

# 4. Accumulate（累加模式）- 对每个节点单独回答，然后拼接
synthesizer = Accumulate()

# 5. NoText（仅检索）- 不经过 LLM，只返回检索结果
synthesizer = NoText()

自定义查询引擎

from llama_index.core.query_engine import CustomQueryEngine
from llama_index.core.llms import LLM
from llama_index.core import Response

class RAGQueryEngine(CustomQueryEngine):
    """自定义 RAG 查询引擎"""

    llm: LLM
    retriever: "BaseRetriever"

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

        # 2. 构建上下文
        context_str = "\n\n---\n\n".join([n.text for n in nodes])

        # 3. 构建提示词
        prompt = f"""
        基于以下上下文信息回答问题。
        如果上下文中没有足够信息，请如实说不知道。

        上下文：
        {context_str}

        问题：{query_str}

        回答：
        """

        # 4. LLM 生成
        response = self.llm.complete(prompt)

        return Response(
            response=response.text,
            source_nodes=nodes
        )

第三章：LLM 的使用与配置

3.1 LLM 抽象层

LlamaIndex 提供了一个统一的 LLM 接口，屏蔽了不同供应商的差异。

from llama_index.core.llms import LLM, ChatMessage, MessageRole

class LLM:
    """LLM 抽象基类的核心接口"""

    # 核心方法
    async def complete(self, prompt: str, **kwargs) -> CompletionResponse: ...
    async def chat(self, messages: list[ChatMessage], **kwargs) -> ChatResponse: ...
    async def stream_complete(self, prompt: str, **kwargs) -> StreamingCallback: ...
    async def stream_chat(self, messages: list[ChatMessage], **kwargs) -> StreamingCallback: ...

    # 工具调用
    async def predict_and_call(self, tools: list[BaseTool], user_msg: str) -> ...
    async def achat_with_tools(self, tools: list[BaseTool], chat_history: list[ChatMessage]) -> ...

3.2 主流 LLM 集成

# === OpenAI ===
from llama_index.llms.openai import OpenAI

llm = OpenAI(
    model="gpt-4o",               # 也支持 gpt-4o-mini, gpt-4-turbo
    temperature=0.3,
    max_tokens=4096,
    api_key="sk-xxx",             # 默认使用环境变量 OPENAI_API_KEY
)

# === Anthropic Claude ===
from llama_index.llms.anthropic import Anthropic

llm = Anthropic(
    model="claude-sonnet-4-20250514",
    temperature=0.7,
    max_tokens=8192,
)

# === 本地 Ollama ===
from llama_index.llms.ollama import Ollama

llm = Ollama(
    model="llama3.3",
    request_timeout=120.0,
    context_window=8000,
)

# === 阿里云通义千问 ===
from llama_index.llms.dashscope import DashScope

llm = DashScope(
    model="qwen-max",
    api_key="sk-xxx",
)

设置全局 LLM

from llama_index.core import Settings

# 设置全局 LLM（所有组件默认使用此 LLM）
Settings.llm = OpenAI(model="gpt-4o", temperature=0.1)

# 设置全局 Embedding 模型
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-small"
)

# 设置全局切分器
Settings.node_parser = SentenceSplitter(chunk_size=1024)

# 设置全局上下文窗口
Settings.context_window = 16384

3.3 LLM 与工具调用

from llama_index.core.tools import FunctionTool

# 定义工具函数
def search_weather(city: str) -> str:
    """查询指定城市的天气"""
    # ... 调用天气 API
    return f"{city}：晴，25°C"

weather_tool = FunctionTool.from_defaults(
    fn=search_weather,
    async_fn=search_weather  # 异步版本（可选）
)

# LLM 根据用户请求自动决定是否调用工具
response = llm.predict_and_call(
    [weather_tool],
    "北京今天天气怎么样？"
)
print(str(response))
# 输出：北京今天天气：晴，25°C

第四章：数据接入与变换

4.1 数据加载器（Reader）

LlamaIndex 通过 LlamaHub 提供了 160+ 数据连接器。

# 加载本地文件目录
from llama_index.core import SimpleDirectoryReader

reader = SimpleDirectoryReader(
    input_dir="./data",
    recursive=True,              # 递归子目录
    required_exts=[".pdf", ".md", ".txt"],  # 仅加载指定扩展名
    file_metadata=lambda f: {"file": f, "loaded_at": str(datetime.now())},
)

documents = reader.load_data()

# 加载 PDF
from llama_index.readers.file import PDFReader
pdf_reader = PDFReader()
pdf_docs = pdf_reader.load_data(file="./report.pdf")

# 加载网页
from llama_index.readers.web import SimpleWebPageReader
web_docs = SimpleWebPageReader().load_data(["https://example.com"])

# 加载数据库
from llama_index.readers.database import DatabaseReader
db_reader = DatabaseReader(
    scheme="postgresql",
    host="localhost",
    port=5432,
    user="postgres",
    password="xxx",
    dbname="mydb",
)
sql_docs = db_reader.load_data(query="SELECT * FROM articles")

自定义 Reader

from llama_index.core.readers import BaseReader
from llama_index.core.schema import Document

class MyAPIReader(BaseReader):
    """自定义 API 数据加载器"""

    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url

    def load_data(self, endpoint: str) -> list[Document]:
        import requests
        response = requests.get(
            f"{self.base_url}/{endpoint}",
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        data = response.json()

        documents = []
        for item in data:
            doc = Document(
                text=item["content"],
                metadata={
                    "id": item["id"],
                    "title": item["title"],
                    "created_at": item["created_at"]
                }
            )
            documents.append(doc)
        return documents

4.2 数据变换管道（IngestionPipeline）

数据变换管道实现了处理链模式，可以对文档进行清洗、切分、嵌入等操作。

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import (
    TitleExtractor,
    QuestionsAnsweredExtractor,
    SummaryExtractor,
)
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.chroma import ChromaVectorStore

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(
            chunk_size=1024,
            chunk_overlap=256,
        ),
        TitleExtractor(llm=llm),           # 自动提取标题
        QuestionsAnsweredExtractor(llm=llm), # 生成可能回答的问题
        SummaryExtractor(llm=llm),           # 生成摘要
        OpenAIEmbedding(model="text-embedding-3-small"),
    ],
    # 可选：直接将结果写入向量库
    vector_store=ChromaVectorStore(
        collection_name="my_docs",
        chroma_client=chroma_client
    )
)

# 执行管道
nodes = pipeline.run(documents=documents)

# 增量更新
pipeline.run(documents=[new_doc], cache_collection="ingestion_cache")

管道缓存

IngestionPipeline 内置缓存机制，避免重复计算：

pipeline = IngestionPipeline(
    transformations=[...],
    cache=IngestionCache(
        collection="pipeline_cache",
        vector_store=ChromaVectorStore(...)
    )
)

# 第一次运行：全部计算
nodes1 = pipeline.run(documents=docs1)

# 第二次运行（新增文档）：只处理新增文档，已有的跳过
nodes2 = pipeline.run(documents=docs1 + docs2)
# 只有 docs2 被真正处理

4.3 自定义转换

from llama_index.core.ingestion import run_transformations
from llama_index.core.schema import TransformComponent

class TextCleaner(TransformComponent):
    """自定义文本清洗转换"""

    @classmethod
    def class_name(cls) -> str:
        return "TextCleaner"

    def __call__(self, nodes: list[TextNode], **kwargs) -> list[TextNode]:
        for node in nodes:
            # 去除多余空白
            node.text = " ".join(node.text.split())
            # 转换统一编码
            import unicodedata
            node.text = unicodedata.normalize("NFKC", node.text)
        return nodes

# 使用
pipeline = IngestionPipeline(
    transformations=[
        TextCleaner(),
        SentenceSplitter(chunk_size=512),
        OpenAIEmbedding(),
    ]
)

第五章：检索增强生成（RAG）深度实践

5.1 基础 RAG 流水线

# ============ 第1步：加载数据 ============
from llama_index.core import SimpleDirectoryReader
documents = SimpleDirectoryReader("./docs").load_data()

# ============ 第2步：构建索引 ============
from llama_index.core import VectorStoreIndex
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

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

index = VectorStoreIndex.from_documents(documents)

# ============ 第3步：查询 ============
query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact",
    streaming=True
)

response = query_engine.query("LlamaIndex 有哪些核心概念？")
print(response)

# 获取来源信息
for node in response.source_nodes:
    print(f"来源: {node.metadata.get('file_path', 'unknown')}")
    print(f"分数: {node.score:.4f}")
    print(f"内容: {node.text[:100]}...")
    print("---")

5.2 高级 RAG：HyDE（假设文档嵌入）

HyDE 的原理：先让 LLM 根据查询生成一个"假设文档"，再用这个文档去向量检索。

from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.query_engine import RetrieverQueryEngine

# HyDE 检索器：先用 LLM 生成假设回答，再用回答去检索
from llama_index.core.retrievers import HyDERetriever

hyde_retriever = HyDERetriever(
    retriever=index.as_retriever(similarity_top_k=5),
    llm=llm,
    # 可选：自定义假设文档生成提示词
    hyde_prompt="请写出一个完美的假设答案，答案应该详细、准确。"
)

query_engine = RetrieverQueryEngine.from_args(
    retriever=hyde_retriever,
    llm=llm
)

5.3 高级 RAG：Agentic RAG（Agent 驱动的检索）

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

# 准备多个专用查询引擎
finance_engine = finance_index.as_query_engine(similarity_top_k=3)
tech_engine = tech_index.as_query_engine(similarity_top_k=3)

# 封装为工具
tools = [
    QueryEngineTool(
        query_engine=finance_engine,
        metadata=ToolMetadata(
            name="finance_qa",
            description="回答关于财务报告的问题。查询应包含具体年份和指标。"
        )
    ),
    QueryEngineTool(
        query_engine=tech_engine,
        metadata=ToolMetadata(
            name="tech_qa",
            description="回答关于技术文档的问题。"
        )
    ),
    # 添加非 RAG 工具
    FunctionTool.from_defaults(fn=search_weather),
]

# Agent Runner 自动选择工具
agent = AgentRunner.from_llm(
    tools=tools,
    llm=llm,
    verbose=True,
    max_function_calls=10,
)

response = agent.chat("2024年公司在AI领域的研发投入是多少？并对比2023年。")

5.4 高级 RAG：Graph RAG（知识图谱增强检索）

from llama_index.core.indices.property_graph import PropertyGraphIndex
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI

# 构建属性图谱索引
graph_index = PropertyGraphIndex.from_documents(
    documents,
    embed_model=OpenAIEmbedding(),
    llm=OpenAI(model="gpt-4o"),
    # 知识图谱提取配置
    kg_extractors=[
        KGExtractor(llm=llm, max_path_depth=3)
    ],
)

# 图谱检索器 - 结合向量搜索和图遍历
from llama_index.core.retrievers import KGTableRetriever
kg_retriever = KGTableRetriever(
    graph_index=graph_index,
    similarity_top_k=3,
    graph_traversal_depth=2,
    include_text=True,
)

query_engine = RetrieverQueryEngine(
    retriever=kg_retriever,
    response_synthesizer=CompactAndRefine()
)

# 查询支持多跳推理
response = query_engine.query("A创投投资了哪些AI公司？这些公司中哪些与B公司有合作？")

5.5 多模态 RAG（v0.14 新特性）

# v0.14 引入的多模态合成能力
from llama_index.core.query_engine import MultiModalQueryEngine
from llama_index.core import SimpleDirectoryReader
from llama_index.indices.multi_modal import MultiModalVectorStoreIndex

# 同时加载文本和图片
documents = SimpleDirectoryReader("./multimodal_data").load_data()

# 多模态索引：文本用 text-embedding，图片用 CLIP
from llama_index.embeddings.clip import ClipEmbedding
from llama_index.core import Settings

Settings.embed_model = ClipEmbedding()

index = MultiModalVectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(
    llm=OpenAI(model="gpt-4o"),  # 支持图文理解的 LLM
    similarity_top_k=3,
)

# 问题可以涉及图片内容
response = query_engine.query("图中的产品有什么特点？")

5.6 RAG 评估

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

# 1. 生成评测数据集
dataset_generator = DatasetGenerator(
    documents[:10],  # 使用前10个文档
    llm=llm,
    num_questions_per_chunk=2,
)
eval_dataset = dataset_generator.generate_dataset_from_nodes()

# 2. 运行评估
faithfulness_evaluator = FaithfulnessEvaluator(llm=llm)
relevancy_evaluator = RelevancyEvaluator(llm=llm)

for query, expected_response in eval_dataset.qr_pairs:
    # 执行查询
    response = query_engine.query(query)

    # 评估忠实度（回答是否基于提供的上下文）
    faith_result = faithfulness_evaluator.evaluate_response(
        query=query, response=response
    )

    # 评估相关性（检索的节点是否与问题相关）
    relev_result = relevancy_evaluator.evaluate_response(
        query=query, response=response
    )

    print(f"Query: {query}")
    print(f"Faithfulness: {faith_result.passing} (Score: {faith_result.score})")
    print(f"Relevancy: {relev_result.passing} (Score: {relev_result.score})")
    print("---")

第六章：Agent 系统

6.1 单 Agent（AgentRunner）

LlamaIndex 的 Agent 基于 LLM + 工具 模式构建。

from llama_index.core.agent import AgentRunner, ReActAgentWorker
from llama_index.core.tools import FunctionTool, QueryEngineTool

# 定义工具
def multiply(a: float, b: float) -> float:
    """将两个数相乘"""
    return a * b

def add(a: float, b: float) -> float:
    """将两个数相加"""
    return a + b

calc_tool = FunctionTool.from_defaults(fn=multiply)
add_tool = FunctionTool.from_defaults(fn=add)

# 创建 Agent
agent = AgentRunner.from_llm(
    tools=[calc_tool, add_tool],
    llm=llm,
    system_prompt="你是一个数学助手。使用工具完成计算。",
    max_function_calls=10,   # 最大工具调用次数
    verbose=True,             # 打印中间步骤
)

# 执行
response = agent.chat("计算 (23 * 45) + 100 的结果")
print(response)

Agent 内部工作流程

用户输入 "计算 (23 * 45) + 100"
    │
    ▼
1. LLM 分析：需要使用工具逐步计算
    │
    ▼
2. LLM 输出：调用 multiply(a=23, b=45)
    │
    ▼
3. 执行工具 → 结果: 1035
    │
    ▼
4. 工具结果回传给 LLM
    │
    ▼
5. LLM 再次分析：还需要 add(1035, 100)
    │
    ▼
6. 执行工具 → 结果: 1135
    │
    ▼
7. 最终回答："(23 * 45) + 100 = 1135"

6.2 AgentWorkflow（v0.12+ 新特性）

AgentWorkflow 是 多 Agent 协作 的官方方案，采用事件驱动的编排方式。

from llama_index.core.agent.workflow import (
    AgentWorkflow,
    AgentWorkflowContext,
    ToolAgentResult,
)

# === 定义多个 Agent ===

# Agent 1：研究助手
research_agent = AgentRunner.from_llm(
    tools=[web_search_tool, finance_qa_tool],
    llm=llm,
    system_prompt="你是研究助手，负责收集和分析信息。",
)

# Agent 2：写作助手
writing_agent = AgentRunner.from_llm(
    tools=[FileWriteTool],
    llm=llm,
    system_prompt="你是写作助手，负责撰写和优化报告内容。",
)

# Agent 3：审核助手
review_agent = AgentRunner.from_llm(
    tools=[],
    llm=llm,
    system_prompt="你是质量审核员，检查报告是否完整、准确。",
)

# === 编排 Workflow ===
workflow = AgentWorkflow(
    agents=[research_agent, writing_agent, review_agent],
    root_agent="research_agent",  # 起始 Agent
    context=AgentWorkflowContext(
        state={
            "topic": "",
            "research_notes": [],
            "draft": "",
            "final_report": "",
        }
    ),
)

# 执行
result = await workflow.run(
    user_msg="写一份关于2025年AI芯片市场的分析报告"
)

# AgentWorkflow 的自动编排过程：
# 1. research_agent 收集资料 → 存到 context.state.research_notes
# 2. writing_agent 撰写草稿 → 存到 context.state.draft
# 3. review_agent 审核修改 → 存到 context.state.final_report

6.3 人机协作（Human-in-the-Loop）

from llama_index.core.agent.workflow import (
    AgentWorkflow,
    HumanInputTool,
)

# 定义人工介入点
human_tool = HumanInputTool(
    description="当你需要用户确认或补充信息时使用此工具。"
)

agent = AgentRunner.from_llm(
    tools=[finance_qa_tool, human_tool],
    llm=llm,
)

response = agent.chat("分析本月销售数据，如果有异常需要用户确认如何处理。")
# Agent 会在必要时暂停等待用户输入

6.4 Agent 的状态记忆

# 默认 Agent 使用 ChatMemoryBuffer 维护对话历史
from llama_index.core.memory import ChatMemoryBuffer

memory = ChatMemoryBuffer(
    token_limit=4096,
    chat_history=[],  # 可选：加载历史对话
)

agent = AgentRunner.from_llm(
    tools=tools,
    llm=llm,
    memory=memory,  # 赋予 Agent 短期记忆
    max_function_calls=10,
)

# 多次对话
agent.chat("帮我查一下苹果公司2024年的营收")
agent.chat("和2023年比增长了多少？")  # Agent 记得上下文
agent.chat("写一封总结邮件给CEO")  # Agent 利用之前的信息

第七章：Workflow 引擎

Workflow 是 LlamaIndex 的事件驱动执行框架（v0.11 引入），允许你构建任意复杂的 Agentic 应用逻辑。

7.1 Workflow 核心概念

Workflow = Step 的有向图

每个 Step:
  - 接收特定 Event 类型
  - 处理逻辑
  - 返回新的 Event
  - 可保持状态

一个 Workflow 的典型结构:
  [StartEvent] ──▶ [Step A] ──▶ [Step B] ──▶ [StopEvent]
                        │            ▲
                        ▼            │
                     [Step C] ───────┘

7.2 构建简单 Workflow

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

class MyRAGWorkflow(Workflow):
    """自定义 RAG Workflow"""

    @step
    async def retrieve(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """步骤1：检索"""
        query = ev.query

        # 检索相关节点
        nodes = self.retriever.retrieve(query)

        # 保存到 Workflow 上下文
        await ctx.set("nodes", nodes)
        await ctx.set("query", query)

        return StopEvent(result=nodes)  # 传递给下一步

    @step
    async def synthesize(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """步骤2：合成（需要 retrieve 先执行完毕）"""
        nodes = await ctx.get("nodes")
        query = await ctx.get("query")

        # 使用 LLM 合成
        context_str = "\n\n".join([n.text for n in nodes])
        response = await self.llm.acomplete(
            f"基于以下内容回答问题：\n{context_str}\n\n问题：{query}"
        )

        return StopEvent(result=str(response))

# 使用
workflow = MyRAGWorkflow(retriever=retriever, llm=llm, verbose=True)
result = await workflow.run(query="LlamaIndex 是什么？")

7.3 分支与循环

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

class AnalyzeWorkflow(Workflow):

    @step
    async def analyze(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """分析步骤 - 根据复杂度决定是否继续"""
        data = ev.data

        # LLM 分析任务复杂度
        response = await self.llm.acomplete(
            f"分析以下任务的复杂度，如果复杂就回复 COMPLEX，否则回复 SIMPLE：\n{data}"
        )

        is_complex = "COMPLEX" in str(response)
        await ctx.set("attempt", 1)

        if is_complex:
            # 返回 Event 触发复杂处理分支
            return ComplexTaskEvent(data=data)
        else:
            return SimpleTaskEvent(data=data)

    @step
    async def complex_handler(self, ctx: Context, ev: ComplexTaskEvent) -> StopEvent:
        """复杂任务处理"""
        data = ev.data
        result = await solve_complex(data)
        return StopEvent(result=result)

    @step
    async def simple_handler(self, ctx: Context, ev: SimpleTaskEvent) -> StopEvent:
        """简单任务处理"""
        data = ev.data
        result = await solve_simple(data)
        return StopEvent(result=result)

    @step
    async def retry_check(self, ctx: Context, ev: ErrorEvent) -> StopEvent:
        """失败重试 - 循环控制"""
        attempt = await ctx.get("attempt", default=0)
        max_retries = 3

        if attempt < max_retries:
            await ctx.set("attempt", attempt + 1)
            return RetryEvent(data=ev.data)  # 重试
        else:
            return StopEvent(result="任务失败，已达最大重试次数")

7.4 并发执行

class ParallelSearchWorkflow(Workflow):

    @step
    async def dispatch(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """分发多个并行搜索任务"""
        query = ev.query

        # 同时触发3个搜索（返回多个 Event 实现并行）
        return StopEvent(result={
            "web": "web_search",
            "news": "news_search",
            "academic": "academic_search",
        })

    @step(num_workers=3)  # 3个工作线程并行
    async def web_search(self, ctx: Context, ev: StartEvent) -> StopEvent:
        query = await ctx.get("query")
        result = await search_web(query)
        await ctx.set("web_result", result)
        return StopEvent(result=result)

    @step(num_workers=3)
    async def news_search(self, ctx: Context, ev: StartEvent) -> StopEvent:
        query = await ctx.get("query")
        result = await search_news(query)
        await ctx.set("news_result", result)
        return StopEvent(result=result)

    @step(num_workers=3)
    async def academic_search(self, ctx: Context, ev: StartEvent) -> StopEvent:
        query = await ctx.get("query")
        result = await search_academic(query)
        await ctx.set("academic_result", result)
        return StopEvent(result=result)

    @step
    async def merge(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """合并所有并行结果"""
        web = await ctx.get("web_result")
        news = await ctx.get("news_result")
        academic = await ctx.get("academic_result")

        merged = await self.llm.acomplete(
            f"综合以下信息：\n网页：{web}\n新闻：{news}\n学术：{academic}"
        )
        return StopEvent(result=str(merged))

7.5 Workflow 流式输出

class StreamingWorkflow(Workflow):

    @step
    async def generate(self, ctx: Context, ev: StartEvent) -> StopEvent:
        """流式生成回答"""
        query = ev.query

        stream = await self.llm.astream_complete(f"请详细回答：{query}")

        async for chunk in stream:
            # 发送 ProgressEvent，前端可以逐字显示
            ctx.write_event_to_stream(
                ProgressEvent(msg=chunk.delta)
            )

        return StopEvent(result="完成")

第八章：存储与管理

8.1 StorageContext

from llama_index.core.storage import StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.storage.docstore.mongodb import MongoDocumentStore
from llama_index.storage.index_store.mongodb import MongoIndexStore

# 自定义存储后端
storage_context = StorageContext.from_defaults(
    vector_store=ChromaVectorStore(
        collection_name="my_docs",
        chroma_client=chroma_client,
    ),
    docstore=MongoDocumentStore.from_uri("mongodb://localhost:27017"),
    index_store=MongoIndexStore.from_uri("mongodb://localhost:27017"),
)

# 构建索引时使用
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
)

# 加载已有索引
from llama_index.core import load_index_from_storage
loaded_index = load_index_from_storage(storage_context)

8.2 向量数据库支持

数据库	包名	特点
ChromaDB	`llama-index-vector-stores-chroma`	轻量、本地
Pinecone	`llama-index-vector-stores-pinecone`	托管、高性能
Weaviate	`llama-index-vector-stores-weaviate`	云原生
Qdrant	`llama-index-vector-stores-qdrant`	自托管可选
Milvus	`llama-index-vector-stores-milvus`	大规模分布式
MongoDB Atlas	`llama-index-vector-stores-mongodb`	已有 MongoDB 集成
Elasticsearch	`llama-index-vector-stores-elasticsearch`	全文+向量混合
PGVector	`llama-index-vector-stores-postgres`	PostgreSQL 扩展

第九章：可观测性与调试

9.1 回调系统

from llama_index.core.callbacks import (
    CallbackManager,
    LlamaDebugHandler,
    TokenCountingHandler,
    CBEventType,
)

# 调试回调 - 打印所有事件
debug_handler = LlamaDebugHandler(
    event_starts_to_ignore=[],  # 默认忽略的事件类型
    event_ends_to_ignore=[],
)

# Token 计数
token_counter = TokenCountingHandler()

Settings.callback_manager = CallbackManager([
    debug_handler,
    token_counter,
])

# 执行查询
response = query_engine.query("LlamaIndex 是什么？")

# 查看调试信息
for event_type, events in debug_handler.event_pairs.items():
    for start_event, end_event in events:
        print(f"事件: {event_type}, 耗时: {end_event.time - start_event.time}")
        print(f"负载: {end_event.payload}")

# 查看 Token 消耗
print(f"LLM Token 使用: {token_counter.total_llm_token_count}")
print(f"Embedding Token 使用: {token_counter.total_embedding_token_count}")

9.2 Arize Phoenix 集成

import phoenix as px
from llama_index.callbacks.arize_phoenix import (
    arize_phoenix_callback_handler,
)

# 启动 Phoenix UI
session = px.launch_app()

# 注册回调
phoenix_handler = arize_phoenix_callback_handler(
    endpoint="http://127.0.0.1:6006/v1/traces",
)

Settings.callback_manager = CallbackManager([phoenix_handler])

9.3 OpenTelemetry 集成

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider

# 初始化 OpenTelemetry
provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces"))
)
trace.set_tracer_provider(provider)

# LlamaIndex 自动集成 OpenTelemetry
# 配置环境变量：
# OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
# OTEL_SERVICE_NAME=llama-index-app

第十章：生产部署最佳实践

10.1 文档索引策略

# 推荐的生产级索引流程
from llama_index.core.node_parser import (
    SemanticSplitterNodeParser,  # 语义切分（v0.12+）
)
from llama_index.core.extractors import (
    KeywordExtractor,
    TitleExtractor,
)

# 语义切分：根据语义边界切分，优于固定字符数
from llama_index.embeddings.openai import OpenAIEmbedding

splitter = SemanticSplitterNodeParser(
    embed_model=OpenAIEmbedding(),
    breakpoint_percentile_threshold=95,  # 相似度断点阈值
    buffer_size=1,  # 上下文窗口大小
)

# 元数据提取
title_extractor = TitleExtractor(llm=llm)
keyword_extractor = KeywordExtractor(llm=llm, keywords=5)

# 组合管道
pipeline = IngestionPipeline(
    transformations=[
        splitter,
        title_extractor,
        keyword_extractor,
        OpenAIEmbedding(),
    ]
)

10.2 混合检索（Hybrid Search）

from llama_index.core.retrievers import (
    VectorIndexRetriever,
    BM25Retriever,
)
from llama_index.core.retrievers import RouterRetriever
from llama_index.core.query_engine import RetrieverQueryEngine

# 向量检索
vector_retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=5,
)

# BM25 关键词检索
bm25_retriever = BM25Retriever.from_defaults(
    index=index,
    similarity_top_k=5,
)

# 融合检索（取并集 + 重排序）
from llama_index.core.retrievers import QueryFusionRetriever

fusion_retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    similarity_top_k=5,
    num_queries=1,  # 对原始查询不做改写
    mode="reciprocal_rerank",  # 使用互惠排名融合
)

query_engine = RetrieverQueryEngine.from_args(
    retriever=fusion_retriever,
    node_postprocessors=[
        SentenceTransformerRerank(
            model="BAAI/bge-reranker-v2-m3",
            top_k=3,
        )
    ]
)

10.3 缓存策略

from llama_index.core.cache import LLMCache, EmbeddingCache
import diskcache

# LLM 回答缓存（相同查询直接返回缓存结果）
llm_cache = LLMCache(
    cache=diskcache.Cache("./llm_cache"),
    similarity_threshold=0.9,  # 相似度超过阈值则命中
)

Settings.llm = OpenAI(model="gpt-4o", cache=llm_cache)

# Embedding 缓存
embed_cache = EmbeddingCache(
    cache=diskcache.Cache("./embed_cache")
)

Settings.embed_model = OpenAIEmbedding(cache=embed_cache)

10.4 异步与并发

# 所有 LlamaIndex 操作都有对应的异步版本
import asyncio

async def async_rag():
    # 异步加载
    documents = await SimpleDirectoryReader("./docs").aload_data()

    # 异步构建索引
    index = await VectorStoreIndex.from_documents(documents)

    # 异步查询
    query_engine = index.as_query_engine()
    response = await query_engine.aquery("什么是 LlamaIndex？")

    # 多个查询并发执行
    queries = [
        "LlamaIndex 的功能",
        "如何安装 LlamaIndex",
        "RAG 的原理"
    ]
    tasks = [query_engine.aquery(q) for q in queries]
    responses = await asyncio.gather(*tasks)

    return responses

responses = asyncio.run(async_rag())

10.5 部署架构建议

                    ┌──────────────┐
                    │   客户端      │
                    │ (前端/API)    │
                    └──────┬──────┘
                           │ HTTP
                    ┌──────▼──────┐
                    │   API 网关   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
        ┌─────▼────┐ ┌────▼────┐ ┌────▼────┐
        │ 查询服务   │ │ 索引服务  │ │ Agent   │
        │ (只读)    │ │ (写入)   │ │ 服务    │
        └─────┬────┘ └─────────┘ └─────────┘
              │
        ┌─────▼────┐
        │ 向量数据库 │
        │ (Chroma/  │
        │  Pinecone)│
        └──────────┘

附录

A. 常用命令速查

# 安装
pip install llama-index

# 创建新项目
npx create-llama@latest my-app

# 查看版本
pip show llama-index

# 列出所有集成包
pip list | grep llama-index

# 更新
pip install --upgrade llama-index

B. 核心类图谱

Settings (全局配置)
  ├── llm → LLM
  ├── embed_model → BaseEmbedding
  ├── node_parser → NodeParser
  ├── callback_manager → CallbackManager
  └── context_window → int

Document → NodeParser → Node
  │                        │
  ▼                        ▼
IngestionPipeline    VectorStoreIndex (or other Index)
                           │
                           ▼
                      Retriever
                           │
                      Postprocessor
                           │
                      ResponseSynthesizer
                           │
                      QueryEngine
                           │
                      Response

C. 版本迁移要点（从 0.10 前迁移）

旧版本 (v0.9.x)	新版本 (v0.14.x)
`from llama_index import ...`	`from llama_index.core import ...`
`ServiceContext`	`Settings`
`GPTVectorStoreIndex`	`VectorStoreIndex`
`LLMPredictor`	直接使用 `Settings.llm`
`node_parser.py`	`from llama_index.core.node_parser`
`response.text`	`str(response)`

D. 术语表

术语	说明
RAG	检索增强生成：从知识库检索相关内容，注入 LLM 上下文
Embedding	嵌入：将文本转换为向量表示
Chunk	分块：将长文本切分为小段落
Rerank	重排序：用专门的模型重新评估检索结果的相关性
HyDE	假设文档嵌入：先生成假设回答再检索
DAG	有向无环图：Workflow 的执行拓扑
Callback	回调：事件驱动的监控/调试机制
Index	索引：节点集合的数据结构
Node	节点：最小语义单元
Workflow	工作流：事件驱动的步骤编排
Agent	具备工具调用能力的 LLM 应用
Tool	工具：Agent 可以调用的外部功能

教程结语

LlamaIndex 已经从一个简单的"数据索引工具"进化为完整的 LLM 应用开发框架。其核心价值在于提供了一个模块化的抽象层，让开发者可以自由组合不同的组件，从简单的 RAG 到复杂的多 Agent 协作系统。

学习路径建议：

🔰 掌握 Document → Index → Retriever → QueryEngine 基础流水线

🔧 理解 Settings 全局配置和各种索引类型的适用场景

🤖 学习 Agent 和工具的集成使用

⚙️ 使用 Workflow 构建自定义执行逻辑

📊 集成可观测性工具进行调试和优化

🚀 探索 Graph RAG、多模态等高级特性

LlamaIndex_精通教程