LangChain and Milvus: scaling RAG to millions of documents

Scaling a RAG system from a prototype to millions of documents changes every architectural decision. The retrieval latency that felt fine at 10k chunks becomes a bottleneck at 10M. The embedding pipeline that ran overnight now needs incremental updates in minutes. This guide walks through the concrete choices that keep LangChain and Milvus performant at scale, with code you can adapt and the tradeoffs you’ll actually face.

Choose the right Milvus deployment model

Milvus runs in three modes: standalone (single node), cluster (distributed), and Zilliz Cloud (managed). For millions of vectors, standalone hits memory and CPU limits fast. Cluster mode separates query nodes, data nodes, and index nodes — letting you scale read and write paths independently.

# docker-compose snippet for local cluster dev (not production)
services:
  etcd:
    image: quay.io/coreos/etcd:v3.5.5
  minio:
    image: minio/minio:RELEASE.2023-03-20
  pulsar:
    image: apachepulsar/pulsar:2.11.0
  milvus:
    image: milvusdb/milvus:v2.3.4
    command: ["milvus", "run", "cluster"]
    environment:
      ETCD_ENDPOINTS: etcd:2379
      MINIO_ADDRESS: minio:9000
      PULSAR_ADDRESS: pulsar://pulsar:6650

In production, use the Helm chart with separate node pools for query and data nodes. Set replicaCount on query nodes for read throughput; scale data nodes for write throughput. The managed Zilliz Cloud option removes operational burden but adds cost and latency variance — evaluate based on your team’s capacity.

Design collections for your access patterns

A common mistake: one giant collection with all documents. This forces every query to scan irrelevant partitions. Instead, partition by logical boundaries — tenant, time range, document type, or language.

from pymilvus import CollectionSchema, FieldSchema, DataType, Collection, utility

def create_partitioned_collection(name: str, dim: int = 1536) -> Collection:
    fields = [
        FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
        FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=dim),
        FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
        FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=256),
        FieldSchema(name="tenant_id", dtype=DataType.VARCHAR, max_length=64),
        FieldSchema(name="created_at", dtype=DataType.INT64),  # unix ms
        FieldSchema(name="metadata", dtype=DataType.JSON),
    ]
    schema = CollectionSchema(fields, description="RAG chunks with tenant partitioning")
    collection = Collection(name, schema, using="default", shards_num=4)
    
    # Create partitions per tenant (or time bucket)
    for tenant in ["tenant_a", "tenant_b", "tenant_c"]:
        collection.create_partition(tenant)
    
    return collection

Partition pruning happens automatically when you filter on the partition key. Always include the partition field in your search expr — Milvus won’t prune without it.

# Query only tenant_a's partition
results = collection.search(
    data=[query_vector],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"nprobe": 32}},
    limit=10,
    expr='tenant_id == "tenant_a"',
    output_fields=["text", "doc_id", "metadata"]
)

Tradeoff: Too many partitions (thousands) increases metadata overhead and slows collection loading. Aim for tens to low hundreds. If you need finer isolation, use the expr filter on a non-partition field instead.

Pick the right index for your scale and latency budget

Milvus supports multiple index types. For millions of vectors, the choice directly determines p99 latency and recall.

Index type	Build time	Memory	Query latency	Recall	Best for
IVF_FLAT	Fast	Low	Medium	High	< 5M vectors, exact recall needed
IVF_SQ8	Fast	Low (quantized)	Medium	Slightly lower	Memory-constrained, < 10M
HNSW	Slow	High	Low	Highest	< 2M, latency-critical
IVF_PQ	Medium	Very low	Fast	Lower	> 10M, approximate OK
DISKANN	Medium	Disk-backed	Low	High	> 50M, cost-sensitive

For most RAG workloads at 5–50M vectors, IVF_PQ with nprobe=32–64 hits the sweet spot. HNSW’s memory footprint grows fast — budget ~1.5x vector size in RAM.

index_params = {
    "metric_type": "COSINE",
    "index_type": "IVF_PQ",
    "params": {"nlist": 4096, "m": 16, "nbits": 8}
}
collection.create_index("embedding", index_params)
collection.load()  # loads index into memory

Pitfall: nlist too low → coarse centroids, poor recall. nlist too high → build time explodes. Rule of thumb: nlist = 4 * sqrt(N) where N is vectors per partition. For 1M vectors, ~4096. Validate with utility.index_building_progress().

Build an incremental embedding pipeline

Re-embedding millions of documents on every update is infeasible. Design for incremental updates from day one.

# langchain_milvus_pipeline.py
from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
import hashlib
import json

class IncrementalEmbeddingPipeline:
    def __init__(self, collection_name: str, embedding_model: str = "text-embedding-3-small"):
        self.embeddings = OpenAIEmbeddings(model=embedding_model)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512,
            chunk_overlap=50,
            separators=["\n\n", "\n", ". ", " ", ""]
        )
        self.vector_store = Milvus(
            embedding_function=self.embeddings,
            collection_name=collection_name,
            connection_args={"uri": "http://localhost:19530"},
            auto_id=True,
            text_field="text",
            vector_field="embedding",
            primary_field="id",
            metadata_field="metadata",
        )
    
    def content_hash(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]
    
    def upsert_documents(self, docs: list[Document], tenant_id: str) -> int:
        """Upsert with deduplication via content hash."""
        chunks = self.splitter.split_documents(docs)
        
        # Add deterministic IDs for upsert semantics
        for chunk in chunks:
            chunk.metadata["tenant_id"] = tenant_id
            chunk.metadata["content_hash"] = self.content_hash(chunk.page_content)
            chunk.metadata["created_at"] = int(time.time() * 1000)
        
        # Milvus upsert via primary key requires knowing existing IDs.
        # Practical approach: delete by content_hash filter, then insert.
        hashes = [c.metadata["content_hash"] for c in chunks]
        expr = f'metadata["content_hash"] in {hashes}'
        self.vector_store.col.delete(expr)
        
        ids = self.vector_store.add_documents(chunks)
        return len(ids)

Key decisions in this pipeline:

Deterministic chunking (fixed separators) ensures reproducible hashes
Delete-then-insert simulates upsert; Milvus 2.4+ adds native upsert() but check your version
Batch size 1000–5000 balances throughput and memory
Run embedding and insertion in separate worker pools — embedding is CPU/GPU bound, insertion is I/O bound

Optimize retrieval for production traffic

Naive similarity search fails at scale: it returns redundant chunks from the same document, misses keyword matches, and ignores recency. Layer three techniques:

1. MMR (Maximal Marginal Relevance) for diversity

from langchain_milvus import Milvus

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 20,           # fetch more candidates
        "fetch_k": 50,     # initial ANN pool
        "lambda_mult": 0.5 # 0 = diversity only, 1 = relevance only
    }
)

2. Hybrid search: vector + BM25

Milvus 2.3+ supports sparse vectors for BM25. Enable it during collection creation:

# Add sparse field to schema
FieldSchema(name="sparse", dtype=DataType.SPARSE_FLOAT_VECTOR)

# Create index for sparse
collection.create_index("sparse", {"index_type": "SPARSE_INVERTED_INDEX", "metric_type": "IP"})

from pymilvus import AnnSearchRequest, WeightedRanker

def hybrid_search(query: str, tenant_id: str, k: int = 10):
    dense_vec = embeddings.embed_query(query)
    sparse_vec = bm25_encoder.encode_query(query)  # fit on your corpus first
    
    dense_req = AnnSearchRequest(
        data=[dense_vec],
        anns_field="embedding",
        param={"metric_type": "COSINE", "params": {"nprobe": 32}},
        limit=k
    )
    sparse_req = AnnSearchRequest(
        data=[sparse_vec],
        anns_field="sparse",
        param={"metric_type": "IP"},
        limit=k
    )
    
    results = collection.hybrid_search(
        reqs=[dense_req, sparse_req],
        ranker=WeightedRanker(0.7, 0.3),  # weight dense higher
        expr=f'tenant_id == "{tenant_id}"',
        limit=k,
        output_fields=["text", "doc_id", "metadata"]
    )
    return results[0]

Tradeoff: BM25 requires fitting on your corpus. Re-fit monthly or when vocabulary shifts. The sparse index adds ~20% storage overhead.

3. Rerank with a cross-encoder

ANN retrieval gets you candidates; a cross-encoder reranks for precision. Keep the reranker small (e.g., bge-reranker-v2-m3 or ms-marco-MiniLM-L-6-v2) and batch requests.

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def rerank(query: str, candidates: list[Document], top_k: int = 5) -> list[Document]:
    pairs = [(query, doc.page_content) for doc in candidates]
    scores = reranker.predict(pairs, batch_size=32, show_progress_bar=False)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, _ in ranked[:top_k]]

Run reranking asynchronously. At 50 candidates, a small cross-encoder adds ~50ms p99 on a T4 GPU — acceptable for most RAG latencies.

Implement query routing for multi-tenant scale

When tenants have vastly different document counts (one has 10k, another 5M), a single collection with partition filtering still loads the full index for the large tenant. Route queries to tenant-specific collections or read replicas.

class TenantAwareRetriever:
    def __init__(self, base_uri: str):
        self.base_uri = base_uri
        self._collection_cache: dict[str, Milvus] = {}
    
    def get_store(self, tenant_id: str) -> Milvus:
        if tenant_id not in self._collection_cache:
            # Convention: one collection per tenant above 100k docs
            # Smaller tenants share a multi-tenant collection
            if self._is_large_tenant(tenant_id):
                coll_name = f"rag_{tenant_id}"
            else:
                coll_name = "rag_shared"
            self._collection_cache[tenant_id] = Milvus(
                embedding_function=embeddings,
                collection_name=coll_name,
                connection_args={"uri": self.base_uri},
            )
        return self._collection_cache[tenant_id]
    
    def retrieve(self, query: str, tenant_id: str, k: int = 5) -> list[Document]:
        store = self.get_store(tenant_id)
        return store.as_retriever(search_kwargs={"k": k, "expr": f'tenant_id == "{tenant_id}"'}).invoke(query)

Pitfall: Collection count explosion. Milvus handles hundreds of collections, but thousands slows meta operations and increases etcd pressure. Consolidate small tenants.

Monitor what matters

You can’t scale what you don’t measure. Instrument these metrics from day one:

# metrics.py
from prometheus_client import Histogram, Counter, Gauge

RETRIEVAL_LATENCY = Histogram(
    "rag_retrieval_latency_seconds",
    "End-to-end retrieval latency",
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5]
)

RERANK_LATENCY = Histogram(
    "rag_rerank_latency_seconds",
    "Cross-encoder rerank latency",
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5]
)

RECALL_AT_K = Gauge("rag_recall_at_k", "Offline recall@k evaluation", ["k"])
INDEX_BUILD_TIME = Histogram("milvus_index_build_seconds", "Index build duration")
QUERY_NODE_CPU = Gauge("milvus_query_node_cpu_percent", "Query node CPU", ["pod"])

Alert on:

p99 retrieval latency > 500ms (investigate nprobe, index type, query node count)
Index build time > 30min for incremental updates (increase build parallelism or reduce nlist)
Query node CPU > 70% sustained (scale query node replicas)
Recall@10 dropping below 0.85 on eval set (re-fit BM25, check embedding drift)

Common pitfalls and how to avoid them

Pitfall	Symptom	Fix
Single collection, no partitions	Query latency grows linearly with total vectors	Partition by tenant/time; filter on partition key
`nprobe` too low	Recall tanks at scale	Set `nprobe = nlist / 16` as starting point; tune via recall eval
No deduplication	Duplicate chunks inflate index, waste retrieval slots	Content-hash upsert pipeline (see code above)
Embedding model mismatch	Query vectors in different space than index	Version your embedding model; re-index on change
Synchronous reranking in request path	p99 latency spikes under load	Async rerank with timeout fallback to ANN results
Ignoring Milvus compaction	Segment count grows, query slows	Schedule `collection.compact()` weekly; monitor `num_segments`
Hardcoded `k=4`	Complex queries miss context	Dynamic `k`: start at 20, rerank to 5–8

Scaling checklist for your next review

Scaling RAG isn’t about one clever trick — it’s a series of boring, correct decisions compounded. Partition early. Index appropriately. Deduplicate religiously. Measure recall continuously. The code above gives you a foundation; your workload will dictate the exact parameters. Start with the checklist, iterate on the numbers, and keep the pipeline incremental.