LLM Vector Database Patterns in Production: Pinecone vs pgvector vs Chroma

import anthropic
 
# Anthropic embeddings (claude-3 family)
# Use for: general text, code, documentation
client = anthropic.Anthropic()
 
# Note: Anthropic uses voyage-* models for embeddings via the API
# Integration via voyageai package:
import voyageai
 
vo = voyageai.Client()
result = vo.embed(
    ["Deploy Kubernetes pods to production", "kubectl apply -f deployment.yaml"],
    model="voyage-code-2",   # Best for code/DevOps content
    input_type="document"
)
embeddings = result.embeddings  # List of 1536-dim vectors

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
 
def chunk_runbook(text: str, source: str) -> list[dict]:
    """
    Chunk a runbook intelligently — preserve section context.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,         # ~600 tokens for voyage-code-2
        chunk_overlap=200,      # Overlap prevents context loss at boundaries
        separators=[
            "\n## ",            # Split on H2 headings first
            "\n### ",           # Then H3
            "\n\n",             # Then paragraphs
            "\n",               # Then lines
            " "                 # Last resort: words
        ]
    )
 
    chunks = splitter.split_text(text)
 
    # Add metadata to each chunk
    return [
        {
            "text": chunk,
            "source": source,
            "chunk_index": i,
            "char_count": len(chunk)
        }
        for i, chunk in enumerate(chunks)
    ]
 
 
# For code files: chunk by function/class, not character count
def chunk_code(code: str, language: str) -> list[dict]:
    """Preserve function boundaries when chunking code."""
    import ast
 
    if language == "python":
        try:
            tree = ast.parse(code)
            chunks = []
            for node in ast.walk(tree):
                if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                    start = node.lineno - 1
                    end = node.end_lineno
                    chunk_text = "\n".join(code.split("\n")[start:end])
                    chunks.append({
                        "text": chunk_text,
                        "type": type(node).__name__,
                        "name": node.name
                    })
            return chunks
        except SyntaxError:
            pass
 
    # Fallback: character-based splitting
    return chunk_runbook(code, language)

import psycopg2
import numpy as np
 
# Setup
conn = psycopg2.connect("postgresql://user:pass@localhost/mydb")
with conn.cursor() as cur:
    cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
    cur.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id SERIAL PRIMARY KEY,
            content TEXT,
            source VARCHAR(500),
            embedding vector(1536),
            metadata JSONB,
            created_at TIMESTAMP DEFAULT NOW()
        )
    """)
    # Create HNSW index for fast ANN search
    cur.execute("""
        CREATE INDEX IF NOT EXISTS documents_embedding_idx
        ON documents
        USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 64)
    """)
    conn.commit()
 
 
def insert_documents(chunks: list[dict], embeddings: list[list[float]]):
    with conn.cursor() as cur:
        for chunk, embedding in zip(chunks, embeddings):
            cur.execute("""
                INSERT INTO documents (content, source, embedding, metadata)
                VALUES (%s, %s, %s, %s)
            """, (
                chunk["text"],
                chunk.get("source", ""),
                embedding,
                json.dumps({k: v for k, v in chunk.items() if k not in ("text", "source")})
            ))
    conn.commit()
 
 
def semantic_search(query_embedding: list[float], top_k: int = 5, source_filter: str = None) -> list[dict]:
    """Search with optional metadata filter — pgvector's killer feature."""
    with conn.cursor() as cur:
        if source_filter:
            cur.execute("""
                SELECT content, source, 1 - (embedding <=> %s::vector) AS similarity
                FROM documents
                WHERE source LIKE %s
                ORDER BY embedding <=> %s::vector
                LIMIT %s
            """, (query_embedding, f"%{source_filter}%", query_embedding, top_k))
        else:
            cur.execute("""
                SELECT content, source, 1 - (embedding <=> %s::vector) AS similarity
                FROM documents
                ORDER BY embedding <=> %s::vector
                LIMIT %s
            """, (query_embedding, query_embedding, top_k))
 
        rows = cur.fetchall()
        return [{"content": r[0], "source": r[1], "similarity": float(r[2])} for r in rows]

from pinecone import Pinecone, ServerlessSpec
 
 
pc = Pinecone(api_key="your-api-key")
 
# Create index (one time)
pc.create_index(
    name="devops-runbooks",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="ap-south-1")
)
 
index = pc.Index("devops-runbooks")
 
 
def upsert_to_pinecone(chunks: list[dict], embeddings: list[list[float]]):
    """Upsert in batches of 100 (Pinecone limit per batch)."""
    vectors = [
        {
            "id": f"chunk-{i}",
            "values": embedding,
            "metadata": {
                "text": chunk["text"][:1000],  # Pinecone metadata limit
                "source": chunk.get("source", ""),
                "chunk_index": chunk.get("chunk_index", 0)
            }
        }
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings))
    ]
 
    # Batch upserts
    batch_size = 100
    for i in range(0, len(vectors), batch_size):
        index.upsert(vectors=vectors[i:i+batch_size])
 
 
def query_pinecone(query_embedding: list[float], top_k: int = 5, filter: dict = None) -> list[dict]:
    results = index.query(
        vector=query_embedding,
        top_k=top_k,
        include_metadata=True,
        filter=filter  # e.g., {"source": {"$eq": "kubernetes-runbooks"}}
    )
 
    return [
        {
            "content": match.metadata.get("text", ""),
            "source": match.metadata.get("source", ""),
            "similarity": match.score
        }
        for match in results.matches
    ]

import chromadb
from chromadb.config import Settings
 
 
client = chromadb.PersistentClient(
    path="./chroma_db",
    settings=Settings(anonymized_telemetry=False)
)
 
collection = client.get_or_create_collection(
    name="devops-docs",
    metadata={"hnsw:space": "cosine"}
)
 
 
def add_to_chroma(chunks: list[dict], embeddings: list[list[float]]):
    collection.add(
        embeddings=embeddings,
        documents=[c["text"] for c in chunks],
        metadatas=[{"source": c.get("source", "")} for c in chunks],
        ids=[f"chunk-{i}" for i in range(len(chunks))]
    )
 
 
def query_chroma(query_embedding: list[float], top_k: int = 5) -> list[dict]:
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=top_k
    )
 
    return [
        {
            "content": doc,
            "source": meta.get("source", ""),
            "similarity": 1 - dist  # Chroma returns distances
        }
        for doc, meta, dist in zip(
            results["documents"][0],
            results["metadatas"][0],
            results["distances"][0]
        )
    ]