LlamaIndex知识库应用开发完全教程

本文系统讲解LlamaIndex（原GPT Index）框架的核心架构、数据处理、索引构建、查询引擎、RAG管线、Agent集成与生产部署，帮助开发者从零构建企业级知识库应用。

目录

1. LlamaIndex架构概览
2. 数据连接器实战
3. 文档分块策略
4. 向量索引与树索引
5. 查询引擎深入
6. RAG管线构建
7. Agent集成
8. 知识图谱集成
9. 评估框架
10. 生产部署

1. LlamaIndex架构概览

LlamaIndex是一个以数据索引和查询为核心的大语言模型应用框架，专为RAG（Retrieval-Augmented Generation）场景设计。其核心理念是：将私有数据结构化后交给LLM高效检索和推理。

1.1 核心模块

LlamaIndex的架构分为五大层次：

1.2 安装与环境配置

# 基础安装
pip install llama-index

# 安装常用扩展
pip install llama-index-readers-web
pip install llama-index-readers-database
pip install llama-index-vector-stores-chroma
pip install llama-index-embeddings-openai
pip install llama-index-llms-openai

设置API密钥（以OpenAI为例）：

import os
os.environ["OPENAI_API_KEY"] = "sk-your-key"

# 或在代码中直接指定
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

llm = OpenAI(model="gpt-4o", api_key="sk-your-key")
embed_model = OpenAIEmbedding(model="text-embedding-3-small")

1.3 最小可运行示例

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader

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

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

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

# 4. 提问
response = query_engine.query("公司的核心业务是什么？")
print(response)

这五行代码就完成了一个完整的RAG应用。接下来我们逐层深入。

2. 数据连接器实战

LlamaIndex拥有丰富的数据连接器生态，支持从PDF、网页、数据库、Notion、Slack等数十种来源加载数据。

2.1 读取PDF文件

from llama_index.readers.file import PDFReader

reader = PDFReader()
documents = reader.load_data(file="./reports/annual_report.pdf")

# 查看文档结构
print(f"文档数量: {len(documents)}")
print(f"第一个文档长度: {len(documents[0].text)} 字符")
print(f"元数据: {documents[0].metadata}")

对于扫描版PDF，需要配合OCR：

from llama_index.readers.file import PDFReader
import pytesseract
from pdf2image import convert_from_path

# 将扫描PDF转为图片后OCR
images = convert_from_path("./scanned_doc.pdf")
text_content = ""
for img in images:
    text_content += pytesseract.image_to_string(img, lang="chi_sim+eng")

from llama_index.core import Document
doc = Document(text=text_content, metadata={"source": "scanned_doc.pdf"})

2.2 读取网页内容

from llama_index.readers.web import SimpleWebPageReader

urls = [
    "https://example.com/about",
    "https://example.com/products",
    "https://example.com/docs/api-reference"
]

reader = SimpleWebPageReader()
documents = reader.load_data(urls=urls)

for doc in documents:
    print(f"URL: {doc.metadata.get('url', 'N/A')}")
    print(f"内容长度: {len(doc.text)} 字符\n")

对于需要JavaScript渲染的动态网页：

from llama_index.readers.web import BeautifulSoupWebReader

# BeautifulSoupReader 支持更灵活的解析
reader = BeautifulSoupWebReader()
documents = reader.load_data(
    urls=["https://example.com/spa-page"],
    # 自定义解析规则
    custom_parsing_fn=lambda soup: soup.find("main").get_text()
)

2.3 读取数据库

from llama_index.readers.database import DatabaseReader
import sqlalchemy

# 连接数据库
engine = sqlalchemy.create_engine(
    "postgresql://user:password@localhost:5432/mydb"
)

reader = DatabaseReader(engine=engine)

# 通过SQL查询加载数据
documents = reader.load_data(
    query="SELECT id, title, content, created_at FROM articles WHERE status = 'published'"
)

# 每行数据会被序列化为一个Document
for doc in documents[:3]:
    print(doc.text)
    print(f"元数据: {doc.metadata}\n")

2.4 批量加载与增量更新

from llama_index.core import SimpleDirectoryReader
from pathlib import Path

def load_with_metadata(directory: str) -> list:
    """带元数据增强的批量加载"""
    reader = SimpleDirectoryReader(
        input_dir=directory,
        recursive=True,
        required_exts=[".pdf", ".txt", ".md"],
        filename_as_id=True
    )
    documents = reader.load_data()

    # 为每个文档添加自定义元数据
    for doc in documents:
        source_path = doc.metadata.get("file_path", "")
        doc.metadata["department"] = _infer_department(source_path)
        doc.metadata["doc_type"] = Path(source_path).suffix
        doc.metadata["indexed_at"] = datetime.now().isoformat()

    return documents

3. 文档分块策略

文档分块（Chunking）是RAG质量的关键环节。块太大导致检索噪音多，块太小导致上下文丢失。

3.1 SentenceSplitter——最常用的分块器

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=512,        # 每个块的目标token数
    chunk_overlap=50,      # 块之间的重叠token数
    separator=" ",         # 分词分隔符
)

nodes = splitter.get_nodes_from_documents(documents)
print(f"生成了 {len(nodes)} 个节点")
print(f"示例节点:\n{nodes[0].text[:200]}...")

3.2 SemanticSplitterNodeParser——语义分块

基于嵌入相似度进行分块，在语义边界处切分，保证每个块的内容语义连贯：

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

embed_model = OpenAIEmbedding(model="text-embedding-3-small")

splitter = SemanticSplitterNodeParser(
    buffer_size=1,                # 连接句子的缓冲区大小
    breakpoint_percentile_threshold=95,  # 语义断点阈值
    embed_model=embed_model,
)

nodes = splitter.get_nodes_from_documents(documents)

3.3 MarkdownNodeParser——结构化文档分块

对Markdown文档按标题层级智能分块，保留文档结构：

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
nodes = parser.get_nodes_from_documents(markdown_documents)

# 每个节点会保留其在文档层级中的位置信息
for node in nodes[:5]:
    print(f"标题层级: {node.metadata.get('header_hierarchy', 'N/A')}")
    print(f"内容: {node.text[:100]}...\n")

3.4 分块策略对比

3.5 最佳实践

# 组合使用：先按结构分块，再微调大小
from llama_index.core.node_parser import (
    MarkdownNodeParser,
    SentenceSplitter,
    HierarchicalNodeParser
)

# 层级分块——为不同查询粒度提供不同大小的节点
parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]  # 三级粒度
)

nodes = parser.get_nodes_from_documents(documents)
# 自动建立父子关系，支持递归检索

4. 向量索引与树索引

4.1 VectorStoreIndex——向量索引

向量索引是最常用的索引类型，通过嵌入模型将文本转为向量，支持语义相似度检索：

from llama_index.core import VectorStoreIndex, Settings
from llama_index.embeddings.openai import OpenAIEmbedding

# 全局配置
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
Settings.chunk_overlap = 50

# 构建索引
index = VectorStoreIndex.from_documents(
    documents,
    show_progress=True
)

# 保存索引到磁盘
index.storage_context.persist(persist_dir="./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)

4.2 配置外部向量数据库

生产环境中，建议使用专业向量数据库：

# Chroma
import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore

chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("my_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)

from llama_index.core import StorageContext
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context
)

# Milvus
from llama_index.vector_stores.milvus import MilvusVectorStore
vector_store = MilvusVectorStore(
    uri="http://localhost:19530",
    collection_name="my_docs",
    dim=1536
)

4.3 SummaryIndex——摘要索引

摘要索引将所有节点串联存储，适合需要遍历全部文档的场景：

from llama_index.core import SummaryIndex

summary_index = SummaryIndex.from_documents(documents)

# 摄要查询——会遍历所有相关节点
summary_engine = summary_index.as_query_engine(
    response_mode="tree_summarize"  # 逐层汇总
)
response = summary_engine.query("请总结这份文档的核心要点")

4.4 TreeIndex——树索引

树索引通过层级结构组织节点，适合多级摘要和主题分类：

from llama_index.core import TreeIndex

tree_index = TreeIndex.from_documents(documents)

# 从树根开始逐层查询
tree_engine = tree_index.as_query_engine(
    child_branch_factor=2  # 每个节点探索的子分支数
)
response = tree_engine.query("文档中关于产品策略的内容有哪些？")

4.5 索引选型指南

5. 查询引擎深入

查询引擎是连接检索和生成的桥梁。LlamaIndex提供了多种查询引擎模式。

5.1 RetrieverQueryEngine——基础查询

from llama_index.core import VectorStoreIndex
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.postprocessor import SimilarityPostprocessor

# 自定义检索器
retriever = index.as_retriever(
    similarity_top_k=5,          # 返回前5个最相关的节点
    filters=None                 # 可添加元数据过滤
)

# 添加后处理器——过滤低质量结果
postprocessor = SimilarityPostprocessor(
    similarity_cutoff=0.7  # 相似度低于0.7的结果被过滤
)

query_engine = RetrieverQueryEngine(
    retriever=retriever,
    node_postprocessors=[postprocessor]
)

response = query_engine.query("如何申请退款？")
print(response)
print(f"\n参考来源: {[n.metadata.get('file_name') for n in response.source_nodes]}")

5.2 SubQuestionQueryEngine——子问题分解

对复杂问题自动拆分为多个子问题，分别查询后汇总：

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

# 为不同数据源创建工具
tools = [
    QueryEngineTool(
        query_engine=product_index.as_query_engine(),
        metadata=ToolMetadata(
            name="product_docs",
            description="包含产品功能、规格和技术文档"
        )
    ),
    QueryEngineTool(
        query_engine=faq_index.as_query_engine(),
        metadata=ToolMetadata(
            name="faq_docs",
            description="包含常见问题解答和客服指南"
        )
    ),
    QueryEngineTool(
        query_engine=finance_index.as_query_engine(),
        metadata=ToolMetadata(
            name="finance_docs",
            description="包含财务报告、定价和退款政策"
        )
    ),
]

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

# 复杂问题会被自动分解
response = query_engine.query(
    "产品A的退款政策是什么？最近一个季度该产品的营收表现如何？"
)
# 自动分解为：
# [product_docs] 产品A的功能和规格
# [finance_docs] 退款政策
# [finance_docs] 最近季度营收

5.3 SQL查询引擎

直接用自然语言查询SQL数据库：

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

engine = create_engine("sqlite:///business.db")
sql_database = SQLDatabase(engine)

query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    tables=["orders", "customers", "products"],
    synthesize_response=True  # 用LLM将SQL结果转为自然语言
)

response = query_engine.query("上个月销售额最高的前10个客户是谁？")
print(response)
# 背景执行: SELECT c.name, SUM(o.amount) as total FROM orders o
#           JOIN customers c ON o.customer_id = c.id
#           WHERE o.date >= '2024-03-01'
#           GROUP BY c.id ORDER BY total DESC LIMIT 10

5.4 自定义查询引擎

from llama_index.core.query_engine import CustomQueryEngine
from llama_index.core.retrievers import BaseRetriever
from llama_index.core import get_response_synthesizer
from llama_index.core import PromptTemplate

class RAGQueryEngine(CustomQueryEngine):
    """自定义RAG查询引擎"""
    retriever: BaseRetriever
    response_synthesizer: object

    def custom_query(self, query_str: str):
        # 1. 检索相关节点
        nodes = self.retriever.retrieve(query_str)

        # 2. 自定义后处理（如去重、重排序）
        unique_nodes = self._deduplicate(nodes)

        # 3. 合成回答
        response = self.response_synthesizer.synthesize(
            query=query_str,
            nodes=unique_nodes
        )
        return response

    def _deduplicate(self, nodes):
        seen = set()
        unique = []
        for node in nodes:
            if node.node_id not in seen:
                seen.add(node.node_id)
                unique.append(node)
        return unique

6. RAG管线构建

6.1 完整的生产级RAG管线

下面是一个包含所有关键组件的完整RAG管线：

from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    StorageContext,
    Settings,
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.postprocessor import (
    SimilarityPostprocessor,
    KeywordNodePostprocessor,
    SentenceEmbeddingOptimizer,
)
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

class ProductionRAGPipeline:
    """生产级RAG管线"""

    def __init__(self, config: dict):
        self.config = config
        self._setup_models()
        self._setup_vector_store()

    def _setup_models(self):
        Settings.llm = OpenAI(
            model=self.config.get("llm_model", "gpt-4o"),
            temperature=0.1,
            max_tokens=2048
        )
        Settings.embed_model = OpenAIEmbedding(
            model=self.config.get("embed_model", "text-embedding-3-small")
        )
        Settings.chunk_size = self.config.get("chunk_size", 512)
        Settings.chunk_overlap = self.config.get("chunk_overlap", 50)

    def _setup_vector_store(self):
        chroma_client = chromadb.PersistentClient(
            path=self.config.get("chroma_path", "./chroma_db")
        )
        collection = chroma_client.get_or_create_collection(
            name=self.config.get("collection_name", "knowledge_base"),
            metadata={"hnsw:space": "cosine"}
        )
        self.vector_store = ChromaVectorStore(chroma_collection=collection)

    def ingest(self, data_dir: str):
        """数据摄入管线"""
        # 加载文档
        reader = SimpleDirectoryReader(
            input_dir=data_dir,
            recursive=True,
            required_exts=[".pdf", ".txt", ".md", ".docx"]
        )
        documents = reader.load_data()

        # 分块
        splitter = SentenceSplitter(
            chunk_size=Settings.chunk_size,
            chunk_overlap=Settings.chunk_overlap
        )
        nodes = splitter.get_nodes_from_documents(documents)

        # 添加元数据
        for i, node in enumerate(nodes):
            node.metadata["chunk_index"] = i
            node.metadata["total_chunks"] = len(nodes)

        # 构建索引
        storage_context = StorageContext.from_defaults(
            vector_store=self.vector_store
        )
        index = VectorStoreIndex(
            nodes=nodes,
            storage_context=storage_context,
            show_progress=True
        )

        print(f"✅ 已索引 {len(nodes)} 个节点")
        return index

    def query(self, question: str, index: VectorStoreIndex):
        """查询管线"""
        retriever = index.as_retriever(
            similarity_top_k=self.config.get("top_k", 5)
        )

        # 多层后处理
        postprocessors = [
            SimilarityPostprocessor(similarity_cutoff=0.65),
            KeywordNodePostprocessor(
                required_keywords=self._extract_keywords(question)
            ),
        ]

        from llama_index.core.query_engine import RetrieverQueryEngine
        query_engine = RetrieverQueryEngine(
            retriever=retriever,
            node_postprocessors=postprocessors
        )

        response = query_engine.query(question)
        return {
            "answer": str(response),
            "sources": [
                {
                    "file": n.metadata.get("file_name", "unknown"),
                    "score": n.score,
                    "snippet": n.text[:200]
                }
                for n in response.source_nodes
            ]
        }

    def _extract_keywords(self, text: str) -> list:
        """简单关键词提取"""
        # 生产环境中可使用jieba等分词工具
        return [w for w in text.split() if len(w) > 1]


# 使用示例
pipeline = ProductionRAGPipeline(config={
    "llm_model": "gpt-4o",
    "embed_model": "text-embedding-3-small",
    "chunk_size": 512,
    "chroma_path": "./chroma_db",
    "collection_name": "company_docs",
    "top_k": 5
})

# 数据摄入
index = pipeline.ingest("./documents")

# 查询
result = pipeline.query("公司的年假政策是什么？", index)
print(f"回答: {result['answer']}")
print(f"参考来源: {result['sources']}")

6.2 混合检索（Hybrid Search）

结合关键词检索和语义检索，提高召回率：

from llama_index.core.retrievers import QueryFusionRetriever

# 融合多种检索策略
retriever = QueryFusionRetriever(
    retrievers=[
        index.as_retriever(similarity_top_k=5),  # 向量检索
        index.as_retriever(mode="keyword"),        # 关键词检索
    ],
    num_queries=3,           # 生成3个查询变体
    use_async=True,
    retriever_weights=[0.7, 0.3]  # 权重分配
)

7. Agent集成

LlamaIndex支持将查询引擎包装为Agent工具，实现自主决策和多步推理。

7.1 Function Calling Agent

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

# 将查询引擎包装为工具
tools = [
    QueryEngineTool(
        query_engine=product_index.as_query_engine(),
        metadata=ToolMetadata(
            name="product_search",
            description="搜索产品文档，回答关于产品功能、使用方法的问题"
        )
    ),
    QueryEngineTool(
        query_engine=hr_index.as_query_engine(),
        metadata=ToolMetadata(
            name="hr_search",
            description="搜索人力资源政策，回答关于假期、报销、入职等问题"
        )
    ),
]

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

# Agent会自主决定使用哪些工具
response = agent.chat("我想了解一下公司的年假政策，以及如何使用内部报销系统")
print(response)

7.2 AgentWorkflow——多Agent协作

from llama_index.core.agent.workflow import AgentWorkflow
from llama_index.core.workflow import Context

# 定义专业Agent
research_agent = FunctionCallingAgent.from_tools(
    tools=[search_tool, database_tool],
    llm=OpenAI(model="gpt-4o"),
    name="research_agent",
    description="负责信息检索和数据分析"
)

writing_agent = FunctionCallingAgent.from_tools(
    tools=[],
    llm=OpenAI(model="gpt-4o"),
    name="writing_agent",
    description="负责撰写报告和文档"
)

review_agent = FunctionCallingAgent.from_tools(
    tools=[],
    llm=OpenAI(model="gpt-4o"),
    name="review_agent",
    description="负责审核和校对内容"
)

# 编排多Agent工作流
workflow = AgentWorkflow(
    agents=[research_agent, writing_agent, review_agent],
    root_agent="research_agent",  # 入口Agent
)

# 执行复杂任务
async def run_workflow():
    ctx = Context(workflow)
    handler = workflow.run(
        user_msg="调研竞品分析并撰写一份报告",
        ctx=ctx
    )
    async for event in handler.stream_events():
        print(f"[{event.__class__.__name__}] {event}")
    result = await handler
    return result

8. 知识图谱集成

知识图谱索引可以抽取实体和关系，支持结构化查询：

from llama_index.core import KnowledgeGraphIndex, KnowledgeGraphStore
from llama_index.core.storage.storage_context import StorageContext

# 使用内置存储
kg_store = SimpleGraphStore()
storage_context = StorageContext.from_defaults(graph_store=kg_store)

# 构建知识图谱索引（自动抽取实体关系）
kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    storage_context=storage_context,
    max_triplets_per_chunk=5,    # 每个chunk最多抽取5个三元组
    include_embeddings=True,     # 同时包含向量嵌入
)

# 查询知识图谱
kg_query_engine = kg_index.as_query_engine(
    response_mode="tree_summarize",
    retriever_mode="keyword"  # 使用关键词匹配实体
)

response = kg_query_engine.query("张三在公司担任什么职位？")

# 可视化知识图谱
from llama_index.core.storage.graphstore import SimpleGraphStore
graph = kg_store.get_networkx_graph()
print(f"节点数: {len(graph.nodes)}")
print(f"边数: {len(graph.edges)}")

8.1 自定义实体抽取

from llama_index.core.indices.knowledge_graph.retrievers import (
    KGRetrieverMode
)

# 使用LLM进行更精确的实体抽取
kg_index = KnowledgeGraphIndex.from_documents(
    documents,
    kg_triple_extract_template="""
    从以下文本中抽取实体和关系三元组。
    格式: (主体, 关系, 客体)
    
    文本: {text}
    
    三元组列表:
    """,
    max_triplets_per_chunk=10,
)

9. 评估框架

9.1 使用LlamaIndex内置评估

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,    # 忠实度——回答是否基于检索到的上下文
    RelevancyEvaluator,       # 相关性——回答是否切题
    CorrectnessEvaluator,     # 正确性——回答是否正确
    SemanticSimilarityEvaluator,  # 语义相似度
)

# 忠实度评估
faithfulness_eval = FaithfulnessEvaluator()
eval_result = faithfulness_eval.evaluate(
    query="公司的退款政策是什么？",
    response="公司提供30天无理由退款...",
    contexts=[retrieved_node.text for node in response.source_nodes]
)
print(f"忠实度分数: {eval_result.score}")  # 0或1
print(f"反馈: {eval_result.feedback}")

# 相关性评估
relevancy_eval = RelevancyEvaluator()
eval_result = relevancy_eval.evaluate(
    query="公司的退款政策是什么？",
    response=response.response,
    contexts=[n.text for n in response.source_nodes]
)
print(f"相关性分数: {eval_result.score}")

9.2 批量评估

from llama_index.core.evaluation import BatchEvalRunner

# 准备测试数据
eval_questions = [
    "公司的核心业务是什么？",
    "如何申请年假？",
    "退款流程是怎样的？",
]
eval_answers = [
    "公司的核心业务是云计算服务...",
    "年假需要提前在OA系统申请...",
    "30天内可通过客服申请退款...",
]

# 批量评估
runner = BatchEvalRunner(
    evaluators={
        "faithfulness": FaithfulnessEvaluator(),
        "relevancy": RelevancyEvaluator(),
        "correctness": CorrectnessEvaluator(),
    },
    show_progress=True
)

eval_results = runner.evaluate_responses(
    queries=eval_questions,
    responses=[query_engine.query(q) for q in eval_questions],
    reference=eval_answers
)

# 汇总结果
for metric, results in eval_results.eval_results.items():
    scores = [r.score for r in results]
    print(f"{metric}: 平均分={sum(scores)/len(scores):.2f}")

10. 生产部署

10.1 FastAPI服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from llama_index.core import load_index_from_storage, StorageContext
import uvicorn

app = FastAPI(title="知识库问答API")

class QueryRequest(BaseModel):
    question: str
    top_k: int = 5

class QueryResponse(BaseModel):
    answer: str
    sources: list[dict]
    latency_ms: float

# 启动时加载索引
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)
query_engine = index.as_query_engine(similarity_top_k=5)

@app.post("/query", response_model=QueryResponse)
async def query_knowledge_base(req: QueryRequest):
    import time
    start = time.time()

    try:
        response = query_engine.query(req.question)
        latency = (time.time() - start) * 1000

        sources = [
            {
                "file": n.metadata.get("file_name", "unknown"),
                "score": round(n.score, 4),
                "snippet": n.text[:300]
            }
            for n in response.source_nodes[:3]
        ]

        return QueryResponse(
            answer=str(response),
            sources=sources,
            latency_ms=round(latency, 2)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/ingest")
async def ingest_documents(directory: str = "./new_docs"):
    """增量更新索引"""
    from llama_index.core import SimpleDirectoryReader, VectorStoreIndex
    reader = SimpleDirectoryReader(input_dir=directory, recursive=True)
    docs = reader.load_data()
    index.insert_nodes(docs)  # 增量插入
    return {"status": "success", "indexed": len(docs)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

10.2 缓存与性能优化

from llama_index.core.cache import CacheType
from llama_index.core import set_cache

# 启用查询缓存
set_cache(CacheType.MEMORY)

# 使用嵌入缓存减少重复计算
from llama_index.core.embeddings import MockEmbedding
# 生产中可用Redis缓存嵌入结果

# 异步查询提升并发
import asyncio

async def batch_query(questions: list[str]):
    tasks = [
        query_engine.aquery(q) for q in questions
    ]
    results = await asyncio.gather(*tasks)
    return results

10.3 监控与日志

import logging
from llama_index.core import set_global_handler

# 集成LangSmith进行追踪
set_global_handler("langsmith", project_name="knowledge-base")

# 或使用自定义日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rag_pipeline")

class MonitoredQueryEngine:
    def __init__(self, engine):
        self.engine = engine

    def query(self, question: str):
        logger.info(f"Query: {question}")
        response = self.engine.query(question)
        logger.info(f"Sources: {[n.metadata.get('file_name') for n in response.source_nodes]}")
        logger.info(f"Answer length: {len(str(response))}")
        return response

10.4 部署清单

总结

LlamaIndex提供了一套从数据摄入到生产部署的完整工具链。核心建议：

1. 分块是关键——投入足够时间调优分块策略，它对RAG质量的影响远超模型选择
2. 先简单后复杂——从VectorStoreIndex开始，按需添加子问题分解、知识图谱等高级功能
3. 评估驱动——用评估框架量化每个环节的效果，避免盲目调参
4. 生产优先——从一开始就考虑增量更新、缓存、监控等生产需求

LlamaIndex生态仍在快速发展，建议关注官方文档和GitHub获取最新特性。