LlamaIndex企业级实战指南：从数据摄取管道到高级RAG架构与Agentic检索系统全流程解析

引言

LlamaIndex是专为构建LLM数据应用而设计的框架，其核心理念是"连接你的数据与大语言模型"。与LangChain侧重于链式编排不同，LlamaIndex专注于数据处理、索引构建和检索增强生成（RAG），在这一领域提供了更精细的抽象和更丰富的工具链。2026年的LlamaIndex已经发展为一个包含LlamaIndex Core、LlamaIndex LLM Packs、LlamaParse（文档解析）、LlamaCloud（托管服务）的完整生态。本文将从数据摄取开始，系统性地讲解文档解析、索引策略、高级检索技术（递归检索、句子窗口检索、自动合并检索）、Agentic RAG、查询引擎、评估框架等核心内容，通过大量可运行的Python代码示例帮助读者掌握企业级RAG系统构建。

一、LlamaIndex核心架构与数据抽象

1.1 核心概念

LlamaIndex的核心数据抽象包括以下层次。Document是最顶层的数据单元，代表一个完整的文档（如一个PDF文件、一篇Markdown文章），包含文本内容和元数据。Node是文档被分块后的最小检索单元，可以是文本块、图像块或其他模态的数据。Index是对Node集合建立的索引结构，支持不同类型的索引（向量索引、关键词索引、知识图谱索引等）。Retriever负责从Index中检索相关Node。QueryEngine将Retriever与LLM结合，生成对用户查询的回答。ResponseSynthesizer负责将检索到的Node合成为最终答案。以下是一个基本的数据流示例：

from llama_index.core import Document, VectorStoreIndex, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import CompactMode
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-large")

# 创建文档
doc = Document(
    text="""
    # FastAPI性能优化指南
    
    FastAPI是一个现代的Python Web框架，基于ASGI标准构建。
    本文将介绍几种提升FastAPI应用性能的方法。
    
    ## 1. 异步数据库访问
    使用asyncpg替代psycopg2可以获得显著的性能提升。
    asyncpg是一个高效的异步PostgreSQL驱动，支持连接池。
    
    ## 2. 响应缓存
    使用Redis对GET请求的响应进行缓存，可以减少数据库查询。
    
    ## 3. 分页优化
    对于大量数据的分页查询，使用游标分页替代偏移分页。
    """,
    metadata={
        "title": "FastAPI性能优化指南",
        "author": "技术团队",
        "category": "backend",
        "tags": ["fastapi", "python", "performance"],
    }
)

# 分块
splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = splitter.get_nodes_from_documents([doc])

print(f"文档分块数: {len(nodes)}")
for i, node in enumerate(nodes):
    print(f"  Node {i}: {node.text[:80]}...")

# 建立索引
index = VectorStoreIndex(nodes)

# 创建查询引擎
query_engine = index.as_query_engine(
    response_mode=CompactMode,
    similarity_top_k=3,
)

# 查询
response = query_engine.query("如何提升FastAPI的数据库性能？")
print(response.response)
print(f"\n来源节点: {[n.metadata for n in response.source_nodes]}")

1.2 LlamaIndex vs LangChain

虽然两个框架都支持RAG，但侧重点不同。LlamaIndex在数据处理、索引结构和检索策略方面提供了更丰富的选项，特别是其高级检索技术（递归检索、句子窗口检索等）在处理复杂文档时表现更优。LangChain在Agent编排、工具调用和多步推理方面更成熟。在实际项目中，两者经常配合使用——用LlamaIndex构建数据处理和检索管道，用LangChain/LangGraph构建Agent层。

二、数据摄取与文档解析

2.1 LlamaParse文档解析

LlamaParse是LlamaIndex推出的AI增强文档解析服务，能将PDF、Word、PPT等复杂文档高效解析为结构化的Markdown。以下是一个使用LlamaParse解析复杂文档的示例：

from llama_parse import LlamaParse
from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage
from llama_index.core.node_parser import MarkdownNodeParser, SentenceSplitter
import os
from pathlib import Path

# 配置LlamaParse
parser = LlamaParse(
    api_key=os.environ.get("LLAMA_CLOUD_API_KEY"),
    result_type="markdown",  # 输出Markdown格式
    premium_mode=True,       # 高精度模式，支持表格和图表
    language="zh",           # 中文优化
)

# 解析PDF文档
documents = parser.load_data("./docs/architecture-design.pdf")

print(f"解析得到 {len(documents)} 个文档")
for doc in documents:
    print(f"  文档: {doc.metadata.get('file_name', 'unknown')}")
    print(f"  内容长度: {len(doc.text)} 字符")
    print(f"  前200字符: {doc.text[:200]}")

# 使用Markdown节点解析器进一步分块
md_parser = MarkdownNodeParser()
md_nodes = md_parser.get_nodes_from_documents(documents)

# 对超长节点进行二次分块
sentence_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
final_nodes = []
for node in md_nodes:
    if len(node.text) > 512:
        sub_nodes = sentence_splitter.get_nodes_from_documents([node])
        final_nodes.extend(sub_nodes)
    else:
        final_nodes.append(node)

print(f"\n最终分块数: {len(final_nodes)}")
for i, node in enumerate(final_nodes[:5]):
    print(f"  Node {i}: [{node.metadata.get('Header_1', 'N/A')}] {node.text[:100]}...")

2.2 多源数据摄取管道

以下是一个支持多种数据源的完整摄取管道实现：

from llama_index.core import Document, VectorStoreIndex
from llama_index.core.node_parser import SentenceSplitter, MarkdownNodeParser
from llama_index.core.extractors import (
    TitleExtractor,
    SummaryExtractor,
    KeywordExtractor,
    QuestionsAnsweredExtractor,
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.readers.file import PyMuPDFReader, UnstructuredReader
from llama_index.readers.github import GithubRepoReader
from llama_index.core.vector_stores import SimpleVectorStore
from typing import Optional
import asyncio
from pathlib import Path
import hashlib

class DataIngestionPipeline:
    """多源数据摄取管道，支持PDF、Markdown、GitHub仓库等。"""

    def __init__(self, persist_dir: str = "./storage"):
        self.persist_dir = Path(persist_dir)
        self.persist_dir.mkdir(parents=True, exist_ok=True)
        self.vector_store = SimpleVectorStore()
        self.splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)

        # 元数据提取器
        self.extractors = [
            TitleExtractor(nodes=5),
            SummaryExtractor(summaries=["self", "prev"]),
            KeywordExtractor(keywords=5),
            QuestionsAnsweredExtractor(questions=3),
        ]

    async def ingest_pdf(self, file_path: str) -> list[Document]:
        """摄取PDF文档。"""
        reader = PyMuPDFReader()
        docs = reader.load(file_path)
        for doc in docs:
            doc.metadata.update({
                "source": file_path,
                "file_type": "pdf",
                "file_name": Path(file_path).name,
            })
        return docs

    async def ingest_markdown(self, file_path: str) -> list[Document]:
        """摄取Markdown文档。"""
        text = Path(file_path).read_text(encoding="utf-8")
        doc = Document(
            text=text,
            metadata={
                "source": file_path,
                "file_type": "markdown",
                "file_name": Path(file_path).name,
            }
        )
        return [doc]

    async def ingest_github_repo(
        self,
        repo: str,
        branch: str = "main",
        filter_dirs: Optional[list[str]] = None,
    ) -> list[Document]:
        """摄取GitHub仓库代码。"""
        reader = GithubRepoReader(
            repo=repo,
            branch=branch,
            filter_directories=(filter_dirs or ["src", "docs"]),
            filter_file_extensions=[".py", ".md", ".txt", ".yaml"],
            verbose=True,
        )
        docs = await reader.aload_data(branch=branch)
        for doc in docs:
            doc.metadata.update({
                "source": f"github:{repo}",
                "file_type": "code",
            })
        return docs

    async def ingest_directory(self, dir_path: str) -> list[Document]:
        """批量摄取目录中的所有支持文件。"""
        all_docs = []
        dir_path = Path(dir_path)

        for file_path in dir_path.rglob("*"):
            if not file_path.is_file():
                continue
            if file_path.suffix.lower() == ".pdf":
                all_docs.extend(await self.ingest_pdf(str(file_path)))
            elif file_path.suffix.lower() == ".md":
                all_docs.extend(await self.ingest_markdown(str(file_path)))
            elif file_path.suffix.lower() in (".py", ".ts", ".go", ".rs"):
                text = file_path.read_text(encoding="utf-8", errors="ignore")
                all_docs.append(Document(
                    text=text,
                    metadata={
                        "source": str(file_path),
                        "file_type": "code",
                        "file_name": file_path.name,
                        "language": file_path.suffix[1:],
                    }
                ))

        return all_docs

    async def process_documents(self, documents: list[Document]) -> list:
        """运行摄取管道：分块 -> 元数据提取 -> 向量化。"""
        pipeline = IngestionPipeline(
            transformations=[
                self.splitter,
                *self.extractors,
            ],
            vector_store=self.vector_store,
        )

        nodes = await pipeline.arun(documents=documents)

        # 为每个节点添加内容哈希（用于去重和增量更新）
        for node in nodes:
            content_hash = hashlib.md5(
                node.text.encode()
            ).hexdigest()
            node.metadata["content_hash"] = content_hash

        return nodes

    async def build_index(self, nodes: list):
        """从处理后的节点构建索引。"""
        index = VectorStoreIndex(nodes, vector_store=self.vector_store)
        index.storage_context.persist(str(self.persist_dir))
        return index

    async def load_existing_index(self):
        """加载已有索引。"""
        from llama_index.core import StorageContext, load_index_from_storage
        storage_context = StorageContext.from_defaults(persist_dir=str(self.persist_dir))
        return load_index_from_storage(storage_context)


# 使用示例
async def main():
    pipeline = DataIngestionPipeline(persist_dir="./storage")

    # 摄取多种来源
    pdf_docs = await pipeline.ingest_pdf("./docs/api-spec.pdf")
    md_docs = await pipeline.ingest_markdown("./docs/architecture.md")
    code_docs = await pipeline.ingest_directory("./src")

    all_docs = pdf_docs + md_docs + code_docs
    print(f"总计摄取 {len(all_docs)} 个文档")

    # 处理文档
    nodes = await pipeline.process_documents(all_docs)
    print(f"生成 {len(nodes)} 个节点")

    # 检查元数据提取结果
    for node in nodes[:3]:
        print(f"\n节点: {node.text[:80]}...")
        print(f"  标题: {node.metadata.get('document_title', 'N/A')}")
        print(f"  摘要: {node.metadata.get('section_summary', 'N/A')[:100]}")
        print(f"  关键词: {node.metadata.get('excerpt_keywords', 'N/A')}")
        print(f"  问题: {node.metadata.get('questions_this_excerpt_can_answer', 'N/A')}")

    # 构建索引
    index = await pipeline.build_index(nodes)
    print(f"\n索引构建完成，存储在 ./storage")

asyncio.run(main())

三、高级检索技术

3.1 句子窗口检索

句子窗口检索的核心思想是在检索时使用单个句子匹配，但在生成答案时使用该句子周围的上下文窗口。这种技术可以提升检索精确度的同时保持答案的上下文完整性：

from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.postprocessor import MetadataReplacementPostProcessor
from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI

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

# 句子级分块
sentence_splitter = SentenceSplitter(
    chunk_size=128,       # 小块用于精确检索
    chunk_overlap=0,
)

# 构建索引
index = VectorStoreIndex(nodes, transformations=[sentence_splitter])

# 配置句子窗口后处理器
# 检索时使用小句子块，但返回时替换为周围窗口
query_engine = index.as_query_engine(
    similarity_top_k=5,
    node_postprocessors=[
        MetadataReplacementPostProcessor(
            target_metadata_key="window",  # 使用window字段替换
        )
    ],
)

# 为了使用句子窗口，需要在分块时保存窗口上下文
from llama_index.core.node_parser import SentenceWindowNodeParser

window_parser = SentenceWindowNodeParser.from_defaults(
    window_size=3,           # 窗口大小（前后各3个句子）
    window_metadata_key="window",
    original_text_metadata_key="original_sentence",
)

# 使用窗口解析器重新构建
window_nodes = window_parser.get_nodes_from_documents(documents)
window_index = VectorStoreIndex(window_nodes)

window_query_engine = window_index.as_query_engine(
    similarity_top_k=5,
    node_postprocessors=[
        MetadataReplacementPostProcessor(
            target_metadata_key="window",
        )
    ],
)

response = window_query_engine.query("FastAPI如何处理异步数据库连接？")
print(response.response)

3.2 自动合并检索

自动合并检索会根据检索结果在文档结构中的位置关系，自动合并相邻的块。如果多个子块被检索到，它们会被合并为父块，从而提供更完整的上下文：

from llama_index.core.node_parser import HierarchicalNodeParser, get_leaf_nodes
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage
from llama_index.core.query_engine import RetrieverQueryEngine

# 层级分块：大块 -> 中块 -> 小块
hierarchical_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128],  # 三级分块
)

# 生成层级节点
hierarchical_nodes = hierarchical_parser.get_nodes_from_documents(documents)
leaf_nodes = get_leaf_nodes(hierarchical_nodes)

# 设置docstore存储层级关系
docstore = SimpleDocumentStore()
docstore.add_documents(hierarchical_nodes)
storage_context = StorageContext.from_defaults(docstore=docstore)

# 只对叶子节点建立向量索引
leaf_index = VectorStoreIndex(leaf_nodes, storage_context=storage_context)

# 创建自动合并检索器
base_retriever = leaf_index.as_retriever(similarity_top_k=6)
auto_merge_retriever = AutoMergingRetriever(
    base_retriever,
    storage_context,
    simple_ratio_thresh=0.4,  # 当子块比例超过40%时合并为父块
)

# 创建查询引擎
query_engine = RetrieverQueryEngine.from_args(auto_merge_retriever)

response = query_engine.query("请总结这篇文档的主要内容")
print(response.response)
print(f"\n检索到的节点数: {len(response.source_nodes)}")

3.3 递归检索

递归检索首先在一个摘要索引中检索，找到相关文档后，再在该文档的详细节点中进行二次检索：

from llama_index.core import SummaryIndex, VectorStoreIndex
from llama_index.core.retrievers import RecursiveRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.response_synthesizers import ResponseMode

# 为每个文档创建摘要和详细索引
all_doc_summaries = []
all_doc_nodes_dict = {}  # 文档ID -> 详细节点列表

for doc in documents:
    doc_id = doc.metadata.get("file_name", str(id(doc)))

    # 创建摘要
    summary_llm = OpenAI(model="gpt-4o-mini")
    summary = summary_llm.complete(f"请用一段话总结以下文档的内容：\n{doc.text[:3000]}")
    summary_node = Document(text=summary.text, metadata={"doc_id": doc_id, **doc.metadata})

    all_doc_summaries.append(summary_node)

    # 创建详细节点
    detail_splitter = SentenceSplitter(chunk_size=512, chunk_overlap=50)
    detail_nodes = detail_splitter.get_nodes_from_documents([doc])
    for node in detail_nodes:
        node.metadata["doc_id"] = doc_id
        all_doc_nodes_dict[node.id_] = node

# 建立摘要索引
summary_index = VectorStoreIndex(all_doc_summaries)
summary_retriever = summary_index.as_retriever(similarity_top_k=2)

# 创建递归检索器
recursive_retriever = RecursiveRetriever(
    "vector",
    retriever_dict={"vector": summary_retriever},
    node_dict=all_doc_nodes_dict,
    verbose=True,
)

query_engine = RetrieverQueryEngine.from_args(
    recursive_retriever,
    response_mode=ResponseMode.COMPACT,
)

response = query_engine.query("哪个文档讨论了数据库连接池的配置？")
print(response.response)

四、Agentic RAG系统

4.1 使用LlamaIndex Agent

Agentic RAG将Agent的决策能力与RAG的检索能力结合，使系统能够根据查询动态选择检索策略、多轮检索、以及组合多个知识源。以下是一个完整的Agentic RAG实现：

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata, FunctionTool
from llama_index.core import VectorStoreIndex, SummaryIndex
from llama_index.llms.openai import OpenAI
from llama_index.core.query_engine import SubQuestionQueryEngine

# 假设我们有多个不同类型的数据索引
code_index = VectorStoreIndex(code_nodes)      # 代码文档
api_index = VectorStoreIndex(api_nodes)        # API文档
architecture_index = VectorStoreIndex(arch_nodes)  # 架构文档
faq_index = SummaryIndex(faq_nodes)            # FAQ（用摘要索引更好）

# 为每个索引创建查询引擎
code_engine = code_index.as_query_engine(similarity_top_k=3)
api_engine = api_index.as_query_engine(similarity_top_k=3)
arch_engine = architecture_index.as_query_engine(similarity_top_k=3)
faq_engine = faq_index.as_query_engine(response_mode="tree_summarize")

# 创建工具
tools = [
    QueryEngineTool(
        query_engine=code_engine,
        metadata=ToolMetadata(
            name="code_search",
            description="搜索代码示例和实现细节。适用于查找函数实现、类定义、代码片段等。",
        ),
    ),
    QueryEngineTool(
        query_engine=api_engine,
        metadata=ToolMetadata(
            name="api_search",
            description="搜索API文档。适用于查找端点定义、请求/响应格式、认证方式等。",
        ),
    ),
    QueryEngineTool(
        query_engine=arch_engine,
        metadata=ToolMetadata(
            name="architecture_search",
            description="搜索架构文档。适用于查找系统设计、技术选型、数据流等。",
        ),
    ),
    QueryEngineTool(
        query_engine=faq_engine,
        metadata=ToolMetadata(
            name="faq_search",
            description="搜索常见问题。适用于查找已知问题的解答。",
        ),
    ),
]

# 添加自定义工具
@FunctionTool.from_defaults
def search_by_date(start_date: str, end_date: str, topic: str) -> str:
    """按日期范围搜索文档。
    
    Args:
        start_date: 开始日期 YYYY-MM-DD
        end_date: 结束日期 YYYY-MM-DD
        topic: 搜索主题
    """
    # 实现日期过滤检索逻辑
    filtered_nodes = [
        n for n in all_nodes
        if start_date <= n.metadata.get("date", "1970-01-01") <= end_date
        and topic.lower() in n.text.lower()
    ]
    if not filtered_nodes:
        return "未找到匹配的文档"
    return "\n".join([n.text[:200] for n in filtered_nodes[:5]])

tools.append(search_by_date)

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

# 使用Agent
response = agent.chat("如何实现用户认证？请从代码示例、API文档和架构设计三个方面回答")
print(response.response)

4.2 子问题查询引擎

对于复杂查询，SubQuestionQueryEngine会将查询分解为多个子问题，分别使用不同的数据源回答：

from llama_index.core.query_engine import SubQuestionQueryEngine

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

# 复杂查询会被自动分解
response = sub_question_engine.query(
    "比较FastAPI和Django在异步处理方面的差异，并给出各自的适用场景"
)

# 查看分解的子问题和各引擎的回答
for sub_q in response.sub_q_responses:
    print(f"子问题: {sub_q.sub_q}")
    print(f"使用引擎: {sub_q.engine_name}")
    print(f"回答: {sub_q.response[:200]}")
    print()

print(f"综合答案: {response.response}")

五、聊天引擎与对话系统

5.1 上下文感知聊天

LlamaIndex的ChatEngine支持多轮对话，并能根据对话历史调整检索策略：

from llama_index.core import VectorStoreIndex
from llama_index.core.chat_engine import CondensePlusContextChatEngine
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.llms.openai import OpenAI

index = VectorStoreIndex(nodes)
retriever = index.as_retriever(similarity_top_k=3)

memory = ChatMemoryBuffer.from_defaults(token_limit=3900)

chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=retriever,
    memory=memory,
    llm=OpenAI(model="gpt-4o", temperature=0.7),
    system_prompt="""你是一个技术文档助手。基于提供的文档上下文回答问题。
    如果上下文中没有相关信息，请说明"文档中未找到相关信息"。
    每次回答后，建议一个相关的后续问题。""",
    verbose=True,
)

# 多轮对话
messages = [
    "什么是游标分页？",
    "它和偏移分页有什么区别？",  # 上下文引用上一轮的游标分页
    "给我一个Python实现的例子",  # 基于前两轮的上下文
]

for msg in messages:
    print(f"\n用户: {msg}")
    response = chat_engine.chat(msg)
    print(f"助手: {response.response}")
    print(f"  [引用: {len(response.source_nodes)} 个来源]")

5.2 带知识图谱的聊天

LlamaIndex支持基于知识图谱的检索，能理解实体之间的关系：

from llama_index.core import KnowledgeGraphIndex
from llama_index.core.graph_stores import SimpleGraphStore
from llama_index.core import StorageContext

# 创建知识图谱索引
graph_store = SimpleGraphStore()
storage_context = StorageContext.from_defaults(graph_store=graph_store)

kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    storage_context=storage_context,
    max_triplets_per_chunk=3,
    include_embeddings=True,
    embed_model=OpenAIEmbedding(model="text-embedding-3-large"),
)

# 查询（支持关系推理）
kg_query_engine = kg_index.as_query_engine(
    include_embeddings=True,
    retriever_mode="hybrid",  # 混合关键词和向量检索
)

response = kg_query_engine.query("FastAPI和asyncpg之间是什么关系？")
print(response.response)

# 导出知识图谱可视化
graph_data = graph_store.to_dict()
print(f"知识图谱包含 {len(graph_data.get('graph', {}))} 个关系")

六、评估框架

6.1 RAG评估

LlamaIndex提供完整的RAG评估工具，包括检索质量和生成质量评估：

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    CorrectnessEvaluator,
    GuidelineEvaluator,
)
from llama_index.llms.openai import OpenAI

# 创建评估器
eval_llm = OpenAI(model="gpt-4o", temperature=0)

faithfulness = FaithfulnessEvaluator(llm=eval_llm)  # 忠实度：答案是否基于检索内容
relevancy = RelevancyEvaluator(llm=eval_llm)        # 相关性：答案是否与问题相关
correctness = CorrectnessEvaluator(llm=eval_llm)    # 正确性：答案是否与参考答案一致

# 评估查询结果
query = "如何优化FastAPI的数据库查询？"
reference_answer = "使用asyncpg异步驱动、连接池、查询缓存和批量查询来优化"

response = query_engine.query(query)

# 评估忠实度（答案是否忠于检索的上下文）
faith_result = faithfulness.evaluate_response(response=response)
print(f"忠实度: {'通过' if faith_result.passing else '失败'} - {faith_result.feedback}")

# 评估相关性（答案是否与问题相关）
rel_result = relevancy.evaluate_response(
    query=query,
    response=response,
)
print(f"相关性: {'通过' if rel_result.passing else '失败'} - {rel_result.feedback}")

# 评估正确性（与参考答案比较）
correct_result = correctness.evaluate(
    query=query,
    response=response.response,
    reference=reference_answer,
)
print(f"正确性: {'通过' if correct_result.passing else '失败'} - {correct_result.feedback}")

6.2 批量评估管道

以下是一个批量评估RAG系统质量的完整管道：

from llama_index.core.evaluation import BatchEvalRunner
import asyncio

# 定义评估数据集
eval_dataset = [
    {"query": "如何实现游标分页？", "reference": "使用WHERE + ORDER BY + LIMIT + cursor参数"},
    {"query": "asyncpg的优势是什么？", "reference": "高性能异步PostgreSQL驱动，支持连接池"},
    {"query": "如何配置Redis缓存？", "reference": "使用redis-py，设置TTL，序列化JSON响应"},
    {"query": "FastAPI中间件如何工作？", "reference": "ASGI中间件，在请求前后执行逻辑"},
    {"query": "什么是Pydantic模型验证？", "reference": "基于Python类型注解的运行时数据验证"},
]

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

async def evaluate_batch():
    results = await runner.aevaluate_queries(
        query_engine,
        queries=[d["query"] for d in eval_dataset],
        references=[d["reference"] for d in eval_dataset],
    )
    return results

results = asyncio.run(evaluate_batch())

# 汇总结果
for metric, eval_results in results.items():
    scores = [1 if r.passing else 0 for r in eval_results]
    avg = sum(scores) / len(scores)
    print(f"{metric}: {avg:.1%} ({sum(scores)}/{len(scores)})")

七、生产部署与优化

7.1 使用外部向量数据库

生产环境中应使用外部向量数据库替代默认的SimpleVectorStore：

from llama_index.vector_stores.qdrant import QdrantVectorStore
from llama_index.core import VectorStoreIndex, StorageContext
from qdrant_client import QdrantClient

# 连接Qdrant向量数据库
client = QdrantClient(
    host=os.environ.get("QDRANT_HOST", "localhost"),
    port=int(os.environ.get("QDRANT_PORT", 6333)),
    api_key=os.environ.get("QDRANT_API_KEY"),
)

vector_store = QdrantVectorStore(
    client=client,
    collection_name="documents",
    batch_size=100,
)

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(nodes, storage_context=storage_context)

7.2 缓存策略

LlamaIndex支持多级缓存以降低成本和延迟：

from llama_index.core import set_global_handler
from llama_index.core.cache import Cache
from llama_index.core.storage.kvstore import SimpleKVStore
from llama_index.core import Settings

# 启用LangSmith追踪
set_global_handler("langsmith")

# 配置缓存
from llama_index.core.embeddings import CacheEmbedding
from llama_index.core.llms import CacheLLM

# 嵌入缓存
cache_store = SimpleKVStore()
Settings.embed_model = CacheEmbedding(
    OpenAIEmbedding(model="text-embedding-3-large"),
    cache=cache_store,
)

# LLM缓存
Settings.llm = CacheLLM(
    OpenAI(model="gpt-4o"),
    cache=cache_store,
)

7.3 流式输出

# 流式输出查询结果
response = query_engine.query("总结文档的主要内容")
for chunk in response.response_gen:
    print(chunk, end="", flush=True)

# 聊天引擎流式
response = chat_engine.stream_chat("如何配置连接池？")
for token in response.response_gen:
    print(token, end="", flush=True)

总结

LlamaIndex作为专注于数据-LLM连接的框架，在RAG系统构建方面提供了最完整的工具链。本文系统性地覆盖了数据摄取管道（含LlamaParse解析和多源摄取）、高级检索技术（句子窗口检索、自动合并检索、递归检索）、Agentic RAG系统（含子问题分解和多数据源路由）、聊天引擎、知识图谱检索以及评估框架等核心内容。关键要点包括：选择正确的检索技术对RAG质量至关重要——句子窗口检索适合精确匹配场景，自动合并检索适合长文档，递归检索适合多文档场景；元数据提取器能显著提升检索质量；Agentic RAG通过动态工具选择处理复杂查询；评估框架是保证RAG系统可靠性的必要工具。随着企业对LLM应用需求的增长，基于LlamaIndex构建的RAG系统将成为知识管理和智能问答场景的核心基础设施。