localvectordb package

Everything below is re-exported at the top level: import it as from localvectordb import LocalVectorDB, not from the submodule it happens to be defined in.

This page renders the whole public surface for browsing, but it deliberately does not own the cross-reference targets – each symbol is documented canonically once, on the page for the module that defines it (linked under Submodules below). Documenting it in both places is what produced Sphinx’s “more than one target found” warnings, since autodoc emits unqualified references from type annotations and cannot choose between two equal targets. Hence :no-index:; removing it re-introduces ~75 warnings.

LocalVectorDB - Document-First Vector Database with SQLite + FAISS

A Python library providing a document-focused vector database built on SQLite, FAISS, and pluggable embedding providers. Offers both local and remote (client-server) usage with a unified API.

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

Bases: LocalTuningMixin, PipelineMixin, SearchMixin, MetadataMixin, CrudMixin, ComparisonMixin, RepairMixin, 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 “nomic-embed-text”
embedding_config (Dict[str, Any], optional) – Configuration for embedding provider
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.

class localvectordb.ChunkerFactory

Bases: object

Factory for creating chunkers

CHUNKERS: dict[str, Type[PositionTrackingChunker]] = {'characters': <class 'localvectordb.chunking.CharChunker'>, 'code-blocks': <class 'localvectordb.chunking.CodeBlockChunker'>, 'lines': <class 'localvectordb.chunking.LineChunker'>, 'paragraphs': <class 'localvectordb.chunking.ParagraphChunker'>, 'sections': <class 'localvectordb.chunking.SectionChunker'>, 'sentences': <class 'localvectordb.chunking.SentenceChunker'>, 'tokens': <class 'localvectordb.chunking.TokenChunker'>, 'words': <class 'localvectordb.chunking.WordChunker'>}

classmethod create_chunker(method: str | Type[PositionTrackingChunker], max_tokens: int = 500, overlap: int = 0, **kwargs) → PositionTrackingChunker: Create a chunker instance

classmethod list_methods() → List[str]: List available chunking methods

class localvectordb.EmbeddingRegistry

Bases: object

Registry for embedding providers with plugin discovery

classmethod register(name: str, provider_class: Type[EmbeddingProvider]) → None: Register a new embedding provider

classmethod get(name: str) → Type[EmbeddingProvider]: Get an embedding provider by name

classmethod create_provider(provider_name: str, model: str, **kwargs: Any) → EmbeddingProvider: Create an embedding provider instance

classmethod list() → List[str]: List all registered providers

classmethod refresh_plugins() → None: Force re-discovery of plugins (useful for testing)

class localvectordb.RerankerRegistry

Bases: object

Registry for reranker providers with plugin discovery.

classmethod register(name: str, provider_class: Type[Reranker]) → None: Register a new reranker provider.

classmethod get(name: str) → Type[Reranker]: Get a reranker provider by name.

classmethod create_reranker(provider_name: str, model: str | None = None, **kwargs: Any) → Reranker: Create a reranker instance.

classmethod list() → List[str]: List all registered reranker providers.

classmethod refresh_plugins() → None: Force re-discovery of plugins (useful for testing).

class localvectordb.RemoteVectorDB(name: str, base_url: str = 'http://127.0.0.1:5000', api_key: str | None = None, *, create_if_not_exists: bool = True, metadata_schema: Dict[str, MetadataField] | None = None, embedding_provider: str = 'ollama', embedding_model: str = 'nomic-embed-text', embedding_config: Dict[str, Any] | None = None, chunking_method: str = 'sentences', chunk_size: int = 500, chunk_overlap: int = 1, enable_gpu: bool = False, enable_fts: bool = True, sqlite_profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'] = 'balanced', sqlite_pragma_overrides: Dict[str, Any] | None = None, request_timeout: int | None = None, authorization_header: str = 'Authorization', max_retries: int = 3, retry_delay: float = 1.0, max_concurrent_requests: int = 5, connection_pool_limits: Limits | None = None)

Bases: TuningMixin, BaseVectorDB

Client for interacting with a LocalVectorDB server.

This client provides the same interface as LocalVectorDB but connects to a remote server via HTTP.

Parameters:

name (str) – Name of the database
base_url (str) – URL of the LocalVectorDB server (e.g., “http://localhost:5000”)
api_key (str, optional) – API key for authentication. If not provided, checks LVDB_API_KEY environment variable. Specify a custom environment variable by passing “$CUSTOM_ENV_VARIABLE” for this parameter.
create_if_not_exists (bool, default=True) – Whether to create the database if it doesn’t exist
metadata_schema (Dict[str, MetadataField], optional) – Schema definition for metadata fields
embedding_provider (str, optional) – Provider for embeddings, by default “ollama”
embedding_model (str, optional) – Model to use for embeddings, by default “nomic-embed-text”
embedding_config (Dict[str, Any], optional) – Configuration for embedding provider
chunking_method (str, optional) – Method for chunking documents, 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, words, lines, characters, paragraphs), NOT tokens — only chunking_method="tokens" shares its unit with chunk_size. Keep it small (e.g. 1-3).
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
request_timeout (int, optional) – Timeout for HTTP requests
authorization_header (str, default = "Authorization") – The server can be configured to accept alternate headers, and the client can too.
max_retries (int) – Max number of retries for a request
retry_delay (int) – The delay after a failed request before trying again
max_concurrent_requests (int) – How many maximum requests for embeddings
connection_pool_limits (httpx.Limits) – Parameters for the httpx client connection pool

__init__(name: str, base_url: str = 'http://127.0.0.1:5000', api_key: str | None = None, *, create_if_not_exists: bool = True, metadata_schema: Dict[str, MetadataField] | None = None, embedding_provider: str = 'ollama', embedding_model: str = 'nomic-embed-text', embedding_config: Dict[str, Any] | None = None, chunking_method: str = 'sentences', chunk_size: int = 500, chunk_overlap: int = 1, enable_gpu: bool = False, enable_fts: bool = True, sqlite_profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'] = 'balanced', sqlite_pragma_overrides: Dict[str, Any] | None = None, request_timeout: int | None = None, authorization_header: str = 'Authorization', max_retries: int = 3, retry_delay: float = 1.0, max_concurrent_requests: int = 5, connection_pool_limits: Limits | None = None)

property embedding_provider: EmbeddingProvider

Return the remote embedding provider instance.

This provider proxies embedding requests to the server, providing API parity with LocalVectorDB while keeping operations server-side.

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

property embedding_model: str: Return the embedding model name

property embedding_dimension: int: Return the dimension of the embeddings

property chunk_size: int: Return the maximum tokens per chunk

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

property chunking_method: str: Return the chunking method

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

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

Insert or update documents in the database

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 database’s metadata schema are stored; other fields are dropped (the server logs a 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 (float, optional) – Skip adding any chunks that are more similar than this value. Good for “pre-deduplication”

Returns:

List of document IDs that were upserted

Return type:

List[str]

Insert new documents into the database

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 database’s metadata schema are stored; other fields are dropped (the server logs a 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

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

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

Count documents matching filters.

Parameters:: filters (Dict[str, Any], optional) – Metadata filters to apply
Returns:: Number of matching documents
Return type:: int

Insert or update documents from files using file extraction.

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 chunks that are too similar to existing chunks
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

Insert new documents from files using file extraction.

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
DuplicateDocumentIDError – If errors=”raise” and document ID conflicts occur

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]

Upsert documents from pre-chunked data.

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
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 upserted

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.

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata
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

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

DuplicateDocumentIDError – If a document ID already exists and errors=”raise”

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

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

Get embeddings for existing chunks in the database

Parameters:: chunk_ids (str | List[str]) – The chunks for which to retrieve the embeddings
Return type:: numpy.ndarray

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]]

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

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 the database)

Return type:

bool

Raises:

DocumentNotFoundError – Raised if doc_id does not exist.

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.

See localvectordb.LocalVectorDB.patch for the contract and op shapes. Raises DocumentNotFoundError (missing), PatchConflictError (stale expect_hash), or PatchError (unmatched/ambiguous/overlapping op).

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: str = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['best', 'average', 'frequency_boost'] = 'frequency_boost', 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 (Literal['documents', 'chunks', 'context', 'enriched']) – Whether to return full documents, individual chunks, or chunks with context
k (int) – Maximum number of results to return
score_threshold (float) – Minimum score threshold (0-1, higher=better)
filters (Optional[Dict[str, Any]]) – Metadata filters. Filter fields must be declared in the database’s metadata schema; unknown fields or unsupported operators are rejected by the server with an error.
vector_weight (float) – Weight for vector search in hybrid mode (0-1)
context_window (int) – Number of chunks before and after to include when return_type=’context’
semantic_dedup_threshold (Optional[float]) – Similarity threshold for semantic deduplication (0-1, higher=more similar)
document_scoring_method (str) – Method for aggregating chunk scores into document scores
document_scoring_options (dict, optional) – Parameters controlling the various scoring methods

Returns:

Search results with normalized scores

Return type:

List[QueryResult]

query_multi_column(query: str, *, columns: List[str] | None = None, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks'] = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, document_scoring_method: Literal['best', 'average', 'frequency_boost'] = 'frequency_boost', document_scoring_options: dict | 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 threshold (0-1, higher=better)
filters (Optional[Dict[str, Any]]) – Metadata filters to apply
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

Returns:

Search results with column attribution

Return type:

List[QueryResult]

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 now supports advanced MongoDB-style filtering with operators like $gt, $lt, $contains, $exists, etc. Raw SQL support has been removed for security reasons.

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}
})

Ordering and pagination:

# Order by field
docs = db.filter(
    where={"category": "tech"},
    order_by="created_at DESC",
    limit=10,
    offset=20
)

Notes

All queries are converted to safe parameterized SQL on the server
Field names are validated against the metadata schema
Raw SQL is no longer supported to prevent injection attacks
JSON fields support special operations like $contains

query_builder() → QueryBuilderInterface

Create a new RemoteQueryBuilder for this database.

The RemoteQueryBuilder provides a fluent API for building complex queries that are executed server-side, including semantic filters. It is typed as the shared QueryBuilderInterface, so the same fluent chain is portable to a LocalVectorDB.

Returns:: A new query builder instance (a RemoteQueryBuilder)
Return type:: QueryBuilderInterface

Examples

Basic query with semantic filter:

results = (db.query_builder()
    .search("machine learning")
    .filter("year", gte_=2020)
    .semantic_filter("methodology", "neural networks", threshold=0.8)
    .limit(10)
    .execute())

Complex multi-criteria query:

results = (db.query_builder()
    .search("quantum computing", search_type="hybrid")
    .filter("publication_type", "paper")
    .filter("citations", gte_=50)
    .semantic_filter("approach", "quantum machine learning", threshold=0.75)
    .order_by("publication_date", "desc")
    .limit(25)
    .execute())

save() → None: Save the database (no-op for remote client)

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

Update the metadata schema for the remote 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, Any]]) – 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) - Dict[str, dict]: Full field configuration objects
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': {'type': 'text', 'indexed': True, 'required': True, 'default_value': 'general'},
    'rating': {'type': 'real', 'default_value': 0.0},
    'tags': {'type': 'json', 'default_value': []}
}

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

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

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]

Examples

Get schema information:

schema_info = db.get_metadata_schema_info()
print(f"Total fields: {schema_info['field_count']}")
print(f"Indexed fields: {schema_info['indexed_fields']}")
print(f"Required fields: {schema_info['required_fields']}")
print(f"Field types: {schema_info['field_types']}")

# Detailed field information
for field_name, field_info in schema_info['fields'].items():
    print(f"{field_name}: {field_info['type']} "
          f"(indexed={field_info['indexed']}, required={field_info['required']})")

property healthy: bool

Check if the remote database server is healthy and accessible.

This property performs a lightweight ping to the server to verify connectivity and server responsiveness. Results are cached for 60 seconds to avoid excessive network requests.

Returns:: True if server is healthy and accessible, False otherwise
Return type:: bool

property closed: bool

Check if the connection to the remote database is closed.

Note: This is a legacy property name. For checking server health, use the ‘healthy’ property instead.

Returns:: False (remote connections don’t have a traditional “closed” state)
Return type:: bool

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

close() → None: Close the database connection and HTTP clients

classmethod database_exists(db_name: str, base_url: str = 'http://127.0.0.1:5000', api_key: str | None = None, authorization_header: str = 'Authorization') → bool: Check if a database exists on the server

async get_stats_async() → Dict[str, Any]: Get database statistics

Insert or update documents in the database

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 database’s metadata schema are stored; other fields are dropped (the server logs a 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 (float, optional) – If provided, skip chunks which have semantic similarity greater than this value (pre-deduplication)

Returns:

List of document IDs that were upserted

Return type:

List[str]

Insert new documents into the database

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 database’s metadata schema are stored; other fields are dropped (the server logs a 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

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Insert or update documents from files using file extraction (async).

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 chunks that are too similar to existing chunks
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

Insert new documents from files using file extraction (async).

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”
max_concurrent_chunks (int) – Maximum concurrent chunk processing (unused for remote)
max_concurrent_embeddings (int) – Maximum concurrent embedding requests (unused for remote)
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
DuplicateDocumentIDError – If errors=”raise” and document ID conflicts occur

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]

Upsert documents from pre-chunked data (async).

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
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
max_concurrent_chunks (int) – Maximum concurrent chunk processing (unused for remote)
max_concurrent_embeddings (int) – Maximum concurrent embedding requests (unused for remote)

Returns:

List of document IDs that were upserted

Return type:

List[str]

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]

Insert documents from pre-chunked data with conflict handling (async).

Parameters:

chunks_by_document (Dict[str, Union[List[Chunk], List[str]]]) – Dictionary mapping document IDs to their chunks
metadata (Optional[Dict[str, Dict[str, Any]]], default=None) – Dictionary mapping document IDs to their metadata
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
max_concurrent_chunks (int) – Maximum concurrent chunk processing (unused for remote)
max_concurrent_embeddings (int) – Maximum concurrent embedding requests (unused for remote)

Returns:

List of document IDs that were actually inserted

Return type:

List[str]

Raises:

DuplicateDocumentIDError – If a document ID already exists and errors=”raise”

async get_async(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]]

async exists_async(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 delete_async(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 count_async(filters: Dict[str, Any] | None = None) → int

Count documents matching filters asynchronously.

Parameters:: filters (Dict[str, Any], optional) – Metadata filters to apply
Returns:: Number of matching documents
Return type:: int

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

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 the database)

Return type:

bool

Raises:

DocumentNotFoundError – Raised if doc_id does not exist.

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().

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: str = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['best', 'average', 'frequency_boost'] = 'frequency_boost', 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 (Literal['documents', 'chunks']) – Whether to return full documents or individual chunks
k (int) – Maximum number of results to return
score_threshold (float) – Minimum score threshold (0-1, higher=better)
filters (Optional[Dict[str, Any]]) – Metadata filters. Filter fields must be declared in the database’s metadata schema; unknown fields or unsupported operators are rejected by the server with an error.
vector_weight (float) – Weight for vector search in hybrid mode (0-1)
context_window (int) – Number of chunks before and after to include when return_type=’context’
semantic_dedup_threshold (Optional[float]) – Similarity threshold for semantic deduplication (0-1, higher=more similar)
document_scoring_method (str) – Method for aggregating chunk scores into document scores
document_scoring_options (dict) – Optional parameters specific to each scoring method

Returns:

Search results with normalized scores

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'] = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, document_scoring_method: Literal['best', 'average', 'frequency_boost'] = 'frequency_boost', document_scoring_options: dict | 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 threshold (0-1, higher=better)
filters (Optional[Dict[str, Any]]) – Metadata filters to apply
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

Returns:

Search results with column attribution

Return type:

List[QueryResult]

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 enhanced metadata filtering

Parameters:

where (Optional[Dict[str, Any]]) – Filter conditions using either simple format or MongoDB-style operators
order_by (Optional[str]) – ORDER BY clause (field name with optional ASC/DESC)
limit (Optional[int]) – Maximum number of results
offset (int) – Number of results to skip

Returns:

Filtered documents

Return type:

List[Document]

async save_async() → None: No-op for remote databases

async close_async() → None: Close HTTP clients

async update_metadata_schema_async(new_schema: str | Dict[str, Any], 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, Any]]) – The new metadata schema to apply. Can be: - str: Schema name from common schemas (e.g., ‘research_papers’) - Dict with field definitions
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': {
        'type': 'text',
        'indexed': True
    },
    'priority': {
        'type': 'integer',
        'default_value': 0
    }
}

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

Apply a common schema:

changes = await db.update_metadata_schema_async('research_papers')

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]

Examples

Get schema information:

schema_info = await db.get_metadata_schema_info_async()
print(f"Total fields: {schema_info['field_count']}")
print(f"Indexed fields: {schema_info['indexed_fields']}")
print(f"Required fields: {schema_info['required_fields']}")
print(f"Field types: {schema_info['field_types']}")

# Detailed field information
for field_name, field_info in schema_info['fields'].items():
    print(f"{field_name}: {field_info['type']} "
          f"(indexed={field_info['indexed']}, required={field_info['required']})")

get_sqlite_tuning() → Dict[str, Any]: Get current SQLite tuning configuration from remote server.

set_sqlite_tuning(profile: str, overrides: Dict[str, Any] | None = None, persist: bool = True) → None: Apply SQLite tuning profile via remote server.

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

Get auto-tuning recommendations from the remote server.

The recommendation is computed on the server (from the server’s own system resources), not the client machine. interactive is not supported over HTTP – pass an explicit workload dict instead.

sqlite_checkpoint(mode: str = 'PASSIVE') → None: Run SQLite WAL checkpoint via remote server.

sqlite_optimize() → None: Run SQLite PRAGMA optimize via remote server.

sqlite_vacuum() → None: Run SQLite VACUUM via remote server.

sqlite_incremental_vacuum(pages: int = 2000) → None: Run incremental VACUUM via remote server.

analyze_system_resources() → Dict[str, Any]: Analyze remote server system resources.

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

query_stream(query: str, *, search_type: str = 'hybrid', return_type: str = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, batch_size: int = 10, context_window: int = 2, context_unit: str = 'chunks', context_truncate: bool = False, document_scoring_method: str = 'frequency_boost', document_scoring_options: Dict[str, Any] | None = None)

Stream query results from the server via SSE.

Yields lists of QueryResult objects as they arrive from the server.

Parameters:

query (str) – The search query text.
search_type (str) – One of ‘vector’, ‘keyword’, ‘hybrid’.
return_type (str) – One of ‘documents’, ‘chunks’, ‘context’, ‘enriched’.
k (int) – Maximum number of results.
batch_size (int) – Number of results per SSE batch.

Yields:

List[QueryResult] – Batches of query results as they stream from the server.

async query_stream_async(query: str, *, search_type: str = 'hybrid', return_type: str = 'documents', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, batch_size: int = 10, context_window: int = 2, context_unit: str = 'chunks', context_truncate: bool = False, document_scoring_method: str = 'frequency_boost', document_scoring_options: Dict[str, Any] | None = None): Async version of query_stream. Yields lists of QueryResult.

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

Compare two documents and return their similarity score.

Parameters:

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

Returns:

Similarity score between the two documents.

Return type:

float

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

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

Compare two documents with detailed chunk-level analysis.

Parameters:

doc_id_1 (str) – First document ID.
doc_id_2 (str) – Second document ID.
chunk_threshold (float) – Threshold for chunk-level similarity matching.

Returns:

Detailed comparison result with per-chunk alignments.

Return type:

DocumentComparisonResult

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

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

Find nearest neighbors for a document.

Parameters:

doc_id (str) – The document ID to find neighbors for.
k (int) – Number of neighbors to return.
score_threshold (float) – Minimum similarity score to include.
filters (dict, optional) – Metadata filter applied to candidate documents.

Returns:

List of nearest neighbor results.

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 version of nearest_neighbors.

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

Compute pairwise similarity matrix for documents.

Parameters:: doc_ids (Optional[List[str]]) – Document IDs to include. If None, uses all documents.
Returns:: Matrix with similarity scores between all document pairs.
Return type:: DocumentSimilarityMatrix

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

localvectordb.VectorDB(name: str, base_path: str | Path, **kwargs) → LocalVectorDB | RemoteVectorDB

Enhanced factory function that returns the appropriate VectorDB instance based on whether base_path looks like a URL or a local path, with optional async support.

This factory automatically handles the differences between local and remote implementations, and between sync and async variants, making it easy to switch between them.

Parameters:

name (str) – Name of the database
base_path (Union[str, Path]) – Path or URL to the database. If it starts with ‘http://’ or ‘https://’, a RemoteVectorDB will be created. Otherwise, a LocalVectorDB will be created.
**kwargs (dict) –
Additional arguments to pass to the appropriate constructor.

These include: - metadata_schema: Dict[str, MetadataField] - Schema for metadata fields - embedding_provider: str - Provider for embeddings (“ollama”, “openai”) - embedding_model: str - Model name for embeddings - embedding_config: Dict[str, Any] - Config for embedding provider - chunking_method: str - Method for chunking (“sentences”, “tokens”, etc.) - chunk_size: int - Maximum tokens per chunk - chunk_overlap: int - Overlap in the method’s own unit, not tokens (except “tokens”); keep small (1-3) - enable_gpu: bool - Whether to use GPU for FAISS - enable_fts: bool - Whether to enable full-text search - create_if_not_exists: bool - Whether to create if not exists

For RemoteVectorDB, these include: - api_key: str - API key for authentication - request_timeout: int - Timeout for HTTP requests - max_retries: int - Number of retry attempts - retry_delay: float - Base delay between retries - connection_pool_limits: httpx.Limits - HTTP connection pool settings

Returns:

An instance of the appropriate vector database class

Return type:

Union[LocalVectorDB, RemoteVectorDB]

Examples

Sync local database:

from localvectordb import VectorDB
from localvectordb.core import MetadataField, MetadataFieldType

# Create a sync local database
db = VectorDB(
    "my_docs",
    "./vector_storage",
    metadata_schema={
        'author': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'date': MetadataField(type=MetadataFieldType.DATE, indexed=True)
    },
    embedding_model="nomic-embed-text",
    chunk_size=500
)

Sync remote database:

# Create a sync remote database connection
db = VectorDB(
    "my_docs",
    "http://localhost:5000",
    api_key="your_api_key",
    metadata_schema={
        'author': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'date': MetadataField(type=MetadataFieldType.DATE, indexed=True)
    }
)

# Async built in
async with db as async_db:
    doc_ids = await async_db.upsert_async(["Document 1", "Document 2"])

Notes

The factory function automatically filters out incompatible parameters for each implementation
Local databases require appropriate dependencies (FAISS, SQLite)
Remote databases require a running LocalVectorDB server

Raises:

ImportError – If required dependencies are not available for the chosen implementation
ValueError – If invalid parameters are provided for the chosen implementation

class localvectordb.Document(id: str, content: str, metadata: Dict[str, ~typing.Any]=<factory>, created_at: datetime | None = None, updated_at: datetime | None = None, content_hash: str | None = None, chunks: List[Chunk] | None = None, sections: List[Section] | None = None)

Bases: object

A document in the vector database

id: str

content: str

metadata: Dict[str, Any]

created_at: datetime | None = None

updated_at: datetime | None = None

content_hash: str | None = None

chunks: List[Chunk] | None = None

sections: List[Section] | None = None

needs_update(new_content: str) → bool: Check if document content has changed

classmethod from_dict(data: dict) → Document | None: Create a Document from a dictionary response

__init__(id: str, content: str, metadata: Dict[str, ~typing.Any]=<factory>, created_at: datetime | None = None, updated_at: datetime | None = None, content_hash: str | None = None, chunks: List[Chunk] | None = None, sections: List[Section] | None = None) → None

class localvectordb.QueryResult(id: str, score: float, type: ~typing.Literal['document', 'chunk', 'section', 'context', 'enriched', 'group', 'aggregation'], content: str, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, document_id: str | None = None, position: ~localvectordb.core.ChunkPosition | None = None)

Bases: object

Result from a search query

id: str

score: float

type: Literal['document', 'chunk', 'section', 'context', 'enriched', 'group', 'aggregation']

content: str

metadata: Dict[str, Any]

document_id: str | None = None

position: ChunkPosition | None = None

get_context(original: str, window: int = 100) → str | None: Get context around chunk (only for chunk results)

classmethod from_dict(data: dict) → QueryResult | None: Create a QueryResult from a dictionary response

__init__(id: str, score: float, type: ~typing.Literal['document', 'chunk', 'section', 'context', 'enriched', 'group', 'aggregation'], content: str, metadata: ~typing.Dict[str, ~typing.Any] = <factory>, document_id: str | None = None, position: ~localvectordb.core.ChunkPosition | None = None) → None

class localvectordb.Chunk(content: str, position: ChunkPosition, tokens: int, index: int, faiss_id: int | None = None, content_hash: str | None = None)

Bases: object

Internal representation of a document chunk.

Encapsulates the content, position metadata, token count, and optional FAISS index identifier for a text segment.

Parameters:

content (str) – The text content of the chunk.
position (ChunkPosition) – The location of this chunk in the original document.
tokens (int) – Number of tokens in this chunk.
index (int) – Sequential index of the chunk within the document.
faiss_id (int, optional) – Identifier in the FAISS index, if applicable.

content: str

position: ChunkPosition

tokens: int

index: int

faiss_id: int | None = None

content_hash: str | None = None

calculate_content_hash() → str: Calculate SHA-256 hash of chunk content

content_equals(other: Chunk) → bool: Check if this chunk has the same content as another chunk

get_context(original: str, window: int = 100) → str: Get chunk with surrounding context from original document

highlight_in_original(original: str) → str: Return original text with chunk highlighted

__init__(content: str, position: ChunkPosition, tokens: int, index: int, faiss_id: int | None = None, content_hash: str | None = None) → None

class localvectordb.ChunkPosition(start: int, end: int, line: int, column: int, end_line: int, end_column: int)

Bases: object

Exact position tracking for a chunk in the original document.

Parameters:

start (int) – Character start position in the original document.
end (int) – Character end position in the original document.
line (int) – Line number in the original document (1-based).
column (int) – Column number in the original document (1-based).
end_line (int) – Line number of end of chunk in original document (1-based)
end_column (int) – Column number of end of chunk in original document (1-based)

start: int

end: int

line: int

column: int

end_line: int

end_column: int

to_dict() → dict

Convert the ChunkPosition to a dictionary.

Returns:: Dictionary representation with keys ‘start’, ‘end’, ‘line’, ‘column’.
Return type:: dict

Examples

>>> pos = ChunkPosition(start=0, end=10, line=1, column=1, end_line=1, end_column=10)
>>> pos.to_dict()
{'start': 0, 'end': 10, 'line': 1, 'column': 1, 'end_line': 1, 'end_column': 10}

classmethod from_dict(data: dict) → ChunkPosition

Create a ChunkPosition instance from a dictionary.

Parameters:: data (dict) – Dictionary with keys ‘start’, ‘end’, ‘line’, ‘column’.
Returns:: The constructed ChunkPosition object.
Return type:: ChunkPosition

Examples

>>> data = {'start': 0, 'end': 10, 'line': 1, 'column': 1, 'end_line': 1, 'end_column': 10}
>>> ChunkPosition.from_dict(data)
ChunkPosition(start=0, end=10, line=1, column=1, end_line=1, end_column=10)

__init__(start: int, end: int, line: int, column: int, end_line: int, end_column: int) → None

class localvectordb.MetadataField(type: MetadataFieldType | str | Type, indexed: bool = False, required: bool = False, default_value: Any = None, embedding_enabled: bool = False, fts_enabled: bool = False)

Bases: object

Defines a metadata field for documents.

Parameters:

type (MetadataFieldType or str or Type) – The type of the metadata field.
indexed (bool, optional) – Whether the field is indexed in the database, by default False.
required (bool, optional) – Whether the field is required, by default False.
default_value (Any, optional) – Default value for the field if not provided, by default None.
embedding_enabled (bool, optional) – Whether this field should have its own embeddings for vector search. Only applicable to TEXT and JSON fields, by default False.
fts_enabled (bool, optional) – Whether this field should have full-text search enabled. Only applicable to TEXT fields, by default False.

type: MetadataFieldType | str | Type

indexed: bool = False

required: bool = False

default_value: Any = None

embedding_enabled: bool = False

fts_enabled: bool = False

__init__(type: MetadataFieldType | str | Type, indexed: bool = False, required: bool = False, default_value: Any = None, embedding_enabled: bool = False, fts_enabled: bool = False) → None

class localvectordb.MetadataFieldType(*values)

Bases: str, Enum

TEXT = 'text'

INTEGER = 'integer'

REAL = 'real'

BOOLEAN = 'boolean'

DATE = 'date'

JSON = 'json'

valid_types() → Tuple[Type[Any], ...]

Bases: object

A section within a document, grouping multiple chunks.

Sections are an overlay on top of existing chunking. They provide a mid-level abstraction between documents and chunks for hierarchical retrieval.

index: int

heading: str | None

heading_level: int | None

start_pos: int

end_pos: int

start_line: int | None = None

end_line: int | None = None

content_hash: str | None = None

metadata: Dict[str, Any] | None = None

faiss_id: int | None = None

chunks: List[Chunk] | None = None

classmethod from_boundary(boundary: SectionBoundary, content_hash: str, faiss_id: int | None = None, chunks: List[Chunk] | None = None) → Section: Create a Section from a SectionBoundary.

__init__(index: int, heading: str | None, heading_level: int | None, start_pos: int, end_pos: int, start_line: int | None = None, end_line: int | None = None, content_hash: str | None = None, metadata: Dict[str, Any] | None = None, faiss_id: int | None = None, chunks: List[Chunk] | None = None) → None

Bases: object

Boundary information for a detected section in a document.

Used during ingestion to track where sections start and end in the original text.

index: int

heading: str | None

heading_level: int | None

start_pos: int

end_pos: int

start_line: int | None = None

end_line: int | None = None

metadata: Dict[str, Any] | None = None

__init__(index: int, heading: str | None, heading_level: int | None, start_pos: int, end_pos: int, start_line: int | None = None, end_line: int | None = None, metadata: Dict[str, Any] | None = None) → None

class localvectordb.ChunkAlignment(chunk_index_1: int, chunk_index_2: int, similarity: float)

Bases: object

Alignment between a chunk in one document and its best match in another.

Parameters:

chunk_index_1 (int) – Chunk index in the first document.
chunk_index_2 (int) – Best-matching chunk index in the second document.
similarity (float) – Cosine similarity between the two chunks.

chunk_index_1: int

chunk_index_2: int

similarity: float

to_dict() → Dict[str, Any]: Serialize to a JSON-compatible dict.

classmethod from_dict(data: Dict[str, Any]) → ChunkAlignment: Reconstruct from a dict produced by to_dict().

__init__(chunk_index_1: int, chunk_index_2: int, similarity: float) → None

class localvectordb.DocumentComparisonResult(doc_id_1: str, doc_id_2: str, overall_similarity: float, chunk_alignments: List[ChunkAlignment], matched_ratio_1: float, matched_ratio_2: float, unmatched_chunks_1: List[int], unmatched_chunks_2: List[int])

Bases: object

Rich comparison result between two documents.

Parameters:

doc_id_1 (str) – ID of the first document.
doc_id_2 (str) – ID of the second document.
overall_similarity (float) – Centroid-level cosine similarity between documents.
chunk_alignments (List[ChunkAlignment]) – Best match per chunk in doc_1, sorted by similarity descending.
matched_ratio_1 (float) – Fraction of doc_1 chunks with a match >= threshold.
matched_ratio_2 (float) – Fraction of doc_2 chunks with a match >= threshold.
unmatched_chunks_1 (List[int]) – Chunk indices in doc_1 with no match >= threshold.
unmatched_chunks_2 (List[int]) – Chunk indices in doc_2 with no match >= threshold.

doc_id_1: str

doc_id_2: str

overall_similarity: float

chunk_alignments: List[ChunkAlignment]

matched_ratio_1: float

matched_ratio_2: float

unmatched_chunks_1: List[int]

unmatched_chunks_2: List[int]

to_dict() → Dict[str, Any]: Serialize to a JSON-compatible dict.

classmethod from_dict(data: Dict[str, Any]) → DocumentComparisonResult: Reconstruct from a dict produced by to_dict().

__init__(doc_id_1: str, doc_id_2: str, overall_similarity: float, chunk_alignments: List[ChunkAlignment], matched_ratio_1: float, matched_ratio_2: float, unmatched_chunks_1: List[int], unmatched_chunks_2: List[int]) → None

class localvectordb.DocumentSimilarityMatrix(matrix: ndarray, doc_ids: List[str], embeddings: ndarray)

Bases: object

NxN similarity matrix for a set of documents.

Parameters:

matrix (np.ndarray) – (N, N) array of pairwise similarity scores.
doc_ids (List[str]) – Ordered document IDs matching rows/columns.
embeddings (np.ndarray) – (N, D) document embeddings used to compute the matrix.

matrix: ndarray

doc_ids: List[str]

embeddings: ndarray

to_dict() → Dict[str, Any]: Serialize to a JSON-compatible dict (arrays become nested lists).

classmethod from_dict(data: Dict[str, Any]) → DocumentSimilarityMatrix: Reconstruct from a dict produced by to_dict().

__init__(matrix: ndarray, doc_ids: List[str], embeddings: ndarray) → None

exception localvectordb.BaseLocalVectorDBException

Bases: Exception

Base class for all LocalVectorDB exceptions.

exception localvectordb.DatabaseError

Bases: BaseLocalVectorDBException

Raised for general database operation failures.

exception localvectordb.DatabaseNotFoundError

Bases: DatabaseError, KeyError

Raised if the Database cannot be found

exception localvectordb.DocumentNotFoundError(message: str, missing_ids: str | List[str] | None = None)

Bases: DatabaseError, KeyError

Raised when one or more requested documents cannot be found

__init__(message: str, missing_ids: str | List[str] | None = None)

exception localvectordb.DuplicateDocumentIDError

Bases: DatabaseError, ValueError

Raised when inserting document(s) and the id(s) already exist

exception localvectordb.IndexIntegrityError

Bases: DatabaseError

Raised when the SQLite rows and the FAISS index disagree in a way that corrupts query results – notably duplicate faiss_id values, which cause one vector to be attributed to two different documents.

Recover with lvdb db <name> repair.

exception localvectordb.UnsupportedIndexOperationError

Bases: DatabaseError

Raised when an operation is not supported by the configured FAISS index type.

Most commonly: IndexHNSWFlat cannot remove vectors, so documents cannot be deleted or replaced in a database backed by it.

exception localvectordb.MetadataFilterError

Bases: DatabaseError, ValueError

Raised when there’s an error in metadata filter specification or processing

exception localvectordb.PatchConflictError(message: str, expected: str | None = None, actual: str | None = None)

Bases: DatabaseError

Raised when a document patch’s expect_hash precondition does not match the stored content_hash – i.e. the document changed since the caller read it.

This is a distinct outcome from “document not found” and “no-op”: the document exists and the ops are valid, but applying them would clobber a concurrent write. Surfaces as HTTP 409 on the server.

__init__(message: str, expected: str | None = None, actual: str | None = None)

exception localvectordb.PatchError

Bases: DatabaseError, ValueError

Raised when a document patch’s ops cannot be applied: an unmatched or ambiguous find, an out-of-range or overlapping splice, or a malformed op.

The whole patch fails atomically – no partial write. Surfaces as HTTP 422 on the server.

exception localvectordb.EmbeddingError

Bases: BaseLocalVectorDBException, RuntimeError

Raised when an embedding provider fails to generate embeddings.

exception localvectordb.OllamaNotFoundError

Bases: EmbeddingError

Raised when Ollama is not installed or not running.

exception localvectordb.ConfigurationError

Bases: BaseLocalVectorDBException, RuntimeError

Raised when configuration is invalid or inconsistent.

exception localvectordb.ValidationError

Bases: BaseLocalVectorDBException, ValueError

Raised when there’s a validation error in input data

exception localvectordb.ConnectionPoolError

Bases: BaseLocalVectorDBException

Raised when a database connection cannot be acquired from the pool.

exception localvectordb.RerankerError

Bases: BaseLocalVectorDBException, RuntimeError

Raised when there’s an error in reranking operations.

exception localvectordb.CursorError

Bases: BaseLocalVectorDBException

Base class for cursor-related errors.

exception localvectordb.CursorExpiredError

Bases: CursorError

Raised when a cursor has expired or been closed.

exception localvectordb.CursorExhaustedError

Bases: CursorError

Raised when attempting to fetch from an exhausted cursor.

Bases: object

Main backup and restore manager for LocalVectorDB.

Provides comprehensive backup and recovery capabilities including full backups, incremental backups, and point-in-time recovery using SQLite’s backup API and FAISS index management.

Parameters:

database_path (Union[str, Path]) – Path to the LocalVectorDB database file
faiss_index_path (Union[str, Path], optional) – Path to the FAISS index file. If None, inferred from database_path.
config (BackupConfig, optional) – Backup configuration. If None, uses default configuration.
db (LocalVectorDB, optional) –
The live database these paths belong to. Pass this to back up a database that is open and may be written to concurrently: the backup then holds the database’s write lock and flushes the FAISS index to disk for the duration of the snapshot, so the SQLite copy and the index copy are mutually consistent.

Without it (the path-only form) the two stores are copied without coordination. A backup taken while a write is in flight can capture SQLite rows whose vectors are not yet in the copied index – i.e. dangling rows. So the path-only form is safe only for a database that is closed, or that is otherwise known to be quiescent.

Examples

Create a full backup of a closed database:

manager = BackupManager("/path/to/mydb.sqlite")
backup_id = manager.create_backup(BackupType.FULL)
print(f"Backup created: {backup_id}")

Create a consistent backup of a live, open database:

manager = BackupManager(db.db_path, db=db)
backup_id = manager.create_backup(BackupType.FULL)

Restore from backup:

manager.restore_backup(backup_id, "/path/to/restore/location")

List available backups:

backups = manager.list_backups()
for backup in backups:
    print(f"{backup.backup_id}: {backup.created_at}")

__init__(database_path: str | Path, faiss_index_path: str | Path | None = None, config: BackupConfig | None = None, db: Any | None = None)

create_backup(backup_type: BackupType = BackupType.FULL, parent_backup_id: str | None = None, backup_id: str | None = None) → str

Create a new backup of the database.

Parameters:

backup_type (BackupType) – Type of backup to create
parent_backup_id (str, optional) – ID of parent backup (required for incremental backups)
backup_id (str, optional) – Custom backup ID. If None, generates a UUID.

Returns:

Unique backup ID

Return type:

str

Raises:

FileNotFoundError – If database file doesn’t exist
ValueError – If incremental backup requested without parent ID

list_backups(backup_type: BackupType | None = None) → List[BackupMetadata]

List available backups.

Parameters:: backup_type (BackupType, optional) – Filter by backup type. If None, returns all backups.
Returns:: List of backup metadata objects
Return type:: List[BackupMetadata]

restore_backup(backup_id: str, restore_location: str | Path | None = None, overwrite_existing: bool = False) → Path

Restore database from backup.

Parameters:

backup_id (str) – ID of backup to restore
restore_location (Union[str, Path], optional) – Directory to restore to. If None, restores to original location.
overwrite_existing (bool) – Whether to overwrite existing files

Returns:

Path to restored database directory

Return type:

Path

Raises:

FileNotFoundError – If backup file not found
ValueError – If restore would overwrite existing files without permission

MAX_PATH_LENGTH = 4096

delete_backup(backup_id: str) → bool

Delete a backup.

Parameters:: backup_id (str) – ID of backup to delete
Returns:: True if backup was deleted successfully
Return type:: bool

cleanup_old_backups(retention_days: int | None = None) → int

Clean up old backups based on retention policy.

Parameters:: retention_days (int, optional) – Number of days to retain backups. If None, uses config value.
Returns:: Number of backups deleted
Return type:: int

verify_backup_streaming(backup_id: str, verify_archive_members: bool = True) → bool

Verify backup integrity using streaming without full extraction.

Parameters:

backup_id (str) – ID of backup to verify
verify_archive_members (bool, default True) – Whether to verify individual archive member checksums. If False, only verifies archive and manifest checksums.

Returns:

True if backup is valid

Return type:

bool

Notes

This method provides efficient verification by streaming archive contents rather than extracting to disk. Useful for large backups or when disk space is limited.

verify_backup(backup_id: str) → bool

Verify backup integrity without restoring.

Parameters:: backup_id (str) – ID of backup to verify
Returns:: True if backup is valid
Return type:: bool

get_backup_info(backup_id: str) → BackupMetadata | None

Get detailed information about a backup.

Parameters:: backup_id (str) – ID of backup to get info for
Returns:: Backup metadata if found
Return type:: BackupMetadata or None

class localvectordb.IncrementalBackupManager(backup_manager: BackupManager)

Bases: object

Specialized manager for incremental backups using WAL tracking.

Implements incremental backup functionality by tracking changes in SQLite’s Write-Ahead Log (WAL) and maintaining FAISS index deltas.

Parameters:: backup_manager (BackupManager) – Parent backup manager instance

__init__(backup_manager: BackupManager)

create_incremental_backup(parent_backup_id: str, backup_id: str | None = None) → str

Create an incremental backup based on changes since parent backup.

Parameters:

parent_backup_id (str) – ID of the parent (full or incremental) backup
backup_id (str, optional) – Custom backup ID. If None, generates a UUID.

Returns:

Unique backup ID for the incremental backup

Return type:

str

Raises:

ValueError – If parent backup not found or WAL mode not enabled
FileNotFoundError – If database files don’t exist

restore_incremental_backup_chain(target_backup_id: str, restore_location: str | Path) → Path

Restore database by applying a chain of incremental backups.

Parameters:

target_backup_id (str) – ID of the target backup (can be full or incremental)
restore_location (Union[str, Path]) – Directory to restore to

Returns:

Path to restored database directory

Return type:

Path

class localvectordb.PointInTimeRecoveryManager(backup_manager: BackupManager, incremental_manager: IncrementalBackupManager)

Bases: object

Point-in-time recovery (PITR) manager for LocalVectorDB.

Provides the ability to restore a database to any point in time within the backup retention window by combining full and incremental backups.

Parameters:

backup_manager (BackupManager) – Parent backup manager instance
incremental_manager (IncrementalBackupManager) – Incremental backup manager instance

__init__(backup_manager: BackupManager, incremental_manager: IncrementalBackupManager)

get_recovery_timeline() → List[Dict[str, Any]]

Get the available recovery timeline showing all recovery points.

Returns:: List of recovery points with timestamps and backup information
Return type:: List[Dict[str, Any]]

find_recovery_point(target_timestamp: datetime, tolerance_minutes: int = 60) → Dict[str, Any] | None

Find the best recovery point for a target timestamp.

Parameters:

target_timestamp (datetime) – Target timestamp for recovery
tolerance_minutes (int) – Maximum tolerance in minutes to find a recovery point

Returns:

Recovery point information if found

Return type:

Dict[str, Any] or None

restore_to_point_in_time(target_timestamp: datetime, restore_location: str | Path, tolerance_minutes: int = 60, dry_run: bool = False) → Dict[str, Any]

Restore database to a specific point in time.

Parameters:

target_timestamp (datetime) – Target timestamp for recovery
restore_location (Union[str, Path]) – Directory to restore to
tolerance_minutes (int) – Maximum tolerance in minutes to find a recovery point
dry_run (bool) – If True, only validate the recovery without actually restoring

Returns:

Recovery operation results with status and details

Return type:

Dict[str, Any]

validate_recovery_timeline() → Dict[str, Any]

Validate the recovery timeline for consistency and completeness.

Returns:: Validation results with any issues found
Return type:: Dict[str, Any]

cleanup_recovery_timeline(max_age_days: int | None = None, keep_full_backups: int = 3, dry_run: bool = False) → Dict[str, Any]

Clean up old backups while maintaining recovery timeline integrity.

Parameters:

max_age_days (int, optional) – Maximum age of backups to keep. If None, uses config retention_days.
keep_full_backups (int) – Minimum number of full backups to keep regardless of age
dry_run (bool) – If True, only simulate cleanup without actually deleting

Returns:

Cleanup operation results

Return type:

Dict[str, Any]

get_recovery_recommendations(target_timestamp: datetime) → Dict[str, Any]

Get recommendations for recovering to a specific point in time.

Parameters:: target_timestamp (datetime) – Target timestamp for recovery
Returns:: Recovery recommendations and analysis
Return type:: Dict[str, Any]

class localvectordb.Migration(database_path: str | Path)

Bases: ABC

Abstract base class for metadata schema migrations.

All migration classes must inherit from this base class and implement the get_schema_changes() and get_rollback_changes() methods for forward and backward migrations.

This class focuses on metadata schema evolution rather than raw SQL operations, leveraging the existing DatabaseSchema functionality.

Variables:

version (str) – Target version for this migration
description (str) – Human-readable description of the migration
dependencies (List[str]) – List of migration versions that must be applied before this one

version: str = '0.0.0'

description: str = 'Base migration'

dependencies: List[str] = []

__init__(database_path: str | Path)

abstractmethod get_schema_changes() → Dict[str, Any]

Get the schema changes to apply in the forward migration.

Returns:

Schema changes specification with the following structure:

{
    'new_schema': Dict[str, MetadataField],  # Complete new schema
    'column_mapping': Dict[str, str],        # Rename mappings: old -> new
    'drop_columns': bool                     # Whether to drop unused columns
}

Return type:

Dict[str, Any]

Examples

Add new fields:

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

Rename and modify fields:

return {
    'new_schema': {
        'author_name': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'created_date': MetadataField(type=MetadataFieldType.DATE, indexed=True)
    },
    'column_mapping': {
        'author': 'author_name',
        'timestamp': 'created_date'
    }
}

abstractmethod get_rollback_changes() → Dict[str, Any]

Get the schema changes to apply for rollback.

Returns:: Schema changes specification for rolling back this migration
Return type:: Dict[str, Any]

validate_prerequisites(current_schema: Dict[str, MetadataField]) → bool

Validate that prerequisites for this migration are met.

Parameters:: current_schema (Dict[str, MetadataField]) – Current metadata schema
Returns:: True if prerequisites are met
Return type:: bool

class localvectordb.MigrationEngine(database_path: str | Path, migrations_directory: str | Path = './migrations', backup_manager: BackupManager | None = None, auto_backup: bool = True)

Bases: object

Core database migration management engine.

Handles discovery, validation, execution, and rollback of database migrations. Integrates with the backup system for safety and the versioning system for tracking applied migrations.

Parameters:

database_path (Union[str, Path]) – Path to the database file
migrations_directory (Union[str, Path], optional) – Directory containing migration scripts. Defaults to “./migrations”
backup_manager (BackupManager, optional) – Backup manager for creating safety backups before migrations
auto_backup (bool) – Whether to automatically create backups before migrations

__init__(database_path: str | Path, migrations_directory: str | Path = './migrations', backup_manager: BackupManager | None = None, auto_backup: bool = True)

discover_migrations() → Dict[str, MigrationScript]

Discover and load all migration scripts from the migrations directory.

Returns:: Dictionary mapping version strings to MigrationScript objects
Return type:: Dict[str, MigrationScript]

get_migration_order() → List[str]

Get the correct order for applying migrations based on dependencies.

Returns:: List of version strings in the order they should be applied
Return type:: List[str]
Raises:: ValueError – If circular dependencies are detected

get_applied_migrations() → List[Dict[str, Any]]

Get list of migrations that have been applied to the database.

Returns:: List of applied migration records
Return type:: List[Dict[str, Any]]

get_pending_migrations(target_version: str | None = None) → List[str]

Get list of migrations that need to be applied.

Parameters:: target_version (str, optional) – Target version to migrate to. If None, migrates to latest.
Returns:: List of migration versions that need to be applied
Return type:: List[str]

migrate(target_version: str | None = None, dry_run: bool = False, create_backup: bool | None = None) → Dict[str, Any]

Apply pending migrations up to the target version.

Parameters:

target_version (str, optional) – Target version to migrate to. If None, migrates to latest.
dry_run (bool) – If True, validate migrations without applying them
create_backup (bool, optional) – Whether to create a backup before migration. If None, uses auto_backup setting.

Returns:

Migration results with status and details

Return type:

Dict[str, Any]

rollback(target_version: str, dry_run: bool = False, create_backup: bool | None = None) → Dict[str, Any]

Rollback migrations to a target version.

Parameters:

target_version (str) – Target version to rollback to
dry_run (bool) – If True, validate rollback without applying it
create_backup (bool, optional) – Whether to create a backup before rollback. If None, uses auto_backup setting.

Returns:

Rollback results with status and details

Return type:

Dict[str, Any]

get_migration_status() → Dict[str, Any]

Get comprehensive status of database migrations.

Returns:: Migration status information
Return type:: Dict[str, Any]

create_migration_template(version: str, description: str, template_type: str = 'basic') → Path

Create a new migration template file.

Parameters:

version (str) – Version for the new migration
description (str) – Description of the migration
template_type (str) – Type of template to create (“basic”, “schema”, “data”)

Returns:

Path to the created migration file

Return type:

Path

class localvectordb.VersionManager(db_path: str | Path)

Bases: object

Manages database version tracking and migration metadata.

Handles SQLite PRAGMA user_version, version metadata in the config table, and migration tracking through the migration_log table.

Parameters:: db_path (Union[str, Path]) – Path to the SQLite database file

__init__(db_path: str | Path)

get_database_version(conn: Connection | None = None) → DatabaseVersion

Get the current database version.

Reads from PRAGMA user_version and falls back to config table if needed.

Parameters:: conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.
Returns:: Current database version
Return type:: DatabaseVersion

set_database_version(version: DatabaseVersion, conn: Connection | None = None) → None

Set the database version using both PRAGMA user_version and config table.

Parameters:

version (DatabaseVersion) – Version to set
conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.

get_migration_history(conn: Connection | None = None) → List[Dict]

Get the history of applied migrations.

Parameters:: conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.
Returns:: List of migration records with version, timestamp, and metadata
Return type:: List[Dict]

record_migration(version: str, rollback_script: str | None = None, checksum: str | None = None, conn: Connection | None = None) → None

Record a completed migration in the migration log.

Parameters:

version (str) – Version that was migrated to
rollback_script (str, optional) – SQL script for rolling back this migration
checksum (str, optional) – Checksum of the migration for integrity verification
conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.

needs_migration(target_version: DatabaseVersion | None = None, conn: Connection | None = None) → bool

Check if database needs migration to target version.

Parameters:

target_version (DatabaseVersion, optional) – Target version to check against. Defaults to CURRENT_SCHEMA_VERSION.
conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.

Returns:

True if migration is needed

Return type:

bool

initialize_version_tracking(conn: Connection | None = None) → None

Initialize version tracking for a new database.

Sets the database to the current schema version and records initial state.

Parameters:: conn (sqlite3.Connection, optional) – Database connection to use. If None, creates a new connection.

class localvectordb.QueryCursor(db: SearchMixin, candidates: List[CursorCandidate], config: CursorConfig, *, ttl_seconds: float = 300.0, default_batch_size: int = 50)

Bases: object

Holds cached search results with lazy SQLite metadata/content loading.

Lifecycle: 1. Created by SearchMixin.query_cursor() which runs FAISS/FTS search 2. Consumer calls fetch_batch() or iterates via stream() / __aiter__() 3. Each batch triggers a SQLite query for only that batch’s metadata 4. Cursor tracks position and expires after configurable TTL 5. Must be closed explicitly or via context manager

Parameters:

db (LocalVectorDBBase) – Reference to the database for lazy SQLite loading.
candidates (list of CursorCandidate) – Scored candidates from FAISS/FTS, sorted by score descending.
config (CursorConfig) – Query configuration captured at cursor creation time.
ttl_seconds (float) – Time-to-live in seconds (default 300 = 5 minutes).
default_batch_size (int) – Default number of results per batch (default 50).

__init__(db: SearchMixin, candidates: List[CursorCandidate], config: CursorConfig, *, ttl_seconds: float = 300.0, default_batch_size: int = 50)

close() → None: Close the cursor and release the database reference.

property closed: bool

property total_candidates: int: Total number of scored candidates (or documents for document return type).

property remaining: int: Number of unfetched candidates.

property is_exhausted: bool

fetch_batch(batch_size: int | None = None) → List[QueryResult]

Fetch the next batch of results, hydrating from SQLite lazily.

Parameters:: batch_size (int, optional) – Number of results to fetch. Defaults to default_batch_size.
Returns:: The next batch. Empty list when exhausted.
Return type:: list of QueryResult
Raises:: CursorExpiredError – If the cursor has been closed or its TTL has elapsed.

fetch_all() → List[QueryResult]: Fetch all remaining results at once.

stream(batch_size: int | None = None) → Iterator[List[QueryResult]]: Yield batches of results as a sync generator.

stream_individual(batch_size: int | None = None) → Iterator[QueryResult]: Yield individual QueryResult objects.

async fetch_batch_async(batch_size: int | None = None) → List[QueryResult]

Async version of fetch_batch. Hydrates from SQLite using the async pool.

Parameters:: batch_size (int, optional) – Number of results to fetch. Defaults to default_batch_size.
Return type:: list of QueryResult

async fetch_all_async() → List[QueryResult]: Fetch all remaining results asynchronously.

async stream_async(batch_size: int | None = None) → AsyncIterator[List[QueryResult]]: Yield batches of results as an async generator.

async stream_individual_async(batch_size: int | None = None) → AsyncIterator[QueryResult]: Yield individual QueryResult objects asynchronously.

class localvectordb.QueryBuilder(db: BaseVectorDB)

Bases: object

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

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

__init__(db: BaseVectorDB)

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

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

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

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

Add exact filter conditions.

Parameters:

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

Returns:

New QueryBuilder instance with filter conditions added

Return type:

QueryBuilder

Examples

Basic filtering:

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

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

Advanced filtering:

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

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

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

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

Add semantic filtering based on conceptual similarity.

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

limit(n: int) → QueryBuilder

Limit the number of results.

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

offset(n: int) → QueryBuilder

Skip the first n results.

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

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

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

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

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

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

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

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

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

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

Parameters:

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

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

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

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

clear_ordering() → QueryBuilder: Remove all ordering clauses.

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

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

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

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

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

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

Find minimum value in a field.

Parameters:

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

Returns:

New QueryBuilder instance with minimum aggregation

Return type:

QueryBuilder

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

Find maximum value in a field.

Parameters:

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

Returns:

New QueryBuilder instance with maximum aggregation

Return type:

QueryBuilder

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

Note

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

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

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

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

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

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

Rerank results using a cross-encoder or reranking model.

Parameters:

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

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

Enable query explanation or return execution plan directly.

Parameters:

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

Returns:

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

Return type:

Union[QueryBuilder, Dict[str, Any]]

Examples

Traditional usage (returns QueryBuilder with explain enabled):

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

New usage (returns execution plan directly):

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

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

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

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

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

Execute the query and return results.

Parameters:

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

Returns:

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

Return type:

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

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

Execute the query asynchronously with native async support.

Parameters:

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

Returns:

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

Return type:

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

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

Create a QueryCursor for this query configuration.

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

Parameters:

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

Return type:

QueryCursor

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

Create a QueryCursor asynchronously.

Parameters:

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

Return type:

QueryCursor

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

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

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

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

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

Get the execution plan for this query without executing it.

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

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

Examples

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

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

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

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

class localvectordb.ExtractorRegistry

Bases: object

Registry for file content extractors.

classmethod register(extractor: Type[BaseExtractor]) → None: Register a new extractor.

classmethod get_extractor(name: str) → BaseExtractor | None: Get an extractor by name.

classmethod list_extractors(available_only: bool = True) → List[str]: List all registered extractors.

classmethod refresh_plugins(): Force re-discovery of plugins (useful for testing)

classmethod get_extractors_for_file(filename: str, mimetype: str | None = None) → List[BaseExtractor]

Get suitable extractors for a file, sorted by priority.

Parameters:

filename (str) – Filename to check
mimetype (Optional[str]) – MIME type hint

Returns:

List of suitable extractors, sorted by priority (highest first)

Return type:

List[BaseExtractor]

classmethod extract_text(file_content: bytes, filename: str, mimetype: str | None = None, **kwargs: Any) → ExtractionResult

Extract text using the best available extractor.

Parameters:

file_content (bytes) – Raw file content
filename (str) – Original filename
mimetype (Optional[str]) – MIME type hint
kwargs – Optional extractor-specific keyword arguments (e.g. security options) forwarded to the selected extractor’s extract_text.

Returns:

Extraction result from the best available extractor

Return type:

ExtractionResult

classmethod get_supported_formats() → Dict[str, Dict[str, Any]]: Get information about all supported formats.

localvectordb.get_extractor_registry(): Get the global extractor registry.

localvectordb.get_profile_description(profile_name: str) → str

Get human-readable description of a profile.

Parameters:: profile_name (str) – Name of the profile
Returns:: Profile description
Return type:: str

localvectordb.is_valid_sqlite_pragma_profile(profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver']) → bool

localvectordb.get_sqlite_pragma_profile(profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'], *, default: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'] | None = None) → SQLitePragmaProfile | None

class localvectordb.FactChecker(databases: LocalVectorDB | list[LocalVectorDB], llm: LLMProvider | Any, model: str | None = None, similarity_threshold: float = 0.3, min_grounding_score: float = 0.7, search_type: str = 'hybrid', top_k: int = 5, max_concurrent: int = 5)

Bases: object

Fact-check LLM-generated text against one or more LocalVectorDB instances.

Parameters:

databases – One or more LocalVectorDB instances to search for evidence.
llm – An Anthropic, OpenAI, or Google GenAI client, or any object implementing the LLMProvider protocol.
model – Model name passed to the LLM provider for claim extraction and polarity classification. Defaults are provider-specific (Haiku for Anthropic, gpt-4o-mini for OpenAI, gemini-2.0-flash for Gemini).
similarity_threshold – Minimum similarity score for a retrieved chunk to be considered relevant.
min_grounding_score – Minimum polarity confidence for a claim to count as grounded.
search_type – Search mode used when querying the databases ("vector", "keyword", or "hybrid").
top_k – Number of chunks to retrieve per claim per database.
max_concurrent – Maximum number of claims to process concurrently.

__init__(databases: LocalVectorDB | list[LocalVectorDB], llm: LLMProvider | Any, model: str | None = None, similarity_threshold: float = 0.3, min_grounding_score: float = 0.7, search_type: str = 'hybrid', top_k: int = 5, max_concurrent: int = 5) → None

async check_async(text: str, sources: list[str] | None = None) → FactCheckResult

Fact-check text asynchronously.

Parameters:

text – The LLM-generated text to verify.
sources – Optional list of document IDs that were used to generate text. When provided, these are searched first; the full database is only queried when no supporting evidence is found or a contradiction is detected.

check(text: str, sources: list[str] | None = None) → FactCheckResult: Synchronous wrapper around check_async().

class localvectordb.FactCheckResult(claims: list[ClaimResult] = <factory>, overall_score: float = 0.0, has_contradictions: bool = False, citation_text: str = '', annotated_text: str | None = None)

Bases: object

Aggregate result of fact-checking a text against one or more databases.

claims: list[ClaimResult]

overall_score: float = 0.0

has_contradictions: bool = False

citation_text: str = ''

annotated_text: str | None = None

__init__(claims: list[ClaimResult] = <factory>, overall_score: float = 0.0, has_contradictions: bool = False, citation_text: str = '', annotated_text: str | None = None) → None

class localvectordb.ClaimResult(claim: str, grounded: bool, confidence: float, source_id: str | None = None, source_excerpt: str | None = None, contradiction: bool = False, polarity: Polarity | None = None, similarity: float | None = None, original_sentence: str | None = None, database_name: str | None = None)

Bases: object

Result of checking a single factual claim against sources.

claim: str

grounded: bool

confidence: float

source_id: str | None = None

source_excerpt: str | None = None

contradiction: bool = False

polarity: Polarity | None = None

similarity: float | None = None

original_sentence: str | None = None

database_name: str | None = None

__init__(claim: str, grounded: bool, confidence: float, source_id: str | None = None, source_excerpt: str | None = None, contradiction: bool = False, polarity: Polarity | None = None, similarity: float | None = None, original_sentence: str | None = None, database_name: str | None = None) → None

class localvectordb.Polarity(*values)

Bases: str, Enum

Relationship between a claim and a source chunk.

SUPPORTS = 'supports'

CONTRADICTS = 'contradicts'

UNRELATED = 'unrelated'

localvectordb package

Subpackages

Submodules