LangChain RAG with Weaviate: a step-by-step guide

If you’re searching for a langchain weaviate rag guide that goes beyond “hello world,” you’ve likely hit the same walls we did: chunking strategies that butcher context, retrieval that returns noise, and embedding costs that scale faster than traffic. This guide walks through a production-grade pipeline — document ingestion, hybrid search, reranking, and observability — with the specific LangChain and Weaviate primitives that actually behave under load.

Why this stack

LangChain gives you composable chains and a stable interface surface. Weaviate gives you a vector database with built-in hybrid search (BM25 + dense vectors), multi-tenancy, and a GraphQL/REST API that doesn’t require a separate control plane. Together they cover the full RAG loop without forcing you into a managed service you can’t debug.

The tradeoff: you own the infrastructure. Weaviate runs as a single binary or Kubernetes deployment. LangChain’s abstraction layer can hide performance cliffs if you don’t understand what each component actually does. We’ll make those boundaries explicit.

Prerequisites and versions

pip install langchain langchain-weaviate langchain-openai weaviate-client

Versions used in this guide:

• langchain>=0.2.0
• langchain-weaviate>=0.1.0
• weaviate-client>=4.0.0 (v4 API, not the legacy v3 client)
• langchain-openai>=0.1.0

Weaviate 1.25+ is required for the hybrid search and multi-tenancy features shown here. If you’re on Weaviate Cloud (WCD), the same client code works — just swap the connection parameters.

Connect to Weaviate

import weaviate
from weaviate.auth import AuthApiKey

# Local dev
client = weaviate.connect_to_local()

# Weaviate Cloud
# client = weaviate.connect_to_wcd(
#     cluster_url="https://your-cluster.weaviate.cloud",
#     auth_credentials=AuthApiKey("your-api-key"),
# )

# Verify connection
assert client.is_ready()
print("Connected to Weaviate:", client.get_meta()["version"])

Pitfall: The v4 client uses connect_to_local() and connect_to_wcd() factory methods. The old weaviate.Client() constructor is deprecated and will not work with LangChain’s Weaviate integrations.

Define the collection schema

Weaviate collections replace the old “class” concept. Define yours with explicit vectorizer configuration — this controls how objects get embedded at import time.

from weaviate.classes.config import Configure, Property, DataType

client.collections.delete("Document")  # clean slate for demo

client.collections.create(
    name="Document",
    description="Source documents for RAG",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(
        model="text-embedding-3-small",
        dimensions=1536,
    ),
    properties=[
        Property(name="content", data_type=DataType.TEXT, description="Chunk text"),
        Property(name="source", data_type=DataType.TEXT, description="File or URL origin"),
        Property(name="page", data_type=DataType.INT, description="Page or section number"),
        Property(name="metadata", data_type=DataType.OBJECT, description="Arbitrary JSON"),
    ],
    # Enable hybrid search (BM25 + vector) by default
    inverted_index_config=Configure.inverted_index(
        index_timestamps=True,
        index_property_length=True,
    ),
)

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

Tradeoff: Using Weaviate’s built-in text2vec_openai vectorizer means embeddings happen at write time, inside Weaviate. This simplifies ingestion but couples you to OpenAI’s embedding API. For multi-provider embedding strategies (or local models), generate embeddings in your application and pass them explicitly — see the “Bring your own embeddings” section below.

Ingest documents with LangChain loaders and splitters

LangChain’s document loaders handle PDFs, HTML, Markdown, Notion, and more. The splitter choice determines retrieval quality more than any other single decision.

from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
import os

def load_and_split(path: str) -> list[Document]:
    if path.endswith(".pdf"):
        loader = PyPDFLoader(path)
    else:
        loader = TextLoader(path, encoding="utf-8")
    
    docs = loader.load()
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    
    chunks = splitter.split_documents(docs)
    
    # Enrich metadata for filtering later
    for i, chunk in enumerate(chunks):
        chunk.metadata.update({
            "chunk_index": i,
            "total_chunks": len(chunks),
        })
    
    return chunks

# Example usage
chunks = load_and_split("./data/your-docs.pdf")
print(f"Created {len(chunks)} chunks")

Pitfall: chunk_size=1000 characters (not tokens). For text-embedding-3-small, this averages ~250-300 tokens per chunk — a reasonable starting point. Measure your actual token distribution with tiktoken before committing.

Pitfall: Overlap of 200 characters helps preserve context across boundaries, but too much overlap wastes embedding budget and dilutes retrieval specificity. Start at 15-20% of chunk size and tune based on eval.

Write chunks to Weaviate via LangChain

LangChain’s WeaviateVectorStore handles the write path. Two modes exist: let Weaviate vectorize (using the collection’s vectorizer) or pass pre-computed vectors.

Mode A: Weaviate vectorizes (simplest)

from langchain_weaviate import WeaviateVectorStore
from langchain_openai import OpenAIEmbeddings

# Uses the collection's configured vectorizer (text2vec_openai)
vector_store = WeaviateVectorStore(
    client=client,
    index_name="Document",
    text_key="content",
    # No embedding= parameter — Weaviate handles it
)

ids = vector_store.add_documents(chunks)
print(f"Inserted {len(ids)} documents")

Mode B: Bring your own embeddings (multi-provider, local models, cost control)

from langchain_openai import OpenAIEmbeddings
# Or: from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

vector_store = WeaviateVectorStore(
    client=client,
    index_name="Document",
    text_key="content",
    embedding=embeddings,  # LangChain computes vectors, sends to Weaviate
)

ids = vector_store.add_documents(chunks)

Tradeoff: Mode A reduces application complexity and network round-trips. Mode B lets you use any embedding model (local, Cohere, Voyage, mixed providers) and gives you per-token usage visibility — critical if you’re routing through a gateway like n4n.ai that meters embedding calls across providers.

Hybrid search: the retrieval workhorse

Pure vector search fails on exact-match queries (error codes, product IDs, proper nouns). Pure keyword search fails on semantic intent. Weaviate’s hybrid search combines both with a single parameter.

from langchain_weaviate import WeaviateVectorStore

vector_store = WeaviateVectorStore(
    client=client,
    index_name="Document",
    text_key="content",
    embedding=embeddings,  # Required for hybrid even if Weaviate vectorizes
)

# alpha=0.5 balances BM25 and vector equally
# alpha=1.0 = pure vector, alpha=0.0 = pure BM25
retriever = vector_store.as_retriever(
    search_type="hybrid",
    search_kwargs={
        "k": 10,
        "alpha": 0.5,
        # Optional: filter by metadata
        "filters": {
            "operator": "Equal",
            "path": ["source"],
            "valueText": "your-docs.pdf",
        },
    },
)

results = retriever.invoke("How do I configure authentication?")
for doc in results:
    print(f"Score: {doc.metadata.get('score', 'N/A'):.3f} | Source: {doc.metadata.get('source')} | Page: {doc.metadata.get('page')}")
    print(doc.page_content[:200])
    print("---")

Pitfall: The alpha parameter is not a magic dial. For technical documentation with lots of exact terminology, try alpha=0.3 (favor BM25). For conversational QA over prose, alpha=0.7 often wins. Run a small eval set to pick.

Pitfall: Hybrid search requires the collection to have inverted_index_config enabled (shown in schema definition). Without it, BM25 scores are unavailable and alpha is ignored.

Rerank for precision

Retrieving 10-20 candidates then reranking with a cross-encoder is the single highest-ROI improvement for RAG quality. LangChain’s ContextualCompressionRetriever wraps this pattern.

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# Cohere Rerank v3.5 is fast and effective
# Alternatives: Jina AI Reranker, BGE-Reranker (local via HuggingFace)
reranker = CohereRerank(model="rerank-v3.5", top_n=5)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=retriever,
)

# Returns top 5 after reranking
reranked = compression_retriever.invoke("How do I configure authentication?")
for doc in reranked:
    print(f"Relevance: {doc.metadata.get('relevance_score', 'N/A'):.3f}")
    print(doc.page_content[:300])
    print("---")

Tradeoff: Reranking adds ~100-300ms latency per query. For high-throughput paths, consider:

• Async reranking with a smaller candidate pool (k=10 → top_n=3)
• Caching reranker scores for repeated queries
• Using a lighter reranker (BGE-Reranker-Large quantized) locally

Build the RAG chain

LangChain’s create_retrieval_chain and create_stuff_documents_chain compose retrieval + generation cleanly.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

system_prompt = """You are a technical assistant. Answer the user's question using only the provided context.
If the context doesn't contain the answer, say you don't know. Cite sources using [source: page] format."""

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "Context:\n{context}\n\nQuestion: {input}"),
])

question_answer_chain = create_stuff_documents_chain(llm, prompt)
rag_chain = create_retrieval_chain(compression_retriever, question_answer_chain)

# Invoke
response = rag_chain.invoke({"input": "How do I configure authentication?"})
print(response["answer"])
print("\nSources:")
for doc in response["context"]:
    print(f"  - {doc.metadata.get('source')}, page {doc.metadata.get('page')}")

Pitfall: create_stuff_documents_chain stuffs all retrieved chunks into a single prompt. With 5 reranked chunks at ~1000 chars each, you’re at ~5k tokens — fine for gpt-4o-mini’s 128k context. If you increase top_n or chunk size, watch the token budget. Use create_map_reduce_documents_chain or create_refine_documents_chain for larger contexts.

Streaming responses

Production UIs need streaming. LangChain’s astream yields tokens as they arrive.

async def stream_answer(question: str):
    async for chunk in rag_chain.astream({"input": question}):
        if "answer" in chunk:
            yield chunk["answer"]
        elif "context" in chunk:
            # Optionally yield sources at the end
            pass

# Usage in FastAPI/Starlette:
# @app.post("/chat")
# async def chat(request: ChatRequest):
#     return StreamingResponse(stream_answer(request.question), media_type="text/plain")

Observability: log retrieval and generation separately

You cannot debug RAG quality without knowing whether retrieval failed or generation hallucinated. Log both.

import json
import time
from contextlib import contextmanager

@contextmanager
def log_stage(stage: str, **kwargs):
    start = time.perf_counter()
    try:
        yield
    finally:
        duration = time.perf_counter() - start
        print(json.dumps({
            "stage": stage,
            "duration_ms": round(duration * 1000, 2),
            **kwargs,
        }))

# Wrap retrieval
with log_stage("retrieval", query=question):
    retrieved = compression_retriever.invoke(question)

# Log retrieved chunk metadata for later analysis
with log_stage("retrieval_results", query=question, num_chunks=len(retrieved)):
    for i, doc in enumerate(retrieved):
        print(json.dumps({
            "rank": i,
            "source": doc.metadata.get("source"),
            "page": doc.metadata.get("page"),
            "relevance_score": doc.metadata.get("relevance_score"),
            "content_preview": doc.page_content[:100],
        }))

# Wrap generation
with log_stage("generation", model="gpt-4o-mini", num_context_chunks=len(retrieved)):
    response = question_answer_chain.invoke({
        "input": question,
        "context": retrieved,
    })

Why this matters: When latency spikes or quality drops, you’ll know immediately whether to investigate the vector index, the reranker, or the LLM. This structure also feeds eval pipelines — pair retrieval logs with human labels to measure nDCG@k over time.

Multi-tenancy: isolate customers or projects

Weaviate’s multi-tenancy lets you partition data logically within a single collection — no separate clusters needed.

# Enable multi-tenancy at collection creation
client.collections.create(
    name="Document",
    # ... other config ...
    multi_tenancy_config=Configure.multi_tenancy(enabled=True),
)

# Create tenants (typically one per customer/project)
collection = client.collections.get("Document")
collection.tenants.create(["customer-acme", "customer-globex", "internal-docs"])

# Write to a specific tenant
with collection.batch.dynamic() as batch:
    for chunk in chunks:
        batch.add_object(
            properties={
                "content": chunk.page_content,
                "source": chunk.metadata.get("source"),
                "page": chunk.metadata.get("page"),
                "metadata": chunk.metadata,
            },
            tenant="customer-acme",
        )

# Query a specific tenant
results = collection.query.hybrid(
    query="authentication config",
    alpha=0.5,
    limit=10,
    tenant="customer-acme",
)

Pitfall: Tenant names must be created before use. The tenants.create() call is idempotent — safe to call on startup.

Tradeoff: Multi-tenancy adds a small overhead per query (tenant routing). For <100 tenants, it’s negligible. For thousands, consider dedicated collections per tenant instead.

Bring your own embeddings: full control

If you need provider diversity, local inference, or exact token accounting, compute embeddings in your app and pass them to Weaviate directly.

from langchain_openai import OpenAIEmbeddings
import numpy as np

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Compute vectors
texts = [doc.page_content for doc in chunks]
vectors = embeddings.embed_documents(texts)

# Write with pre-computed vectors
with collection.batch.dynamic() as batch:
    for doc, vector in zip(chunks, vectors):
        batch.add_object(
            properties={
                "content": doc.page_content,
                "source": doc.metadata.get("source"),
                "page": doc.metadata.get("page"),
                "metadata": doc.metadata,
            },
            vector=vector,
            tenant="customer-acme",
        )

# Query with pre-computed query vector
query_vector = embeddings.embed_query("authentication config")
results = collection.query.hybrid(
    query="authentication config",  # Still used for BM25
    vector=query_vector,            # Used for dense search
    alpha=0.5,
    limit=10,
    tenant="customer-acme",
)

Why bother: This pattern lets you swap embedding providers without re-indexing (just re-embed new docs), run local models via Ollama or TGI, and get exact per-request token counts for cost allocation. If you’re routing LLM calls through a gateway that meters usage, apply the same pattern to embeddings.

Common failure modes and fixes

Scaling considerations

• Write throughput: Weaviate’s HNSW index builds incrementally. For bulk loads (>100k objects), disable vector indexing during import (collection.config.update(vector_index_config=Configure.VectorIndex.hnsw(ef_construction=0))), import, then re-enable and trigger async indexing.
• Read throughput: Horizontal scaling via Weaviate replicas. Each replica serves queries independently. Configure replication_factor in collection settings.
• Memory: HNSW indexes live in RAM. Rule of thumb: ~1.5-2x vector size in memory. For 1M vectors at 1536 dims (float32), expect ~12-16 GB RAM for the index alone.
• Quantization: Weaviate supports PQ (product quantization) and BQ (binary quantization) to reduce memory 4-8x with minimal recall loss. Enable via vector_index_config=Configure.VectorIndex.hnsw(quantizer=Configure.VectorIndex.Quantizer.pq()).

What to build next

This pipeline gets you to a working, observable RAG system. From here, the highest-leverage investments are:

1. Evaluation harness — synthetic QA pairs from your docs, measure retrieval recall@k and answer correctness
2. Query rewriting — decompose complex questions, rewrite for better retrieval (HyDE, step-back prompting)
3. Adaptive retrieval — route simple queries to BM25-only, complex to hybrid+rerank
4. Feedback loop — log user thumbs-up/down, mine for hard negatives, retrain reranker

The LangChain Weaviate RAG guide pattern scales because each component is swappable. You can replace the reranker, the LLM, the embedding model, or the vector store without rewriting the orchestration logic. That modularity is the real product.