Spaces:
Sleeping
Sleeping
| """Novelty-gap search: scan the indexed literature cheaply, then read the papers that matter. | |
| The pipeline is deliberately two-speed. A wide scan ranks hundreds of indexed works with | |
| hybrid retrieval, reference edges and embeddings, and costs no LLM calls; only the handful | |
| of papers that could actually settle the claim are pulled as open-access full text and read | |
| by the large model. The question it answers is not "what is relevant" but "has this specific | |
| claim been tested, and if not, which facet of it is still open". | |
| Full text is the one thing the corpus cannot supply: the snapshot carries abstracts and | |
| metadata, and an abstract states what a paper claims rather than what it measured. | |
| """ | |
| import argparse | |
| import asyncio | |
| import json | |
| import logging | |
| import math | |
| import os | |
| import re | |
| import sys | |
| from dataclasses import asdict, dataclass, field | |
| from app.config import Settings, settings | |
| from app.embeddings import get_embedder | |
| from app.llm import LLMService, Usage | |
| from app.models import Claim, NoveltyAssessment | |
| from app.paper_text import FullTextClient | |
| from app.repository import ElasticRepository, lexical_query | |
| logger = logging.getLogger(__name__) | |
| # Reading a paper the scan already ranked far from the claim buys nothing. | |
| READ_MIN_SIMILARITY = 0.35 | |
| REFERENCE_CAP = 200 | |
| class Candidate: | |
| """A scanned work: metadata, how it was reached, and how close it sits to the claim.""" | |
| id: str | |
| title: str | |
| abstract: str | |
| year: int | None | |
| venue: str | |
| url: str | |
| doi: str | |
| pmid: str | |
| cited_by_count: int | |
| is_review: bool | |
| relation: str # retrieved | reference | |
| referenced_works: list[str] = field(default_factory=list) | |
| embedding: list[float] | None = None | |
| score: float = 0.0 | |
| def to_candidate(document: dict, relation: str) -> Candidate | None: | |
| """An indexed study document as a scan candidate; retracted work is never a precedent.""" | |
| if document.get("is_retracted") or not document.get("id"): | |
| return None | |
| url = document.get("url", "") or "" | |
| return Candidate( | |
| id=document["id"], | |
| title=document.get("title", ""), | |
| abstract=document.get("abstract", "") or "", | |
| year=document.get("year"), | |
| venue=document.get("venue", "") or "", | |
| url=url, | |
| doi=url if "doi.org" in url else "", | |
| pmid=(document.get("pmids") or [""])[0], | |
| cited_by_count=document.get("cited_by_count") or 0, | |
| is_review=bool(document.get("is_review")), | |
| relation=relation, | |
| referenced_works=document.get("referenced_works") or [], | |
| embedding=document.get("embedding"), | |
| ) | |
| def _title_key(value: str) -> str: | |
| return re.sub(r"[^a-z0-9]+", " ", value.lower()).strip() | |
| def _same_title(left: str, right: str) -> bool: | |
| """Compare titles through punctuation, case and markup differences between sources.""" | |
| first, second = _title_key(left), _title_key(right) | |
| return bool(first) and bool(second) and (first == second or first in second or second in first) | |
| def cosine(a: list[float], b: list[float]) -> float: | |
| denominator = math.sqrt(sum(x * x for x in a)) * math.sqrt(sum(y * y for y in b)) | |
| return sum(x * y for x, y in zip(a, b)) / denominator if denominator else 0.0 | |
| class NoveltyEngine: | |
| def __init__( | |
| self, | |
| llm: LLMService | None = None, | |
| config: Settings = settings, | |
| repo: ElasticRepository | None = None, | |
| fulltext: FullTextClient | None = None, | |
| embedder=None, | |
| ): | |
| self.config = config | |
| self.llm = llm or LLMService(config) | |
| self.repo = repo or ElasticRepository(config) | |
| self.fulltext = fulltext or FullTextClient() | |
| self.embedder = embedder | |
| async def close(self) -> None: | |
| await asyncio.gather(self.repo.close(), self.fulltext.close(), self.llm.close()) | |
| async def embed(self, texts: list[str]) -> list[list[float]]: | |
| if self.embedder is None: | |
| self.embedder = get_embedder(self.config.embedding_model, self.config.embedding_device) | |
| return await asyncio.to_thread(self.embedder.embed_documents, texts) | |
| async def parse(self, hypothesis: str, usage: Usage) -> Claim: | |
| try: | |
| return await self.llm.parse_claim(hypothesis, usage) | |
| except Exception as exc: | |
| logger.warning("Claim parsing unavailable: %s", type(exc).__name__) | |
| return Claim( | |
| intervention="", system="", outcome="", direction="", | |
| queries=[hypothesis[:250]], synonyms=[], | |
| ) | |
| async def scan(self, hypothesis: str, claim: Claim, limit: int) -> list[Candidate]: | |
| """Hybrid-retrieve candidates, then follow reference edges out of the closest papers. | |
| The edge walk is what a keyword query cannot do: a paper that tested this claim and | |
| phrased it differently is usually one reference away from a paper that phrases it | |
| our way, and reviews carry hand-screened lists of exactly that work. | |
| """ | |
| pico = { | |
| "intervention": claim.intervention, | |
| "outcome": claim.outcome, | |
| "population": claim.system, | |
| "synonyms": claim.synonyms, | |
| } | |
| query = lexical_query(pico, " ".join([hypothesis, *claim.queries])) | |
| vector = (await self.embed([self.target(hypothesis, claim)]))[0] | |
| documents, _ = await self.repo.retrieve(query, vector) | |
| if not documents: | |
| # The search pipeline wants a precise evidence pool; a novelty scan wants recall, | |
| # because the reader, not the query, decides whether a paper settles the claim. | |
| documents, _ = await self.repo.retrieve({"match_all": {}}, vector) | |
| found: dict[str, Candidate] = {} | |
| for document in documents: | |
| candidate = to_candidate(document, "retrieved") | |
| if candidate and candidate.id not in found: | |
| found[candidate.id] = candidate | |
| if not found: | |
| return [] | |
| ranked = await self.rank(vector, list(found.values())) | |
| seeds = ranked[:6] | |
| references = [ | |
| ref | |
| for ref in dict.fromkeys(r for seed in seeds for r in seed.referenced_works) | |
| if ref not in found | |
| ][:REFERENCE_CAP] | |
| expanded = [ | |
| candidate | |
| for document in await self.repo.get_many(references) | |
| if (candidate := to_candidate(document, "reference")) | |
| ] | |
| if expanded: | |
| ranked = sorted([*ranked, *await self.rank(vector, expanded)], key=lambda c: -c.score) | |
| return ranked[:limit] | |
| def target(self, hypothesis: str, claim: Claim) -> str: | |
| return " ".join(filter(None, [hypothesis, claim.intervention, claim.system, claim.outcome])) | |
| async def rank(self, vector: list[float], candidates: list[Candidate]) -> list[Candidate]: | |
| """Prefer the indexed embedding the ingestion pass already paid for; embed the rest.""" | |
| missing = [c for c in candidates if not c.embedding or len(c.embedding) != len(vector)] | |
| if missing: | |
| vectors = await self.embed([f"{c.title}. {c.abstract}" for c in missing]) | |
| for candidate, embedding in zip(missing, vectors): | |
| candidate.embedding = embedding | |
| for candidate in candidates: | |
| candidate.score = cosine(vector, candidate.embedding or []) | |
| return sorted(candidates, key=lambda c: -c.score) | |
| async def read( | |
| self, hypothesis: str, candidate: Candidate, usage: Usage, warnings: list[str] | |
| ) -> dict: | |
| """Pull the actual paper and have the large model report what its results establish.""" | |
| text = await self.fulltext.fetch( | |
| candidate.id, doi=candidate.doi, pmid=candidate.pmid, abstract=candidate.abstract | |
| ) | |
| row = { | |
| **{k: v for k, v in asdict(candidate).items() if k not in ("embedding", "referenced_works")}, | |
| "availability": text.availability, | |
| "textSource": text.source, | |
| "textChars": text.chars, | |
| } | |
| # OpenAlex merges of MAG records can attach a DOI to another work's title and year, | |
| # so the publisher record wins for anything shown next to the text we actually read. | |
| if text.title and not _same_title(text.title, candidate.title): | |
| row.update(title=text.title, indexTitle=candidate.title, metadataConflict=True) | |
| if text.year and text.year != candidate.year: | |
| row.update(year=text.year, indexYear=candidate.year) | |
| if text.venue and text.venue != candidate.venue: | |
| row.update(venue=text.venue, indexVenue=candidate.venue) | |
| warnings.append( | |
| f"Index metadata disagreed with the publisher record for '{text.title[:60]}'; " | |
| "the resolved title, year and venue are shown." | |
| ) | |
| if text.availability == "unavailable" or not self.config.openai_api_key: | |
| return row | |
| body = text.evidence_text() | |
| try: | |
| reading = await self.llm.read_paper(hypothesis, row, body, usage) | |
| except ValueError: | |
| warnings.append( | |
| f"A reading of '{candidate.title[:60]}' lacked verbatim support and was discarded." | |
| ) | |
| return row | |
| except Exception as exc: | |
| logger.warning("Paper reading unavailable: %s", type(exc).__name__) | |
| warnings.append("Some papers could not be read; their abstracts remain visible.") | |
| return row | |
| return {**row, **reading.model_dump()} | |
| async def assess(self, hypothesis: str, scan: int = 120, read: int = 5) -> dict: | |
| usage = Usage() | |
| warnings: list[str] = [] | |
| if not self.config.openai_api_key: | |
| warnings.append( | |
| "OPENAI_API_KEY is not configured: papers were scanned and retrieved but not read." | |
| ) | |
| claim = await self.parse(hypothesis, usage) | |
| scanned = await self.scan(hypothesis, claim, scan) | |
| if not scanned: | |
| return { | |
| "hypothesis": hypothesis, | |
| "claim": claim.model_dump(), | |
| "scanned": 0, | |
| "papers": [], | |
| "verdict": "insufficient_evidence", | |
| "summary": "No candidate literature was retrieved, so novelty cannot be assessed. This is a retrieval gap, not evidence that the claim is untested.", | |
| "gaps": [], | |
| "settled": [], | |
| "warnings": [*warnings, "The indexed corpus returned no candidates; it may not cover this field yet."], | |
| "usage": usage.summary([]), | |
| } | |
| targets = [c for c in scanned if not c.is_review and c.score >= READ_MIN_SIMILARITY][:read] | |
| papers = await asyncio.gather( | |
| *(self.read(hypothesis, candidate, usage, warnings) for candidate in targets) | |
| ) | |
| read_full = [p for p in papers if p.get("availability") == "full_text"] | |
| if targets and not read_full: | |
| warnings.append( | |
| "No open-access full text was available for the closest papers; readings fall back to abstracts." | |
| ) | |
| assessment = await self.judge(hypothesis, claim, papers, scanned, usage, warnings) | |
| return { | |
| "hypothesis": hypothesis, | |
| "claim": claim.model_dump(), | |
| "scanned": len(scanned), | |
| "readFullText": len(read_full), | |
| "papers": papers, | |
| "nearest": [ | |
| {"id": c.id, "title": c.title, "year": c.year, "score": round(c.score, 4), | |
| "relation": c.relation, "url": c.url} | |
| for c in scanned[: max(read * 4, 20)] | |
| ], | |
| **assessment, | |
| "warnings": warnings, | |
| "usage": usage.summary([c.abstract for c in scanned]), | |
| } | |
| async def judge( | |
| self, | |
| hypothesis: str, | |
| claim: Claim, | |
| papers: list[dict], | |
| scanned: list[Candidate], | |
| usage: Usage, | |
| warnings: list[str], | |
| ) -> dict: | |
| readings = [p for p in papers if p.get("coverage")] | |
| direct = [p for p in readings if p["coverage"] == "tests_claim"] | |
| fallback = { | |
| "verdict": "already_done" if direct else | |
| ("incremental" if any(p["coverage"] == "tests_related" for p in readings) | |
| else "insufficient_evidence"), | |
| "summary": ( | |
| f"{len(scanned)} candidate works were scanned and {len(readings)} were read. " | |
| f"{len(direct)} tested this claim directly. " | |
| "This summary is computed from the readings, without narration." | |
| ), | |
| "gaps": list(dict.fromkeys(f for p in readings for f in p.get("facets_untested", [])))[:4], | |
| "settled": list(dict.fromkeys(f for p in readings for f in p.get("facets_settled", [])))[:4], | |
| } | |
| if not readings or not self.config.openai_api_key: | |
| return fallback | |
| payload = { | |
| "hypothesis": hypothesis, | |
| "claim": claim.model_dump(), | |
| "scannedCount": len(scanned), | |
| "readings": [ | |
| { | |
| k: p.get(k) | |
| for k in ("title", "year", "url", "availability", "coverage", "system_tested", | |
| "intervention_tested", "outcome_measured", "finding", | |
| "facets_settled", "facets_untested") | |
| } | |
| for p in readings | |
| ], | |
| } | |
| try: | |
| assessment: NoveltyAssessment = await self.llm.judge_novelty(payload, usage) | |
| except Exception as exc: | |
| logger.warning("Novelty narration unavailable: %s", type(exc).__name__) | |
| warnings.append("AI narration was unavailable; the verdict is computed from readings.") | |
| return fallback | |
| return assessment.model_dump() | |
| def format_report(report: dict) -> str: | |
| lines = [ | |
| "=" * 100, | |
| f"HYPOTHESIS: {report['hypothesis']}", | |
| "=" * 100, | |
| f"claim : {json.dumps(report['claim'])}", | |
| f"scan : {report['scanned']} works ranked, " | |
| f"{report.get('readFullText', 0)} read as full text, {len(report['papers'])} read in total", | |
| "", | |
| f"VERDICT : {report['verdict'].upper()}", | |
| f"{report['summary']}", | |
| "", | |
| ] | |
| if report.get("gaps"): | |
| lines += ["OPEN GAPS:", *(f" - {gap}" for gap in report["gaps"]), ""] | |
| if report.get("settled"): | |
| lines += ["ALREADY SETTLED:", *(f" - {item}" for item in report["settled"]), ""] | |
| for paper in report["papers"]: | |
| lines += [ | |
| "-" * 100, | |
| f"[{paper.get('coverage', 'not_read')}] {paper['title'][:110]} ({paper.get('year')})", | |
| f" similarity {paper['score']:.3f} via {paper['relation']} | " | |
| f"{paper.get('availability')} {paper.get('textSource', '')} " | |
| f"({paper.get('textChars', 0)} chars) | {paper.get('url', '')}", | |
| ] | |
| if paper.get("finding"): | |
| lines += [ | |
| f" system : {paper.get('system_tested')}", | |
| f" measured : {paper.get('outcome_measured')}", | |
| f" finding : {paper.get('finding')}", | |
| *(f" quote : \"{quote[:220]}\"" for quote in paper.get("quotes", [])), | |
| ] | |
| usage = report["usage"] | |
| lines += [ | |
| "-" * 100, | |
| f"llm calls {usage['calls']} | tokens in {usage['inputTokens']} out {usage['outputTokens']} " | |
| f"| estimated ${usage['estimatedUsd']:.4f}", | |
| ] | |
| if report.get("warnings"): | |
| lines += ["warnings:", *(f" ! {w}" for w in report["warnings"])] | |
| return "\n".join(lines) | |
| async def run(hypothesis: str, scan: int, read: int, as_json: bool) -> int: | |
| engine = NoveltyEngine() | |
| try: | |
| report = await engine.assess(hypothesis, scan=scan, read=read) | |
| finally: | |
| await engine.close() | |
| print(json.dumps(report, indent=2) if as_json else format_report(report)) | |
| return 0 | |
| def main(argv: list[str] | None = None) -> int: | |
| parser = argparse.ArgumentParser(description="Check whether a hypothesis is already tested.") | |
| parser.add_argument("hypothesis") | |
| parser.add_argument("--scan", type=int, default=120, help="candidate works to rank") | |
| parser.add_argument("--read", type=int, default=5, help="papers to pull and read in full") | |
| parser.add_argument("--json", action="store_true") | |
| args = parser.parse_args(argv) | |
| logging.basicConfig(level=os.environ.get("LOG_LEVEL", "WARNING")) | |
| return asyncio.run(run(args.hypothesis, args.scan, args.read, args.json)) | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |