LlamaIndex 核心组件：搭建 RAG 流水线

这篇适合会写一点 Python、想让模型读取自己资料的读者。读完你能说清 RAG 的五个阶段，用十几行代码完成加载、切块、向量化、存储和查询，并知道怎样给回答打分。

LlamaIndex 组件和 RAG 是什么

大模型的知识来自训练数据，但它未必见过你的资料，也不一定掌握最新的内容。检索增强生成（RAG，Retrieval-Augmented Generation）的做法是先从你自己的数据里找出相关内容，再把它交给模型。LlamaIndex 把这条流程拆成若干组件。本文重点讲 QueryEngine，因为它可以作为 RAG 工具交给智能体使用^[1]。

一条完整的 RAG 流程通常分五个阶段：加载、索引、存储、查询和评测^[1]。

第一步：加载文档

加载就是把数据从它所在的地方搬进工作流，这些地方可以是文本文件、PDF、网站、数据库或 API^[1]。LlamaIndex 有三种主要的加载方式：

• SimpleDirectoryReader：内置加载器，读取本地文件夹里的多种文件类型；
• LlamaParse：LlamaIndex 官方的 PDF 解析工具，以托管 API 的形式提供；
• LlamaHub：数百个数据加载库的注册表，可以接入各种数据源^[1]。

入门时用 SimpleDirectoryReader 最简单，它会把文件转换成 Document 对象^[1]：

from llama_index.core import SimpleDirectoryReader

reader = SimpleDirectoryReader(input_dir="path/to/directory")
documents = reader.load_data()

第二步：切块并向量化

Document 通常太长，不适合直接检索，需要切成更小的片段，LlamaIndex 叫它 Node。一个 Node 就是文档中的一段文字，同时保留对原 Document 的引用^[1]。

IngestionPipeline 可以用两个变换一次完成切块和向量化：SentenceSplitter 按自然的句子边界切分，HuggingFaceEmbedding 把每个片段转换成向量，也就是能表达语义的一组数字^[1]。下面的示例使用 BAAI/bge-small-en-v1.5 作为嵌入模型，chunk_overlap=0 表示相邻片段不重叠^[1]：

from llama_index.core import Document
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.ingestion import IngestionPipeline

# create the pipeline with transformations
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_overlap=0),
        HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"),
    ]
)

nodes = await pipeline.arun(documents=[Document.example()])

第三步：存入向量库并建立索引

向量需要有地方存放。示例使用 Chroma，先安装它的 LlamaIndex 集成^[1]：

pip install llama-index-vector-stores-chroma

然后把流水线和向量库连起来。chunk_size=25 让每个片段更短，运行流水线时向量会直接写入 Chroma，并保存在指定的目录里^[1]：

import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore

db = chromadb.PersistentClient(path="./alfred_chroma_db")
chroma_collection = db.get_or_create_collection("alfred")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=25, chunk_overlap=0),
        HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5"),
    ],
    vector_store=vector_store,
)

索引不只有向量一种。对于任何一种索引策略，都可以用多种方式查询，包括子查询、多步查询和混合策略，并结合大语言模型与 LlamaIndex 的数据结构来完成^[1]。除了向量，索引还可以借助多种元数据策略，依据文档的属性找到上下文相关的数据^[1]。

索引负责让查询找到相关片段。VectorStoreIndex 把问题和片段放进同一个向量空间，按相似度匹配。建索引时要用入库时同一个嵌入模型，两边的向量才能保持一致^[1]：

from llama_index.core import VectorStoreIndex
from llama_index.embeddings.huggingface import HuggingFaceEmbedding

embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en-v1.5")
index = VectorStoreIndex.from_vector_store(vector_store, embed_model=embed_model)

第四步：查询

索引可以转换成三种查询接口^[1]：

• as_retriever：返回带相似度分数的 NodeWithScore 列表，只取匹配的片段；
• as_query_engine：针对单个问题返回一段回答，智能体式的问答最常用它；
• as_chat_engine：多轮对话，保留对话历史。

使用 as_query_engine 时要传入一个大语言模型来写回答。下面的示例用 HuggingFaceInferenceAPI 调用 Qwen/Qwen2.5-Coder-32B-Instruct，并把 response_mode 设为 tree_summarize^[1]：

from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")
query_engine = index.as_query_engine(
    llm=llm,
    response_mode="tree_summarize",
)
query_engine.query("What is the meaning of life?")
# The meaning of life is 42

回答怎样整理，由 ResponseSynthesizer 决定，常用的有三种策略^[1]：

• refine：依次处理检索到的每个片段，逐步修正答案，每个片段调用一次模型，调用次数最多；
• compact：默认策略，先把片段拼接起来再处理，调用次数更少；
• tree_summarize：逐段生成答案，并组织成树状结构。

如果需要精细控制查询的每一步，可以使用低层组合 API，它能调整查询过程的每个环节，也能和 Workflows 搭配使用^[1]。

第五步：评测与可观测性

模型不一定每次都答对，所以要给回答打分。LlamaIndex 内置了三个基于大语言模型的评测器^[1]：

• FaithfulnessEvaluator：检查答案是否有上下文支持；
• AnswerRelevancyEvaluator：检查答案是否切题；
• CorrectnessEvaluator：检查答案是否正确。

下面的代码用 FaithfulnessEvaluator 判断一次查询的结果是否通过，结果保存在 eval_result.passing 中^[1]：

from llama_index.core.evaluation import FaithfulnessEvaluator

query_engine = # from the previous section
llm = # from the previous section

# query index
evaluator = FaithfulnessEvaluator(llm=llm)
response = query_engine.query(
    "What battles took place in New York City in the American Revolution?"
)
eval_result = evaluator.evaluate_response(response=response)
eval_result.passing

流水线变复杂以后，单次评测不够用，还需要观察每个组件的表现。LlamaIndex 可以接入基于 Arize Phoenix 的 LlamaTrace 回调，安装命令如下^[1]：

pip install -U llama-index-callbacks-arize-phoenix

使用前先在 LlamaTrace 注册账号，在账号设置里生成 API key，再把它设为 PHOENIX_API_KEY 环境变量^[1]。启用追踪的代码如下：

import llama_index
import os

PHOENIX_API_KEY = "<PHOENIX_API_KEY>"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"api_key={PHOENIX_API_KEY}"
llama_index.core.set_global_handler(
    "arize_phoenix",
    endpoint="https://llamatrace.com/v1/traces"
)

下一步

这些组件组合起来，就是一条可用的 RAG 流水线。接下来可以把 QueryEngine 包装成工具，交给智能体调用。想了解更多组件，可以阅读 LlamaIndex 官方文档中的组件指南和 RAG 指南^[1]。

本文改编自 Hugging Face 开源课程《Agents Course》第 2 单元关于 LlamaIndex 组件的内容^[1]，原文以 Apache-2.0 许可证发布。