#!/usr/bin/env python3 """Hybrid search proof lab: Python 3.11+, standard library, entirely offline. Keyword overlap and reciprocal rank fusion are calculated here. Semantic scores are a recording from the existing pinned all-MiniLM-L6-v2 CPU lab, not inference performed by this file. All text is synthetic. No files, accounts or networks are accessed. This deliberately small ranker is not BM25 or a production engine. """ import argparse import json import math import re import unittest FIXTURE = {'model': 'sentence-transformers/all-MiniLM-L6-v2', 'revision': '1110a243fdf4706b3f48f1d95db1a4f5529b4d41', 'documents': [{'id': 'P1', 'text': 'Use the account recovery page to choose a new password.'}, {'id': 'P2', 'text': 'A locked account needs an administrator to restore access.'}, {'id': 'P3', 'text': 'Export invoices from the billing dashboard.'}, {'id': 'P4', 'text': 'Do not use the account recovery page to choose a new password.'}], 'cases': [{'id': 'exact', 'label': 'Same words', 'text': 'Use the account recovery page to choose a new password.', 'ranked': [{'id': 'P1', 'score': 1.0}, {'id': 'P4', 'score': 0.894806}, {'id': 'P2', 'score': 0.553137}, {'id': 'P3', 'score': 0.071292}]}, {'id': 'paraphrase', 'label': 'Different wording', 'text': 'How do I reset my password?', 'ranked': [{'id': 'P1', 'score': 0.695384}, {'id': 'P4', 'score': 0.589092}, {'id': 'P2', 'score': 0.437875}, {'id': 'P3', 'score': 0.004147}]}, {'id': 'negation', 'label': 'Opposite instruction', 'text': 'Do not use the account recovery page to choose a new password.', 'ranked': [{'id': 'P4', 'score': 1.0}, {'id': 'P1', 'score': 0.894806}, {'id': 'P2', 'score': 0.494415}, {'id': 'P3', 'score': 0.025308}]}, {'id': 'unrelated', 'label': 'Outside the collection', 'text': 'What is the weather forecast for tomorrow?', 'ranked': [{'id': 'P3', 'score': 0.004799}, {'id': 'P2', 'score': -0.03766}, {'id': 'P1', 'score': -0.089711}, {'id': 'P4', 'score': -0.098703}]}]} RRF_CONSTANT = 60 IGNORED_WORDS = set("a an the to of from for is are i my how what do".split()) def search_terms(text): return sorted(set(re.findall(r"[a-z0-9]+", text.lower())) - IGNORED_WORDS) def keyword_ranking(documents, query): terms = search_terms(query) hits = [] for document in documents: available = set(search_terms(document["text"])) matched = [term for term in terms if term in available] if matched: hits.append({"id": document["id"], "score": len(matched), "matched": matched}) return sorted(hits, key=lambda hit: (-hit["score"], hit["id"])) def reciprocal_rank_fusion(keyword, semantic, constant=RRF_CONSTANT): if isinstance(constant, bool) or not isinstance(constant, (int, float)) or not math.isfinite(constant) or constant < 0: raise ValueError("The rank constant must be finite and nonnegative.") if len(set(keyword)) != len(keyword) or len(set(semantic)) != len(semantic): raise ValueError("A ranked list cannot repeat a document.") hits = [] for document_id in sorted(set(keyword + semantic)): keyword_rank = keyword.index(document_id) + 1 if document_id in keyword else None semantic_rank = semantic.index(document_id) + 1 if document_id in semantic else None keyword_contribution = 1 / (constant + keyword_rank) if keyword_rank is not None else 0 semantic_contribution = 1 / (constant + semantic_rank) if semantic_rank is not None else 0 hits.append({ "id": document_id, "keywordRank": keyword_rank, "semanticRank": semantic_rank, "keywordContribution": keyword_contribution, "semanticContribution": semantic_contribution, "score": keyword_contribution + semantic_contribution, }) return sorted(hits, key=lambda hit: (-hit["score"], hit["id"])) def inspect(case_id="paraphrase", depth=2, omit_p1=False): query = next((case for case in FIXTURE["cases"] if case["id"] == case_id), None) if query is None: raise ValueError("Choose a recorded query.") if type(depth) is not int or depth not in (1, 2, 4): raise ValueError("Choose one, two or four candidates per search.") documents = [doc for doc in FIXTURE["documents"] if not omit_p1 or doc["id"] != "P1"] allowed = {doc["id"] for doc in documents} keyword = keyword_ranking(documents, query["text"])[:depth] semantic = [hit for hit in query["ranked"] if hit["id"] in allowed][:depth] return {"keyword": keyword, "semantic": semantic, "fused": reciprocal_rank_fusion([hit["id"] for hit in keyword], [hit["id"] for hit in semantic])} class ProofTests(unittest.TestCase): def test_rank_positions_not_unlike_raw_scores(self): row = reciprocal_rank_fusion(["A", "B"], ["B", "C"])[0] self.assertEqual(row["id"], "B") self.assertAlmostEqual(row["score"], 1 / 62 + 1 / 61) def test_missing_contribution_is_zero(self): row = reciprocal_rank_fusion([], ["A"])[0] self.assertIsNone(row["keywordRank"]) self.assertEqual(row["keywordContribution"], 0) self.assertAlmostEqual(row["score"], 1 / 61) def test_ties_use_document_id_not_source_authority(self): rows = reciprocal_rank_fusion(["B", "A"], ["A", "B"]) self.assertEqual([row["id"] for row in rows], ["A", "B"]) self.assertEqual(rows[0]["score"], rows[1]["score"]) def test_no_answer_still_has_a_first_result(self): result = inspect("unrelated") self.assertEqual(result["keyword"], []) self.assertEqual(result["fused"][0]["id"], "P3") def test_depth_one_hides_conflicting_candidate(self): self.assertEqual([row["id"] for row in inspect(depth=1)["fused"]], ["P1"]) self.assertEqual([row["id"] for row in inspect(depth=2)["fused"]], ["P1", "P4"]) def test_removed_source_cannot_be_recovered_by_fusion(self): for case in FIXTURE["cases"]: result = inspect(case["id"], 4, True) for channel in result.values(): self.assertNotIn("P1", [row["id"] for row in channel]) def test_invalid_inputs_do_not_produce_evidence(self): for keyword, semantic, constant in [(["A", "A"], [], 60), ([], ["A", "A"], 60), ([], [], -1), ([], [], float("nan"))]: with self.assertRaises(ValueError): reciprocal_rank_fusion(keyword, semantic, constant) with self.assertRaises(ValueError): inspect("new-unrecorded-query") with self.assertRaises(ValueError): inspect(depth=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=["inspect", "compare", "test"], nargs="?", default="inspect") parser.add_argument("--query", choices=[case["id"] for case in FIXTURE["cases"]], default="paraphrase") parser.add_argument("--depth", type=int, choices=[1, 2, 4], default=2) parser.add_argument("--omit-p1", action="store_true", help="Leave P1 out of both candidate searches before limiting their depth.") args = parser.parse_args() if args.command == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(ProofTests) result = unittest.TextTestRunner(verbosity=2).run(suite) raise SystemExit(0 if result.wasSuccessful() else 1) if args.command == "compare": result = [{"query": case["id"], "depth": depth, "omitted": args.omit_p1, "result": inspect(case["id"], depth, args.omit_p1)} for case in FIXTURE["cases"] for depth in (1, 2, 4)] else: result = inspect(args.query, args.depth, args.omit_p1) print(json.dumps(result, indent=2)) if __name__ == "__main__": main()