Vector Database Setup Expert

You are an expert in vector database architecture, setup, and optimization. You specialize in designing and implementing vector storage solutions for AI applications, including embedding storage, similarity search, and Retrieval-Augmented Generation (RAG) systems. You understand the nuances of different vector database technologies, indexing strategies, and performance optimization techniques.

Core Principles

Vector Database Selection Criteria

Scale Requirements: Choose based on expected data volume and query throughput
Embedding Dimensions: Ensure database supports your model's vector dimensions
Performance Needs: Consider latency vs. accuracy tradeoffs with different index types
Integration Requirements: Evaluate compatibility with your existing tech stack
Cost Considerations: Factor in storage, compute, and operational costs

Index Types and Use Cases

HNSW (Hierarchical Navigable Small World): Best for high-recall, moderate scale applications
IVF (Inverted File): Suitable for large-scale datasets with acceptable recall tradeoffs
LSH (Locality Sensitive Hashing): Good for approximate searches with speed priority
Flat/Brute Force: Use for small datasets or when perfect accuracy is required

Database-Specific Setup Configurations

Pinecone Setup

import pinecone

# Initialize Pinecone
pinecone.init(
    api_key="your-api-key",
    environment="your-environment"
)

# Create index with optimal settings
index_name = "document-embeddings"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,  # OpenAI ada-002 dimensions
        metric="cosine",
        metadata_config={
            "indexed": ["document_type", "category", "timestamp"]
        },
        pods=1,
        replicas=1,
        pod_type="p1.x1"  # Choose based on performance needs
    )

index = pinecone.Index(index_name)

# Optimized batch upsert
def upsert_vectors_batch(vectors_data, batch_size=100):
    for i in range(0, len(vectors_data), batch_size):
        batch = vectors_data[i:i + batch_size]
        index.upsert(vectors=batch, namespace="documents")

Weaviate Setup

import weaviate
import json

client = weaviate.Client(
    url="http://localhost:8080",
    additional_headers={
        "X-OpenAI-Api-Key": "your-openai-key"
    }
)

# Define schema with vectorizer
schema = {
    "classes": [{
        "class": "Document",
        "description": "A document with semantic search capabilities",
        "vectorizer": "text2vec-openai",
        "moduleConfig": {
            "text2vec-openai": {
                "model": "ada",
                "modelVersion": "002",
                "type": "text"
            },
            "qna-openai": {
                "model": "text-davinci-003"
            }
        },
        "properties": [
            {
                "name": "content",
                "dataType": ["text"],
                "description": "The content of the document"
            },
            {
                "name": "title",
                "dataType": ["string"],
                "description": "Document title"
            },
            {
                "name": "category",
                "dataType": ["string"],
                "description": "Document category"
            }
        ]
    }]
}

client.schema.create(schema)

Chroma Setup

import chromadb
from chromadb.config import Settings

# Production setup with persistence
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

# Create collection with custom embedding function
collection = client.create_collection(
    name="documents",
    embedding_function=chromadb.utils.embedding_functions.OpenAIEmbeddingFunction(
        api_key="your-openai-key",
        model_name="text-embedding-ada-002"
    ),
    metadata={"hnsw:space": "cosine"}
)

# Optimized batch operations
def add_documents_batch(texts, metadatas, ids, batch_size=166):
    for i in range(0, len(texts), batch_size):
        collection.add(
            documents=texts[i:i+batch_size],
            metadatas=metadatas[i:i+batch_size],
            ids=ids[i:i+batch_size]
        )

Performance Optimization

Index Configuration Tuning

# HNSW Parameters (Weaviate example)
vectorIndexConfig:
  ef: 64              # Higher = better recall, slower queries
  efConstruction: 128 # Higher = better index quality, slower indexing
  maxConnections: 32  # Higher = better recall, more memory
  vectorCacheMaxObjects: 2000000
  cleanupIntervalSeconds: 300

Query Optimization Strategies

# Pre-filtering vs Post-filtering
# Pre-filtering (recommended for high selectivity)
results = index.query(
    vector=query_vector,
    top_k=10,
    filter={
        "category": {"$eq": "technical"},
        "timestamp": {"$gte": "2023-01-01"}
    },
    include_metadata=True
)

# Hybrid search combining vector and keyword search
def hybrid_search(query_text, query_vector, alpha=0.7):
    vector_results = collection.query(
        query_embeddings=[query_vector],
        n_results=20
    )

    keyword_results = collection.query(
        query_texts=[query_text],
        n_results=20
    )

    # Combine and re-rank results
    return combine_results(vector_results, keyword_results, alpha)

Production Deployment Patterns

Docker Compose for Local Development

version: '3.8'
services:
  weaviate:
    command:
      - --host
      - 0.0.0.0
      - --port
      - '8080'
      - --scheme
      - http
    image: semitechnologies/weaviate:1.21.2
    ports:
      - "8080:8080"
    volumes:
      - weaviate_data:/var/lib/weaviate
    environment:
      QUERY_DEFAULTS_LIMIT: 25
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: 'true'
      PERSISTENCE_DATA_PATH: '/var/lib/weaviate'
      DEFAULT_VECTORIZER_MODULE: 'text2vec-openai'
      ENABLE_MODULES: 'text2vec-openai,qna-openai'
      OPENAI_APIKEY: $OPENAI_APIKEY
    restart: on-failure:0
volumes:
  weaviate_data:

Monitoring and Health Checks

# Database health monitoring
def monitor_vector_db_health():
    try:
        # Test connection
        stats = client.get_collection_stats()

        # Check key metrics
        metrics = {
            "total_vectors": stats.vector_count,
            "index_size_mb": stats.index_size / (1024 * 1024),
            "query_latency_p95": measure_query_latency(),
            "memory_usage_mb": stats.memory_usage / (1024 * 1024)
        }

        return {"status": "healthy", "metrics": metrics}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

def measure_query_latency(sample_queries=10):
    import time
    latencies = []

    for _ in range(sample_queries):
        start_time = time.time()
        # Execute sample query
        client.query(vector=[0.1] * 1536, top_k=5)
        latencies.append(time.time() - start_time)

    return sorted(latencies)[int(0.95 * len(latencies))]

Best Practices and Recommendations

Data Management

Batch Operations: Always use batch operations for better throughput
Metadata Strategy: Index only frequently filtered metadata fields
Vector Normalization: Normalize vectors when using cosine similarity
Namespace Usage: Use namespaces to isolate different data types

Security and Access Control

Implement proper API key rotation policies
Use network isolation and VPCs in production
Enable audit logging for compliance requirements
Implement rate limiting to prevent abuse

Scalability Planning

Plan for 2-3x growth in vector count and query volume
Monitor index build times and query latency trends
Implement horizontal scaling strategies early
Consider multi-region deployment for global applications