Vector Databases Compared: Pinecone vs Chroma vs Weaviate for AI Agents

import chromadb
from chromadb.utils import embedding_functions

# In-memory client for testing (no persistence)
client = chromadb.Client()

# Persistent client for development
client = chromadb.PersistentClient(path="./chroma_db")

# Use sentence-transformers for embeddings (local, free)
embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="all-MiniLM-L6-v2"
)

collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=embedding_fn
)

# Add documents
documents = [
    "LangGraph is a framework for building stateful agent workflows.",
    "Pinecone is a managed vector database with serverless scaling.",
    "RAG systems combine retrieval with language model generation.",
]

collection.add(
    documents=documents,
    ids=["doc1", "doc2", "doc3"],
    metadatas=[{"source": "tech_docs"} for _ in documents]
)

# Semantic query
results = collection.query(
    query_texts=["how do I scale my vector search?"],
    n_results=2,
    where={"source": "tech_docs"}  # metadata filtering
)

for doc, distance in zip(results["documents"][0], results["distances"][0]):
    print(f"Distance: {distance:.4f} | {doc[:80]}...")

from pinecone import Pinecone, ServerlessSpec
from sentence_transformers import SentenceTransformer
import time

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

index_name = "ai-agent-knowledge"

# Create serverless index if it doesn't exist
if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=384,  # all-MiniLM-L6-v2 output dimension
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )
    # Wait for index to be ready
    while not pc.describe_index(index_name).status["ready"]:
        time.sleep(1)

index = pc.Index(index_name)
model = SentenceTransformer("all-MiniLM-L6-v2")

# Upsert vectors
documents = [
    {"id": "doc1", "text": "Agent memory enables persistent context across sessions.", "category": "memory"},
    {"id": "doc2", "text": "Tool calling allows LLMs to interact with external systems.", "category": "tools"},
    {"id": "doc3", "text": "Serverless vector databases scale to zero when idle.", "category": "infrastructure"},
]

vectors = []
for doc in documents:
    embedding = model.encode(doc["text"]).tolist()
    vectors.append({
        "id": doc["id"],
        "values": embedding,
        "metadata": {"text": doc["text"], "category": doc["category"]}
    })

index.upsert(vectors=vectors)

# Query with metadata filtering
query_embedding = model.encode("how do AI agents remember things?").tolist()
results = index.query(
    vector=query_embedding,
    top_k=2,
    filter={"category": {"$in": ["memory", "tools"]}},
    include_metadata=True
)

for match in results["matches"]:
    print(f"Score: {match['score']:.4f} | {match['metadata']['text']}")

import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
from weaviate.classes.query import MetadataQuery

# Connect to Weaviate Cloud (or local Docker instance)
client = weaviate.connect_to_weaviate_cloud(
    cluster_url="https://your-cluster.weaviate.network",
    auth_credentials=Auth.api_key("your-weaviate-api-key")
)

# Define schema with vectorizer
if not client.collections.exists("TechArticle"):
    client.collections.create(
        name="TechArticle",
        vectorizer_config=Configure.Vectorizer.text2vec_openai(),  # or text2vec_huggingface
        properties=[
            Property(name="title", data_type=DataType.TEXT),
            Property(name="content", data_type=DataType.TEXT),
            Property(name="category", data_type=DataType.TEXT),
        ]
    )

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

# Insert objects (Weaviate handles vectorization automatically)
with collection.batch.dynamic() as batch:
    batch.add_object(properties={
        "title": "Introduction to LangGraph",
        "content": "LangGraph enables building stateful multi-agent workflows as directed graphs.",
        "category": "frameworks"
    })
    batch.add_object(properties={
        "title": "RAG Pipeline Design Patterns",
        "content": "Effective RAG systems balance retrieval precision with generation quality.",
        "category": "architecture"
    })

# Hybrid search: combine vector similarity with keyword matching
results = collection.query.hybrid(
    query="stateful agent workflows",
    alpha=0.75,  # 0 = pure BM25, 1 = pure vector
    limit=3,
    return_metadata=MetadataQuery(score=True)
)

for obj in results.objects:
    print(f"Score: {obj.metadata.score:.4f} | {obj.properties['title']}")

client.close()

from langchain_chroma import Chroma
from langchain_pinecone import PineconeVectorStore
from langchain_weaviate import WeaviateVectorStore
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# Swap any of these into the same retriever interface:
vectorstore = Chroma(persist_directory="./db", embedding_function=embeddings)
# vectorstore = PineconeVectorStore(index_name="my-index", embedding=embeddings)
# vectorstore = WeaviateVectorStore(client=client, index_name="TechArticle", text_key="content", embedding=embeddings)

# Universal retriever interface—works with any LangChain vectorstore
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# Use directly in a LangGraph node
def retrieve_context(state):
    query = state["messages"][-1].content
    docs = retriever.invoke(query)
    context = "\n\n".join(doc.page_content for doc in docs)
    return {**state, "context": context}

from sentence_transformers import SentenceTransformer
import numpy as np

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

# Benchmark retrieval quality on your actual data before choosing
corpus = ["document 1 text...", "document 2 text...", "document 3 text..."]
queries = ["test query 1", "test query 2"]

corpus_embeddings = model.encode(corpus, normalize_embeddings=True)
query_embeddings = model.encode(queries, normalize_embeddings=True)

# Cosine similarity (fast with normalized embeddings = dot product)
scores = np.dot(query_embeddings, corpus_embeddings.T)
print(scores)