Building a RAG pipeline with ManyVector

Retrieval-augmented generation (RAG) grounds your LLM responses in private knowledge. Instead of the model hallucinating an answer, it retrieves relevant document chunks from your vector store and uses them as context. This guide builds a complete RAG pipeline using ManyVector.

Architecture overview

A minimal RAG system has four components:

1. Ingestion pipeline — chunk documents, generate embeddings, upsert to ManyVector
2. Query pipeline — embed the user query, retrieve top-k chunks, inject into LLM context
3. ManyVector — stores embeddings in your S3/GCS/R2 bucket, serves hybrid vector + full-text search
4. LLM — generates grounded answers using the retrieved context

Step 1: Chunk your documents

Split documents into overlapping chunks. Overlap ensures important context isn’t cut off at a boundary:

def chunk_document(text: str, chunk_size: int = 400, overlap: int = 80) -> list[str]:
    words = text.split()
    chunks = []
    i = 0
    while i < len(words):
        chunk = " ".join(words[i : i + chunk_size])
        chunks.append(chunk)
        i += chunk_size - overlap
    return chunks

For production, use a semantic chunker (LangChain, LlamaIndex, or chonkie) that splits on paragraph boundaries rather than word counts.

Step 2: Set up ManyVector

from manyvector import ManyVector
import openai

mv_client = ManyVector(
    api_key="mv-...",
    backend={"provider": "s3", "bucket": "my-rag-bucket", "region": "us-east-1", ...}
)
oai_client = openai.OpenAI(api_key="sk-...")

ns = mv_client.create_namespace(
    name="company-knowledge-base",
    dimensions=1536,
    metric="cosine",
)

Step 3: Ingest documents

import hashlib

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

def ingest_document(document_id: str, text: str, metadata: dict):
    chunks = chunk_document(text)
    batch_size = 100

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i : i + batch_size]
        embeddings = embed_batch(batch)

        records = [
            {
                "id": f"{document_id}-chunk-{i + j}",
                "vector": emb,
                "text": chunk,
                "document_id": document_id,
                "chunk_index": i + j,
                **metadata,
            }
            for j, (chunk, emb) in enumerate(zip(batch, embeddings))
        ]
        ns.upsert(records)

    print(f"Ingested {len(chunks)} chunks from {document_id}")

Step 4: Build the retrieval function

Use hybrid search for best recall — vector search catches semantic matches, BM25 catches exact keywords:

def retrieve_context(
    query: str,
    top_k: int = 5,
    filters: dict | None = None,
) -> list[str]:
    # Embed the query
    query_embedding = embed_batch([query])[0]

    # Hybrid retrieval: vector + BM25 fused via RRF
    results = ns.query(
        vector=query_embedding,
        query_text=query,
        search_mode="hybrid",
        alpha=0.7,       # 70% vector, 30% BM25
        top_k=top_k,
        filters=filters,
    )

    return [r.attributes["text"] for r in results]

Step 5: Build the generation function

def rag_answer(
    question: str,
    filters: dict | None = None,
    model: str = "gpt-4o",
) -> str:
    # Retrieve relevant chunks
    context_chunks = retrieve_context(question, top_k=5, filters=filters)
    context = "\n\n---\n\n".join(context_chunks)

    # Generate answer grounded in context
    response = oai_client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant. Answer the question using only "
                    "the provided context. If the context doesn't contain enough "
                    "information to answer, say so clearly.\n\n"
                    f"Context:\n{context}"
                ),
            },
            {"role": "user", "content": question},
        ],
    )
    return response.choices[0].message.content

Step 6: Test the pipeline

# Ingest some documents
ingest_document(
    "engineering-faq",
    "Our vector search uses HNSW for approximate nearest-neighbour retrieval...",
    {"source": "internal-docs", "department": "engineering"},
)

# Query with a filter to only search engineering docs
answer = rag_answer(
    "How does our search handle approximate nearest neighbours?",
    filters={"department": {"$eq": "engineering"}},
)
print(answer)

Improving recall with re-ranking

For higher quality results, add a cross-encoder re-ranker after retrieval. Retrieve top-20 with ManyVector, then re-rank to top-5 with a cross-encoder:

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def retrieve_and_rerank(query: str, top_k: int = 5) -> list[str]:
    # Over-retrieve with ManyVector
    results = ns.query(
        vector=embed_batch([query])[0],
        query_text=query,
        search_mode="hybrid",
        top_k=20,
    )
    candidates = [r.attributes["text"] for r in results]

    # Re-rank with cross-encoder
    scores = reranker.predict([(query, c) for c in candidates])
    ranked = sorted(zip(scores, candidates), reverse=True)
    return [text for _, text in ranked[:top_k]]

Namespace strategy

Structure namespaces around your data boundaries:

Namespace	Contents
knowledge-base-prod	Production RAG corpus
knowledge-base-staging	Fork of prod for testing
knowledge-base-v2-eval	Re-indexed with new embedding model

Fork before making large changes:

# Fork production before re-indexing with a new model
fork = mv_client.fork_namespace(
    source="knowledge-base-prod",
    name="knowledge-base-v2-eval",
)
# Re-embed into fork, then promote if recall improves

Production checklist

• Set chunk size and overlap based on your LLM’s context window
• Enable hybrid search with a tuned alpha value (start at 0.7)
• Add metadata filtering to scope retrieval to relevant document sets
• Monitor retrieval quality using your eval framework
• Set up namespace branching before model migrations
• Configure S3 lifecycle policies for cost optimization