infrastructure-search-literature

Literature Search Submodule

Multi-source literature search modelled after Paperclip's agent-native abstractions.

Quick Search

from infrastructure.search.literature import (
    LiteratureClient, SearchQuery, ArxivBackend, CrossrefBackend
)

client = LiteratureClient([ArxivBackend(), CrossrefBackend(mailto="[email protected]")])
result = client.search(SearchQuery(text="protein language model fitness", max_results=20))

print(f"{len(result)} unique papers from {len(result.per_source_counts)} sources")
print(f"Errors: {result.errors}")    # {} when all backends succeeded
for paper in result.papers[:5]:
    print(f"  [{paper.score:.2f}] {paper.title} ({paper.year}) — {paper.doi or paper.url}")

Backends

from infrastructure.search.literature import (
    LocalBackend, ArxivBackend, CrossrefBackend, PaperclipBackend
)

# Offline / reproducible — searches a JSON corpus on disk.
local = LocalBackend("data/curated_corpus.json")

# Public APIs, no auth.
arxiv = ArxivBackend()
crossref = CrossrefBackend(mailto="[email protected]")

# Paperclip — requires API key.
import os
paperclip = PaperclipBackend(api_key=os.environ["PAPERCLIP_API_KEY"])

Filters

SearchQuery(
    text="adam optimizer",
    max_results=50,
    year_min=2014, year_max=2025,
    sources=["arxiv"],          # subset of configured backends
)

Caching

from infrastructure.search.literature import SearchCache

cache = SearchCache("output/search_cache", ttl_seconds=86400)
client = LiteratureClient([ArxivBackend()], cache=cache)

# First call hits arXiv; second is a deterministic file read.
client.search(SearchQuery(text="x"))
client.search(SearchQuery(text="x"))

# Force refresh:
client.search(SearchQuery(text="x"), use_cache=False)

Deduplication

from infrastructure.search.literature import merge_papers

unique = merge_papers([*result_a.papers, *result_b.papers])

Priority: DOI → arXiv id → normalised (title, year). Higher-scored copy wins; missing fields on the winner are filled from the loser ("union of evidence").