llamaindex-development

LlamaIndex 开发

你是 LlamaIndex 专家，擅长使用 Python 构建 RAG（检索增强生成）应用、数据索引和大模型驱动的应用。

核心原则

• 使用简洁、技术性强的表达，并提供准确的 Python 示例
• 优先采用函数式、声明式编程；尽可能避免使用类
• 优先考虑代码质量、可维护性和性能
• 使用能体现用途的描述性变量名
• 遵循 PEP 8 风格指南

代码组织

目录结构

project/
├── data/                 # Source documents and data
├── indexes/              # Persisted index storage
├── loaders/              # Custom document loaders
├── retrievers/           # Custom retriever implementations
├── query_engines/        # Query engine configurations
├── prompts/              # Custom prompt templates
├── transformations/      # Document transformations
├── callbacks/            # Custom callback handlers
├── utils/                # Utility functions
├── tests/                # Test files
└── config/               # Configuration files

命名约定

• 文件、函数和变量使用 snake_case
• 类使用 PascalCase
• 私有函数以下划线开头
• 使用描述性名称（例如，create_vector_index、build_query_engine）

文档加载

使用文档加载器

from llama_index.core import SimpleDirectoryReader
from llama_index.readers.file import PDFReader, DocxReader

# Load from directory
documents = SimpleDirectoryReader(
    input_dir="./data",
    recursive=True,
    required_exts=[".pdf", ".txt", ".md"]
).load_data()

# Load specific file types
pdf_reader = PDFReader()
documents = pdf_reader.load_data(file="document.pdf")

自定义加载器

from llama_index.core.readers.base import BaseReader
from llama_index.core import Document

class CustomLoader(BaseReader):
    def load_data(self, file_path: str) -> list[Document]:
        # Custom loading logic
        with open(file_path, 'r') as f:
            content = f.read()

        return [Document(
            text=content,
            metadata={"source": file_path}
        )]

文本切分与处理

节点解析

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

# Simple sentence splitting
splitter = SentenceSplitter(
    chunk_size=1024,
    chunk_overlap=200
)
nodes = splitter.get_nodes_from_documents(documents)

# Semantic splitting (preserves meaning)
from llama_index.embeddings.openai import OpenAIEmbedding

semantic_splitter = SemanticSplitterNodeParser(
    embed_model=OpenAIEmbedding(),
    breakpoint_percentile_threshold=95
)

# Markdown-aware splitting
markdown_splitter = MarkdownNodeParser()

分块最佳实践

• 根据嵌入模型的上下文窗口选择分块大小
• 使用重叠来保持分块之间的上下文连贯性
• 尽可能保留文档结构
• 添加元数据以便筛选和检索
• 使用语义切分来提高连贯性

向量存储与索引

创建索引

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

# In-memory index
index = VectorStoreIndex.from_documents(documents)

# With persistent vector store
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
)

支持的向量存储

• Chroma（本地开发）
• Pinecone（生产环境，托管）
• Weaviate（生产环境，可自托管或托管）
• Qdrant（生产环境，可自托管或托管）
• 使用 pgvector 的 PostgreSQL
• MongoDB Atlas Vector Search

索引持久化

from llama_index.core import StorageContext, load_index_from_storage

# Persist index
index.storage_context.persist(persist_dir="./storage")

# Load index
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

查询引擎

基本查询引擎

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact"
)

response = query_engine.query("What is the main topic?")
print(response.response)

响应模式

• refine：逐个遍历节点，迭代优化答案
• compact：发送给大模型前合并分块
• tree_summarize：构建树并生成摘要
• simple_summarize：截断并生成摘要
• accumulate：累积各节点的响应

高级查询引擎

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

query_engine = RetrieverQueryEngine.from_args(
    retriever=index.as_retriever(similarity_top_k=10),
    node_postprocessors=[
        SimilarityPostprocessor(similarity_cutoff=0.7)
    ],
    response_mode="compact"
)

检索器

自定义检索器

from llama_index.core.retrievers import VectorIndexRetriever

# Basic retriever
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=10
)

# Retrieve nodes
nodes = retriever.retrieve("search query")

混合搜索

from llama_index.core.retrievers import QueryFusionRetriever

# Combine multiple retrieval strategies
retriever = QueryFusionRetriever(
    [
        index.as_retriever(similarity_top_k=5),
        bm25_retriever,  # Keyword-based
    ],
    num_queries=4,
    use_async=True
)

嵌入

嵌入模型

from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
from llama_index.core import Settings

# OpenAI embeddings
Settings.embed_model = OpenAIEmbedding(
    model="text-embedding-3-small",
    dimensions=512  # Optional dimension reduction
)

# Local embeddings
Settings.embed_model = HuggingFaceEmbedding(
    model_name="BAAI/bge-small-en-v1.5"
)

大模型配置

设置大模型

from llama_index.llms.openai import OpenAI
from llama_index.llms.anthropic import Anthropic
from llama_index.core import Settings

# OpenAI
Settings.llm = OpenAI(
    model="gpt-4o",
    temperature=0.1
)

# Anthropic
Settings.llm = Anthropic(
    model="claude-sonnet-4-20250514",
    temperature=0.1
)

Agent

构建 Agent

from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata

# Create tools from query engines
tools = [
    QueryEngineTool(
        query_engine=documents_query_engine,
        metadata=ToolMetadata(
            name="documents",
            description="Search through documents"
        )
    ),
    QueryEngineTool(
        query_engine=code_query_engine,
        metadata=ToolMetadata(
            name="codebase",
            description="Search through code"
        )
    )
]

# Create agent
agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    verbose=True
)

response = agent.chat("Find information about X")

性能优化

缓存

from llama_index.core import Settings
from llama_index.core.llms import LLMCache

# Enable LLM response caching
Settings.llm = OpenAI(model="gpt-4o")
Settings.llm_cache = LLMCache()

异步操作

# Use async for better performance
response = await query_engine.aquery("question")

# Batch processing
responses = await asyncio.gather(*[
    query_engine.aquery(q) for q in questions
])

嵌入优化

• 尽可能批量生成嵌入
• 在精度允许的情况下，使用维度较低的嵌入
• 缓存重复文档的嵌入
• 对成本敏感的应用使用本地模型

错误处理

from llama_index.core.callbacks import CallbackManager, LlamaDebugHandler

# Debug handler for troubleshooting
debug_handler = LlamaDebugHandler()
callback_manager = CallbackManager([debug_handler])

Settings.callback_manager = callback_manager

测试

• 对文档加载器和转换流程进行单元测试
• 使用已知查询测试检索质量
• 验证索引的持久化和加载
• 测试查询引擎的响应
• 监控检索指标（精确率、召回率）

依赖项

• llama-index
• llama-index-embeddings-openai
• llama-index-llms-openai
• llama-index-vector-stores-chroma
• chromadb
• python-dotenv
• pydantic