localvectordb.query_builder module

LocalVectorDB Query Builder - SQL-like interface for vector database queries.

This module provides a high-level, fluent interface for building complex queries against vector databases. It combines the power of vector similarity search with traditional database operations like filtering, grouping, aggregation, and sorting, all through a SQL-like API.

The QueryBuilder supports multiple search modes (vector, keyword, hybrid), semantic filtering, result reranking, and both synchronous and asynchronous execution.

Core Features

Multi-modal Search: Vector similarity, keyword (FTS), and hybrid search
Advanced Filtering: Exact filters with operators, semantic similarity filters
SQL-like Operations: GROUP BY, aggregations (COUNT, SUM, AVG, etc.), HAVING, ORDER BY
Result Processing: Pagination, deduplication, reranking, context windows
Async Support: Native async execution with optimized semantic filtering
Performance: Query planning, execution hints, batch processing
Debugging: Query explanation, execution statistics, SQL-like previews

Examples

Basic Vector Search:

# Simple semantic search
results = (db.query_builder()
    .search("machine learning algorithms")
    .limit(10)
    .execute())

Advanced Filtering and Search Combinations:

# Complex multi-field query with exact and semantic filters
results = (db.query_builder()
    .search("deep learning frameworks")
    .filter("year", gte_=2020, lt_=2024)          # Year range
    .filter("category", "AI")                     # Exact match
    .semantic_filter("methodology", "supervised learning", threshold=0.75)
    .order_by("year", "desc")
    .limit(50)
    .execute())

Asynchronous Execution:

# Async query with semantic filtering
async def search_research_papers():
    results = await (db.query_builder()
        .search("quantum computing algorithms")
        .filter("year", gte_=2022)
        .semantic_filter("approach", "quantum machine learning", threshold=0.8)
        .order_by("citation_count", "desc")
        .limit(25)
        .execute_async())
    return results

class localvectordb.query_builder.SimilarityMetric(*values)

Bases: Enum

Supported similarity metrics for semantic filtering.

COSINE = 'cosine'

EUCLIDEAN = 'euclidean'

DOT_PRODUCT = 'dot_product'

MANHATTAN = 'manhattan'

classmethod from_value(value: SimilarityMetric | str) → SimilarityMetric

Coerce a metric to the enum, accepting the string forms sent over HTTP.

The remote query-builder path (and the /query-builder endpoint) pass metric as a plain string, and the client’s public API uses "dot" as the short form of dot_product. Without this coercion a remote semantic_filter with a non-default metric stored a bare string that never compared equal to any enum member, raising “Unsupported similarity metric” at execution time.

class localvectordb.query_builder.SearchClause(field: str, query: str, weight: float = 1.0, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', score_threshold: float | None = None)

Bases: object

Represents a search operation on a specific field.

field: str

query: str

weight: float = 1.0

search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid'

score_threshold: float | None = None

__init__(field: str, query: str, weight: float = 1.0, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', score_threshold: float | None = None) → None

class localvectordb.query_builder.SemanticFilter(field: str, concept: str, threshold: float, metric: SimilarityMetric = SimilarityMetric.COSINE, embedding_model: str | None = None)

Bases: object

Semantic filtering based on conceptual similarity with async support.

field: str

concept: str

threshold: float

metric: SimilarityMetric = 'cosine'

embedding_model: str | None = None

async apply_async(documents: List[Document], db: BaseVectorDB) → List[Document]: Apply semantic filtering with async embedding generation.

apply(documents: List[Document], db: BaseVectorDB) → List[Document]: Apply semantic filtering synchronously.

__init__(field: str, concept: str, threshold: float, metric: SimilarityMetric = SimilarityMetric.COSINE, embedding_model: str | None = None) → None

class localvectordb.query_builder.AggregationClause(field: str, function: Literal['count', 'sum', 'avg', 'min', 'max', 'std', 'var'], alias: str | None = None)

Bases: object

Represents an aggregation operation.

field: str

function: Literal['count', 'sum', 'avg', 'min', 'max', 'std', 'var']

alias: str | None = None

__init__(field: str, function: Literal['count', 'sum', 'avg', 'min', 'max', 'std', 'var'], alias: str | None = None) → None

class localvectordb.query_builder.QueryBuilder(db: BaseVectorDB)

Bases: object

Fluent interface for building complex vector database queries with async support.

This class provides a SQL-like interface for building sophisticated search and filter operations against vector databases.

__init__(db: BaseVectorDB)

clone() → QueryBuilder: Create a copy of this QueryBuilder for chaining.

search(query: str, search_type=None, vector_weight=None, score_threshold=None) → QueryBuilder: Search the content for query text.

search_field(field: str, query: str) → QueryBuilder: Find records where field contains query.

filter(field: str | None = None, value: Any = None, **kwargs: Any) → QueryBuilder

Add exact filter conditions.

Parameters:

field (str, optional) – Field to filter on
value (Any, optional) – Value to filter for
kwargs (dict) – Key-value pairs for filtering multiple fields, or operator suffixes for advanced filtering

Returns:

New QueryBuilder instance with filter conditions added

Return type:

QueryBuilder

Examples

Basic filtering:

# Simple equality filter
builder.filter("year", 2023)

# Multiple fields
builder.filter(year=2023, category="AI")

Advanced filtering:

# Range conditions
builder.filter("year", gt_=2020, lt_=2024)

# Field exists
builder.filter("tags", exists_=True)

Raises:: ValueError – If field is not a string when provided directly

semantic_filter(field: str, concept: str, threshold: float = 0.8, metric: SimilarityMetric | str = SimilarityMetric.COSINE) → QueryBuilder

Add semantic filtering based on conceptual similarity.

metric accepts either a SimilarityMetric or its string form ("cosine", "euclidean", "dot"/"dot_product", "manhattan"); strings are coerced so the remote/query-builder path works identically to the local one.

limit(n: int) → QueryBuilder

Limit the number of results.

Parameters:: n (int) – Maximum number of results to return
Returns:: New QueryBuilder instance with limit applied
Return type:: QueryBuilder
Raises:: ValueError – If n is not positive

offset(n: int) → QueryBuilder

Skip the first n results.

Parameters:: n (int) – Number of results to skip
Returns:: New QueryBuilder instance with offset applied
Return type:: QueryBuilder
Raises:: ValueError – If n is negative

vector(query, score_threshold=None) → QueryBuilder: Use vector search.

keyword(query, score_threshold=None) → QueryBuilder: Use keyword search.

hybrid(query, vector_weight: float = 0.5, score_threshold=None) → QueryBuilder: Use hybrid search with specified vector weight.

semantic_dedup(threshold: float) → QueryBuilder: Enable semantic deduplication with similarity threshold.

documents(scoring_method: Literal['best', 'average', 'frequency_boost'] = 'frequency_boost', scoring_options: dict | None = None) → QueryBuilder: Return full documents in results (default).

chunks() → QueryBuilder: Return individual chunks in results with position information.

sections() → QueryBuilder: Return section-level results (requires hierarchical_embeddings).

search_level(level: Literal['chunks', 'sections', 'documents', 'fused'], section_weight: float | None = None) → QueryBuilder

Set the retrieval level (‘chunks’, ‘sections’, ‘documents’, or ‘fused’).

Parameters:

level (str) – Which level to search. ‘fused’ blends chunk retrieval with section (raw-span) retrieval and requires a hierarchical database.
section_weight (float, optional) – Weight on the section leg for level='fused' (0-1). Ignored otherwise.

context(window_size: int = 2) → QueryBuilder: Return chunks with context window.

order_by(field: str, direction: str = 'desc') → QueryBuilder: Add ordering by specified field.

order_by_score(direction: str = 'desc') → QueryBuilder: Order results by relevance score.

clear_ordering() → QueryBuilder: Remove all ordering clauses.

group_by(*fields: str) → QueryBuilder: Group results by one or more fields.

aggregate(field: str, function: Literal['count', 'sum', 'avg', 'min', 'max', 'std', 'var'], alias: str | None = None) → QueryBuilder: Add an aggregation function.

count_by(field: str = '*', alias: str | None = None) → QueryBuilder: Count documents/chunks, optionally grouped by field.

sum_by(field: str, alias: str | None = None) → QueryBuilder: Sum numeric values in a field.

avg_by(field: str, alias: str | None = None) → QueryBuilder: Calculate average of numeric values in a field.

min_by(field: str, alias: str | None = None) → QueryBuilder

Find minimum value in a field.

Parameters:

field (str) – Field name containing values
alias (str, optional) – Alias for the minimum result

Returns:

New QueryBuilder instance with minimum aggregation

Return type:

QueryBuilder

max_by(field: str, alias: str | None = None) → QueryBuilder

Find maximum value in a field.

Parameters:

field (str) – Field name containing values
alias (str, optional) – Alias for the maximum result

Returns:

New QueryBuilder instance with maximum aggregation

Return type:

QueryBuilder

having(field: str, operator: str, value: Any) → QueryBuilder: Add HAVING clause for post-aggregation filtering.

Note

Unlike filter(), having supports only the comparison operators eq, ne, gt, gte, lt and lte. Set/pattern operators such as in, like or contains are not available for HAVING clauses.

having_count(operator: str, value: int, alias: str = 'count') → QueryBuilder: Add HAVING clause for count aggregations.

rerank(method: str, **config) → QueryBuilder: Add reranking configuration for result post-processing.

rerank_by_recency(date_field: str = 'updated_at', weight: float = 1.0) → QueryBuilder: Rerank results by recency (newer documents ranked higher).

rerank_by_diversity(field: str, weight: float = 1.0) → QueryBuilder: Rerank results to promote diversity in specified field.

rerank_by_model(provider: str, model: str | None = None, top_k: int | None = None, **config) → QueryBuilder

Rerank results using a cross-encoder or reranking model.

Parameters:

provider (str) – Reranker provider name (e.g., “sentence_transformers”, “jina”, “huggingface”, “mock”).
model (str, optional) – Model name. If None, provider default is used.
top_k (int, optional) – Maximum results to keep after reranking.
**config – Additional configuration passed to the reranker.

explain(detailed: bool = False, return_plan: bool = False) → QueryBuilder | Dict[str, Any]

Enable query explanation or return execution plan directly.

Parameters:

detailed (bool, optional) – If True, includes additional details in explanation/plan
return_plan (bool, optional) – If True, returns the execution plan dict instead of QueryBuilder. If False (default), returns QueryBuilder with explanation enabled.

Returns:

If return_plan=False: QueryBuilder with explanation enabled If return_plan=True: Execution plan dictionary

Return type:

Union[QueryBuilder, Dict[str, Any]]

Examples

Traditional usage (returns QueryBuilder with explain enabled):

results = (db.query_builder()
    .search("machine learning")
    .explain(detailed=True)
    .execute())

New usage (returns execution plan directly):

plan = (db.query_builder()
    .search("machine learning")
    .explain(detailed=True, return_plan=True))
print(f"Query will execute: {plan['steps']}")

validate() → Dict[str, Any]: Validate the current query configuration and return validation results.

debug_info() → Dict[str, Any]: Get detailed debugging information about the current query state.

execute(*, streaming: Literal[False] = False, batch_size: int = 100) → List[QueryResult]

execute(*, streaming: Literal[True], batch_size: int = 100) → Iterator[List[QueryResult]]

Execute the query and return results.

Parameters:

streaming (bool, default = False) – If True, return an iterator that yields batches of results instead of all results at once
batch_size (int, default = 100) – Size of each batch when streaming is enabled

Returns:

Query results. Returns List when streaming=False, Iterator[List] when streaming=True

Return type:

Union[List[QueryResult], Iterator[List[QueryResult]]]

async execute_async(*, streaming: bool = False, batch_size: int = 100) → List[QueryResult] | Iterator[List[QueryResult]]

Execute the query asynchronously with native async support.

Parameters:

streaming (bool, default = False) – If True, return an iterator that yields batches of results instead of all results at once
batch_size (int, default = 100) – Size of each batch when streaming is enabled

Returns:

Query results. Returns List when streaming=False, Iterator[List] when streaming=True

Return type:

Union[List[QueryResult], Iterator[List[QueryResult]]]

cursor(batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Create a QueryCursor for this query configuration.

The cursor performs FAISS/FTS search once and lazily loads content/metadata from SQLite in batches as you iterate.

Parameters:

batch_size (int) – Default number of results per batch (default 50).
cursor_ttl (float) – Cursor time-to-live in seconds (default 300).

Return type:

QueryCursor

async cursor_async(batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Create a QueryCursor asynchronously.

Parameters:

batch_size (int) – Default number of results per batch (default 50).
cursor_ttl (float) – Cursor time-to-live in seconds (default 300).

Return type:

QueryCursor

stream(batch_size: int = 100) → Iterator[List[QueryResult]]: Stream results in batches using a cursor (single FAISS/FTS search).

async stream_async(batch_size: int = 100): Stream results in batches asynchronously using a cursor.

count() → int: Get count of matching results without returning them.

async count_async() → int: Get count of matching results asynchronously.

get_execution_plan(detailed: bool = False) → Dict[str, Any]

Get the execution plan for this query without executing it.

This method allows you to preview how the query will be executed, including the steps, estimated cost, and optimizations that will be applied.

Parameters:: detailed (bool, optional) – If True, includes additional details like field usage and optimization hints
Returns:: Execution plan containing: - steps: List of execution steps - estimated_cost: Relative cost estimate - query_type: Type of query (search, filter, hybrid) - optimizations: List of applied optimizations - details: Additional details if detailed=True
Return type:: Dict[str, Any]

Examples

>>> plan = (db.query_builder()
...     .search("machine learning")
...     .filter("year", gte=2020)
...     .get_execution_plan())
>>> print(f"Query type: {plan['query_type']}")
>>> print(f"Steps: {plan['steps']}")

async get_execution_plan_async(detailed: bool = False) → Dict[str, Any]

Get the execution plan for this query without executing it (async version).

Parameters:: detailed (bool, optional) – If True, includes additional details like field usage and optimization hints
Returns:: Execution plan with same structure as get_execution_plan()
Return type:: Dict[str, Any]

class localvectordb.query_builder.QueryExecutor(query_builder: QueryBuilder)

Bases: _QueryExecutorBase

Synchronous executor for QueryBuilder queries.

execute() → List[QueryResult]: Execute the query synchronously with full feature support.

count() → int: Get count of matching results.

cursor(batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor: Create a QueryCursor by delegating to db.query_cursor().

stream(batch_size: int = 100) → Iterator[List[QueryResult]]: Stream results in batches using a cursor (single FAISS/FTS search).

class localvectordb.query_builder.AsyncQueryExecutor(query_builder: QueryBuilder)

Bases: _QueryExecutorBase

Asynchronous executor for QueryBuilder queries with native database async support.

async execute() → List[QueryResult]: Execute the query asynchronously with full feature support.

async count() → int: Get count of matching results asynchronously.

async cursor(batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor: Create a QueryCursor asynchronously by delegating to db.query_cursor_async().

async stream(batch_size: int = 100): Stream results in batches asynchronously using a cursor.