"""Evidence ranking, deduplication, and truncation.""" SOURCE_PRIORITY = { "DMD_ExonSkip_Lookup": -2, "Variant_Therapy_Lookup": -2, "Supportive_Care_Lookup": -2, "ACMG_SF": -1, "ACMG_Guideline": -1, "CPIC": -1, "N1C_Eligibility": -1, "FDA": 0, "DailyMed": 0, "GeneReviews": 1, "ClinGen": 1, "ClinGen_Validity": 1, "ClinVar": 2, "Orphanet": 2, "GenCC": 2, "PharmGKB": 2, "HPO_MONDO": 2, "ClinicalTrials.gov": 3, "Ensembl": 4, "Ensembl_VEP": 4, "UniProt": 4, "ChEMBL": 5, "PubMed": 5, "PHAROS": 5, "DGIdb": 5, "Open Targets": 6, "OMIM": 7, "gnomAD": 8, } def _estimate_tokens(text: str) -> int: return len(text) // 4 def truncate_evidence(evidence: list[dict], max_tokens: int = 3000) -> list[dict]: """Rank, deduplicate, and truncate evidence to fit within token budget.""" # Deduplicate by URL, keeping the first occurrence — but merge the # strongest `_disease_relevance` across duplicates. The same trial can be # retrieved by several queries (broad gene search vs. targeted condition # search) that score disease-relevance differently; keeping only the first # (often a low-relevance gene-query hit) would let a 0.0 duplicate mask a # 1.0 hit and wrongly disqualify the right trial (-> spurious "None"). seen_urls: dict[str, dict] = {} unique = [] for e in evidence: url = e["url"] if url not in seen_urls: seen_urls[url] = e unique.append(e) elif e.get("_disease_relevance", 0) > seen_urls[url].get("_disease_relevance", 0): seen_urls[url]["_disease_relevance"] = e.get("_disease_relevance") def sort_key(e): base = SOURCE_PRIORITY.get(e["source_name"], 99) snippet = e.get("snippet", "") if "maps to exon" in snippet or "DOMAIN MATCH" in snippet: return (-1, base) return (0, base) unique.sort(key=sort_key) selected = [] total_tokens = 0 for e in unique: snippet_tokens = _estimate_tokens(e["snippet"]) if total_tokens + snippet_tokens > max_tokens: break selected.append(e) total_tokens += snippet_tokens return selected