Vector DB Patterns

Vector Database Patterns

Qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue

client = QdrantClient(url="http://localhost:6333")

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

# Upsert vectors
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=i,
            vector=embedding.tolist(),
            payload={"text": chunk, "source": filename, "page": page_num}
        )
        for i, (embedding, chunk, filename, page_num) in enumerate(data)
    ]
)

# Search with metadata filter
results = client.search(
    collection_name="documents",
    query_vector=query_embedding.tolist(),
    query_filter=Filter(
        must=[FieldCondition(key="source", match=MatchValue(value="policy.pdf"))]
    ),
    limit=5,
    with_payload=True,
)

for r in results:
    print(f"score={r.score:.3f} text={r.payload['text'][:100]}")

Pinecone

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="YOUR_API_KEY")

# Create index
pc.create_index(
    name="documents",
    dimension=1536,  # text-embedding-3-small
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)

index = pc.Index("documents")

# Upsert in batches
batch_size = 100
vectors = [
    (f"doc-{i}", embedding, {"text": chunk, "category": category})
    for i, (embedding, chunk, category) in enumerate(data)
]

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

# Query with namespace isolation
results = index.query(
    vector=query_embedding,
    top_k=10,
    namespace="production",
    filter={"category": {"$in": ["policy", "procedure"]}},
    include_metadata=True,
)

Chroma (local/embedded)

import chromadb
from chromadb.utils import embedding_functions

# Persistent local store
client = chromadb.PersistentClient(path="./chroma_db")

ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-small-en-v1.5"
)

collection = client.get_or_create_collection(
    name="documents",
    embedding_function=ef,
    metadata={"hnsw:space": "cosine"},
)

# Add documents (auto-embeds)
collection.add(
    documents=chunks,
    metadatas=[{"source": s, "page": p} for s, p in sources],
    ids=[f"chunk-{i}" for i in range(len(chunks))],
)

# Query
results = collection.query(
    query_texts=["What is the refund policy?"],
    n_results=5,
    where={"source": "terms.pdf"},
    include=["documents", "metadatas", "distances"],
)

Weaviate

import weaviate
from weaviate.classes.config import Configure, Property, DataType

client = weaviate.connect_to_local()

# Schema
client.collections.create(
    name="Document",
    vectorizer_config=Configure.Vectorizer.text2vec_transformers(),
    properties=[
        Property(name="text", data_type=DataType.TEXT),
        Property(name="source", data_type=DataType.TEXT),
        Property(name="category", data_type=DataType.TEXT),
    ],
)

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

# Insert
with collection.batch.dynamic() as batch:
    for chunk, source, category in data:
        batch.add_object({"text": chunk, "source": source, "category": category})

# Hybrid search (BM25 + vector)
results = collection.query.hybrid(
    query="refund policy terms",
    alpha=0.5,       # 0=BM25 only, 1=vector only
    limit=5,
    filters=weaviate.classes.query.Filter.by_property("category").equal("policy"),
)

Embedding Best Practices

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("BAAI/bge-base-en-v1.5")

# Batch encode efficiently
def embed_chunks(texts: list[str], batch_size: int = 64) -> np.ndarray:
    embeddings = model.encode(
        texts,
        batch_size=batch_size,
        show_progress_bar=True,
        normalize_embeddings=True,  # For cosine similarity
        convert_to_numpy=True,
    )
    return embeddings

# Add query prefix for BGE models
query = "What is the return policy?"
query_embedding = model.encode(f"Represent this sentence for searching relevant passages: {query}")

# Matryoshka embeddings — truncate for speed
full_embedding = model.encode(text)  # 768-dim
small_embedding = full_embedding[:256]  # Truncate, still valid with Matryoshka training
small_embedding = small_embedding / np.linalg.norm(small_embedding)