similarity-search-patterns

name	similarity-search-patterns
description	Implement efficient similarity search with vector databases. Use when building semantic search, implementing nearest neighbor queries, or optimizing retrieval performance.

Similarity Search Patterns

Patterns for implementing efficient similarity search in production systems.

When to Use This Skill

• Building semantic search systems
• Implementing RAG retrieval
• Creating recommendation engines
• Optimizing search latency
• Scaling to millions of vectors
• Combining semantic and keyword search

Core Concepts

1. Distance Metrics

Metric	Formula	Best For
Cosine	1 - (A·B)/(‖A‖‖B‖)	Normalized embeddings
Euclidean (L2)	√Σ(a-b)²	Raw embeddings
Dot Product	A·B	Magnitude matters
Manhattan (L1)	Σ	a-b

2. Index Types

┌─────────────────────────────────────────────────┐
│                 Index Types                      │
├─────────────┬───────────────┬───────────────────┤
│    Flat     │     HNSW      │    IVF+PQ         │
│ (Exact)     │ (Graph-based) │ (Quantized)       │
├─────────────┼───────────────┼───────────────────┤
│ O(n) search │ O(log n)      │ O(√n)             │
│ 100% recall │ ~95-99%       │ ~90-95%           │
│ Small data  │ Medium-Large  │ Very Large        │
└─────────────┴───────────────┴───────────────────┘

Templates

Template 1: Pinecone Implementation

from pinecone import Pinecone, ServerlessSpec
from typing import List, Dict, Optional
import hashlib

class PineconeVectorStore:
    def __init__(
        self,
        api_key: str,
        index_name: str,
        dimension: int = 1536,
        metric: str = "cosine"
    ):
        self.pc = Pinecone(api_key=api_key)

        # Create index if not exists
        if index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=index_name,
                dimension=dimension,
                metric=metric,
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )

        self.index = self.pc.Index(index_name)

    def upsert(
        self,
        vectors: List[Dict],
        namespace: str = ""
    ) -> int:
        """
        Upsert vectors.
        vectors: [{"id": str, "values": List[float], "metadata": dict}]
        """
        # Batch upsert
        batch_size = 100
        total = 0

        for i in range(0, len(vectors), batch_size):
            batch = vectors[i:i + batch_size]
            self.index.upsert(vectors=batch, namespace=namespace)
            total += len(batch)

        return total

    def search(
        self,
        query_vector: List[float],
        top_k: int = 10,
        namespace: str = "",
        filter: Optional[Dict] = None,
        include_metadata: bool = True
    ) -> List[Dict]:
        """Search for similar vectors."""
        results = self.index.query(
            vector=query_vector,
            top_k=top_k,
            namespace=namespace,
            filter=filter,
            include_metadata=include_metadata
        )

        return [
            {
                "id": match.id,
                "score": match.score,
                "metadata": match.metadata
            }
            for match in results.matches
        ]

    def search_with_rerank(
        self,
        query: str,
        query_vector: List[float],
        top_k: int = 10,
        rerank_top_n: int = 50,
        namespace: str = ""
    ) -> List[Dict]:
        """Search and rerank results."""
        # Over-fetch for reranking
        initial_results = self.search(
            query_vector,
            top_k=rerank_top_n,
            namespace=namespace
        )

        # Rerank with cross-encoder or LLM
        reranked = self._rerank(query, initial_results)

        return reranked[:top_k]

    def _rerank(self, query: str, results: List[Dict]) -> List[Dict]:
        """Rerank results using cross-encoder."""
        from sentence_transformers import CrossEncoder

        model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

        pairs = [(query, r["metadata"]["text"]) for r in results]
        scores = model.predict(pairs)

        for result, score in zip(results, scores):
            result["rerank_score"] = float(score)

        return sorted(results, key=lambda x: x["rerank_score"], reverse=True)

    def delete(self, ids: List[str], namespace: str = ""):
        """Delete vectors by ID."""
        self.index.delete(ids=ids, namespace=namespace)

    def delete_by_filter(self, filter: Dict, namespace: str = ""):
        """Delete vectors matching filter."""
        self.index.delete(filter=filter, namespace=namespace)

Template 2: Qdrant Implementation

from qdrant_client import QdrantClient
from qdrant_client.http import models
from typing import List, Dict, Optional

class QdrantVectorStore:
    def __init__(
        self,
        url: str = "localhost",
        port: int = 6333,
        collection_name: str = "documents",
        vector_size: int = 1536
    ):
        self.client = QdrantClient(url=url, port=port)
        self.collection_name = collection_name

        # Create collection if not exists
        collections = self.client.get_collections().collections
        if collection_name not in [c.name for c in collections]:
            self.client.create_collection(
                collection_name=collection_name,
                vectors_config=models.VectorParams(
                    size=vector_size,
                    distance=models.Distance.COSINE
                ),
                # Optional: enable quantization for memory efficiency
                quantization_config=models.ScalarQuantization(
                    scalar=models.ScalarQuantizationConfig(
                        type=models.ScalarType.INT8,
                        quantile=0.99,
                        always_ram=True
                    )
                )
            )

    def upsert(self, points: []) -> :
        
        qdrant_points = [
            models.PointStruct(
                =p[],
                vector=p[],
                payload=p.get(, {})
            )
             p  points
        ]

        .client.upsert(
            collection_name=.collection_name,
            points=qdrant_points
        )
         (points)

     () -> []:
        
        results = .client.search(
            collection_name=.collection_name,
            query_vector=query_vector,
            limit=limit,
            query_filter=,
            score_threshold=score_threshold
        )

         [
            {
                : r.,
                : r.score,
                : r.payload
            }
             r  results
        ]

     () -> []:
        
        conditions = []

         must_conditions:
            conditions.extend([
                models.FieldCondition(
                    key=c[],
                    =models.MatchValue(value=c[])
                )
                 c  must_conditions
            ])

         = models.Filter(must=conditions)  conditions  

         .search(query_vector, limit=limit, =)

     () -> []:
        
        
        results = .client.search(
            collection_name=.collection_name,
            query_vector=models.NamedVector(
                name=,
                vector=dense_vector
            ),
            limit=limit
        )
         [{: r., : r.score, : r.payload}  r  results]

Template 3: pgvector with PostgreSQL

import asyncpg
from typing import List, Dict, Optional
import numpy as np

class PgVectorStore:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string

    async def init(self):
        """Initialize connection pool and extension."""
        self.pool = await asyncpg.create_pool(self.connection_string)

        async with self.pool.acquire() as conn:
            # Enable extension
            await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")

            # Create table
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS documents (
                    id TEXT PRIMARY KEY,
                    content TEXT,
                    metadata JSONB,
                    embedding vector(1536)
                )
            """)

            # Create index (HNSW for better performance)
            await conn.execute("""
                CREATE INDEX IF NOT EXISTS documents_embedding_idx
                ON documents
                USING hnsw (embedding vector_cosine_ops)
                WITH (m = 16, ef_construction = 64)
            """)

    async def upsert(self, documents: List[]):
        
          .pool.acquire()  conn:
             conn.executemany(
                ,
                [
                    (
                        doc[],
                        doc[],
                        doc.get(, {}),
                        np.array(doc[]).tolist()
                    )
                     doc  documents
                ]
            )

      () -> []:
        
        query = 

        params = [query_embedding]

         filter_metadata:
            conditions = []
             key, value  filter_metadata.items():
                params.append(value)
                conditions.append()
            query +=  + .join(conditions)

        query += 
        params.append(limit)

          .pool.acquire()  conn:
            rows =  conn.fetch(query, *params)

         [
            {
                : row[],
                : row[],
                : row[],
                : row[]
            }
             row  rows
        ]

      () -> []:
        
          .pool.acquire()  conn:
            rows =  conn.fetch(
                ,
                query_embedding, query_text, limit, vector_weight
            )

         [(row)  row  rows]

Template 4: Weaviate Implementation

import weaviate
from weaviate.util import generate_uuid5
from typing import List, Dict, Optional

class WeaviateVectorStore:
    def __init__(
        self,
        url: str = "http://localhost:8080",
        class_name: str = "Document"
    ):
        self.client = weaviate.Client(url=url)
        self.class_name = class_name
        self._ensure_schema()

    def _ensure_schema(self):
        """Create schema if not exists."""
        schema = {
            "class": self.class_name,
            "vectorizer": "none",  # We provide vectors
            "properties": [
                {"name": "content", "dataType": ["text"]},
                {"name": "source", "dataType": ["string"]},
                {"name": "chunk_id", "dataType": ["int"]}
            ]
        }

        if not self.client.schema.exists(self.class_name):
            self.client.schema.create_class(schema)

     ():
        
         .client.batch  batch:
            batch.batch_size = 

             doc  documents:
                batch.add_data_object(
                    data_object={
                        : doc[],
                        : doc.get(, ),
                        : doc.get(, )
                    },
                    class_name=.class_name,
                    uuid=generate_uuid5(doc[]),
                    vector=doc[]
                )

     () -> []:
        
        query = (
            .client.query
            .get(.class_name, [, , ])
            .with_near_vector({: query_vector})
            .with_limit(limit)
            .with_additional([, ])
        )

         where_filter:
            query = query.with_where(where_filter)

        results = query.do()

         [
            {
                : item[][],
                : item[],
                : item[],
                :  - item[][]
            }
             item  results[][][.class_name]
        ]

     () -> []:
        
        results = (
            .client.query
            .get(.class_name, [, ])
            .with_hybrid(query=query, vector=query_vector, alpha=alpha)
            .with_limit(limit)
            .with_additional([])
            .do()
        )

         [
            {
                : item[],
                : item[],
                : item[][]
            }
             item  results[][][.class_name]
        ]

Best Practices

Do's

• Use appropriate index - HNSW for most cases
• Tune parameters - ef_search, nprobe for recall/speed
• Implement hybrid search - Combine with keyword search
• Monitor recall - Measure search quality
• Pre-filter when possible - Reduce search space

Don'ts

• Don't skip evaluation - Measure before optimizing
• Don't over-index - Start with flat, scale up
• Don't ignore latency - P99 matters for UX
• Don't forget costs - Vector storage adds up

Resources

• Pinecone Docs
• Qdrant Docs
• pgvector
• Weaviate Docs