localvectordb.database.base module

class localvectordb.database.base.BaseVectorDB

Bases: ABC

Abstract base class defining the interface for vector databases.

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

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

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

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

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

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

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

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

Update a document’s content and/or metadata.

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

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

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

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

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

abstractmethod query(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['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.

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

Create a QueryCursor for streaming results with lazy hydration.

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

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

Create a QueryCursor for async streaming results with lazy hydration.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

query_builder() → QueryBuilderInterface

Create a new QueryBuilder for this database.

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

Examples

Basic search:

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

Complex multi-field search with semantic filtering:

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

Async usage:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

abstractmethod async query_async(query: str, *, search_type: Literal['vector', 'keyword', 'hybrid'] = 'hybrid', return_type: Literal['documents', 'chunks', 'sections', 'context', 'enriched'] = 'documents', search_level: Literal['chunks', 'sections', 'documents', 'fused'] = 'chunks', k: int = 10, score_threshold: float = 0.0, filters: Dict[str, Any] | None = None, vector_weight: float = 0.5, section_weight: float = 0.65, context_window: int = 2, context_unit: Literal['chunks', 'tokens', 'words', 'characters'] = 'chunks', context_truncate: bool = False, semantic_dedup_threshold: float | None = None, document_scoring_method: Literal['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 asynchronously.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Return the stored chunks for a document.

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

class localvectordb.database.base.LocalVectorDBBase(name: str, base_path: str | Path = '.lvdb', *, metadata_schema: Dict[str, Any] | None = None, doc_id_pattern: str = 'doc_{idx}', embedding_provider: str = 'ollama', embedding_model: str = 'nomic-embed-text', embedding_config: Dict[str, Any] | None = None, chunking_method: str | Any = 'sentences', chunk_size: int = 500, chunk_overlap: int = 1, batch_size: int = 100, faiss_index_type: Literal['IndexFlatL2', 'IndexFlatIP', 'IndexHNSWFlat', 'IndexLSH'] = 'IndexFlatL2', faiss_index_hnsw_flat_neighbors: int | None = None, faiss_index_lsh_bits: int | None = None, enable_gpu: bool = False, enable_fts: bool = True, connection_pool_size: int = 10, sqlite_profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'] = 'balanced', sqlite_pragma_overrides: Dict[str, Any] | None = None, create_if_not_exists: bool = True)

Bases: BaseVectorDB, ABC

The abstract base class that defines the attributes and helper methods used by the various mixins.

__init__(name: str, base_path: str | Path = '.lvdb', *, metadata_schema: Dict[str, Any] | None = None, doc_id_pattern: str = 'doc_{idx}', embedding_provider: str = 'ollama', embedding_model: str = 'nomic-embed-text', embedding_config: Dict[str, Any] | None = None, chunking_method: str | Any = 'sentences', chunk_size: int = 500, chunk_overlap: int = 1, batch_size: int = 100, faiss_index_type: Literal['IndexFlatL2', 'IndexFlatIP', 'IndexHNSWFlat', 'IndexLSH'] = 'IndexFlatL2', faiss_index_hnsw_flat_neighbors: int | None = None, faiss_index_lsh_bits: int | None = None, enable_gpu: bool = False, enable_fts: bool = True, connection_pool_size: int = 10, sqlite_profile: Literal['balanced', 'fast_ingest', 'read_optimized', 'durable', 'memory_saver'] = 'balanced', sqlite_pragma_overrides: Dict[str, Any] | None = None, create_if_not_exists: bool = True)

schema: DatabaseSchema

chunker: PositionTrackingChunker

connection_pool: ConnectionPool

async_connection_pool: AsyncConnectionPool | None

index: Index | None

db_path: str | Path

async_max_connections: int

pipeline_queue_size: int

abstract property is_memory_only: bool