localvectordb.database package

LocalVectorDB is assembled from mixins and defines no members of its own, so :inherited-members: is what makes this page document anything at all. The argument ABC names the ancestor to stop at: members from ABC and above (i.e. object) are skipped, while everything the mixins contribute is kept. Drop the option and the class documents zero methods – silently, with a clean build.

class localvectordb.database.LocalVectorDB(*args: Any, **kwargs: Any)

Bases: LocalTuningMixin, PipelineMixin, SearchMixin, MetadataMixin, CrudMixin, ComparisonMixin, RepairMixin, DiagnoseMixin, LocalVectorDBCore

Document-first vector database with SQLite + FAISS + embeddings

This is the main interface for LocalVectorDB v1.0, designed around documents rather than chunks. All chunking is handled internally.

Parameters:

name (str) – Database name (used for file naming)
base_path (str, optional) – Directory to store database files, by default “.lvdb”
metadata_schema (str | Dict[str, MetadataField], optional) – Schema definition for metadata fields
doc_id_pattern (str, optional) – Pattern for auto-generating document IDs, by default “doc_{idx}”
embedding_provider (str, optional) – Embedding provider name, by default “ollama”
embedding_model (str, optional) – Embedding model name, by default “embeddinggemma”
embedding_config (Dict[str, Any], optional) –
Configuration passed to the embedding provider. Beyond provider-specific options (base_url, api_key, …), two keys apply to every provider:

document_prefix / query_prefix
Instruction prefixes for asymmetric retrieval models, prepended when embedding for storage and for search respectively. Both default to the model’s known training prefix (embeddinggemma, nomic-embed-text, snowflake-arctic-embed*, bge-*-en, e5-*); an unrecognised model gets none. Pass "" to force no prefix, or auto_prefix=False to disable the lookup.

An existing database reuses the prefixes it was built with – they are part of its vector space – so these take effect on a new database, or on a re-ingest. See the Embeddings guide for details.
reranker_config (Dict[str, Any], optional) – Persisted default reranker for this database, applied by every query() unless overridden per call (reranker=False disables for one call). Same shape as query()’s per-call reranker_config: {"provider": ..., "model": ..., **kwargs}. Saved in the database’s config (like the embedding provider) so the CLI, MCP and server all inherit it. For hosted providers store credentials as environment references (api_key="$MY_KEY_VAR"), never raw secrets. Change or clear it later with set_default_reranker(). On reopen, an explicit value here overrides (and re-persists over) the saved one, with a warning; None means “use what is saved”.
chunking_method (str, optional) – Chunking method, by default “sentences”
chunk_size (int, optional) – Maximum tokens per chunk, by default 500
chunk_overlap (int, optional) – Overlap between consecutive chunks, by default 1. Measured in the unit of chunking_method (sentences for “sentences”, tokens for “tokens”, words for “words”, lines for “lines”/”code-blocks”, characters for “characters”, paragraphs for “paragraphs”), NOT tokens — only “tokens” shares its unit with chunk_size. Keep it small (e.g. 1-3); a value larger than the number of units a chunk holds produces highly redundant chunks.
enable_gpu (bool, optional) – Whether to use GPU for FAISS, by default False
enable_fts (bool, optional) – Whether to enable full-text search, by default True
create_if_not_exists (bool, default = True) – If False, raises DatabaseNotFoundError if the database doesn’t exist.

__init__(*args: Any, **kwargs: Any)

analyze_system_resources() → Dict[str, Any]

Analyze system resources for tuning recommendations.

Returns:: System resource information
Return type:: Dict[str, Any]

auto_tune(workload: Dict[str, Any] | None = None, interactive: bool = False, apply: bool = False) → Dict[str, Any]

Get auto-tuning recommendations based on system and workload.

Parameters:

workload (Dict[str, Any], optional) – Workload characteristics. If None and interactive=True, will prompt user.
interactive (bool, optional) – Whether to run interactive interview for workload, by default False
apply (bool, optional) – Whether to apply the recommended settings, by default False

Returns:

Tuning recommendation containing: - profile_name: Recommended profile - pragma_overrides: Recommended pragma overrides - reasoning: List of reasoning explanations - estimated_memory_mb: Estimated memory usage

Return type:

Dict[str, Any]

property batch_size: int: Batch size for processing.

checkpoint_if_wal_large(wal_mb_threshold: int = 128) → bool: Check if WAL file is large and checkpoint if needed.

property chunk_delimiter: str: The delimiter used by chunking_method="delimiter" (ignored otherwise).

property chunk_overlap: int: Overlap between chunks, in the unit of chunking_method (not tokens unless the method is "tokens").

chunk_similarity_matrix(doc_id_1: str, doc_id_2: str | None = None) → ChunkSimilarityMatrix

Compute the full chunk-level pairwise similarity matrix.

When doc_id_2 is None, computes self-similarity within doc_id_1 (useful for chord diagrams).

Parameters:

doc_id_1 (str) – First document ID.
doc_id_2 (str, optional) – Second document ID. Defaults to doc_id_1.

Return type:

ChunkSimilarityMatrix

property chunk_size: int: Return the maximum tokens per chunk.

property chunking_method: str: Return the chunking method.

close(): Close the database

async close_async(): Close async resources

property closed: bool: Return whether the database connection is closed.

compare_documents(doc_id_1: str, doc_id_2: str) → float

Return cosine similarity [0, 1] between two documents (centroid-based).

Parameters:

doc_id_1 (str) – First document ID.
doc_id_2 (str) – Second document ID.

Returns:

Cosine similarity normalised to [0, 1].

Return type:

float

async compare_documents_async(doc_id_1: str, doc_id_2: str) → float: Async twin of compare_documents().

compare_documents_detailed(doc_id_1: str, doc_id_2: str, chunk_threshold: float = 0.7) → DocumentComparisonResult

Rich chunk-level comparison between two documents.

For each chunk in doc_id_1, finds the best-matching chunk in doc_id_2 (and vice-versa). A chunk counts as “matched” if its best-match similarity meets chunk_threshold.

Parameters:

doc_id_1 (str) – First document ID.
doc_id_2 (str) – Second document ID.
chunk_threshold (float) – Minimum similarity for a chunk pair to count as “matched”.

Return type:

DocumentComparisonResult

async compare_documents_detailed_async(doc_id_1: str, doc_id_2: str, chunk_threshold: float = 0.7) → DocumentComparisonResult: Async twin of compare_documents_detailed().

count(filters: Dict[str, Any] | None = None) → int

Count documents matching filter criteria

Parameters:: filters (Optional[Dict[str, Any]]) – Filter criteria using MongoDB-style syntax, by default None
Returns:: Number of documents matching the criteria
Return type:: int

async count_async(filters: Dict[str, Any] | None = None) → int

Async count documents matching filter criteria

Parameters:: filters (Optional[Dict[str, Any]]) – Filter criteria using MongoDB-style syntax, by default None
Returns:: Number of documents matching the criteria
Return type:: int

delete(ids: str | List[str]) → int

Delete documents

Parameters:: ids (Union[str, List[str]]) – Document ID(s) to delete
Returns:: Number of documents deleted
Return type:: int

async delete_async(ids: str | List[str]) → int

Async delete documents by ID

Parameters:: ids (Union[str, List[str]]) – Document ID(s) to delete
Returns:: Number of documents deleted
Return type:: int

diagnose(sample: int = 5000) → DiagnoseReport

Measure which retrieval regime this corpus is in.

Reports encoder coverage of chunk text (text past the context window never enters any vector), section length against the encoder window, the chunk->section reachability ceiling, chunks-per-document / chunks-per-section fanout, and the health of each keyword (FTS) leg.

Token counts use the encoder’s own tokenizer where one is importable (sentence-transformers, local HuggingFace, OpenAI); otherwise they are estimates and the report says so rather than printing a confident percentage.

Parameters:: sample (int) – Maximum number of chunks to re-tokenize when an exact tokenizer is available (sections are capped lower). Sampling is evenly strided over id order, so repeat runs on an unchanged database measure the same rows. With no exact tokenizer, chunk coverage uses the stored per-chunk token counts over the full corpus instead.
Returns:: Structured results; report.summary renders them for humans.
Return type:: DiagnoseReport

async diagnose_async(sample: int = 5000) → DiagnoseReport: Async variant of diagnose(); same measurement, same report.

property embedding_dimension: int: Return the dimension of the embeddings.

property embedding_model: str: Return the embedding model name.

property embedding_provider: EmbeddingProvider: Return the embedding provider name or instance.

exists(ids: str | List[str]) → bool | List[bool]

Check if documents exist

Parameters:: ids (Union[str, List[str]]) – Document ID(s) to check
Returns:: Existence status for each ID
Return type:: Union[bool, List[bool]]

async exists_async(ids: str | list[str]) → bool | list[bool]

Async check if a document exists

Parameters:: ids (str) – Document ID to check
Returns:: True if document exists, False otherwise
Return type:: bool or list[bool]

filter(where: Dict[str, Any] | None = None, order_by: str | None = None, limit: int | None = None, offset: int = 0) → List[Document]

Filter documents using enhanced metadata filtering

This method supports advanced MongoDB-style filtering with operators like $gt, $lt, $contains, $exists, etc.

Parameters:

where (Optional[Dict[str, Any]]) –
Filter conditions using either simple format or MongoDB-style operators.

Simple format:
```
{"author": "John Doe", "year": 2023}
```
Advanced format with operators:
```
{
    "author": {"$eq": "John Doe"},
    "year": {"$gte": 2020, "$lte": 2024},
    "tags": {"$contains": "python"},
    "rating": {"$in": [4, 5]},
    "$and": [
        {"category": "tech"},
        {"$or": [{"lang": "en"}, {"lang": "es"}]}
    ]
}
```
Supported operators:
- Comparison: $eq, $ne, $gt, $lt, $gte, $lte, $in, $nin
- String: $like, $ilike, $contains, $startswith, $endswith
- Existence: $exists, $not_exists
- Type: $type
- Logical: $and, $or, $not
- JSON: $contains, $not_contains (for JSON fields)
order_by (Optional[str]) – ORDER BY clause (field name with optional ASC/DESC) Examples: “created_at DESC”, “author ASC”, “rating”
limit (Optional[int]) – Maximum number of results
offset (int) – Number of results to skip

Returns:

Filtered documents

Return type:

List[Document]

Examples

Simple filtering:

# Simple equality
docs = db.filter(where={"author": "John Doe"})

# Multiple conditions (AND)
docs = db.filter(where={"author": "John Doe", "year": 2023})

Advanced filtering:

# Range queries
docs = db.filter(where={
    "year": {"$gte": 2020, "$lte": 2024},
    "rating": {"$gt": 4.0}
})

# String operations
docs = db.filter(where={
    "title": {"$contains": "python"},
    "author": {"$startswith": "Dr."}
})

# List operations
docs = db.filter(where={
    "category": {"$in": ["tech", "science"]},
    "tags": {"$contains": "tutorial"}
})

# Logical operations
docs = db.filter(where={
    "$and": [
        {"year": {"$gte": 2020}},
        {"$or": [
            {"author": "John Doe"},
            {"author": "Jane Smith"}
        ]}
    ]
})

# Existence checks
docs = db.filter(where={
    "optional_field": {"$exists": False},
    "required_field": {"$exists": True}
})

Notes

All queries are converted to safe parameterized SQL
Field names are validated against the metadata schema
JSON fields support special operations like $contains

async filter_async(where: Dict[str, Any] | None = None, order_by: str | None = None, limit: int | None = None, offset: int = 0) → List[Document]

Async filter documents by metadata criteria

Parameters:

where (Dict[str, Any]) – Filter criteria using MongoDB-style syntax
order_by (Optional[str]) – SQL-style ORDER BY clause (e.g., ‘created_at DESC’), by default None
limit (Optional[int]) – Maximum number of documents to return, by default None
offset (Optional[int]) – Number of documents to skip, by default 0

Returns:

Documents matching the filter criteria

Return type:

List[Document]

property fts_enabled: bool: Return whether full-text search is enabled.

get(ids: str | List[str]) → Document | List[Document]

Retrieve documents by ID

Parameters:: ids (Union[str, List[str]]) – Document ID(s) to retrieve
Returns:: Retrieved document(s)
Return type:: Union[Document, List[Document]]
Raises:: DocumentNotFoundError – If any requested documents are not found

async get_async(ids: str | List[str]) → Document | List[Document]

Async retrieve documents by ID

Parameters:: ids (Union[str, List[str]]) – Document ID(s) to retrieve
Returns:: The requested document(s)
Return type:: Union[Document, List[Document]]
Raises:: DocumentNotFoundError – If any requested documents are not found

async get_async_pool_stats() → Dict[str, Any]

Get async connection-pool statistics.

Renamed from get_async_stats for v0.1.0: the old name collided with (and read as) the async twin of get_stats while actually returning pool internals. This reports the async pool only.

get_chunk_embeddings(chunk_ids: str | List[str]) → ndarray

Returns embeddings for chunks given by chunk_ids”

Parameters:: chunk_ids (str | List[str]) – The chunk_ids for which to return embeddings
Return type:: np.ndarray

get_chunks(document_id: str, indices: List[int] | None = None) → List[Chunk]

Retrieve the persisted chunks of a document, ordered by chunk index.

Returns the chunks exactly as they were stored at ingest time (content plus full character/line position), unlike get(), which only returns the whole-document content.

Parameters:

document_id (str) – The document whose chunks to retrieve.
indices (list[int], optional) – If given, only chunks whose chunk_index appears in this list are returned; unknown indices are silently skipped. When None (default), every chunk of the document is returned.

Returns:

Chunks with full position information, ordered by chunk_index. Empty if the document has no chunks (or none match indices).

Return type:

list[Chunk]

Notes

Synchronous only; the CLI get command is the sole consumer. Add an async twin if a future async caller needs one.

get_default_reranker() → Dict[str, Any] | None: Return a copy of this database’s persisted default reranker config, if any.

get_metadata_schema_info() → Dict[str, Any]

Get detailed information about the current metadata schema

Returns:: Dictionary containing: - fields: Dict of field definitions - field_count: Number of fields - indexed_fields: List of indexed field names - required_fields: List of required field names - field_types: Summary of field types used
Return type:: Dict[str, Any]

async get_metadata_schema_info_async() → Dict[str, Any]

Get detailed information about the current metadata schema asynchronously

Returns:: Dictionary containing: - fields: Dict of field definitions - field_count: Number of fields - indexed_fields: List of indexed field names - required_fields: List of required field names - field_types: Summary of field types used
Return type:: Dict[str, Any]

get_sqlite_tuning() → Dict[str, Any]: Get current SQLite tuning configuration from local database.

get_stats() → Dict[str, Any]

Get database statistics

Returns:: A dict with the following keys: - documents - chunks - index_vectors - embedding_dimension - embedding_provider - embedding_model - chunking_method - chunk_size - chunk_overlap - fts_enabled - schema_version (on-disk table-layout version) and created_by_version
Return type:: dict[str, int|str|bool]

async get_stats_async() → Dict[str, Any]

Async twin of get_stats() (database statistics).

Mirrors RemoteVectorDB.get_stats_async so await db.get_stats_async() works against either backend. Statistics collection is lightweight sync SQLite work, so this delegates to the sync path.

grep(pattern: str, *, regex: bool = False, ignore_case: bool = False, whole_word: bool = False, context: int = 0, before_context: int | None = None, after_context: int | None = None, prefix: str | None = None, where: Dict[str, Any] | None = None, max_count: int | None = None, limit: int | None = None) → List[GrepMatch]

Lexical, line-oriented search over document content – like grep.

This is exact/regex substring matching, deliberately separate from query(). query does ranked semantic/keyword retrieval; grep finds literal or regex matches and reports where they are (document id, line number, column span, and optional surrounding lines). Agents use it alongside vector and keyword search when they know a precise string or pattern to look for. Results are returned in document-id then line order, not by relevance.

Parameters:

pattern (str) – The text to search for. A literal substring by default; a regular expression when regex=True.
regex (bool) – Treat pattern as a Python regular expression. Default False (literal match).
ignore_case (bool) – Case-insensitive matching. Default False.
whole_word (bool) – Require the match to fall on word boundaries (wraps the pattern in \b...\b). Default False.
context (int) – Number of adjacent lines to include both before and after each match (like grep -C). Default 0.
before_context (Optional[int]) – Lines of leading context (like grep -B). Overrides context for the “before” side when set.
after_context (Optional[int]) – Lines of trailing context (like grep -A). Overrides context for the “after” side when set.
prefix (Optional[str]) – Restrict the scan to documents whose id starts with this literal prefix (case-sensitive). Pair with list_prefixes() to grep within a virtual “folder”.
where (Optional[Dict[str, Any]]) – Restrict the scan to documents matching this metadata filter (same syntax as filter()).
max_count (Optional[int]) – Stop after this many matches per document (like grep -m).
limit (Optional[int]) – Stop after this many matches in total across all documents.

Returns:

One entry per match, in document-id then line-number order.

Return type:

List[GrepMatch]

Notes

Matching runs line-by-line over the stored document content. Narrow the corpus with prefix / where on large databases, since every matched document is scanned.

property hierarchical_embeddings: bool

Insert new documents into the database with pipeline processing

Parameters:

documents (Union[str, List[str]]) – Document text(s) to add
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Only fields declared in the metadata schema are stored; fields not in the schema are dropped with a logged warning.
ids (Optional[Union[str, List[str]]]) – Document IDs (auto-generated if not provided)
batch_size (int) – Batch size for processing, by default 100
similarity_threshold (Optional[float]) – Skip chunks that are too similar to existing chunks
errors (Literal["ignore", "raise"]) – How to handle document ID conflicts, by default “raise”

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

async insert_async(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2, **kwargs: Any) → List[str]

Insert new documents into the database with async pipeline

Parameters:

documents (Union[str, List[str]]) – Document text(s) to add
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Only fields declared in the metadata schema are stored; fields not in the schema are dropped with a logged warning.
ids (Optional[Union[str, List[str]]]) – Document IDs (auto-generated if not provided)
batch_size (int) – Batch size for processing, by default 100
errors (Literal["ignore", "raise"]) – How to handle document ID conflicts, by default “raise”
similarity_threshold (Optional[float]) – Skip chunks that are too similar to existing chunks
max_concurrent_chunks (int, default=3) – Maximum concurrent chunking operations
max_concurrent_embeddings (int, default=2) – Maximum concurrent embedding operations

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

insert_from_chunks(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise') → List[str]

Insert documents from pre-chunked data with conflict handling.

Similar to upsert_from_chunks but fails on duplicate document IDs unless configured to ignore them.

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks. Chunks can be either: - List[Chunk]: Full Chunk objects with position information - List[str]: Simple strings that will be converted to Chunk objects
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata. If None, empty metadata is used for all documents.
batch_size (int, default=100) – Number of embeddings to generate at once
similarity_threshold (Optional[float], default=None) – If provided, filters out chunks that are too similar to existing chunks
errors (Literal["ignore", "raise"], default="raise") – How to handle document ID conflicts: - “raise”: Raise DuplicateDocumentIDError - “ignore”: Skip existing documents and continue

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

DuplicateDocumentIDError – If a document ID already exists and errors=”raise”
ValueError – If chunk data is invalid or metadata doesn’t match schema

async insert_from_chunks_async(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2) → List[str]

Async version of insert_from_chunks - Insert documents from pre-chunked data with conflict handling.

Similar to upsert_from_chunks_async but fails on duplicate document IDs unless configured to ignore them.

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks. Chunks can be either: - List[Chunk]: Full Chunk objects with position information - List[str]: Simple strings that will be converted to Chunk objects
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata. If None, empty metadata is used for all documents.
batch_size (int, default=None) – Number of embeddings to generate at once. If None, uses default from configuration.
similarity_threshold (Optional[float], default=None) – If provided, filters out chunks that are too similar to existing chunks
errors (Literal["ignore", "raise"], default="raise") – How to handle document ID conflicts: - “raise”: Raise DuplicateDocumentIDError - “ignore”: Skip existing documents and continue
max_concurrent_chunks (int, default=3) – Maximum number of concurrent chunk processing operations
max_concurrent_embeddings (int, default=2) – Maximum number of concurrent embedding operations

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

DuplicateDocumentIDError – If a document ID already exists and errors=”raise”
ValueError – If chunk data is invalid or metadata doesn’t match schema

Insert new documents from files using file extraction.

Uses the ExtractorRegistry to automatically extract text from files based on file extension and MIME type, then calls the regular insert method.

Parameters:

file_paths (Union[str, Path, List[Union[str, Path]]]) – Path(s) to files to extract and insert
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Will be merged with extracted metadata.
ids (Optional[Union[str, List[str]]]) – Document IDs. If not provided, will use filename without extension.
batch_size (int) – Batch size for processing, by default 100
similarity_threshold (Optional[float]) – Skip chunks that are too similar to existing chunks
errors (Literal["ignore", "raise"]) – How to handle document ID conflicts, by default “raise”
extractor_kwargs (Optional[Dict[str, Any]]) – Additional keyword arguments passed to the extractor

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

FileNotFoundError – If any of the specified files don’t exist
ValueError – If extraction fails for any file and no fallback is available
DuplicateDocumentIDError – If errors=”raise” and document ID conflicts occur

Async insert new documents from files using file extraction.

Uses the ExtractorRegistry to automatically extract text from files based on file extension and MIME type, then calls the regular insert_async method.

Parameters:

file_paths (Union[str, Path, List[Union[str, Path]]]) – Path(s) to files to extract and insert
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Will be merged with extracted metadata.
ids (Optional[Union[str, List[str]]]) – Document IDs. If not provided, will use filename without extension.
batch_size (int) – Batch size for processing
similarity_threshold (Optional[float]) – Skip chunks that are too similar to existing chunks
errors (Literal["ignore", "raise"]) – How to handle document ID conflicts, by default “raise”
max_concurrent_chunks (int, default=3) – Maximum concurrent chunking operations
max_concurrent_embeddings (int, default=2) – Maximum concurrent embedding operations
extractor_kwargs (Optional[Dict[str, Any]]) – Additional keyword arguments passed to the extractor

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

FileNotFoundError – If any of the specified files don’t exist
ValueError – If extraction fails for any file and no fallback is available
DuplicateDocumentIDError – If errors=”raise” and document ID conflicts occur

property is_memory_only: bool

list_prefixes(prefix: str = '', delimiter: str = '/') → PrefixListing

List the immediate children of a document-id prefix, S3-style.

Treats delimiter as a virtual path separator over document ids and rolls documents up to their first segment beneath prefix. A useful pattern is to use relative paths as manual document ids (docs/reports/q1) and then navigate them like folders – there are no real directories, only ids that share a prefix.

Parameters:

prefix (str) – Literal id prefix to list beneath. Pass "" (the default) to list the top level. For folder-like navigation, include the trailing delimiter ("docs/" lists the children of docs/, whereas "docs" lists children of every id beginning with docs).
delimiter (str) – Virtual path separator. Defaults to "/".

Returns:

prefixes are the virtual sub-folders (common prefixes) with a recursive document count; documents are the leaf documents that live directly at this level.

Return type:

PrefixListing

Examples

db.upsert(["..."], ids=["docs/reports/q1"])
listing = db.list_prefixes("docs/")
for folder in listing.prefixes:
    print(folder.path, folder.count)   # e.g. "docs/reports/" 1

Notes

Matching is case-sensitive (SQLite GLOB / BINARY collation), which is the desired behaviour for path-like keys.
A document whose id equals prefix exactly is not reported as its own child.

list_sqlite_profiles() → Dict[str, str]

List available SQLite tuning profiles.

Returns:: Dictionary mapping profile names to descriptions
Return type:: Dict[str, str]

property metadata_schema: Dict[str, MetadataField]: Return the metadata schema.

nearest_neighbors(doc_id: str, k: int = 5, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None) → List[QueryResult]

Return the k most similar documents to doc_id.

Parameters:

doc_id (str) – Reference document.
k (int) – Maximum number of neighbours to return.
score_threshold (float) – Minimum similarity score to include.
filters (dict, optional) – Metadata filter dict applied to candidates. Filter fields must be declared in the metadata schema; unknown fields or unsupported operators raise DatabaseError.

Returns:

Sorted by score descending; the reference document is excluded.

Return type:

List[QueryResult]

async nearest_neighbors_async(doc_id: str, k: int = 5, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None) → List[QueryResult]: Async twin of nearest_neighbors().

pairwise_similarity_matrix(doc_ids: List[str] | None = None) → DocumentSimilarityMatrix

Compute an NxN similarity matrix for all (or selected) documents.

Parameters:: doc_ids (list of str, optional) – Specific document IDs. If None, all documents are used.
Return type:: DocumentSimilarityMatrix

async pairwise_similarity_matrix_async(doc_ids: List[str] | None = None) → DocumentSimilarityMatrix: Async twin of pairwise_similarity_matrix().

patch(doc_id: str, ops: List[Dict[str, Any]], *, expect_hash: str | None = None, metadata: Dict[str, Any] | None = None) → PatchResult

Patch a document’s content with find/replace or span-splice ops.

Unlike update() (which replaces the whole content string), patch applies targeted edits resolved against the document’s current content, so a caller need not re-send the untouched remainder. See localvectordb.patching for the op shapes.

Parameters:

doc_id (str) – Document ID to patch.
ops (List[Dict[str, Any]]) – Patch ops (splice / replace / append / prepend), resolved against the original content, non-overlapping, applied atomically.
expect_hash (Optional[str]) – If given and it does not equal the stored content_hash, the patch fails with PatchConflictError instead of clobbering a concurrent write.
metadata (Optional[Dict[str, Any]]) – Metadata merged with existing (same semantics as update()).

Returns:

updated is False only when the ops produced content identical to what is stored and no metadata changed.

Return type:

PatchResult

Raises:

DocumentNotFoundError – If doc_id does not exist.
PatchConflictError – If expect_hash is given and does not match the stored hash.
PatchError – If an op is unmatched, ambiguous, overlapping, or out of range.

async patch_async(doc_id: str, ops: List[Dict[str, Any]], *, expect_hash: str | None = None, metadata: Dict[str, Any] | None = None) → PatchResult: Patch a document’s content asynchronously. Same contract as patch().

ping() → bool: Check if the database is accessible. Override in subclasses.

query(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] | None = None, search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]

Unified query interface for all search types

Parameters:

query (str) – Query text
search_type (Literal['vector', 'keyword', 'hybrid']) – Type of search to perform
return_type (Optional[Literal['documents', 'chunks', 'sections', 'context', 'enriched']]) – The unit to report hits in: whole documents, individual chunks, sections, chunks with context, or enriched chunks with intra-document context. Defaults to None, meaning “whatever unit search_level searched”: documents for the default chunk search, sections for search_level='sections'. Pass a value to override – notably return_type='documents' with search_level='sections' ranks documents by their best-matching section.
k (int) – Maximum number of results to return
score_threshold (float) – Minimum score to keep (0-1, higher=better). For search_type="hybrid" each leg is min-max normalized within this query’s own candidate pool, so scores are not comparable across queries or across different k: the threshold cuts on rank position within the pool, not on absolute match quality, and is not a portable bar you can tune once and reuse.
filters (Optional[Dict[str, Any]]) – Metadata filters. Filter fields must be declared in the metadata schema (or be reserved columns like id/created_at); unknown fields or unsupported operators raise DatabaseError.
vector_weight (float) – Weight for vector search in hybrid mode (0-1). At search_level='fused' it applies within each granularity – the chunk leg and the section leg each blend their own vector and BM25 scores by this weight – before section_weight blends the two legs. The best value is corpus-dependent: 0.5 is the measured argmax on qasper and MAUD, but a corpus whose documents repeat the same terms in every section (Wikipedia, in our Natural Questions runs) wants ~0.9, because BM25 cannot discriminate within a document there.
search_level (Literal['chunks', 'sections', 'documents', 'fused']) – Which retrieval level to search. ‘chunks’ (default) is the normal path. ‘sections’/’documents’ search the hierarchical indices directly. ‘fused’ blends chunk retrieval with section (raw-span) retrieval. All three require hierarchical_embeddings and raise ValueError without it. ‘sections’ and ‘fused’ report either ‘sections’ or ‘documents’; ‘documents’ reports only ‘documents’. Every level honours search_type; none silently downgrades to vector-only.
section_weight (float) – Weight on the section leg when search_level='fused' (0-1): 0.0 is chunk-only, 1.0 is section-only. Default 0.65. Like vector_weight this is regime-specific rather than universal – the measured argmax is 0.65 on qasper, 0.35 on MAUD and 0.80 on Natural Questions – so treat the default as a starting point and tune it if the corpus matters. Ignored for other search levels.
context_window (int) – Size of the context to assemble for return_type=’context’/’enriched’. Interpreted in the units given by context_unit. When context_unit='chunks' (default): number of chunks before and after to include (context) or number of similar chunks to enrich. When context_unit is ‘tokens’/’words’/’characters’: an approximate budget for the assembled context content.
context_unit (Literal['chunks', 'tokens', 'words', 'characters']) – Unit in which context_window is measured, by default ‘chunks’. With a non-chunk unit, neighbouring/similar chunks are added whole, greedily, until the next one would exceed the budget (the matched chunk is always kept). Only applies to return_type=’context’/’enriched’.
context_truncate (bool) – When True and context_unit is a token/word/character budget, the assembled context is hard-truncated to exactly the budget (cutting the final chunk if needed). By default False (whole chunks only). This is the only way to guarantee the result never exceeds the budget when a single chunk is larger than it.
semantic_dedup_threshold (Optional[float]) – Similarity threshold for semantic deduplication (0-1, higher=more similar)
document_scoring_method (DocumentScoringMethod) – Method for aggregating chunk scores into document scores. One of: {“auto”, “best”, “average”, “frequency_boost”, “percentile”}. For detailed explanations and guidance on selecting the appropriate method, see the Document Scoring documentation.
document_scoring_options (dict, optional) –
Parameters to pass to the scoring method function. For complete parameter documentation and examples, see the Document Scoring documentation.

Common parameters by method:
- frequency_boost
  
  frequency_bias0.0 - 1.0, default = 0.3
  The ratio of the frequency multiplier to apply. Higher favors documents with more matching chunks
- percentile
  
  percentile0.0 - 1.0, default = 0.9
  Which order statistic of the chunk scores to take. 1.0 is exactly best; lower values soften it toward the mean. Document targets only, and only worth it at a wide candidate pool – see DocumentScoringMethod.
reranker (object or False, optional) – A reranker instance whose rerank() re-scores the candidate pool. Pass False to disable reranking for this one call, including the database’s persisted default reranker (see set_default_reranker); None (the default) means “use the database default, if one is configured”.
reranker_config (dict, optional) – Config from which the server/factory constructs a reranker, e.g. {"provider": "jina", "model": "jina-reranker-v2-base-multilingual"}. Overrides the database’s persisted default for this call.
rerank_k (int, optional) – Size of the candidate pool to fetch and hand to the reranker before truncating to k. Only has an effect when reranking is active (a reranker or reranker_config is supplied, or the database has a persisted default reranker). Defaults to 5*k (clamped to at most 200); a reranker given only k candidates cannot improve recall, since it never sees the results ranked just below the cutoff.

Returns:

Search results with normalized scores

Return type:

List[QueryResult]

async query_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] | None = None, search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]

Async query the database using vector, keyword, or hybrid search

Parameters:

query (str) – Search query text
search_type (Literal['vector', 'keyword', 'hybrid']) – Type of search to perform, by default ‘hybrid’
return_type (Optional[Literal['documents', 'chunks', 'sections', 'context', 'enriched']]) – The unit to report hits in. Defaults to None, meaning “whatever unit search_level searched” – documents for the default chunk search, sections for search_level='sections'. See query().
k (int) – Maximum number of results to return, by default 10
score_threshold (float) – Minimum score to keep (0-1, higher=better). For search_type="hybrid" each leg is min-max normalized within this query’s own candidate pool, so scores are not comparable across queries or across different k: the threshold cuts on rank position within the pool, not on absolute match quality, and is not a portable bar you can tune once and reuse., by default 0.0
filters (Optional[Dict[str, Any]]) – Metadata filters to apply, by default None. Filter fields must be declared in the metadata schema; unknown fields or unsupported operators raise DatabaseError.
vector_weight (float) – Weight for vector search in hybrid mode (0-1), by default 0.7
context_window (int) – Size of the assembled context for return_type=’context’/’enriched’, measured in context_unit (chunks before/after or similar-chunk count when ‘chunks’; an approximate token/word/character budget otherwise), by default 2
context_unit (Literal['chunks', 'tokens', 'words', 'characters']) – Unit in which context_window is measured, by default ‘chunks’.
context_truncate (bool) – Hard-truncate the assembled context to exactly the token/word/character budget (only applies with a non-chunk context_unit), by default False.
semantic_dedup_threshold (Optional[float]) – Similarity threshold for semantic deduplication (0-1, higher=more similar), by default None
document_scoring_method (DocumentScoringMethod) – Method for aggregating chunk scores into document scores, by default “frequency_boost” For detailed explanations and guidance on selecting the appropriate method, see the Document Scoring documentation.
document_scoring_options (dict, optional) – Parameters for the document_scoring_method (to choose overall scores for documents from chunk results). For complete parameter documentation and examples, see the Document Scoring documentation.
rerank_k (int, optional) – Size of the candidate pool to fetch and hand to the reranker before truncating to k. Only has an effect when reranking is active (a reranker or reranker_config is supplied, or the database has a persisted default reranker; reranker=False disables the default for this call). Defaults to 5*k (clamped to at most 200). See query() for the rationale.

Returns:

Search results with normalized scores

Return type:

List[QueryResult]

query_builder() → QueryBuilder: Returns a QueryBuilder for the database.

query_cursor(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Create a QueryCursor for streaming results with lazy hydration.

Performs the FAISS/FTS search once, caches scored candidates, and returns a cursor that lazily loads content/metadata from SQLite per batch.

Parameters match query() with the addition of:

Parameters:

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

Returns:

A cursor that can be iterated to fetch results in batches.

Return type:

QueryCursor

Raises:

ValueError – If a reranker or reranker_config is supplied. Reranking requires scoring the fully materialized result set, which is incompatible with lazy cursor hydration; use query() instead.

async query_cursor_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Async version of query_cursor. Returns a QueryCursor for async iteration.

Raises ValueError if a reranker/reranker_config is supplied; reranking is incompatible with lazy cursor hydration (use query_async()).

query_multi_column(query: str, *, columns: List[str] | None = None, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'enriched'] = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]

Query across multiple columns (main content + embedding-enabled metadata fields)

Parameters:

query (str) – Query text
columns (Optional[List[str]]) – Specific columns to search. If None, searches all embedding-enabled fields plus main content. Use ‘content’ for main document content.
search_type (Literal['vector', 'keyword', 'hybrid']) – Type of search to perform
return_type (Literal['documents', 'chunks']) – Whether to return full documents or individual chunks
k (int) – Maximum number of results to return
score_threshold (float) – Minimum score to keep (0-1, higher=better). For search_type="hybrid" each leg is min-max normalized within this query’s own candidate pool, so scores are not comparable across queries or across different k: the threshold cuts on rank position within the pool, not on absolute match quality, and is not a portable bar you can tune once and reuse.
filters (Optional[Dict[str, Any]]) – Metadata filters to apply. Filter fields must be declared in the metadata schema; unknown fields or unsupported operators raise DatabaseError.
vector_weight (float) – Weight for vector search in hybrid mode (0-1)
document_scoring_method (DocumentScoringMethod) – Method for aggregating chunk scores into document scores
document_scoring_options (dict, optional) – Parameters for the scoring method
reranker (object or False, optional) – A reranker instance applied ONCE to the merged multi-column pool (never per column leg, which would re-score legs inconsistently). False disables reranking for this call, including the database’s persisted default; None uses the default if set.
reranker_config (dict, optional) – Config from which a reranker is constructed for the merged pool; overrides the database’s persisted default for this call.
rerank_k (int, optional) – Merged-pool width handed to the reranker before truncating to k. Defaults to 5*k (clamped to at most 200).

Returns:

Search results with column attribution

Return type:

List[QueryResult]

async query_multi_column_async(query: str, *, columns: List[str] | None = None, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'context', 'enriched'] = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]

Async query across multiple columns (main content + embedding-enabled metadata fields)

Parameters:

query (str) – Query text
columns (Optional[List[str]]) – Specific columns to search. If None, searches all embedding-enabled fields plus main content. Use ‘content’ for main document content.
search_type (Literal['vector', 'keyword', 'hybrid']) – Type of search to perform
return_type (Literal['documents', 'chunks']) – Whether to return full documents or individual chunks
k (int) – Maximum number of results to return
score_threshold (float) – Minimum score to keep (0-1, higher=better). For search_type="hybrid" each leg is min-max normalized within this query’s own candidate pool, so scores are not comparable across queries or across different k: the threshold cuts on rank position within the pool, not on absolute match quality, and is not a portable bar you can tune once and reuse.
filters (Optional[Dict[str, Any]]) – Metadata filters to apply. Filter fields must be declared in the metadata schema; unknown fields or unsupported operators raise DatabaseError.
vector_weight (float) – Weight for vector search in hybrid mode (0-1)
document_scoring_method (DocumentScoringMethod) – Method for aggregating chunk scores into document scores
document_scoring_options (dict, optional) – Parameters for the scoring method
reranker (object or False, optional) – A reranker instance applied ONCE to the merged multi-column pool (never per column leg, which would re-score legs inconsistently). False disables reranking for this call, including the database’s persisted default; None uses the default if set.
reranker_config (dict, optional) – Config from which a reranker is constructed for the merged pool; overrides the database’s persisted default for this call.
rerank_k (int, optional) – Merged-pool width handed to the reranker before truncating to k. Defaults to 5*k (clamped to at most 200).

Returns:

Search results with column attribution

Return type:

List[QueryResult]

query_stream(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, batch_size: int = 50) → Iterator[List[QueryResult]]

Stream query results in batches. Convenience wrapper around query_cursor().

Yields:: list of QueryResult – Each yield is a batch of results.

async query_stream_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, batch_size: int = 50) → AsyncIterator[List[QueryResult]]

Async stream query results in batches. Convenience wrapper around query_cursor_async().

Yields:: list of QueryResult – Each yield is a batch of results.

rebuild_hierarchical_embeddings() → None

Rebuild section and document FAISS indices from existing data.

This is useful when opening an existing database with hierarchical_embeddings=True for the first time, or to rebuild after data corruption.

repair(dry_run: bool = False) → RepairReport

Rebuild the FAISS indices from SQLite, reassigning every id.

Parameters:: dry_run – Report what is wrong without modifying anything.

save(): Save the database.

async save_async(): Saves the database asynchronously

property section_vector_strategy: str | None

How sections are represented – "centroid" or "rawspan".

None when the database is not hierarchical.

set_default_reranker(config: Dict[str, Any] | None, persist: bool = True) → None

Set (or clear, with None) the database’s default reranker.

Validates the provider name against the registry without constructing the reranker, invalidates the cached instance, and – when persist – writes only its own default_reranker config key (deleting it when clearing), following the set_sqlite_tuning pattern.

set_sqlite_tuning(profile: str, overrides: Dict[str, Any] | None = None, persist: bool = True) → None: Apply SQLite tuning profile to local database.

sqlite_checkpoint(mode: str = 'PASSIVE') → None: Run SQLite WAL checkpoint operation.

sqlite_incremental_vacuum(pages: int = 2000) → None: Run incremental VACUUM operation.

sqlite_optimize() → None: Run SQLite PRAGMA optimize.

sqlite_vacuum() → None: Run SQLite VACUUM operation.

update(doc_id: str, content: str | None = None, metadata: Dict[str, Any] | None = None) → bool

Update a document’s content and/or metadata

Parameters:

doc_id (str) – Document ID to update
content (Optional[str]) – New content (if None, content is not updated)
metadata (Optional[Dict[str, Any]]) – New metadata (merged with existing)

Returns:

True if document was updated, False if no updates needed (content and metadata already match database)

Return type:

bool

Raises:

DocumentNotFoundError – Raised if doc_id does not exist.

async update_async(doc_id: str, content: str | None = None, metadata: Dict[str, Any] | None = None) → bool

Update a document’s content and/or metadata asynchronously.

Parameters:

doc_id (str) – Document ID to update
content (Optional[str]) – New content (if None, content is not updated)
metadata (Optional[Dict[str, Any]]) – New metadata (merged with existing metadata)

Returns:

True if document was updated, False if no updates needed (content and metadata already match database)

Return type:

bool

Raises:

DocumentNotFoundError – Raised if doc_id does not exist.

Examples

Update content only:

updated = await db.update_async("doc1", content="New content")

Update metadata only:

updated = await db.update_async("doc1", metadata={"status": "reviewed"})

Update both content and metadata:

updated = await db.update_async(
    "doc1",
    content="Updated content",
    metadata={"last_modified": datetime.now()}
)

Notes

If content is updated, the document will be re-chunked and re-embedded
Metadata updates are merged with existing metadata (not replaced)
Content changes trigger full document reprocessing for consistency
Uses async database operations for better performance

update_metadata_schema(new_schema, drop_columns: bool = False, column_mapping: dict | None = None) → Dict[str, Any]

Update the metadata schema for the database

This method allows you to add new metadata fields, modify existing ones, or remove fields from the schema. Existing document data is preserved.

Parameters:

new_schema (Union[str, Dict[str, MetadataField]]) – The new metadata schema to apply. Can be: - str: Schema name from common schemas (e.g., ‘research_papers’) - Dict[str, MetadataField]: Complete field definitions - Dict[str, str]: Simple type-only definitions (e.g., {‘field’: ‘text’}) - Dict[str, tuple]: Tuple definitions (type, indexed) or (type, indexed, required)
drop_columns (bool, default=False) – Whether to actually drop columns that are no longer in the schema. If False, columns are kept but removed from schema for safety.
column_mapping (dict, optional) – Optionally provide a mapping dict with old-column (key) -> new-column (value)

Returns:

Summary of changes made including: - added_fields: List of newly added field names - removed_fields: List of removed field names - modified_fields: List of modified fields with change details - populated_defaults: List of fields where default values were populated - dropped_columns: List of actually dropped columns (if drop_columns=True) - warnings: List of warnings about potential issues - errors: List of any errors encountered

Return type:

Dict[str, Any]

Examples

Add new metadata fields:

new_schema = {
    'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
    'priority': MetadataField(type=MetadataFieldType.INTEGER, default_value=0),
    'tags': MetadataField(type=MetadataFieldType.JSON)
}

changes = db.update_metadata_schema(new_schema)
print(f"Added fields: {changes['added_fields']}")

Use shorthand syntax:

new_schema = {
    'category': 'text',  # Simple type
    'priority': ('integer', False, True),  # (type, indexed, required)
    'rating': ('real', True)  # (type, indexed)
}

changes = db.update_metadata_schema(new_schema)

Apply a common schema:

changes = db.update_metadata_schema('research_papers')

Notes

Field names cannot conflict with reserved columns: id, content, content_hash, created_at, updated_at
Removed fields are removed from the schema but columns are kept for data safety
Type changes are recorded but don’t modify existing data (SQLite limitation)
Index changes are applied immediately
Changes are applied in a transaction and rolled back on error

async update_metadata_schema_async(new_schema, drop_columns: bool = False, column_mapping: dict | None = None) → Dict[str, Any]

Update the metadata schema for the database asynchronously

This method allows you to add new metadata fields, modify existing ones, or remove fields from the schema. Existing document data is preserved.

Parameters:

new_schema (Union[str, Dict[str, MetadataField]]) – The new metadata schema to apply. Can be: - str: Schema name from common schemas (e.g., ‘research_papers’) - Dict[str, MetadataField]: Complete field definitions - Dict[str, str]: Simple type-only definitions (e.g., {‘field’: ‘text’}) - Dict[str, tuple]: Tuple definitions (type, indexed) or (type, indexed, required)
drop_columns (bool, default=False) – Whether to actually drop columns that are no longer in the schema. If False, columns are kept but removed from schema for safety.
column_mapping (dict, optional) – Optionally provide a mapping dict with old-column (key) -> new-column (value)

Returns:

Summary of changes made including: - added_fields: List of newly added field names - removed_fields: List of removed field names - modified_fields: List of modified fields with change details - populated_defaults: List of fields where default values were populated - dropped_columns: List of actually dropped columns (if drop_columns=True) - warnings: List of warnings about potential issues - errors: List of any errors encountered

Return type:

Dict[str, Any]

Examples

Add new metadata fields:

new_schema = {
    'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
    'priority': MetadataField(type=MetadataFieldType.INTEGER, default_value=0),
    'tags': MetadataField(type=MetadataFieldType.JSON)
}

changes = await db.update_metadata_schema_async(new_schema)
print(f"Added fields: {changes['added_fields']}")

Use shorthand syntax:

new_schema = {
    'category': 'text',  # Simple type
    'priority': ('integer', False, True),  # (type, indexed, required)
    'rating': ('real', True)  # (type, indexed)
}

changes = await db.update_metadata_schema_async(new_schema)

Apply a common schema:

changes = await db.update_metadata_schema_async('research_papers')

Notes

Field names cannot conflict with reserved columns: id, content, content_hash, created_at, updated_at
Removed fields are removed from the schema but columns are kept for data safety
Type changes are recorded but don’t modify existing data (SQLite limitation)
Index changes are applied immediately
Changes are applied in a transaction and rolled back on error

Insert or update documents in the database with pipeline processing

This enhanced version uses a 3-stage pipeline to overlap chunking, embedding generation, and database operations for 2-3x better throughput.

Parameters:

documents (Union[str, List[str]]) – Document text(s) to add
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Only fields declared in the metadata schema are stored; fields not in the schema are dropped with a logged warning.
ids (Optional[Union[str, List[str]]]) – Document IDs (auto-generated if not provided)
batch_size (int) – Batch size for processing, by default 100
similarity_threshold (Optional[float]) – Skip adding chunks that are more similar than this value
errors (Literal["ignore", "raise"]) – How to handle a document that fails to embed or write. "raise" (default) raises IngestError naming the failed IDs after the documents that succeeded are committed; "ignore" returns only the IDs that landed.

Returns:

List of document IDs that were upserted

Return type:

List[str]

async upsert_async(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2, errors: Literal['ignore', 'raise'] = 'raise', **kwargs: Any) → List[str]

Async upsert with pipeline processing for maximum throughput

Parameters:

documents (Union[str, List[str]]) – Document text(s) to add
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Only fields declared in the metadata schema are stored; fields not in the schema are dropped with a logged warning.
ids (Optional[Union[str, List[str]]]) – Document IDs (auto-generated if not provided)
batch_size (int) – Batch size for processing
similarity_threshold (Optional[float]) – Skip adding chunks that are more similar than this value
max_concurrent_chunks (int, default=3) – Maximum concurrent chunking operations
max_concurrent_embeddings (int, default=2) – Maximum concurrent embedding operations
**kwargs (Any) – Any other parameter accepted by upsert().

Returns:

List of document IDs that were upserted

Return type:

List[str]

upsert_from_chunks(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise') → List[str]

Insert or update documents from pre-chunked data with pipeline processing.

This method allows you to directly provide chunks for documents, bypassing the chunking step and enabling more efficient processing of pre-processed documents.

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks. Chunks can be either: - List[Chunk]: Full Chunk objects with position information - List[str]: Simple strings that will be converted to Chunk objects
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata. If None, empty metadata is used for all documents.
batch_size (int, default=100) – Number of embeddings to generate at once
similarity_threshold (Optional[float], default=None) – If provided, filters out chunks that are too similar to existing chunks

Returns:

List of document IDs that were processed

Return type:

List[str]

Raises:

ValueError – If chunk data is invalid or metadata doesn’t match schema

async upsert_from_chunks_async(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2, errors: Literal['ignore', 'raise'] = 'raise') → List[str]

Async version of upsert_from_chunks - Insert or update documents from pre-chunked data.

This method allows you to directly provide chunks for documents, bypassing the chunking step and enabling more efficient processing of pre-processed documents.

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks. Chunks can be either: - List[Chunk]: Full Chunk objects with position information - List[str]: Simple strings that will be converted to Chunk objects
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata. If None, empty metadata is used for all documents.
batch_size (int, default=None) – Number of embeddings to generate at once. If None, uses default from configuration.
similarity_threshold (Optional[float], default=None) – If provided, filters out chunks that are too similar to existing chunks
max_concurrent_chunks (int, default=3) – Maximum number of concurrent chunk processing operations
max_concurrent_embeddings (int, default=2) – Maximum number of concurrent embedding operations

Returns:

List of document IDs that were processed

Return type:

List[str]

Raises:

ValueError – If chunk data is invalid or metadata doesn’t match schema

Insert or update documents from files using file extraction.

Uses the ExtractorRegistry to automatically extract text from files based on file extension and MIME type, then calls the regular upsert method.

Parameters:

file_paths (Union[str, Path, List[Union[str, Path]]]) – Path(s) to files to extract and upsert
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Will be merged with extracted metadata.
ids (Optional[Union[str, List[str]]]) – Document IDs. If not provided, will use filename without extension.
batch_size (int) – Batch size for processing, by default 100
similarity_threshold (Optional[float]) – Skip adding chunks that are more similar than this value
extractor_kwargs (Optional[Dict[str, Any]]) – Additional keyword arguments passed to the extractor

Returns:

List of document IDs that were upserted

Return type:

List[str]

Raises:

FileNotFoundError – If any of the specified files don’t exist
ValueError – If extraction fails for any file and no fallback is available

Async insert or update documents from files using file extraction.

Uses the ExtractorRegistry to automatically extract text from files based on file extension and MIME type, then calls the regular upsert_async method.

Parameters:

file_paths (Union[str, Path, List[Union[str, Path]]]) – Path(s) to files to extract and upsert
metadata (Optional[Union[Dict[str, Any], List[Dict[str, Any]]]]) – Metadata for documents. Will be merged with extracted metadata.
ids (Optional[Union[str, List[str]]]) – Document IDs. If not provided, will use filename without extension.
batch_size (int) – Batch size for processing
similarity_threshold (Optional[float]) – Skip adding chunks that are more similar than this value
max_concurrent_chunks (int, default=3) – Maximum concurrent chunking operations
max_concurrent_embeddings (int, default=2) – Maximum concurrent embedding operations
extractor_kwargs (Optional[Dict[str, Any]]) – Additional keyword arguments passed to the extractor

Returns:

List of document IDs that were upserted

Return type:

List[str]

Raises:

FileNotFoundError – If any of the specified files don’t exist
ValueError – If extraction fails for any file and no fallback is available

visualize_chord(doc_id: str, similarity_threshold: float = 0.7, min_chunk_distance: int = 3, chunk_labels: bool = False, interactive: bool = False, **kwargs)

Chord (Circos-style) diagram for chunk self-similarity.

Parameters:

doc_id (str) – Document ID.
similarity_threshold (float) – Minimum similarity for a chord to be drawn.
min_chunk_distance (int) – Minimum index distance between chunks for a chord.
chunk_labels (bool) – Label each arc segment with its index.
interactive (bool) – Use plotly instead of matplotlib.
**kwargs – Forwarded to the plotting function – notably labels, which puts arbitrary text (the section each chunk falls in, say) on each arc instead of its index, in both the matplotlib and the interactive renderer.

Return type:

matplotlib.figure.Figure or plotly.graph_objects.Figure

visualize_documents(doc_ids: List[str] | None = None, method: str = 'tsne', color_by: str | None = None, n_clusters: int | None = None, interactive: bool = False, **kwargs)

Project document embeddings to 2-D and plot.

Parameters:

doc_ids (list of str, optional) – Documents to include. All documents if None.
method (str) – "tsne" or "pca".
color_by (str, optional) – Metadata field name used for point colouring.
n_clusters (int, optional) – If set, cluster embeddings and colour by cluster.
interactive (bool) – Use plotly for interactive plots instead of matplotlib.

Return type:

matplotlib.figure.Figure or plotly.graph_objects.Figure

visualize_queries(queries: List[str], doc_ids: List[str] | None = None, method: str = 'tsne', interactive: bool = False, **kwargs)

Visualise how queries relate to the document embedding space.

Parameters:

queries (list of str) – Query strings to overlay on the map.
doc_ids (list of str, optional) – Documents to include. All documents if None.
method (str) – Dimensionality reduction method.
interactive (bool) – Use plotly instead of matplotlib.

Return type:

matplotlib.figure.Figure or plotly.graph_objects.Figure

visualize_synteny(doc_id_1: str, doc_id_2: str, similarity_threshold: float = 0.7, orientation: str = 'horizontal', chunk_labels: bool = False, interactive: bool = False, **kwargs)

Synteny ribbon diagram comparing chunks of two documents.

Parameters:

doc_id_1 (str) – First document ID.
doc_id_2 (str) – Second document ID.
similarity_threshold (float) – Minimum similarity for a ribbon to be drawn.
orientation (str) – "horizontal" or "vertical".
chunk_labels (bool) – Label each chunk segment with its index.
interactive (bool) – Use plotly instead of matplotlib.
**kwargs – Forwarded to the plotting function – notably labels_1 and labels_2, which put arbitrary text (a section heading, say) on each chunk segment instead of its index, in both the matplotlib and the interactive renderer.

Return type:

matplotlib.figure.Figure or plotly.graph_objects.Figure

connection_pool

async_connection_pool

index

schema

chunker

pipeline_worker_timeout

section_index

document_index

name

class localvectordb.database.BaseVectorDB

Bases: ABC

Abstract base class defining the interface for vector databases.

This class defines the common interface that both LocalVectorDB and RemoteVectorDB must implement, allowing QueryBuilder and other components to work with either implementation seamlessly.

abstractmethod upsert(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None) → List[str]: Insert or update documents in the database.

abstractmethod insert(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise') → List[str]: Insert new documents into the database.

abstractmethod get(ids: str | List[str]) → Document | List[Document]: Retrieve documents by ID.

abstractmethod exists(ids: str | List[str]) → bool | List[bool]: Check if documents exist.

abstractmethod count(filters: Dict[str, Any] | None = None) → int: Count the number of documents in the database

abstractmethod delete(ids: str | List[str]) → int: Delete documents.

abstractmethod update(doc_id: str, content: str | None = None, metadata: Dict[str, Any] | None = None) → bool

Update a document’s content and/or metadata.

Returns True if the document was updated, False if no update was needed (content and metadata already match what is stored). Raises DocumentNotFoundError if doc_id does not exist – “no-op” and “not found” are distinct outcomes and must not be collapsed into one another.

abstractmethod patch(doc_id: str, ops: List[Dict[str, Any]], *, expect_hash: str | None = None, metadata: Dict[str, Any] | None = None) → PatchResult

Patch a document’s content with find/replace or span-splice ops.

ops resolve against the document’s current content (character offsets) and are applied atomically; metadata is merged as in update(). This shares update()’s three-outcome contract and adds a fourth:

Missing doc_id -> raises DocumentNotFoundError.
Ops produce content identical to what is stored (and no metadata delta) -> PatchResult.updated is False.
expect_hash given and it does not match the stored content_hash -> raises PatchConflictError (never collapsed into either of the above).
Unmatched/ambiguous find or overlapping/out-of-range ops -> raises PatchError.

abstractmethod query(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]: Unified query interface for all search types.

query_cursor(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: DocumentScoringMethod = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Create a QueryCursor for streaming results with lazy hydration.

Intentionally concrete (not @abstractmethod): cursor support is optional, so backends that do not provide it (e.g. RemoteVectorDB) can inherit this default, which raises if a caller actually invokes it.

async query_cursor_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: DocumentScoringMethod = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, batch_size: int = 50, cursor_ttl: float = 300.0) → QueryCursor

Create a QueryCursor for async streaming results with lazy hydration.

Intentionally concrete (not @abstractmethod): cursor support is optional, so backends that do not provide it (e.g. RemoteVectorDB) can inherit this default, which raises if a caller actually invokes it.

abstractmethod filter(where: Dict[str, Any] | None = None, order_by: str | None = None, limit: int | None = None, offset: int = 0) → List[Document]: Filter documents using metadata filtering.

abstract property embedding_model: str: Return the embedding model name.

abstract property embedding_provider: EmbeddingProvider: Return the embedding provider name or instance.

abstract property embedding_dimension: int: Return the dimension of the embeddings.

abstract property chunk_size: int: Return the maximum tokens per chunk.

abstract property chunk_overlap: int: Return the chunk overlap, in the unit of chunking_method (not tokens unless the method is "tokens").

abstract property chunking_method: str: Return the chunking method.

abstract property fts_enabled: bool: Return whether full-text search is enabled.

abstract property metadata_schema: Dict[str, MetadataField]: Return the metadata schema.

abstractmethod get_stats() → Dict[str, Any]: Get database statistics.

abstractmethod async get_stats_async() → Dict[str, Any]: Get database statistics asynchronously (async twin of get_stats).

abstract property closed: bool: Return whether the database connection is closed.

abstractmethod update_metadata_schema(new_schema: str | Dict[str, MetadataField], drop_columns: bool = False, column_mapping: dict | None = None) → Dict[str, Any]: Update the metadata schema.

abstractmethod get_metadata_schema_info() → Dict[str, Any]: Get detailed information about the current metadata schema.

abstractmethod save() → None: Save the database.

abstractmethod close() → None: Close the database.

query_builder() → QueryBuilderInterface

Create a new QueryBuilder for this database.

Returns:: A new QueryBuilder instance for building complex queries
Return type:: QueryBuilder

Examples

Basic search:

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

Complex multi-field search with semantic filtering:

results = (db.query_builder()
    .search_field("title", "neural networks", weight=0.3)
    .search_field("content", "deep learning", weight=0.7)
    .semantic_filter("methodology", "supervised learning", threshold=0.8)
    .filter("year", gte=2020)
    .hybrid(vector_weight=0.6)
    .limit(20)
    .execute())

Async usage:

results = await (db.query_builder()
    .search("machine learning")
    .semantic_filter("category", "research")
    .execute_async())

ping() → bool: Check if the database is accessible. Override in subclasses.

abstractmethod upsert_from_chunks(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None) → List[str]

abstractmethod insert_from_chunks(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise') → List[str]

abstractmethod upsert_from_file(file_paths: str | Path | List[str | Path], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, extractor_kwargs: Dict[str, Any] | None = None) → List[str]

abstractmethod insert_from_file(file_paths: str | Path | List[str | Path], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', extractor_kwargs: Dict[str, Any] | None = None) → List[str]

abstractmethod async upsert_async(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, **kwargs: Any) → List[str]: Insert or update documents in the database asynchronously.

abstractmethod async insert_async(documents: str | List[str], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', **kwargs: Any) → List[str]: Insert new documents into the database asynchronously.

abstractmethod async upsert_from_chunks_async(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2) → List[str]

abstractmethod async insert_from_chunks_async(chunks_by_document: Dict[str, List[Chunk] | List[str]], metadata: Dict[str, Dict[str, Any]] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2) → List[str]

abstractmethod async upsert_from_file_async(file_paths: str | Path | List[str | Path], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2, extractor_kwargs: Dict[str, Any] | None = None) → List[str]

abstractmethod async insert_from_file_async(file_paths: str | Path | List[str | Path], metadata: Dict[str, Any] | List[Dict[str, Any]] | None = None, ids: str | List[str] | None = None, batch_size: int | None = None, similarity_threshold: float | None = None, errors: Literal['ignore', 'raise'] = 'raise', max_concurrent_chunks: int = 3, max_concurrent_embeddings: int = 2, extractor_kwargs: Dict[str, Any] | None = None) → List[str]

abstractmethod async get_async(ids: str | List[str]) → Document | List[Document]: Retrieve documents by ID asynchronously.

abstractmethod async exists_async(ids: str | List[str]) → bool | List[bool]: Check if documents exist asynchronously.

abstractmethod async delete_async(ids: str | List[str]) → int: Delete documents asynchronously.

abstractmethod async count_async(filters: Dict[str, Any] | None = None) → int

abstractmethod async update_async(doc_id: str, content: str | None = None, metadata: Dict[str, Any] | None = None) → bool

Update a document’s content and/or metadata asynchronously.

Same contract as update(): False means “no update needed”, and a missing document raises DocumentNotFoundError.

abstractmethod async patch_async(doc_id: str, ops: List[Dict[str, Any]], *, expect_hash: str | None = None, metadata: Dict[str, Any] | None = None) → PatchResult: Patch a document’s content asynchronously. Same contract as patch().

abstractmethod async query_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['auto', 'best', 'average', 'frequency_boost', 'percentile'] = 'auto', document_scoring_options: dict | None = None, reranker: Any | None = None, reranker_config: Dict[str, Any] | None = None, rerank_k: int | None = None) → List[QueryResult]: Unified query interface for all search types asynchronously.

abstractmethod async filter_async(where: Dict[str, Any] | None = None, order_by: str | None = None, limit: int | None = None, offset: int = 0) → List[Document]: Filter documents using metadata filtering asynchronously.

abstractmethod async save_async() → None: Save the database asynchronously.

abstractmethod async close_async() → None: Close the database asynchronously.

abstractmethod async update_metadata_schema_async(new_schema: str | Dict[str, MetadataField], drop_columns: bool = False, column_mapping: Dict[str, str] | None = None) → Dict[str, Any]: Update metadata schema asynchronously.

abstractmethod async get_metadata_schema_info_async() → Dict[str, Any]: Get metadata schema information asynchronously.

abstractmethod get_chunk_embeddings(chunk_ids: str | List[str]) → ndarray: Return the raw embedding vectors for one or more chunk IDs.

abstractmethod compare_documents(doc_id_1: str, doc_id_2: str) → float: Return the [0, 1] similarity between two documents.

abstractmethod async compare_documents_async(doc_id_1: str, doc_id_2: str) → float: Async twin of compare_documents().

abstractmethod compare_documents_detailed(doc_id_1: str, doc_id_2: str, chunk_threshold: float = 0.7) → DocumentComparisonResult: Return a rich chunk-level comparison between two documents.

abstractmethod async compare_documents_detailed_async(doc_id_1: str, doc_id_2: str, chunk_threshold: float = 0.7) → DocumentComparisonResult: Async twin of compare_documents_detailed().

abstractmethod nearest_neighbors(doc_id: str, k: int = 5, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None) → List[QueryResult]: Return the k documents most similar to doc_id.

abstractmethod async nearest_neighbors_async(doc_id: str, k: int = 5, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None) → List[QueryResult]: Async twin of nearest_neighbors().

abstractmethod pairwise_similarity_matrix(doc_ids: List[str] | None = None) → DocumentSimilarityMatrix: Return an NxN document similarity matrix.

abstractmethod async pairwise_similarity_matrix_async(doc_ids: List[str] | None = None) → DocumentSimilarityMatrix: Async twin of pairwise_similarity_matrix().

get_chunks(document_id: str, indices: List[int] | None = None) → List[Chunk]

Return the stored chunks for a document.

Intentionally concrete (not @abstractmethod): chunk retrieval is a local-only capability. RemoteVectorDB inherits this raising default so a local↔remote swap fails loudly here rather than with AttributeError.

__init__()

class localvectordb.database.TuningMixin

Bases: ABC

Mixin class providing SQLite tuning interface for vector databases.

This mixin defines the common interface for SQLite performance tuning that is implemented by both LocalVectorDB and RemoteVectorDB classes.

abstractmethod get_sqlite_tuning() → Dict[str, Any]

Get current SQLite tuning configuration.

Returns:: Current tuning configuration containing: - profile: Current profile name - pragmas: Current pragma settings - overrides: Profile overrides
Return type:: Dict[str, Any]

abstractmethod set_sqlite_tuning(profile: str, overrides: Dict[str, Any] | None = None, persist: bool = True) → None

Apply SQLite tuning profile with optional overrides.

Parameters:

profile (str) – Name of the tuning profile to apply
overrides (Dict[str, Any], optional) – Pragma overrides for the profile
persist (bool, optional) – Whether to persist settings to database config, by default True

Raises:

ValueError – If profile name is not recognized

abstractmethod sqlite_checkpoint(mode: str = 'PASSIVE') → None

Run SQLite WAL checkpoint operation.

Parameters:: mode (str, optional) – Checkpoint mode (PASSIVE, FULL, RESTART, TRUNCATE), by default “PASSIVE”

abstractmethod sqlite_optimize() → None: Run SQLite PRAGMA optimize to update query planner statistics.

abstractmethod sqlite_vacuum() → None: Run SQLite VACUUM operation.

Warning

This operation requires exclusive database access and may take significant time.

abstractmethod sqlite_incremental_vacuum(pages: int = 2000) → None

Run incremental VACUUM operation.

Parameters:: pages (int, optional) – Number of pages to reclaim, by default 2000

list_sqlite_profiles() → Dict[str, str]

List available SQLite tuning profiles.

Returns:: Dictionary mapping profile names to descriptions
Return type:: Dict[str, str]

analyze_system_resources() → Dict[str, Any]

Analyze system resources for tuning recommendations.

Returns:: System resource information
Return type:: Dict[str, Any]

auto_tune(workload: Dict[str, Any] | None = None, interactive: bool = False, apply: bool = False) → Dict[str, Any]

Get auto-tuning recommendations based on system and workload.

Parameters:

workload (Dict[str, Any], optional) – Workload characteristics. If None and interactive=True, will prompt user.
interactive (bool, optional) – Whether to run interactive interview for workload, by default False
apply (bool, optional) – Whether to apply the recommended settings, by default False

Returns:

Tuning recommendation containing: - profile_name: Recommended profile - pragma_overrides: Recommended pragma overrides - reasoning: List of reasoning explanations - estimated_memory_mb: Estimated memory usage

Return type:

Dict[str, Any]

checkpoint_if_wal_large(wal_mb_threshold: int = 128) → bool

Check if WAL file is large and checkpoint if needed.

Parameters:: wal_mb_threshold (int, optional) – WAL size threshold in MB, by default 128
Returns:: True if checkpoint was performed, False otherwise
Return type:: bool

__init__()

class localvectordb.database.RepairReport(dry_run: bool = False, duplicate_ids: List[int] = <factory>, orphan_vectors: List[int] = <factory>, dangling_rows: List[int] = <factory>, reconstructed: int = 0, reembedded: int = 0, dropped: int = 0, base_index_type: str = '', sections_rebuilt: int = 0, documents_rebuilt: int = 0)

Bases: object

What repair found, and what it did (or would do, under dry_run).

dry_run: bool = False

duplicate_ids: List[int]

orphan_vectors: List[int]

dangling_rows: List[int]

reconstructed: int = 0

reembedded: int = 0

dropped: int = 0

base_index_type: str = ''

sections_rebuilt: int = 0

documents_rebuilt: int = 0

property healthy: bool

property summary: str

__init__(dry_run: bool = False, duplicate_ids: List[int] = <factory>, orphan_vectors: List[int] = <factory>, dangling_rows: List[int] = <factory>, reconstructed: int = 0, reembedded: int = 0, dropped: int = 0, base_index_type: str = '', sections_rebuilt: int = 0, documents_rebuilt: int = 0) → None

class localvectordb.database.DiagnoseReport(database: str = '', documents: int = 0, chunks: int = 0, sections: int = 0, embedding_provider: str = '', embedding_model: str = '', schema_version: int | None = None, expected_schema_version: int = 0, created_by_version: str | None = None, context_tokens: int | None = None, tokens_exact: bool = False, token_source: str = '', chunk_coverage: float | None = None, truncated_chunk_share: float | None = None, chunks_measured: int = 0, median_chunk_tokens: int | None = None, sections_over_context_share: float | None = None, median_section_tokens: int | None = None, sections_measured: int = 0, chunkless_section_share: float | None = None, mean_chunks_per_document: float | None = None, median_chunks_per_document: int | None = None, mean_chunks_per_section: float | None = None, fts_status: Dict[str, str]=<factory>, warnings: List[str] = <factory>, notes: List[str] = <factory>)

Bases: object

What diagnose() measured, and which regime it puts the corpus in.

database: str = ''

documents: int = 0

chunks: int = 0

sections: int = 0

embedding_provider: str = ''

embedding_model: str = ''

schema_version: int | None = None: On-disk table-layout version (config.schema_version) against the version this package expects; they match after any successful open.

expected_schema_version: int = 0

created_by_version: str | None = None: localvectordb version that created the file, when recorded.

context_tokens: int | None = None: The encoder’s context window in tokens, or None when no provider reports one.

tokens_exact: bool = False: True when token counts came from the encoder’s own tokenizer.

token_source: str = ''

chunk_coverage: float | None = None: sum(min(tokens, context)) / sum(tokens) over the measured chunks; None when the context is unknown (there is no cap to measure against).

truncated_chunk_share: float | None = None

chunks_measured: int = 0

median_chunk_tokens: int | None = None

sections_over_context_share: float | None = None: Share of sections whose text exceeds one encoder window. These are not truncated – rawspan vectors are windowed and mean-pooled – but pooled vectors degrade as the span grows, which is a regime, not a defect.

median_section_tokens: int | None = None

sections_measured: int = 0

chunkless_section_share: float | None = None: Share of sections unreachable by chunk->section roll-up (return_type=”sections”): sections with neither a chunk_sections overlap row nor a midpoint owner. ~0 on a healthy database, since chunks credit every section they overlap; 1 - share is a hard recall ceiling.

mean_chunks_per_document: float | None = None

median_chunks_per_document: int | None = None

mean_chunks_per_section: float | None = None

fts_status: Dict[str, str]

“ok”, “missing”, “disabled”, or “stale (n vs m)”.

Type:: Per FTS table

warnings: List[str]

notes: List[str]

property healthy: bool

property summary: str

__init__(database: str = '', documents: int = 0, chunks: int = 0, sections: int = 0, embedding_provider: str = '', embedding_model: str = '', schema_version: int | None = None, expected_schema_version: int = 0, created_by_version: str | None = None, context_tokens: int | None = None, tokens_exact: bool = False, token_source: str = '', chunk_coverage: float | None = None, truncated_chunk_share: float | None = None, chunks_measured: int = 0, median_chunk_tokens: int | None = None, sections_over_context_share: float | None = None, median_section_tokens: int | None = None, sections_measured: int = 0, chunkless_section_share: float | None = None, mean_chunks_per_document: float | None = None, median_chunks_per_document: int | None = None, mean_chunks_per_section: float | None = None, fts_status: Dict[str, str]=<factory>, warnings: List[str] = <factory>, notes: List[str] = <factory>) → None

localvectordb.database.open_for_repair(name: str, base_path: str, **kwargs: Any) → Any

Open a database bypassing the on-open integrity check.

_verify_integrity raises on precisely the databases repair exists to fix, so the normal constructor (and get_ctx_db) cannot reach them.

The constructor validates its default embedding provider before it loads the database’s saved config, which would make repairing a database require whichever provider happens to be the default to be reachable. Read the saved provider from SQLite first so repair works offline whenever it does not need to re-embed.

Submodules

localvectordb.database.base module
- BaseVectorDB
- LocalVectorDBBase