Weaviate: Collections, Hybrid Search and RAG Reference Guide

# Docker — single node with text2vec + generative-openai bolted on
docker run -d --name weaviate -p 8080:8080 -p 50051:50051 \
  -e ENABLE_MODULES="text2vec-openai,generative-openai,reranker-cohere" \
  -e DEFAULT_VECTORIZER_MODULE=text2vec-openai \
  -e OPENAI_APIKEY=$OPENAI_API_KEY \
  -e PERSISTENCE_DATA_PATH=/var/lib/weaviate \
  cr.weaviate.io/semitechnologies/weaviate:1.25

# Python client v4 — sync + async, gRPC under the hood
pip install "weaviate-client>=4.5"

# JS / TS — v3 client
npm install weaviate-client

# Health check
curl http://localhost:8080/v1/.well-known/ready
# → 200 OK once nodes are up

import weaviate
import weaviate.classes.config as wvcc
from weaviate.classes.config import Property, DataType

client = weaviate.connect_to_local()           # or connect_to_weaviate_cloud(...)

# Idempotent create
if not client.collections.exists("Article"):
    client.collections.create(
        name="Article",
        # Vectorize with OpenAI; only the "body" property feeds the vector
        vectorizer_config=wvcc.Configure.Vectorizer.text2vec_openai(
            model="text-embedding-3-small",
        ),
        generative_config=wvcc.Configure.Generative.openai(
            model="gpt-4o-mini",
        ),
        properties=[
            Property(name="title",     data_type=DataType.TEXT),
            Property(name="body",      data_type=DataType.TEXT),
            Property(name="tags",      data_type=DataType.TEXT_ARRAY),
            Property(name="published", data_type=DataType.DATE),
            Property(name="rating",    data_type=DataType.NUMBER),
        ],
        # HNSW + quantization
        vector_index_config=wvcc.Configure.VectorIndex.hnsw(
            quantizer=wvcc.Configure.VectorIndex.Quantizer.pq(),
        ),
    )

client.close()

import weaviate
from weaviate.classes.query import Filter, MetadataQuery, HybridFusion

client = weaviate.connect_to_local()
col = client.collections.get("Article")

# Pure semantic (vector) search
near = col.query.near_text(
    query="how does retrieval-augmented generation work?",
    limit=5,
    filters=Filter.by_property("tags").contains_any(["rag", "embeddings"]),
    return_metadata=MetadataQuery(distance=True, score=True),
)

# BM25-only (keyword)
bm = col.query.bm25(
    query="RAG retrieval",
    query_properties=["title^2", "body"],   # boost title
    limit=5,
)

# Hybrid — server-side fusion. alpha=0 → pure BM25, 1 → pure vector
hybrid = col.query.hybrid(
    query="RAG retrieval generation",
    alpha=0.5,
    fusion_type=HybridFusion.RELATIVE_SCORE,   # or RANKED (RRF)
    limit=5,
    return_metadata=MetadataQuery(score=True, explain_score=True),
)

for o in hybrid.objects:
    print(f"{o.metadata.score:.3f}  {o.properties['title']}")

client.close()

# Minimal Weaviate RAG — schema, ingest, hybrid retrieve, RAG generate
import os, weaviate
import weaviate.classes.config as wvcc
from weaviate.classes.config import Property, DataType
from weaviate.classes.query import Filter

client = weaviate.connect_to_local(
    headers={"X-OpenAI-Api-Key": os.environ["OPENAI_API_KEY"]},
)

NAME = "Article"
if not client.collections.exists(NAME):
    client.collections.create(
        name=NAME,
        vectorizer_config=wvcc.Configure.Vectorizer.text2vec_openai(),
        generative_config=wvcc.Configure.Generative.openai(model="gpt-4o-mini"),
        properties=[
            Property(name="title", data_type=DataType.TEXT),
            Property(name="body",  data_type=DataType.TEXT),
            Property(name="tags",  data_type=DataType.TEXT_ARRAY),
        ],
    )

col = client.collections.get(NAME)

# Batch insert — Weaviate calls the vectorizer for you
with col.batch.dynamic() as batch:
    for d in [
        {"title": "What is RAG?", "body": "Retrieval-augmented generation joins...",
         "tags": ["rag", "intro"]},
        {"title": "Hybrid search", "body": "BM25 + vectors fuse for...",
         "tags": ["rag", "hybrid"]},
    ]:
        batch.add_object(properties=d)

# Hybrid retrieve + RAG generate in one call
res = col.generate.hybrid(
    query="explain RAG",
    alpha=0.5, limit=3,
    filters=Filter.by_property("tags").contains_any(["rag"]),
    grouped_task="Answer the question using only the retrieved articles.",
)
print(res.generated)
client.close()