Vector search powers the semantic layer of modern AI applications — from RAG pipelines and recommendation engines to image similarity and anomaly detection. Unlike keyword search, which matches on exact tokens, vector search finds results based on meaning by comparing high-dimensional embeddings using distance metrics such as cosine similarity or dot product.
This guide covers the full stack: generating embeddings, choosing and operating a vector database, building a query pipeline with reranking, and scaling to production.
Vector Search Architecture
A production vector search system has four stages. Text (or another modality) is encoded into a dense vector by an embedding model. Those vectors are stored in a purpose-built vector database that maintains an approximate nearest-neighbor (ANN) index. At query time, the user’s question is embedded with the same model, and the ANN index returns the top-K most similar stored vectors in milliseconds.
flowchart LR
A[Raw Documents] --> B[Embedding Model]
B --> C[(Vector Database\nANN Index)]
Q[User Query] --> B2[Same Embedding Model]
B2 --> D{ANN Search\nTop-K}
C --> D
D --> E[Ranked Results]
The critical constraint is that query and document embeddings must come from the same model. Switching models requires re-embedding your entire corpus.
Embedding Model Choices
| Model | Dimensions | Best For |
|---|---|---|
text-embedding-3-small (OpenAI) |
1536 | General English text, low latency |
text-embedding-3-large (OpenAI) |
3072 | Higher accuracy, multilingual |
embed-multilingual-v3 (Cohere) |
1024 | Multilingual corpora |
BAAI/bge-base-en-v1.5 (OSS) |
768 | Self-hosted, no API cost |
BAAI/bge-m3 (OSS) |
1024 | Multilingual self-hosted |
For cost-sensitive production workloads, open-source models via sentence-transformers run comfortably on a single GPU and match OpenAI’s smaller models on most benchmarks.
Embedding Generation
The embedding layer is a thin wrapper around your chosen model. The key design decisions are batching (critical for throughput) and how you combine multiple text fields before encoding. Concatenating title and body into a single string typically outperforms embedding them separately.
#!/usr/bin/env python3
"""Embedding generation with batching support."""
from openai import OpenAI
from typing import List, Dict
class EmbeddingGenerator:
"""Generate text embeddings via OpenAI or a drop-in compatible API."""
def __init__(self, model: str = "text-embedding-3-small"):
self.client = OpenAI()
self.model = model
def embed_text(self, text: str) -> List[float]:
"""Embed a single string."""
response = self.client.embeddings.create(model=self.model, input=text)
return response.data[0].embedding
def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[List[float]]:
"""Embed a list of strings in batches to stay within API limits."""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i : i + batch_size]
response = self.client.embeddings.create(model=self.model, input=batch)
all_embeddings.extend(item.embedding for item in response.data)
return all_embeddings
def embed_documents(self, documents: List[Dict]) -> List[Dict]:
"""Embed a list of documents, concatenating title + content."""
texts = [f"{d.get('title', '')} {d.get('content', '')}" for d in documents]
embeddings = self.embed_batch(texts)
return [
{"id": d["id"], "embedding": emb, "text": texts[i], "metadata": {k: v for k, v in d.items() if k not in ("id", "content")}}
for i, (d, emb) in enumerate(zip(documents, embeddings))
]
For self-hosted deployments, swap EmbeddingGenerator for the sentence-transformers variant below. The interface is identical, so the rest of the pipeline requires no changes.
from sentence_transformers import SentenceTransformer
from typing import List
class OpenSourceEmbedder:
"""Drop-in embedder using sentence-transformers (no API key required)."""
def __init__(self, model_name: str = "BAAI/bge-base-en-v1.5"):
self.model = SentenceTransformer(model_name)
def embed_text(self, text: str) -> List[float]:
return self.model.encode(text).tolist()
def embed_batch(self, texts: List[str]) -> List[List[float]]:
return self.model.encode(texts, batch_size=64, show_progress_bar=True).tolist()
Vector Databases
Once you have embeddings, you need a store that can run ANN queries at low latency. The three most widely deployed options are Pinecone (managed SaaS), Milvus (self-hosted or cloud), and Weaviate (self-hosted or cloud, with built-in vectorization). Choose based on your ops maturity and data residency requirements.
Pinecone
Pinecone is fully managed — no infrastructure to run. You create a serverless index, upsert vectors with metadata, and query with optional metadata filters. This makes it ideal for teams that want to ship quickly without managing clusters.
#!/usr/bin/env python3
"""Pinecone vector store operations."""
from pinecone import Pinecone, ServerlessSpec
from typing import List, Dict, Optional
class PineconeVectorStore:
"""Thin wrapper around the Pinecone client."""
def __init__(self, api_key: str, index_name: str, dimension: int = 1536):
self.client = Pinecone(api_key=api_key)
self.index_name = index_name
self._ensure_index(dimension)
self.index = self.client.Index(index_name)
def _ensure_index(self, dimension: int):
if self.index_name not in self.client.list_indexes().names():
self.client.create_index(
name=self.index_name,
dimension=dimension,
metric="cosine",
spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
def upsert(self, vectors: List[Dict]):
"""Upsert a list of {id, embedding, metadata} dicts."""
self.index.upsert(
vectors=[{"id": v["id"], "values": v["embedding"], "metadata": v.get("metadata", {})} for v in vectors]
)
def search(self, query_vector: List[float], top_k: int = 10, filter_dict: Optional[Dict] = None) -> List[Dict]:
"""Return top-K results, optionally filtered by metadata."""
results = self.index.query(vector=query_vector, top_k=top_k, filter=filter_dict, include_metadata=True, include_values=False)
return [{"id": m["id"], "score": m["score"], "metadata": m.get("metadata", {})} for m in results["matches"]]
Milvus
Milvus is the go-to choice for on-premises deployments or when you need fine-grained control over indexing parameters. The example below uses IVF_FLAT, which is a good baseline; for billion-scale datasets consider HNSW or IVF_PQ.
#!/usr/bin/env python3
"""Milvus vector store with IVF_FLAT index."""
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
from typing import List, Dict
class MilvusVectorStore:
"""Milvus collection wrapper."""
def __init__(self, collection_name: str, dimension: int = 1536, host: str = "localhost", port: int = 19530):
connections.connect(host=host, port=port)
self.collection = self._get_or_create(collection_name, dimension)
def _get_or_create(self, name: str, dimension: int) -> Collection:
from pymilvus import utility
if utility.has_collection(name):
return Collection(name)
schema = CollectionSchema([
FieldSchema("id", DataType.INT64, is_primary=True, auto_id=False),
FieldSchema("vector", DataType.FLOAT_VECTOR, dim=dimension),
FieldSchema("text", DataType.VARCHAR, max_length=65535),
])
col = Collection(name, schema)
col.create_index("vector", {"metric_type": "IP", "index_type": "IVF_FLAT", "params": {"nlist": 128}})
return col
def insert(self, vectors: List[Dict]):
self.collection.insert([
[v["id"] for v in vectors],
[v["embedding"] for v in vectors],
[v["text"][:65535] for v in vectors],
])
self.collection.flush()
def search(self, query_vector: List[float], top_k: int = 10) -> List[Dict]:
self.collection.load()
hits = self.collection.search(
data=[query_vector], anns_field="vector",
param={"metric_type": "IP", "params": {"nprobe": 10}},
limit=top_k, output_fields=["text"],
)
return [{"id": h.id, "score": h.distance, "text": h.entity.get("text")} for h in hits[0]]
Query Pipeline with Reranking
A common production pattern is retrieve-then-rerank: fetch a larger candidate set (top-20 or top-50) from the ANN index cheaply, then run a cross-encoder to reorder by true relevance. The cross-encoder is slower but much more accurate because it attends to both query and document together.
flowchart TD
Q[Query] --> E[Embed Query]
E --> ANN[ANN Search\ntop-20 candidates]
ANN --> CE[Cross-Encoder\nRerank]
CE --> R[Return top-5]
#!/usr/bin/env python3
"""Semantic search with optional cross-encoder reranking."""
from sentence_transformers import CrossEncoder
from typing import List, Dict
class SemanticSearchApp:
"""Combines an embedder and a vector store into a search interface."""
def __init__(self, embedder, vector_store):
self.embedder = embedder
self.store = vector_store
self._reranker = None
def index(self, documents: List[Dict]) -> int:
embedded = self.embedder.embed_documents(documents)
self.store.upsert(embedded)
return len(embedded)
def search(self, query: str, top_k: int = 5) -> List[Dict]:
return self.store.search(self.embedder.embed_text(query), top_k=top_k)
def search_with_rerank(self, query: str, candidates: int = 20, top_k: int = 5) -> List[Dict]:
"""Retrieve a wider candidate set, then rerank with a cross-encoder."""
if self._reranker is None:
self._reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
results = self.search(query, top_k=candidates)
scores = self._reranker.predict([(query, r["text"]) for r in results])
for r, score in zip(results, scores):
r["rerank_score"] = float(score)
return sorted(results, key=lambda x: x["rerank_score"], reverse=True)[:top_k]
Scaling to Production
Single-node vector databases handle tens of millions of vectors comfortably. Beyond that, you need horizontal sharding. The simplest strategy is consistent hashing: assign each document to a shard deterministically by ID, then fan out queries to all shards and merge the results.
#!/usr/bin/env python3
"""Fan-out search across multiple vector store shards."""
from typing import List, Dict
def federated_search(query_vector: List[float], shards: list, top_k: int = 10) -> List[Dict]:
"""Query all shards in parallel and return the global top-K."""
import concurrent.futures
all_results: List[Dict] = []
with concurrent.futures.ThreadPoolExecutor() as pool:
futures = [pool.submit(shard.search, query_vector, top_k) for shard in shards]
for f in concurrent.futures.as_completed(futures):
all_results.extend(f.result())
# Deduplicate by ID and keep highest score
seen: Dict[str, Dict] = {}
for r in all_results:
if r["id"] not in seen or r["score"] > seen[r["id"]]["score"]:
seen[r["id"]] = r
return sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:top_k]
Deployment Checklist
Before going to production, verify these points:
- Index warm-up — Milvus and Weaviate require
collection.load()before the first query. Add a readiness probe to your service that performs a single search. - Batch ingestion throughput — Pinecone Serverless handles ~100 upserts/s per namespace; use parallel workers for bulk loads.
- Metadata filtering cardinality — High-cardinality filters (e.g., per-user namespaces) should use separate namespaces or collections, not metadata fields, to avoid index bloat.
- Embedding model versioning — Pin the exact model version. OpenAI occasionally releases updated embedding models whose outputs are incompatible with older vectors.
- Monitoring — Track p99 search latency, recall (via offline evaluation), and index size growth. Milvus exposes Prometheus metrics out of the box.
Related Articles
- Building Production LLM Applications
- Vector Databases: Pinecone, Milvus, Weaviate
- Hybrid Search RAG Complete Guide
Comments