Vector Database Guide

pip install chromadb openai

import chromadb
from openai import OpenAI

client = OpenAI()
chroma = chromadb.Client()
collection = chroma.create_collection("docs")

def embed(text: str) -> list[float]:
    res = client.embeddings.create(model="text-embedding-3-small", input=text)
    return res.data[0].embedding

# Add documents
texts = ["The sky is blue.", "Python is a programming language.", "Dogs are loyal pets."]
collection.add(
    ids=[f"doc_{i}" for i in range(len(texts))],
    embeddings=[embed(t) for t in texts],
    documents=texts,
)

# Query
results = collection.query(query_embeddings=[embed("What color is the sky?")], n_results=2)
print(results["documents"])
# [['The sky is blue.', 'Dogs are loyal pets.']]

pip install pinecone-client openai

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="YOUR_API_KEY")
pc.create_index(
    name="my-index",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
index = pc.Index("my-index")

# Upsert vectors
index.upsert(vectors=[
    {"id": "doc1", "values": embed("Hello world"), "metadata": {"source": "blog"}},
    {"id": "doc2", "values": embed("Vector databases are fast"), "metadata": {"source": "docs"}},
])

# Query
results = index.query(vector=embed("fast database"), top_k=3, include_metadata=True)
for match in results["matches"]:
    print(match["id"], match["score"], match["metadata"])

docker run -p 6333:6333 qdrant/qdrant
pip install qdrant-client

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

client = QdrantClient("localhost", port=6333)
client.create_collection(
    collection_name="articles",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

client.upsert(
    collection_name="articles",
    points=[
        PointStruct(id=1, vector=embed("AI is transforming industries"), payload={"category": "tech"}),
        PointStruct(id=2, vector=embed("Recipe for chocolate cake"), payload={"category": "food"}),
    ],
)

results = client.search(
    collection_name="articles",
    query_vector=embed("machine learning applications"),
    limit=5,
)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

texts = ["Vector databases are fast.", "I love machine learning."]
embeddings = model.encode(texts)  # shape: (2, 384)

def embed_batch(texts: list[str], batch_size: int = 100) -> list[list[float]]:
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        res = openai_client.embeddings.create(model="text-embedding-3-small", input=batch)
        all_embeddings.extend([r.embedding for r in res.data])
    return all_embeddings

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # characters per chunk
    chunk_overlap=64,      # overlap to preserve context across chunks
    separators=["\n\n", "\n", ".", " "],
)

chunks = splitter.split_text(long_document)

# Pinecone
results = index.query(vector=query_embedding, top_k=10)

# Qdrant
results = client.search(collection_name="docs", query_vector=query_embedding, limit=10)

# Chroma
results = collection.query(query_embeddings=[query_embedding], n_results=10)

# Pinecone: filter during query
results = index.query(
    vector=query_embedding,
    top_k=10,
    filter={"source": {"$eq": "blog"}, "year": {"$gte": 2023}},
)

# Qdrant: filter with payload conditions
from qdrant_client.models import Filter, FieldCondition, MatchValue

results = client.search(
    collection_name="articles",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[FieldCondition(key="category", match=MatchValue(value="tech"))]
    ),
    limit=10,
)

# Weaviate hybrid search
result = (
    client.query
    .get("Article", ["title", "body"])
    .with_hybrid(query="transformer neural network", alpha=0.5)  # 0=BM25, 1=vector
    .with_limit(10)
    .do()
)

from openai import OpenAI
import chromadb

openai_client = OpenAI()
chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection("knowledge_base")

def embed(text: str) -> list[float]:
    return openai_client.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding

def index_documents(docs: list[dict]):
    """docs: list of {"id": str, "text": str, "metadata": dict}"""
    collection.upsert(
        ids=[d["id"] for d in docs],
        embeddings=[embed(d["text"]) for d in docs],
        documents=[d["text"] for d in docs],
        metadatas=[d["metadata"] for d in docs],
    )

def rag_query(question: str, k: int = 5) -> str:
    # Retrieve
    results = collection.query(query_embeddings=[embed(question)], n_results=k)
    context = "\n\n".join(results["documents"][0])

    # Generate
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": "Answer using only the provided context. If unsure, say so.",
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}",
            },
        ],
    )
    return response.choices[0].message.content

# Usage
index_documents([
    {"id": "1", "text": "MCP was created by Anthropic in 2024.", "metadata": {"source": "blog"}},
    {"id": "2", "text": "Vector databases store high-dimensional embeddings.", "metadata": {"source": "docs"}},
])

answer = rag_query("Who created MCP?")
print(answer)

res = openai_client.embeddings.create(
    model="text-embedding-3-small",
    input="Hello world",
    dimensions=256,  # reduce from 1536 → 256
)

# Qdrant scalar quantization
from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType

client.create_collection(
    collection_name="compressed",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(type=ScalarType.INT8, quantile=0.99, always_ram=True)
    ),
)

import hashlib

_cache: dict[str, list[float]] = {}

def cached_embed(text: str) -> list[float]:
    key = hashlib.sha256(text.encode()).hexdigest()
    if key not in _cache:
        _cache[key] = embed(text)
    return _cache[key]

# Use per-tenant namespaces (Pinecone) or collections (Qdrant/Chroma)
index.upsert(vectors=vectors, namespace=f"tenant_{user_id}")
results = index.query(vector=query_vec, top_k=10, namespace=f"tenant_{user_id}")

# Bad: storing sensitive data in metadata
payload = {"user_email": "alice@example.com", "ssn": "123-45-6789", "text": chunk}

# Good: store only non-sensitive, lookupable fields
payload = {"document_id": "doc_42", "source": "handbook", "section": "onboarding"}

# Sanitize retrieved chunks before inserting into prompts
import html

def safe_context(chunks: list[str]) -> str:
    sanitized = [html.escape(c) for c in chunks]
    return "\n\n---\n\n".join(sanitized)

from pathlib import Path
from openai import OpenAI
import chromadb

openai_client = OpenAI()
chroma = chromadb.PersistentClient(path="./chroma_store")
collection = chroma.get_or_create_collection("documents")

def ingest_folder(folder: str):
    for path in Path(folder).rglob("*.txt"):
        text = path.read_text()
        chunks = [text[i:i+500] for i in range(0, len(text), 450)]
        for i, chunk in enumerate(chunks):
            vec = openai_client.embeddings.create(
                model="text-embedding-3-small", input=chunk
            ).data[0].embedding
            collection.upsert(
                ids=[f"{path.stem}_{i}"],
                embeddings=[vec],
                documents=[chunk],
                metadatas=[{"file": path.name, "chunk": i}],
            )

def search(query: str, k: int = 5):
    vec = openai_client.embeddings.create(
        model="text-embedding-3-small", input=query
    ).data[0].embedding
    return collection.query(query_embeddings=[vec], n_results=k)

ingest_folder("./my_docs")
results = search("quarterly revenue targets")
for doc, meta in zip(results["documents"][0], results["metadatas"][0]):
    print(f"[{meta['file']}] {doc[:120]}...")

from PIL import Image
import torch
from transformers import CLIPProcessor, CLIPModel

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def embed_image(image_path: str) -> list[float]:
    image = Image.open(image_path)
    inputs = processor(images=image, return_tensors="pt")
    with torch.no_grad():
        features = model.get_image_features(**inputs)
    return features[0].numpy().tolist()

def embed_text_clip(text: str) -> list[float]:
    inputs = processor(text=[text], return_tensors="pt", padding=True)
    with torch.no_grad():
        features = model.get_text_features(**inputs)
    return features[0].numpy().tolist()

# Index images into Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

qdrant = QdrantClient(":memory:")
qdrant.create_collection("images", vectors_config=VectorParams(size=512, distance=Distance.COSINE))

image_paths = ["cat.jpg", "dog.jpg", "car.jpg"]
qdrant.upsert("images", points=[
    PointStruct(id=i, vector=embed_image(p), payload={"path": p})
    for i, p in enumerate(image_paths)
])

# Text-to-image search
results = qdrant.search("images", query_vector=embed_text_clip("a fluffy animal"), limit=3)
for r in results:
    print(r.payload["path"], r.score)

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("all-MiniLM-L6-v2")

records = [
    "Customer complaint: product arrived broken",
    "Feedback: item was damaged on delivery",
    "New feature request: dark mode support",
    "Bug report: app crashes on startup",
    "Issue: received a broken product",
]

embeddings = model.encode(records, normalize_embeddings=True)
similarity_matrix = np.dot(embeddings, embeddings.T)

threshold = 0.85
for i in range(len(records)):
    for j in range(i + 1, len(records)):
        if similarity_matrix[i][j] >= threshold:
            score = float(similarity_matrix[i][j])
            print(f"Duplicate (score={score:.2f}):")
            print(f"  A: {records[i]}")
            print(f"  B: {records[j]}\n")