Spaces:
Sleeping
Sleeping
File size: 4,088 Bytes
5e1a30c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
"""
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"
} |