Vector Databases — Embeddings, Similarity Search, Indexing & RAG Pipelines

# Generate embeddings with OpenAI
from openai import OpenAI
client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Vector databases store semantic meaning as mathematical vectors"
)
vector = response.data[0].embedding
print(f"Dimension: {len(vector)}")
print(f"First 5 values: {vector[:5]}")

from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="your-api-key")

# Serverless index
pc.create_index(
    name="rag-index",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

index = pc.Index("rag-index")

# Upsert vectors
index.upsert([
    ("id1", [0.1, 0.2, ...], {"text": "Document about AI"}),
    ("id2", [0.3, 0.4, ...], {"text": "Document about databases"}),
])

# Query
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=5,
    include_metadata=True
)
print(results.matches[0].metadata["text"])

import weaviate
import weaviate.classes as wvc

client = weaviate.connect_to_local()

# Create collection with auto-schema
collection = client.collections.create(
    name="Documents",
    vectorizer_config=wvc.config.Configure.Vectorizer.none(),
    properties=[
        wvc.config.Property(name="title", data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="content", data_type=wvc.config.DataType.TEXT),
        wvc.config.Property(name="category", data_type=wvc.config.DataType.TEXT),
    ]
)

# Insert with pre-generated vectors
collection.data.insert(
    properties={"title": "AI Overview", "content": "...", "category": "tech"},
    vector=[0.1, 0.2, ...]
)

# Hybrid search (vector + keyword)
response = collection.query.hybrid(
    query="artificial intelligence",
    alpha=0.75,  # 75% vector, 25% keyword
    limit=5
)

import chromadb

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

collection.add(
    documents=["Document about AI", "Document about databases"],
    metadatas=[{"category": "tech"}, {"category": "tech"}],
    ids=["doc1", "doc2"]
)

results = collection.query(
    query_texts=["machine learning"],
    n_results=2
)
print(results["documents"])

from qdrant_client import QdrantClient, models

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

client.create_collection(
    collection_name="products",
    vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE),
)

client.upsert(
    collection_name="products",
    points=[
        models.PointStruct(id=1, vector=[0.1, 0.2, ...],
            payload={"name": "Laptop", "price": 999.99, "category": "electronics"}),
    ]
)

# Query with metadata filter
client.search(
    collection_name="products",
    query_vector=[0.15, 0.25, ...],
    query_filter=models.Filter(
        must=[
            models.FieldCondition(key="category", match=models.MatchValue(value="electronics")),
            models.FieldCondition(key="price", range=models.Range(lt=1500)),
        ]
    ),
    limit=5
)

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

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

schema = CollectionSchema([
    FieldSchema("id", DataType.INT64, is_primary=True),
    FieldSchema("embedding", DataType.FLOAT_VECTOR, dim=1536),
    FieldSchema("title", DataType.VARCHAR, max_length=200),
])
collection = Collection("documents", schema)

# Create IVF_FLAT index
collection.create_index("embedding", {
    "index_type": "IVF_FLAT",
    "metric_type": "COSINE",
    "params": {"nlist": 256}
})

collection.load()
results = collection.search(
    data=[[0.1, 0.2, ...]],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"nprobe": 10}},
    limit=5,
)

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    return np.sqrt(np.sum((a - b) ** 2))

vec1 = np.array([0.1, 0.2, 0.3])
vec2 = np.array([0.15, 0.25, 0.35])
vec3 = np.array([0.9, 0.8, 0.7])

print(f"Cosine(vec1, vec2) = {cosine_similarity(vec1, vec2):.4f}")
print(f"Cosine(vec1, vec3) = {cosine_similarity(vec1, vec3):.4f}")

# HNSW configuration in Pinecone
index = pc.Index("hnsw-index")
# HNSW parameters (Pinecone handles this automatically)
# M = 16      - connections per node
# ef_construction = 200 - build quality vs speed
# ef_search = 100      - query accuracy vs speed

# IVF configuration in Milvus
index_params = {
    "index_type": "IVF_FLAT",
    "params": {"nlist": 256},  # Number of clusters
}
# nprobe = 10 (number of clusters to search at query time)

# Complete RAG pipeline
from openai import OpenAI
import chromadb

client = OpenAI()
vector_db = chromadb.Client().get_or_create_collection("knowledge_base")

# Step 1: Index documents
def index_document(doc_id, text, metadata=None):
    embedding = client.embeddings.create(
        model="text-embedding-3-small", input=text
    ).data[0].embedding
    
    vector_db.add(
        embeddings=[embedding],
        documents=[text],
        metadatas=[metadata or {}],
        ids=[doc_id]
    )

# Step 2: Retrieve
def retrieve(query, k=3):
    query_emb = client.embeddings.create(
        model="text-embedding-3-small", input=query
    ).data[0].embedding
    
    results = vector_db.query(query_embeddings=[query_emb], n_results=k)
    return results["documents"][0]

# Step 3: Generate
def rag_answer(query):
    context = retrieve(query)
    context_text = "\n\n".join(context)
    
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Answer using the provided context."},
            {"role": "user", "content": f"Context:\n{context_text}\n\nQuestion: {query}"}
        ]
    )
    return response.choices[0].message.content

# Example
index_document("1", "Vector databases store embeddings for similarity search")
print(rag_answer("What do vector databases store?"))

import chromadb
from sentence_transformers import SentenceTransformer

# Use a local embedding model
model = SentenceTransformer("all-MiniLM-L6-v2")
client = chromadb.Client()
collection = client.create_collection("demo")

documents = [
    "Python is a programming language",
    "Vector databases enable semantic search",
    "Neural networks learn from data",
    "Databases store structured information",
]

# Embed and index
embeddings = model.encode(documents).tolist()
collection.add(
    embeddings=embeddings,
    documents=documents,
    ids=[f"doc{i}" for i in range(len(documents))]
)

# Search
query = "machine learning"
query_emb = model.encode([query]).tolist()
results = collection.query(query_embeddings=query_emb, n_results=2)
print("Query:", query)
print("Results:", results["documents"][0])