search

metasearch, fuse, rerank, and research

Parallel backends via ddgs, fused with RRF, then BM25 (or flashrank). google() uses a stealth browser for real Google ranking. research() searches, reads, and returns a cited digest.

Client

One shared DDGS with TTL cache. Env vars from ddgs_env() cover SSL/proxy.

---

source

get_ddgs

def get_ddgs(
    timeout:int=10
):

Call self as a function.

---

source

ddgs_env

def ddgs_env()->dict:

Proxy + CA settings for ddgs, read from the environment (HTTPS_PROXY, SSL_CERT_FILE, …).

Backends

Each backend keeps its own rank order. We do not use ddgs’ aggregator (random backends + Wikipedia bias).

---

source

backends

def backends(
    category:str='text'
)->list:

Every ddgs backend available for category.

Fuse

RRF over per-backend ranks; norm_url dedups. engines lists which backends hit.

---

source

rrf

def rrf(
    lists:list, # ranked result lists, one per backend
    k:int=60, # RRF damping constant
)->list:

Reciprocal rank fusion: score each url as sum 1/(k+rank) across the lists it appears in.

Rerank

method='lexical' (BM25), 'flashrank', 'auto', or 'none' (fused list as-is).

---

source

rerank

def rerank(
    q:str, # search query
    hits:list, # fused hits (from `rrf`) or any list of result dicts
    n:int=None, # keep the top n (None = all)
    method:str='none', # none | lexical | flashrank | auto
    mix:float=0.4, # weight on the retrieval score vs the reranker (1 = ignore the reranker)
)->list:

Re-order hits by relevance to q, blending the reranker with the retrieval score.

---

source

flashrank_scores

def flashrank_scores(
    q:str, texts:list, model:str=None, cache_dir:str=None
)->list | None:

Cross-encoder scores from flashrank, or None when it isn’t installed (pip install flashrank).

---

source

bm25

def bm25(
    q:str, docs:list, k1:float=1.5, b:float=0.75
)->list:

BM25 scores for docs against q, with idf computed from docs themselves (no index needed).

Where & when

region='auto' resolves country from the query and language from the asker separately (fr-en for an English question about France). timelimit='d'|'w'|'m'|'y' restricts recency.

---

source

infer_region

def infer_region(
    q:str, # the query, read for a place and a language
    dflt:str='us-en', # used for whichever half the query does not settle
)->str:

ddgs country-lang from q, or dflt when auto has nothing to say.

---

source

infer_language

def infer_language(
    q:str
)->str:

Language code from function words in q, or None when unclear/tied.

---

source

infer_country

def infer_country(
    q:str
)->str:

ISO country from q, or None when none/ambiguous.

search

Fan-out is independent of n. pages>1 widens the candidate pool.

---

source

extract

def extract(
    url, **kw
):

Call self as a function.

---

source

search

def search(
    q:str, # search query
    category:str='text', # text | images | news | videos | books
    n:int=10, # number of results to return
    fuse:bool=True, # fan out across backends and fuse with RRF (text only)
    backend:str=None, # restrict to a single ddgs backend, e.g. 'google'
    method:str='none', # rerank method: none | lexical | flashrank | auto
    pages:int=1, # result pages to pull from each backend (raises the candidate pool)
    region:str='auto', # ddgs region, or 'auto' to read the country off the query
    timelimit:str=None, # d | w | m | y — only results from the last day/week/month/year
    timeout:int=8, **kw
)->list: # extra kwargs passed to ddgs (max_results, safesearch, page)

Search the web. Text results are fused across backends with RRF; pass method=‘lexical’ or ‘flashrank’ to rerank.

google

Stealth browser past Google’s JS wall. Same result shape as search.

---

source

google

def google(
    q:str, # search query
    n:int=10, # number of results
    lang:str=None, # `hl`; defaults to the language half of `region`
    region:str='auto', # ddgs-style region, or 'auto' to read the country off the query
    timelimit:str=None, # d | w | m | y — Google's own `tbs=qdr:` filter
)->list:

Search Google directly via a stealth browser (real Google ranking). Returns dicts, same shape as search().

research

plan → search each facet → curate → fetch readable pages → cited digest. Bot-wall 200s go to dropped and are backfilled. See 06_quality for plan/curate.

# pagination: ranks run on across pages, and page 2's repeats of page 1 don't get counted twice
_seen = []
def _fake(be, q, region='us-en', page=1, timeout=8, tries=2, **kw):
    _seen.append((be, page))
    start = (page-1)*2
    return [dict(title=f'{be}{i}', href=f'https://e.com/{be}/{i}', body='', engine=be, rank=i-start+1)
            for i in range(start, start+3)]        # 3 hits/page, overlapping the next page by one
_real, _engine_hits = _engine_hits, _fake
try:
    one = _backend_lists('q', ('brave',), pages=1)
    two = _backend_lists('q', ('brave',), pages=2)
finally: _engine_hits = _real

assert sorted(_seen) == [('brave',1), ('brave',1), ('brave',2)], _seen   # one job per (backend, page)
assert len(one[0]) == 3 and len(two[0]) == 5                             # 3+3 minus the one repeat
assert [h['rank'] for h in two[0]] == [1,2,3,4,5]                        # continuous, not 1,2,3,1,2,3
assert len({h['href'] for h in two[0]}) == 5                             # deduped within the backend

---

source

research

def research(
    q:str, # search query
    n:int=5, # number of readable sources to gather
    engine:str='search', # 'search' (fused metasearch) or 'google' (stealth browser ranking)
    sel:str=None, # CSS selector to narrow each page before markdown
    chars:int=4000, # max markdown chars kept per page
    auto:bool=True, # auto-escalate fetching (plain->heavy->stealthy->session) per page
    region:str='auto', # ddgs region, or 'auto' to read the country off the query
    timelimit:str=None, # d | w | m | y — only sources from the last day/week/month/year
    curated:bool=True, # drop mirror domains and rank by source authority before reading
    intent:str='auto', # authority class ('policy', 'docs', …), 'auto' to read it off `q`
    facets:int=4, # searches to fan out over when `q` holds more than one question; 1 disables
    rewrite:NoneType=None, # optional `f(q) -> list[str]` (a model) in place of the heuristic planner
    focused:bool=True, # keep query-relevant passages instead of the first `chars` chars
    candidates:int=None, # how many hits to draw on when backfilling (default 3n)
    pages:int=1, # result pages per backend (raise it when `candidates` exceeds one page)
    concurrency:int=8, **kw
)->dict: # extra kwargs passed to fetch()

Query -> read the top n readable* results -> a cited markdown corpus. Returns {query, sources, digest, dropped, region, curation}.*

---

source

focus

def focus(
    md:str, # page markdown
    q:str, # query to keep relevant to
    chars:int=4000, # character budget
    reserve:float=0.3, # fraction of the budget spent on the top of the page before ranking
    lead:float=0.6, # extra score for blocks near the top (intros define things)
)->str:

Keep the parts of md that answer q, in document order, within chars.

---

source

usable

def usable(
    pg, min_chars:int=400
)->bool:

Did this fetch return a readable page, or an error / bot wall / empty JS shell?

---

source

page_date

def page_date(
    page
)->str:

Publication date from JSON-LD / meta / URL path, or None.

Tests

# empty query is a no-op for every entry point (no network)
assert search('') == []
assert research('')['sources'] == []
assert rrf([]) == [] and rerank('x', []) == []

# Google SERP parser works on a static fixture (no browser/network)
_serp = ('<html><body><div class="MjjYud"><div class="g">'
         '<a href="https://ex.com/p"><h3>Example Page</h3></a>'
         '<div class="VwiC3b">A short snippet.</div></div></div>'
         '<div><a href="https://www.google.com/aclk"><h3>Ad</h3></a></div></body></html>')
_g = _parse_google(_serp, 5)
assert len(_g) == 1, f'expected 1 organic result, got {len(_g)}'   # google.com ad link filtered out
assert _g[0]['href'] == 'https://ex.com/p' and _g[0]['title'] == 'Example Page'
assert _g[0]['content'] == 'A short snippet.'

# urls that differ only by scheme/www/trailing slash/tracking params are one page
assert norm_url('https://www.x.com/a/?utm_source=t#frag') == norm_url('http://x.com/a')
assert norm_url('https://x.com/a') != norm_url('https://x.com/b')

_g = [dict(href='https://x.com/1', rank=1, engine='google'), dict(href='https://x.com/8', rank=8, engine='google')]
_b = [dict(href='https://x.com/8', rank=8, engine='brave'),  dict(href='https://x.com/2', rank=2, engine='brave')]

# two urls, one backend each: the better-ranked one wins. ddgs ties them and orders by thread completion.
_solo = rrf([_g[:1], _b[1:]])
assert [h['href'] for h in _solo] == ['https://x.com/1', 'https://x.com/2'], _solo

# agreement across backends still outranks a single strong hit — that part of ddgs' instinct is right
_f = rrf([_g, _b])
assert _f[0]['href'] == 'https://x.com/8' and _f[0]['engines'] == ['google', 'brave']
assert len(_f) == 3                                                    # deduped across backends

# nothing is hoisted by domain: wikipedia earns its place from rank like everything else
_w = rrf([[dict(href='https://en.wikipedia.org/wiki/Rust', rank=7, engine='google'),
           dict(href='https://actix.rs/', rank=1, engine='google')]])
assert _w[0]['href'] == 'https://actix.rs/', _w

# BM25 ranks the doc that actually covers the query terms
_docs = ['sqlite wal mode benchmark write ahead log', 'a recipe for banana bread', 'wal mode']
_s = bm25('sqlite wal mode benchmark', _docs)
assert _s[0] == max(_s) and _s[1] == 0.0, _s

# rerank pulls the on-topic hit up past a higher-retrieval-score off-topic one
_hits = [dict(href='https://a.com/cake', title='Banana bread', body='baking', score=1.0),
         dict(href='https://b.com/sqlite-wal-benchmark', title='SQLite WAL benchmark', body='wal vs journal', score=0.9)]
assert rerank('sqlite wal benchmark', _hits, method='lexical')[0]['href'].startswith('https://b.com')

# focus() keeps the passage that answers the query, not the first N chars
_md = ('Cookie notice. We use cookies.\n\n' + 'Navigation menu home about contact.\n\n'*8 +
       'WAL mode allows readers and writers to proceed concurrently, unlike the rollback journal.\n\n' +
       'Subscribe to our newsletter.\n\n'*8)
_f = focus(_md, 'wal mode concurrent readers writers', chars=200)
assert 'concurrently' in _f and len(_f) <= 260, _f
assert 'concurrently' not in _md[:200]      # head truncation would have missed it

# a Cloudflare interstitial is a 200 that must NOT reach the corpus
class _P:
    def __init__(self, status=200, html=''): self.status, self.html_content, self.html = status, html, html
assert not usable(_P(200, '<html><head><title>Just a moment...</title></head><body>Enable JavaScript and cookies to continue</body></html>'))
assert not usable(_P(403, '<html>' + 'x'*500 + '</html>'))
assert not usable(None)
assert usable(_P(200, '<html><body>' + 'real prose about sqlite. '*40 + '</body></html>'))
assert _why(_P(429)) == 'http 429' and _why(None) == 'fetch failed'

# where and what-language are separate questions with separate answers
assert infer_country('rebuild my garage in australia') == 'au'
assert infer_country('garage rebuild sydney quotes') == 'au'           # cities count
assert infer_country('planning permission in the UK') == 'uk'
assert infer_country('compare australia and germany building codes') is None   # a comparison: neither
assert infer_country('garage rebuild cost') is None
assert infer_country('give us a quote') is None                        # bare 'us' is the pronoun
assert infer_country('australianism') is None                          # word boundaries

assert infer_language('what is the best way to do this for my house') == 'en'
assert infer_language('welche Vorschriften gelten für eine Garage in Österreich') == 'de'
assert infer_language('garage') is None                                # one word is not evidence
assert infer_language('la') is None                                    # shared across languages

# the join: country from the subject, language from the asker
assert infer_region('I want to rebuild my garage in australia, what is the regulation') == 'au-en'
assert infer_region('garage rebuild sydney') == 'au-en'
assert infer_region('do I need planning permission for a garage in the UK') == 'uk-en'
assert infer_region('what is the cost of living in france') == 'fr-en'  # French sources, English pages
assert infer_region('welche Vorschriften gelten für eine Garage in Österreich') == 'at-de'
assert infer_region('aktuelle KI Regulierung Österreich Unternehmen 2026') == 'at-de'  # no function words: fall to the country
assert infer_region('compare australia and germany building codes') == 'us-en'         # abstains to the default
assert infer_region('garage rebuild cost') == 'us-en'
assert infer_region('', dflt='au-en') == 'au-en' and infer_region(None) == 'us-en'
assert _region('garage in australia', 'uk-en') == 'uk-en'              # explicit beats inferred

# region and timelimit reach every backend: search() used to build the fused call without **kw,
# so `timelimit=` was accepted, ignored, and indistinguishable from having no effect.
_calls = []
def _spy(be, q, region='us-en', page=1, timeout=8, tries=2, **kw):
    _calls.append(dict(be=be, region=region, **kw))
    return [dict(title=be, href=f'https://e.com/{be}', body='', engine=be, rank=1)]
_real, _engine_hits = _engine_hits, _spy
try:
    search('sqlite wal mode', timelimit='y')
    assert [c['timelimit'] for c in _calls] == ['y']*len(TEXT_BACKENDS), _calls
    assert {c['region'] for c in _calls} == {'us-en'}

    _calls.clear(); search('rebuild my garage in australia')
    assert {c['region'] for c in _calls} == {'au-en'}, _calls
    assert not any('timelimit' in c for c in _calls)          # absent, not None: ddgs defaults it

    _calls.clear(); search('garage in australia', region='uk-en', timelimit='m', safesearch='off')
    assert {c['region'] for c in _calls} == {'uk-en'}
    assert all(c['timelimit'] == 'm' and c['safesearch'] == 'off' for c in _calls)

    _calls.clear(); assert search('  ') == [] and _calls == []   # empty query still touches no backend
finally: _engine_hits = _real

# a page's publication date, from the three places a page states it -- and None when it states none
class _D:
    def __init__(self, html='', url=''): self.html, self.url = html, url

_ld = lambda s: _D(f'<script type="application/ld+json">{s}</script>')
assert page_date(_ld('{"@type":"Article","datePublished":"2024-03-07T09:00:00Z"}')) == '2024-03-07'
assert page_date(_ld('{"@graph":[{"@type":"WebSite"},{"datePublished":"2019-11-02"}]}')) == '2019-11-02'
assert page_date(_ld('{"datePublished":"not a date","dateModified":"2024-05-05"}')) == '2024-05-05'
assert page_date(_D('<meta property="article:published_time" content="2022-06-15T10:11:12+00:00">')) == '2022-06-15'
assert page_date(_D('<meta name="DC.date.issued" content="2021-1-5">')) == '2021-01-05'      # zero-padded
assert page_date(_D('<meta name="viewport" content="width=1"><meta name="date" content="2025-12-31">')) == '2025-12-31'
assert page_date(_D('<time datetime="2023-08-01">last August</time>')) == '2023-08-01'
assert page_date(_D('', 'https://blog.example.com/2018/09/wal-mode')) == '2018-09-01'   # month only -> day 01
assert page_date(_D('<p>nothing dated here</p>', 'https://e.com/x')) is None            # undated stays undated

# a malformed JSON-LD block must not swallow the meta tag underneath it
assert page_date(_D('<script type="application/ld+json">{oops</script>'
                    '<meta name="pubdate" content="2020-02-29">')) == '2020-02-29'

# the month/day alternation has to prefer the longest branch, or 2019-11-02 reads as 2019-01-01
assert _iso('2019-11-02') == '2019-11-02' and _iso('2025-12-31') == '2025-12-31'
assert _iso('v1.2 build 7') is None

# research() reports the region it resolved, and keeps one shape whether or not it found anything
_r = research('')
assert _r['sources'] == [] and _r['digest'] == ''
assert sorted(_r) == ['curation', 'digest', 'dropped', 'plan', 'query', 'region', 'sources']
assert _r['plan'] == []
assert _r['region'] == 'us-en' and research('', region='au-en')['region'] == 'au-en'

# text search — every backend queried in parallel, fused with RRF, then reranked
for r in search('fasthtml python web framework', n=5):
    print(f"{r['score']:.4f} {str(r.get('engines')):40s} {r['title'][:50]} — {r['href']}")

0.0320 ['brave', 'startpage']                   Python FastHTML: A Beginner’s Guide with Examples  — https://www.datacamp.com/tutorial/python-fasthtml
0.0318 ['brave', 'startpage']                   FastHTML - Modern web applications in pure Python — https://fastht.ml/
0.0315 ['brave', 'startpage']                   FastHTML: Revolutionizing Web Development with Pyt — https://www.geeksforgeeks.org/python/fasthtml-modern-web-application-in-pure-python/
0.0315 ['brave', 'startpage']                   python-fasthtml · PyPI — https://pypi.org/project/python-fasthtml/
0.0296 ['brave', 'startpage']                   GitHub - AnswerDotAI/fasthtml: The fastest way to  — https://github.com/AnswerDotAI/fasthtml

for r in search('red panda', 'images',n=3): print(r['title'], '—', r['image'])

Happy Red Panda — https://animalfactguide.com/wp-content/uploads/2020/12/red_panda_220-scaled.jpg
Cute Red Panda — https://i.natgeofe.com/k/63b1a8a7-0081-493e-8b53-81d01261ab5d/red-panda-full-body_16x9.jpg
Red Panda Animal Endangered Red Pandas Seized From Smuggler In Laos, — https://cincinnatizoo.org/wp-content/uploads/2012/04/Red-Panda-Image-for-Endangered-Excursion-scaled.jpg

# news search
for r in search('open source LLMs', 'news', n=3, pages=2): print(r['date'], r['title'])

2025-03-20T10:45:00+00:00 The Open-Source LLM Revolution: Transforming Enterprise AI For A New Era
2026-07-25T00:15:00+00:00 Nvidia, other tech giants caution against open-source AI ban in open letter
2023-06-20T00:00:00+00:00 Open-source AI chatbots are booming — what does this mean for researchers?

for r in videos(q='aditya hrudayam', n=1): print(r)

ERROR: query "aditya hrudayam" page 1: Unable to download API page: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1018) (caused by CertificateVerifyError('[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1018)')); please report this issue on  https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using  yt-dlp -U

search_yt error: ERROR: query "aditya hrudayam" page 1: Unable to download API page: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1018) (caused by CertificateVerifyError('[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1018)')); please report this issue on  https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template. Confirm you are on the latest version using  yt-dlp -U

# real Google ranking via stealth browser (slower; use when you need Google specifically)
for r in google('vedicreader website', n=5): print(r['title'], '—', r['href'])

[2026-08-09 15:39:05] INFO: Fetched (200) <GET https://www.google.com/search?q=vedicreader+website&hl=en&num=7&sei=dRJ4arGzHuGihvcP0anygAk> (referer: https://www.google.com/)

VedicReader — https://vedicreader.com/
VedicReader — https://vedicreader.com/a/lgn
vedicreader.com to help absorb vedic texts easily — https://www.reddit.com/r/AdvaitaVedanta/comments/1lpr29l/vedicreadercom_to_help_absorb_vedic_texts_easily/
VedSearch — Read the Vedas in Sanskrit, Hindi & English — https://vedsearch.org/
Ved Portal - Search & Read — https://www.xn--j2b3a4c.com/en/

# research(): search -> read the top n results -> one cited markdown corpus
res = research('what is the raft consensus algorithm', n=3)
assert set(res) == {'query', 'sources', 'digest', 'dropped'}, res.keys()
assert res['query'] and isinstance(res['sources'], list) and res['sources']
assert all({'title','href','md','tier'} <= set(s) for s in res['sources'])
assert res['digest']
print(len(res['sources']), 'sources |', len(res['digest']), 'chars |', len(res['dropped']), 'skipped')
for d in res['dropped']: print('  skipped:', d['reason'], d['href'])

[2026-08-09 15:39:10] INFO: Fetched (200) <GET https://www.geeksforgeeks.org/system-design/raft-consensus-algorithm/> (referer: https://www.google.com/)
[2026-08-09 15:39:10] INFO: Fetched (200) <GET https://raft.github.io/> (referer: https://www.google.com/)
[2026-08-09 15:39:10] INFO: Fetched (200) <GET https://en.wikipedia.org/wiki/Raft_(algorithm)> (referer: https://www.google.com/)

3 sources | 10171 chars | 0 skipped