tessl/pypi-llama-index
Interface between LLMs and your data for building retrieval-augmented generation (RAG) applications
—
Pending
retrievers.mddocs/
Retrievers
Advanced retrieval strategies including fusion, hierarchical, and routing approaches for sophisticated document retrieval patterns and enhanced context gathering.
Capabilities
Base Retriever Interface
Foundation class for all retrieval strategies with standardized interface and configuration options.
class BaseRetriever:
"""
Base class for document retrievers.
Args:
callback_manager: Callback manager for events
object_map: Object mapping for node resolution
verbose: Enable verbose logging
"""
def __init__(
self,
callback_manager=None,
object_map=None,
verbose=False,
**kwargs
): ...
def retrieve(self, str_or_query_bundle):
"""
Retrieve relevant nodes for query.
Args:
str_or_query_bundle: Query string or QueryBundle object
Returns:
List[NodeWithScore]: Retrieved nodes with relevance scores
"""
async def aretrieve(self, str_or_query_bundle):
"""Async version of retrieve method."""
def get_modules(self):
"""Get retriever modules for introspection."""Vector Index Retrievers
Core retrievers for vector-based semantic similarity search with various configuration options.
class VectorIndexRetriever(BaseRetriever):
"""
Vector index retriever for semantic similarity search.
Args:
index: Vector store index instance
similarity_top_k: Number of top similar nodes to retrieve
vector_store_query_mode: Query mode ("default", "sparse", "hybrid")
alpha: Alpha parameter for hybrid search
doc_ids: Filter by specific document IDs
filters: Metadata filters for retrieval
"""
def __init__(
self,
index,
similarity_top_k=2,
vector_store_query_mode="default",
alpha=None,
doc_ids=None,
filters=None,
**kwargs
): ...
class VectorIndexAutoRetriever(BaseRetriever):
"""
Auto-retriever with dynamic query analysis and metadata filtering.
Args:
index: Vector store index
vector_store_info: Metadata about vector store schema
similarity_top_k: Number of nodes to retrieve
empty_query_top_k: Fallback retrieval count for empty queries
max_top_k_comparisons: Maximum comparisons for auto-selection
"""
def __init__(
self,
index,
vector_store_info,
similarity_top_k=2,
empty_query_top_k=10,
max_top_k_comparisons=10,
**kwargs
): ...Vector Retriever Usage Example:
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.vector_stores import MetadataFilter, MetadataFilters
# Basic vector retrieval
vector_retriever = VectorIndexRetriever(
index=vector_index,
similarity_top_k=5
)
# Filtered retrieval
filters = MetadataFilters(filters=[
MetadataFilter(key="document_type", value="research_paper"),
MetadataFilter(key="year", value=2023, operator=">=")
])
filtered_retriever = VectorIndexRetriever(
index=vector_index,
similarity_top_k=3,
filters=filters
)
# Retrieve documents
nodes = vector_retriever.retrieve("machine learning applications")
for node in nodes:
print(f"Score: {node.score}, Content: {node.node.get_content()[:100]}...")Hierarchical and Auto-Merging Retrievers
Advanced retrievers that work with hierarchical document structures and automatic chunk merging.
class AutoMergingRetriever(BaseRetriever):
"""
Auto-merging retriever for hierarchical document chunks.
Retrieves fine-grained chunks and automatically merges them into larger
context windows when consecutive chunks are retrieved.
Args:
vector_retriever: Base vector retriever for initial retrieval
storage_context: Storage context for document resolution
similarity_top_k: Number of initial chunks to retrieve
repack: Whether to repack merged nodes
"""
def __init__(
self,
vector_retriever,
storage_context,
similarity_top_k=2,
repack=True,
**kwargs
): ...
class RecursiveRetriever(BaseRetriever):
"""
Recursive retriever for nested document structures.
Recursively retrieves from hierarchical indices, following references
and building comprehensive context through multiple retrieval levels.
Args:
root_id: Root node identifier for recursive traversal
retriever_dict: Dictionary mapping node IDs to retrievers
query_transform_fn: Function to transform queries at each level
max_depth: Maximum recursion depth
"""
def __init__(
self,
root_id,
retriever_dict,
query_transform_fn=None,
max_depth=10,
**kwargs
): ...Hierarchical Retriever Example:
from llama_index.core.retrievers import AutoMergingRetriever
from llama_index.core.node_parser import HierarchicalNodeParser
from llama_index.core import VectorStoreIndex
# Create hierarchical document structure
node_parser = HierarchicalNodeParser.from_defaults(
chunk_sizes=[2048, 512, 128] # Multi-level chunking
)
# Build index with hierarchical nodes
documents = SimpleDirectoryReader("data").load_data()
nodes = node_parser.get_nodes_from_documents(documents)
index = VectorStoreIndex(nodes, storage_context=storage_context)
# Create auto-merging retriever
base_retriever = VectorIndexRetriever(
index=index,
similarity_top_k=6
)
auto_merging_retriever = AutoMergingRetriever(
vector_retriever=base_retriever,
storage_context=storage_context,
similarity_top_k=6,
repack=True
)
# Retrieve with automatic merging
merged_nodes = auto_merging_retriever.retrieve(
"What are the latest developments in AI?"
)Query Fusion Retriever
Advanced retriever that generates multiple query variations and fuses results for comprehensive retrieval.
class QueryFusionRetriever(BaseRetriever):
"""
Query fusion retriever for comprehensive document retrieval.
Generates multiple query variations, retrieves results for each,
and fuses the results using advanced ranking algorithms.
Args:
retrievers: List of retrievers to query
similarity_top_k: Number of nodes per retrieval
num_queries: Number of query variations to generate
mode: Fusion mode ("reciprocal_rerank", "relative_score", "simple")
use_async: Enable asynchronous retrieval
query_gen_prompt: Template for query generation
"""
def __init__(
self,
retrievers,
similarity_top_k=2,
num_queries=4,
mode="reciprocal_rerank",
use_async=True,
query_gen_prompt=None,
**kwargs
): ...Query Fusion Example:
from llama_index.core.retrievers import QueryFusionRetriever
# Create multiple retrievers for fusion
vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=5)
keyword_retriever = KeywordTableRetriever(index=keyword_index, similarity_top_k=5)
# Create fusion retriever
fusion_retriever = QueryFusionRetriever(
retrievers=[vector_retriever, keyword_retriever],
similarity_top_k=3,
num_queries=4, # Generate 4 query variations
mode="reciprocal_rerank",
use_async=True
)
# Retrieve with query fusion
fused_results = fusion_retriever.retrieve("artificial intelligence applications")
# Results are automatically ranked and deduplicated
for node in fused_results:
print(f"Fused Score: {node.score}")Router Retriever
Intelligent retriever that routes queries to appropriate specialized retrievers based on query analysis.
class RouterRetriever(BaseRetriever):
"""
Router retriever for intelligent retrieval routing.
Analyzes queries and routes them to the most appropriate retriever
from a collection of specialized retrievers.
Args:
selector: Selector for choosing appropriate retriever
retriever_tools: List of retriever tools with descriptions
default_retriever: Fallback retriever if routing fails
"""
def __init__(
self,
selector,
retriever_tools,
default_retriever=None,
**kwargs
): ...
class BaseSelector:
"""Base class for retriever selection logic."""
def select(self, choices, query):
"""Select appropriate retriever based on query."""Router Retriever Example:
from llama_index.core.retrievers import RouterRetriever
from llama_index.core.tools import RetrieverTool
from llama_index.core.selectors import LLMSingleSelector
# Define specialized retrievers
technical_retriever = VectorIndexRetriever(
index=technical_index,
similarity_top_k=5
)
business_retriever = VectorIndexRetriever(
index=business_index,
similarity_top_k=5
)
# Create retriever tools with descriptions
retriever_tools = [
RetrieverTool.from_defaults(
retriever=technical_retriever,
description="Technical documentation and API references"
),
RetrieverTool.from_defaults(
retriever=business_retriever,
description="Business processes and company policies"
)
]
# Create router with LLM-based selection
selector = LLMSingleSelector.from_defaults()
router_retriever = RouterRetriever(
selector=selector,
retriever_tools=retriever_tools,
default_retriever=technical_retriever
)
# Route queries automatically
tech_results = router_retriever.retrieve("API authentication methods")
business_results = router_retriever.retrieve("expense reporting procedures")Transform Retriever
Retriever wrapper that applies transformations to queries or retrieved nodes for enhanced processing.
class TransformRetriever(BaseRetriever):
"""
Transform retriever for query and node processing.
Applies transformations to queries before retrieval and/or to
retrieved nodes before returning results.
Args:
retriever: Base retriever to wrap
query_transform: Function to transform queries
node_transform: Function to transform retrieved nodes
transform_metadata: Whether to transform node metadata
"""
def __init__(
self,
retriever,
query_transform=None,
node_transform=None,
transform_metadata=True,
**kwargs
): ...Transform Retriever Example:
from llama_index.core.retrievers import TransformRetriever
def expand_query(query_str):
"""Expand query with synonyms and related terms."""
# Custom query expansion logic
synonyms = {
"AI": "artificial intelligence machine learning",
"ML": "machine learning artificial intelligence"
}
expanded = query_str
for term, expansion in synonyms.items():
if term in query_str:
expanded = f"{query_str} {expansion}"
return expanded
def enhance_nodes(nodes):
"""Add custom metadata to retrieved nodes."""
for node in nodes:
# Add relevance category
content = node.node.get_content().lower()
if "technical" in content:
node.node.metadata["category"] = "technical"
elif "business" in content:
node.node.metadata["category"] = "business"
return nodes
# Create transform retriever
transform_retriever = TransformRetriever(
retriever=base_retriever,
query_transform=expand_query,
node_transform=enhance_nodes
)
# Queries and results are automatically transformed
results = transform_retriever.retrieve("AI applications")Specialized Index Retrievers
Retrievers designed for specific index types with optimized retrieval strategies.
class KeywordTableRetriever(BaseRetriever):
"""
Keyword table retriever for exact keyword matching.
Args:
index: Keyword table index
mode: Retrieval mode ("DEFAULT", "SIMPLE", "RAKE")
max_chunks_per_query: Maximum chunks to return
"""
def __init__(
self,
index,
mode="DEFAULT",
max_chunks_per_query=10,
**kwargs
): ...
class KnowledgeGraphRAGRetriever(BaseRetriever):
"""
Knowledge graph RAG retriever for graph-based retrieval.
Args:
storage_context: Storage context with graph store
service_context: Service context (deprecated)
entity_extract_fn: Function to extract entities from queries
synonym_expand_fn: Function to expand entity synonyms
retrieval_depth: Graph traversal depth
max_entities: Maximum entities to extract per query
"""
def __init__(
self,
storage_context,
service_context=None,
entity_extract_fn=None,
synonym_expand_fn=None,
retrieval_depth=2,
max_entities=10,
**kwargs
): ...
class TreeSelectLeafRetriever(BaseRetriever):
"""
Tree index retriever that selects specific leaf nodes.
Args:
index: Tree index instance
child_branch_factor: Number of child branches to explore
"""
def __init__(
self,
index,
child_branch_factor=1,
**kwargs
): ...Postprocessor Integration
Retrievers with integrated postprocessing for result refinement and reranking.
class BasePostprocessor:
"""Base class for retrieval postprocessors."""
def postprocess_nodes(self, nodes, query_bundle=None):
"""Postprocess retrieved nodes."""
class SimilarityPostprocessor(BasePostprocessor):
"""
Similarity-based postprocessor for score filtering.
Args:
similarity_cutoff: Minimum similarity score threshold
"""
def __init__(self, similarity_cutoff=0.7): ...
class KeywordNodePostprocessor(BasePostprocessor):
"""
Keyword-based postprocessor for content filtering.
Args:
required_keywords: Keywords that must be present
exclude_keywords: Keywords that exclude nodes
"""
def __init__(
self,
required_keywords=None,
exclude_keywords=None
): ...
class SentenceEmbeddingOptimizer(BasePostprocessor):
"""
Sentence embedding optimizer for context optimization.
Args:
embed_model: Embedding model for optimization
percentile_cutoff: Percentile cutoff for sentence selection
threshold_cutoff: Absolute threshold for sentence selection
"""
def __init__(
self,
embed_model=None,
percentile_cutoff=None,
threshold_cutoff=None
): ...Postprocessor Usage Example:
from llama_index.core.postprocessor import (
SimilarityPostprocessor,
KeywordNodePostprocessor,
SentenceEmbeddingOptimizer
)
# Create retriever with postprocessors
retriever = VectorIndexRetriever(
index=index,
similarity_top_k=10 # Retrieve more, then filter
)
# Chain multiple postprocessors
postprocessors = [
SimilarityPostprocessor(similarity_cutoff=0.7),
KeywordNodePostprocessor(required_keywords=["machine learning"]),
SentenceEmbeddingOptimizer(percentile_cutoff=0.5)
]
# Use with query engine
query_engine = RetrieverQueryEngine.from_args(
retriever=retriever,
node_postprocessors=postprocessors
)
response = query_engine.query("machine learning applications")Ensemble and Multi-Modal Retrievers
Advanced retrievers for combining multiple retrieval strategies and handling multi-modal content.
class EnsembleRetriever(BaseRetriever):
"""
Ensemble retriever combining multiple retrieval strategies.
Args:
retrievers: List of retrievers to ensemble
weights: Weights for each retriever (optional)
combine_mode: How to combine results ("weighted_sum", "rank_fusion")
"""
def __init__(
self,
retrievers,
weights=None,
combine_mode="rank_fusion",
**kwargs
): ...
class MultiModalRetriever(BaseRetriever):
"""
Multi-modal retriever for text and image content.
Args:
vector_retriever: Text vector retriever
image_retriever: Image vector retriever
mode: Retrieval mode ("text", "image", "both")
fusion_strategy: Strategy for combining modalities
"""
def __init__(
self,
vector_retriever,
image_retriever=None,
mode="both",
fusion_strategy="score_fusion",
**kwargs
): ...Custom Retriever Implementation
Framework for implementing custom retrieval strategies with full control over the retrieval process.
class CustomRetriever(BaseRetriever):
"""
Custom retriever implementation template.
Args:
custom_retrieve_fn: Custom retrieval function
**kwargs: BaseRetriever arguments
"""
def __init__(self, custom_retrieve_fn, **kwargs):
super().__init__(**kwargs)
self._custom_retrieve_fn = custom_retrieve_fn
def _retrieve(self, query_bundle):
"""Custom retrieval logic."""
return self._custom_retrieve_fn(query_bundle)Custom Retriever Example:
from llama_index.core.retrievers import BaseRetriever
from llama_index.core.schema import NodeWithScore
class SemanticCacheRetriever(BaseRetriever):
"""Custom retriever with semantic caching."""
def __init__(self, base_retriever, cache_threshold=0.95, **kwargs):
super().__init__(**kwargs)
self.base_retriever = base_retriever
self.cache = {} # Simple in-memory cache
self.cache_threshold = cache_threshold
def _retrieve(self, query_bundle):
query_str = query_bundle.query_str
# Check semantic cache
cached_result = self._check_cache(query_str)
if cached_result:
return cached_result
# Retrieve from base retriever
nodes = self.base_retriever.retrieve(query_bundle)
# Cache results
self._cache_results(query_str, nodes)
return nodes
def _check_cache(self, query_str):
"""Check if semantically similar query exists in cache."""
# Implementation for semantic similarity check
# Return cached results if similarity > threshold
return None # Placeholder
def _cache_results(self, query_str, nodes):
"""Cache retrieval results."""
self.cache[query_str] = nodes
# Use custom retriever
cached_retriever = SemanticCacheRetriever(
base_retriever=vector_retriever,
cache_threshold=0.95
)Install with Tessl CLI
npx tessl i tessl/pypi-llama-index