vd #

supported_filter_operators: frozenset = frozenset({'$and', '$eq', '$exists', '$gt', '$gte', '$in', '$lt', '$lte', '$ne', '$nin', '$not', '$or'})#

the full language. Adapters narrow this; search() validates against it.

Type:: Filter operators this backend can honor. Default

upsert(document)[source]#

Insert or replace document (equivalent to self[doc.id] = doc).

Return type:: None

class vd.AsyncAbstractClient(*, embedder=None, **config)[source]#

Bases: object

Base class for native async clients (the async sibling of vd.AbstractClient).

A backend implements create_collection(), get_collection(), delete_collection() and list_collections() as coroutines / async generators; get_or_create_collection(), the client escape hatch, close() and async with support come for free. Register the class with register_async_backend() so connect_async() returns it.

Parameters:

embedder (Optional[Callable[[str], list[float]]]) – A text -> vector function handed to every collection.
**config – Backend-specific connection configuration.

backend_name: str = ''#: The registry name of this backend (set by register_async_backend()).

property client: Any#: The raw async backend client — a supported, documented escape hatch.

async close()[source]#

Release backend resources (closes the raw client if it can).

Return type:: None

abstractmethod async create_collection(name, *, dimension=None, metric='cosine', **index_config)[source]#

Create a new collection; raise ValueError if it exists.

Return type:: AsyncAbstractCollection

abstractmethod async delete_collection(name)[source]#

Drop a collection; raise KeyError if absent.

Return type:: None

abstractmethod async get_collection(name)[source]#

Return an existing collection; raise KeyError if absent.

Return type:: AsyncAbstractCollection

async get_or_create_collection(name, *, dimension=None, metric='cosine', **index_config)[source]#

Return collection name, creating it if missing.

Return type:: AsyncAbstractCollection

abstractmethod list_collections()[source]#

Async-iterate collection names.

Return type:: AsyncIterator[str]

native_async: bool = True#: Real non-blocking I/O through the backend’s async SDK.

class vd.AsyncAbstractCollection[source]#

Bases: _CollectionPolicy

Base class for native async collections (the async sibling of vd.AbstractCollection).

A backend subclasses this and implements async raw primitives; the user-facing AsyncCollection surface is provided here, with the same input coercion, embedding, dimension checks, filter validation and egress handling as the sync base (both share one policy mixin).

Subclass responsibilities (async raw primitives)#

async _write_many(docs): Upsert documents; each vector is set and dimension-checked.
async _read(key) -> Document: Fetch one document; raise KeyError if absent.
async _drop(key): Delete one document; raise KeyError if absent.
_keys() -> AsyncIterator[str]: An async generator of document ids.
async _count() -> int: Number of documents.
async _query(vector, *, limit, filter, **kwargs) -> list[SearchResult]: Raw nearest-neighbor search; filter is the canonical AST.

async add_documents(documents, *, batch_size=100)[source]#

Batch upsert — mirrors vd.AbstractCollection.add_documents().

Return type:: None

async count()[source]#

Return the number of documents.

Return type:: int

async delete(key)[source]#

Delete a document; raises KeyError if absent.

Return type:: None

async get(key)[source]#

Fetch one document; raises KeyError if absent.

Return type:: Document

async keys()[source]#

Yield document ids.

Return type:: AsyncIterator[str]

property native: Any#: The raw backend handle (escape hatch), or None.

native_async: bool = True#: Real non-blocking I/O through the backend’s async SDK.

async search(query, *, limit=10, filter=None, egress=None, **kwargs)[source]#

Yield the limit documents most similar to query.

Same contract as vd.AbstractCollection.search().

Return type:: AsyncIterator[dict[str, Any]]

async set(key, value)[source]#

Insert or replace a document (idempotent upsert).

Return type:: None

supported_filter_operators: frozenset = frozenset({'$and', '$eq', '$exists', '$gt', '$gte', '$in', '$lt', '$lte', '$ne', '$nin', '$not', '$or'})#

the full language. Adapters narrow this; search() validates against it.

Type:: Filter operators this backend can honor. Default

async upsert(document)[source]#

Insert or replace document.

Return type:: None

class vd.AsyncClient(*args, **kwargs)[source]#

Bases: Protocol

The async sibling of Client.

Same operations — collection create / fetch / drop / list — exposed as awaitables and async iterators. Construct via vd.connect_async().

class vd.AsyncClientWrapper(sync_client)[source]#

Bases: object

Adapt a sync Client to the AsyncClient contract by dispatching every method to asyncio.to_thread().

Use connect_async() rather than instantiating this directly.

Parameters:: sync_client (Any) – A live Client (typically obtained from vd.connect()).

native_async#

Always False for this wrapper.

Type:: bool

property client: Any#: Pass through to the wrapped client’s client.

async close()[source]#

Release backend resources. Calls close() on the sync client if present.

Return type:: None

async create_collection(name, *, dimension=None, metric='cosine', **index_config)[source]#

Create a new collection; raise ValueError if it exists.

Return type:: AsyncCollection

async delete_collection(name)[source]#

Drop a collection; raise KeyError if absent.

Return type:: None

async get_collection(name)[source]#

Return an existing collection; raise KeyError if absent.

Return type:: AsyncCollection

async get_or_create_collection(name, *, dimension=None, metric='cosine', **index_config)[source]#

Return collection name, creating it if missing.

Return type:: AsyncCollection

async list_collections()[source]#

Yield collection names.

Return type:: AsyncIterator[str]

property sync: Any#: The underlying sync Client — a documented escape hatch.

class vd.AsyncCollection(*args, **kwargs)[source]#

Bases: Protocol

The async sibling of Collection.

Same conceptual surface — storage + search — but every method is awaitable and iterators are AsyncIterator. The mapping interface is exposed as explicit get / set / delete / keys / count methods (the stdlib’s MutableMapping ABC has no async counterpart; explicit methods are the Motor / aiopg convention).

Construct via vd.connect_async(); the universal AsyncCollectionWrapper in vd.asynchronous adapts every backend to this protocol by dispatching to the sync API through asyncio.to_thread(). Backends with native async SDKs override the wrapper and additionally satisfy SupportsNativeAsync.

class vd.AsyncCollectionWrapper(sync_collection)[source]#

Bases: object

Adapt a sync Collection to the AsyncCollection contract by dispatching every method to asyncio.to_thread().

Use connect_async() rather than instantiating this directly — it will pick this wrapper or a native async adapter as appropriate.

Parameters:: :: (sync_collection) – A live Collection (typically obtained from a Client).

native_async#

Always False for this wrapper. The wrapper still satisfies SupportsNativeAsync structurally (the attribute is present), but the boolean tells callers that I/O is happening in a thread pool rather than on the event loop. Prefer a native implementation for high-concurrency workloads.

Type:: bool

async add_documents(documents, *, batch_size=100)[source]#

Batch upsert — mirrors add_documents().

Return type:: None

async count()[source]#

Return the number of documents.

Return type:: int

async delete(key)[source]#

Delete a document; raises KeyError if absent.

Return type:: None

async get(key)[source]#

Fetch one document; raises KeyError if absent.

Return type:: Document

async keys()[source]#

Yield document ids.

Return type:: AsyncIterator[str]

property native: Any#: Pass through to the wrapped collection’s native.

native_async: bool = False#: This wrapper offloads to a thread pool; it doesn’t do non-blocking I/O.

async search(query, *, limit=10, filter=None, egress=None, **kwargs)[source]#

Yield the limit documents most similar to query.

The underlying search runs once on a worker thread; results stream from memory. (Most backends’ sync search already returns a list or a fully-realized iterator under the hood.)

Return type:: AsyncIterator[dict[str, Any]]

async set(key, value)[source]#

Insert or replace a document (idempotent upsert).

Return type:: None

property sync: Any#: The underlying sync Collection — a documented escape hatch.

async upsert(document)[source]#

Insert or replace document.

Return type:: None

class vd.BM25Index(collection, *, filter=None, tokenize=<function _tokenize>)[source]#

Bases: object

A reusable Okapi BM25 index over a vd collection’s stored text.

Builds the query-independent term statistics — per-document token lists, document frequencies, document lengths, and the mean length — once in __init__(), then answers many queries against them via search(). This is the build-once / query-many companion to bm25_lexical_search() (which builds a throwaway index for a single query): for batch evaluation or any repeated querying of the same collection it turns an O(N · Q) workload (re-tokenizing every document on every query) into O(N + scoring · Q).

Construction is O(N) in the collection size; each search() is O(matching documents). Fine for prototypes and collections up to ~100k documents; for larger workloads switch to a backend with a native text index (weaviate, elasticsearch, redis, lancedb, …).

Parameters:

collection (Collection) – Any vd Collection (or mapping-like id -> obj exposing .text and .metadata). Documents whose text is empty contribute zero score and are dropped at build time.
filter (Optional[dict]) – Canonical vd metadata filter, applied once at build time (via vd.filters.matches_filter()) so the index covers only the matching documents and its statistics reflect that subset.
tokenize (Callable[[str], list[str]]) – Tokenizer (default: lowercased \w+ tokens). Pass a custom one for stemming, CJK, etc.

Examples

>>> import vd
>>> c = vd.connect('memory').create_collection('t', dimension=2)
>>> c['a'] = vd.Document(id='a', text='the quick brown fox', vector=[1.0, 0.0])
>>> c['b'] = vd.Document(id='b', text='lazy dog sleeps', vector=[0.0, 1.0])
>>> index = vd.BM25Index(c)
>>> index.search('quick fox', limit=1)[0]['id']
'a'

search(query_text, *, limit=10, k1=1.5, b=0.75)[source]#

Okapi BM25 scores for query_text over the indexed documents.

Returns result dicts in the same shape as Collection.search() — {"id", "text", "score", "metadata"} — sorted by descending score. k1 / b are the standard Okapi hyperparameters (scoring-time, so one index can be queried with different settings).

Return type:: list[dict[str, Any]]

exception vd.BackendNotInstalledError[source]#

Bases: VdError, ImportError

Raised when a known backend’s Python package is not installed.

Distinct from an unknown backend name (a plain ValueError): the backend exists in vd’s provider registry, but its client library is missing. The message carries the pip install command to fix it.

class vd.Client(*args, **kwargs)[source]#

Bases: Protocol

A live connection to one backend: Mapping[str, Collection].

Collections are created explicitly (so create-time parameters such as dimension and metric can be supplied) and fetched either by get_collection() or by mapping access client[name].

create_collection(name, *, dimension=None, metric='cosine', **index_config)[source]#

Create a new collection; raise ValueError if it exists.

Return type:: Collection

delete_collection(name)[source]#

Drop a collection; raise KeyError if absent.

Return type:: None

get_collection(name)[source]#

Return an existing collection; raise KeyError if absent.

Return type:: Collection

list_collections()[source]#

Iterate collection names.

Return type:: Iterator[str]

class vd.Collection(*args, **kwargs)[source]#

Bases: Protocol

A collection of documents: MutableMapping[str, Document] + search.

The mapping half is storage; search() is the single retrieval extension. This minimal surface is everything vd’s tooling depends on. Batch insertion is an optional capability — see SupportsBatch.

search(query, *, limit=10, filter=None, egress=None, **kwargs)[source]#

Return the limit documents most similar to query.

Return type:: Iterator[dict[str, Any]]

class vd.Document(id, text='', vector=None, metadata=<factory>)[source]#

Bases: object

The unit stored in a Collection.

Parameters:

id (str) – Unique identifier; the key under which the document lives in a collection.
text (str) – The text content. May be empty for vector-first use cases where no text is associated with a vector.
vector (Optional[list[float]]) – The embedding. If None when written, the collection embeds text with its client’s embedder — or raises EmbeddingRequiredError if none is configured.
metadata (dict[str, Any]) – Arbitrary metadata, used for filtering and carried through search results.

Examples

>>> doc = Document(id="doc1", text="Hello world")
>>> doc.id, doc.text, doc.metadata
('doc1', 'Hello world', {})
>>> Document(id="v1", vector=[0.1, 0.2]).text
''

exception vd.EmbeddingRequiredError[source]#

Bases: VdError, RuntimeError

Raised when text is given but no embedder is configured.

vd operates on vectors. Passing raw text to collection[key] = text or collection.search(text) only works when the Client was created with an embedder. Otherwise, pass a Document with a vector (or a pre-computed query vector) directly.

exception vd.StaticIndexError[source]#

Bases: VdError

Raised on a write to a static (immutable) index.

Some backends — notably a plain FAISS flat index — build an index that cannot accept incremental __setitem__ / __delitem__ after creation. Such collections set AbstractCollection.supports_incremental_writes to False and raise this on write. Callers branch on that flag before triggering the error, and use the adapter’s documented rebuild() path.

class vd.SupportsBatch(*args, **kwargs)[source]#

Bases: Protocol

A collection that supports efficient batch insertion.

add_documents and upsert are not part of the minimal Collection contract. Every adapter built on AbstractCollection happens to provide them, but generic code should still feature-discover:

if isinstance(collection, SupportsBatch):
    collection.add_documents(many_docs, batch_size=256)

class vd.SupportsHybrid(*args, **kwargs)[source]#

Bases: Protocol

A collection that supports native hybrid (dense + lexical) search.

Hybrid search has no syntactic convergence across vector databases, so it is an opt-in capability, never baseline. Prefer the top-level vd.hybrid_search() — it dispatches to this protocol when the collection implements it and falls back to a pure-Python BM25 + RRF fusion otherwise. Feature-discover directly only when you specifically need to refuse the fallback path:

if isinstance(collection, SupportsHybrid):
    hits = collection.hybrid_search("query text", limit=20)

The portable contract is Reciprocal Rank Fusion (every native backend supports it). Weighted-blend (alpha) and other backend-specific fusion variants are accepted via **kwargs and documented per adapter — they are not portable across backends.

Parameters:

query (str or list[float]) – Query text (embedded via the collection’s embedder if configured) or a pre-computed query vector for the dense side.
query_text (str, optional) – Explicit text for the lexical side. Defaults to query when query is a string. Required when query is a vector.
limit (int) – Number of fused results to return.
filter (dict, optional) – Canonical vd metadata filter applied to both sub-searches.
k_dense (int, optional) – How many results to fetch from each sub-search before fusion. Both default to max(4 * limit, 50). Widen for higher recall.
k_lexical (int, optional) – How many results to fetch from each sub-search before fusion. Both default to max(4 * limit, 50). Widen for higher recall.
rrf_k (int) – Reciprocal Rank Fusion constant (typically 60).
egress (callable, optional) – Transform applied to each fused result before it is yielded.
**kwargs – Backend-specific knobs (e.g. alpha=0.7 on weaviate, ranker="weighted" on milvus). Documented per adapter.

class vd.SupportsNativeAsync(*args, **kwargs)[source]#

Bases: Protocol

Marker protocol set on async clients/collections that use a backend’s native async SDK rather than the universal asyncio.to_thread() wrapper.

Why care: in high-concurrency event-loop apps (FastAPI, Starlette, etc.), a to_thread-wrapped backend still blocks a worker thread per request. For real non-blocking I/O, prefer collections that satisfy this protocol. The wrapper sets this attribute to False; native adapters set it to True. isinstance(c, SupportsNativeAsync) matches both — check c.native_async for the boolean.

class vd.TimeIndexedCollection(collection, *, ts_field='ts', ts_parser=<function to_datetime>)[source]#

Bases: MutableMapping

Time-indexed wrapper over any vd Collection.

Maintains a sorted (ts_epoch, id) index alongside the underlying collection. Each stored document MUST carry a timestamp in its metadata under ts_field (default "ts"). The stored value is normalized to an ISO-8601 string so backend-side filtering remains usable.

Parameters:

collection (MutableMapping) – Any vd Collection (MutableMapping + search).
ts_field (str) – Metadata key holding the timestamp.
ts_parser (Callable[[Any], datetime]) – Optional custom parser Any -> datetime. Defaults to to_datetime().

Notes

The index is rebuilt on construction from whatever the underlying collection already contains (so the wrapper is safe to re-wrap a persisted collection across process restarts).

property base: MutableMapping#: The wrapped underlying collection.

query_window(start=None, end=None, *, filt=None)[source]#

Yield documents with start <= ts < end, in chronological order.

start / end may be None for half-open infinity. filt is an optional MongoDB-style predicate applied to document metadata, evaluated client-side (so it works on any backend).

Return type:: Iterator[Document]

reindex()[source]#

Force-rebuild the in-memory time index from the underlying collection.

Return type:: None

search_window(query, *, start=None, end=None, limit=10, filt=None, **kwargs)[source]#

Semantic search restricted to a time window.

Builds a metadata filter on ts_field and delegates to the underlying collection’s search. Falls back to a client-side post-filter for backends that don’t honor the filter.

Return type:: Iterator[dict]

time_range()[source]#

Return (min_ts, max_ts) as aware datetimes, or None if empty.

Return type:: Optional[tuple[datetime, datetime]]

>>> from vd import connect, Document
>>> import hashlib
>>> emb = lambda t: [b/128.0-1.0 for b in hashlib.md5(t.encode()).digest()[:4]]
>>> col = connect('memory', embedder=emb).create_collection('t')
>>> t = TimeIndexedCollection(col)
>>> t['a'] = Document(id='a', text='x', metadata={'ts': '2025-01-01'})
>>> t['b'] = Document(id='b', text='y', metadata={'ts': '2025-03-01'})
>>> [d.isoformat() for d in t.time_range()]
['2025-01-01T00:00:00+00:00', '2025-03-01T00:00:00+00:00']

window_iter(window='1d', *, start=None, end=None, reducer=<function count_docs>, skip_empty=False, align=True)[source]#

Yield (window_start, window_end, reducer_value) over fixed windows.

Parameters:

window (Union[str, timedelta, int, float]) – Window size. See parse_window() for accepted forms.
start (Union[str, datetime, int, float, None]) – Override the data range. Default: actual min/max ts in the index.
end (Union[str, datetime, int, float, None]) – Override the data range. Default: actual min/max ts in the index.
reducer (Callable[[Iterable[Document]], Any]) – Callable taking the iterable of in-window ``Document``s. Default is count_docs(). See also mean_vector().
skip_empty (bool) – If True, omit windows that contained zero documents.
align (bool) – If True (default), align start to the previous midnight (for daily windows) or to window-rounded boundary so downstream joins are clean. If False, use the literal start.

Return type:

Iterator[tuple[datetime, datetime, Any]]

exception vd.UnsupportedCapabilityError[source]#

Bases: VdError, NotImplementedError

Raised when an operation needs a capability the backend lacks.

Prefer feature-discovery — isinstance(collection, SupportsHybrid) — over catching this, but it is the clear, typed fallback when an optional operation is called on a backend that does not implement it.

exception vd.UnsupportedFilterError[source]#

Bases: VdError, ValueError

Raised when a metadata filter uses an operator a backend cannot honor.

The canonical, backend-agnostic filter language lives in vd.filters (a MongoDB-style JSON dialect). When a filter uses an operator outside a backend’s documented subset — or one that does not exist at all — this is raised, so the caller can simplify the filter or drop to the backend’s native filter via the escape hatch (collection.native).

exception vd.VdError[source]#

Bases: Exception

Base class for every error vd raises on its own behalf.

class vd.WindowSlice(start, end)[source]#

Bases: object

A time window: [start, end).

vd.benchmark_insert(collection, n_documents=100, *, text_length=100, batch_size=10)[source]#

Benchmark document insertion performance.

Parameters:

collection (Collection) – Collection to benchmark
n_documents (int) – Number of documents to insert
text_length (int) – Length of test documents
batch_size (int) – Batch size for insertion

Returns:

Benchmark results

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> results = vd.benchmark_insert(docs, n_documents=50)

vd.benchmark_search(collection, query, *, n_queries=100, limit=10)[source]#

Benchmark search performance on a collection.

Parameters:

collection (Collection) – Collection to benchmark
query (str) – Query text to use
n_queries (int) – Number of queries to run
limit (int) – Number of results per query

Returns:

Benchmark results with:

total_time: Total time for all queries
avg_latency: Average query latency
min_latency: Minimum latency
max_latency: Maximum latency
p50, p95, p99: Latency percentiles
queries_per_second: Throughput

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> # Add some documents...
>>> results = vd.benchmark_search(docs, "test query", n_queries=50)

vd.bm25_lexical_search(collection, query_text, *, limit=10, filter=None, k1=1.5, b=0.75)[source]#

Brute-force BM25 lexical search over a vd collection’s stored text.

Builds a throwaway BM25Index over collection and runs a single query against it. Used as the default lexical side of hybrid_search() when a collection does not implement SupportsHybrid.

Cost is O(N) in the collection size — fine for prototypes and collections up to ~100k documents. **For repeated queries over the same collection, build a** BM25Index once and call BM25Index.search() per query instead of calling this function in a loop — the term statistics are then computed once rather than on every call. For larger workloads, switch to a backend with native hybrid search (weaviate, elasticsearch, redis, lancedb, …) or pass a custom lexical_search callable to hybrid_search() that consults a real text index.

Parameters:

collection (Collection) – Any vd Collection. Documents whose text is empty contribute zero score and are filtered out of the result.
query_text (str) – The lexical query.
limit (int) – Maximum number of results.
filter (Optional[dict]) – Canonical vd metadata filter. Applied client-side via vd.filters.matches_filter().
k1 (float) – BM25 hyperparameters. Defaults match the standard Okapi BM25.
b (float) – BM25 hyperparameters. Defaults match the standard Okapi BM25.

Returns:

Result dicts in the same shape as Collection.search() — {"id", "text", "score", "metadata"} — sorted by descending score.

Return type:

list[dict[str, Any]]

Examples

>>> import vd
>>> c = vd.connect('memory').create_collection('t', dimension=2)
>>> c['a'] = vd.Document(id='a', text='the quick brown fox', vector=[1.0, 0.0])
>>> c['b'] = vd.Document(id='b', text='lazy dog sleeps', vector=[0.0, 1.0])
>>> hits = bm25_lexical_search(c, 'quick fox', limit=1)
>>> hits[0]['id']
'a'

vd.check_requirements(backend, *, verbose=True)[source]#

Diagnose whether backend is ready to use, and say what to do if not.

Runs an installed-check plus archetype-specific checks (embedded / server / managed), then computes the single most useful next step.

Parameters:

backend (str) – A provider name (see vd.list_all_backends()).
verbose (bool) – Print a human-readable report (in addition to returning the dict).

Returns:

{"backend", "archetype", "ok", "checks", "next_step"} where checks is a list of {"name", "ok", "detail"} records.

Return type:

Examples

>>> report = check_requirements('memory', verbose=False)
>>> report['ok']
True

vd.chunk_documents(documents, chunk_size=500, *, overlap=50, strategy='chars', id_template='{doc_id}_chunk_{chunk_num}', preserve_metadata=True)[source]#

Chunk multiple documents while preserving metadata.

Parameters:

documents (Iterator[tuple[str, str | dict]]) – Iterator of (doc_id, text) or (doc_id, text, metadata) tuples
chunk_size (int) – Size of each chunk
overlap (int) – Overlap between chunks
strategy (str) – Chunking strategy (see chunk_text)
id_template (str) – Template for chunk IDs. Can use {doc_id} and {chunk_num}
preserve_metadata (bool) – Whether to copy metadata to all chunks

Yields:

tuple – (chunk_id, chunk_text, metadata) tuples

Return type:

Iterator[tuple[str, str, dict]]

Examples

>>> docs = [('doc1', 'Long text...', {'author': 'Alice'})]
>>> chunks = list(chunk_documents(docs, chunk_size=20))
>>> len(chunks) >= 1
True

vd.chunk_text(text, chunk_size=500, *, overlap=50, strategy='chars', preserve_sentences=True)[source]#

Chunk text into smaller pieces.

Parameters:

text (str) – Text to chunk
chunk_size (int) – Target size of each chunk (in characters or tokens depending on strategy)
overlap (int) – Number of characters/tokens to overlap between chunks
strategy (str) –
Chunking strategy:
- ’chars’: Character-based chunking
- ’words’: Word-based chunking
- ’sentences’: Sentence-based chunking
- ’paragraphs’: Paragraph-based chunking
preserve_sentences (bool) – Try to avoid breaking sentences when using chars/words strategy

Returns:

List of text chunks

Return type:

list[str]

Examples

>>> text = "This is sentence one. This is sentence two. This is sentence three."
>>> chunks = chunk_text(text, chunk_size=30, strategy='chars')
>>> len(chunks) >= 2
True

>>> chunks = chunk_text(text, strategy='sentences')
>>> len(chunks)
3

vd.clean_text(text, *, lowercase=False, remove_extra_whitespace=True, remove_urls=False, remove_emails=False, remove_numbers=False, remove_punctuation=False)[source]#

Clean and normalize text.

Parameters:

text (str) – Text to clean
lowercase (bool) – Convert to lowercase
remove_extra_whitespace (bool) – Collapse multiple spaces/newlines
remove_urls (bool) – Remove URLs
remove_emails (bool) – Remove email addresses
remove_numbers (bool) – Remove numbers
remove_punctuation (bool) – Remove punctuation

Returns:

Cleaned text

Return type:

str

Examples

>>> text = "Hello   World!  Visit https://example.com"
>>> clean_text(text, remove_urls=True)
'Hello World! Visit'
>>> clean_text(text, lowercase=True, remove_punctuation=True)
'hello world visit httpsexamplecom'

vd.collection_stats(collection)[source]#

Compute comprehensive statistics for a collection.

Parameters:

collection (Collection) – Collection to analyze

Returns:

Statistics including:

total_documents: Number of documents
avg_text_length: Average text length in characters
min_text_length: Minimum text length
max_text_length: Maximum text length
total_chars: Total characters across all documents
metadata_fields: Set of all metadata fields used
metadata_field_counts: Count of documents with each metadata field
embedding_dimension: Dimension of embeddings (if available)
has_vectors: Number of documents with vectors

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> docs['doc1'] = ("Hello", {'category': 'greeting'})
>>> stats = vd.collection_stats(docs)
>>> print(stats['total_documents'])
1

vd.compare_backends(names, *, characteristics=None)[source]#

Return a {name: {characteristic: value}} table for the given providers.

Return type:: dict[str, dict[str, Any]]

vd.connect(backend, *, embedder=None, **backend_kwargs)[source]#

Connect to a vector database backend and return its Client.

This is the single entry point of vd. Switching vector databases is a one-argument change here.

Parameters:

backend (str) – Backend name: "memory", "chroma", "qdrant", "faiss", "lancedb", "sqlite_vec", "duckdb", "pgvector", "pinecone", … Run vd.list_backends() for what is installed.
embedder (Optional[Callable[[str], list[float]]]) – A text -> vector function. Supply it only if you want the convenience of passing raw text to collection[key] = "text" and collection.search("query text"). vd never embeds on its own — with no embedder, pass Document objects with vectors and pre-computed query vectors.
**backend_kwargs – Backend-specific connection options (persist_directory, url, api_key, path, …). See each adapter’s docstring.

Returns:

A connected client — a Mapping of collection name to collection.

Return type:

Client

Examples

>>> client = connect('memory')
>>> client = connect('chroma', persist_directory='./db')
>>> client = connect('qdrant', url='http://localhost:6333')

async vd.connect_async(backend, *, native=True, **kwargs)[source]#

Async sibling of vd.connect().

Returns an AsyncClient. When the backend has a native async client (see list_async_backends(); today qdrant), its registered factory decides: it may return a native client doing real non-blocking I/O (qdrant with url=) or the wrapper (embedded qdrant, whose async client would block the loop). Every other backend goes through the universal AsyncClientWrapper, built on asyncio.to_thread(). Check client.native_async to tell them apart.

Parameters:

backend (str) – Backend name — same vocabulary as vd.connect().
native (bool) – Use the backend’s native async client when one is registered (default). False forces the to_thread wrapper around the sync adapter.
**kwargs – Forwarded to the native client’s constructor, or to vd.connect() for the wrapper. Both take the same arguments (embedder, url, path, …).

Returns:

A live async client. await once at session start:

client = await vd.connect_async("memory")

Return type:

AsyncClient

Examples

>>> import asyncio, vd
>>> async def go():
...     client = await vd.connect_async("memory")
...     col = await client.create_collection("docs", dimension=2)
...     await col.set("a", vd.Document(id="a", text="x", vector=[1.0, 0.0]))
...     return await col.count()
>>> asyncio.run(go())
1

vd.connect_from_config(path=None, *, profile=None, apply_env=True, embedder=None, **overrides)[source]#

Connect to a backend using configuration from a file.

Parameters:

path (Union[str, Path, None]) – Path to configuration file. If not provided, searches for default config files.
profile (Optional[str]) – Profile name to use from configuration. Defaults to ‘default’ or the VD_PROFILE environment variable.
apply_env (bool) – Whether to apply environment variable overrides
embedder (Optional[Callable[[str], list[float]]]) – Optional text -> vector convenience embedder, passed to vd.connect(). A vd config file describes the *backend connection*, not embedding — embedding stays the caller’s concern.
**overrides – Additional keyword arguments to override configuration values

Returns:

Connected client instance

Return type:

Client

Examples

>>> # With a config file
>>> client = connect_from_config('vd.yaml')

>>> # With a specific profile
>>> client = connect_from_config('vd.yaml', profile='production')

>>> # With environment variable VD_PROFILE=dev
>>> client = connect_from_config()

>>> # With overrides
>>> client = connect_from_config('vd.yaml', persist_directory='./data')

vd.copy_collection(source, target, *, batch_size=100, preserve_vectors=True)[source]#

Copy a collection with flexible source/target specification.

Parameters:

source (tuple[str, str] | Collection) – Either a Collection object or (backend_name, collection_name) tuple
target (tuple[str, str, dict] | Collection) – Either a Collection object or (backend_name, collection_name, config) tuple
batch_size (int) – Batch size for copying
preserve_vectors (bool) – Whether to preserve vectors

Returns:

Migration statistics

Return type:

Examples

>>> import vd
>>> # Copy between backends
>>> stats = vd.copy_collection(
...     source=('memory', 'docs'),
...     target=('chroma', 'docs', {'persist_directory': './data'}),
... )

vd.cosine_similarity(vec1, vec2)[source]#

Cosine similarity of two vectors (1.0 identical, 0.0 orthogonal).

Return type:: float

Examples

>>> cosine_similarity([1.0, 0.0], [1.0, 0.0])
1.0
>>> cosine_similarity([1.0, 0.0], [0.0, 1.0])
0.0

vd.count_docs(docs)[source]#

len reducer that also handles generator inputs.

Return type:: int

>>> count_docs(iter([1, 2, 3]))
3

vd.create_example_config(format='yaml')[source]#

Generate an example configuration file content.

Parameters:: format (str) – Format of configuration: ‘yaml’ or ‘toml’
Returns:: Example configuration as a string
Return type:: str

Examples

>>> yaml_config = create_example_config('yaml')
>>> print(yaml_config)
>>> toml_config = create_example_config('toml')

vd.deduplicate_results(results, *, key='id', keep='first')[source]#

Remove duplicate results.

Parameters:

results (Iterator[dict[str, Any]]) – Search results
key (str) – Field to check for duplicates
keep (str) – Which duplicate to keep: ‘first’ or ‘highest_score’

Yields:

dict – Deduplicated results

Return type:

Examples

>>> results = [
...     {'id': 'doc1', 'score': 0.9},
...     {'id': 'doc1', 'score': 0.8},
...     {'id': 'doc2', 'score': 0.7}
... ]
>>> unique = list(deduplicate_results(iter(results)))
>>> len(unique)
2

vd.euclidean_distance(vec1, vec2)[source]#

Euclidean (L2) distance between two vectors.

Return type:: float

Examples

>>> euclidean_distance([1.0, 0.0], [1.0, 0.0])
0.0

vd.export_collection(collection, output_path, *, format='jsonl', **kwargs)[source]#

Export a collection to a file in the specified format.

Parameters:

collection (Collection) – Collection to export
output_path (Union[str, Path]) – Output file/directory path
format (str) – Export format: ‘jsonl’, ‘json’, ‘directory’
**kwargs – Additional format-specific options

Returns:

Number of documents exported

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> vd.export_collection(docs, 'backup.jsonl')

vd.export_to_directory(collection, output_dir, *, include_vectors=True)[source]#

Export collection as a directory with one JSON file per document.

Useful for version control and easy browsing.

Parameters:

collection (Collection) – Collection to export
output_dir (Union[str, Path]) – Output directory path
include_vectors (bool) – Whether to include vectors

Returns:

Number of documents exported

Return type:

vd.export_to_json(collection, output_path, *, include_vectors=True, indent=2)[source]#

Export a collection to JSON format.

Creates a JSON array of all documents.

Parameters:

collection (Collection) – Collection to export
output_path (Union[str, Path]) – Output file path
include_vectors (bool) – Whether to include embedding vectors
indent (Optional[int]) – JSON indentation (None for compact)

Returns:

Number of documents exported

Return type:

vd.export_to_jsonl(collection, output_path, *, include_vectors=True)[source]#

Export a collection to JSONL (JSON Lines) format.

Each line is a JSON object representing a document.

Parameters:

collection (Collection) – Collection to export
output_path (Union[str, Path]) – Output file path
include_vectors (bool) – Whether to include embedding vectors in the export

Returns:

Number of documents exported

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> docs['doc1'] = "Hello"
>>> vd.export_to_jsonl(docs, 'backup.jsonl')
1

vd.extract_metadata(text, *, extract_title=True, extract_length=True, extract_word_count=True, extract_language=False)[source]#

Extract metadata from text.

Parameters:

text (str) – Text to analyze
extract_title (bool) – Extract first line as title
extract_length (bool) – Add text length
extract_word_count (bool) – Add word count
extract_language (bool) – Detect language (requires langdetect)

Returns:

Extracted metadata

Return type:

Examples

>>> text = "My Title\n\nThis is the content."
>>> meta = extract_metadata(text)
>>> meta['title']
'My Title'
>>> meta['char_count']
30

vd.find_duplicates(collection, *, threshold=0.95, method='cosine')[source]#

Find near-duplicate documents in a collection.

Parameters:

collection (Collection) – Collection to analyze
threshold (float) – Similarity threshold above which documents are considered duplicates
method (str) – Similarity method: ‘cosine’ or ‘exact’

Returns:

List of (doc_id1, doc_id2, similarity) tuples for duplicates

Return type:

list[tuple[str, str, float]]

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> docs['doc1'] = "Hello world"
>>> docs['doc2'] = "Hello world"
>>> duplicates = vd.find_duplicates(docs)
>>> len(duplicates) > 0
True

vd.find_outliers(collection, *, n_neighbors=5, threshold=0.3)[source]#

Find outlier documents (those dissimilar to their neighbors).

Parameters:

collection (Collection) – Collection to analyze
n_neighbors (int) – Number of neighbors to consider
threshold (float) – Average similarity threshold below which a document is an outlier

Returns:

List of (doc_id, avg_similarity) for outliers

Return type:

list[tuple[str, float]]

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> # Add some documents...
>>> outliers = vd.find_outliers(docs)

vd.get_backend_characteristics()[source]#

Return a compact {name: characteristics} map for comparison tooling.

Return type:: dict[str, dict[str, Any]]

vd.get_backend_info(name)[source]#

Return one provider’s metadata with installed/has_adapter flags.

Return type:: dict[str, Any]

vd.get_install_instructions(name)[source]#

Return a human-readable setup blurb for one provider.

Return type:: str

vd.health_check_backend(backend_name, **config)[source]#

Check if a backend is healthy and accessible.

Parameters:

backend_name (str) – Backend name to check
**config – Backend-specific configuration

Returns:

Health report with keys:

status: ‘healthy’, ‘unhealthy’, or ‘unavailable’
available: Whether backend is installed
registered: Whether backend is registered
message: Status message
details: Additional details (if connected successfully)

Return type:

Examples

>>> import vd
>>> status = vd.health_check_backend('memory')
>>> print(status['status'])
'healthy'

vd.health_check_collection(collection)[source]#

Check collection health and compute basic stats.

Parameters:: collection (Collection) – Collection to check
Returns:: Health report
Return type:: dict[str, Any]

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> status = vd.health_check_collection(docs)

vd.hybrid_search(collection, query, *, query_text=None, limit=10, filter=None, k_dense=None, k_lexical=None, rrf_k=60, lexical_search=None, egress=None, **kwargs)[source]#

Hybrid (dense + lexical) search that works on any vd Collection.

Dispatches to the collection’s native hybrid_search when it implements SupportsHybrid (efficient, server-side). Otherwise fuses the collection’s own dense search() with a client-side lexical scan (default: bm25_lexical_search()) via **Reciprocal Rank Fusion**.

The portable contract is RRF. Backend-specific knobs (weighted blend alpha, fusion-type variants, native ranker choices) are accepted via **kwargs and forwarded to the adapter when it has a native implementation; they are ignored by the client-side fallback.

Parameters:

collection (Collection) – Any vd Collection — native-hybrid or not.
query (Union[str, list[float]]) – Query text (embedded by the collection if it has an embedder) or a pre-computed query vector. When query is a vector, query_text is required.
query_text (Optional[str]) – Explicit text for the lexical side. Defaults to query when query is a string.
limit (int) – Number of fused results to return.
filter (Optional[dict]) – Canonical vd metadata filter, applied to both sub-searches.
k_dense (Optional[int]) – How many results to fetch from each sub-search before fusion. Default is max(4 * limit, 50) for each side. Widen for higher recall.
k_lexical (Optional[int]) – How many results to fetch from each sub-search before fusion. Default is max(4 * limit, 50) for each side. Widen for higher recall.
rrf_k (int) – Reciprocal Rank Fusion constant (typically 60).
lexical_search (Optional[Callable[..., list[dict[str, Any]]]]) – Custom lexical_search(collection, query_text, *, limit, filter, \*\*kwargs) -> list[SearchResult]. Defaults to bm25_lexical_search(). Used only on the fallback path; on the native path it is ignored with a UserWarning.
egress (Optional[Callable[[dict[str, Any]], Any]]) – Per-result transform applied before yielding.
**kwargs – Extra options. On the native path they are forwarded to the adapter (e.g. alpha=0.7 on weaviate). On the fallback path they are ignored.

Yields:

dict – Fused result dicts. score is the RRF score on the fallback path, or the adapter’s fused score on the native path.

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> col = client.create_collection('docs', dimension=2)
>>> col['a'] = vd.Document(id='a', text='cats purr',
...                        vector=[1.0, 0.0])
>>> col['b'] = vd.Document(id='b', text='dogs bark',
...                        vector=[0.0, 1.0])
>>> hits = list(vd.hybrid_search(col, [0.9, 0.1], query_text='cats',
...                              limit=1))
>>> hits[0]['id']
'a'

async vd.hybrid_search_async(collection, query, *, query_text=None, limit=10, filter=None, k_dense=None, k_lexical=None, rrf_k=60, lexical_search=None, egress=None, **kwargs)[source]#

Async sibling of vd.hybrid_search().

For a wrapped sync collection, runs vd.hybrid_search() (native hybrid if the backend has it, else the client-side BM25 + RRF fallback) on a worker thread. For a native async collection it awaits the collection’s own hybrid_search if it has one, and otherwise fuses the collection’s async dense search with a client-side BM25 scan (O(N): it reads every document) via RRF. On a native collection, an async def lexical_search receives the async collection; a sync one receives a materialized {id: Document} dict (every document is read per call) and runs on a worker thread. Either way the awaitable + async iterator interface stays uniform.

Parameters mirror vd.hybrid_search() exactly; see that function for the full docs.

Yields:: dict – Fused result dicts.
Return type:: AsyncIterator[dict[str, Any]]

Examples

>>> import asyncio, vd
>>> async def go():
...     client = await vd.connect_async("memory")
...     col = await client.create_collection("docs", dimension=2)
...     await col.set("a", vd.Document(id="a", text="cats",
...                                    vector=[1.0, 0.0]))
...     await col.set("b", vd.Document(id="b", text="dogs",
...                                    vector=[0.0, 1.0]))
...     hits = []
...     async for h in vd.hybrid_search_async(col, [0.9, 0.1],
...                                           query_text="cats", limit=1):
...         hits.append(h["id"])
...     return hits
>>> asyncio.run(go())
['a']

vd.id_and_score(result)[source]#

Egress: keep (id, score).

Return type:: tuple[str, float]

vd.id_only(result)[source]#

Egress: keep only the document id.

Return type:: str

vd.id_text_score(result)[source]#

Egress: keep (id, text, score).

Return type:: tuple[str, str, float]

vd.import_collection(collection, input_path, *, format=None, **kwargs)[source]#

Import documents into a collection from a file.

Parameters:

collection (Collection) – Collection to import into
input_path (Union[str, Path]) – Input file/directory path
format (Optional[str]) – Import format: ‘jsonl’, ‘json’, ‘directory’ If None, inferred from file extension
**kwargs – Additional format-specific options

Returns:

Number of documents imported

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> vd.import_collection(docs, 'backup.jsonl')

vd.import_from_directory(collection, input_dir, *, batch_size=100, skip_existing=False, pattern='*.json')[source]#

Import documents from a directory of JSON files.

Parameters:

collection (Collection) – Collection to import into
input_dir (Union[str, Path]) – Input directory path
batch_size (int) – Batch size for adding documents
skip_existing (bool) – If True, skip documents with IDs that already exist
pattern (str) – File pattern to match

Returns:

Number of documents imported

Return type:

vd.import_from_json(collection, input_path, *, batch_size=100, skip_existing=False)[source]#

Import documents from JSON format into a collection.

Parameters:

collection (Collection) – Collection to import into
input_path (Union[str, Path]) – Input file path
batch_size (int) – Batch size for adding documents
skip_existing (bool) – If True, skip documents with IDs that already exist

Returns:

Number of documents imported

Return type:

vd.import_from_jsonl(collection, input_path, *, batch_size=100, skip_existing=False)[source]#

Import documents from JSONL format into a collection.

Parameters:

collection (Collection) – Collection to import into
input_path (Union[str, Path]) – Input file path
batch_size (int) – Batch size for adding documents
skip_existing (bool) – If True, skip documents with IDs that already exist

Returns:

Number of documents imported

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> vd.import_from_jsonl(docs, 'backup.jsonl')
1

vd.install_backend(backend, *, run=False)[source]#

Return (and optionally run) the pip install command for backend.

Parameters:

backend (str) – Provider name.
run (bool) – If True, actually invoke pip in the current interpreter. If False (the default), only return the command — the caller decides.

Returns:

The pip command (or a note that nothing is needed).

Return type:

str

vd.install_command(name)[source]#

Return the pip install command that makes name usable.

A backend vd has an adapter for installs through vd’s own extra (vd[<backend>]), which pins exactly the client libraries that adapter imports. The extra is double-quoted so the command is safe to paste into zsh, bash, PowerShell and cmd. Providers without an adapter get their raw client package(s).

Return type:: str

Examples

>>> install_command('qdrant')
'pip install "vd[qdrant]"'
>>> install_command('opensearch')
'pip install opensearch-py'
>>> install_command('memory')
'memory needs no installation (built into vd)'

vd.list_all_backends()[source]#

Return every provider with live installed / has_adapter flags added.

Return type:: dict[str, dict[str, Any]]

vd.list_async_backends()[source]#

Return the names of backends with a registered native async client.

Every other backend still works with connect_async(), through the universal to_thread wrapper.

Return type:: list[str]

vd.list_available_backends()[source]#

Return providers vd can connect() right now.

A backend is available iff its adapter module imported successfully — which happens only when its client library is installed. This is exactly the set of registered backends.

Return type:: list[str]

vd.list_backends()[source]#

Return the names of all backends with a registered (importable) adapter.

Return type:: list[str]

vd.load_config(path=None, *, format=None)[source]#

Load configuration from a file.

Automatically detects format from file extension if not specified.

Parameters:

path (Union[str, Path, None]) – Path to configuration file. If not provided, looks for default config files in: ./vd.yaml, ./vd.yml, ./vd.toml, ~/.vd/config.yaml, etc.
format (Optional[str]) – Configuration format: ‘yaml’ or ‘toml’. Auto-detected from extension if not provided.

Returns:

Configuration dictionary

Return type:

dict

Examples

>>> config = load_config('vd.yaml')
>>> config = load_config('vd.toml')
>>> config = load_config()  # Looks for default config files

vd.matches_filter(metadata, filter)[source]#

Return True if metadata satisfies the MongoDB-style filter.

An empty or None filter matches everything. Unknown operators raise UnsupportedFilterError — they never silently match.

Parameters:

metadata (Mapping[str, Any]) – A document’s metadata dict.
filter (Optional[dict[str, Any]]) – A filter in the canonical vd dialect (see the module docstring).

Return type:

bool

Examples

>>> matches_filter({'year': 2024}, None)
True
>>> matches_filter({'year': 2024, 'cat': 'tech'},
...                {'year': {'$gte': 2020}, 'cat': 'tech'})
True
>>> matches_filter({'views': 50}, {'views': {'$gte': 10, '$lte': 100}})
True

vd.mean_vector(docs)[source]#

Element-wise mean of document embeddings. None if empty / no vectors.

Return type:: Optional[list[float]]

>>> from vd.base import Document
>>> mean_vector([
...     Document(id='a', text='', vector=[1.0, 2.0]),
...     Document(id='b', text='', vector=[3.0, 4.0]),
... ])
[2.0, 3.0]
>>> mean_vector([]) is None
True

vd.metadata_distribution(collection, field, *, top_n=None)[source]#

Get the distribution of values for a metadata field.

Parameters:

collection (Collection) – Collection to analyze
field (str) – Metadata field name
top_n (Optional[int]) – If specified, return only the top N most common values

Returns:

Mapping of field values to their counts

Return type:

dict[Any, int]

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> docs['doc1'] = ("Hello", {'category': 'A'})
>>> docs['doc2'] = ("World", {'category': 'A'})
>>> docs['doc3'] = ("Test", {'category': 'B'})
>>> dist = vd.metadata_distribution(docs, 'category')
>>> print(dist)
{'A': 2, 'B': 1}

vd.migrate_client(source_client, target_client, *, collection_names=None, batch_size=100, preserve_vectors=True, progress_callback=None)[source]#

Migrate all (or selected) collections from one client to another.

Parameters:

source_client (Client) – Source database client
target_client (Client) – Target database client
collection_names (Optional[list[str]]) – Specific collections to migrate. If None, migrates all.
batch_size (int) – Batch size for migration
preserve_vectors (bool) – Whether to preserve vectors
progress_callback (Optional[Callable[[str, int, int], None]]) – Function called with (collection_name, current, total)

Returns:

Overall migration statistics

Return type:

Examples

>>> import vd
>>> source = vd.connect('memory')
>>> target = vd.connect('chroma', persist_directory='./backup')
>>> stats = vd.migrate_client(source, target)

vd.migrate_collection(source_collection, target_collection, *, batch_size=100, preserve_vectors=True, progress_callback=None, skip_existing=False)[source]#

Migrate a collection from one backend to another.

Parameters:

source_collection (Collection) – Source collection to migrate from
target_collection (Collection) – Target collection to migrate to
batch_size (int) – Number of documents to migrate per batch
preserve_vectors (bool) – Whether to preserve pre-computed vectors
progress_callback (Optional[Callable[[int, int], None]]) – Function called with (current, total) to report progress
skip_existing (bool) – If True, skip documents that already exist in target

Returns:

Migration statistics with keys:

total: Total documents in source
migrated: Number of documents migrated
skipped: Number of documents skipped
failed: Number of failures
errors: List of error messages

Return type:

Examples

>>> import vd
>>> # Create source and target
>>> source_client = vd.connect('memory')
>>> target_client = vd.connect('chroma', persist_directory='./data')
>>> source = source_client.get_collection('my_docs')
>>> target = target_client.create_collection('my_docs')
>>>
>>> # Migrate
>>> stats = vd.migrate_collection(source, target)
>>> print(f"Migrated {stats['migrated']} documents")

vd.multi_query_search(collection, queries, *, limit=10, combine='interleave', filter=None, **kwargs)[source]#

Search with multiple queries and combine results.

Parameters:

collection (Collection) – Collection to search
queries (list[str]) – Multiple query strings
limit (int) – Total number of results to return
combine (str) –
How to combine results:
- ’interleave’: Interleave results from each query
- ’concatenate’: Concatenate all results
- ’union’: Remove duplicates across queries
- ’best’: Take best results across all queries
filter (Optional[dict]) – Metadata filter
**kwargs – Additional search options

Yields:

dict – Search results

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> results = vd.multi_query_search(
...     docs,
...     ["What is AI?", "How does ML work?"],
...     limit=10
... )

vd.normalize_document_input(doc_input, *, auto_id=True)[source]#

Normalize a flexible document input to a Document.

Accepted shapes: a Document; a str (just text); a tuple (text, id), (text, metadata), or (text, id, metadata).

Parameters:

doc_input (Union[str, tuple, Document]) – The input to normalize.
auto_id (bool) – When the input carries no id, generate one (vs. leaving it empty).

Return type:

Document

Examples

>>> normalize_document_input(("Hello", "doc1")).id
'doc1'
>>> normalize_document_input(("Hello", {"k": "v"})).metadata
{'k': 'v'}
>>> normalize_document_input("Hello world").id.startswith('doc_')
True

vd.normalize_whitespace(text)[source]#

Normalize whitespace in text.

Replaces tabs, multiple spaces, and multiple newlines with single versions.

Parameters:: text (str) – Text to normalize
Returns:: Normalized text
Return type:: str

Examples

>>> normalize_whitespace("Hello\t\tWorld  \n\n\nTest")
'Hello World \nTest'

vd.parse_window(window)[source]#

Parse a window spec into a timedelta.

Strings use a trailing unit char: "1d", "4h", "30m", "15s", "1w". Numbers are treated as seconds. A timedelta is returned as-is.

Return type:: timedelta

>>> parse_window('1d') == timedelta(days=1)
True
>>> parse_window('4h') == timedelta(hours=4)
True
>>> parse_window(3600) == timedelta(hours=1)
True

vd.print_backends_table()[source]#

Print every known vector database, grouped by deployment archetype.

Return type:: None

vd.print_comparison(names)[source]#

Print a side-by-side comparison table of the given providers.

Return type:: None

vd.print_recommendation(**kwargs)[source]#

Run recommend_backend() and print the recommendation readably.

Return type:: None

vd.provider(name)[source]#

Return one provider’s metadata, or None if name is unknown.

Return type:: Optional[dict[str, Any]]

vd.providers()[source]#

Return the full provider registry as {name: metadata}.

Return type:: dict[str, dict[str, Any]]

vd.reciprocal_rank_fusion(result_lists, *, k=60)[source]#

Combine multiple result lists using Reciprocal Rank Fusion.

RRF is a simple yet effective way to combine rankings from multiple sources.

Parameters:

result_lists (list[list[dict[str, Any]]]) – Multiple lists of search results
k (int) – Constant for RRF formula (typically 60)

Returns:

Combined and re-ranked results

Return type:

list[dict[str, Any]]

Examples

>>> results1 = [{'id': 'doc1', 'score': 0.9}, {'id': 'doc2', 'score': 0.8}]
>>> results2 = [{'id': 'doc2', 'score': 0.95}, {'id': 'doc3', 'score': 0.7}]
>>> combined = reciprocal_rank_fusion([results1, results2])

vd.recommend_backend(*, corpus_size='medium', persistence=True, can_run_docker=True, cloud_ok=True, budget='free', existing_db=None, needs_hybrid=False, air_gapped=False)[source]#

Recommend a vector database from a few yes/no facts about the situation.

A direct encoding of the decision framework in the report’s §4. Returns a primary pick, a runner-up, and the reasoning trail.

Parameters:

corpus_size (str) – Rough vector count: tiny <100k, small <10M, medium ~10M, large <100M, huge >100M.
persistence (bool) – Must data survive a process restart?
can_run_docker (bool) – Can the user run Docker / operate a server process?
cloud_ok (bool) – Is a managed cloud service acceptable (vs. on-prem only)?
budget (str) – Free-tier-only, or is paid acceptable?
existing_db (Optional[str]) – A database the user already operates — strongly biases the pick.
needs_hybrid (bool) – Need keyword + vector ranking fused in one query?
air_gapped (bool) – Must run with zero network / zero telemetry?

Returns:

{"primary", "runner_up", "reasoning", "alternatives"}.

Return type:

Examples

>>> rec = recommend_backend(corpus_size='tiny', persistence=False)
>>> rec['primary']
'memory'
>>> rec = recommend_backend(existing_db='postgres')
>>> rec['primary']
'pgvector'

vd.register_async_backend(name, factory=None)[source]#

Use as a class decorator on an AsyncAbstractClient subclass, or call it with a factory (a class or a function, sync or async, taking the connect_async() keyword arguments). Once registered, connect_async() returns the factory’s client instead of the to_thread wrapper.

Examples

>>> @register_async_backend('example')
... class ExampleAsyncClient(AsyncAbstractClient):
...     ...

vd.register_backend(name)[source]#

Class decorator: register an adapter Client under name.

Return type:: Callable[[type], type]

Examples

>>> from vd.base import AbstractClient
>>> @register_backend('example')
... class ExampleClient(AbstractClient):
...     ...

vd.sample_collection(collection, n, *, method='random', seed=None)[source]#

Sample document IDs from a collection.

Parameters:

collection (Collection) – Collection to sample from
n (int) – Number of documents to sample
method (str) – Sampling method: ‘random’, ‘first’, ‘diverse’
seed (Optional[int]) – Random seed for reproducibility

Returns:

Sampled document IDs

Return type:

list[str]

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> # Add 100 documents...
>>> sample = vd.sample_collection(docs, 10, method='random')
>>> len(sample)
10

vd.save_config(config, path, *, format=None)[source]#

Save configuration to a file.

Parameters:

config (dict) – Configuration dictionary to save
path (Union[str, Path]) – Path to save configuration file
format (Optional[str]) – Format to save: ‘yaml’ or ‘toml’. Auto-detected from extension if not provided.

Return type:

None

Examples

>>> config = {
...     'profiles': {
...         'dev': {'backend': 'memory'},
...         'prod': {'backend': 'chroma', 'persist_directory': './data'}
...     }
... }
>>> save_config(config, 'vd.yaml')

vd.search_similar_to_document(collection, doc_id, *, limit=10, exclude_self=True, filter=None, **kwargs)[source]#

Find documents similar to a specific document.

Parameters:

collection (Collection) – Collection to search
doc_id (str) – ID of the reference document
limit (int) – Number of similar documents to return
exclude_self (bool) – Whether to exclude the reference document from results
filter (Optional[dict]) – Metadata filter
**kwargs – Additional search options

Yields:

dict – Search results

Return type:

Examples

>>> import vd
>>> client = vd.connect('memory')
>>> docs = client.create_collection('test')
>>> similar = vd.search_similar_to_document(docs, 'doc1', limit=5)

vd.setup_guide(backend)[source]#

Return a full, copy-pasteable setup playbook for backend.

Covers: the pip install, a Docker one-liner for server backends, the environment variables for managed backends, a verify command, and the relevant documentation links.

Return type:: str

vd.skills_dir()[source]#

Return the path to the bundled AI-agent skills directory.

Return type:: Path

vd.text_only(result)[source]#

Egress: keep only the text. >>> text_only({'text': 'hi'}) -> 'hi'.

Return type:: str

vd.to_datetime(ts)[source]#

Coerce a timestamp-like value into a tz-aware UTC datetime.

Accepts ISO-8601 strings (with or without timezone), date-only strings ("2025-03-13"), epoch seconds (int or float), and datetime objects. Naive datetimes / strings are assumed UTC.

Return type:: datetime

>>> to_datetime('2025-03-13T09:00:00').isoformat()
'2025-03-13T09:00:00+00:00'
>>> to_datetime('2025-03-13').isoformat()
'2025-03-13T00:00:00+00:00'
>>> to_datetime(1741856400).isoformat()
'2025-03-13T09:00:00+00:00'

vd.to_iso(ts)[source]#

ISO-8601 (UTC) string suitable for cross-backend metadata storage.

Return type:: str

>>> to_iso('2025-03-13T09:00:00')
'2025-03-13T09:00:00+00:00'

vd.truncate_text(text, max_length, *, suffix='...')[source]#

Truncate text to maximum length.

Parameters:

text (str) – Text to truncate
max_length (int) – Maximum length
suffix (str) – Suffix to add to truncated text

Returns:

Truncated text

Return type:

str

Examples

>>> truncate_text("This is a long text", 10)
'This is...'

vd.validate_collection(collection)[source]#

Validate collection integrity and identify issues.

Parameters:

collection (Collection) – Collection to validate

Returns:

Validation report with:

valid: Whether collection is valid
issues: List of issue descriptions
warnings: List of warning messages
stats: Basic stats

Return type: