RemoteVectorDB Client

The RemoteVectorDB client provides a powerful HTTP interface to LocalVectorDB servers with near-perfect API parity to the local LocalVectorDB implementation. This enables a seamless development-to-production workflow: rapidly prototype with LocalVectorDB locally, then deploy to production using RemoteVectorDB without changing your code.

Overview

RemoteVectorDB is the HTTP client for LocalVectorDB that maintains complete API compatibility with the local implementation. This design philosophy enables developers to:

1. Develop Locally: Start with LocalVectorDB for rapid prototyping with no server setup

2. Test Remotely: Switch to RemoteVectorDB for integration testing

3. Deploy to Production: Use the same code in production with distributed servers

4. Scale Seamlessly: Add servers and load balancing without code changes

The client communicates with LocalVectorDB servers via a RESTful HTTP API, supporting:

• Full document lifecycle management (CRUD operations)

• Vector, keyword, and hybrid search

• Metadata filtering and schema management

• SQLite tuning and optimization

• Async operations for high-performance applications

• Automatic retry logic and connection pooling

Key Features

Near-Perfect API Parity

Every method available in LocalVectorDB is also available in RemoteVectorDB, including:

• All document operations (upsert, insert, get, update, delete)

• All query methods (query, filter, query_multi_column)

• Metadata schema management

• Database tuning operations

• Async variants of all methods

Drop-in Replacement

Switch between local and remote with a single line change:

# Development: Local database
db = LocalVectorDB("mydb", "./data")

# Production: Remote database (same API!)
db = RemoteVectorDB("mydb", "https://vectordb.company.com")

VectorDB Factory Pattern

The VectorDB() factory automatically selects the right implementation:

from localvectordb import VectorDB

# Automatically creates LocalVectorDB
db = VectorDB("mydb", "./data")

# Automatically creates RemoteVectorDB
db = VectorDB("mydb", "https://vectordb.company.com")

Built-in Resilience

• Automatic retry with exponential backoff

• Connection pooling for efficiency

• Timeout configuration

• Comprehensive error handling

Authentication Support

• API key authentication

• Environment variable configuration

• Custom authorization headers

Development Workflow

The LocalVectorDB ecosystem is designed for a natural progression from development to production:

1. Start with LocalVectorDB

Begin development using LocalVectorDB for immediate productivity:

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

# Local development database
db = LocalVectorDB(
    name="products",
    base_path="./dev_data",
    metadata_schema={
        'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'price': MetadataField(type=MetadataFieldType.REAL, indexed=True),
        'in_stock': MetadataField(type=MetadataFieldType.BOOLEAN)
    }
)

# Develop your application
db.upsert(documents, metadata=metadata)
results = db.query("search term", filters={"category": "electronics"})

2. Use VectorDB Factory for Flexibility

Refactor to use the VectorDB factory for environment-based configuration:

from localvectordb import VectorDB
import os

# Configuration from environment
db_path = os.getenv("VECTOR_DB_PATH", "./dev_data")

# Works with both local paths and URLs!
db = VectorDB(
    name="products",
    base_path=db_path,  # "./dev_data" or "https://api.company.com"
    api_key=os.getenv("LVDB_API_KEY"),  # Ignored for local, used for remote
    metadata_schema={
        'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'price': MetadataField(type=MetadataFieldType.REAL, indexed=True),
        'in_stock': MetadataField(type=MetadataFieldType.BOOLEAN)
    }
)

3. Deploy with RemoteVectorDB

In production, simply change the configuration:

# Production: Set environment variables
# VECTOR_DB_PATH=https://vectordb.company.com
# LVDB_API_KEY=your-secure-api-key

# Same code works in production!
db = VectorDB(
    name="products",
    base_path=os.getenv("VECTOR_DB_PATH"),
    api_key=os.getenv("LVDB_API_KEY")
)

Connection and Authentication

Basic Connection

Connect to a LocalVectorDB server:

from localvectordb.client import RemoteVectorDB

# Basic connection (no authentication)
db = RemoteVectorDB(
    name="my_database",
    base_url="http://localhost:8000"
)

# HTTPS connection with authentication
db = RemoteVectorDB(
    name="my_database",
    base_url="https://vectordb.company.com",
    api_key="your-api-key"
)

Authentication Methods

RemoteVectorDB supports multiple authentication configurations:

Direct API Key:

db = RemoteVectorDB(
    name="mydb",
    base_url="https://api.example.com",
    api_key="sk-abc123def456"
)

Environment Variable (default: LVDB_API_KEY):

# Automatically reads from LVDB_API_KEY
db = RemoteVectorDB(
    name="mydb",
    base_url="https://api.example.com"
)

Custom Environment Variable:

# Use a custom environment variable
db = RemoteVectorDB(
    name="mydb",
    base_url="https://api.example.com",
    api_key="$MY_CUSTOM_API_KEY"  # Reads from MY_CUSTOM_API_KEY env var
)

Connection Configuration

Fine-tune connection behavior:

from httpx import Limits

db = RemoteVectorDB(
    name="mydb",
    base_url="https://api.example.com",
    api_key="your-api-key",

    # Timeout configuration
    request_timeout=30,  # Request timeout in seconds

    # Retry configuration
    max_retries=3,
    retry_delay=1.0,  # Base delay for exponential backoff

    # Connection pooling
    connection_pool_limits=Limits(
        max_keepalive_connections=20,
        max_connections=100,
        keepalive_expiry=30.0
    ),

    # Concurrent request limits
    max_concurrent_requests=5,

    # Custom auth header (for non-standard servers)
    authorization_header="X-API-Key"
)

Complete API Reference

RemoteVectorDB implements all methods from LocalVectorDB with both synchronous and asynchronous variants.

Document Operations

Upsert Documents:

# Sync: Add or update documents
doc_ids = db.upsert(
    documents=["Document 1", "Document 2"],
    metadata=[
        {"author": "Alice", "year": 2024},
        {"author": "Bob", "year": 2023}
    ],
    ids=["doc1", "doc2"]  # Optional custom IDs
)

# Async variant
doc_ids = await db.upsert_async(documents, metadata, ids)

Insert Documents:

# Sync: Insert new documents (fails if IDs exist)
doc_ids = db.insert(documents, metadata, ids)

# Async variant
doc_ids = await db.insert_async(documents, metadata, ids)

Get Documents:

# Get single document
doc = db.get("doc1")

# Get multiple documents
docs = db.get(["doc1", "doc2", "doc3"])

# Async variants
doc = await db.get_async("doc1")
docs = await db.get_async(["doc1", "doc2", "doc3"])

Update Documents:

# Update document content and/or metadata
db.update(
    id="doc1",
    content="Updated content",  # Optional
    metadata={"author": "Alice Smith", "updated": True}  # Optional
)

# Async variant
await db.update_async(id, content, metadata)

Patch Documents (in-place edit):

# Edit a document without re-sending its whole content. Ops resolve against
# the current content (character offsets), touch disjoint spans, and apply
# atomically. Returns a PatchResult(updated, new_hash, ops_applied).
result = db.patch("doc1", [{"op": "replace", "find": "draft", "replace": "final"}])

# Span splice + append/prepend
db.patch("doc1", [{"op": "splice", "start": 0, "end": 5, "text": "Hello"}])
db.patch("doc1", [{"op": "append", "text": " (revised)"}])

# Optimistic concurrency: fail with PatchConflictError if the stored content
# changed since you read it. An unmatched/ambiguous find raises PatchError.
doc = db.get("doc1")
db.patch("doc1", [{"op": "replace", "find": "v1", "replace": "v2"}],
         expect_hash=doc.content_hash)

# Async variant
result = await db.patch_async("doc1", [{"op": "append", "text": "!"}])

Note

patch is an interface win, not a compute win: it re-embeds the document through the normal upsert path (unchanged chunks reuse their vectors). Its value is that an agent need not regenerate — and risk corrupting — the whole document to change one line, and can guard against lost updates with expect_hash.

Delete Documents:

# Delete single or multiple documents
deleted_count = db.delete("doc1")
deleted_count = db.delete(["doc1", "doc2"])

# Async variants
deleted_count = await db.delete_async(ids)

Check Existence:

# Check if documents exist
exists = db.exists("doc1")  # Returns: bool
exist_list = db.exists(["doc1", "doc2"])  # Returns: List[bool]

# Async variants
exists = await db.exists_async(ids)

Count Documents:

# Count all documents
total = db.count()

# Count with filters
count = db.count(filters={"author": "Alice", "year": {"$gte": 2023}})

# Async variant
count = await db.count_async(filters)

File Operations

Upsert from Files:

# Process files and add to database
doc_ids = db.upsert_from_file(
    file_paths=["document.pdf", "report.docx"],
    metadata=[
        {"source": "research", "confidential": False},
        {"source": "internal", "confidential": True}
    ],
    ids=["pdf1", "docx1"],  # Optional
    # Optional security options passed through to the extractor
    extractor_kwargs={"strip_dangerous_elements": True}
)

# Async variant
doc_ids = await db.upsert_from_file_async(
    file_paths, metadata, ids
)

Insert from Files:

# Insert new files (fails if IDs exist)
doc_ids = db.insert_from_file(
    file_paths, metadata, ids, errors="raise"
)

# Async variant
doc_ids = await db.insert_from_file_async(
    file_paths, metadata, ids, errors="raise"
)

Chunk Operations

Upsert Chunks Directly:

from localvectordb.core import Chunk, ChunkPosition

# upsert_from_chunks takes a dict mapping each document ID to its chunks.
# Chunks may be plain strings or Chunk objects. A Chunk's fields are:
# content, position (a ChunkPosition), tokens, index, and the optional
# faiss_id / content_hash.
chunks_by_document = {
    "doc1": [
        Chunk(
            content="First chunk content",
            position=ChunkPosition(
                start=0, end=100, line=1, column=1, end_line=5, end_column=20
            ),
            tokens=18,
            index=0,
        ),
        Chunk(
            content="Second chunk content",
            position=ChunkPosition(
                start=100, end=200, line=5, column=21, end_line=9, end_column=15
            ),
            tokens=20,
            index=1,
        ),
    ]
}

# Upsert chunks (per-document metadata is optional)
db.upsert_from_chunks(chunks_by_document, metadata={"doc1": {"section": "intro"}})

# A list of plain strings per document also works:
db.upsert_from_chunks({"doc2": ["chunk one text", "chunk two text"]})

# Async variant
await db.upsert_from_chunks_async(chunks_by_document)

Get Chunk Embeddings:

# Retrieve embeddings for specific chunks
embeddings = db.get_chunk_embeddings(["chunk1", "chunk2"])

# Returns numpy array of shape (n_chunks, embedding_dim)

Search and Query Operations

Unified Query Interface:

# Vector search (semantic)
results = db.query(
    query="machine learning algorithms",
    search_type="vector",
    k=10,
    filters={"year": {"$gte": 2020}}
)

# Keyword search (FTS)
results = db.query(
    query="exact phrase match",
    search_type="keyword",
    k=10
)

# Hybrid search (combined)
results = db.query(
    query="neural networks",
    search_type="hybrid",
    k=10,
    vector_weight=0.5  # 50% vector, 50% keyword (keyword weight = 1 - vector_weight)
)

# Async variants
results = await db.query_async(query, **kwargs)

# Hierarchical search (server database created with hierarchical_embeddings=True)
results = db.query(
    query="how do I rotate the API key?",
    search_level="sections",   # "chunks" (default), "sections", or "documents"
)
# ...or section-grouped results from the chunk index:
results = db.query(query="billing", return_type="sections")

RemoteVectorDB.query accepts the same search_level="sections"/ "documents" and return_type="sections" options as the local database; see Hierarchical Embeddings. The options are forwarded to the server, which must host a database created with hierarchical_embeddings=True.

Note

search_level="fused" is not available on remote databases — a fused query on RemoteVectorDB raises NotImplementedError (the server query route does not accept it). Use a local LocalVectorDB for fused retrieval.

Multi-Column Search:

# Search the main content plus embedding-enabled metadata columns.
# Pass a single query string and the columns to search (use "content"
# for the main document text). If columns is omitted, all
# embedding-enabled fields plus the main content are searched.
results = db.query_multi_column(
    "neural networks",
    columns=["content", "title", "abstract"],
    search_type="vector",
    k=10
)

# Async variant
results = await db.query_multi_column_async(
    "neural networks", columns=["content", "title", "abstract"]
)

MongoDB-style Filtering:

# Filter documents with complex conditions
docs = db.filter(
    where={
        "author": "Alice",
        "year": {"$gte": 2020, "$lt": 2025},
        "tags": {"$in": ["AI", "ML"]},
        "$or": [
            {"published": True},
            {"internal_review": "approved"}
        ]
    },
    order_by="year DESC",
    limit=20,
    offset=0
)

# Async variant
docs = await db.filter_async(where, **kwargs)

Metadata Operations

Update Metadata Schema:

from localvectordb.core import MetadataField, MetadataFieldType

# Add new metadata fields
new_schema = {
    'new_field': MetadataField(
        type=MetadataFieldType.TEXT,
        indexed=True
    ),
    'score': MetadataField(
        type=MetadataFieldType.REAL,
        indexed=True,
        default_value=0.0
    )
}
db.update_metadata_schema(
    new_schema,
    drop_columns=False  # Keep columns removed from the schema (safe default)
)

# Async variant
await db.update_metadata_schema_async(new_schema, drop_columns=False)

Get Schema Information:

# Retrieve current metadata schema
schema_info = db.get_metadata_schema_info()
# Returns: {
#     "fields": {...},
#     "field_count": 5,
#     "indexed_fields": [...],
#     "required_fields": [...],
#     "field_types": {...}
# }

# Async variant
schema_info = await db.get_metadata_schema_info_async()

Database Tuning Operations

RemoteVectorDB inherits from TuningMixin, providing remote access to SQLite tuning:

# Get current tuning settings
settings = db.get_sqlite_tuning()

# Apply tuning profile
db.set_sqlite_tuning(
    profile="fast_ingest",  # or "balanced", "read_optimized", "durable", "memory_saver"
    overrides={"cache_size": -262144}  # 256MB cache
)

# Maintenance operations
db.sqlite_checkpoint(mode="FULL")  # WAL checkpoint
db.sqlite_optimize()  # Optimize query planner
db.sqlite_vacuum()  # Reclaim space
db.sqlite_incremental_vacuum(pages=1000)  # Incremental cleanup

# System analysis
resources = db.analyze_system_resources()

# Conditional checkpoint
was_checkpointed = db.checkpoint_if_wal_large(wal_mb_threshold=128)

Database Management

Get Statistics:

stats = db.get_stats()
# Returns: {
#     "total_documents": 10000,
#     "total_chunks": 50000,
#     "vector_dim": 768,
#     "db_size_mb": 1024.5,
#     "indexes": [...],
#     ...
# }

# Async variant
stats = await db.get_stats_async()

Health Checks:

# Check if server is healthy (property, not a method)
is_healthy = db.healthy

# Ping server (with caching)
is_alive = db.ping(force=False)  # Uses cached result if recent
is_alive = db.ping(force=True)   # Forces new ping

Connection Management:

# Check if connection is closed (property, not a method)
is_closed = db.closed

# Save any pending changes
db.save()
await db.save_async()

# Close connection and cleanup
db.close()
await db.close_async()

Context Manager Support:

# Automatic cleanup with context manager
with RemoteVectorDB("mydb", "http://localhost:8000") as db:
    db.upsert(documents)
    results = db.query("search")
# Connection automatically closed

# Async context manager
async with RemoteVectorDB("mydb", "http://localhost:8000") as db:
    await db.upsert_async(documents)
    results = await db.query_async("search")

Async Operations

Every method in RemoteVectorDB has an async variant for high-performance applications:

Basic Async Pattern

All async methods follow the naming convention of adding _async suffix:

import asyncio
from localvectordb.client import RemoteVectorDB

async def main():
    db = RemoteVectorDB("mydb", "http://localhost:8000")

    # Async document operations
    doc_ids = await db.upsert_async(["Doc 1", "Doc 2"])
    doc = await db.get_async(doc_ids[0])
    await db.update_async(doc_ids[0], content="Updated")
    await db.delete_async(doc_ids)

    # Async search
    results = await db.query_async("search term", k=5)

    # Async schema management
    await db.update_metadata_schema_async(updates)

    # Cleanup
    await db.close_async()

asyncio.run(main())

Concurrent Operations

Leverage async for concurrent operations:

async def process_batch(db, batch_data):
    tasks = []

    # Create concurrent upsert tasks
    for content, metadata in batch_data:
        task = db.upsert_async([content], [metadata])
        tasks.append(task)

    # Execute all upserts concurrently
    doc_ids_lists = await asyncio.gather(*tasks)
    return [ids[0] for ids in doc_ids_lists]

Async Context Manager

Use async context manager for automatic cleanup:

async def search_documents():
    async with RemoteVectorDB("mydb", "http://localhost:8000") as db:
        # Concurrent searches
        results = await asyncio.gather(
            db.query_async("machine learning", search_type="vector"),
            db.query_async("machine learning", search_type="keyword"),
            db.filter_async({"author": "Alice"})
        )
        return results
    # Connection automatically closed

Code Examples

Migration from Local to Remote

Here’s how to migrate existing LocalVectorDB code to RemoteVectorDB:

Original Local Code:

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

# Local database
db = LocalVectorDB(
    name="products",
    base_path="./data",
    metadata_schema={
        'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'price': MetadataField(type=MetadataFieldType.REAL, indexed=True)
    }
)

# Operations
db.upsert(["Product A", "Product B"], metadata=[...])
results = db.query("electronics", filters={"price": {"$lt": 1000}})

Migrated to Remote (Option 1: Direct):

from localvectordb.client import RemoteVectorDB
from localvectordb.core import MetadataField, MetadataFieldType

# Remote database - same API!
db = RemoteVectorDB(
    name="products",
    base_url="https://vectordb.company.com",
    api_key="your-api-key",
    metadata_schema={
        'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'price': MetadataField(type=MetadataFieldType.REAL, indexed=True)
    }
)

# Exact same operations work!
db.upsert(["Product A", "Product B"], metadata=[...])
results = db.query("electronics", filters={"price": {"$lt": 1000}})

Migrated to Remote (Option 2: Using Factory):

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

# Environment-based configuration
# Development: DB_PATH="./data"
# Production: DB_PATH="https://vectordb.company.com"

db = VectorDB(
    name="products",
    base_path=os.getenv("DB_PATH", "./data"),
    api_key=os.getenv("LVDB_API_KEY"),  # Only used if remote
    metadata_schema={
        'category': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
        'price': MetadataField(type=MetadataFieldType.REAL, indexed=True)
    }
)

# Same operations for both local and remote!
db.upsert(["Product A", "Product B"], metadata=[...])
results = db.query("electronics", filters={"price": {"$lt": 1000}})

RAG Application Example

Build a Retrieval-Augmented Generation system:

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

class RAGSystem:
    def __init__(self, db_path: str):
        # Works with both local and remote databases
        self.db = VectorDB(
            name="knowledge_base",
            base_path=db_path,
            api_key=os.getenv("LVDB_API_KEY"),
            metadata_schema={
                'source': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
                'section': MetadataField(type=MetadataFieldType.TEXT, indexed=True),
                'last_updated': MetadataField(type=MetadataFieldType.DATE, indexed=True)
            },
            embedding_model="nomic-embed-text",
            chunk_size=512,
            chunk_overlap=50
        )

    def add_documents(self, file_paths: list, source: str):
        """Add documents from files"""
        metadata = [{"source": source} for _ in file_paths]
        return self.db.upsert_from_file(file_paths, metadata=metadata)

    def retrieve_context(self, query: str, max_docs: int = 5):
        """Retrieve relevant context for a query"""
        results = self.db.query(
            query=query,
            search_type="hybrid",
            k=max_docs,
            vector_weight=0.5
        )

        contexts = []
        for result in results:
            contexts.append({
                'content': result.content,
                'source': result.metadata.get('source', 'Unknown'),
                'score': result.score
            })

        return contexts

    async def retrieve_context_async(self, query: str, max_docs: int = 5):
        """Async version for high-performance scenarios"""
        results = await self.db.query_async(
            query=query,
            search_type="hybrid",
            k=max_docs,
            vector_weight=0.5
        )
        # Process results...
        return contexts

# Usage - works identically for local and remote!

# Development
rag = RAGSystem("./local_data")

# Production
rag = RAGSystem("https://vectordb.company.com")

# Add knowledge base
rag.add_documents(["doc1.pdf", "doc2.md"], source="technical_docs")

# Retrieve context
contexts = rag.retrieve_context("How does the system handle errors?")

Document Processing Pipeline

Process documents with metadata extraction:

import asyncio
from datetime import datetime
from localvectordb import VectorDB

class DocumentProcessor:
    def __init__(self, db_url: str):
        self.db = VectorDB("documents", db_url)

    async def process_document_batch(self, documents: list):
        """Process documents in parallel"""
        tasks = []

        for doc in documents:
            task = self.process_single_document(doc)
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)

        successful = [r for r in results if not isinstance(r, Exception)]
        failed = [r for r in results if isinstance(r, Exception)]

        return {
            'successful': len(successful),
            'failed': len(failed),
            'errors': failed
        }

    async def process_single_document(self, doc: dict):
        """Process a single document"""
        # Extract metadata
        metadata = {
            'title': doc.get('title', 'Untitled'),
            'author': doc.get('author', 'Unknown'),
            'processed_at': datetime.now().isoformat(),
            'word_count': len(doc['content'].split()),
            'language': doc.get('language', 'en')
        }

        # Upsert to database
        doc_id = await self.db.upsert_async(
            documents=[doc['content']],
            metadata=[metadata],
            ids=[doc.get('id')]
        )

        return doc_id[0]

    async def search_recent(self, query: str, days: int = 7):
        """Search recently processed documents"""
        from datetime import datetime, timedelta

        cutoff_date = (datetime.now() - timedelta(days=days)).isoformat()

        results = await self.db.query_async(
            query=query,
            search_type="hybrid",
            filters={
                "processed_at": {"$gte": cutoff_date}
            },
            k=20
        )

        return results

Error Handling

RemoteVectorDB provides comprehensive error handling:

Exception Types

from localvectordb.exceptions import (
    DatabaseNotFoundError,
    DocumentNotFoundError,
    DuplicateDocumentIDError,
    EmbeddingError,
    DatabaseError
)

try:
    db = RemoteVectorDB("nonexistent", "http://localhost:8000")
except DatabaseNotFoundError as e:
    print(f"Database not found: {e}")

try:
    doc = db.get("missing_id")
except DocumentNotFoundError as e:
    print(f"Document not found: {e}")

try:
    db.insert(["content"], ids=["existing_id"])
except DuplicateDocumentIDError as e:
    print(f"Document already exists: {e}")

Retry Logic

RemoteVectorDB includes automatic retry with exponential backoff:

db = RemoteVectorDB(
    name="mydb",
    base_url="http://localhost:8000",
    max_retries=3,  # Retry failed requests up to 3 times
    retry_delay=1.0  # Base delay, exponentially increases
)

# Requests will automatically retry on:
# - Network errors
# - 5xx server errors
# - Timeout errors

Connection Health Monitoring

Monitor and handle connection health:

import time

def ensure_healthy_connection(db: RemoteVectorDB):
    """Ensure database connection is healthy"""
    if db.closed:  # property, not a method
        raise RuntimeError("Database connection is closed")

    if not db.ping():
        # Try to wait and retry
        time.sleep(1)
        if not db.ping(force=True):
            raise RuntimeError("Database server is not responding")

    if not db.healthy:  # property, not a method
        raise RuntimeError("Database is not healthy")

    return True

Performance Optimization

Connection Pooling

Optimize connection reuse:

from httpx import Limits

# Configure connection pooling for high-throughput
db = RemoteVectorDB(
    name="mydb",
    base_url="http://localhost:8000",
    connection_pool_limits=Limits(
        max_keepalive_connections=50,  # Keep more connections alive
        max_connections=200,  # Allow more total connections
        keepalive_expiry=60.0  # Keep connections alive longer
    ),
    max_concurrent_requests=20  # More concurrent embedding requests
)

Batch Operations

Use batch operations for efficiency:

# Inefficient: Multiple round trips
for doc in documents:
    db.upsert([doc['content']], [doc['metadata']])

# Efficient: Single batch operation
contents = [doc['content'] for doc in documents]
metadata = [doc['metadata'] for doc in documents]
db.upsert(contents, metadata)

Async Concurrency

Leverage async for parallel operations:

async def parallel_search(db, queries):
    """Execute multiple searches in parallel"""
    tasks = [
        db.query_async(query, k=10)
        for query in queries
    ]

    results = await asyncio.gather(*tasks)
    return results

Use Appropriate Search Types

Choose the right search type for your use case:

# Fast: Keyword search for exact matches
results = db.query("ERROR: Connection failed", search_type="keyword")

# Accurate: Vector search for semantic similarity
results = db.query("network connectivity issues", search_type="vector")

# Balanced: Hybrid for best of both
results = db.query("connection problems", search_type="hybrid")

Database URI Support

RemoteVectorDB supports URI-based connections:

from localvectordb.factory import from_uri

# Local database
db = from_uri("lvdb:///path/to/databases/mydb")

# Remote HTTP
db = from_uri("lvdb+http://localhost:8000/mydb")

# Remote HTTPS with auth
db = from_uri("lvdb+https://api_key@vectordb.company.com/mydb")

# With query parameters
db = from_uri(
    "lvdb+https://vectordb.company.com/mydb?"
    "chunk_size=1000&embedding_model=nomic-embed-text"
)

Best Practices

1. Use Environment Variables for Configuration

   Store sensitive configuration in environment variables:

   # .env file
   VECTOR_DB_URL=https://vectordb.company.com
   LVDB_API_KEY=sk-abc123def456
   
   # Application code
   db = VectorDB(
       "mydb",
       os.getenv("VECTOR_DB_URL"),
       api_key=os.getenv("LVDB_API_KEY")
   )
   

2. Implement Health Checks

   Monitor database health in production:

   async def health_check():
       try:
           if not db.healthy:  # property, not a method
               alert_ops_team("Database unhealthy")
               return False
           return True
       except Exception as e:
           alert_ops_team(f"Health check failed: {e}")
           return False
   

3. Use Context Managers

   Ensure proper cleanup with context managers:

   async def process_documents():
       async with RemoteVectorDB("mydb", url) as db:
           # Process documents
           await db.upsert_async(documents)
       # Connection automatically closed
   

4. Handle Errors Gracefully

   Implement comprehensive error handling:

   from localvectordb.exceptions import BaseLocalVectorDBException
   
   try:
       results = db.query("search term")
   except BaseLocalVectorDBException as e:
       logger.error(f"Database operation failed: {e}")
       # Implement fallback logic
   except Exception as e:
       logger.critical(f"Unexpected error: {e}")
       # Alert and fail safely
   

5. Optimize for Your Use Case

   • High Write Throughput: Use batch operations and async upserts

   • Low Latency Queries: Keep connection pools warm, use caching

   • Large Documents: Use file operations with chunking

   • Complex Filtering: Create appropriate indexes on metadata fields

See Also

• Quickstart - Get started with LocalVectorDB

• Server - Deploy LocalVectorDB server

• Query Types and Return Modes - Query methods and options

• Metadata Filtering Guide - Metadata schema and filtering

• Embeddings - Embedding providers configuration

• Performance Tuning Guide - Performance tuning guide