aiohttp
免费Use when building Python async HTTP services or clients with aiohttp - web server routing, middleware, WebSocket, SSE, streaming, client sessions, pytest-aiohttp testing, or troubleshooting SSL and timeout issues
中文简介正在准备中。当前显示原文说明。
Use when building RAG or LLM applications with LlamaIndex - data loaders, node parsing, vector stores, retrievers and rerankers, query engines, agents and workflows, streaming, or evaluation
中文简介正在准备中。当前显示原文说明。
查看安装方法安装前可以查看交给 AI 代理的指令内容。
Complete guide for building RAG applications with LlamaIndex framework.
LlamaIndex is a data framework for LLM applications, providing tools for data ingestion, indexing, and retrieval.
Key Characteristics:
# Basic installation (v0.14+)
pip install llama-index
# With OpenAI
pip install llama-index-llms-openai
pip install llama-index-embeddings-openai
# With workflow engine support
pip install llama-index-core
# With vector stores
pip install llama-index-vector-stores-chroma
pip install llama-index-vector-stores-pinecone
pip install llama-index-vector-stores-qdrant
# With evaluation
pip install llama-index-llms-openai ragas
llama_index.coreimport os
from llama_index.core import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# Set API key
os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
# Configure global settings
Settings.llm = OpenAI(model="gpt-4o", temperature=0.0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
Settings.chunk_overlap = 50
# Unified imports in v0.14+
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext, load_index_from_storage
# Load documents
documents = SimpleDirectoryReader("./docs").load_data()
# Create index
index = VectorStoreIndex.from_documents(documents)
# Create query engine
query_engine = index.as_query_engine()
# Query
response = query_engine.query("What is the main topic?")
print(response)
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llama_index.embeddings.openai import OpenAIEmbedding
import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore
# Setup ChromaDB
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("my-index")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# Load and index
documents = SimpleDirectoryReader("./docs").load_data()
index = VectorStoreIndex.from_documents(
documents,
storage_context=storage_context,
embed_model=OpenAIEmbedding(),
)
# Persist
storage_context.persist()
# Load from disk
index = VectorStoreIndex.from_vector_store(
vector_store,
embed_model=OpenAIEmbedding(),
)
from llama_index.core import SimpleDirectoryReader, Document
# Load from directory
documents = SimpleDirectoryReader(
input_dir="./docs",
required_exts=[".pdf", ".txt", ".md"],
exclude=["*.tmp"],
recursive=True,
).load_data()
# Load specific files
documents = SimpleDirectoryReader(
input_files=["./file1.pdf", "./file2.txt"]
).load_data()
# Create documents manually
documents = [
Document(text="Content here", metadata={"source": "manual"}),
]
# With metadata extraction
def custom_metadata_func(file_path: str) -> dict:
return {
"file_path": file_path,
"file_name": os.path.basename(file_path),
}
documents = SimpleDirectoryReader(
input_dir="./docs",
file_metadata=custom_metadata_func,
).load_data()
from llama_index.core import Document
from typing import List
class CustomDataReader:
"""Custom data loader."""
def load_data(self, source: str) -> List[Document]:
documents = []
# Load from custom source
# Example: API, database, etc.
data = self._fetch_from_source(source)
for item in data:
doc = Document(
text=item["content"],
metadata={
"source": source,
"id": item["id"],
"timestamp": item["timestamp"],
},
)
documents.append(doc)
return documents
def _fetch_from_source(self, source: str):
# Implement data fetching
pass
# Usage
reader = CustomDataReader()
documents = reader.load_data("api://endpoint")
Chunk boundaries directly determine citation quality and retrieval accuracy. Choose based on your data structure and query patterns.
Use sentence splitting when:
Use semantic chunking when:
# ❌ PITFALL: Ignoring semantic boundaries
splitter = SentenceSplitter(
chunk_size=512,
chunk_overlap=50,
# No paragraph or section awareness
)
# ✅ BETTER: Respect document structure
splitter = SentenceSplitter(
chunk_size=512,
chunk_overlap=50,
paragraph_separator="\n\n", # Split on paragraphs first
secondary_chunking_regex=r"\n#{1,6} ", # Don't break across headings
)
Common overlap mistakes:
Metadata enables filtering and improves retrieval precision. Attach these at ingest:
| Metadata Field | Why It Matters |
|---|---|
source (file path) | Cite exact document when answering |
section_heading | Preserve document hierarchy in context |
page_number | Critical for PDFs and scanned documents |
document_date | Filter by recency for time-sensitive queries |
tenant_id / PROJECT_KEY | Multi-tenant isolation via metadata filters |
from llama_index.core import Document
def load_with_metadata(file_path: str) -> List[Document]:
"""Load documents with essential metadata for retrieval."""
# Parse file to extract structure
content, headings, page_numbers = parse_document(file_path)
return [
Document(
text=chunk.text,
metadata={
"source": file_path,
"section": chunk.section,
"page": page_numbers[chunk.start_line],
"tenant_id": "my-tenant", # Replace with actual tenant
},
)
for chunk in split_by_structure(content, headings)
]
Never guess whether a chunking change helped. Measure:
# Simple hit rate evaluation
def evaluate_chunking(queries_with_relevant_chunks, retriever, top_k=5):
hits = 0
for query, relevant_ids in queries_with_relevant_chunks:
nodes = retriever.retrieve(query)
retrieved_ids = {n.node.node_id for n in nodes}
if relevant_ids & retrieved_ids: # Any overlap = hit
hits += 1
return hits / len(queries_with_relevant_chunks)
Vector store setup follows the same pattern across backends:
from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# 1. Initialize backend client
db = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = db.get_or_create_collection("my-index")
# 2. Wrap as LlamaIndex vector store
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
# 3. Create index with storage context
index = VectorStoreIndex.from_documents(documents, storage_context=storage_context)
Backend-specific notes:
PersistentClient for local storage, EphemeralClient for testingcollection_name and ensure vector size matches embed_modelFor detailed setup per backend, see references/retrieval.md.
Symptom-first guide to fixing retrieval problems.
Symptom: Response references information from the wrong document or section.
Check:
source and section metadata?Fix:
# Apply metadata filters to narrow retrieval
query_engine = index.as_query_engine(
filters=MetadataFilter(
key="tenant_id", value="my-tenant" # Replace with actual filter
)
)
# Or filter at retriever level
retriever = index.as_retriever(
filters=MetadataFilter(key="source", value="specific-file.pdf")
)
Symptom: Query returns chunks that don't match the question intent.
Check:
similarity_top_k too low?Fix:
# Ensure consistent embedding model
from llama_index.embeddings.openai import OpenAIEmbedding
embed_model = OpenAIEmbedding(model="text-embedding-3-small")
# At index time
index = VectorStoreIndex.from_documents(documents, embed_model=embed_model)
# At query time (must match)
query_engine = index.as_query_engine(
embed_model=embed_model, # Same model as index time
similarity_top_k=10, # Retrieve more before reranking
)
Symptom: Responses lack specificity, sound like boilerplate.
Check:
similarity_top_k too low (<5)?Fix:
from llama_index.core.postprocessor import SentenceTransformerRerank
# Retrieve more, then rerank
query_engine = index.as_query_engine(
similarity_top_k=20,
node_postprocessors=[
SentenceTransformerRerank(
model="cross-encoder/ms-marco-MiniLM-L-6-v2",
top_n=5,
),
],
)
Symptom: Keyword + vector search gives unpredictable results.
Check:
Fix:
from llama_index.core.retrievers import QueryFusionRetriever
# Tune query fusion parameters
fusion_retriever = QueryFusionRetriever(
retrievers=[vector_retriever, keyword_retriever],
similarity_top_k=10,
num_queries=3,
mode="reciprocal_rerank", # Or "weighted_sum"
)
# For strict matching, use AND mode
hybrid_retriever = HybridRetriever(
vector_index, keyword_index, mode="AND" # Only return nodes in both
)
Advanced topics are split into reference files loaded on demand:
# ❌ BAD: Too small chunks lose context
splitter = SentenceSplitter(chunk_size=64)
# ❌ BAD: Too large chunks dilute relevance
splitter = SentenceSplitter(chunk_size=8192)
# ✅ GOOD: Balanced chunk size
splitter = SentenceSplitter(
chunk_size=512, # For embeddings
chunk_overlap=50, # ~10% overlap
)
# ❌ BAD: No reranking, few results
query_engine = index.as_query_engine(similarity_top_k=3)
# ✅ GOOD: Retrieve more, rerank
query_engine = index.as_query_engine(
similarity_top_k=20,
node_postprocessors=[
SentenceTransformerRerank(top_n=5),
],
)
# ❌ BAD: Load all documents at once
documents = SimpleDirectoryReader("./huge_folder").load_data()
# ✅ GOOD: Process in batches
from llama_index.core import StorageContext
for batch in document_batches:
nodes = splitter.get_nodes_from_documents(batch)
index.insert_nodes(nodes)
# ❌ BAD: Index created with different embedding
# Pinecone index: 1536 dimensions
# Using: 768 dimension embeddings
# ✅ GOOD: Match dimensions
embed_model = OpenAIEmbedding(model="text-embedding-3-small") # 1536 dims
# Create Pinecone index with same dimensions
from llama_index.core import (
VectorStoreIndex,
SimpleDirectoryReader,
Document,
Settings,
StorageContext,
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# Basic RAG
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("query")
# With reranking
query_engine = index.as_query_engine(
similarity_top_k=20,
node_postprocessors=[reranker],
)
# Streaming
query_engine = index.as_query_engine(streaming=True)
for token in query_engine.query("query").response_gen:
print(token)
# Chat
chat_engine = index.as_chat_engine()
response = chat_engine.chat("message")
# Agent
agent = ReActAgent.from_tools(tools, llm=llm)
response = agent.chat("message")
还没有评价。欢迎分享你的使用感受。
无需登录即可发表。欢迎分享使用感受、提问,或说明在哪些环境下无法运行。
还没有评论。来写第一条吧。
Use when building Python async HTTP services or clients with aiohttp - web server routing, middleware, WebSocket, SSE, streaming, client sessions, pytest-aiohttp testing, or troubleshooting SSL and timeout issues
中文简介正在准备中。当前显示原文说明。
Use when doing structural code search and rewriting - ast-grep linting, refactoring, multi-language patterns
中文简介正在准备中。当前显示原文说明。
Use when building GBA games with the BPCore Lua engine - entity, sprite and tilemap functions, SRAM save and load, link cable multiplayer protocol, camera and scrolling, or optimization patterns
中文简介正在准备中。当前显示原文说明。
Use when running background tasks with Celery - worker and broker configuration (Redis, RabbitMQ), task routing by name vs queue, chains/groups/chords, retry patterns (autoretry_for, retry_backoff), acks_late semantics, failure detection, and monitoring with Flower
中文简介正在准备中。当前显示原文说明。
Use when building Django applications - security hardening, authentication and permissions, ORM optimization, PostgreSQL features, Django 6.0, migrations, testing, and ecosystem libraries
中文简介正在准备中。当前显示原文说明。
Use when customizing Django Admin - save_formset, get_search_results, formsets, queryset optimization, db_index, custom URLs
中文简介正在准备中。当前显示原文说明。