Interface between LLMs and your data
—
Components for processing and refining retrieved results, including similarity filtering, reranking, metadata replacement, and recency scoring. Postprocessors enhance retrieval quality by applying various refinement strategies to improve relevance and remove irrelevant content.
Foundation interface for all postprocessing operations with standardized node processing methods.
class BaseNodePostprocessor:
"""
Base interface for node postprocessing operations.
Parameters:
- callback_manager: Optional[CallbackManager], callback management system
"""
def __init__(self, callback_manager: Optional[CallbackManager] = None): ...
def postprocess_nodes(
self,
nodes: List[NodeWithScore],
query_bundle: Optional[QueryBundle] = None
) -> List[NodeWithScore]:
"""
Process and refine retrieved nodes.
Parameters:
- nodes: List[NodeWithScore], nodes to postprocess
- query_bundle: Optional[QueryBundle], original query for context
Returns:
- List[NodeWithScore], processed and refined nodes
"""
def _postprocess_nodes(
self,
nodes: List[NodeWithScore],
query_bundle: Optional[QueryBundle] = None
) -> List[NodeWithScore]:
"""Internal postprocessing method to be implemented by subclasses."""Filters nodes based on relevance scores and similarity thresholds to remove low-quality results.
class SimilarityPostprocessor(BaseNodePostprocessor):
"""
Postprocessor that filters nodes based on similarity score thresholds.
Parameters:
- similarity_cutoff: Optional[float], minimum similarity score to retain nodes
"""
def __init__(self, similarity_cutoff: Optional[float] = None): ...Filters nodes based on keyword inclusion or exclusion criteria for content-based filtering.
class KeywordNodePostprocessor(BaseNodePostprocessor):
"""
Postprocessor for keyword-based node filtering.
Parameters:
- required_keywords: Optional[List[str]], keywords that must be present
- exclude_keywords: Optional[List[str]], keywords that must not be present
- lang: str, language for keyword matching
"""
def __init__(
self,
required_keywords: Optional[List[str]] = None,
exclude_keywords: Optional[List[str]] = None,
lang: str = "en"
): ...Enhances nodes by adding adjacent or related content for better context understanding.
class PrevNextNodePostprocessor(BaseNodePostprocessor):
"""
Postprocessor that adds previous and next nodes for enhanced context.
Parameters:
- docstore: BaseDocumentStore, document store for node relationships
- num_nodes: int, number of previous/next nodes to include
- mode: str, inclusion mode (previous, next, or both)
"""
def __init__(
self,
docstore: BaseDocumentStore,
num_nodes: int = 1,
mode: str = "both"
): ...
class AutoPrevNextNodePostprocessor(BaseNodePostprocessor):
"""
Automatic previous/next node inclusion with intelligent boundary detection.
Parameters:
- docstore: BaseDocumentStore, document store for node relationships
- num_nodes: int, number of nodes to include in each direction
"""
def __init__(
self,
docstore: BaseDocumentStore,
num_nodes: int = 1
): ...Reorders and optimizes nodes for long context scenarios to improve model performance.
class LongContextReorder(BaseNodePostprocessor):
"""
Reorders nodes to optimize performance in long context scenarios.
Long context reordering places the most relevant information at the beginning
and end of the context window where language models pay more attention.
"""
def __init__(self): ...Applies recency-based scoring and filtering to prioritize recent or time-relevant content.
class FixedRecencyPostprocessor(BaseNodePostprocessor):
"""
Postprocessor that applies fixed recency scoring based on date metadata.
Parameters:
- top_k: int, number of top recent nodes to return
- date_key: str, metadata key containing date information
- service_context: Optional[ServiceContext], service context for processing
"""
def __init__(
self,
top_k: int = 1,
date_key: str = "date",
service_context: Optional[ServiceContext] = None
): ...
class EmbeddingRecencyPostprocessor(BaseNodePostprocessor):
"""
Recency postprocessor using embedding-based similarity for temporal relevance.
Parameters:
- embed_model: Optional[BaseEmbedding], embedding model for similarity computation
- similarity_cutoff: float, minimum similarity threshold
- date_key: str, metadata key containing date information
- service_context: Optional[ServiceContext], service context for processing
"""
def __init__(
self,
embed_model: Optional[BaseEmbedding] = None,
similarity_cutoff: float = 0.7,
date_key: str = "date",
service_context: Optional[ServiceContext] = None
): ...
class TimeWeightedPostprocessor(BaseNodePostprocessor):
"""
Time-weighted relevance scoring that balances content relevance with recency.
Parameters:
- time_decay: float, decay factor for time-based scoring
- time_access_refresh: bool, whether to refresh access times
- top_k: int, number of top nodes to return
"""
def __init__(
self,
time_decay: float = 0.99,
time_access_refresh: bool = True,
top_k: int = 1
): ...Removes or masks personally identifiable information (PII) and sensitive data from retrieved content.
class PIINodePostprocessor(BaseNodePostprocessor):
"""
Postprocessor for detecting and removing personally identifiable information.
Parameters:
- pii_node_info_key: str, metadata key for storing PII information
- pii_str_tmpl: str, template for PII replacement strings
- service_context: Optional[ServiceContext], service context for processing
"""
def __init__(
self,
pii_node_info_key: str = "__pii_node_info__",
pii_str_tmpl: str = "[PII_REMOVED]",
service_context: Optional[ServiceContext] = None
): ...
class NERPIINodePostprocessor(BaseNodePostprocessor):
"""
Named Entity Recognition-based PII detection and removal postprocessor.
Parameters:
- pii_node_info_key: str, metadata key for PII information
- pii_str_tmpl: str, template for PII replacement
- ner_model_name: str, name of NER model to use
- service_context: Optional[ServiceContext], service context for processing
"""
def __init__(
self,
pii_node_info_key: str = "__pii_node_info__",
pii_str_tmpl: str = "[PII_REMOVED]",
ner_model_name: str = "StanfordAIMI/stanford-deidentifier-base",
service_context: Optional[ServiceContext] = None
): ...Advanced reranking using language models and specialized algorithms to improve result ordering.
class LLMRerank(BaseNodePostprocessor):
"""
LLM-based reranking postprocessor for improved result ordering.
Parameters:
- choice_batch_size: int, batch size for LLM processing
- top_n: int, number of top nodes to return after reranking
- service_context: Optional[ServiceContext], service context for LLM operations
- choice_select_prompt: Optional[PromptTemplate], prompt for node selection
- choice_batch_select_prompt: Optional[PromptTemplate], prompt for batch selection
- llm: Optional[LLM], language model for reranking
"""
def __init__(
self,
choice_batch_size: int = 10,
top_n: int = 10,
service_context: Optional[ServiceContext] = None,
choice_select_prompt: Optional[PromptTemplate] = None,
choice_batch_select_prompt: Optional[PromptTemplate] = None,
llm: Optional[LLM] = None
): ...
class StructuredLLMRerank(BaseNodePostprocessor):
"""
Structured LLM reranking with explicit scoring criteria and rationale.
Parameters:
- llm: Optional[LLM], language model for structured reranking
- top_n: int, number of top nodes to return
- choice_batch_size: int, batch size for processing
"""
def __init__(
self,
llm: Optional[LLM] = None,
top_n: int = 10,
choice_batch_size: int = 10
): ...
class SentenceTransformerRerank(BaseNodePostprocessor):
"""
Sentence transformer-based reranking for semantic similarity.
Parameters:
- model: str, sentence transformer model name
- top_n: int, number of top nodes to return
- device: Optional[str], device for model computation (cpu, cuda)
- keep_retrieval_score: bool, whether to preserve original retrieval scores
"""
def __init__(
self,
model: str = "cross-encoder/ms-marco-MiniLM-L-2-v2",
top_n: int = 10,
device: Optional[str] = None,
keep_retrieval_score: bool = False
): ...Optimizes embedding-based operations and enhances semantic understanding of retrieved content.
class SentenceEmbeddingOptimizer(BaseNodePostprocessor):
"""
Optimizer for sentence embeddings to improve semantic retrieval quality.
Parameters:
- embed_model: Optional[BaseEmbedding], embedding model for optimization
- percentile_cutoff: Optional[float], percentile cutoff for optimization
- threshold_cutoff: Optional[float], absolute threshold for optimization
- mode: str, optimization mode (percentile, threshold, or auto)
"""
def __init__(
self,
embed_model: Optional[BaseEmbedding] = None,
percentile_cutoff: Optional[float] = None,
threshold_cutoff: Optional[float] = None,
mode: str = "percentile"
): ...Processes and transforms node metadata to enhance content understanding and presentation.
class MetadataReplacementPostProcessor(BaseNodePostprocessor):
"""
Postprocessor for replacing and transforming node metadata.
Parameters:
- target_metadata_key: str, metadata key to replace or transform
- new_metadata_key: str, new key name for transformed metadata
- replacement_function: Optional[Callable], function for metadata transformation
"""
def __init__(
self,
target_metadata_key: str,
new_metadata_key: str = "new_metadata",
replacement_function: Optional[Callable] = None
): ...Advanced relevance scoring and document-level processing for improved result quality.
class DocumentWithRelevance:
"""
Document wrapper with relevance scoring for postprocessing operations.
Parameters:
- document: Document, the original document
- relevance_score: float, computed relevance score
- metadata: Optional[dict], additional relevance metadata
"""
def __init__(
self,
document: Document,
relevance_score: float,
metadata: Optional[dict] = None
): ...
@property
def text(self) -> str:
"""Get document text content."""
@property
def doc_id(self) -> str:
"""Get document identifier."""from llama_index.core.postprocessor import SimilarityPostprocessor
from llama_index.core.schema import NodeWithScore, TextNode
# Create test nodes with scores
nodes = [
NodeWithScore(node=TextNode(text="Machine learning algorithms"), score=0.85),
NodeWithScore(node=TextNode(text="Deep learning techniques"), score=0.72),
NodeWithScore(node=TextNode(text="Unrelated content here"), score=0.45),
NodeWithScore(node=TextNode(text="Neural network architectures"), score=0.78)
]
# Filter by similarity threshold
similarity_filter = SimilarityPostprocessor(similarity_cutoff=0.7)
filtered_nodes = similarity_filter.postprocess_nodes(nodes)
print(f"Original nodes: {len(nodes)}")
print(f"Filtered nodes: {len(filtered_nodes)}")
for node in filtered_nodes:
print(f"Score: {node.score:.2f}, Text: {node.text}")from llama_index.core.postprocessor import KeywordNodePostprocessor
# Keyword filtering
keyword_filter = KeywordNodePostprocessor(
required_keywords=["machine", "learning"],
exclude_keywords=["unrelated", "spam"]
)
filtered_by_keywords = keyword_filter.postprocess_nodes(nodes)
print(f"Keyword filtered nodes: {len(filtered_by_keywords)}")from llama_index.core.postprocessor import LLMRerank
from llama_index.core.llms import MockLLM
# Initialize LLM reranker
llm = MockLLM()
reranker = LLMRerank(
llm=llm,
top_n=3,
choice_batch_size=5
)
# Rerank nodes based on relevance
reranked_nodes = reranker.postprocess_nodes(
nodes,
query_bundle=QueryBundle(query_str="What is machine learning?")
)
print("Reranked results:")
for i, node in enumerate(reranked_nodes):
print(f"{i+1}. Score: {node.score:.2f}, Text: {node.text}")from llama_index.core.postprocessor import PrevNextNodePostprocessor
from llama_index.core.storage.docstore import SimpleDocumentStore
# Setup document store with node relationships
docstore = SimpleDocumentStore()
# Add nodes with relationships to docstore
# docstore.add_documents([...])
# Context enhancement postprocessor
context_enhancer = PrevNextNodePostprocessor(
docstore=docstore,
num_nodes=1,
mode="both"
)
# Add context to retrieved nodes
enhanced_nodes = context_enhancer.postprocess_nodes(nodes)
print("Enhanced nodes with context:")
for node in enhanced_nodes:
print(f"Enhanced text length: {len(node.text)}")from llama_index.core.postprocessor import FixedRecencyPostprocessor
from datetime import datetime, timedelta
# Create nodes with date metadata
recent_nodes = [
NodeWithScore(
node=TextNode(
text="Latest ML research findings",
metadata={"date": datetime.now().isoformat()}
),
score=0.75
),
NodeWithScore(
node=TextNode(
text="Historical ML overview",
metadata={"date": (datetime.now() - timedelta(days=365)).isoformat()}
),
score=0.80
)
]
# Prioritize recent content
recency_processor = FixedRecencyPostprocessor(
top_k=1,
date_key="date"
)
recent_filtered = recency_processor.postprocess_nodes(recent_nodes)
print("Most recent content:")
for node in recent_filtered:
print(f"Date: {node.node.metadata['date']}")
print(f"Text: {node.text}")from llama_index.core.postprocessor import PIINodePostprocessor
# Nodes with potential PII
pii_nodes = [
NodeWithScore(
node=TextNode(text="Contact John Doe at john.doe@email.com for more info"),
score=0.80
),
NodeWithScore(
node=TextNode(text="The phone number is 555-123-4567"),
score=0.75
)
]
# Remove PII from nodes
pii_remover = PIINodePostprocessor(pii_str_tmpl="[REDACTED]")
sanitized_nodes = pii_remover.postprocess_nodes(pii_nodes)
print("Sanitized content:")
for node in sanitized_nodes:
print(f"Text: {node.text}")from llama_index.core.postprocessor import LongContextReorder
# Reorder for long context optimization
long_context_reorder = LongContextReorder()
reordered_nodes = long_context_reorder.postprocess_nodes(nodes)
print("Reordered for long context:")
for i, node in enumerate(reordered_nodes):
print(f"Position {i}: {node.text[:50]}...")from llama_index.core.postprocessor import SentenceTransformerRerank
# Advanced semantic reranking
sentence_reranker = SentenceTransformerRerank(
model="cross-encoder/ms-marco-MiniLM-L-2-v2",
top_n=3,
keep_retrieval_score=True
)
# Note: This requires actual sentence-transformers library
# reranked_semantic = sentence_reranker.postprocess_nodes(
# nodes,
# query_bundle=QueryBundle(query_str="machine learning algorithms")
# )# Chain multiple postprocessors
postprocessors = [
SimilarityPostprocessor(similarity_cutoff=0.6),
KeywordNodePostprocessor(required_keywords=["machine", "learning"]),
LLMRerank(llm=llm, top_n=2)
]
# Apply postprocessors in sequence
processed_nodes = nodes
for processor in postprocessors:
processed_nodes = processor.postprocess_nodes(processed_nodes)
print(f"Final processed nodes: {len(processed_nodes)}")
for node in processed_nodes:
print(f"Final result: {node.text}")# Postprocessor modes and configurations
class PostprocessorMode(str, Enum):
SIMILARITY = "similarity"
KEYWORD = "keyword"
LLM_RERANK = "llm_rerank"
RECENCY = "recency"
PII_REMOVAL = "pii_removal"
# Default configuration values
DEFAULT_SIMILARITY_CUTOFF = 0.7
DEFAULT_TOP_N = 10
DEFAULT_BATCH_SIZE = 10
DEFAULT_PII_TEMPLATE = "[PII_REMOVED]"
DEFAULT_DATE_KEY = "date"Install with Tessl CLI
npx tessl i tessl/pypi-llama-index-core