LlamaIndex数据框架完全教程
一、概述与核心理念
1.1 什么是LlamaIndex
LlamaIndex（原名GPT Index）是由Jerry Liu于2022年11月创建的开源数据框架，专注于为大语言模型（LLM）提供高效的数据连接、索引和查询能力。如果说LangChain是LLM应用的"胶水框架"，那么LlamaIndex就是专注于"数据层"的专家框架——它解决的核心问题是：如何让LLM高效地访问和理解你的私有数据。
LlamaIndex的核心理念建立在"检索增强生成"（RAG）的基础之上，但它远不止是一个简单的RAG工具。它提供了从数据摄入、索引构建、查询优化到评估验证的完整工具链，是构建企业级知识密集型应用的理想选择。
1.2 核心设计理念
LlamaIndex的设计遵循以下几个核心原则：
- 数据优先：一切以数据的高效利用为中心，提供丰富的数据连接器和索引结构
- 模块化架构：每个组件都可以独立使用或替换，提供极大的灵活性
- 查询优化：不仅仅是简单的语义搜索，还支持多种高级查询策略
- 生产就绪：内置评估框架、可观测性和部署工具
- 渐进式复杂度：从几行代码的简单用例到复杂的自定义Pipeline，都能胜任
1.3 与其他框架的定位对比
| 特性 | LlamaIndex | LangChain |
|---|---|---|
| 核心定位 | 数据索引与检索 | LLM应用编排 |
| RAG支持 | 原生深度支持 | 通过组件组合实现 |
| 索引类型 | 丰富（向量、树、关键词、知识图谱等） | 主要依赖外部向量数据库 |
| 评估框架 | 内置完善的评估模块 | 通过LangSmith实现 |
| Agent能力 | 支持但非核心 | 核心能力 |
| 学习曲线 | 专注领域较易上手 | 概念较多，学习曲线较陡 |
1.4 安装与环境配置
# 安装LlamaIndex核心包
pip install llama-index
# 或者安装核心包和常用集成
pip install llama-index-core llama-index-llms-openai llama-index-embeddings-openai
# 安装文档解析器
pip install llama-index-readers-file llama-parse
# 安装向量存储集成
pip install llama-index-vector-stores-chroma llama-index-vector-stores-faiss
# 安装评估模块
pip install llama-index-evaluation
环境配置：
import os
# OpenAI配置
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# LlamaParse配置（可选，用于高级文档解析）
os.environ["LLAMA_CLOUD_API_KEY"] = "your-llama-cloud-api-key"
1.5 快速上手
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
# 从目录加载文档
documents = SimpleDirectoryReader("./data").load_data()
# 创建向量索引
index = VectorStoreIndex.from_documents(documents)
# 创建查询引擎
query_engine = index.as_query_engine()
# 查询
response = query_engine.query("这个文档的主要内容是什么？")
print(response)
仅需5行代码，就能构建一个完整的文档问答系统。这就是LlamaIndex的魅力所在。
二、数据连接器（Data Connectors）
2.1 数据连接器概述
数据连接器是LlamaIndex的数据摄入层，负责从各种来源加载数据并转换为统一的Document格式。LlamaIndex提供了超过160种数据连接器，覆盖了几乎所有常见的数据源类型。
2.2 文件系统连接器
from llama_index.core import SimpleDirectoryReader
# 加载目录下的所有文件
documents = SimpleDirectoryReader(
input_dir="./documents",
recursive=True, # 递归子目录
required_exts=[".pdf", ".txt", ".md"], # 文件类型过滤
filename_as_id=True # 使用文件名作为文档ID
).load_data()
print(f"加载了 {len(documents)} 个文档")
# 查看文档结构
for doc in documents[:3]:
print(f"文档ID: {doc.id_}")
print(f"内容长度: {len(doc.text)} 字符")
print(f"元数据: {doc.metadata}")
print("---")
2.3 数据库连接器
from llama_index.readers.database import DatabaseReader
# 连接SQL数据库
reader = DatabaseReader(
uri="postgresql://user:password@localhost:5432/mydb"
)
# 加载表数据
documents = reader.load_data(
queries=["SELECT * FROM products WHERE category = 'electronics'"]
)
2.4 API连接器
from llama_index.readers.web import SimpleWebPageReader
# 加载网页内容
documents = SimpleWebPageReader().load_data(
urls=[
"https://example.com/page1",
"https://example.com/page2"
]
)
# 使用BeautifulSoup进行更精细的网页解析
from llama_index.readers.web import BeautifulSoupWebReader
reader = BeautifulSoupWebReader()
documents = reader.load_data(
urls=["https://example.com"],
custom_parsing_fn=lambda soup: soup.find("article").get_text()
)
2.5 自定义数据连接器
from llama_index.core import Document
from llama_index.core.readers.base import BaseReader
from typing import List, Optional, Any
import json
class CustomAPIReader(BaseReader):
"""自定义API数据连接器示例"""
def __init__(self, api_base: str, api_key: str):
self.api_base = api_base
self.api_key = api_key
def load_data(self, **load_kwargs: Any) -> List[Document]:
"""从自定义API加载数据"""
import httpx
# 调用API获取数据
response = httpx.get(
f"{self.api_base}/articles",
headers={"Authorization": f"Bearer {self.api_key}"},
params=load_kwargs
)
articles = response.json()
# 转换为LlamaIndex Document格式
documents = []
for article in articles:
doc = Document(
text=article["content"],
metadata={
"title": article["title"],
"author": article["author"],
"date": article["published_at"],
"source": "custom_api"
},
id_=article["id"]
)
documents.append(doc)
return documents
# 使用自定义连接器
reader = CustomAPIReader(
api_base="https://api.example.com",
api_key="your-api-key"
)
documents = reader.load_data(limit=100)
2.6 数据预处理与清洗
from llama_index.core import Document
from llama_index.core.node_parser import SentenceSplitter
import re
def preprocess_documents(documents: List[Document]) -> List[Document]:
"""文档预处理管道"""
processed = []
for doc in documents:
text = doc.text
# 1. 清理HTML标签
text = re.sub(r'<[^>]+>', '', text)
# 2. 清理多余空白
text = re.sub(r'\s+', ' ', text).strip()
# 3. 去除特殊字符
text = re.sub(r'[^\w\s\u4e00-\u9fff.,!?;:()（）【】「」]', '', text)
# 4. 长度过滤
if len(text) < 50: # 太短的文档跳过
continue
# 创建处理后的文档
processed_doc = Document(
text=text,
metadata={**doc.metadata, "processed": True},
id_=doc.id_
)
processed.append(processed_doc)
return processed
三、索引类型（Index Types）
3.1 索引概述
索引是LlamaIndex的核心概念，它决定了数据如何被组织和检索。LlamaIndex提供了多种索引类型，每种都有其适用场景。
3.2 向量索引（VectorStoreIndex）
向量索引是最常用的索引类型，它将文档转换为向量嵌入并存储，支持高效的语义相似性搜索。
from llama_index.core import VectorStoreIndex, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# 配置全局设置
Settings.llm = OpenAI(model="gpt-4o", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 1024
Settings.chunk_overlap = 200
# 创建向量索引
index = VectorStoreIndex.from_documents(
documents,
show_progress=True # 显示进度条
)
# 从已有索引加载
# index = VectorStoreIndex.from_vector_store(vector_store)
# 基本查询
query_engine = index.as_query_engine(
similarity_top_k=5, # 返回最相似的5个节点
response_mode="compact" # 紧凑模式，节省token
)
response = query_engine.query("什么是机器学习？")
print(response)
# 流式查询
query_engine = index.as_query_engine(streaming=True)
streaming_response = query_engine.query("详细解释深度学习的原理")
for text in streaming_response.response_gen:
print(text, end="", flush=True)
3.3 树形索引（TreeIndex）
树形索引构建了一个层次化的摘要树，适合需要多级摘要和全局概览的场景。
from llama_index.core import TreeIndex
# 创建树形索引
tree_index = TreeIndex.from_documents(
documents,
num_children=10 # 每个节点的子节点数量
)
# 查询：从叶节点逐层向上摘要
query_engine = tree_index.as_query_engine(
child_branch_factor=2, # 每次选择2个子分支
response_mode="tree_summarize" # 树形摘要模式
)
response = query_engine.query("请总结这份文档的核心要点")
print(response)
3.4 关键词索引（KeywordTableIndex）
关键词索引通过提取文档的关键词来建立索引，适合需要精确关键词匹配的场景。
from llama_index.core import KeywordTableIndex
# 创建关键词索引
keyword_index = KeywordTableIndex.from_documents(
documents,
max_keywords_per_chunk=10 # 每个块提取的最大关键词数
)
# 查询
query_engine = keyword_index.as_query_engine(
similarity_top_k=5
)
response = query_engine.query("Python编程语言的特点")
print(response)
3.5 知识图谱索引（KnowledgeGraphIndex）
知识图谱索引从文档中提取实体和关系，构建结构化的知识图谱，适合需要实体关系推理的场景。
from llama_index.core import KnowledgeGraphIndex, KnowledgeGraphStore
from llama_index.core import StorageContext
# 创建知识图谱索引
kg_index = KnowledgeGraphIndex.from_documents(
documents,
max_triplets_per_chunk=5, # 每个块提取的最大三元组数
show_progress=True,
# 可选：使用自定义的KG存储
# storage_context=StorageContext.from_defaults(kg_store=your_kg_store)
)
# 查询知识图谱
query_engine = kg_index.as_query_engine(
response_mode="tree_summarize",
include_text=True # 包含原始文本作为上下文
)
response = query_engine.query("Python的创始人是谁？他还有什么其他成就？")
print(response)
# 可视化知识图谱（生成NetworkX图）
# kg_visualize = kg_index.get_networkx_graph()
3.6 组合索引（Composable Graph）
LlamaIndex支持将多个索引组合成一个图，实现跨文档、跨类型的联合查询。
from llama_index.core import ComposableGraph, VectorStoreIndex
# 假设有多个文档集合
finance_docs = SimpleDirectoryReader("./finance").load_data()
tech_docs = SimpleDirectoryReader("./tech").load_data()
# 分别创建索引
finance_index = VectorStoreIndex.from_documents(finance_docs)
tech_index = VectorStoreIndex.from_documents(tech_docs)
# 创建组合索引
from llama_index.core.indices.keyword_table import KeywordTableIndex
# 方法1：使用子索引
summary_index = KeywordTableIndex.from_documents([
Document(text="财务报告索引: 包含公司财务相关的所有文档"),
Document(text="技术文档索引: 包含技术架构和开发文档")
])
# 将子索引挂载到摘要节点
for node in summary_index.docstore.docs.values():
if "财务" in node.text:
node.metadata["index"] = finance_index
elif "技术" in node.text:
node.metadata["index"] = tech_index
# 方法2：直接使用ComposableGraph
graph = ComposableGraph.from_indices(
KeywordTableIndex,
[finance_index, tech_index],
index_summaries=[
"财务报告索引: 包含公司财务数据、收入报告、预算分析等",
"技术文档索引: 包含系统架构、API文档、开发指南等"
]
)
# 查询组合图
query_engine = graph.as_query_engine()
response = query_engine.query("公司的收入增长情况如何？")
print(response)
3.7 索引持久化与加载
from llama_index.core import StorageContext, load_index_from_storage
# 持久化索引
storage_context = StorageContext.from_defaults()
index.storage_context.persist(persist_dir="./storage")
# 从持久化存储加载索引
storage_context = StorageContext.from_defaults(persist_dir="./storage")
loaded_index = load_index_from_storage(storage_context)
# 查询加载的索引
query_engine = loaded_index.as_query_engine()
response = query_engine.query("测试查询")
四、检索策略与查询引擎
4.1 检索策略概述
检索是RAG系统的关键环节，直接影响回答的质量。LlamaIndex提供了多种检索策略。
4.2 基本检索模式
# 1. 语义检索（默认）
retriever = index.as_retriever(
similarity_top_k=5
)
nodes = retriever.retrieve("什么是深度学习？")
# 查看检索结果
for node in nodes:
print(f"分数: {node.score:.4f}")
print(f"内容: {node.node.text[:200]}")
print(f"来源: {node.node.metadata.get('file_name', '未知')}")
print("---")
# 2. MMR检索（最大边际相关性）
retriever_mmr = index.as_retriever(
retriever_mode="mmr",
similarity_top_k=10,
mmr_threshold=0.5
)
# 3. 混合检索（结合关键词和语义）
retriever_hybrid = index.as_retriever(
retriever_mode="hybrid",
similarity_top_k=5,
keyword_top_k=3
)
4.3 高级查询引擎
from llama_index.core.query_engine import (
RetrieverQueryEngine,
SubQuestionQueryEngine,
RetryQueryEngine,
RouterQueryEngine
)
from llama_index.core.selectors import LLMSingleSelector
# 1. 子问题查询引擎 - 将复杂问题分解为子问题
from llama_index.core.tools import QueryEngineTool, ToolMetadata
# 创建不同领域的查询引擎工具
finance_tool = QueryEngineTool(
query_engine=finance_index.as_query_engine(),
metadata=ToolMetadata(
name="finance_engine",
description="用于回答关于公司财务、收入、利润、预算等财务相关问题"
)
)
tech_tool = QueryEngineTool(
query_engine=tech_index.as_query_engine(),
metadata=ToolMetadata(
name="tech_engine",
description="用于回答关于技术架构、API、系统设计等技术相关问题"
)
)
# 创建子问题查询引擎
sub_question_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=[finance_tool, tech_tool],
verbose=True
)
# 复杂问题会自动分解为子问题
response = sub_question_engine.query(
"公司的技术投入对财务表现有什么影响？"
)
print(response)
# 2. 路由查询引擎 - 根据问题自动选择最合适的查询引擎
router_engine = RouterQueryEngine(
selector=LLMSingleSelector.from_defaults(),
query_engine_tools=[finance_tool, tech_tool],
verbose=True
)
response = router_engine.query("Python框架的选择建议")
print(response)
4.4 自定义检索器
from llama_index.core.retrievers import BaseRetriever
from llama_index.core.schema import QueryBundle, NodeWithScore
from typing import List
class HybridRetriever(BaseRetriever):
"""自定义混合检索器：结合向量检索和关键词检索"""
def __init__(self, vector_retriever, keyword_retriever, alpha=0.5):
self.vector_retriever = vector_retriever
self.keyword_retriever = keyword_retriever
self.alpha = alpha # 向量检索权重
def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]:
# 向量检索
vector_nodes = self.vector_retriever.retrieve(query_bundle)
# 关键词检索
keyword_nodes = self.keyword_retriever.retrieve(query_bundle)
# 合并并重新排序
node_scores = {}
for i, node in enumerate(vector_nodes):
node_id = node.node.node_id
vector_score = node.score * self.alpha
node_scores[node_id] = (node, vector_score)
for i, node in enumerate(keyword_nodes):
node_id = node.node.node_id
keyword_score = node.score * (1 - self.alpha)
if node_id in node_scores:
existing_node, existing_score = node_scores[node_id]
node_scores[node_id] = (existing_node, existing_score + keyword_score)
else:
node_scores[node_id] = (node, keyword_score)
# 按综合分数排序
sorted_nodes = sorted(
node_scores.values(),
key=lambda x: x[1],
reverse=True
)
return [NodeWithScore(node=n.node, score=n.score) for n, s in sorted_nodes[:10]]
# 使用自定义检索器
vector_retriever = index.as_retriever(similarity_top_k=10)
keyword_retriever = KeywordTableIndex.from_documents(documents).as_retriever(similarity_top_k=10)
hybrid = HybridRetriever(vector_retriever, keyword_retriever, alpha=0.7)
nodes = hybrid.retrieve("机器学习算法比较")
4.5 查询变换与优化
from llama_index.core.indices.query.query_transform import HyDEQueryTransform
from llama_index.core.query_engine import TransformQueryEngine
# HyDE（假设文档嵌入）变换
# 先让LLM生成一个假设性的答案文档，再用这个文档进行检索
hyde_transform = HyDEQueryTransform(
llm=OpenAI(model="gpt-4o"),
include_original=True # 同时包含原始查询
)
hyde_engine = TransformQueryEngine(
query_engine=index.as_query_engine(),
query_transform=hyde_transform
)
response = hyde_engine.query("如何优化数据库查询性能？")
print(response)
# Step-back查询变换 - 生成更通用的问题
from llama_index.core.indices.query.query_transform.base import StepDecomposeQueryTransform
step_back_transform = StepDecomposeQueryTransform(
llm=OpenAI(model="gpt-4o"),
verbose=True
)
step_back_engine = TransformQueryEngine(
query_engine=index.as_query_engine(),
query_transform=step_back_transform
)
五、自定义RAG Pipeline
5.1 RAG Pipeline概述
LlamaIndex允许你完全自定义RAG的每一个环节，从数据处理到最终输出。
5.2 完整的自定义RAG Pipeline
from llama_index.core import VectorStoreIndex, Settings
from llama_index.core.node_parser import SentenceSplitter, SemanticSplitterNodeParser
from llama_index.core.extractors import (
TitleExtractor,
QuestionsAnsweredExtractor,
SummaryExtractor,
KeywordExtractor
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# 配置模型
Settings.llm = OpenAI(model="gpt-4o")
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
# 创建数据处理管道
pipeline = IngestionPipeline(
transformations=[
# 1. 文本分割
SentenceSplitter(
chunk_size=1024,
chunk_overlap=200
),
# 2. 元数据提取
TitleExtractor(
llm=OpenAI(model="gpt-4o-mini"),
nodes=5 # 使用5个节点来提取标题
),
QuestionsAnsweredExtractor(
llm=OpenAI(model="gpt-4o-mini"),
questions=3 # 每个块生成3个可能的问题
),
KeywordExtractor(
llm=OpenAI(model="gpt-4o-mini"),
keywords=5 # 每个块提取5个关键词
),
# 3. 嵌入生成（最后一步）
OpenAIEmbedding(model="text-embedding-3-small")
]
)
# 运行管道
nodes = pipeline.run(documents=documents)
print(f"处理后节点数: {len(nodes)}")
# 查看增强后的节点
for node in nodes[:2]:
print(f"标题: {node.metadata.get('document_title', '无')}")
print(f"关键词: {node.metadata.get('excerpt_keywords', '无')}")
print(f"问题: {node.metadata.get('questions_this_excerpt_can_answer', '无')}")
print(f"内容预览: {node.text[:200]}")
print("---")
5.3 语义分割器
from llama_index.core.node_parser import SemanticSplitterNodeParser
# 语义分割器 - 根据语义相似度来分割文本
semantic_splitter = SemanticSplitterNodeParser(
buffer_size=1,
breakpoint_percentile_threshold=95,
embed_model=OpenAIEmbedding(model="text-embedding-3-small")
)
# 语义分割会保持语义相关的段落在一起
semantic_nodes = semantic_splitter.get_nodes_from_documents(documents)
print(f"语义分割后节点数: {len(semantic_nodes)}")
# 对比普通分割
from llama_index.core.node_parser import SentenceSplitter
regular_splitter = SentenceSplitter(chunk_size=1024)
regular_nodes = regular_splitter.get_nodes_from_documents(documents)
print(f"普通分割后节点数: {len(regular_nodes)}")
5.4 自定义响应合成器
from llama_index.core.response_synthesizers import (
CompactAndRefine,
TreeSummarize,
Refine,
SimpleSummarize
)
from llama_index.core import get_response_synthesizer
# 1. Refine模式 - 逐个处理节点，逐步精炼答案
refine_synthesizer = get_response_synthesizer(
response_mode="refine",
verbose=True
)
# 2. Compact模式 - 尽可能多地将节点打包到单个LLM调用中
compact_synthesizer = get_response_synthesizer(
response_mode="compact"
)
# 3. Tree Summarize模式 - 树形递归摘要
tree_synthesizer = get_response_synthesizer(
response_mode="tree_summarize",
num_children=5
)
# 4. 自定义Prompt的响应合成器
from llama_index.core.prompts import PromptTemplate
custom_qa_prompt = PromptTemplate(
"""你是一个专业的技术文档分析师。
请根据以下上下文信息，准确回答用户的问题。
规则：
1. 只使用上下文中的信息来回答
2. 如果上下文中没有相关信息，明确说明
3. 引用具体的上下文来源
4. 使用结构化的格式（标题、列表等）
上下文信息：
{context_str}
问题：{query_str}
回答："""
)
custom_synthesizer = get_response_synthesizer(
text_qa_template=custom_qa_prompt,
response_mode="compact"
)
# 使用自定义合成器创建查询引擎
query_engine = index.as_query_engine(
response_synthesizer=custom_synthesizer,
similarity_top_k=5
)
response = query_engine.query("如何设计高可用的系统架构？")
print(response)
5.5 后处理器
from llama_index.core.postprocessor import (
SimilarityPostprocessor,
KeywordNodePostprocessor,
MetadataReplacementPostProcessor,
SentenceEmbeddingOptimizer
)
# 1. 相似度过滤 - 移除低于阈值的节点
similarity_processor = SimilarityPostprocessor(
similarity_cutoff=0.7
)
# 2. 关键词过滤 - 只保留包含特定关键词的节点
keyword_processor = KeywordNodePostprocessor(
required_keywords=["Python", "机器学习"],
exclude_keywords=["广告", "推广"]
)
# 3. 元数据替换 - 用元数据字段替换节点文本
metadata_processor = MetadataReplacementPostProcessor(
target_metadata_key="window" # 使用滑动窗口上下文
)
# 4. 句子嵌入优化 - 移除不相关的句子
sentence_optimizer = SentenceEmbeddingOptimizer(
percentile_cutoff=0.5, # 保留相似度前50%的句子
context_before=1, # 保留前1句上下文
context_after=1 # 保留后1句上下文
)
# 组合后处理器
query_engine = index.as_query_engine(
node_postprocessors=[
similarity_processor,
keyword_processor,
sentence_optimizer
],
similarity_top_k=10 # 先检索较多节点，再通过后处理过滤
)
六、评估框架（Evaluation Module）
6.1 评估的重要性
在生产环境中部署RAG系统时，评估是必不可少的环节。LlamaIndex提供了完善的评估框架，可以系统地衡量检索质量和回答质量。
6.2 检索质量评估
from llama_index.core.evaluation import (
RetrieverEvaluator,
generate_question_context_pairs
)
from llama_index.core import VectorStoreIndex
import pandas as pd
# 1. 生成评估数据集
# 使用LLM为文档生成问题-上下文对
qa_dataset = generate_question_context_pairs(
nodes=nodes[:50], # 使用前50个节点
llm=OpenAI(model="gpt-4o"),
num_questions_per_chunk=2,
qa_generate_prompt_tmpl="""\
根据以下上下文信息，生成{num_questionsPerChunk}个相关的问题。
问题应该能够仅从这个上下文中得到回答。
上下文：
{context_str}
问题列表：
"""
)
# 2. 评估检索器
retriever = index.as_retriever(similarity_top_k=5)
retriever_evaluator = RetrieverEvaluator.from_metric_names(
["mrr", "hit_rate"],
retriever=retriever
)
# 运行评估
eval_results = retriever_evaluator.evaluate_dataset(qa_dataset)
# 查看结果
df = pd.DataFrame(eval_results)
print(f"平均MRR: {df['mrr'].mean():.4f}")
print(f"平均Hit Rate: {df['hit_rate'].mean():.4f}")
# 3. 不同检索策略的对比评估
retrievers = {
"vector": index.as_retriever(similarity_top_k=5),
"mmr": index.as_retriever(retriever_mode="mmr", similarity_top_k=5),
"hybrid": index.as_retriever(retriever_mode="hybrid", similarity_top_k=5)
}
results = {}
for name, retriever in retrievers.items():
evaluator = RetrieverEvaluator.from_metric_names(
["mrr", "hit_rate"],
retriever=retriever
)
eval_result = evaluator.evaluate_dataset(qa_dataset)
df = pd.DataFrame(eval_result)
results[name] = {
"mrr": df["mrr"].mean(),
"hit_rate": df["hit_rate"].mean()
}
# 对比结果
comparison_df = pd.DataFrame(results).T
print("\n=== 检索策略对比 ===")
print(comparison_df)
6.3 回答质量评估
from llama_index.core.evaluation import (
FaithfulnessEvaluator,
RelevancyEvaluator,
CorrectnessEvaluator,
GuidelineEvaluator
)
# 1. 忠实度评估 - 回答是否基于检索到的上下文
faithfulness_eval = FaithfulnessEvaluator(
llm=OpenAI(model="gpt-4o")
)
# 2. 相关性评估 - 回答是否与问题相关
relevancy_eval = RelevancyEvaluator(
llm=OpenAI(model="gpt-4o")
)
# 3. 正确性评估 - 回答是否与参考答案一致
correctness_eval = CorrectnessEvaluator(
llm=OpenAI(model="gpt-4o")
)
# 4. 准则评估 - 根据自定义准则评估
guideline_eval = GuidelineEvaluator(
llm=OpenAI(model="gpt-4o"),
guidelines="""
评估标准：
1. 回答必须准确，不能包含错误信息
2. 回答必须完整，覆盖问题的所有方面
3. 回答必须有条理，使用清晰的结构
4. 回答必须引用来源
"""
)
# 运行评估
query_engine = index.as_query_engine()
test_questions = [
"什么是机器学习？",
"深度学习和传统机器学习有什么区别？",
"如何选择合适的算法？"
]
for question in test_questions:
response = query_engine.query(question)
# 评估
faithfulness_result = faithfulness_eval.evaluate_response(
query=question,
response=response
)
relevancy_result = relevancy_eval.evaluate_response(
query=question,
response=response
)
print(f"问题: {question}")
print(f"忠实度: {faithfulness_result.passing} (得分: {faithfulness_result.score:.2f})")
print(f"相关性: {relevancy_result.passing} (得分: {relevancy_result.score:.2f})")
print(f"回答: {response.response[:200]}...")
print("---")
6.4 批量评估与基准测试
from llama_index.core.evaluation import BatchEvalRunner
import asyncio
# 创建批量评估器
runner = BatchEvalRunner(
{
"faithfulness": FaithfulnessEvaluator(llm=OpenAI(model="gpt-4o")),
"relevancy": RelevancyEvaluator(llm=OpenAI(model="gpt-4o")),
},
workers=4 # 并行工作线程数
)
# 准备评估数据
eval_queries = ["什么是Python？", "机器学习的应用场景？", "深度学习的优势？"]
eval_responses = [query_engine.query(q) for q in eval_queries]
# 批量评估
async def run_batch_eval():
eval_results = await runner.aevaluate_responses(
queries=eval_queries,
responses=eval_responses
)
return eval_results
eval_results = asyncio.run(run_batch_eval())
# 汇总结果
for metric, results in eval_results.items():
scores = [r.score for r in results]
print(f"{metric}: 平均={sum(scores)/len(scores):.4f}, 最小={min(scores):.4f}, 最大={max(scores):.4f}")
七、与LangChain对比与集成
7.1 功能对比详解
| 功能维度 | LlamaIndex | LangChain |
|---|---|---|
| 核心定位 | 数据框架（RAG专家） | LLM应用编排框架 |
| 数据摄入 | 原生160+连接器 | 通过Community包支持 |
| 索引类型 | 向量/树/关键词/知识图谱/组合 | 主要依赖外部向量数据库 |
| 检索策略 | MMR/混合/子问题/路由/HyDE | 基础检索+自定义 |
| 评估 | 内置完善评估框架 | 通过LangSmith |
| Agent | 支持但非核心 | 核心能力，LangGraph |
| 输出解析 | 内置结构化输出 | Pydantic集成 |
| 流式支持 | 原生支持 | LCEL原生支持 |
| 生产工具 | LlamaDeploy | LangGraph Platform |
7.2 集成使用
LlamaIndex和LangChain可以无缝集成，发挥各自优势。
# 方式1：在LlamaIndex中使用LangChain的LLM和Embeddings
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from llama_index.core import Settings
from llama_index.llms.langchain import LangChainLLM
from llama_index.embeddings.langchain import LangchainEmbedding
# 包装LangChain组件
lc_llm = ChatOpenAI(model="gpt-4o")
lc_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = LangChainLLM(lc_llm)
embed_model = LangchainEmbedding(lc_embeddings)
# 设置为LlamaIndex默认组件
Settings.llm = llm
Settings.embed_model = embed_model
# 现在LlamaIndex使用LangChain的模型
index = VectorStoreIndex.from_documents(documents)
# 方式2：在LangChain中使用LlamaIndex的检索器
from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document as LCDocument
class LlamaIndexRetriever(BaseRetriever):
"""将LlamaIndex检索器适配为LangChain接口"""
def __init__(self, llama_retriever):
self.llama_retriever = llama_retriever
def _get_relevant_documents(self, query: str):
nodes = self.llama_retriever.retrieve(query)
return [
LCDocument(
page_content=node.node.text,
metadata={**node.node.metadata, "score": node.score}
)
for node in nodes
]
# 使用
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llama_retriever = index.as_retriever(similarity_top_k=5)
lc_retriever = LlamaIndexRetriever(llama_retriever)
# 在LangChain RAG链中使用
prompt = ChatPromptTemplate.from_template(
"根据以下上下文回答问题：\n\n{context}\n\n问题：{question}"
)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": lc_retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| ChatOpenAI(model="gpt-4o")
| StrOutputParser()
)
result = rag_chain.invoke("什么是深度学习？")
print(result)
7.3 何时选择哪个框架
选择LlamaIndex当：
- 主要需求是构建RAG系统
- 需要多种索引类型和检索策略
- 需要完善的评估框架
- 数据密集型应用
- 需要快速原型开发
选择LangChain当：
- 需要复杂的Agent和工具调用
- 需要与多种外部服务集成
- 需要复杂的链式编排逻辑
- 需要LangSmith的调试监控能力
- 多步骤、有状态的工作流
两者结合使用：
- 大型项目中，用LlamaIndex处理数据层，LangChain处理应用层
- 需要两者各自优势的场景
八、LlamaParse文档解析
8.1 LlamaParse简介
LlamaParse是LlamaIndex提供的云端文档解析服务，特别擅长处理复杂的文档格式，如包含表格、图表、公式的PDF文件。
8.2 基本使用
from llama_parse import LlamaParse
from llama_index.core import VectorStoreIndex
# 配置解析器
parser = LlamaParse(
result_type="markdown", # 输出格式：markdown或text
num_workers=4, # 并行处理线程数
verbose=True,
language="chinese", # 文档语言
# 高级选项
skip_diagonal_text=False, # 是否跳过对角线文字
invalidate_cache=False, # 是否清除缓存
do_not_cache=False, # 是否禁用缓存
)
# 解析PDF文件
documents = parser.load_data("./complex_document.pdf")
print(f"解析出 {len(documents)} 个文档")
for doc in documents[:3]:
print(f"内容长度: {len(doc.text)} 字符")
print(f"内容预览: {doc.text[:300]}")
print("---")
# 直接创建索引
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("文档中的表格数据是什么？")
print(response)
8.3 高级配置
# 多文件解析
documents = parser.load_data([
"./document1.pdf",
"./document2.pdf",
"./document3.pdf"
])
# 自定义解析指令
parser = LlamaParse(
result_type="markdown",
parsing_instruction="""
请特别注意以下内容的提取：
1. 表格：保持表格结构，转换为Markdown表格格式
2. 图表：提取图表中的关键数据和描述
3. 公式：转换为LaTeX格式
4. 代码块：保持代码格式不变
5. 页眉页脚：忽略页眉页脚内容
""",
disable_ocr=False, # 启用OCR处理扫描件
invalidate_cache=True
)
# 与向量存储集成
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# 解析并存储
documents = parser.load_data("./technical_report.pdf")
# 创建Chroma向量存储
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("tech_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
# 构建索引
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context
)
8.4 解析质量优化
# 使用预处理和后处理提升解析质量
def enhanced_parse(file_path: str) -> List[Document]:
"""增强的文档解析流程"""
# 1. 使用LlamaParse解析
parser = LlamaParse(
result_type="markdown",
language="chinese",
verbose=True
)
documents = parser.load_data(file_path)
# 2. 后处理：清理和增强
processed_docs = []
for doc in documents:
text = doc.text
# 移除页码标记
import re
text = re.sub(r'第\s*\d+\s*页', '', text)
# 修复常见的OCR错误
text = text.replace('。', '。') # 标点规范化
# 移除多余的空行
text = re.sub(r'\n{3,}', '\n\n', text)
processed_doc = Document(
text=text.strip(),
metadata={**doc.metadata, "source": file_path}
)
if len(processed_doc.text) > 100: # 过滤太短的块
processed_docs.append(processed_doc)
return processed_docs
九、多模态索引
9.1 多模态概述
LlamaIndex支持对图像、音频等多模态数据进行索引和查询，能够处理包含图表、截图、照片的复杂文档。
9.2 图像索引
from llama_index.core import SimpleDirectoryReader
from llama_index.multi_modal_llms.openai import OpenAIMultiModal
from llama_index.core import VectorStoreIndex
# 加载图像文件
image_documents = SimpleDirectoryReader(
input_dir="./images",
required_exts=[".jpg", ".png", ".jpeg", ".gif"]
).load_data()
# 使用多模态LLM
multi_modal_llm = OpenAIMultiModal(
model="gpt-4o",
max_new_tokens=1000
)
# 为图像生成描述
from llama_index.core.schema import ImageDocument
def describe_images(image_docs):
"""为图像生成文本描述"""
described_docs = []
for img_doc in image_docs:
response = multi_modal_llm.complete(
prompt="请详细描述这张图片的内容，包括主要元素、文字、布局等。",
image_documents=[img_doc]
)
# 创建包含图像描述的文本文档
text_doc = Document(
text=f"图像描述：{response.text}\n\n原始图像路径：{img_doc.metadata.get('file_path', '未知')}",
metadata={**img_doc.metadata, "image_description": response.text}
)
described_docs.append(text_doc)
return described_docs
# 生成图像描述
text_docs = describe_images(image_documents)
# 创建文本索引
index = VectorStoreIndex.from_documents(text_docs)
query_engine = index.as_query_engine()
# 查询图像相关内容
response = query_engine.query("图片中有哪些图表？")
print(response)
9.3 多模态RAG
from llama_index.core.schema import ImageNode
from llama_index.core.query_engine import MultimodalQueryEngine
from llama_index.multi_modal_llms.openai import OpenAIMultiModal
# 创建多模态索引
from llama_index.core.indices.multi_modal import MultiModalVectorStoreIndex
# 准备文本和图像数据
text_nodes = [n for n in nodes if not isinstance(n, ImageNode)]
image_nodes = [n for n in nodes if isinstance(n, ImageNode)]
# 创建多模态索引
multi_modal_index = MultiModalVectorStoreIndex(
nodes=text_nodes + image_nodes
)
# 创建多模态查询引擎
multi_modal_llm = OpenAIMultiModal(model="gpt-4o")
query_engine = multi_modal_index.as_query_engine(
multi_modal_llm=multi_modal_llm,
similarity_top_k=5
)
# 查询
response = query_engine.query("请分析报告中的图表趋势")
print(response)
9.4 音频和视频处理
# 音频转录和索引
import whisper
def transcribe_audio(audio_path: str) -> Document:
"""将音频转录为文本并创建文档"""
model = whisper.load_model("base")
result = model.transcribe(audio_path, language="Chinese")
# 按段落组织转录文本
segments = []
for segment in result["segments"]:
start = segment["start"]
end = segment["end"]
text = segment["text"]
segments.append(f"[{start:.1f}s - {end:.1f}s] {text}")
full_text = "\n".join(segments)
return Document(
text=full_text,
metadata={
"source": audio_path,
"language": result.get("language", "unknown"),
"duration": result["segments"][-1]["end"] if result["segments"] else 0
}
)
# 转录并索引
audio_doc = transcribe_audio("./meeting_recording.mp3")
index = VectorStoreIndex.from_documents([audio_doc])
query_engine = index.as_query_engine()
response = query_engine.query("会议中讨论了哪些关键议题？")
print(response)
十、生产部署最佳实践
10.1 索引持久化与缓存
from llama_index.core import StorageContext, load_index_from_storage
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.storage.index_store import SimpleIndexStore
from llama_index.vector_stores.faiss import FaissVectorStore
import faiss
# 创建持久化存储
def create_persisted_index(documents, persist_dir="./storage"):
"""创建并持久化索引"""
import os
os.makedirs(persist_dir, exist_ok=True)
# 创建FAISS向量存储
d = 1536 # OpenAI embedding维度
faiss_index = faiss.IndexFlatL2(d)
vector_store = FaissVectorStore(faiss_index=faiss_index)
# 创建存储上下文
storage_context = StorageContext.from_defaults(
docstore=SimpleDocumentStore(),
index_store=SimpleIndexStore(),
vector_store=vector_store
)
# 创建索引
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context
)
# 持久化
storage_context.persist(persist_dir=persist_dir)
return index
def load_persisted_index(persist_dir="./storage"):
"""从持久化存储加载索引"""
# 重建向量存储
vector_store = FaissVectorStore.from_persist_dir(persist_dir)
# 创建存储上下文
storage_context = StorageContext.from_defaults(
persist_dir=persist_dir,
vector_store=vector_store
)
# 加载索引
index = load_index_from_storage(storage_context)
return index
# 增量更新索引
def update_index(index, new_documents):
"""增量添加新文档到索引"""
for doc in new_documents:
index.insert(doc)
return index
10.2 性能优化
from llama_index.core import Settings
from llama_index.embeddings.openai import OpenAIEmbedding
import asyncio
# 1. 批量嵌入优化
Settings.embed_model = OpenAIEmbedding(
model="text-embedding-3-small",
embed_batch_size=100 # 批量处理嵌入请求
)
# 2. 并行处理
async def parallel_query(queries: list, query_engine, max_concurrent=5):
"""并行处理多个查询"""
semaphore = asyncio.Semaphore(max_concurrent)
async def single_query(query):
async with semaphore:
return await query_engine.aquery(query)
tasks = [single_query(q) for q in queries]
results = await asyncio.gather(*tasks)
return results
# 3. 缓存机制
from llama_index.core.storage.kvstore import SimpleKVStore
from llama_index.core import get_response_synthesizer
class CachedQueryEngine:
"""带缓存的查询引擎"""
def __init__(self, query_engine, cache_dir="./cache"):
self.query_engine = query_engine
self.cache = SimpleKVStore(cache_dir)
def query(self, question: str):
# 检查缓存
cached = self.cache.get(question)
if cached:
return cached["response"]
# 执行查询
response = self.query_engine.query(question)
# 存入缓存
self.cache.put(question, {"response": response.response})
return response.response
10.3 监控与日志
import logging
from llama_index.core import set_global_handler
from llama_index.core.callbacks import CallbackManager
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("llamaindex")
# 使用LlamaIndex的回调系统
from llama_index.core.callbacks import LlamaDebugHandler
debug_handler = LlamaDebugHandler()
callback_manager = CallbackManager([debug_handler])
# 全局设置回调
from llama_index.core import Settings
Settings.callback_manager = callback_manager
# 查询后查看统计
query_engine = index.as_query_engine()
response = query_engine.query("测试查询")
# 获取统计信息
print("=== 查询统计 ===")
print(f"总耗时: {debug_handler.flush_event_logs()}")
print(f"LLM调用次数: {len(debug_handler.llm_inputs)}")
10.4 安全性考虑
# 1. 输入验证和过滤
def sanitize_query(query: str) -> str:
"""清理和验证查询输入"""
import re
# 移除潜在的注入内容
query = re.sub(r'[<>{}]', '', query)
# 限制长度
if len(query) > 1000:
raise ValueError("查询过长，请缩短输入")
# 过滤空查询
if not query.strip():
raise ValueError("查询不能为空")
return query.strip()
# 2. 访问控制
class AuthorizedQueryEngine:
"""带访问控制的查询引擎"""
def __init__(self, query_engine, allowed_users=None):
self.query_engine = query_engine
self.allowed_users = allowed_users or []
def query(self, question: str, user_id: str):
if user_id not in self.allowed_users:
raise PermissionError(f"用户 {user_id} 无权访问")
sanitized = sanitize_query(question)
return self.query_engine.query(sanitized)
# 3. 数据脱敏
def mask_sensitive_data(text: str) -> str:
"""脱敏处理"""
import re
# 手机号
text = re.sub(r'1[3-9]\d{9}', '1**********', text)
# 身份证号
text = re.sub(r'\d{17}[\dXx]', '******************', text)
# 邮箱
text = re.sub(r'[\w.]+@[\w.]+', '***@***.***', text)
return text
十一、实战案例：构建企业文档智能助手
11.1 系统概述
我们将构建一个完整的企业文档智能助手，具备以下功能：
- 支持多种文档格式（PDF、Word、Markdown、网页）
- 智能文档分类和标签
- 多级检索策略
- 对话式问答
- 文档摘要生成
- 引用溯源
11.2 完整实现
"""
企业文档智能助手 - 完整实现
功能：文档摄入、索引构建、智能问答、评估优化
"""
import os
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime
from llama_index.core import (
VectorStoreIndex, SimpleDirectoryReader, Document,
Settings, StorageContext, load_index_from_storage,
get_response_synthesizer
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.evaluation import (
FaithfulnessEvaluator, RelevancyEvaluator
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.faiss import FaissVectorStore
from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.core.prompts import PromptTemplate
# ==================== 1. 配置管理 ====================
@dataclass
class AssistantConfig:
"""助手配置"""
llm_model: str = "gpt-4o"
embedding_model: str = "text-embedding-3-small"
chunk_size: int = 1024
chunk_overlap: int = 200
similarity_top_k: int = 5
storage_dir: str = "./enterprise_storage"
max_input_length: int = 2000
# ==================== 2. 文档管理器 ====================
class DocumentManager:
"""文档管理器：负责文档的加载、预处理和管理"""
def __init__(self, config: AssistantConfig):
self.config = config
self.documents: List[Document] = []
self.metadata_index: Dict[str, Dict] = {}
def load_from_directory(self, directory: str, file_types: List[str] = None):
"""从目录加载文档"""
required_exts = file_types or [".pdf", ".txt", ".md", ".docx"]
reader = SimpleDirectoryReader(
input_dir=directory,
recursive=True,
required_exts=required_exts,
filename_as_id=True
)
documents = reader.load_data()
self.documents.extend(documents)
# 建立元数据索引
for doc in documents:
self.metadata_index[doc.id_] = {
"source": doc.metadata.get("file_path", "未知"),
"loaded_at": datetime.now().isoformat(),
"length": len(doc.text)
}
print(f"已加载 {len(documents)} 个文档")
return documents
def add_document(self, text: str, metadata: Dict = None):
"""手动添加文档"""
doc = Document(
text=text,
metadata=metadata or {}
)
self.documents.append(doc)
return doc
def preprocess(self) -> List[Document]:
"""预处理所有文档"""
processed = []
for doc in self.documents:
# 清理文本
text = doc.text.strip()
if len(text) < 50: # 跳过太短的文档
continue
# 添加处理标记
doc.metadata["processed"] = True
doc.metadata["processed_at"] = datetime.now().isoformat()
processed.append(doc)
self.documents = processed
return processed
# ==================== 3. 索引管理器 ====================
class IndexManager:
"""索引管理器：负责索引的创建、持久化和加载"""
def __init__(self, config: AssistantConfig):
self.config = config
self.index: Optional[VectorStoreIndex] = None
# 配置全局设置
Settings.llm = OpenAI(model=config.llm_model, temperature=0)
Settings.embed_model = OpenAIEmbedding(model=config.embedding_model)
Settings.chunk_size = config.chunk_size
Settings.chunk_overlap = config.chunk_overlap
def build_index(self, documents: List[Document]):
"""构建索引"""
# 创建节点解析器
node_parser = SentenceSplitter(
chunk_size=self.config.chunk_size,
chunk_overlap=self.config.chunk_overlap
)
# 创建向量存储
import faiss
d = 1536 # embedding维度
faiss_index = faiss.IndexFlatL2(d)
vector_store = FaissVectorStore(faiss_index=faiss_index)
# 创建存储上下文
storage_context = StorageContext.from_defaults(
vector_store=vector_store
)
# 构建索引
self.index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context,
transformations=[node_parser],
show_progress=True
)
print(f"索引构建完成，包含 {len(self.index.docstore.docs)} 个节点")
return self.index
def persist(self):
"""持久化索引"""
if self.index is None:
raise ValueError("索引尚未构建")
os.makedirs(self.config.storage_dir, exist_ok=True)
self.index.storage_context.persist(
persist_dir=self.config.storage_dir
)
print(f"索引已持久化到 {self.config.storage_dir}")
def load(self):
"""加载持久化的索引"""
if not os.path.exists(self.config.storage_dir):
raise FileNotFoundError(f"存储目录不存在: {self.config.storage_dir}")
vector_store = FaissVectorStore.from_persist_dir(
self.config.storage_dir
)
storage_context = StorageContext.from_defaults(
persist_dir=self.config.storage_dir,
vector_store=vector_store
)
self.index = load_index_from_storage(storage_context)
print("索引加载完成")
return self.index
# ==================== 4. 查询引擎 ====================
class SmartQueryEngine:
"""智能查询引擎：支持多种查询模式"""
def __init__(self, index: VectorStoreIndex, config: AssistantConfig):
self.index = index
self.config = config
# 自定义QA提示模板
self.qa_prompt = PromptTemplate(
"""你是一位专业的企业知识助手。请根据以下上下文信息，准确、专业地回答用户的问题。
回答规则：
1. 严格基于提供的上下文信息回答，不要编造信息
2. 如果上下文中没有相关信息，明确告知用户
3. 引用具体的来源文档（如果可用）
4. 使用清晰的结构化格式（标题、列表等）
5. 对于技术问题，提供详细的解释和示例
6. 保持专业和友好的语气
上下文信息：
{context_str}
用户问题：{query_str}
请回答："""
)
# 创建基础查询引擎
self.base_engine = index.as_query_engine(
similarity_top_k=config.similarity_top_k,
node_postprocessors=[
SimilarityPostprocessor(similarity_cutoff=0.6)
],
text_qa_template=self.qa_prompt,
response_mode="compact"
)
def query(self, question: str) -> Dict:
"""执行查询"""
# 输入验证
if not question.strip():
return {"answer": "请输入您的问题", "sources": []}
if len(question) > self.config.max_input_length:
return {"answer": "问题过长，请缩短到2000字以内", "sources": []}
# 执行查询
response = self.base_engine.query(question)
# 提取来源信息
sources = []
for node in response.source_nodes:
sources.append({
"content": node.node.text[:200],
"score": float(node.score) if node.score else 0,
"metadata": dict(node.node.metadata) if node.node.metadata else {}
})
return {
"answer": response.response,
"sources": sources,
"query": question
}
def stream_query(self, question: str):
"""流式查询"""
streaming_engine = self.index.as_query_engine(
streaming=True,
similarity_top_k=self.config.similarity_top_k,
text_qa_template=self.qa_prompt
)
response = streaming_engine.query(question)
return response.response_gen
# ==================== 5. 评估系统 ====================
class EvaluationSystem:
"""评估系统：用于评估查询质量"""
def __init__(self):
self.faithfulness_eval = FaithfulnessEvaluator(
llm=OpenAI(model="gpt-4o")
)
self.relevancy_eval = RelevancyEvaluator(
llm=OpenAI(model="gpt-4o")
)
def evaluate_response(self, query: str, response) -> Dict:
"""评估单个响应"""
faithfulness = self.faithfulness_eval.evaluate_response(
query=query, response=response
)
relevancy = self.relevancy_eval.evaluate_response(
query=query, response=response
)
return {
"faithfulness": {
"passing": faithfulness.passing,
"score": faithfulness.score,
"feedback": faithfulness.feedback
},
"relevancy": {
"passing": relevancy.passing,
"score": relevancy.score,
"feedback": relevancy.feedback
}
}
def batch_evaluate(self, qa_pairs: List[Dict], query_engine) -> Dict:
"""批量评估"""
results = []
for qa in qa_pairs:
query = qa["question"]
reference = qa.get("reference", "")
response = query_engine.query(query)
eval_result = self.evaluate_response(query, response)
results.append({
"query": query,
"response": response.response,
"evaluation": eval_result
})
# 汇总统计
faith_scores = [r["evaluation"]["faithfulness"]["score"] for r in results]
relev_scores = [r["evaluation"]["relevancy"]["score"] for r in results]
return {
"results": results,
"summary": {
"avg_faithfulness": sum(faith_scores) / len(faith_scores),
"avg_relevancy": sum(relev_scores) / len(relev_scores),
"total_queries": len(results)
}
}
# ==================== 6. 主程序 ====================
def main():
"""主程序：演示完整的文档助手流程"""
# 配置
config = AssistantConfig(
llm_model="gpt-4o",
embedding_model="text-embedding-3-small",
chunk_size=1024,
similarity_top_k=5,
storage_dir="./enterprise_storage"
)
# 1. 初始化文档管理器
print("=== 初始化文档管理器 ===")
doc_manager = DocumentManager(config)
doc_manager.load_from_directory("./documents")
doc_manager.preprocess()
# 2. 构建索引
print("\n=== 构建索引 ===")
index_manager = IndexManager(config)
index = index_manager.build_index(doc_manager.documents)
index_manager.persist()
# 3. 创建查询引擎
print("\n=== 创建查询引擎 ===")
query_engine = SmartQueryEngine(index, config)
# 4. 测试查询
print("\n=== 测试查询 ===")
test_questions = [
"公司的技术架构是怎样的？",
"最近的财务报告有哪些关键数据？",
"新员工入职流程是什么？"
]
for question in test_questions:
print(f"\n问题: {question}")
result = query_engine.query(question)
print(f"回答: {result['answer'][:200]}...")
print(f"来源数: {len(result['sources'])}")
# 5. 评估
print("\n=== 评估 ===")
evaluator = EvaluationSystem()
eval_pairs = [
{
"question": "什么是深度学习？",
"reference": "深度学习是机器学习的一个子领域，使用多层神经网络学习数据的层次化表示。"
}
]
eval_results = evaluator.batch_evaluate(eval_pairs, query_engine)
print(f"平均忠实度: {eval_results['summary']['avg_faithfulness']:.4f}")
print(f"平均相关性: {eval_results['summary']['avg_relevancy']:.4f}")
if __name__ == "__main__":
main()
11.3 系统架构优化建议
向量数据库选择：
- 小规模（<10万文档）：FAISS本地存储
- 中规模（10万-100万）：Chroma或Milvus
- 大规模（>100万）：Pinecone、Weaviate或Qdrant
索引策略：
- 频繁更新的数据使用增量索引
- 静态数据使用全量索引
- 不同类型的文档使用不同的索引类型
查询优化：
- 实施查询缓存减少重复查询
- 使用混合检索提高召回率
- 对长文档使用层次化索引
监控与运维：
- 监控查询延迟和成功率
- 设置异常告警
- 定期评估和优化索引质量
十二、常见问题解答
Q1: LlamaIndex和LangChain如何选择？
选择LlamaIndex：主要需求是RAG，需要多种索引和检索策略，需要内置评估框架。
选择LangChain：需要复杂的Agent、工具调用、多步编排。
两者结合：大型项目中，LlamaIndex处理数据层，LangChain处理应用层。
Q2: 如何优化RAG系统的回答质量？
- 数据质量：确保输入文档的质量和相关性
- 分块策略：使用语义分割而非固定长度分割
- 嵌入模型：选择适合你领域的嵌入模型
- 检索策略：使用MMR或混合检索提高多样性
- 后处理：使用相似度过滤和元数据过滤
- 提示优化：精心设计系统提示和QA模板
- 评估迭代：使用评估框架持续优化
Q3: 如何处理多语言文档？
# 使用多语言嵌入模型
embed_model = OpenAIEmbedding(model="text-embedding-3-small") # 支持多语言
# 在查询时指定语言
Settings.llm = OpenAI(
model="gpt-4o",
system_prompt="请使用与文档相同的语言回答问题。如果文档是中文，用中文回答。"
)
Q4: 如何降低API调用成本？
- 使用更小的模型（如gpt-4o-mini）处理简单任务
- 实施缓存机制，对相同查询返回缓存结果
- 优化chunk_size，减少嵌入调用次数
- 使用批量API而非逐个调用
- 对非关键任务使用本地模型
Q5: 索引应该多久更新一次？
- 实时数据：使用增量更新，每次有新文档就更新
- 每日更新：适合大多数业务场景
- 每周更新：适合变化不频繁的文档
- 按需更新：适合静态知识库
十三、总结
LlamaIndex作为专注于数据索引和检索的框架，为构建RAG系统提供了全面而深入的解决方案。本教程涵盖了从基础概念到高级应用的完整知识体系：
- 数据连接器：支持160+数据源，覆盖几乎所有常见的数据格式
- 索引类型：向量、树形、关键词、知识图谱等多种索引，满足不同场景需求
- 检索策略：语义检索、MMR、混合检索、子问题分解等高级策略
- 自定义Pipeline：灵活的管道配置，支持深度定制
- 评估框架：内置完善的评估工具，支持系统化的质量保证
- 与LangChain集成：无缝对接，发挥各自优势
- LlamaParse：强大的文档解析能力，处理复杂文档
- 多模态支持：图像、音频等多模态数据索引
- 生产部署：持久化、性能优化、安全性等生产级特性
掌握LlamaIndex，你就拥有了构建企业级知识密集型应用的核心能力。随着LLM技术的不断发展，数据框架的重要性将日益凸显。
下一步学习建议：
- 深入学习LlamaParse的高级功能
- 探索LlamaDeploy的生产部署方案
- 研究自定义嵌入模型的训练和微调
- 参与LlamaIndex开源社区，贡献代码和反馈
本教程基于LlamaIndex v0.11+版本编写，部分API可能随版本更新而变化，请以官方文档为准。
