core

Building blocks for litesearch

Introduction

We often have to go through a whole bunch of hoops to get documents processed and ready for searching through them. litesearch plans to make this as easy as possible by providing simple building blocks to set up a database with FTS5 and vector search capabilities.

porter unicode61, the FTS5 default, splits on _ and stems the pieces, so fts_search indexes as ft+search. apsw’s UAX#29 tokenizer joins on _, so identifiers survive and porter still stems prose. sanskrit only adds colocated tokens, so the chain is additive and is the default: a store’s tokenizer is fixed at table creation, and the first document should not decide it.

sanskrit sits inside porter. Outside it, the fold is computed on porter’s output: Devanagari passes porter untouched, so धर्मक्षेत्रे indexes as dharmaksetre while the ASCII query stems to dharmaksetr and misses. Measured: 3 of 14 ordinary Devanagari words unreachable by their own ASCII spelling.

_tokenizers_ok probes the whole chain, not its names one at a time. simplify wraps another tokenizer and raises when asked for alone, so name-by-name probing fails on a working chain.

busy_window widens apsw’s busy timeout for a block and restores it. Refcounted, so overlapping windows cannot strand the widened value.

source

process_content

def process_content(
    store, # target Table (hash-id store)
    content, # iterable of chunk dicts
    embed:bool=True, # embed content before upsert
    emb_fn:NoneType=None, # embedder, required when embed=True
    hash_id_columns:tuple=('content',), # columns the row id is hashed over; add 'node_id' to keep identical text in different sections distinct
    parallel:bool=False, # widen the busy timeout so concurrent writers wait rather than fail
    chunk:int=5000, # rows per transaction
    **kw
):

Embed and upsert chunks in chunk-sized transactions. parallel=True gives concurrent writers a 30-second busy timeout.

source

write_txn

def write_txn(
    db
):

Transaction using BEGIN IMMEDIATE; nested calls join the open transaction.

source

db_lock

def db_lock(
    db
):

Re-entrant lock guarding litesearch operations on one connection.

source

busy_window

def busy_window(
    busy_ms:int=30000, *dbs
):

Widen apsw’s busy timeout for the duration of the block, then restore it.

source

embed_chunk

def embed_chunk(
    chunk, # iterable of dicts with a 'content' key
    emb_fn, # callable: list[str] -> list of vectors (np arrays)
    **kw
):

Embed each chunk’s content via emb_fn and attach embedding bytes. Skips blank content.

source

content_id

def content_id(
    content
):

Call self as a function.

source

sql_in

def sql_in(
    col, vals, # SQL IN clause; callers guard against empty vals
):

Call self as a function.

get_store and get_tree run idempotent DDL. Read paths call them per query. ensured caches their table handles per connection, so later reads run no DDL.

source

Database.ensured

def ensured():

Table handles built on this connection, keyed by store configuration.

source

Database.forget_ensured

def forget_ensured(
    store:str=None
):

Forget cached table handles and ANN indices after dropping stores.

Simple Docs table setup

source

Database.get_store

def get_store(
    name:str='store', # table name
    hash:bool=False, # whether to create hash index on content
    ann:bool=False, # also register a usearch HNSW index (sidecar) for this store
    ndim:int=None, # embedding dims for the ANN index (inferred on first sync if None)
    metric:str='cosine', # ANN distance metric (cosine,inner,sqeuclidean,divergence)
    dtype:type=float16, # ANN vector dtype
    connectivity:int=None, # HNSW connectivity (usearch default if None)
    expansion_add:int=None, # HNSW add expansion
    expansion_search:int=None, # HNSW search expansion
    index_path:str=None, # sidecar path (defaults to <db>.<name>.usearch; None db -> in-memory)
    tokenize:str=None, # FTS5 tokenizer (default: apsw unicodewords chain, else 'porter')
    **kw
):

Create an FTS5 store with optional content hashes and a usearch HNSW index.

source

upsert_all

def upsert_all(
    tbl, # target Table, with `pk` declared as its primary key
    rows, # dicts of scalars, upserted on `pk`
    pk, # primary-key column name or tuple of them
    chunk:int=2000, # rows per `executemany` batch
):

Upsert rows with one prepared statement instead of two per row.

source

Database.get_index

def get_index(
    name:str='ann_store', # ANN store name
    load:bool=True, # load the sidecar file if present
):

Build (and cache) the usearch Index for an ANN store. Requires a known ndim.

source

Database.ann_indices

def ann_indices():

Per-connection cache of loaded usearch Index objects, keyed by store name.

_dtype_check warns when dtype cannot be what these vectors were stored as.

The failure is silent and total. model2vec returns float32 and FastEncode float16. Store one and search as the other, and distance_cosine_f16 reads f32 bytes as twice as many f16 values, whose inf/nan patterns return SQL NULL for every row. The vector leg comes back in rowid order and a hybrid search degrades to FTS.

source

source

source

Table.rebuild_index

def rebuild_index(
    dtype:NoneType=None
):

Rebuild the ANN Index from scratch using the embedding blobs in SQLite. Returns index size.

source

Table.sync

def sync(
    content, # iterable of chunk dicts with 'content'
    key_col:str=None, # scope existing rows by this column's values (e.g. 'path','package'); None = whole table
    emb_fn:NoneType=None, # embedder for new/changed chunks
    embed:bool=True, # embed before upsert
    force:bool=False, # treat all content as new (skip diff)
    parallel:bool=False, # chunked BEGIN IMMEDIATE writes; needs database(busy_timeout=...)
    chunk:int=5000, # rows per transaction when parallel=True
):

Smart hash-diff update: delete stale rows, embed+upsert changed, mirror into the ANN index. Returns {changed,same,removed}.

source

source

rowid_sel

def rowid_sel():

Call self as a function.

Loading in bulk

FTS5 triggers index each row as it is written. bulk_load removes them and rebuilds FTS once after the load. Use it only for offline loads: FTS stays stale until the rebuild, and rebuilding a small update costs more than trigger maintenance.

bulk_load

def bulk_load():

Suspend this table’s FTS5 triggers for a bulk insert, then rebuild the index once.

Neighbours of an indexed row

ann_neighbors answers “what is near this row” using the vector usearch already holds, rather than re-embedding its text. Cheap enough to fire from a cursor move.

ann_neighbors

Rows nearest an already-indexed row, “what else looks like this”.

Reuses the stored vector rather than re-embedding the row, so no model is loaded and no text is tokenised: the whole call is one HNSW probe plus one rowid IN (...) fetch.

source

Table.ann_neighbors

def ann_neighbors(
    key:int, # usearch key (rowid) to search from
    limit:int=15, # neighbours to return
    columns:list=None, # columns to return (rowid always included)
    where:str=None, # additional where clause
    where_args:dict=None, # args for where clause
    include_self:bool=False, # keep the anchor row in the results
    dtype:type=float16, # embedding dtype
):

Rows nearest an already-indexed row — “what else looks like this”.

source

Table.ann_vec

def ann_vec(
    key:int, # usearch key (the row's rowid)
    dtype:type=float16, # fallback dtype when the registry has none
):

The vector the index already holds for key, or None when the key is not indexed.

source

rrf_merge

def rrf_merge(
    fts, vec, k:int=60, limit:int=50, id_key:str='rowid'
)->list:

Two-list rrf_all. Deprecated: call rrf_all directly.

source

rrf_all

def rrf_all(
    lists, # ranked result lists to fuse
    k:int=60, # rank at which a hit is worth half the top spot
    limit:int=50, # results returned
    id_key:str='rowid', # the key rows are joined on
    weights:NoneType=None, # one weight per list; None means all equal
)->list:

Reciprocal Rank Fusion over any number of ranked lists. A row in two lists outranks a row in one.

from litesearch.core import rrf_all, rrf_merge

fts = [dict(rowid=1, t='a'), dict(rowid=2, t='b'), dict(rowid=3, t='c')]
vec = [dict(rowid=3, t='c'), dict(rowid=4, t='d')]

# rowid 3 is 3rd on one list and 1st on the other, which beats anything ranked once
fused = rrf_all([fts, vec])
assert [r['rowid'] for r in fused][:2] == [3, 1]
assert {r['rowid'] for r in fused} == {1, 2, 3, 4}
assert rrf_merge(fts, vec) == fused

# a weight is a thumb on the scale: doubling the vector leg lifts its top hit past the FTS top hit
assert [r['rowid'] for r in rrf_all([fts, vec], weights=[1.0, 2.0])][:2] == [3, 4]

register_tokenizers returns None on success, so _fts_tokenizers = register(...) was always falsy and get_store fell back to bare porter. The UAX#29 chain had never reached a store.

Registered on every connection, because FTS5 resolves a tokenizer by name at query time.

source

database

def database(
    pth_or_uri:str=':memory:', # the database name or URL
    wal:bool=True, # use WAL mode
    sem_search:bool=True, # enable usearch extensions
    busy_timeout:int=None, # ms to wait on a locked db; None = apsw default. Set for parallel ingestion
    **kw
)->Database: # additional args to pass to apswutils database

Set up a database connection and load usearch extensions.

Parallel ingestion. apswutils runs PRAGMA optimize inside apsw’s open hook with a 100 ms busy timeout, so concurrent writers raised BusyError and dropped whole batches silently.

import threading, tempfile, os, apsw, apsw.bestpractice
_orig = apsw.bestpractice.connection_busy_timeout.__defaults__

# write_txn commits, rolls back on error, and nests without opening a second transaction
_wdb = database(sem_search=False); _wt = _wdb.get_store(name='w', hash=True)
with write_txn(_wdb): _wt.insert({'content':'a'})
try:
    with write_txn(_wdb): _wt.insert({'content':'b'}); raise ValueError('boom')
except ValueError: pass
assert len(_wt()) == 1, 'a failed write_txn must roll back'
with write_txn(_wdb):
    with write_txn(_wdb): _wt.insert({'content':'c'})   # nested is a no-op
assert len(_wt()) == 2 and not _wdb.conn.in_transaction

busy_window widens inside, and only the outermost exit restores

with busy_window(30_000):
    assert apsw.bestpractice.connection_busy_timeout.__defaults__ == (30_000,)
    with busy_window(30_000): pass
    assert apsw.bestpractice.connection_busy_timeout.__defaults__ == (30_000,), 'inner exit must not restore'
assert apsw.bestpractice.connection_busy_timeout.__defaults__ == _orig

8 threads into one database needs both halves: busy_timeout= on database(), which opens inside a busy_window, and parallel=True on process_content, which chunks writes into BEGIN IMMEDIATE transactions. The open dominates: with the stock 100 ms timeout this load passed 9 of 15 runs.

_cp = os.path.join(tempfile.mkdtemp(), 'parallel.db')
_cmain = database(_cp, sem_search=False, busy_timeout=30_000); _cmain.get_store(name='store', hash=True)
_cerrs, _cconns, _cesc = [], [], []
_prev_hook = threading.excepthook
threading.excepthook = lambda a: _cesc.append(type(a.exc_value).__name__)  # thread errors never reach us otherwise
def _cingest(tid):
    try:
        _d = database(_cp, sem_search=False, busy_timeout=30_000); _cconns.append(_d)
        process_content(_d.t['store'], [{'content':f't{tid}-r{i}', 'metadata':'{}'} for i in range(200)],
                        embed=False, parallel=True, chunk=50)
    except Exception as e: _cerrs.append((tid, type(e).__name__))
_cts = [threading.Thread(target=_cingest, args=(i,)) for i in range(8)]
for _t in _cts: _t.start()
for _t in _cts: _t.join()
threading.excepthook = _prev_hook
assert not _cerrs and not _cesc, (_cerrs, _cesc)
assert list(_cmain.conn.execute('SELECT count(*) FROM store'))[0][0] == 8*200, 'rows lost under contention'
assert apsw.bestpractice.connection_busy_timeout.__defaults__ == _orig, 'busy_window leaked'
# `db_lock` is what makes the read-modify-write paths safe when threads share *one* connection.
# The busy_window test above does not reach it: there every thread opens its own connection, so
# they contend in SQLite rather than in python. Here the hash-id upserts interleave on the same
# handle, which is the case the lock exists for.
_sp = os.path.join(tempfile.mkdtemp(), 'shared.db')
_smain = database(_sp, sem_search=False); _smain.get_store(name='store', hash=True)
_sst = _smain.t['store']
assert db_lock(_smain) is db_lock(_smain), 'one lock per connection, not one per call'
_serrs, _sesc = [], []
_prev_hook = threading.excepthook
threading.excepthook = lambda a: _sesc.append(type(a.exc_value).__name__)
def _singest(tid):
    try: process_content(_sst, [{'content':f's{tid}-r{i}', 'metadata':'{}'} for i in range(100)],
                         embed=False, chunk=25)
    except Exception as e: _serrs.append((tid, type(e).__name__))
_sts = [threading.Thread(target=_singest, args=(i,)) for i in range(8)]
for _t in _sts: _t.start()
for _t in _sts: _t.join()
threading.excepthook = _prev_hook
assert not _serrs and not _sesc, (_serrs, _sesc)
assert list(_smain.conn.execute('SELECT count(*) FROM store'))[0][0] == 8*100, 'rows lost on a shared connection'
assert apsw.bestpractice.connection_busy_timeout.__defaults__ == _orig, 'busy_window leaked'
# Table.sync forwards parallel/chunk down to process_content, so hash-diff syncs (what kosha's
# package ingestion uses) get the same chunked BEGIN IMMEDIATE writes as a direct process_content.
_sdb = database(sem_search=False, busy_timeout=30_000); _sst = _sdb.get_store(name='sy', hash=True)
_srows = [{'content': f'row {i}', 'metadata': '{}'} for i in range(120)]
assert _sst.sync(_srows, embed=False, parallel=True, chunk=25) == {'changed':120,'same':0,'removed':0}
assert len(_sst()) == 120
assert _sst.sync(_srows, embed=False, parallel=True, chunk=25) == {'changed':0,'same':120,'removed':0}  # idempotent
assert _sst.sync(_srows[:60], embed=False, parallel=True, chunk=25) == {'changed':0,'same':60,'removed':60}
assert len(_sst()) == 60

source

rerank_hits

def rerank_hits(
    q, hits, model:NoneType=None, limit:NoneType=None, text_col:str='content'
):

Reorder search hits by a flashrank cross-encoder on (q, hit[text_col]); returns the top limit.

source

Database.search

def search(
    q:str, # query string
    emb:bytes, # embedding vector
    columns:list=None, # columns to return
    where:str=None, # additional where clause
    where_args:dict=None, # args for where clause
    limit:int | None=50, # limit on number of results
    offset:int | None=None, # offset for results (ignored when rrf=True)
    table_name:str='store', # table name
    emb_col:str='embedding', # embedding column name
    emb_metric:str='cosine', # embedding distance metric (cosine,sqeuclidean,inner,divergence)
    rrf:bool=True, # rerank results with reciprocal rank fusion
    rrf_k:int=60, # RRF k parameter
    dtype:type=float16, # embedding dtype
    id_key:str='rowid', # key to join RRF results on
    quote:bool=True, # quote FTS query to disable special chars (ignored when fts_pre=True)
    fts_pre:bool=True, # send the FTS leg through `pre()`: keywords, wildcards, OR
    ann:bool=False, # use the HNSW ANN index for the vector leg falls back to the exact vec scan when `where` is set
    parallel:bool=False, # not used. kept for backward compatability. to be removed in later releases
    reranking:bool=False, # rerank the merged (rrf) hits with a flashrank cross-encoder
    rerank_model:str=None, # flashrank model name (None -> fast default)
):

Search the litesearch store with fts and vector search combined.

Two legs, one connection, two threads is what apsw refuses, and it bought nothing anyway: the work is in consuming the vector leg, not in issuing the two queries, so threading them only added pool overhead. Concurrent now means a connection each, and the default is sequential.

def _v(i, n=8): return (np.full(n, i, dtype=np.float16)/16).tobytes()
_db = database(str(Path(mkdtemp())/'p.db'))
_st = _db.get_store()
_st.insert_all([dict(content=f'rank fusion over leg {i}', embedding=_v(i)) for i in range(200)])
_seq = _db.search('rank fusion', _v(3), ['content'], limit=5)
_par = _db.search('rank fusion', _v(3), ['content'], limit=5, parallel=True)
test_eq([h['content'] for h in _seq], [h['content'] for h in _par])   # same answer either way
assert _seq, _seq
# an in-memory database has no file to reopen, so the concurrent path steps back to sequential
_mem = database(':memory:'); _mem.get_store().insert_all([dict(content='rank fusion', embedding=_v(3))])
assert _mem.search('rank fusion', _v(3), ['content'], limit=5, parallel=True)
/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/ipykernel_91989/3569258371.py:28: UserWarning: 'store': every distance came back null or 0, which usually means dtype='f16' does not match how the embeddings were stored (e.g. float32 vectors from model2vec searched as float16). The ranking is meaningless.
  warnings.warn(f'{tbl.name!r}: every distance came back null or 0, which usually means dtype={want!r} '

The FTS leg goes through pre() by default: stopwords out, wildcard per term, terms OR-ed. Quoting every token makes FTS an implicit AND, so a query with one absent term matched nothing.

_d = database(); _s = _d.get_store('store')
_s.insert_all([dict(content="The Pathfinder's transition activities shall help all types of researchers",
                    embedding=np.zeros(4, dtype=np.float16).tobytes()),
               dict(content='Value added tax shall be chargeable on the supply of goods',
                    embedding=np.ones(4, dtype=np.float16).tobytes())])
_qv = np.zeros(4, dtype=np.float16).tobytes()
_q  = 'Pathfinder researchers assistance'          # 'assistance' is absent from both rows

assert _d.search(_q, _qv, columns=['content'], rrf=False, fts_pre=False)['fts'] == []
_got = _d.search(_q, _qv, columns=['content'], rrf=False, fts_pre=True)['fts']
assert len(_got) == 1 and 'Pathfinder' in _got[0]['content']

# a query of pure stopwords cannot be pre()-processed into anything; fall back rather than crash
assert _d.search('of the', _qv, columns=['content'], limit=2) is not None

reranking reorders by content relevance (downloads a flashrank model; run manually, not CI)

import numpy as np, json
_rdb = database(); _rst = _rdb.get_store('rr', ann=True, ndim=4)
_rst.insert_all([{'content':'The Eiffel Tower is in Paris, France.','embedding':np.array([1,0,0,0],dtype=np.float16).tobytes(),'metadata':json.dumps({'doc_id':'a'})},
                 {'content':'Cats are small mammals kept as pets.','embedding':np.array([0,1,0,0],dtype=np.float16).tobytes(),'metadata':json.dumps({'doc_id':'b'})}])
_rst.rebuild_index()
_q = 'where is the eiffel tower'
# query vector deliberately favors 'b'; the cross-encoder should still surface 'a' first
_hits = _rdb.search(_q, np.array([0,1,0,0],dtype=np.float16).tobytes(), columns=['metadata'], limit=2, table_name='rr', reranking=True)
assert json.loads(_hits[0]['metadata'])['doc_id']=='a'

A float32 store searched as float16: every distance comes back NULL and nothing is raised. The guard turns that into a warning instead of a silently useless vector leg.

# Table names are identifiers, including names such as a routed `client-old` shelf.
_hdb = database(); _hst = _hdb.get_store('client-old', ann=True, ndim=4)
_hv = np.array([1,0,0,0], dtype=np.float16)
_hst.insert(dict(content='hyphenated shelf', embedding=_hv.tobytes()))
assert _hst.vec_search(_hv.tobytes(), ['content'])[0]['content'] == 'hyphenated shelf'
assert _hst.rebuild_index() == 1
assert _hst.ann_search(_hv.tobytes(), ['content'])[0]['content'] == 'hyphenated shelf'
import warnings
_rng = np.random.default_rng(0)
_nrm = lambda x: x/np.linalg.norm(x)
_a, _b = (_nrm(_rng.normal(size=64)).astype(np.float32) for _ in range(2))
_d = database(); _s = _d.get_store('store')
_s.insert_all([dict(content='a', embedding=_a.tobytes()), dict(content='b', embedding=_b.tobytes())])
with warnings.catch_warnings(record=True) as _w:
    warnings.simplefilter('always')
    _s.vec_search(_a.tobytes(), ['content'], dtype=np.float16)
    assert len(_w) == 1 and 'distance came back' in str(_w[0].message), [str(x.message) for x in _w]
with warnings.catch_warnings(record=True) as _w:          # dtype matched: silent
    warnings.simplefilter('always')
    assert _s.vec_search(_a.tobytes(), ['content'], dtype=np.float32)[0]['content'] == 'a'
    assert not _w, [str(x.message) for x in _w]

an ANN-registered store knows its dtype, so the mismatch is caught exactly rather than guessed

_d2 = database(); _s2 = _d2.get_store('store', ann=True, ndim=64, dtype=np.float32)
_s2.insert_all([dict(content='a', embedding=_a.tobytes())])
with warnings.catch_warnings(record=True) as _w:
    warnings.simplefilter('always')
    _s2.vec_search(_a.tobytes(), ['content'], dtype=np.float16)
    assert len(_w) == 1 and 'registered as' in str(_w[0].message), [str(x.message) for x in _w]
db = database()

The fastlite database is set up with usearch extensions. Let’s run some distance calculations.

embs = dict(
    v1=np.ones((100,),dtype=np.float32).tobytes(),      # vector of ones
    v2=np.zeros((100,),dtype=np.float32).tobytes(),     # vector of zeros
    v3=np.full((100,),0.25,dtype=np.float32).tobytes()  # vector of 0.25s
)
def dist_q(metric):
    return db.q(f'''
        select
            distance_{metric}_f32(:v1,:v2) as {metric}_v1_v2,
            distance_{metric}_f32(:v1,:v3) as {metric}_v1_v3,
            distance_{metric}_f32(:v2,:v3) as {metric}_v2_v3
    ''', embs)

for fn in ['sqeuclidean', 'divergence', 'inner', 'cosine']: print(dist_q(fn))
[{'sqeuclidean_v1_v2': 100.0, 'sqeuclidean_v1_v3': 56.25, 'sqeuclidean_v2_v3': 6.25}]
[{'divergence_v1_v2': 34.657352447509766, 'divergence_v1_v3': 12.046551704406738, 'divergence_v2_v3': 8.66433334350586}]
[{'inner_v1_v2': 1.0, 'inner_v1_v3': -24.0, 'inner_v2_v3': 1.0}]
[{'cosine_v1_v2': 1.0, 'cosine_v1_v3': 0.0, 'cosine_v2_v3': 1.0}]
db.get_store()
if 'store' in db.t: print('store is created')
print('detected fts table: ',db.t.store.detect_fts())
print('Search results:', len(db.search('h',np.zeros((100,)).tobytes()))) # there is no data yet, so should be 0
store is created
detected fts table:  store_fts
Search results: 0

We can also create a store with hash index on content. Useful for code search applications

st=db.get_store(name='my_store', hash=True)
st.insert_all([dict(content='hello world', embedding=np.ones((100,),dtype=np.float16).tobytes()),
                           dict(content='hi there', embedding=np.full((100,),0.5,dtype=np.float16).tobytes()),
                           dict(content='goodbye now', embedding=np.zeros((100,),dtype=np.float16).tobytes())],upsert=True,hash_id='id')
st(select='id,content')
[{'id': '250ce2bffa97ab21fa9ab2922d19993454a0cf28', 'content': 'hello world'},
 {'id': 'c89f43361891bfab9290bcebf182fa5978f89700', 'content': 'hi there'},
 {'id': '882293d5e5c3d3e04e8e0c4f7c01efba904d0932', 'content': 'goodbye now'}]

Let’s run a search again.

db.search(q='hello', emb=np.full((100,),0.25, dtype=np.float16).tobytes(), columns=['content'], table_name='my_store',limit=2, rrf=False, quote=True)
/var/folders/kg/9vdw4mdd1fs58svgh4k1qhr09x7dqh/T/ipykernel_91989/3569258371.py:28: UserWarning: 'my_store': every distance came back null or 0, which usually means dtype='f16' does not match how the embeddings were stored (e.g. float32 vectors from model2vec searched as float16). The ranking is meaningless.
  warnings.warn(f'{tbl.name!r}: every distance came back null or 0, which usually means dtype={want!r} '
{'fts': [{'content': 'hello world', 'rank': -0.5108256237659907}],
 'vec': [{'content': 'hello world', '_dist': 0.0},
  {'content': 'hi there', '_dist': 0.0}]}
db.t.my_store.fts_search(q='hello', columns=['content'], limit=2)
[{'content': 'hello world', 'rank': -0.5108256237659907}]
db.search(q='hello', emb=np.full((100,),0.25, dtype=np.float16).tobytes(), columns=['content'], table_name='my_store',limit=1,offset=1, rrf=False, quote=True)
{'fts': [], 'vec': [{'content': 'hi there', '_dist': 0.0}]}

Now, let’s try the same but with a broader query.

db.search(q='goodbye OR hi', emb=np.full((100,),0,dtype=np.float16).tobytes(), columns=['content'], table_name='my_store',limit=2, quote=True)
[{'rowid': 3,
  'content': 'goodbye now',
  'rank': -0.5108256237659907,
  '_rrf_score': 0.03306010928961749},
 {'rowid': 2,
  'content': 'hi there',
  'rank': -0.5108256237659907,
  '_rrf_score': 0.016666666666666666}]

You can use different kind of embedding metrics as well. The default is cosine. Let’s try with divergence distance

db.search(q='goodbye OR hi', emb=np.full((100,),0,dtype=np.float16).tobytes(), columns=['content'], table_name='my_store',limit=2, emb_metric='divergence', quote=True)
[{'rowid': 2,
  'content': 'hi there',
  'rank': -0.5108256237659907,
  '_rrf_score': 0.03306010928961749},
 {'rowid': 3,
  'content': 'goodbye now',
  'rank': -0.5108256237659907,
  '_rrf_score': 0.03306010928961749}]

ANN store, usearch HNSW

get_store(ann=True) keeps the SQLite table as the source of truth and maintains a usearch HNSW index as a rebuildable sidecar. Table.sync hash-diffs and mirrors changes into it; db.search(..., ann=True) uses it for the vector leg.

The tests below use a deterministic embedder.

# tiny deterministic embedder: content -> 4-d f16 vector
_vecs = {'A':[1,0,0,0], 'B':[0,1,0,0], 'C':[0,0,1,0], 'B2':[0,1,1,0], 'D':[0,0,0,1]}
def emb_fn(texts): return [np.array(_vecs.get(t, [0.01,0.01,0.01,0.01]), dtype=np.float16) for t in texts]

adb = database()
ast_ = adb.get_store('ann_store', hash=True, ann=True, metric='cosine')  # ndim inferred on first sync
assert 'usearch_indices' in adb.t, 'registry table created'
assert adb._ann_meta('ann_store')['ndim'] is None, 'ndim unknown until first sync'

stats = ast_.sync([dict(content=c) for c in ['A','B','C']], emb_fn=emb_fn)
print('sync stats:', stats)
assert stats == dict(changed=3, same=0, removed=0)
assert adb._ann_meta('ann_store')['ndim'] == 4, 'ndim inferred and persisted'
assert adb.get_index('ann_store').size == 3, 'index has 3 vectors'
sync stats: {'changed': 3, 'same': 0, 'removed': 0}
# ANN vector search returns the nearest doc (query == C's vector)
qC = np.array([0,0,1,0], dtype=np.float16).tobytes()
hit = ast_.ann_search(qC, columns=['content'], limit=1)
print('ann top hit:', hit)
assert hit[0]['content'] == 'C'
ann top hit: [{'content': 'C', 'rowid': 3, '_dist': 0.0}]
# hybrid search with ann=True (fts + HNSW -> rrf)
res = adb.search('C', qC, columns=['content'], table_name='ann_store', ann=True, limit=3, quote=True)
assert res[0]['content'] == 'C', res
# smart update: A unchanged, B->B2 (changed), C removed, D new
stats = ast_.sync([dict(content=c) for c in ['A','B2','D']], emb_fn=emb_fn)
print('update stats:', stats)
assert stats == dict(changed=2, same=1, removed=2), stats   # add B2,D ; remove old-B,C
assert {r['content'] for r in ast_()} == {'A','B2','D'}
assert adb.get_index('ann_store').size == 3
# D is now findable, C is gone
qD = np.array([0,0,0,1], dtype=np.float16).tobytes()
assert ast_.ann_search(qD, columns=['content'], limit=1)[0]['content'] == 'D'
update stats: {'changed': 2, 'same': 1, 'removed': 2}
# force=True: skip the diff, re-ingest everything (nothing stale/removed, all re-embedded)
stats = ast_.sync([dict(content=c) for c in ['A','B2','D']], emb_fn=emb_fn, force=True)
assert stats == dict(changed=3, same=0, removed=0), stats
assert adb.get_index('ann_store').size == 3   # re-adds overwrite existing keys
assert ast_.ann_search(qD, columns=['content'], limit=1)[0]['content'] == 'D'
# rebuild the index from the SQLite blobs (recovery path) -- drop cached index, rebuild
adb.ann_indices.clear()
n = ast_.rebuild_index()
assert n == 3, n
assert ast_.ann_search(qD, columns=['content'], limit=1)[0]['content'] == 'D'
# persistence: on-disk db saves the .usearch sidecar; reopening loads it (no re-embed)
import tempfile
p = str(Path(tempfile.mkdtemp())/'ann.db')
d1 = database(p)
s1 = d1.get_store('ann_store', hash=True, ann=True)
s1.sync([dict(content=c) for c in ['A','B','C']], emb_fn=emb_fn)
assert Path(d1._ann_meta('ann_store')['path']).exists(), 'sidecar saved'
d1.close()

d2 = database(p)                       # fresh connection, index loaded lazily from sidecar
assert d2.get_index('ann_store').size == 3, 'index restored from disk'
assert d2.t.ann_store.ann_search(qC, columns=['content'], limit=1)[0]['content'] == 'C'
d2.close()
import numpy as np, json
# ANN on an id-pk (non-hash) store: rebuild_index + ann_search must not KeyError on 'rowid'
_annt = database().get_store('annrowid', ann=True, ndim=4, metric='cosine')
_vs = np.array([[1,0,0,0],[0,1,0,0],[0,0,1,0],[0,1,0,0]], dtype=np.float16)
_annt.insert_all([{'content':f'c{i}','embedding':_vs[i].tobytes(),'metadata':json.dumps({'doc_id':f'd{i}'})} for i in range(4)])
assert _annt.rebuild_index() == 4
_h = _annt.ann_search(np.array([1,0,0,0],dtype=np.float16).tobytes(), columns=['metadata'], limit=2)
assert _h and json.loads(_h[0]['metadata'])['doc_id']=='d0'
assert _h[0]['_dist'] is not None
_h1= _annt.ann_search(np.array([0,1,0,0],dtype=np.float16).tobytes(), columns=['metadata'], where="json_extract(metadata, '$.doc_id')='d3'",limit=2)
assert _h1 and json.loads(_h1[0]['metadata'])['doc_id']=='d3'
assert _h1[0]['_dist'] is not None
# regression: ann_search must restrict to the ANN candidate keys even when `where` is None.
# A precedence slip (`a + b if where else ''`) dropped the rowid filter entirely, turning every
# unfiltered ANN query into a full-table scan -- correct top hit, but O(table) rows and latency.
_big = database().get_store('annbig', ann=True, ndim=4, metric='cosine')
_bv = np.random.RandomState(0).randn(50,4).astype(np.float16)
_bv[0] = np.array([1,0,0,0], dtype=np.float16)
_big.insert_all([{'content':f'b{i}','embedding':_bv[i].tobytes()} for i in range(50)])
assert _big.rebuild_index() == 50
_qb = np.array([1,0,0,0], dtype=np.float16).tobytes()
_rb = _big.ann_search(_qb, columns=['content'], limit=2)          # no where clause
assert _rb[0]['content'] == 'b0', _rb[:1]
assert len(_rb) == 2, f'ann_search returned {len(_rb)} rows for limit=2'
# where_args must reach db.q as query params (not as **kwargs, which raises TypeError)
_rb2 = _big.ann_search(_qb, columns=['content'], limit=5, where='content = :c', where_args={'c':'b0'})
assert [r['content'] for r in _rb2] == ['b0'], _rb2
# _dist must be a plain python float: numpy scalars are not JSON-serializable, which breaks
# any caller that json.dumps() the hits (kosha's --as-json, its daemon and MCP server).
import json as _json
_json.dumps(_rb[0])
'{"content": "b0", "rowid": 1, "_dist": 0.0}'
# regression: ANN + a selective `where` must not lose recall. usearch post-filters its candidate
# window (Index.search has no predicate hook), so a match ranked past that window is invisible.
# db.search now routes filtered vector queries to the exact vec_search (WHERE applied before ranking).
_fdb = database()
_fst = _fdb.get_store('annfilter', ann=True, ndim=4, metric='cosine')
import json as _json
_near, _far = np.array([1,0,0,0], dtype=np.float16), np.array([0,0,0,1], dtype=np.float16)
# 20 rows sit exactly on the query vector; the single row we actually want is far from it.
_rows = [{'content':f'near{i}','embedding':_near.tobytes(),'metadata':_json.dumps({'tag':'other'})} for i in range(20)]
_rows.append({'content':'wanted','embedding':_far.tobytes(),'metadata':_json.dumps({'tag':'keep'})})
_fst.insert_all(_rows)
assert _fst.rebuild_index() == 21
_qf = _near.tobytes()
_wh = "json_extract(metadata,'$.tag')='keep'"
# the bug: raw ANN post-filter starves recall -- none of the nearest limit*2 candidates carry tag='keep'
assert _fst.ann_search(_qf, columns=['content'], limit=2, where=_wh) == [], 'ANN post-filter should miss the far match'
# the fix: db.search(ann=True, where=...) falls back to the exact vec scan and surfaces the filtered match
_res = _fdb.search('wanted', _qf, columns=['content','metadata'], where=_wh, table_name='annfilter', ann=True, rrf=False)
assert [r['content'] for r in _res['vec']] == ['wanted'], _res['vec']
# ann_neighbors: nearest rows to an already-indexed row, no re-embedding
_nb = database().get_store('nbstore', ann=True, ndim=4, metric='cosine')
_nv = np.array([[1,0,0,0],[0.99,0.01,0,0],[0.7,0.7,0,0],[0,0,1,0]], dtype=np.float16)
_nb.insert_all([{'content':f'n{i}','embedding':_nv[i].tobytes()} for i in range(4)])
assert _nb.rebuild_index() == 4
assert _nb.ann_vec(1) is not None and _nb.ann_vec(99) is None, 'vector by key; None for an unknown key'
_nn = _nb.ann_neighbors(1, limit=2, columns=['content'])
assert [r['content'] for r in _nn] == ['n1','n2'], _nn      # n0 itself excluded, n1 is its near-duplicate
assert _nb.ann_neighbors(1, limit=1, columns=['content'], include_self=True)[0]['content'] == 'n0'
assert _nb.ann_neighbors(99) == [], 'unknown key returns nothing rather than raising'