# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-resolve-conflicting-rankings-between-vector-similarity-and-keyword-search-results.md # Remnant Warrant: https://corpusfetch.com/c/how-should-an-instance-resolve-conflicting-rankings-between-vector-similarity-and-keyword-search-results.warrant.json # Markdown SHA-256: cbd02a8e74b9c3ae5f42510415da8ffa1bad0b13ee0f4ca3e62cb4ecf1ad6b63 # Code AST SHA-256: 1496724c6d73342ea66d9d9bdd5972e795d7c673cbf74411bf3927c1576a3c71 # Verified Classes: ArbitrationVerdict, HybridRanking, ArbitrationPolicy, ArbitrationManifest, HybridSearchArbitrationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ArbitrationVerdict(Enum): RANKINGS_AGREE = auto() VECTOR_PRIVILEGED = auto() KEYWORD_PRIVILEGED = auto() ESCALATE_TO_HUMAN = auto() @dataclass(frozen=True) class HybridRanking: query_hash: str vector_ranking: tuple[str, ...] keyword_ranking: tuple[str, ...] semantic_similarity_peak: float disagreement_magnitude: float @dataclass(frozen=True) class ArbitrationPolicy: semantic_similarity_threshold: float max_acceptable_disagreement: float @dataclass(frozen=True) class ArbitrationManifest: verdict: ArbitrationVerdict chosen_ranking: tuple[str, ...] directive: str manifest_digest: str class HybridSearchArbitrationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_ranking( cls, ranking: HybridRanking, policy: ArbitrationPolicy ) -> ArbitrationManifest: # Gate 1: Agreement check top_k = min(5, len(ranking.vector_ranking), len(ranking.keyword_ranking)) if ranking.vector_ranking[:top_k] == ranking.keyword_ranking[:top_k]: digest = cls._hash(json.dumps({ "query": ranking.query_hash, "top_k": tuple(ranking.vector_ranking[:top_k]) }, sort_keys=True)) return ArbitrationManifest( verdict=ArbitrationVerdict.RANKINGS_AGREE, chosen_ranking=ranking.vector_ranking, directive="PROCEED: Top-K rankings agree. Use fused ranking.", manifest_digest=digest ) # Gate 2: Severe disagreement check if ranking.disagreement_magnitude > policy.max_acceptable_disagreement: return ArbitrationManifest( verdict=ArbitrationVerdict.ESCALATE_TO_HUMAN, chosen_ranking=tuple(), directive=f"HALT: Disagreement magnitude {ranking.disagreement_magnitude:.2f} exceeds threshold {policy.max_acceptable_disagreement}. Escalate.", manifest_digest=cls._hash(f"ESCALATE:{ranking.query_hash}") ) # Gate 3: Semantic similarity privilege if ranking.semantic_similarity_peak >= policy.semantic_similarity_threshold: digest = cls._hash(json.dumps({ "query": ranking.query_hash, "chosen": "VECTOR", "similarity": f"{ranking.semantic_similarity_peak:.3f}" }, sort_keys=True)) return ArbitrationManifest( verdict=ArbitrationVerdict.VECTOR_PRIVILEGED, chosen_ranking=ranking.vector_ranking, directive=f"PROCEED: Peak similarity {ranking.semantic_similarity_peak:.2f} privileges vector ranking.", manifest_digest=digest ) # Gate 4: Keyword privilege digest = cls._hash(json.dumps({ "query": ranking.query_hash, "chosen": "KEYWORD" }, sort_keys=True)) return ArbitrationManifest( verdict=ArbitrationVerdict.KEYWORD_PRIVILEGED, chosen_ranking=ranking.keyword_ranking, directive="PROCEED: Semantic similarity below threshold. Privilege keyword ranking.", manifest_digest=digest )