Using Vector Databases

Vector Databases for AI Applications

When to Use This Skill

Use this skill when implementing:

• RAG (Retrieval-Augmented Generation) systems for AI chatbots
• Semantic search capabilities (meaning-based, not just keyword)
• Recommendation systems based on similarity
• Multi-modal AI (unified search across text, images, audio)
• Document similarity and deduplication
• Question answering over private knowledge bases

Quick Decision Framework

1. Vector Database Selection

START: Choosing a Vector Database

EXISTING INFRASTRUCTURE?
├─ Using PostgreSQL already?
│  └─ pgvector (<10M vectors, tight budget)
│      See: references/pgvector.md
│
└─ No existing vector database?
   │
   ├─ OPERATIONAL PREFERENCE?
   │  │
   │  ├─ Zero-ops managed only
   │  │  └─ Pinecone (fully managed, excellent DX)
   │  │      See: references/pinecone.md
   │  │
   │  └─ Flexible (self-hosted or managed)
   │     │
   │     ├─ SCALE: <100M vectors + complex filtering ⭐
   │     │  └─ Qdrant (RECOMMENDED)
   │     │      • Best metadata filtering
   │     │      • Built-in hybrid search (BM25 + Vector)
   │     │      • Self-host: Docker/K8s
   │     │      • Managed: Qdrant Cloud
   │     │      See: references/qdrant.md
   │     │
   │     ├─ SCALE: >100M vectors + GPU acceleration
   │     │  └─ Milvus / Zilliz Cloud
   │     │      See: references/milvus.md
   │     │
   │     ├─ Embedded / No server
   │     │  └─ LanceDB (serverless, edge deployment)
   │     │
   │     └─ Local prototyping
   │        └─ Chroma (simple API, in-memory)

2. Embedding Model Selection

REQUIREMENTS?

├─ Best quality (cost no object)
│  └─ Voyage AI voyage-3 (1024d)
│      • 9.74% better than OpenAI on MTEB
│      • ~$0.12/1M tokens
│      See: references/embedding-strategies.md
│
├─ Enterprise reliability
│  └─ OpenAI text-embedding-3-large (3072d)
│      • Industry standard
│      • ~$0.13/1M tokens
│      • Maturity shortening: reduce to 256/512/1024d
│
├─ Cost-optimized
│  └─ OpenAI text-embedding-3-small (1536d)
│      • ~$0.02/1M tokens (6x cheaper)
│      • 90-95% of large model performance
│
├─ Multilingual (100+ languages)
│  └─ Cohere embed-v3 (1024d)
│      • ~$0.10/1M tokens
│
└─ Self-hosted / Privacy-critical
   ├─ English: nomic-embed-text-v1.5 (768d, Apache 2.0)
   ├─ Multilingual: BAAI/bge-m3 (1024d, MIT)
   └─ Long docs: jina-embeddings-v2 (768d, 8K context)

Core Concepts

Document Chunking Strategy

Recommended defaults for most RAG systems:

• Chunk size: 512 tokens (not characters)
• Overlap: 50 tokens (10% overlap)

Why these numbers?

• 512 tokens balances context vs. precision
• Too small (128-256): Fragments concepts, loses context
• Too large (1024-2048): Dilutes relevance, wastes LLM tokens
• 50 token overlap ensures sentences aren't split mid-context

See references/chunking-patterns.md for advanced strategies by content type.

Hybrid Search (Vector + Keyword)

Hybrid Search = Vector Similarity + BM25 Keyword Matching

User Query: "OAuth refresh token implementation"
           │
    ┌──────┴──────┐
    │             │
Vector Search   Keyword Search
(Semantic)      (BM25)
    │             │
Top 20 docs   Top 20 docs
    │             │
    └──────┬──────┘
           │
   Reciprocal Rank Fusion
   (Merge + Re-rank)
           │
    Final Top 5 Results

Why hybrid matters:

• Vector captures semantic meaning ("OAuth refresh" ≈ "token renewal")
• Keyword ensures exact matches ("refresh_token" literal)
• Combined provides best retrieval quality

See references/hybrid-search.md for implementation details.

Getting Started

Python + Qdrant Example

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

# 1. Initialize client
client = QdrantClient("localhost", port=6333)

# 2. Create collection
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE)
)

# 3. Insert documents with embeddings
points = [
    PointStruct(
        id=idx,
        vector=embedding,  # From OpenAI/Voyage/etc
        payload={
            "text": chunk_text,
            "source": "docs/api.md",
            "section": "Authentication"
        }
    )
    for idx, (embedding, chunk_text) in enumerate(chunks)
]
client.upsert(collection_name="documents", points=points)

# 4. Search with metadata filtering
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=5,
    query_filter={
        "must": [
            {"key": "section", "match": {"value": "Authentication"}}
        ]
    }
)

For complete examples, see examples/qdrant-python/.

TypeScript + Qdrant Example

import { QdrantClient } from '@qdrant/js-client-rest';

const client = new QdrantClient({ url: 'http://localhost:6333' });

// Create collection
await client.createCollection('documents', {
  vectors: { size: 1024, distance: 'Cosine' }
});

// Insert documents
await client.upsert('documents', {
  points: chunks.map((chunk, idx) => ({
    id: idx,
    vector: chunk.embedding,
    payload: {
      text: chunk.text,
      source: chunk.source
    }
  }))
});

// Search
const results = await client.search('documents', {
  vector: queryEmbedding,
  limit: 5,
  filter: {
    must: [
      { key: 'source', match: { value: 'docs/api.md' } }
    ]
  }
});

For complete examples, see examples/typescript-rag/.

RAG Pipeline Architecture

Complete Pipeline Components

1. INGESTION
   ├─ Document Loading (PDF, web, code, Office)
   ├─ Text Extraction & Cleaning
   ├─ Chunking (semantic, recursive, code-aware)
   └─ Embedding Generation (batch, rate-limited)

2. INDEXING
   ├─ Vector Store Insertion (batch upsert)
   ├─ Index Configuration (HNSW, distance metric)
   └─ Keyword Index (BM25 for hybrid search)

3. RETRIEVAL (Query Time)
   ├─ Query Processing (expansion, embedding)
   ├─ Hybrid Search (vector + keyword)
   ├─ Filtering & Post-Processing (metadata, MMR)
   └─ Re-Ranking (cross-encoder, LLM-based)

4. GENERATION
   ├─ Context Construction (format chunks, citations)
   ├─ Prompt Engineering (system + context + query)
   ├─ LLM Inference (streaming, temperature tuning)
   └─ Response Post-Processing (citations, validation)

5. EVALUATION (Production Critical)
   ├─ Retrieval Metrics (precision, recall, relevancy)
   ├─ Generation Metrics (faithfulness, correctness)
   └─ System Metrics (latency, cost, satisfaction)

Essential Metadata for Production RAG

Critical for filtering and relevance:

metadata = {
    # SOURCE TRACKING
    "source": "docs/api-reference.md",
    "source_type": "documentation",  # code, docs, logs, chat
    "last_updated": "2025-12-01T12:00:00Z",

    # HIERARCHICAL CONTEXT
    "section": "Authentication",
    "subsection": "OAuth 2.1",
    "heading_hierarchy": ["API Reference", "Authentication", "OAuth 2.1"],

    # CONTENT CLASSIFICATION
    "content_type": "code_example",  # prose, code, table, list
    "programming_language": "python",

    # FILTERING DIMENSIONS
    "product_version": "v2.0",
    "audience": "enterprise",  # free, pro, enterprise

    # RETRIEVAL HINTS
    "chunk_index": 3,
    "total_chunks": 12,
    "has_code": True
}

Why metadata matters:

• Enables filtering BEFORE vector search (reduces search space)
• Improves relevance through targeted retrieval
• Supports multi-tenant systems (filter by user/org)
• Enables versioned documentation (filter by product version)

Evaluation with RAGAS

Use scripts/evaluate_rag.py for automated evaluation:

from ragas import evaluate
from ragas.metrics import (
    faithfulness,       # Answer grounded in context
    answer_relevancy,   # Answer addresses query
    context_recall,     # Retrieved docs cover ground truth
    context_precision   # Retrieved docs are relevant
)

# Test dataset
test_data = {
    "question": ["How do I refresh OAuth tokens?"],
    "answer": ["Use /token with refresh_token grant..."],
    "contexts": [["OAuth refresh documentation..."]],
    "ground_truth": ["POST to /token with grant_type=refresh_token"]
}

# Evaluate
results = evaluate(test_data, metrics=[
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
])

# Production targets:
# faithfulness: >0.90 (minimal hallucination)
# answer_relevancy: >0.85 (addresses user query)
# context_recall: >0.80 (sufficient context retrieved)
# context_precision: >0.75 (minimal noise)

Performance Optimization

Embedding Generation

• Batch processing: 100-500 chunks per batch
• Caching: Cache embeddings by content hash
• Rate limiting: Respect API provider limits (exponential backoff)

Vector Search

• Index type: HNSW (Hierarchical Navigable Small World) for most cases
• Distance metric: Cosine for normalized embeddings
• Pre-filtering: Apply metadata filters before vector search
• Result diversity: Use MMR (Maximal Marginal Relevance) to reduce redundancy

Cost Optimization

• Embedding model: Consider text-embedding-3-small for budget constraints
• Dimension reduction: Use maturity shortening (3072d → 1024d)
• Caching: Implement semantic caching for repeated queries
• Batch operations: Group insertions/updates for efficiency

Common Workflows

1. Building a RAG Chatbot

• Vector database: Qdrant (self-hosted or cloud)
• Embeddings: OpenAI text-embedding-3-large
• Chunking: 512 tokens, 50 overlap, semantic splitter
• Search: Hybrid (vector + BM25)
• Integration: Frontend with ai-chat skill

See examples/qdrant-python/ for complete implementation.

2. Semantic Search Engine

• Vector database: Qdrant or Pinecone
• Embeddings: Voyage AI voyage-3 (best quality)
• Chunking: Content-type specific (see chunking-patterns.md)
• Search: Hybrid with re-ranking
• Filtering: Pre-filter by metadata (date, category, etc.)

3. Code Search

• Vector database: Qdrant
• Embeddings: OpenAI text-embedding-3-large
• Chunking: AST-based (function/class boundaries)
• Metadata: Language, file path, imports
• Search: Hybrid with language filtering

See examples/qdrant-python/ for code-specific implementation.

Integration with Other Skills

Frontend Skills

• ai-chat: Vector DB powers RAG pipeline behind chat interface
• search-filter: Replace keyword search with semantic search
• data-viz: Visualize embedding spaces, similarity scores

Backend Skills

• databases-relational: Hybrid approach using pgvector extension
• api-patterns: Expose semantic search via REST/GraphQL
• observability: Monitor embedding quality and retrieval metrics

Multi-Language Support

Python (Primary)

• Client: qdrant-client
• Framework: LangChain, LlamaIndex
• See: examples/qdrant-python/

Rust

• Client: qdrant-client (1,549 code snippets in Context7)
• Framework: Raw Rust for performance-critical systems
• See: examples/rust-axum-vector/

TypeScript

• Client: @qdrant/js-client-rest
• Framework: LangChain.js, integration with Next.js
• See: examples/typescript-rag/

Go

• Client: qdrant-go
• Use case: High-performance microservices

Troubleshooting

Poor Retrieval Quality

1. Check chunking strategy (too large/small?) 2. Verify metadata filtering (too restrictive?) 3. Try hybrid search instead of vector-only 4. Implement re-ranking stage 5. Evaluate with RAGAS metrics

Slow Performance

1. Use HNSW index (not Flat) 2. Pre-filter with metadata before vector search 3. Reduce vector dimensions (maturity shortening) 4. Batch operations (insertions, searches) 5. Consider GPU acceleration (Milvus)

High Costs

1. Switch to text-embedding-3-small 2. Implement semantic caching 3. Reduce chunk overlap 4. Use self-hosted embeddings (nomic, bge-m3) 5. Batch embedding generation

Qdrant Context7 Documentation

Primary resource: /llmstxt/qdrant_tech_llms-full_txt

• Trust score: High
• Code snippets: 10,154
• Quality score: 83.1

Access via Context7:

resolve-library-id({ libraryName: "Qdrant" })
get-library-docs({
  context7CompatibleLibraryID: "/llmstxt/qdrant_tech_llms-full_txt",
  topic: "hybrid search collections python",
  mode: "code"
})

Additional Resources

Reference Documentation

• references/qdrant.md - Comprehensive Qdrant guide
• references/pgvector.md - PostgreSQL pgvector extension
• references/milvus.md - Milvus/Zilliz for billion-scale
• references/embedding-strategies.md - Embedding model comparison
• references/chunking-patterns.md - Advanced chunking techniques

Code Examples

• examples/qdrant-python/ - FastAPI + Qdrant RAG pipeline
• examples/pgvector-prisma/ - PostgreSQL + Prisma integration
• examples/typescript-rag/ - TypeScript RAG with Hono

Automation Scripts

• scripts/generate_embeddings.py - Batch embedding generation
• scripts/benchmark_similarity.py - Performance benchmarking
• scripts/evaluate_rag.py - RAGAS-based evaluation

---

Next Steps: 1. Choose vector database based on scale and infrastructure 2. Select embedding model based on quality vs. cost trade-off 3. Implement chunking strategy for the content type 4. Set up hybrid search for production quality 5. Evaluate with RAGAS metrics 6. Optimize for performance and cost

Hybrid Search Example (Vector + BM25)

Python implementation of hybrid search combining vector similarity and BM25 keyword matching with Reciprocal Rank Fusion.

Features

• Vector similarity search (semantic)
• BM25 keyword search (exact matching)
• Reciprocal Rank Fusion (RRF) for result merging
• Qdrant hybrid search implementation
• Performance comparison (vector-only vs. keyword-only vs. hybrid)

Prerequisites

• Python 3.10+
• Qdrant running (Docker or managed)

Installation

pip install -r requirements.txt

Usage

from hybrid_search import HybridSearchEngine

# Initialize
engine = HybridSearchEngine(
    qdrant_url="localhost",
    collection_name="documents"
)

# Hybrid search
results = engine.search(
    query="OAuth refresh token implementation",
    limit=5
)

for result in results:
    print(f"Score: {result['score']:.3f}")
    print(f"Text: {result['text'][:100]}...")
    print()

Running

# Run example
python hybrid_search_example.py

# Run comparison benchmark
python compare_search_methods.py

How Hybrid Search Works

User Query: "OAuth refresh tokens"
           │
    ┌──────┴──────┐
    │             │
Vector Search   Keyword Search
(Semantic)      (BM25)
    │             │
Top 20 docs   Top 20 docs
    │             │
    └──────┬──────┘
           │
   Reciprocal Rank Fusion
   (Merge + Re-rank)
           │
    Final Top 5 Results

Benefits of Hybrid Search

• Vector search: Captures semantic meaning ("refresh" ≈ "renewal")
• Keyword search: Ensures exact matches aren't missed ("refresh_token" literal)
• Combined: Best retrieval quality

Project Structure

hybrid-search/
├── hybrid_search.py              # Main implementation
├── hybrid_search_example.py      # Usage example
├── compare_search_methods.py     # Benchmark comparison
├── requirements.txt              # Dependencies
└── README.md                     # This file

See individual files for implementation details.

pgvector + Prisma Integration Example

TypeScript/Node.js example using PostgreSQL with pgvector extension and Prisma ORM.

Features

• pgvector PostgreSQL extension
• Prisma ORM integration
• Vector similarity search
• OpenAI embeddings
• TypeScript type safety

Prerequisites

• Node.js 18+
• PostgreSQL 15+ with pgvector extension
• OpenAI API key

Setup

# Install dependencies
npm install

# Set up database
cp .env.example .env
# Edit .env with your DATABASE_URL and OPENAI_API_KEY

# Run migrations
npx prisma migrate dev

# Generate Prisma client
npx prisma generate

Usage

import { PrismaClient } from '@prisma/client';
import { generateEmbedding, similaritySearch } from './vector-search';

const prisma = new PrismaClient();

// Insert document with embedding
const embedding = await generateEmbedding('Document content');
await prisma.$executeRaw`
  INSERT INTO documents (content, embedding, metadata)
  VALUES (${content}, ${embedding}::vector, ${metadata}::jsonb)
`;

// Search
const results = await similaritySearch('search query', 5);
console.log(results);

Running

npm run dev

Project Structure

pgvector-prisma/
├── package.json
├── tsconfig.json
├── prisma/
│   └── schema.prisma
├── src/
│   ├── index.ts
│   └── vector-search.ts
└── README.md

See individual files for implementation details.

Qdrant Python RAG Pipeline Example

Complete implementation of a production-ready RAG (Retrieval-Augmented Generation) pipeline using Qdrant, LangChain, and OpenAI.

Features

• Document ingestion with semantic chunking
• Hybrid search (vector + BM25 keyword)
• Metadata filtering
• RAGAS evaluation
• FastAPI REST API
• Docker Compose deployment

Prerequisites

• Python 3.10+
• Docker and Docker Compose
• OpenAI API key

Installation

# Install dependencies
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env
# Edit .env and add your OPENAI_API_KEY

Running with Docker Compose

# Start Qdrant
docker-compose up -d

# Run the application
python main.py

Project Structure

qdrant-python/
├── main.py              # FastAPI application
├── rag_pipeline.py      # RAG implementation
├── document_loader.py   # Document ingestion
├── evaluation.py        # RAGAS evaluation
├── requirements.txt     # Python dependencies
├── docker-compose.yml   # Qdrant deployment
├── .env.example         # Environment template
└── README.md            # This file

API Endpoints

Ingest Documents

POST /ingest
{
  "file_path": "path/to/document.md",
  "metadata": {
    "source_type": "documentation",
    "product_version": "v2.0"
  }
}

Search

POST /search
{
  "query": "How do I implement OAuth refresh tokens?",
  "limit": 5,
  "filter": {
    "source_type": "documentation"
  }
}

Generate Answer (RAG)

POST /generate
{
  "query": "Explain OAuth refresh tokens",
  "limit": 5
}

Usage Example

from rag_pipeline import RAGPipeline

# Initialize
rag = RAGPipeline(
    qdrant_url="localhost",
    collection_name="documents",
    embedding_model="text-embedding-3-large"
)

# Ingest documents
rag.ingest_document(
    file_path="docs/api-reference.md",
    metadata={"source": "api-docs"}
)

# Search
results = rag.search(
    query="OAuth implementation",
    limit=5,
    filter={"source": "api-docs"}
)

# Generate answer
answer = rag.generate_answer(
    query="How do I implement OAuth?",
    limit=5
)
print(answer)

Evaluation

Run RAGAS evaluation:

python evaluation.py

Metrics:

• Faithfulness: >0.90 (minimal hallucination)
• Answer Relevancy: >0.85 (addresses query)
• Context Recall: >0.80 (sufficient context)
• Context Precision: >0.75 (minimal noise)

Configuration

Edit .env:

# OpenAI API Key
OPENAI_API_KEY=your-api-key-here

# Qdrant Configuration
QDRANT_URL=localhost
QDRANT_PORT=6333

# Embedding Model
EMBEDDING_MODEL=text-embedding-3-large
EMBEDDING_DIMENSIONS=1024

# Chunking Strategy
CHUNK_SIZE=512
CHUNK_OVERLAP=50

# LLM Configuration
LLM_MODEL=gpt-4-turbo-preview
LLM_TEMPERATURE=0.1

Production Deployment

1. Enable Qdrant authentication 2. Set up TLS/HTTPS 3. Configure rate limiting 4. Add monitoring (Prometheus/Grafana) 5. Set up backups 6. Use managed Qdrant Cloud for high availability

Troubleshooting

Issue: Qdrant connection fails

# Check if Qdrant is running
docker ps | grep qdrant

# View logs
docker logs qdrant

Issue: Poor retrieval quality

• Adjust chunking strategy (chunk size, overlap)
• Try hybrid search instead of vector-only
• Add metadata filtering
• Implement re-ranking

Issue: High costs

• Switch to text-embedding-3-small
• Implement semantic caching
• Reduce chunk overlap
• Use self-hosted embeddings

License

MIT

Rust + Axum + Qdrant Vector Search

High-performance vector search API using Rust, Axum web framework, and Qdrant vector database.

Features

• Axum async web framework
• Qdrant vector database integration
• OpenAI embeddings generation
• Semantic search endpoints
• Type-safe with compile-time checks

Files

• main.rs - Axum server with vector search routes
• models.rs - Request/response types
• qdrant.rs - Qdrant client wrapper
• Cargo.toml - Dependencies

Setup

# 1. Install Rust
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

# 2. Start Qdrant
docker run -p 6333:6333 qdrant/qdrant

# 3. Set environment variables
export OPENAI_API_KEY="your-key"
export QDRANT_URL="http://localhost:6333"

# 4. Run
cargo run --release

API Endpoints

# Index document
curl -X POST http://localhost:3000/documents \
  -H "Content-Type: application/json" \
  -d '{"text": "Rust is a systems programming language", "metadata": {"source": "docs"}}'

# Search
curl -X POST http://localhost:3000/search \
  -H "Content-Type: application/json" \
  -d '{"query": "programming languages", "limit": 5}'

Performance

• Throughput: 10,000+ req/s (single core)
• Latency: <5ms (p99)
• Memory: ~10MB baseline

See source files for implementation details.

TypeScript RAG with Hono + Qdrant

Full-stack RAG (Retrieval-Augmented Generation) application using TypeScript, Hono, and Qdrant.

Stack

• Backend: Hono (edge-first framework)
• Vector DB: Qdrant
• Embeddings: Voyage AI / OpenAI
• LLM: OpenAI GPT-4
• Deployment: Cloudflare Workers / Vercel

Files

• server.ts - Hono API server
• rag-chain.ts - RAG pipeline logic
• qdrant-client.ts - Qdrant integration
• streaming.ts - SSE streaming responses
• package.json - Dependencies

Setup

# 1. Install dependencies
npm install

# 2. Start Qdrant
docker run -p 6333:6333 qdrant/qdrant

# 3. Configure environment
cp .env.example .env
# Edit .env with your API keys

# 4. Run development server
npm run dev

API Endpoints

# Index documents
curl -X POST http://localhost:3000/api/index \
  -H "Content-Type: application/json" \
  -d '{"documents": [{"text": "...", "metadata": {}}]}'

# RAG query (streaming)
curl http://localhost:3000/api/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"query": "What is vector search?"}'

# Search only (no LLM)
curl http://localhost:3000/api/search \
  -H "Content-Type: application/json" \
  -d '{"query": "vector databases", "limit": 5}'

Deployment

Cloudflare Workers

npm run deploy:cloudflare

Vercel

vercel deploy

See source files for complete implementation.

Document Chunking Strategies for RAG

Table of Contents

• Overview
• Default Strategy (Works for 80% of Cases)
• Chunking by Content Type
• Implementation Strategies
• 1. Recursive Character Splitting (General Purpose)
• 2. Semantic Chunking (Intelligent Boundaries)
• 3. Code-Aware Chunking (AST-Based)
• 4. Markdown-Aware Chunking
• 5. Fixed-Size Chunking (Simple, Fast)
• Advanced Patterns
• Hierarchical Chunking (Parent-Child)
• Sliding Window Chunking
• Sentence-Based Chunking
• Metadata Enrichment During Chunking
• Chunking PDFs
• Chunking for Different RAG Patterns
• Extractive QA (Precise Answers)
• Conversational RAG (Context Understanding)
• Summarization
• Code Search
• Evaluating Chunking Quality
• Common Mistakes to Avoid
• 1. Using Character Count Instead of Tokens
• 2. No Overlap
• 3. Ignoring Content Structure
• 4. Chunks Too Large
• Performance Optimization
• Parallel Chunking
• Caching Chunked Documents
• Production Checklist
• Additional Resources

Overview

Chunking splits long documents into smaller pieces that fit within embedding model context windows and provide focused semantic meaning for retrieval.

Core principle: Balance between context (larger chunks) and precision (smaller chunks).

Default Strategy (Works for 80% of Cases)

CHUNK_SIZE = 512  # tokens, not characters
CHUNK_OVERLAP = 50  # tokens (10% overlap)

Why these numbers?

• 512 tokens: Sweet spot between context and precision
• Too small (128-256): Fragments concepts, loses context
• Too large (1024-2048): Dilutes relevance, wastes LLM tokens
• 50 token overlap: Ensures sentences/paragraphs aren't split awkwardly

Chunking by Content Type

Implementation Strategies

1. Recursive Character Splitting (General Purpose)

from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken

# Token counting function (OpenAI tokenizer)
def tiktoken_len(text):
    tokenizer = tiktoken.get_encoding('cl100k_base')
    tokens = tokenizer.encode(text, disallowed_special=())
    return len(tokens)

# Create splitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    length_function=tiktoken_len,
    separators=["\n\n", "\n", ". ", " ", ""]  # Try in order
)

# Split documents
chunks = splitter.split_text(document)

How it works: 1. Try splitting on \n\n (paragraphs) 2. If chunks still too large, try \n (lines) 3. Then . (sentences) 4. Then (words) 5. Finally split by character

Best for:

• General text documents
• Unknown/mixed content types
• Markdown files
• README files

2. Semantic Chunking (Intelligent Boundaries)

from langchain.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # or "standard_deviation"
    breakpoint_threshold_amount=95  # Top 5% semantic differences
)

chunks = splitter.split_text(document)

How it works: 1. Generates embeddings for sentences 2. Computes similarity between consecutive sentences 3. Splits where similarity drops significantly (semantic shift)

Best for:

• Structured documentation
• Technical articles with clear sections
• Content with logical flow
• When preserving semantic coherence is critical

Trade-off: Slower (generates embeddings for splitting), but better quality.

3. Code-Aware Chunking (AST-Based)

from langchain.text_splitter import Language
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Python code splitter
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=200,  # lines of code
    chunk_overlap=0   # Don't split functions
)

chunks = python_splitter.split_text(python_code)

# Supported languages
# Language.PYTHON, .JS, .TS, .JAVA, .CPP, .GO, .RUST, .RUBY, etc.

Separators for Python:

[
    "\nclass ",      # Class definitions
    "\ndef ",       # Function definitions
    "\n\tdef ",     # Indented methods
    "\n\n",         # Blank lines
    "\n",           # Lines
    " ",            # Spaces
]

Best for:

• Code search
• Documentation generation
• Code analysis RAG systems

4. Markdown-Aware Chunking

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Split by headers
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)

md_chunks = markdown_splitter.split_text(markdown_document)

# Then apply recursive splitting to long sections
text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
final_chunks = text_splitter.split_documents(md_chunks)

Best for:

• Documentation sites
• GitHub README files
• Technical blogs
• Preserving document hierarchy

5. Fixed-Size Chunking (Simple, Fast)

def fixed_size_chunks(text, chunk_size=512, overlap=50):
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = " ".join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

Best for:

• Prototyping
• Uniform content (chat logs, logs)
• When speed matters more than quality

Drawbacks:

• May split sentences/paragraphs awkwardly
• Doesn't respect semantic boundaries

Advanced Patterns

Hierarchical Chunking (Parent-Child)

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Parent chunks (large context)
parent_splitter = RecursiveCharacterTextSplitter(
    chunk_size=2048,
    chunk_overlap=200
)

# Child chunks (for retrieval)
child_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50
)

# Create hierarchy
parent_chunks = parent_splitter.split_text(document)
all_child_chunks = []

for parent in parent_chunks:
    child_chunks = child_splitter.split_text(parent)
    for child in child_chunks:
        all_child_chunks.append({
            "child_text": child,
            "parent_text": parent  # Store parent for context expansion
        })

# Retrieve child, return parent to LLM
# This gives precise retrieval + full context

Benefits:

• Retrieve with precision (small chunks)
• Provide LLM with full context (large chunks)
• Best of both worlds

Use case: Complex technical documentation where context is critical.

Sliding Window Chunking

def sliding_window_chunks(text, window_size=512, step_size=256):
    words = text.split()
    chunks = []

    for i in range(0, len(words), step_size):
        chunk = " ".join(words[i:i + window_size])
        chunks.append(chunk)
        if i + window_size >= len(words):
            break

    return chunks

When to use:

• Need high overlap for dense coverage
• Content has critical information at unpredictable locations
• Willing to trade storage for retrieval quality

Sentence-Based Chunking

from langchain.text_splitter import SentenceTransformersTokenTextSplitter

splitter = SentenceTransformersTokenTextSplitter(
    chunk_overlap=50,
    tokens_per_chunk=512
)

chunks = splitter.split_text(document)

Best for:

• Preserving sentence boundaries
• Natural language text
• When readability matters

Metadata Enrichment During Chunking

def chunk_with_metadata(document, source, section):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=50,
        length_function=tiktoken_len
    )

    chunks = splitter.split_text(document)

    # Enrich with metadata
    enriched_chunks = []
    for idx, chunk in enumerate(chunks):
        enriched_chunks.append({
            "text": chunk,
            "source": source,
            "section": section,
            "chunk_index": idx,
            "total_chunks": len(chunks),
            "has_code": "```" in chunk,
            "char_count": len(chunk),
            "token_count": tiktoken_len(chunk)
        })

    return enriched_chunks

Chunking PDFs

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Load PDF
loader = PyPDFLoader("document.pdf")
pages = loader.load()

# Chunk with page context
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)

all_chunks = []
for page in pages:
    chunks = splitter.split_text(page.page_content)
    for chunk in chunks:
        all_chunks.append({
            "text": chunk,
            "page_number": page.metadata['page'],
            "source": "document.pdf"
        })

Chunking for Different RAG Patterns

Extractive QA (Precise Answers)

• Chunk size: 256-384 tokens
• Overlap: 30-50 tokens
• Strategy: Sentence-based
• Goal: Find exact answer location

Conversational RAG (Context Understanding)

• Chunk size: 512-768 tokens
• Overlap: 50-100 tokens
• Strategy: Semantic chunking
• Goal: Provide conversational context

Summarization

• Chunk size: 1024-2048 tokens
• Overlap: 100-200 tokens
• Strategy: Hierarchical (parent-child)
• Goal: Capture full context for summaries

Code Search

• Chunk size: 100-200 lines
• Overlap: 0 (function boundaries)
• Strategy: AST-based
• Goal: Preserve code structure

Evaluating Chunking Quality

def evaluate_chunks(chunks):
    """Evaluate chunking quality metrics."""
    metrics = {
        "total_chunks": len(chunks),
        "avg_length": sum(len(c) for c in chunks) / len(chunks),
        "min_length": min(len(c) for c in chunks),
        "max_length": max(len(c) for c in chunks),
        "std_dev": np.std([len(c) for c in chunks])
    }
    return metrics

# Good chunking: Low std_dev (uniform sizes)
# Bad chunking: High std_dev (inconsistent)

Common Mistakes to Avoid

1. Using Character Count Instead of Tokens

# ❌ BAD: Character count
splitter = RecursiveCharacterTextSplitter(chunk_size=2000)  # 2000 chars

# ✅ GOOD: Token count
def tiktoken_len(text):
    tokenizer = tiktoken.get_encoding('cl100k_base')
    return len(tokenizer.encode(text))

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,  # 512 tokens
    length_function=tiktoken_len
)

2. No Overlap

# ❌ BAD: No overlap, may split sentences
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=0)

# ✅ GOOD: 10% overlap
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)

3. Ignoring Content Structure

# ❌ BAD: Treat code like prose
general_splitter.split_text(python_code)

# ✅ GOOD: Use language-specific splitter
python_splitter = RecursiveCharacterTextSplitter.from_language(Language.PYTHON)
python_splitter.split_text(python_code)

4. Chunks Too Large

# ❌ BAD: Chunks too large, dilutes relevance
splitter = RecursiveCharacterTextSplitter(chunk_size=2048)

# ✅ GOOD: 512 tokens balances context and precision
splitter = RecursiveCharacterTextSplitter(chunk_size=512)

Performance Optimization

Parallel Chunking

from concurrent.futures import ThreadPoolExecutor

def chunk_document(doc):
    splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
    return splitter.split_text(doc)

# Process multiple documents in parallel
with ThreadPoolExecutor(max_workers=10) as executor:
    all_chunks = list(executor.map(chunk_document, documents))

Caching Chunked Documents

import hashlib
import pickle

def get_cached_chunks(document, cache_dir="chunk_cache"):
    # Hash document content
    doc_hash = hashlib.sha256(document.encode()).hexdigest()
    cache_path = f"{cache_dir}/{doc_hash}.pkl"

    # Check cache
    if os.path.exists(cache_path):
        with open(cache_path, 'rb') as f:
            return pickle.load(f)

    # Chunk and cache
    chunks = chunk_document(document)
    with open(cache_path, 'wb') as f:
        pickle.dump(chunks, f)

    return chunks

Production Checklist

• [ ] Use token count, not character count
• [ ] Apply 10% overlap (50 tokens for 512 chunk size)
• [ ] Choose strategy based on content type
• [ ] Enrich chunks with metadata (source, section, index)
• [ ] Test chunking with sample documents
• [ ] Validate chunk sizes (min, max, avg)
• [ ] Consider hierarchical chunking for complex docs
• [ ] Cache chunked documents for efficiency
• [ ] Monitor chunk quality over time
• [ ] Document chunking strategy for the team

Additional Resources

• LangChain Text Splitters: https://python.langchain.com/docs/modules/data_connection/document_transformers/
• Chunking Strategies Guide: https://www.pinecone.io/learn/chunking-strategies/
• RAG Chunking Best Practices: https://docs.anthropic.com/en/docs/build-with-claude/rag

Embedding Generation Strategies

Table of Contents

• Overview
• Embedding Model Comparison (2025)
• Quality Benchmark: MTEB (Massive Text Embedding Benchmark)
• Managed Embedding APIs
• Voyage AI (Highest Quality)
• OpenAI (Enterprise Standard)
• OpenAI text-embedding-3-small (Cost-Optimized)
• Cohere (Multilingual Leader)
• Google text-embedding-004
• Self-Hosted (Open Source) Models
• nomic-embed-text-v1.5 (English, Best Open Source)
• BAAI/bge-m3 (Multilingual)
• jina-embeddings-v2 (Long Documents)
• Batch Processing Strategies
• API Rate Limiting
• Caching by Content Hash
• Embedding for Different Content Types
• Text Documents
• Code
• Queries (Search)
• Cost Optimization Strategies
• 1. Dimension Reduction (Maturity Shortening)
• 2. Use Smaller Models for Less Critical Content
• 3. Self-Host for High Volume
• Quality vs. Cost Trade-Off Matrix
• Monitoring Embedding Quality
• Migration Between Embedding Models
• Best Practices
• 1. Normalize Embeddings
• 2. Consistent Preprocessing
• 3. Separate Query and Document Embeddings (if supported)
• Additional Resources

Overview

Embedding models convert text, images, audio, or code into dense vector representations that capture semantic meaning. Choosing the right embedding model balances quality, cost, latency, and deployment requirements.

Embedding Model Comparison (2025)

Quality Benchmark: MTEB (Massive Text Embedding Benchmark)

Higher scores indicate better semantic understanding across diverse tasks.

Voyage AI Advantage: 9.74% better performance than OpenAI on MTEB, making it the quality leader.

Managed Embedding APIs

Voyage AI (Highest Quality)

import voyageai

client = voyageai.Client(api_key="your-api-key")

# Generate embeddings
embeddings = client.embed(
    texts=["Document text 1", "Document text 2"],
    model="voyage-3",  # or voyage-large-2, voyage-code-2
    input_type="document"  # or "query" for search queries
)

# Extract vectors
vectors = [emb.embedding for emb in embeddings.embeddings]

When to use:

• Best quality matters more than cost
• High-stakes search applications
• Enterprise RAG systems requiring maximum accuracy
• Budget allows ~$0.12/1M tokens

OpenAI (Enterprise Standard)

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# Generate embeddings
response = client.embeddings.create(
    input=["Document text 1", "Document text 2"],
    model="text-embedding-3-large"  # or text-embedding-3-small
)

# Extract vectors
vectors = [item.embedding for item in response.data]

Dimension Reduction (Maturity Shortening):

# Reduce 3072d to 1024d for cost savings
response = client.embeddings.create(
    input=["Document text"],
    model="text-embedding-3-large",
    dimensions=1024  # Can be: 256, 512, 1024, 1536, 3072
)

When to use:

• Enterprise reliability required
• Integration simplicity valued
• Dimension reduction needed (3072d → 1024d)
• Cost: $0.13/1M tokens acceptable

OpenAI text-embedding-3-small (Cost-Optimized)

response = client.embeddings.create(
    input=["Document text"],
    model="text-embedding-3-small"  # 1536 dimensions
)

When to use:

• Budget constraints critical
• 90-95% of large model quality acceptable
• Cost: $0.02/1M tokens (6x cheaper than large)
• High-volume applications

Cohere (Multilingual Leader)

import cohere

client = cohere.Client(api_key="your-api-key")

# Generate embeddings
response = client.embed(
    texts=["Document text 1", "Document text 2"],
    model="embed-v3",  # or embed-english-v3.0
    input_type="search_document"  # or "search_query", "classification"
)

vectors = response.embeddings

When to use:

• Global applications (100+ languages)
• Non-English content dominant
• Need input type optimization (document vs. query)
• Cost: ~$0.10/1M tokens

Google text-embedding-004

from google.cloud import aiplatform

# Generate embeddings via Vertex AI
embeddings = aiplatform.TextEmbedding.from_pretrained(
    model_name="text-embedding-004"
)

vectors = embeddings.get_embeddings(["Document text"])

When to use:

• Already using GCP ecosystem
• Vertex AI integration required
• Cost: ~$0.025/1M tokens
• 768 dimensions sufficient

Self-Hosted (Open Source) Models

nomic-embed-text-v1.5 (English, Best Open Source)

from sentence_transformers import SentenceTransformer

# Load model
model = SentenceTransformer('nomic-ai/nomic-embed-text-v1.5', trust_remote_code=True)

# Generate embeddings
embeddings = model.encode([
    "Document text 1",
    "Document text 2"
], convert_to_numpy=True)

Specifications:

• Dimensions: 768
• License: Apache 2.0
• Context length: 8192 tokens
• Quality: Competitive with commercial models
• Cost: Free (infrastructure only)

When to use:

• Privacy-critical applications
• English-only content
• Self-hosting required
• Budget for GPU infrastructure

BAAI/bge-m3 (Multilingual)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-m3')

# Generate embeddings
embeddings = model.encode([
    "English text",
    "中文文本",
    "Texte français"
], convert_to_numpy=True)

Specifications:

• Dimensions: 1024
• License: MIT
• Languages: 100+ languages
• Context length: 8192 tokens

When to use:

• Multilingual content
• Self-hosting required
• Strong performance needed
• License flexibility important

jina-embeddings-v2 (Long Documents)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('jinaai/jina-embeddings-v2-base-en')

# Generate embeddings for long documents
embeddings = model.encode([
    long_technical_doc  # Up to 8192 tokens
], convert_to_numpy=True)

Specifications:

• Dimensions: 768
• License: Apache 2.0
• Context length: 8192 tokens (2x most models)
• Best for: Technical documentation, long articles

When to use:

• Documents exceed 512 tokens regularly
• Technical documentation embedding
• Self-hosting preferred
• Long-form content

Batch Processing Strategies

API Rate Limiting

import time
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(multiplier=1, min=4, max=60), stop=stop_after_attempt(5))
def generate_embeddings_with_retry(texts, model="text-embedding-3-large"):
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

# Batch processing
batch_size = 100  # Adjust based on API limits
all_embeddings = []

for i in range(0, len(documents), batch_size):
    batch = documents[i:i+batch_size]
    embeddings = generate_embeddings_with_retry(batch)
    all_embeddings.extend(embeddings)
    time.sleep(1)  # Rate limiting

Caching by Content Hash

import hashlib
import json

class EmbeddingCache:
    def __init__(self, cache_file="embedding_cache.json"):
        self.cache_file = cache_file
        try:
            with open(cache_file, 'r') as f:
                self.cache = json.load(f)
        except FileNotFoundError:
            self.cache = {}

    def get_hash(self, text):
        return hashlib.sha256(text.encode()).hexdigest()

    def get(self, text):
        hash_key = self.get_hash(text)
        return self.cache.get(hash_key)

    def set(self, text, embedding):
        hash_key = self.get_hash(text)
        self.cache[hash_key] = embedding
        self._save()

    def _save(self):
        with open(self.cache_file, 'w') as f:
            json.dump(self.cache, f)

# Usage
cache = EmbeddingCache()

def get_embedding_cached(text):
    cached = cache.get(text)
    if cached:
        return cached

    # Generate new embedding
    response = client.embeddings.create(input=[text], model="text-embedding-3-large")
    embedding = response.data[0].embedding
    cache.set(text, embedding)
    return embedding

Embedding for Different Content Types

Text Documents

# Standard text embedding
def embed_document(text, model="text-embedding-3-large"):
    response = client.embeddings.create(input=[text], model=model)
    return response.data[0].embedding

Code

# Use code-specific models
import voyageai

client = voyageai.Client(api_key="your-api-key")

# Voyage Code-2 optimized for code
embeddings = client.embed(
    texts=["def hello_world():\n    print('Hello, world!')"],
    model="voyage-code-2",
    input_type="document"
)

# Alternative: OpenAI with code prefix
response = openai_client.embeddings.create(
    input=["CODE:\ndef hello_world():\n    print('Hello, world!')"],
    model="text-embedding-3-large"
)

Queries (Search)

# Use query-specific input type
import voyageai

client = voyageai.Client(api_key="your-api-key")

# Mark as query (not document)
query_embedding = client.embed(
    texts=["How do I implement OAuth refresh tokens?"],
    model="voyage-3",
    input_type="query"  # Optimizes for search queries
).embeddings[0].embedding

# Cohere query optimization
import cohere
cohere_client = cohere.Client(api_key="your-api-key")

query_embedding = cohere_client.embed(
    texts=["search query"],
    model="embed-v3",
    input_type="search_query"  # vs. "search_document"
).embeddings[0]

Cost Optimization Strategies

1. Dimension Reduction (Maturity Shortening)

# OpenAI: Reduce from 3072d to 1024d
response = client.embeddings.create(
    input=["text"],
    model="text-embedding-3-large",
    dimensions=1024  # 3x fewer dimensions = lower storage/compute
)

Savings:

• Storage: 3x reduction
• Vector search: 2-3x faster
• Quality loss: <5% in most cases

2. Use Smaller Models for Less Critical Content

# High-value content: Use voyage-3
important_embeddings = voyage_client.embed(
    texts=important_docs,
    model="voyage-3"
)

# Supplementary content: Use text-embedding-3-small
supplementary_embeddings = openai_client.embeddings.create(
    input=supplementary_docs,
    model="text-embedding-3-small"  # 6x cheaper
)

3. Self-Host for High Volume

# One-time setup cost, zero per-request cost
model = SentenceTransformer('nomic-ai/nomic-embed-text-v1.5')

# Free embeddings after infrastructure cost
embeddings = model.encode(millions_of_documents)

Break-even analysis:

• API cost: $0.12/1M tokens
• Self-hosted: GPU instance ~$500/month
• Break-even: ~4M tokens/month (~2M documents)

Quality vs. Cost Trade-Off Matrix

Monitoring Embedding Quality

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# Test semantic similarity
test_pairs = [
    ("OAuth refresh token", "token renewal mechanism"),  # Should be similar
    ("OAuth refresh token", "chocolate cake recipe")     # Should be different
]

for text1, text2 in test_pairs:
    emb1 = get_embedding(text1)
    emb2 = get_embedding(text2)
    similarity = cosine_similarity([emb1], [emb2])[0][0]
    print(f"'{text1}' <-> '{text2}': {similarity:.3f}")

Migration Between Embedding Models

# Regenerate all embeddings when changing models
def migrate_embeddings(old_collection, new_model):
    # 1. Retrieve all documents
    docs = old_collection.scroll(limit=10000)

    # 2. Generate new embeddings
    new_embeddings = []
    for doc in docs:
        embedding = generate_embedding(doc.payload['text'], new_model)
        new_embeddings.append(embedding)

    # 3. Create new collection with new dimensions
    new_collection = create_collection(
        name="documents_v2",
        vector_size=len(new_embeddings[0])
    )

    # 4. Insert with new embeddings
    new_collection.upsert(new_embeddings, payloads=[d.payload for d in docs])

Best Practices

1. Normalize Embeddings

import numpy as np

def normalize_embedding(embedding):
    return embedding / np.linalg.norm(embedding)

# Normalized embeddings work better with cosine distance

2. Consistent Preprocessing

def preprocess_text(text):
    # Consistent preprocessing for embeddings
    text = text.lower().strip()
    text = re.sub(r'\s+', ' ', text)  # Normalize whitespace
    return text

3. Separate Query and Document Embeddings (if supported)

# Document embedding
doc_emb = voyage_client.embed(
    texts=[document],
    model="voyage-3",
    input_type="document"
)

# Query embedding
query_emb = voyage_client.embed(
    texts=[query],
    model="voyage-3",
    input_type="query"
)

Additional Resources

• MTEB Leaderboard: https://huggingface.co/spaces/mteb/leaderboard
• Voyage AI Docs: https://docs.voyageai.com/
• OpenAI Embeddings Guide: https://platform.openai.com/docs/guides/embeddings
• Sentence Transformers: https://www.sbert.net/
• Cohere Embeddings: https://docs.cohere.com/docs/embeddings

Hybrid Search: Combining Vector and Keyword Search

Table of Contents

• Overview
• Why Hybrid Search
• Core Concepts
• Vector Search (Semantic)
• Keyword Search (BM25)
• Fusion Strategies
• Reciprocal Rank Fusion (RRF)
• Weighted Scoring Approaches
• Implementation by Platform
• Qdrant
• Pinecone
• Weaviate
• pgvector
• Advanced Patterns
• Performance Optimization
• Evaluation

Overview

Hybrid search combines vector similarity search (semantic understanding) with traditional keyword search (exact matching) to achieve superior retrieval quality. This approach addresses the limitations of each method when used in isolation.

Key benefit: Retrieval quality improvement of 15-30% over vector-only search in production RAG systems.

Why Hybrid Search

Vector Search Limitations

Strengths:

• Captures semantic meaning ("OAuth token renewal" ≈ "refresh authentication credentials")
• Handles synonyms and paraphrasing automatically
• Language-agnostic similarity

Weaknesses:

• May miss exact technical terms (API names, error codes)
• Can retrieve semantically similar but contextually wrong results
• Struggles with acronyms and identifiers

Keyword Search Limitations

Strengths:

• Exact matches for technical terms ("refresh_token" literal)
• Fast lookup for known phrases
• Deterministic and explainable

Weaknesses:

• Misses synonyms and paraphrases
• No understanding of semantic meaning
• Sensitive to typos and variations

Hybrid Advantage

Combining both methods provides:

• Recall: Vector search finds conceptually similar content
• Precision: Keyword search ensures important terms are present
• Robustness: Complementary strengths cover each method's weaknesses

Core Concepts

Vector Search (Semantic)

Uses embedding models to convert text into high-dimensional vectors, then computes similarity using distance metrics.

Distance metrics:

• Cosine similarity: Most common, works with normalized embeddings
• Euclidean distance: Absolute distance in vector space
• Dot product: Efficient for normalized vectors

Formula (Cosine):

similarity = (A · B) / (||A|| × ||B||)

Keyword Search (BM25)

BM25 (Best Matching 25) is a probabilistic ranking function for keyword matching.

Formula:

BM25(D,Q) = Σ IDF(qi) × (f(qi,D) × (k1 + 1)) / (f(qi,D) + k1 × (1 - b + b × |D| / avgdl))

Where:

• f(qi,D): Term frequency of query term qi in document D
• |D|: Document length
• avgdl: Average document length
• k1: Term frequency saturation (typical: 1.2)
• b: Length normalization (typical: 0.75)
• IDF(qi): Inverse document frequency

Key parameters:

• k1 = 1.2: Controls term frequency saturation
• b = 0.75: Controls document length penalty

Fusion Strategies

Three primary approaches to combine vector and keyword results:

1. Reciprocal Rank Fusion (RRF) - Rank-based merging 2. Weighted Scoring - Linear combination of normalized scores 3. Re-ranking - Two-stage retrieval with cross-encoder

Reciprocal Rank Fusion (RRF)

RRF combines ranked lists without requiring normalized scores, making it robust and implementation-agnostic.

Algorithm

Formula:

RRF_score(d) = Σ 1 / (k + rank_i(d))

Where:

• d: Document
• rank_i(d): Rank of document d in result set i
• k: Constant (typical: 60)

Example Calculation

Query: "OAuth refresh token implementation"

Vector Search Results:
1. Doc A (rank 1)
2. Doc B (rank 2)
3. Doc C (rank 3)

Keyword Search Results:
1. Doc B (rank 1)
2. Doc D (rank 2)
3. Doc A (rank 3)

RRF Scores (k=60):
Doc A: 1/(60+1) + 1/(60+3) = 0.0164 + 0.0159 = 0.0323
Doc B: 1/(60+2) + 1/(60+1) = 0.0161 + 0.0164 = 0.0325 ← Winner
Doc C: 1/(60+3) + 0 = 0.0159
Doc D: 0 + 1/(60+2) = 0.0161

Final Ranking: B, A, D, C

Characteristics

Advantages:

• No score normalization required
• Robust to score scale differences
• Simple to implement
• Works across different search systems

Disadvantages:

• Ignores absolute score magnitudes
• Treats all rank positions equally within the formula
• May not fully leverage high-confidence matches

Tuning the k Parameter

• k = 60: Default, balanced approach
• k = 10-30: Emphasizes top-ranked results more strongly
• k = 100-200: More gradual rank decay, considers more results

Weighted Scoring Approaches

Linear combination of normalized vector and keyword scores.

Normalization Methods

Min-Max Normalization:

normalized_score = (score - min_score) / (max_score - min_score)

Z-Score Normalization:

normalized_score = (score - mean_score) / std_dev

Weighted Combination

Formula:

hybrid_score = (alpha × vector_score) + ((1 - alpha) × keyword_score)

Where alpha ∈ [0, 1] controls the balance.

Recommended alpha values:

• alpha = 0.7: Semantic-heavy (general Q&A, conversational)
• alpha = 0.5: Balanced (most RAG applications)
• alpha = 0.3: Keyword-heavy (technical documentation, code search)

Adaptive Weighting

Adjust alpha based on query characteristics:

def calculate_alpha(query: str) -> float:
    """
    Adjust semantic vs keyword weight based on query type.
    """
    # Technical terms suggest keyword importance
    technical_terms = ["API", "function", "class", "error", "code"]
    has_technical = any(term in query for term in technical_terms)

    # Question words suggest semantic importance
    question_words = ["how", "why", "what", "when", "explain"]
    is_question = any(word in query.lower() for word in question_words)

    if has_technical and not is_question:
        return 0.3  # Keyword-heavy
    elif is_question and not has_technical:
        return 0.7  # Semantic-heavy
    else:
        return 0.5  # Balanced

Implementation by Platform

Architecture Overview

User Query: "OAuth refresh token implementation"
           │
    ┌──────┴──────┐
    │             │
Vector Search   BM25 Search
(Semantic)      (Keyword)
    │             │
Top 20 docs   Top 20 docs
    │             │
    └──────┬──────┘
           │
   Reciprocal Rank Fusion
   (Merge + Re-rank)
           │
    Final Top 5 Results

---

Qdrant

Qdrant provides built-in hybrid search with fusion support.

Basic Hybrid Search

from qdrant_client import QdrantClient
from qdrant_client.models import (
    SearchRequest,
    Prefetch,
    QueryEnum,
    FusionQuery
)

client = QdrantClient("localhost", port=6333)

# Hybrid search with RRF fusion
results = client.query_points(
    collection_name="documents",
    prefetch=[
        # Vector search prefetch
        Prefetch(
            query=query_vector,
            limit=20
        ),
        # Keyword search prefetch
        Prefetch(
            query=QueryEnum(text="OAuth refresh token"),
            using="text-index",  # Named text index
            limit=20
        )
    ],
    query=FusionQuery(fusion="rrf"),  # Reciprocal Rank Fusion
    limit=5
)

Collection Setup for Hybrid Search

from qdrant_client.models import (
    VectorParams,
    Distance,
    TextIndexParams,
    TextIndexType,
    TokenizerType
)

# Create collection with vector and text indexes
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE
    )
)

# Create named text index for BM25
client.create_payload_index(
    collection_name="documents",
    field_name="text",
    field_schema=TextIndexParams(
        type=TextIndexType.TEXT,
        tokenizer=TokenizerType.WORD,
        min_token_len=2,
        max_token_len=20,
        lowercase=True
    ),
    field_type="text"
)

Advanced: Custom Fusion Weights

# Weighted scoring instead of RRF
results = client.query_points(
    collection_name="documents",
    prefetch=[
        Prefetch(query=query_vector, limit=20),
        Prefetch(query=QueryEnum(text=query_text), using="text-index", limit=20)
    ],
    query=FusionQuery(
        fusion="score",  # Use score-based fusion
        weights=[0.7, 0.3]  # 70% vector, 30% keyword
    ),
    limit=5
)

Pinecone

Pinecone supports hybrid search through sparse-dense vectors.

Setup with Sparse Embeddings

from pinecone import Pinecone, ServerlessSpec
from pinecone_text.sparse import BM25Encoder

# Initialize
pc = Pinecone(api_key="your-api-key")
bm25 = BM25Encoder()

# Create index with hybrid support
index = pc.create_index(
    name="hybrid-search",
    dimension=1024,
    metric="dotproduct",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

# Fit BM25 on corpus
bm25.fit(documents)

# Upsert with sparse and dense vectors
index.upsert(
    vectors=[
        {
            "id": str(i),
            "values": dense_vector,  # From OpenAI/Voyage
            "sparse_values": bm25.encode_documents(doc),
            "metadata": {"text": doc}
        }
        for i, doc in enumerate(documents)
    ]
)

Hybrid Query

# Query with both dense and sparse
query_dense = get_embedding(query_text)
query_sparse = bm25.encode_queries(query_text)

results = index.query(
    vector=query_dense,
    sparse_vector=query_sparse,
    top_k=5,
    alpha=0.5  # Balance between dense (1.0) and sparse (0.0)
)

Custom Alpha Tuning

def hybrid_search(query: str, alpha: float = 0.5):
    """
    Hybrid search with configurable semantic vs keyword balance.

    Args:
        query: Search query string
        alpha: Weight for dense vectors (0.0 = sparse only, 1.0 = dense only)
    """
    query_dense = get_embedding(query)
    query_sparse = bm25.encode_queries(query)

    results = index.query(
        vector=query_dense,
        sparse_vector=query_sparse,
        top_k=10,
        alpha=alpha
    )

    return results

Weaviate

Weaviate provides native hybrid search with automatic score normalization.

Basic Hybrid Query

import weaviate
from weaviate.classes.query import HybridFusion

client = weaviate.connect_to_local()

collection = client.collections.get("Document")

# Hybrid search with RRF
results = collection.query.hybrid(
    query="OAuth refresh token implementation",
    alpha=0.5,  # 0.5 = balanced, 0 = BM25 only, 1 = vector only
    fusion_type=HybridFusion.RANKED,  # RRF fusion
    limit=5
)

for item in results.objects:
    print(f"Score: {item.metadata.score}")
    print(f"Text: {item.properties['text']}")

Schema Definition

import weaviate.classes.config as wc

# Create collection with vectorizer
client.collections.create(
    name="Document",
    vectorizer_config=wc.Configure.Vectorizer.text2vec_openai(
        model="text-embedding-3-large"
    ),
    properties=[
        wc.Property(
            name="text",
            data_type=wc.DataType.TEXT,
            tokenization=wc.Tokenization.WORD  # Enable BM25
        )
    ]
)

Fusion Types

from weaviate.classes.query import HybridFusion

# Ranked Fusion (RRF)
results_rrf = collection.query.hybrid(
    query="OAuth implementation",
    fusion_type=HybridFusion.RANKED,
    limit=5
)

# Relative Score Fusion (weighted)
results_relative = collection.query.hybrid(
    query="OAuth implementation",
    fusion_type=HybridFusion.RELATIVE_SCORE,
    alpha=0.7,
    limit=5
)

pgvector

PostgreSQL with pgvector requires manual implementation of hybrid search.

Schema Setup

-- Enable extensions
CREATE EXTENSION IF NOT EXISTS vector;
CREATE EXTENSION IF NOT EXISTS pg_trgm;  -- For text search

-- Create table with vector and text search
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1024),
    metadata JSONB,
    ts_vector tsvector GENERATED ALWAYS AS (to_tsvector('english', content)) STORED
);

-- Indexes
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
CREATE INDEX ON documents USING GIN (ts_vector);

RRF Implementation in SQL

WITH vector_search AS (
    SELECT
        id,
        content,
        1 - (embedding <=> $1::vector) AS vector_score,
        ROW_NUMBER() OVER (ORDER BY embedding <=> $1::vector) AS vector_rank
    FROM documents
    ORDER BY embedding <=> $1::vector
    LIMIT 20
),
keyword_search AS (
    SELECT
        id,
        content,
        ts_rank(ts_vector, plainto_tsquery('english', $2)) AS keyword_score,
        ROW_NUMBER() OVER (ORDER BY ts_rank(ts_vector, plainto_tsquery('english', $2)) DESC) AS keyword_rank
    FROM documents
    WHERE ts_vector @@ plainto_tsquery('english', $2)
    ORDER BY ts_rank(ts_vector, plainto_tsquery('english', $2)) DESC
    LIMIT 20
),
rrf_scores AS (
    SELECT
        COALESCE(v.id, k.id) AS id,
        COALESCE(v.content, k.content) AS content,
        COALESCE(1.0 / (60 + v.vector_rank), 0) + COALESCE(1.0 / (60 + k.keyword_rank), 0) AS rrf_score
    FROM vector_search v
    FULL OUTER JOIN keyword_search k ON v.id = k.id
)
SELECT id, content, rrf_score
FROM rrf_scores
ORDER BY rrf_score DESC
LIMIT 5;

Python Implementation with SQLAlchemy

from sqlalchemy import text
from pgvector.sqlalchemy import Vector

def hybrid_search_rrf(
    session,
    query_text: str,
    query_vector: list[float],
    k: int = 60,
    limit: int = 5
):
    """
    Hybrid search using RRF in PostgreSQL.
    """
    sql = text("""
        WITH vector_search AS (
            SELECT
                id,
                content,
                metadata,
                ROW_NUMBER() OVER (ORDER BY embedding <=> :vector) AS rank
            FROM documents
            ORDER BY embedding <=> :vector
            LIMIT 20
        ),
        keyword_search AS (
            SELECT
                id,
                content,
                metadata,
                ROW_NUMBER() OVER (
                    ORDER BY ts_rank(ts_vector, plainto_tsquery('english', :query)) DESC
                ) AS rank
            FROM documents
            WHERE ts_vector @@ plainto_tsquery('english', :query)
            ORDER BY ts_rank(ts_vector, plainto_tsquery('english', :query)) DESC
            LIMIT 20
        )
        SELECT
            COALESCE(v.id, k.id) AS id,
            COALESCE(v.content, k.content) AS content,
            COALESCE(v.metadata, k.metadata) AS metadata,
            (COALESCE(1.0 / (:k + v.rank), 0) +
             COALESCE(1.0 / (:k + k.rank), 0)) AS score
        FROM vector_search v
        FULL OUTER JOIN keyword_search k ON v.id = k.id
        ORDER BY score DESC
        LIMIT :limit
    """)

    results = session.execute(
        sql,
        {
            "vector": query_vector,
            "query": query_text,
            "k": k,
            "limit": limit
        }
    ).fetchall()

    return results

Weighted Scoring Approach

def hybrid_search_weighted(
    session,
    query_text: str,
    query_vector: list[float],
    alpha: float = 0.5,
    limit: int = 5
):
    """
    Hybrid search using weighted score combination.

    Args:
        alpha: Weight for vector score (0.0 = keyword only, 1.0 = vector only)
    """
    sql = text("""
        WITH vector_search AS (
            SELECT
                id,
                content,
                metadata,
                1 - (embedding <=> :vector) AS score
            FROM documents
            ORDER BY embedding <=> :vector
            LIMIT 20
        ),
        keyword_search AS (
            SELECT
                id,
                content,
                metadata,
                ts_rank(ts_vector, plainto_tsquery('english', :query)) AS score
            FROM documents
            WHERE ts_vector @@ plainto_tsquery('english', :query)
            ORDER BY ts_rank(ts_vector, plainto_tsquery('english', :query)) DESC
            LIMIT 20
        ),
        normalized AS (
            SELECT
                id,
                content,
                metadata,
                'vector' AS source,
                (score - MIN(score) OVER ()) /
                NULLIF(MAX(score) OVER () - MIN(score) OVER (), 0) AS norm_score
            FROM vector_search
            UNION ALL
            SELECT
                id,
                content,
                metadata,
                'keyword' AS source,
                (score - MIN(score) OVER ()) /
                NULLIF(MAX(score) OVER () - MIN(score) OVER (), 0) AS norm_score
            FROM keyword_search
        )
        SELECT
            id,
            content,
            metadata,
            SUM(
                CASE
                    WHEN source = 'vector' THEN :alpha * norm_score
                    WHEN source = 'keyword' THEN (1 - :alpha) * norm_score
                END
            ) AS hybrid_score
        FROM normalized
        GROUP BY id, content, metadata
        ORDER BY hybrid_score DESC
        LIMIT :limit
    """)

    results = session.execute(
        sql,
        {
            "vector": query_vector,
            "query": query_text,
            "alpha": alpha,
            "limit": limit
        }
    ).fetchall()

    return results

---

Advanced Patterns

Two-Stage Retrieval with Re-ranking

Combine hybrid search with cross-encoder re-ranking for maximum quality.

from sentence_transformers import CrossEncoder

# Stage 1: Hybrid retrieval (top 20)
initial_results = hybrid_search(query, top_k=20)

# Stage 2: Re-rank with cross-encoder (top 5)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
pairs = [[query, result.text] for result in initial_results]
scores = reranker.predict(pairs)

# Sort by reranking scores
reranked = sorted(
    zip(initial_results, scores),
    key=lambda x: x[1],
    reverse=True
)[:5]

Query Expansion

Expand queries before hybrid search to improve recall.

from openai import OpenAI

client = OpenAI()

def expand_query(query: str) -> list[str]:
    """
    Generate query variations for improved recall.
    """
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Generate 3 alternative phrasings of the query that maintain the same intent."
            },
            {
                "role": "user",
                "content": query
            }
        ]
    )

    variations = response.choices[0].message.content.split('\n')
    return [query] + variations

# Search with multiple query variations
def multi_query_hybrid_search(query: str):
    queries = expand_query(query)
    all_results = []

    for q in queries:
        results = hybrid_search(q, top_k=10)
        all_results.extend(results)

    # Deduplicate and merge scores
    merged = {}
    for result in all_results:
        if result.id not in merged:
            merged[result.id] = result
        else:
            merged[result.id].score += result.score

    return sorted(merged.values(), key=lambda x: x.score, reverse=True)[:5]

Metadata-Filtered Hybrid Search

Apply metadata filters before hybrid search to constrain the search space.

# Qdrant example
from qdrant_client.models import Filter, FieldCondition, MatchValue

results = client.query_points(
    collection_name="documents",
    prefetch=[
        Prefetch(
            query=query_vector,
            limit=20,
            filter=Filter(
                must=[
                    FieldCondition(
                        key="product_version",
                        match=MatchValue(value="v2.0")
                    ),
                    FieldCondition(
                        key="content_type",
                        match=MatchValue(value="documentation")
                    )
                ]
            )
        ),
        Prefetch(
            query=QueryEnum(text=query_text),
            using="text-index",
            limit=20,
            filter=Filter(
                must=[
                    FieldCondition(
                        key="product_version",
                        match=MatchValue(value="v2.0")
                    )
                ]
            )
        )
    ],
    query=FusionQuery(fusion="rrf"),
    limit=5
)

Performance Optimization

Prefetch Strategies

Optimal prefetch limits:

• Vector search: 20-50 results
• Keyword search: 20-50 results
• Final fusion: 5-10 results

Rationale:

• Larger prefetch improves fusion quality
• Diminishing returns beyond 50 results
• Balance quality vs. latency

Index Optimization

Vector indexes:

• Use HNSW for <10M vectors
• Use IVF for >10M vectors
• Tune ef_construct and m parameters

Text indexes:

• Use inverted indexes (GIN in PostgreSQL)
• Configure stopwords appropriately
• Consider language-specific tokenizers

Caching

Implement semantic caching for repeated queries:

from functools import lru_cache
import hashlib

def query_hash(query: str) -> str:
    """
    Create hash of query for caching.
    """
    return hashlib.md5(query.encode()).hexdigest()

@lru_cache(maxsize=1000)
def cached_hybrid_search(query_hash: str, query: str, alpha: float):
    """
    Cache hybrid search results.
    """
    return hybrid_search(query, alpha=alpha)

# Usage
qhash = query_hash(user_query)
results = cached_hybrid_search(qhash, user_query, alpha=0.5)

Batch Processing

Process multiple queries in parallel:

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def batch_hybrid_search(queries: list[str], alpha: float = 0.5):
    """
    Execute multiple hybrid searches concurrently.
    """
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor(max_workers=10) as executor:
        tasks = [
            loop.run_in_executor(executor, hybrid_search, q, alpha)
            for q in queries
        ]
        results = await asyncio.gather(*tasks)

    return results

Evaluation

Metrics

Retrieval Quality:

• Precision@k: Proportion of relevant results in top k
• Recall@k: Proportion of all relevant results found in top k
• MRR (Mean Reciprocal Rank): Average inverse rank of first relevant result
• NDCG (Normalized Discounted Cumulative Gain): Ranking quality metric

Production Targets:

• Precision@5: >0.80
• Recall@10: >0.70
• MRR: >0.75

A/B Testing Framework

from dataclasses import dataclass
from typing import List

@dataclass
class SearchVariant:
    name: str
    alpha: float
    fusion_type: str  # "rrf" or "weighted"
    k_param: int = 60

def evaluate_variants(
    queries: List[str],
    ground_truth: dict,
    variants: List[SearchVariant]
):
    """
    Compare different hybrid search configurations.
    """
    results = {}

    for variant in variants:
        precision_scores = []
        recall_scores = []

        for query in queries:
            if variant.fusion_type == "rrf":
                search_results = hybrid_search_rrf(
                    query,
                    k=variant.k_param
                )
            else:
                search_results = hybrid_search_weighted(
                    query,
                    alpha=variant.alpha
                )

            # Calculate metrics
            relevant = ground_truth[query]
            retrieved = [r.id for r in search_results]

            precision = len(set(retrieved) & set(relevant)) / len(retrieved)
            recall = len(set(retrieved) & set(relevant)) / len(relevant)

            precision_scores.append(precision)
            recall_scores.append(recall)

        results[variant.name] = {
            "precision": sum(precision_scores) / len(precision_scores),
            "recall": sum(recall_scores) / len(recall_scores)
        }

    return results

# Example usage
variants = [
    SearchVariant("vector_heavy", alpha=0.7, fusion_type="weighted"),
    SearchVariant("balanced", alpha=0.5, fusion_type="weighted"),
    SearchVariant("keyword_heavy", alpha=0.3, fusion_type="weighted"),
    SearchVariant("rrf_default", alpha=0.5, fusion_type="rrf", k_param=60),
    SearchVariant("rrf_aggressive", alpha=0.5, fusion_type="rrf", k_param=20)
]

results = evaluate_variants(test_queries, ground_truth_data, variants)

RAGAS Integration

from ragas import evaluate
from ragas.metrics import context_precision, context_recall

def evaluate_hybrid_rag(
    questions: list[str],
    ground_truth: list[str],
    alpha: float = 0.5
):
    """
    Evaluate RAG system with hybrid search using RAGAS.
    """
    contexts = []

    for question in questions:
        results = hybrid_search(question, alpha=alpha)
        contexts.append([r.text for r in results])

    dataset = {
        "question": questions,
        "ground_truth": ground_truth,
        "contexts": contexts
    }

    scores = evaluate(
        dataset,
        metrics=[context_precision, context_recall]
    )

    return scores

# Compare different alpha values
for alpha in [0.3, 0.5, 0.7]:
    scores = evaluate_hybrid_rag(test_questions, test_answers, alpha=alpha)
    print(f"Alpha {alpha}: Precision={scores['context_precision']:.3f}, "
          f"Recall={scores['context_recall']:.3f}")

Performance Comparison

Benchmark (MTEB retrieval tasks):

Conclusion: Hybrid provides 12-point recall improvement at minimal latency cost.

---

Summary

Hybrid search combines the semantic understanding of vector search with the precision of keyword matching, providing 15-30% improvement in retrieval quality for RAG applications.

Key takeaways:

• Use RRF for simplicity and robustness across platforms
• Use weighted scoring when fine-tuning search behavior
• Start with alpha=0.5 (balanced), adjust based on query characteristics
• Implement two-stage retrieval (hybrid + re-ranking) for maximum quality
• Evaluate with real queries and ground truth data

Platform recommendations:

• Qdrant: Built-in RRF, best metadata filtering
• Pinecone: Sparse-dense vectors, fully managed
• Weaviate: Native hybrid with multiple fusion types
• pgvector: Manual implementation, full SQL control

Milvus: Billion-Scale Vector Database

Table of Contents

• Overview
• When to Use Milvus
• Ideal For:
• Not Ideal For:
• Architecture Options
• Standalone Mode
• Cluster Mode
• Zilliz Cloud (Managed)
• Installation
• Docker (Standalone)
• Docker Compose (with etcd, MinIO)
• Python Client
• Installation
• Basic Usage
• Index Types
• IVF_FLAT
• IVF_SQ8 (Scalar Quantization)
• HNSW (Recommended)
• GPU Indexes (CUDA Required)
• Filtering with Scalar Fields
• Hybrid Search (Vector + Scalar)
• Partition Management (Multi-Tenancy)
• Performance Optimization
• Batch Operations
• Search Parameters Tuning
• Resource Configuration
• Monitoring
• Zilliz Cloud (Managed Milvus)
• Use Cases
• Billion-Scale Semantic Search
• Recommendation Systems
• Anomaly Detection
• Production Checklist
• Additional Resources

Overview

Milvus is an open-source vector database designed for billion-scale vector similarity search with GPU acceleration support.

When to Use Milvus

Ideal For:

• >100M vectors - Optimized for massive scale
• GPU acceleration - Leverage CUDA for 10-100x faster search
• Enterprise features - Role-based access control, multi-tenancy
• Distributed deployments - Horizontal scaling across nodes
• High throughput - Millions of searches per second

Not Ideal For:

• <10M vectors - Overkill for small datasets (use Qdrant/pgvector)
• Simple use cases - More complex than Qdrant
• Resource-constrained - Requires more infrastructure
• Rapid prototyping - Longer setup time

Architecture Options

Standalone Mode

• Single-node deployment
• Up to 100M vectors
• Good for development and small production workloads

Cluster Mode

• Distributed deployment
• Billions of vectors
• Horizontal scaling
• Production-grade high availability

Zilliz Cloud (Managed)

• Fully managed Milvus
• Serverless options available
• Auto-scaling
• Built-in monitoring

Installation

Docker (Standalone)

# Pull Milvus
docker pull milvusdb/milvus:latest

# Run standalone
docker run -d --name milvus \
  -p 19530:19530 \
  -p 9091:9091 \
  -v milvus_data:/var/lib/milvus \
  milvusdb/milvus:latest

Docker Compose (with etcd, MinIO)

version: '3.5'

services:
  etcd:
    image: quay.io/coreos/etcd:latest
    environment:
      - ETCD_AUTO_COMPACTION_MODE=revision
      - ETCD_AUTO_COMPACTION_RETENTION=1000
      - ETCD_QUOTA_BACKEND_BYTES=4294967296

  minio:
    image: minio/minio:latest
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: minio server /minio_data

  milvus:
    image: milvusdb/milvus:latest
    depends_on:
      - etcd
      - minio
    ports:
      - "19530:19530"
      - "9091:9091"
    volumes:
      - milvus_data:/var/lib/milvus

Python Client

Installation

pip install pymilvus

Basic Usage

from pymilvus import (
    connections, Collection, CollectionSchema, FieldSchema, DataType
)

# Connect
connections.connect("default", host="localhost", port="19530")

# Define schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="metadata", dtype=DataType.JSON)
]
schema = CollectionSchema(fields, description="Documents collection")

# Create collection
collection = Collection(name="documents", schema=schema)

# Insert data
entities = [
    [text1, text2, text3],                    # text field
    [[0.1]*1024, [0.2]*1024, [0.3]*1024],   # embedding field
    [{"source": "doc1"}, {"source": "doc2"}, {"source": "doc3"}]  # metadata
]
collection.insert(entities)

# Create index (HNSW recommended)
index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {"M": 16, "efConstruction": 256}
}
collection.create_index(field_name="embedding", index_params=index_params)

# Load collection to memory
collection.load()

# Search
search_params = {"metric_type": "COSINE", "params": {"ef": 64}}
results = collection.search(
    data=[[0.1]*1024],
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr="metadata['source'] == 'doc1'"  # Filtering
)

Index Types

IVF_FLAT

• Best for: <1M vectors
• Memory: High (stores full vectors)
• Speed: Fast
• Recall: Good

index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "COSINE",
    "params": {"nlist": 1024}
}

IVF_SQ8 (Scalar Quantization)

• Best for: 1M-10M vectors
• Memory: Medium (8-bit quantization)
• Speed: Very fast
• Recall: Good

index_params = {
    "index_type": "IVF_SQ8",
    "metric_type": "COSINE",
    "params": {"nlist": 1024}
}

HNSW (Recommended)

• Best for: All scales
• Memory: High
• Speed: Fastest
• Recall: Excellent

index_params = {
    "index_type": "HNSW",
    "metric_type": "COSINE",
    "params": {
        "M": 16,              # Edges per node (8-64)
        "efConstruction": 256  # Build quality (40-500)
    }
}

GPU Indexes (CUDA Required)

# GPU_IVF_FLAT - Fastest with GPU
index_params = {
    "index_type": "GPU_IVF_FLAT",
    "metric_type": "COSINE",
    "params": {"nlist": 1024}
}

# GPU_IVF_PQ - GPU + Product Quantization
index_params = {
    "index_type": "GPU_IVF_PQ",
    "metric_type": "COSINE",
    "params": {"nlist": 1024, "m": 8, "nbits": 8}
}

Filtering with Scalar Fields

# Define schema with scalar fields for filtering
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=100),
    FieldSchema(name="timestamp", dtype=DataType.INT64),
    FieldSchema(name="metadata", dtype=DataType.JSON)
]

# Search with filtering
results = collection.search(
    data=[[0.1]*1024],
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr='category == "documentation" and timestamp > 1640000000000'
)

Hybrid Search (Vector + Scalar)

# Pre-filter with scalar, then vector search
results = collection.search(
    data=[[0.1]*1024],
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr='metadata["product_version"] == "v2.0"',
    output_fields=["text", "metadata"]
)

Partition Management (Multi-Tenancy)

# Create partitions for different tenants/categories
collection.create_partition("org_1")
collection.create_partition("org_2")

# Insert into specific partition
collection.insert(entities, partition_name="org_1")

# Search in specific partition
results = collection.search(
    data=[[0.1]*1024],
    anns_field="embedding",
    param=search_params,
    limit=5,
    partition_names=["org_1"]
)

Performance Optimization

Batch Operations

# Batch insert (10K-50K vectors per batch)
batch_size = 10000
for i in range(0, len(embeddings), batch_size):
    batch_embeddings = embeddings[i:i+batch_size]
    batch_texts = texts[i:i+batch_size]
    collection.insert([batch_texts, batch_embeddings])

Search Parameters Tuning

# Adjust ef for recall vs. speed trade-off
search_params = {
    "metric_type": "COSINE",
    "params": {
        "ef": 128  # Higher = better recall, slower (default: 64)
    }
}

Resource Configuration

# Configure Milvus resources
queryNode:
  replicas: 3
  resources:
    limits:
      memory: 32Gi
      cpu: 8
    requests:
      memory: 16Gi
      cpu: 4

dataNode:
  replicas: 2
  resources:
    limits:
      memory: 16Gi

Monitoring

# Get collection stats
stats = collection.get_stats()
print(f"Row count: {stats['row_count']}")

# Get query node metrics
from pymilvus import utility
metrics = utility.get_query_segment_info("documents")

Zilliz Cloud (Managed Milvus)

from pymilvus import connections

# Connect to Zilliz Cloud
connections.connect(
    alias="default",
    uri="https://your-cluster.cloud.zilliz.com:19530",
    token="your-api-key"
)

Use Cases

Billion-Scale Semantic Search

• E-commerce product search (100M+ products)
• Video similarity search (millions of videos)
• Image search engines

Recommendation Systems

• Content recommendations at Netflix scale
• Product recommendations for large catalogs
• User behavior-based recommendations

Anomaly Detection

• Security threat detection across billions of events
• Fraud detection in financial transactions
• Network intrusion detection

Production Checklist

• [ ] Deploy in cluster mode for high availability
• [ ] Configure resource limits (CPU, memory, GPU)
• [ ] Set up monitoring (Prometheus, Grafana)
• [ ] Enable authentication and authorization
• [ ] Configure backups (snapshots)
• [ ] Tune index parameters (M, efConstruction)
• [ ] Set up load balancing
• [ ] Test failover and recovery
• [ ] Monitor query performance
• [ ] Configure partitions for multi-tenancy

Additional Resources

• Official Docs: https://milvus.io/docs
• GitHub: https://github.com/milvus-io/milvus
• Zilliz Cloud: https://cloud.zilliz.com
• Performance Tuning: https://milvus.io/docs/tune.md
• Slack Community: https://milvusio.slack.com

pgvector: PostgreSQL Vector Extension

Table of Contents

• Overview
• When to Use pgvector
• Ideal For:
• Not Ideal For:
• Installation
• PostgreSQL Extension
• Docker Setup
• Managed Services
• Python Integration
• Installation
• Basic Usage
• Distance Metrics
• Indexing for Performance
• IVFFlat Index
• HNSW Index (Better Performance)
• Index Comparison
• Prisma Integration
• Schema Definition
• Usage
• Drizzle ORM Integration
• Filtering with Metadata
• JSONB Filtering
• Relational Filtering
• Batch Operations
• Bulk Insert
• Batch Search
• Hybrid Search with pg_search
• Performance Optimization
• Query Tuning
• Table Partitioning
• Materialized Views
• Monitoring and Maintenance
• Check Index Usage
• Vacuum and Analyze
• Monitor Performance
• Common Patterns
• Multi-Tenant RAG
• Versioned Documentation
• Code Search
• Migration from Other Vector DBs
• From Qdrant to pgvector
• Limitations
• Scale Limits
• Filtering Performance
• Feature Gaps
• When to Migrate Away
• Production Checklist
• Additional Resources

Overview

pgvector is an open-source PostgreSQL extension for vector similarity search. It enables vector operations within your existing PostgreSQL database without requiring additional infrastructure.

When to Use pgvector

Ideal For:

• Already using PostgreSQL - No new infrastructure required
• <10M vectors - Performance is good up to this scale
• Tight budget - Leverage existing PostgreSQL servers
• Simple use cases - Basic similarity search with limited filtering
• Relational + vector hybrid - Join vector search with relational data

Not Ideal For:

• >10M vectors - Performance degrades significantly
• Complex metadata filtering - Slower than specialized vector DBs
• High-throughput search - Limited compared to Qdrant/Milvus
• Hybrid search - No built-in BM25, requires extensions (pg_search)

Installation

PostgreSQL Extension

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

Docker Setup

# docker-compose.yml
version: '3.8'

services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
      POSTGRES_DB: vectordb
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

Managed Services

• Supabase - Built-in pgvector support
• Neon - Serverless PostgreSQL with pgvector
• AWS RDS - pgvector available on PostgreSQL 15+
• Google Cloud SQL - pgvector available
• Azure Database for PostgreSQL - pgvector available

Python Integration

Installation

pip install psycopg2-binary pgvector

Basic Usage

import psycopg2
from pgvector.psycopg2 import register_vector

# Connect
conn = psycopg2.connect(
    host="localhost",
    database="vectordb",
    user="postgres",
    password="password"
)
conn.autocommit = True

# Register vector type
register_vector(conn)

# Create table
cur = conn.cursor()
cur.execute("""
    CREATE TABLE documents (
        id SERIAL PRIMARY KEY,
        content TEXT,
        embedding vector(1024),
        metadata JSONB
    )
""")

# Insert vectors
embedding = [0.1] * 1024
cur.execute(
    "INSERT INTO documents (content, embedding, metadata) VALUES (%s, %s, %s)",
    ("Document content", embedding, {"source": "docs/api.md"})
)

# Search by similarity (L2 distance)
query_embedding = [0.1] * 1024
cur.execute(
    """
    SELECT id, content, embedding <-> %s AS distance
    FROM documents
    ORDER BY embedding <-> %s
    LIMIT 5
    """,
    (query_embedding, query_embedding)
)
results = cur.fetchall()

Distance Metrics

pgvector supports three distance metrics:

# L2 distance (Euclidean)
cur.execute(
    "SELECT * FROM documents ORDER BY embedding <-> %s LIMIT 5",
    (query_vector,)
)

# Inner product (negated dot product)
cur.execute(
    "SELECT * FROM documents ORDER BY embedding <#> %s LIMIT 5",
    (query_vector,)
)

# Cosine distance (1 - cosine similarity)
cur.execute(
    "SELECT * FROM documents ORDER BY embedding <=> %s LIMIT 5",
    (query_vector,)
)

Which to use?

• Cosine (`<=>`) - Most common for embeddings (OpenAI, Voyage, etc.)
• L2 (`<->`) - When vectors are not normalized
• Inner product (`<#>`) - For normalized vectors, equivalent to cosine

Indexing for Performance

IVFFlat Index

-- Create IVFFlat index (faster than sequential scan)
CREATE INDEX ON documents USING ivfflat (embedding vector_l2_ops)
WITH (lists = 100);

-- For cosine distance
CREATE INDEX ON documents USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

-- For inner product
CREATE INDEX ON documents USING ivfflat (embedding vector_ip_ops)
WITH (lists = 100);

Lists parameter:

• Formula: lists = rows / 1000 (for 1M rows, use lists=1000)
• Trade-off: More lists = faster search, less accurate
• Recommendation: Start with lists = sqrt(rows)

HNSW Index (Better Performance)

-- Create HNSW index (better recall than IVFFlat)
CREATE INDEX ON documents USING hnsw (embedding vector_l2_ops);

-- For cosine distance
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

-- Configure parameters
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

HNSW parameters:

• m: Edges per node (default: 16, higher = better recall, more memory)
• ef_construction: Index build quality (default: 64, higher = better index)

Index Comparison

Prisma Integration

Schema Definition

// schema.prisma
generator client {
  provider = "prisma-client-js"
  previewFeatures = ["postgresqlExtensions"]
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
  extensions = [vector]
}

model Document {
  id        Int      @id @default(autoincrement())
  content   String
  embedding Unsupported("vector(1024)")
  metadata  Json
  createdAt DateTime @default(now())

  @@map("documents")
}

Usage

import { PrismaClient } from '@prisma/client';
import { Prisma } from '@prisma/client';

const prisma = new PrismaClient();

// Insert document with embedding
await prisma.$executeRaw`
  INSERT INTO documents (content, embedding, metadata)
  VALUES (${content}, ${embedding}::vector, ${metadata}::jsonb)
`;

// Similarity search
const results = await prisma.$queryRaw<Array<{
  id: number;
  content: string;
  distance: number;
}>>`
  SELECT id, content, embedding <=> ${queryEmbedding}::vector AS distance
  FROM documents
  ORDER BY embedding <=> ${queryEmbedding}::vector
  LIMIT 5
`;

Drizzle ORM Integration

import { pgTable, serial, text, vector, jsonb } from 'drizzle-orm/pg-core';
import { drizzle } from 'drizzle-orm/node-postgres';
import { sql } from 'drizzle-orm';
import { Pool } from 'pg';

// Define schema
export const documents = pgTable('documents', {
  id: serial('id').primaryKey(),
  content: text('content').notNull(),
  embedding: vector('embedding', { dimensions: 1024 }),
  metadata: jsonb('metadata')
});

// Initialize
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const db = drizzle(pool);

// Insert
await db.insert(documents).values({
  content: 'Document content',
  embedding: Array(1024).fill(0.1),
  metadata: { source: 'docs/api.md' }
});

// Search
const results = await db.execute(sql`
  SELECT id, content, embedding <=> ${queryEmbedding}::vector AS distance
  FROM ${documents}
  ORDER BY embedding <=> ${queryEmbedding}::vector
  LIMIT 5
`);

Filtering with Metadata

JSONB Filtering

-- Filter by metadata before vector search
SELECT id, content, embedding <=> %s AS distance
FROM documents
WHERE metadata->>'source_type' = 'documentation'
  AND metadata->>'product_version' = 'v2.0'
ORDER BY embedding <=> %s
LIMIT 5;

-- Create index on JSONB for performance
CREATE INDEX idx_metadata_source ON documents
USING gin ((metadata->>'source_type'));

Relational Filtering

-- Join with relational tables
SELECT d.id, d.content, d.embedding <=> %s AS distance
FROM documents d
JOIN organizations o ON d.org_id = o.id
WHERE o.plan = 'enterprise'
  AND d.is_active = true
ORDER BY d.embedding <=> %s
LIMIT 5;

Batch Operations

Bulk Insert

from psycopg2.extras import execute_values

# Prepare batch
data = [
    (content, embedding, metadata)
    for content, embedding, metadata in chunks
]

# Bulk insert
execute_values(
    cur,
    """
    INSERT INTO documents (content, embedding, metadata)
    VALUES %s
    """,
    data,
    template="(%s, %s, %s::jsonb)"
)

Batch Search

# Search with multiple vectors
query_vectors = [embedding1, embedding2, embedding3]

for query_vector in query_vectors:
    cur.execute(
        """
        SELECT id, content, embedding <=> %s AS distance
        FROM documents
        ORDER BY embedding <=> %s
        LIMIT 5
        """,
        (query_vector, query_vector)
    )
    results = cur.fetchall()

Hybrid Search with pg_search

Install pg_search extension for BM25 keyword search:

CREATE EXTENSION IF NOT EXISTS pg_search;

-- Add full-text search column
ALTER TABLE documents ADD COLUMN content_tsv tsvector;

-- Update tsvector column
UPDATE documents
SET content_tsv = to_tsvector('english', content);

-- Create index
CREATE INDEX idx_content_tsv ON documents USING gin(content_tsv);

-- Hybrid search (combine vector + keyword)
WITH vector_results AS (
  SELECT id, content, embedding <=> %s AS distance,
         0.7 AS weight
  FROM documents
  ORDER BY embedding <=> %s
  LIMIT 20
),
keyword_results AS (
  SELECT id, content, ts_rank(content_tsv, query) AS rank,
         0.3 AS weight
  FROM documents, plainto_tsquery('OAuth refresh tokens') query
  WHERE content_tsv @@ query
  ORDER BY rank DESC
  LIMIT 20
)
SELECT DISTINCT ON (id) id, content,
       (vr.distance * vr.weight + kr.rank * kr.weight) AS score
FROM vector_results vr
FULL OUTER JOIN keyword_results kr USING (id)
ORDER BY score
LIMIT 5;

Performance Optimization

Query Tuning

-- Set search parameters (trade recall for speed)
SET ivfflat.probes = 10;  -- Default: 1, higher = better recall, slower

SET hnsw.ef_search = 40;  -- Default: 40, higher = better recall

Table Partitioning

-- Partition by organization for multi-tenant
CREATE TABLE documents (
    id SERIAL,
    org_id INTEGER NOT NULL,
    content TEXT,
    embedding vector(1024),
    metadata JSONB
) PARTITION BY LIST (org_id);

-- Create partitions
CREATE TABLE documents_org1 PARTITION OF documents FOR VALUES IN (1);
CREATE TABLE documents_org2 PARTITION OF documents FOR VALUES IN (2);

-- Create indexes per partition
CREATE INDEX ON documents_org1 USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON documents_org2 USING hnsw (embedding vector_cosine_ops);

Materialized Views

-- Pre-compute frequently accessed subsets
CREATE MATERIALIZED VIEW recent_docs AS
SELECT id, content, embedding, metadata
FROM documents
WHERE created_at > NOW() - INTERVAL '30 days';

-- Create index on materialized view
CREATE INDEX ON recent_docs USING hnsw (embedding vector_cosine_ops);

-- Refresh periodically
REFRESH MATERIALIZED VIEW recent_docs;

Monitoring and Maintenance

Check Index Usage

-- Check if indexes are being used
EXPLAIN ANALYZE
SELECT id, content, embedding <=> %s AS distance
FROM documents
ORDER BY embedding <=> %s
LIMIT 5;

Vacuum and Analyze

-- Regular maintenance
VACUUM ANALYZE documents;

-- Rebuild indexes if needed
REINDEX TABLE documents;

Monitor Performance

-- Check table size
SELECT pg_size_pretty(pg_total_relation_size('documents'));

-- Check index sizes
SELECT indexname, pg_size_pretty(pg_relation_size(indexname::regclass))
FROM pg_indexes
WHERE tablename = 'documents';

Common Patterns

Multi-Tenant RAG

-- Filter by organization
SELECT id, content, embedding <=> %s AS distance
FROM documents
WHERE org_id = %s
ORDER BY embedding <=> %s
LIMIT 5;

-- Partition by organization for better performance

Versioned Documentation

-- Filter by version
SELECT id, content, embedding <=> %s AS distance
FROM documents
WHERE metadata->>'product_version' = 'v2.0'
ORDER BY embedding <=> %s
LIMIT 5;

Code Search

-- Filter by programming language
SELECT id, content, embedding <=> %s AS distance
FROM documents
WHERE metadata->>'content_type' = 'code'
  AND metadata->>'language' = 'python'
ORDER BY embedding <=> %s
LIMIT 5;

Migration from Other Vector DBs

From Qdrant to pgvector

from qdrant_client import QdrantClient
import psycopg2
from pgvector.psycopg2 import register_vector

# Source: Qdrant
qdrant = QdrantClient("localhost", port=6333)

# Destination: PostgreSQL
conn = psycopg2.connect(database="vectordb")
register_vector(conn)
cur = conn.cursor()

# Scroll through Qdrant
offset = None
while True:
    response = qdrant.scroll(
        collection_name="documents",
        limit=1000,
        offset=offset
    )

    points = response[0]
    for point in points:
        cur.execute(
            "INSERT INTO documents (id, content, embedding, metadata) VALUES (%s, %s, %s, %s)",
            (point.id, point.payload['text'], point.vector, point.payload)
        )

    offset = response[1]
    if offset is None:
        break

conn.commit()

Limitations

Scale Limits

• Performance degrades beyond 10M vectors
• Memory requirements increase linearly with vector count
• Index build time can be slow for large datasets

Filtering Performance

• JSONB queries are slower than specialized vector DBs
• Complex filters can bypass index usage
• Pre-filtering may not leverage index

Feature Gaps

• No built-in hybrid search (requires pg_search)
• No distributed clustering (single-node only)
• Limited query optimization compared to Qdrant/Milvus

When to Migrate Away

Consider migrating to Qdrant/Milvus/Pinecone if:

• Vector count exceeds 10M
• Need complex metadata filtering at scale
• Require hybrid search (vector + BM25) out of the box
• Need distributed/clustered deployment
• Search latency becomes unacceptable (>100ms p95)

Production Checklist

• [ ] Enable pgvector extension
• [ ] Create HNSW index for production workloads
• [ ] Set up connection pooling (PgBouncer)
• [ ] Configure autovacuum settings
• [ ] Monitor index usage with EXPLAIN ANALYZE
• [ ] Set up backups (pg_dump)
• [ ] Tune PostgreSQL parameters (shared_buffers, work_mem)
• [ ] Create JSONB indexes on frequently filtered fields
• [ ] Set up monitoring (query performance, table size)
• [ ] Test failover and recovery

Additional Resources

• GitHub: https://github.com/pgvector/pgvector
• Supabase Guide: https://supabase.com/docs/guides/ai/vector-databases
• Neon Guide: https://neon.tech/docs/extensions/pgvector
• Performance Tuning: https://github.com/pgvector/pgvector#performance