Building production RAG pipelines on object storage

Most RAG tutorials end where production begins. They show you how to embed a PDF, store the chunks in a vector database, and run a query. Then you ship to production and discover that chunking strategy matters enormously, that retrieval quality degrades as your corpus grows, that multi-tenant isolation is harder than expected, and that your vector database bill is becoming a line item on the CFO’s radar.

This post covers what the tutorials skip: how to build a RAG pipeline that stays correct, performant, and affordable as your corpus scales from thousands to hundreds of millions of chunks.

The pipeline architecture

A production RAG pipeline has five distinct stages:

Documents → Chunking → Embedding → Indexing → Retrieval → LLM

Each stage has failure modes, quality tradeoffs, and operational concerns. Let’s go through them.

Stage 1: Document ingestion and chunking

Why chunking matters

LLMs have context windows. Embedding models have context windows. You can’t embed an entire 200-page document as one vector — the embedding would be a blurry average of everything in the document. Instead, you split documents into chunks and embed each chunk separately.

The chunk size determines retrieval granularity. Too small (e.g., sentence-level): chunks lack context, answers require stitching multiple chunks. Too large (e.g., page-level): chunks contain multiple topics, the embedding is diffuse, you retrieve the right page but the wrong answer is buried in it.

Practical chunk sizes

For most text RAG workloads, the sweet spot is 512-1024 tokens with 10-20% overlap between adjacent chunks.

from typing import Iterator

def chunk_document(text: str, chunk_size: int = 768, overlap: int = 128) -> Iterator[str]:
    tokens = tokenizer.encode(text)
    step = chunk_size - overlap
    for i in range(0, len(tokens), step):
        chunk = tokens[i:i + chunk_size]
        if len(chunk) < 64:  # Skip tiny trailing chunks
            break
        yield tokenizer.decode(chunk)

Semantic chunking

Fixed-size chunking often splits in the middle of a paragraph. Semantic chunking tries to split at paragraph or section boundaries. A simple implementation:

def semantic_chunk(text: str, max_tokens: int = 1024) -> list[str]:
    paragraphs = text.split("\n\n")
    chunks, current = [], []
    current_tokens = 0
    
    for para in paragraphs:
        para_tokens = len(tokenizer.encode(para))
        if current_tokens + para_tokens > max_tokens and current:
            chunks.append("\n\n".join(current))
            current, current_tokens = [], 0
        current.append(para)
        current_tokens += para_tokens
    
    if current:
        chunks.append("\n\n".join(current))
    return chunks

Metadata to always store

Every chunk should carry metadata: document ID, page number, section heading, creation date, author, source URL. This metadata enables:

• Metadata filtering at query time (“only search docs from Q4 2025”)
• Deduplication on re-ingest
• Attribution in LLM responses
• Selective updates when documents change

Stage 2: Embedding

Model selection

The embedding model determines your semantic space. Switching models requires re-embedding your entire corpus — a significant operational cost. Choose your model before you have millions of documents.

Current recommendations:

• text-embedding-3-small (OpenAI, 1536 dims): Best price/performance for general English text. Good multilingual coverage.
• text-embedding-3-large (OpenAI, 3072 dims): 10-15% better on MTEB benchmarks. Worth it for knowledge-intensive applications.
• Cohere embed-english-v3 (1024 dims): Competitive quality, lower per-token cost at high volume.
• self-hosted (E5-large, bge-m3): For air-gapped requirements or very high volume.

Batching for efficiency

Embedding APIs are network-bound. Batch as aggressively as the rate limits allow:

import asyncio
from openai import AsyncOpenAI

async def embed_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    client = AsyncOpenAI()
    response = await client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

async def embed_all(chunks: list[str], batch_size: int = 512) -> list[list[float]]:
    tasks = [
        embed_batch(chunks[i:i+batch_size])
        for i in range(0, len(chunks), batch_size)
    ]
    results = await asyncio.gather(*tasks)
    return [emb for batch in results for emb in batch]

Stage 3: Indexing into object storage

With embeddings in hand, you upsert into ManyVector. Store both the vector and the original chunk text as metadata — retrieval returns the text, which goes into the LLM context:

import manyvector

client = manyvector.ManyVector(api_key="mv-...")
ns = client.namespace("company-docs")

def index_chunks(chunks: list[dict], embeddings: list[list[float]]):
    records = []
    for chunk, embedding in zip(chunks, embeddings):
        records.append({
            "id": chunk["id"],
            "vector": embedding,
            "text": chunk["text"],           # stored, returned at query time
            "doc_id": chunk["doc_id"],
            "page": chunk["page"],
            "section": chunk["section"],
            "date": chunk["date"],
            "source": chunk["source_url"],
        })
    
    # Upsert in batches of 1000
    for i in range(0, len(records), 1000):
        ns.upsert(records[i:i+1000])

Idempotent ingestion

Use stable chunk IDs so re-ingesting a document is idempotent. A good ID scheme: {doc_id}:{chunk_index}. Re-upserting the same ID with the same vector is a no-op. Re-upserting with a new vector (because the document changed) correctly updates the chunk.

def chunk_id(doc_id: str, chunk_index: int) -> str:
    return f"{doc_id}:{chunk_index:05d}"

Handling document updates

When a document changes, you need to:

1. Delete all old chunks for that document
2. Re-chunk, re-embed, re-index

With ManyVector’s metadata filtering, you can find and delete all chunks for a given doc_id:

ns.delete(filter={"doc_id": {"$eq": old_doc_id}})

Stage 4: Retrieval

The basic retrieval call

def retrieve(query: str, top_k: int = 8, filters: dict = None) -> list[dict]:
    embedding = embed_query(query)
    
    results = ns.query(
        vector=embedding,
        text=query,          # enable hybrid search
        top_k=top_k,
        fusion="rrf",
        filters=filters,
        include_text=True,   # return stored text
    )
    
    return [
        {
            "text": r.text,
            "score": r.score,
            "doc_id": r.metadata["doc_id"],
            "source": r.metadata["source"],
        }
        for r in results
    ]

Multi-query retrieval

LLMs are good at generating multiple phrasings of a question. Running retrieval on multiple query variants improves recall:

async def multi_query_retrieve(original_query: str, top_k: int = 8) -> list[dict]:
    # Generate variants with an LLM
    variants = await generate_query_variants(original_query, n=3)
    all_queries = [original_query] + variants
    
    # Retrieve for each variant
    all_results = await asyncio.gather(*[
        retrieve_async(q, top_k=top_k) for q in all_queries
    ])
    
    # Deduplicate by doc_id, take highest score per document
    seen = {}
    for results in all_results:
        for r in results:
            doc_id = r["doc_id"]
            if doc_id not in seen or r["score"] > seen[doc_id]["score"]:
                seen[doc_id] = r
    
    return sorted(seen.values(), key=lambda x: x["score"], reverse=True)[:top_k]

Re-ranking

The vector + BM25 retrieval returns the most likely relevant chunks. A cross-encoder re-ranker verifies this ranking with more compute. Re-rankers are slow (they run the query and each document through a full encoder together), so run them on top-20 results and return top-5:

from cohere import Client

co = Client(api_key="...")

def rerank(query: str, results: list[dict], top_n: int = 5) -> list[dict]:
    reranked = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=[r["text"] for r in results],
        top_n=top_n,
    )
    return [results[r.index] for r in reranked.results]

The retrieval → re-rank pattern is worth the added latency (~100-200ms) for knowledge-intensive applications. The quality improvement is significant.

Stage 5: LLM synthesis

Pass retrieved chunks as context:

def synthesize(query: str, context_chunks: list[dict]) -> str:
    context = "\n\n".join([
        f"[Source: {c['source']}]\n{c['text']}"
        for c in context_chunks
    ])
    
    messages = [
        {
            "role": "system",
            "content": "Answer the question using only the provided context. If the context doesn't contain the answer, say so. Cite your sources."
        },
        {
            "role": "user",
            "content": f"Context:\n{context}\n\nQuestion: {query}"
        }
    ]
    
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        temperature=0,
    )
    return response.choices[0].message.content

Operational concerns

Multi-tenant isolation

For SaaS RAG (each customer has their own knowledge base), use per-tenant namespaces:

def get_tenant_namespace(tenant_id: str):
    return client.namespace(f"tenant-{tenant_id}")

With object-storage-backed namespaces, each tenant’s data is isolated within their own set of segment files in your bucket. There’s no risk of cross-tenant data leakage at the storage layer.

Incremental updates

Documents change. New documents are added. Your ingestion pipeline needs to handle this without full re-indexing:

def process_document_update(doc_id: str, new_content: str):
    # Delete old chunks
    ns.delete(filter={"doc_id": {"$eq": doc_id}})
    
    # Re-chunk, re-embed, re-index
    chunks = semantic_chunk(new_content)
    embeddings = embed_all(chunks)
    index_chunks(
        [{"id": chunk_id(doc_id, i), "text": c, "doc_id": doc_id, ...} 
         for i, c in enumerate(chunks)],
        embeddings
    )

Monitoring retrieval quality

Log every retrieval call with the query, retrieved chunks, and eventually the user’s feedback. Use this to:

• Identify queries with consistently low-quality retrievals
• Detect corpus coverage gaps (users asking questions your corpus can’t answer)
• Evaluate embedding model upgrades via forked namespace comparison

The key metric is not retrieval latency — it’s whether the retrieved chunks actually contain the answer. Build an eval set of (query, expected_answer_source) pairs and measure retrieval hit rate weekly.

Cost at scale

For a corpus of 10M chunks (typical medium-sized enterprise knowledge base):

• 10M chunks × 1536 dims × 4 bytes = 61.4GB vector storage
• At S3 Standard ($0.023/GB): $1.41/month for vector storage
• S3 GET costs for retrieval: at 1M queries/day, ~$1.50/day

Total object storage cost: under $50/month for a moderately busy knowledge base. Compare this to managed vector database costs for 10M vectors at equivalent query volume: typically $200-500/month.

The savings compound as the corpus grows. At 100M chunks, the gap between object-storage-native and RAM-first architectures is a 10-15x cost difference — and growing.

Next steps

The patterns described here are production-ready, but tuning for your specific workload matters. Start with:

1. Measure your baseline retrieval quality with an eval set before you optimize anything.
2. Experiment with chunk size on a small sample before indexing your full corpus.
3. Enable hybrid search if your users search for specific terminology.
4. Add re-ranking if retrieval quality is the bottleneck (usually is for knowledge-intensive applications).