LlamaIndex 教程：构建智能检索增强生成系统的全面指南

# 安装核心包
pip install llama-index

# 安装特定集成组件（以 OpenAI 为例）
pip install llama-index-llms-openai
pip install llama-index-embeddings-openai

# 安装向量数据库集成（以 Chroma 为例）
pip install llama-index-vector-stores-chroma

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

from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置语言模型
Settings.llm = OpenAI(model="gpt-4o", temperature=0.1)

# 配置嵌入模型
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 配置文本切分参数
Settings.chunk_size = 512
Settings.chunk_overlap = 50

from llama_index.core import SimpleDirectoryReader, Document

# 从目录加载所有文件
documents = SimpleDirectoryReader("./data").load_data()

# 加载单个文件
documents = SimpleDirectoryReader(input_files=["./document.pdf"]).load_data()

# 手动创建文档对象
doc = Document(
    text="这是文档的正文内容",
    metadata={
        "source": "manual_input",
        "author": "张三",
        "date": "2024-01-01"
    }
)

from llama_index.core.extractors import (
    TitleExtractor,
    QuestionsAnsweredExtractor,
    SummaryExtractor,
    KeywordExtractor
)
from llama_index.core.ingestion import IngestionPipeline

# 构建数据摄入流水线
pipeline = IngestionPipeline(
    transformations=[
        TitleExtractor(nodes=5),
        QuestionsAnsweredExtractor(questions=3),
        KeywordExtractor(keywords=10),
    ]
)

nodes = pipeline.run(documents=documents)

from llama_index.core.node_parser import (
    SentenceSplitter,
    SemanticSplitterNodeParser
)

# 基于句子的切分器
sentence_splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50,
    paragraph_separator="\n\n"
)

# 语义感知切分器
semantic_splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,
    embed_model=Settings.embed_model
)

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# 使用持久化向量数据库
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("my_collection")
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,
    show_progress=True
)

from llama_index.core import SummaryIndex

summary_index = SummaryIndex.from_documents(documents)
query_engine = summary_index.as_query_engine(
    response_mode="tree_summarize"
)
response = query_engine.query("请对这份文档进行全面总结")

from llama_index.core import StorageContext, load_index_from_storage

# 持久化保存索引
index.storage_context.persist(persist_dir="./storage")

# 从磁盘加载索引
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,           # 检索最相似的5个节点
    response_mode="compact",      # 响应模式
    streaming=True                # 启用流式输出
)

# 执行查询
response = query_engine.query("什么是检索增强生成技术？")
print(response.response)

# 查看参考来源
for node in response.source_nodes:
    print(f"相关度评分: {node.score:.4f}")
    print(f"文本内容: {node.node.text[:200]}")

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

# 配置混合检索
retriever = QueryFusionRetriever(
    retrievers=[
        vector_retriever,
        keyword_retriever
    ],
    similarity_top_k=5,
    num_queries=4,      # 自动生成多个查询变体以提升召回率
    mode="reciprocal_rerank",
    use_async=True
)

query_engine = RetrieverQueryEngine(retriever=retriever)

from llama_index.core.postprocessor import (
    SentenceTransformerRerank,
    CohereRerank,
    LLMRerank
)

# 使用 Cohere 重排序模型
cohere_rerank = CohereRerank(
    api_key="your-cohere-api-key",
    top_n=3
)

# 使用本地 SentenceTransformer 模型重排序
local_rerank = SentenceTransformerRerank(
    model="cross-encoder/ms-marco-MiniLM-L-2-v2",
    top_n=3
)

query_engine = index.as_query_engine(
    similarity_top_k=10,
    node_postprocessors=[cohere_rerank]
)

from llama_index.core.chat_engine import CondensePlusContextChatEngine

chat_engine = index.as_chat_engine(
    chat_mode="condense_plus_context",
    verbose=True,
    system_prompt="你是一个专业的知识库助手，请基于提供的文档内容回答用户问题。"
)

# 进行多轮对话
response1 = chat_engine.chat("什么是 LlamaIndex？")
print(response1.response)

response2 = chat_engine.chat("它与 LangChain 有什么区别？")  # 系统能理解"它"指代 LlamaIndex
print(response2.response)

# 重置对话历史
chat_engine.reset()

from llama_index.core.node_parser import SentenceWindowNodeParser
from llama_index.core.postprocessor import MetadataReplacementPostProcessor

# 创建句子窗口节点解析器
node_parser = SentenceWindowNodeParser.from_defaults(
    window_size=3,              # 前后各保留3个句子
    window_metadata_key="window",
    original_text_metadata_key="original_text"
)

# 在查询时替换为窗口内容
postprocessor = MetadataReplacementPostProcessor(
    target_metadata_key="window"
)

query_engine = index.as_query_engine(
    node_postprocessors=[postprocessor]
)

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

# 创建层次化节点解析器
node_parser = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128]  # 三层结构：父-中-子
)

all_nodes = node_parser.get_nodes_from_documents(documents)
leaf_nodes = get_leaf_nodes(all_nodes)

# 构建文档存储
docstore = SimpleDocumentStore()
docstore.add_documents(all_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)
retriever = AutoMergingRetriever(
    base_retriever,
    storage_context,
    verbose=True
)

from llama_index.core.indices.query.query_transform import HyDEQueryTransform
from llama_index.core.query_engine import TransformQueryEngine

hyde_transform = HyDEQueryTransform(include_original=True)
hyde_query_engine = TransformQueryEngine(
    query_engine,
    query_transform=hyde_transform
)

response = hyde_query_engine.query("解释一下 Transformer 架构的注意力机制")

from llama_index.core.query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector
from llama_index.core.tools import QueryEngineTool

# 定义多个查询引擎工具
vector_tool = QueryEngineTool.from_defaults(
    query_engine=vector_query_engine,
    description="适用于特定问题的精确查询，例如查找具体事实、数据或定义"
)

summary_tool = QueryEngineTool.from_defaults(
    query_engine=summary_query_engine,
    description="适用于需要对文档进行整体总结或概述的查询"
)

# 创建路由查询引擎
router_engine = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[vector_tool, summary_tool],
    verbose=True
)

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import FunctionTool, QueryEngineTool
import json

# 定义自定义工具函数
def calculate_statistics(data: str) -> str:
    """计算一组数字的统计信息，包括均值、最大值和最小值"""
    numbers = json.loads(data)
    return json.dumps({
        "mean": sum(numbers) / len(numbers),
        "max": max(numbers),
        "min": min(numbers),
        "count": len(numbers)
    })

# 将函数包装为工具
stats_tool = FunctionTool.from_defaults(
    fn=calculate_statistics,
    name="calculate_statistics",
    description="用于计算数字列表的统计信息"
)

# 将查询引擎包装为工具
knowledge_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="knowledge_base",
    description="用于查询企业知识库中的信息"
)

# 创建 ReAct 代理
agent = ReActAgent.from_tools(
    tools=[stats_tool, knowledge_tool],
    llm=Settings.llm,
    verbose=True,
    max_iterations=10
)

response = agent.chat("请查询我们产品的销售数据，并计算其统计信息")
print(response.response)

from llama_index.core.agent import FunctionCallingAgent

agent = FunctionCallingAgent.from_tools(
    tools=[stats_tool, knowledge_tool],
    llm=Settings.llm,
    verbose=True,
    allow_parallel_tool_calls=True  # 允许并行调用多个工具
)

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

# 定义专业化子代理
research_agent = ReActAgent.from_tools(
    tools=[web_search_tool, knowledge_tool],
    name="research_agent",
    description="负责信息检索与研究"
)

analysis_agent = ReActAgent.from_tools(
    tools=[stats_tool, chart_tool],
    name="analysis_agent",
    description="负责数据分析与可视化"
)

# 构建多代理工作流
workflow = AgentWorkflow(
    agents=[research_agent, analysis_agent],
    root_agent="research_agent"
)

from llama_index.core.ingestion import IngestionPipeline, IngestionCache
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.storage.kvstore.redis import RedisKVStore

# 配置缓存以避免重复处理已处理的文档
cache = IngestionCache(
    cache=RedisKVStore.from_host_and_port("localhost", 6379),
    collection="ingest_cache"
)

# 构建完整的数据摄入流水线
pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=50),
        TitleExtractor(nodes=5),
        OpenAIEmbedding(),
    ],
    vector_store=vector_store,
    cache=cache
)

# 支持异步执行以提升处理效率
import asyncio
nodes = asyncio.run(pipeline.arun(documents=new_documents))

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,      # 忠实度：回答是否基于检索内容
    RelevancyEvaluator,          # 相关性：检索内容是否与查询相关
    CorrectnessEvaluator,        # 正确性：回答是否准确
    RetrieverEvaluator           # 检索器评估
)
from llama_index.core.evaluation import BatchEvalRunner

# 创建评估器
faithfulness_evaluator = FaithfulnessEvaluator(llm=Settings.llm)
relevancy_evaluator = RelevancyEvaluator(llm=Settings.llm)

# 准备测试数据集
queries = ["什么是 RAG 技术？", "LlamaIndex 有哪些主要组件？"]
expected_answers = ["...", "..."]  # 可选：提供标准答案

# 批量评估
runner = BatchEvalRunner(
    {"faithfulness": faithfulness_evaluator, "relevancy": relevancy_evaluator},
    workers=4
)

eval_results = asyncio.run(
    runner.aevaluate_queries(
        query_engine,
        queries=queries
    )
)

# 输出评估报告
for metric, results in eval_results.items():
    scores = [r.score for r in results]
    print(f"{metric}: 平均分 = {sum(scores)/len(scores):.3f}")

import phoenix as px
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

# 启动 Phoenix 本地追踪服务
px.launch_app()

# 初始化追踪仪器
LlamaIndexInstrumentor().instrument()

# 此后所有的 LlamaIndex 操作将被自动追踪记录
response = query_engine.query("测试查询")

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

# 为每份合同文档创建独立的查询引擎工具
contract_tools = [
    QueryEngineTool.from_defaults(
        query_engine=VectorStoreIndex.from_documents([doc]).as_query_engine(),
        name=f"contract_{i}",
        description=f"用于查询第{i}份合同的条款内容"
    )
    for i, doc in enumerate(contract_documents)
]

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

# 执行跨文档对比查询
response = sub_question_engine.query(
    "比较这三份合同在违约责任条款上的主要差异"
)

import asyncio

# 异步查询
async def process_queries(queries: list[str]):
    tasks = [query_engine.aquery(q) for q in queries]
    responses = await asyncio.gather(*tasks)
    return responses

responses = asyncio.run(process_queries(query_list))

from llama_index.core.storage.kvstore import SimpleKVStore
from llama_index.embeddings.openai import OpenAIEmbedding

# 配置嵌入缓存
embed_model = OpenAIEmbedding()
# LlamaIndex 会自动缓存已计算的嵌入向量

from llama_index.core.llms import OpenAI
from llama_index.core.cache import InMemoryCache

# 配置 LLM 响应缓存
llm = OpenAI(model="gpt-4o")