""" Identity Reranker implementation for Modular Retriever Architecture. This module provides a no-op reranker that returns results unchanged, useful for disabling reranking while maintaining the same interface. """ import logging from typing import List, Dict, Any, Tuple from src.core.interfaces import Document from .base import Reranker logger = logging.getLogger(__name__) class IdentityReranker(Reranker): """ Identity reranker that returns results unchanged. This is a no-op implementation that passes through the original ranking without modification. Useful for: - Disabling reranking while maintaining interface compatibility - Testing and baseline comparisons - Performance benchmarking - Fallback when reranking models are unavailable Features: - Zero computational overhead - Preserves original ranking order - Maintains same interface as other rerankers - Can be enabled/disabled for consistency Example: config = { "enabled": True # Even when enabled, does nothing } reranker = IdentityReranker(config) results = reranker.rerank(query, documents, initial_scores) # results will be identical to input order """ def __init__(self, config: Dict[str, Any]): """ Initialize identity reranker. Args: config: Configuration dictionary with: - enabled: Whether reranking is enabled (default: True) Note: Even when enabled, this reranker does nothing """ self.config = config self.enabled = config.get("enabled", True) self._initialized = True # Always initialized since it's a no-op logger.info(f"IdentityReranker initialized with enabled={self.enabled}") def rerank( self, query: str, documents: List[Document], initial_scores: List[float] ) -> List[Tuple[int, float]]: """ Return documents in original order without reranking. Args: query: The search query (unused) documents: List of candidate documents (unused) initial_scores: Initial relevance scores from fusion Returns: List of (document_index, score) tuples in original order """ if not self.enabled: return [] # Return original ranking unchanged return [(i, score) for i, score in enumerate(initial_scores)] def is_enabled(self) -> bool: """ Check if reranking is enabled. Returns: True if reranking should be performed """ return self.enabled def get_reranker_info(self) -> Dict[str, Any]: """ Get information about the reranker. Returns: Dictionary with reranker configuration and statistics """ return { "type": "identity", "enabled": self.enabled, "description": "No-op reranker that preserves original ranking", "computational_overhead": 0.0, "modifies_ranking": False } def enable(self) -> None: """Enable reranking (no-op for identity reranker).""" self.enabled = True logger.info("Identity reranker enabled (no-op)") def disable(self) -> None: """Disable reranking.""" self.enabled = False logger.info("Identity reranker disabled") def get_performance_info(self) -> Dict[str, Any]: """ Get performance information about the reranker. Returns: Dictionary with performance characteristics """ return { "average_latency_ms": 0.0, "memory_usage_mb": 0.0, "cpu_usage_percent": 0.0, "gpu_usage_percent": 0.0, "throughput_docs_per_second": float('inf'), "description": "Identity reranker has zero computational overhead" }