RAG Pipeline Shootout: Pinecone 1.10 vs. Weaviate 1.25 vs. Redis 8.0 for 1M Document Indexes

import os
import time
import pinecone
from pinecone import Pinecone, ServerlessSpec
from sentence_transformers import SentenceTransformer
import pandas as pd
from tqdm import tqdm

# Initialize Pinecone 1.10 client
# Official client: https://github.com/pinecone-io/pinecone-python-client
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

# Configuration
INDEX_NAME = "pubmed-1m-768d"
DIMENSION = 768
METRIC = "cosine"
REGION = "us-east-1"
BATCH_SIZE = 100  # Pinecone max batch size per API call

def init_pinecone_index():
    """Create or connect to existing Pinecone index with HNSW config"""
    if INDEX_NAME not in pc.list_indexes().names():
        print(f"Creating index {INDEX_NAME}...")
        pc.create_index(
            name=INDEX_NAME,
            dimension=DIMENSION,
            metric=METRIC,
            spec=ServerlessSpec(cloud="aws", region=REGION),
            # HNSW parameters tuned for 1M doc workload
            metadata_config={"indexed": ["pubmed_id", "journal"]},
            pod_type="s1.x1"  # Equivalent to benchmark hardware tier
        )
        # Wait for index to be ready
        while not pc.describe_index(INDEX_NAME).status["ready"]:
            time.sleep(1)
        print("Index ready.")
    return pc.Index(INDEX_NAME)

def embed_and_upsert_batch(model, batch_df):
    """Embed text and upsert to Pinecone with error handling"""
    try:
        # Generate 384d embeddings, scale to 768d for benchmark parity
        embeddings = model.encode(batch_df["abstract"].tolist(), show_progress_bar=False)
        scaled_embeddings = [list(emb) + [0.0]*384 for emb in embeddings]  # Pad to 768d

        # Prepare vectors for upsert
        vectors = []
        for idx, row in batch_df.iterrows():
            vectors.append({
                "id": row["pubmed_id"],
                "values": scaled_embeddings[idx % len(scaled_embeddings)],
                "metadata": {
                    "journal": row["journal"],
                    "publication_date": row["pub_date"],
                    "abstract": row["abstract"][:500]  # Truncate for storage
                }
            })

        # Upsert with retry logic for rate limits
        max_retries = 3
        for attempt in range(max_retries):
            try:
                index.upsert(vectors=vectors, batch_size=BATCH_SIZE)
                return True
            except pinecone.exceptions.PineconeRateLimitError as e:
                print(f"Rate limit hit, retrying in {2**attempt}s...")
                time.sleep(2**attempt)
            except Exception as e:
                print(f"Upsert error: {str(e)}")
                if attempt == max_retries -1:
                    raise
        return False
    except Exception as e:
        print(f"Batch processing failed: {str(e)}")
        return False

if __name__ == "__main__":
    # Load 1M document dataset (PubMed abstracts)
    # Dataset source: https://github.com/ncbi/datasets
    df = pd.read_csv("pubmed_1m_abstracts.csv", chunksize=1000)
    model = SentenceTransformer("all-MiniLM-L6-v2")
    index = init_pinecone_index()

    total_upserted = 0
    start_time = time.time()

    for chunk in tqdm(df, total=1000, desc="Upserting to Pinecone"):
        success = embed_and_upsert_batch(model, chunk)
        if success:
            total_upserted += len(chunk)
        else:
            print(f"Failed to upsert chunk, skipping...")

    elapsed = time.time() - start_time
    print(f"Upserted {total_upserted} documents in {elapsed:.2f}s")
    print(f"Write throughput: {total_upserted / elapsed:.2f} docs/s")

import os
import time
import weaviate
from weaviate.classes.init import Auth
from weaviate.classes.config import Configure, Property, DataType
from sentence_transformers import SentenceTransformer
import pandas as pd
from tqdm import tqdm

# Initialize Weaviate 1.25 client (self-hosted)
# Official client: https://github.com/weaviate/weaviate-python-client
client = weaviate.connect_to_local(
    host="localhost",
    port=8080,
    grpc_port=50051,
    auth_credentials=Auth.api_key(os.getenv("WEAVIATE_API_KEY")) if os.getenv("WEAVIATE_API_KEY") else None
)

# Configuration
CLASS_NAME = "PubmedAbstract"
DIMENSION = 768
BATCH_SIZE = 200  # Weaviate max batch size for optimal throughput

def init_weaviate_schema():
    """Create Weaviate schema with HNSW config tuned for 1M docs"""
    if client.collections.exists(CLASS_NAME):
        print(f"Collection {CLASS_NAME} already exists, skipping creation.")
        return client.collections.get(CLASS_NAME)

    print(f"Creating collection {CLASS_NAME}...")
    collection = client.collections.create(
        name=CLASS_NAME,
        description="PubMed 2024 abstracts for RAG benchmarking",
        properties=[
            Property(name="pubmed_id", data_type=DataType.TEXT),
            Property(name="journal", data_type=DataType.TEXT),
            Property(name="publication_date", data_type=DataType.DATE),
            Property(name="abstract", data_type=DataType.TEXT),
        ],
        vectorizer_config=Configure.Vectorizer.none(),  # We provide our own embeddings
        # HNSW parameters for 1M doc workload: balance latency and recall
        vector_index_config=Configure.VectorIndex.hnsw(
            distance_metric=Configure.VectorIndex.Distance.COSINE,
            ef_construction=256,
            max_connections=32,
            ef=128  # Tune for p99 latency target
        ),
        # Enable BM25 keyword search on abstract field
        inverted_index_config=Configure.InvertedIndex(
            index_timestamps=True,
            index_property_length=True
        )
    )
    print("Collection created.")
    return collection

def embed_and_upsert_weaviate(model, collection, batch_df):
    """Embed and upsert batch to Weaviate with error handling"""
    try:
        # Generate 384d embeddings, scale to 768d for parity
        embeddings = model.encode(batch_df["abstract"].tolist(), show_progress_bar=False)
        scaled_embeddings = [list(emb) + [0.0]*384 for emb in embeddings]

        # Prepare objects for batch import
        objects = []
        for idx, row in batch_df.iterrows():
            objects.append({
                "pubmed_id": row["pubmed_id"],
                "journal": row["journal"],
                "publication_date": row["pub_date"],
                "abstract": row["abstract"][:500]
            })

        # Batch upsert with retry logic
        max_retries = 3
        for attempt in range(max_retries):
            try:
                # Upsert with vectors
                collection.data.insert_many(
                    objects=objects,
                    vectors=[scaled_embeddings[i] for i in range(len(objects))]
                )
                return True
            except weaviate.exceptions.WeaviateRateLimitError as e:
                print(f"Rate limit hit, retrying in {2**attempt}s...")
                time.sleep(2**attempt)
            except Exception as e:
                print(f"Weaviate upsert error: {str(e)}")
                if attempt == max_retries -1:
                    raise
        return False
    except Exception as e:
        print(f"Batch processing failed: {str(e)}")
        return False

if __name__ == "__main__":
    # Load dataset
    df = pd.read_csv("pubmed_1m_abstracts.csv", chunksize=1000)
    model = SentenceTransformer("all-MiniLM-L6-v2")
    collection = init_weaviate_schema()

    total_upserted = 0
    start_time = time.time()

    for chunk in tqdm(df, total=1000, desc="Upserting to Weaviate"):
        success = embed_and_upsert_weaviate(model, collection, chunk)
        if success:
            total_upserted += len(chunk)
        else:
            print(f"Failed to upsert chunk, skipping...")

    elapsed = time.time() - start_time
    print(f"Upserted {total_upserted} documents in {elapsed:.2f}s")
    print(f"Write throughput: {total_upserted / elapsed:.2f} docs/s")
    client.close()

import os
import time
import redis
from redisvl.index import SearchIndex
from redisvl.schema import IndexSchema
from sentence_transformers import SentenceTransformer
import pandas as pd
from tqdm import tqdm

# Initialize Redis 8.0 client (self-hosted)
# redis-py client: https://github.com/redis/redis-py
# redisvl: https://github.com/redis/redisvl
redis_client = redis.Redis(
    host="localhost",
    port=6379,
    password=os.getenv("REDIS_PASSWORD"),
    decode_responses=False  # Binary for vector data
)

# Configuration
INDEX_NAME = "pubmed-1m-768d"
DIMENSION = 768
BATCH_SIZE = 500  # Redis max batch size for high throughput

def init_redis_index():
    """Create Redis HNSW index with RediSearch and vector support"""
    # Define index schema for hybrid search (vector + keyword)
    schema = IndexSchema.from_dict({
        "index": {
            "name": INDEX_NAME,
            "prefix": ["pubmed:"],
            "storage_type": "hash"
        },
        "fields": [
            {"name": "pubmed_id", "type": "text"},
            {"name": "journal", "type": "text", "sortable": True},
            {"name": "publication_date", "type": "numeric", "sortable": True},
            {"name": "abstract", "type": "text", "weight": 1.0},  # For BM25
            {
                "name": "embedding",
                "type": "vector",
                "attrs": {
                    "dims": DIMENSION,
                    "distance_metric": "cosine",
                    "algorithm": "hnsw",
                    "ef_construction": 256,
                    "m": 32,
                    "ef_runtime": 128
                }
            }
        ]
    })

    # Create or load existing index
    idx = SearchIndex(schema=schema, redis_client=redis_client)
    if not idx.exists():
        print(f"Creating Redis index {INDEX_NAME}...")
        idx.create()
        print("Index created.")
    else:
        print(f"Index {INDEX_NAME} already exists.")
    return idx

def embed_and_upsert_redis(model, idx, batch_df):
    """Embed and upsert batch to Redis with error handling"""
    try:
        # Generate 384d embeddings, scale to 768d for parity
        embeddings = model.encode(batch_df["abstract"].tolist(), show_progress_bar=False)
        scaled_embeddings = [list(emb) + [0.0]*384 for emb in embeddings]

        # Prepare hash keys and data for batch upsert
        keys = [f"pubmed:{row['pubmed_id']}" for _, row in batch_df.iterrows()]
        data = []
        for idx, row in batch_df.iterrows():
            data.append({
                "pubmed_id": row["pubmed_id"],
                "journal": row["journal"],
                "publication_date": int(time.mktime(time.strptime(row["pub_date"], "%Y-%m-%d"))),
                "abstract": row["abstract"][:500],
                "embedding": scaled_embeddings[idx % len(scaled_embeddings)]
            })

        # Batch upsert with pipeline for performance
        max_retries = 3
        for attempt in range(max_retries):
            try:
                pipe = redis_client.pipeline()
                for i in range(len(keys)):
                    pipe.hset(keys[i], mapping=data[i])
                pipe.execute()
                return True
            except redis.exceptions.RedisError as e:
                print(f"Redis error: {str(e)}, retrying in {2**attempt}s...")
                time.sleep(2**attempt)
            except Exception as e:
                print(f"Redis upsert error: {str(e)}")
                if attempt == max_retries -1:
                    raise
        return False
    except Exception as e:
        print(f"Batch processing failed: {str(e)}")
        return False

if __name__ == "__main__":
    # Load dataset
    df = pd.read_csv("pubmed_1m_abstracts.csv", chunksize=1000)
    model = SentenceTransformer("all-MiniLM-L6-v2")
    idx = init_redis_index()

    total_upserted = 0
    start_time = time.time()

    for chunk in tqdm(df, total=1000, desc="Upserting to Redis"):
        success = embed_and_upsert_redis(model, idx, chunk)
        if success:
            total_upserted += len(chunk)
        else:
            print(f"Failed to upsert chunk, skipping...")

    elapsed = time.time() - start_time
    print(f"Upserted {total_upserted} documents in {elapsed:.2f}s")
    print(f"Write throughput: {total_upserted / elapsed:.2f} docs/s")
    redis_client.close()

# Weaviate 1.25 HNSW tuning example
collection = client.collections.create(
    name="rag-docs",
    vector_index_config=Configure.VectorIndex.hnsw(
        distance_metric=Configure.VectorIndex.Distance.COSINE,
        ef_construction=256,  # Higher = better recall, slower indexing
        max_connections=32,   # Higher = better recall, more storage
        ef=128               # Higher = lower latency, more compute
    )
)

# Pinecone 1.10 batched upsert with retry
def upsert_with_retry(index, vectors, max_retries=3):
    for attempt in range(max_retries):
        try:
            index.upsert(vectors=vectors, batch_size=100)
            return True
        except pinecone.exceptions.PineconeRateLimitError:
            time.sleep(2**attempt)
        except Exception as e:
            print(f"Upsert failed: {e}")
            if attempt == max_retries -1:
                return False
    return False

# Redis 8.0 hybrid search query example
from redisvl.query import VectorQuery, TextQuery, HybridQuery

# Vector query for semantic similarity
vector_q = VectorQuery(
    vector=query_embedding,
    vector_field="embedding",
    num_results=10
)

# Text query for keyword matching
text_q = TextQuery(
    text="diabetes treatment",
    fields=["abstract"]
)

# Hybrid query combining both
hybrid_q = HybridQuery(
    queries=[vector_q, text_q],
    weights=[0.7, 0.3]  # 70% vector, 30% keyword
)

results = idx.search(hybrid_q)