Building a Modular Retrieval‑Augmented Generation Pipeline with Vector Stores

Opening: why RAG trips up the unwary

When you wire a large language model straight to a vector database without thinking about the surrounding plumbing, the first thing that tends to break is latency. A request that should finish in a couple of hundred milliseconds suddenly stretches to seconds because the retrieval step pulls from a remote service that is overloaded, or because the embedding index was built on stale data and returns irrelevant matches. The symptom looks like a flaky AI feature – users get hallucinations or empty answers – and the root cause is usually a missing layer of resilience. In this post I walk through a modular design that keeps the retrieval part predictable, whether you run FAISS on‑prem, spin up Milvus in a Kubernetes cluster, or call a managed Pinecone endpoint.

Background: why modularity matters now

Production teams are moving from proof‑of‑concept notebooks to microservices that serve millions of queries per day. At that scale the choice of vector store becomes a cost and performance decision, not just a convenience. On‑prem solutions like FAISS give you full control over hardware and can be tuned for sub‑millisecond ANN search, but you have to manage sharding, replication, and backups yourself. Cloud‑native options such as Pinecone or Milvus‑managed services offload operational burden, yet they introduce network hops and pricing tiers that can surprise you when traffic spikes. A modular pipeline lets you swap the store without rewriting the surrounding code, and it gives you a place to plug in latency guards and fallback strategies that keep the user experience smooth.

Technical deep dive

Below is a stripped‑down Python example that shows the core pieces of a RAG pipeline built around langchain‑style abstractions. The design separates three concerns:

• Embedding provider: a function that turns raw text into a fixed‑size vector. I use sentence‑transformers because it balances speed and quality for most domain vocabularies.
• Vector store adapter: a thin wrapper that implements add_documents, similarity_search, and delete_documents. The wrapper can point at FAISS, Milvus, or Pinecone based on a simple configuration flag.
• Retrieval‑augmented generation step: fetches the top‑k chunks, formats a prompt, and sends it to the LLM.

Here is the code. Notice the use of try/except around the search call – that is where we insert fallback logic for missing or stale embeddings.

import os
from typing import List
from sentence_transformers import SentenceTransformer

# 1. Embedding provider ----------------------------------------------------
model = SentenceTransformer("all-MiniLM-L6-v2")

def embed_text(texts: List[str]):
    return model.encode(texts, convert_to_tensor=False).tolist()

# 2. Vector store adapter --------------------------------------------------
class VectorStore:
    def __init__(self, backend: str, index_path: str = None, url: str = None):
        self.backend = backend.lower()
        if self.backend == "faiss":
            from faiss import IndexFlatL2
            self.index = IndexFlatL2(model.get_sentence_embedding_dimension())
            self.id_to_doc = {}
        elif self.backend == "milvus":
            from pymilvus import Collection, connections, FieldSchema, CollectionSchema, DataType
            connections.connect("default", host=url.split(":")[0], port=url.split(":")[1])
            fields = [
                FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
                FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=model.get_sentence_embedding_dimension())
            ]
            schema = CollectionSchema(fields, "rag_collection")
            self.collection = Collection("rag_collection", schema)
        elif self.backend == "pinecone":
            import pinecone
            pinecone.init(api_key=os.getenv("PINECONE_API_KEY"), environment="us-west1-gcp")
            self.index = pinecone.Index("rag-index")
        else:
            raise ValueError(f"Unsupported backend: {backend}")

    def add_documents(self, docs: List[str]):
        embeddings = embed_text(docs)
        if self.backend == "faiss":
            ids = list(range(len(self.id_to_doc), len(self.id_to_doc) + len(docs)))
            self.index.add_with_ids(np.array(embeddings).astype('float32'), np.array(ids))
            for i, doc in zip(ids, docs):
                self.id_to_doc[i] = doc
        elif self.backend == "milvus":
            ids = list(range(self.collection.num_entities, self.collection.num_entities + len(docs)))
            self.collection.insert([ids, embeddings])
        elif self.backend == "pinecone":
            ids = [str(uuid.uuid4()) for _ in docs]
            self.index.upsert(vectors=zip(ids, embeddings), namespace="rag")
        return ids

    def similarity_search(self, query: str, k: int = 5):
        query_vec = embed_text([query])[0]
        try:
            if self.backend == "faiss":
                D, I = self.index.search(np.array([query_vec]).astype('float32'), k)
                return [self.id_to_doc[i] for i in I[0] if i != -1]
            elif self.backend == "milvus":
                search_params = {"metric_type": "L2", "params": {"nprobe": 10}}
                results = self.collection.search([query_vec], "embedding", param=search_params, limit=k)
                return [hit.entity.get("id") for hit in results[0]]
            elif self.backend == "pinecone":
                resp = self.index.query(vector=query_vec, top_k=k, include_metadata=True, namespace="rag")
                return [match.metadata["text"] for match in resp.matches]
        except Exception as e:
            # fallback: return empty list and let the caller decide what to do
            print(f"Retrieval error: {e}")
            return []

# 3. Generation step ------------------------------------------------------
def rag_generate(query: str, store: VectorStore, llm):
    chunks = store.similarity_search(query, k=4)
    if not chunks:
        # Simple fallback – ask the model without context
        prompt = f"Answer the question without external context: {query}"
    else:
        context = "\n---\n".join(chunks)
        prompt = f"Use the following excerpts to answer the question.\n{context}\n\nQuestion: {query}\nAnswer:"
    return llm.complete(prompt)

The snippet demonstrates a few trade‑offs:

• FAISS lives in memory, so similarity_search is fast but you lose durability unless you serialize the index after each batch.
• Milvus requires a running service; you get built‑in replication but you also inherit network latency.
• Pinecone hides the infrastructure, but you pay per query and you are limited to the region you provision.
• The try/except block gives you a clean place to add a cache miss counter, trigger a re‑embedding job, or fall back to a pure LLM answer.

Common pitfalls and what tends to go wrong

Even with a clean abstraction, several things bite developers:

• Stale embeddings – if the source documents change but you don’t rebuild the index, the similarity scores become misleading. A lightweight scheduler that re‑embeds only the changed files keeps the index fresh without full rebuilds.
• Dimension mismatch – switching the embedding model without updating the index schema throws obscure errors from the ANN library. Always version the schema alongside the model.
• Unbounded result sets – asking for too many neighbors (e.g., k=100) on a small index can cause the search to fall back to a brute‑force scan, inflating latency dramatically.
• Network jitter – cloud vector stores add a round‑trip. If your service runs inside the same VPC, use private endpoints; otherwise, add a retry‑with‑backoff wrapper around the search call.
• Memory pressure – loading a million‑vector FAISS index into RAM on a modest VM leads to swapping and timeouts. Consider IVF‑based indexes or chunk the data across multiple shards.

All of these issues can be mitigated by a health‑check endpoint that reports index size, average query latency, and a “stale‑percentage” metric derived from a checksum of source documents.

Practical implementation guide

Here is a step‑by‑step checklist you can copy into your repo’s README:

1. Pick the backend. For a prototype on a single server, start with FAISS. For horizontal scaling, provision Milvus in a Kubernetes stateful set or enable a Pinecone index in the same cloud region as your API.
2. Define the schema. Record the embedding dimension, a unique document ID, and a text field for fallback retrieval. Keep the schema version in a config.yaml file.
3. Set up the embedding service. Install sentence‑transformers version 2.2.2 and pin the model name. Write a small CLI that reads a directory of markdown files, computes embeddings, and writes them to the chosen store.
4. Implement a refresh job. Use watchdog or a file‑system event stream to detect changes. When a file updates, recompute its vector and call delete_documents followed by add_documents on the store.
5. Wrap the search call. Add a decorator that measures latency, retries on TimeoutError, and falls back to an empty list after three attempts. Log the event with a correlation ID for tracing.
6. Integrate with the LLM. Build the prompt exactly as shown in the code example – separate chunks with a clear delimiter, and keep the total token count under the model’s limit (usually 4 KB for GPT‑3.5‑turbo).
7. Expose health metrics. Use Prometheus gauges for index_size, average_latency_ms, and stale_ratio. Alert if latency crosses a threshold you measured during load testing.
8. Deploy and monitor. Deploy the service as a Docker container, configure the backend endpoint via environment variables, and watch the latency graph. If you see spikes, consider adding a local cache (e.g., LRU with functools.lru_cache) for the most frequent queries.

Following these steps gives you a pipeline that can be swapped from FAISS to Milvus to Pinecone by changing a single line in .env – no code rewrite needed.

Closing thoughts: what to keep in mind

The biggest lesson I keep returning to is that retrieval is a first‑class citizen in a RAG system, not an afterthought. Treat the vector store like any other database: monitor its health, version its schema, and plan for data drift. Latency is often the make‑or‑break factor, so measure it early and decide whether the convenience of a managed service outweighs the raw speed of an on‑prem index. Finally, don’t forget the fallback path – a well‑written prompt that can answer without external context is a safety net that saves you from silent failures.

Take the next step

If you want a ready‑to‑run starter repo that follows this pattern, grab it from akkistech.com and start swapping vector stores today.