RAG Pipelines and Vector Databases: Production Retrieval-Augmented Generation

1. RAG Architecture Overview

Retrieval-Augmented Generation (RAG) is the dominant pattern for grounding LLM responses in external knowledge. Instead of relying solely on a model's parametric memory (which is frozen at training time and prone to hallucination), RAG retrieves relevant documents from a knowledge base at inference time and includes them in the prompt context. This gives the model access to current, domain-specific, and verifiable information -- the difference between an LLM that guesses and one that cites sources.

A production RAG system has three phases: ingestion (processing documents into searchable chunks with embeddings stored in a vector database), retrieval (finding the most relevant chunks for a given query using vector similarity, keyword search, or hybrid approaches), and generation (feeding retrieved context to an LLM to produce a grounded answer). Each phase has its own set of engineering decisions that compound into overall system quality.

2. Document Processing

The quality of your RAG system is bounded by the quality of your document processing. Garbage in, garbage out applies doubly here: poorly extracted text leads to bad chunks, which lead to irrelevant retrievals, which lead to hallucinated answers. Invest heavily in document processing -- it is the highest-leverage improvement you can make to a RAG pipeline.

3. Chunking Strategies

Chunking determines the granularity of your knowledge base. Chunks too large dilute the embedding with irrelevant information and waste context window tokens. Chunks too small lose the context needed for the LLM to produce coherent answers. The optimal chunk size depends on your document type, embedding model, and use case -- but most production systems land between 256 and 1024 tokens per chunk.

4. Embedding Models

Embedding models convert text into dense vectors that capture semantic meaning. The choice of embedding model determines retrieval quality: it sets the ceiling for how well your RAG system can match queries to relevant documents. As of April 2026, the landscape includes both proprietary API models (OpenAI, Cohere, Google, Voyage) and open-source models (BGE, E5, GTE, Jina) that rival or exceed proprietary options on benchmarks like MTEB.

5. Vector Databases Comparison

Vector databases store embeddings and enable fast similarity search at scale. The choice depends on your deployment model (managed vs self-hosted), scale requirements (thousands vs billions of vectors), query patterns (pure vector vs hybrid search), and existing infrastructure. All production-grade vector databases support HNSW indexing, metadata filtering, and sub-100ms query latency for million-scale collections.

6. Retrieval Strategies

Retrieval is where most RAG quality issues originate. A perfect embedding model with a naive top-k retrieval strategy will underperform a decent embedding model with thoughtful retrieval engineering. The strategies below progress from simple to advanced, and production systems typically combine multiple approaches.

7. Reranking

Reranking is the highest-leverage post-retrieval improvement. Initial retrieval (embedding similarity) is a fast but imprecise first pass. Rerankers use more expensive models to re-score and reorder the top-k results based on deep query-document interaction. A bi-encoder (embedding model) processes query and document independently; a cross-encoder processes them jointly, attending to interactions between query and document tokens. This joint processing catches relevance signals that independent encoding misses.

8. Advanced RAG Patterns

Standard RAG (retrieve-then-generate) hits a ceiling for complex questions that require multi-step reasoning, self-correction, or heterogeneous data sources. Advanced RAG patterns extend the basic architecture with agentic behavior, graph-based knowledge, multi-modal retrieval, and self-evaluation loops. These patterns increase system complexity but unlock capabilities that simple RAG cannot achieve.

9. Evaluation (RAGAS Framework)

You cannot improve what you do not measure. RAG evaluation is harder than standard ML evaluation because you need to assess both retrieval quality and generation quality independently, and their interaction. The RAGAS (Retrieval Augmented Generation Assessment) framework is the most widely adopted evaluation toolkit, providing automated metrics that correlate well with human judgments without requiring ground-truth labels.

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# Prepare evaluation dataset
eval_data = Dataset.from_dict({
    "question": questions,
    "answer": generated_answers,
    "contexts": retrieved_contexts,      # list[list[str]]
    "ground_truth": reference_answers     # for context_recall
})

# Run evaluation
results = evaluate(
    dataset=eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=ChatOpenAI(model="gpt-4o"),       # evaluator LLM
    embeddings=OpenAIEmbeddings()
)
print(results)
# {'faithfulness': 0.87, 'answer_relevancy': 0.91,
#  'context_precision': 0.78, 'context_recall': 0.82}

10. Production Deployment

Moving RAG from a notebook to production involves caching, streaming, multi-tenancy, monitoring, and cost optimization. The difference between a demo and a production system is not the retrieval algorithm -- it is the infrastructure that makes the algorithm reliable, fast, and affordable at scale.

11. Code Examples

Three implementation approaches from high-level frameworks to direct API calls. LangChain provides the fastest path to a working RAG system with the most abstractions. LlamaIndex is purpose-built for RAG with deeper indexing primitives. Direct API calls give maximum control and minimum dependencies for teams that prefer explicit code over framework magic.

LangChain RAG Pipeline

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Qdrant
from langchain_community.document_loaders import PyMuPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# 1. Load and chunk documents
loader = PyMuPDFLoader("technical_manual.pdf")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_documents(docs)

# 2. Embed and store in Qdrant
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Qdrant.from_documents(
    chunks, embeddings,
    url="http://localhost:6333",
    collection_name="tech_manual"
)

# 3. Create retrieval chain with reranking
retriever = vectorstore.as_retriever(
    search_type="mmr",          # Maximum Marginal Relevance
    search_kwargs={"k": 10, "fetch_k": 25, "lambda_mult": 0.7}
)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer based on the context below. Cite sources.
If the context doesn't contain the answer, say so.

Context: {context}"""),
    ("human", "{input}")
])

llm = ChatOpenAI(model="gpt-4o", temperature=0)
chain = create_retrieval_chain(
    retriever,
    create_stuff_documents_chain(llm, prompt)
)

# 4. Query
result = chain.invoke({"input": "How do I configure failover?"})
print(result["answer"])

LlamaIndex RAG Pipeline

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.pinecone import PineconeVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.anthropic import Anthropic
from llama_index.core.postprocessor import SentenceTransformerRerank
from pinecone import Pinecone

# 1. Configure models
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-large")
Settings.llm = Anthropic(model="claude-sonnet-4-5-20250514", temperature=0)

# 2. Load documents with metadata
documents = SimpleDirectoryReader(
    input_dir="./knowledge_base",
    recursive=True,
    filename_as_id=True
).load_data()

# 3. Parse with sentence-aware splitting
parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)

# 4. Create Pinecone index
pc = Pinecone(api_key="your-api-key")
pinecone_index = pc.Index("rag-production")
vector_store = PineconeVectorStore(pinecone_index=pinecone_index)

index = VectorStoreIndex.from_documents(
    documents, vector_store=vector_store
)

# 5. Query with reranking
reranker = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3", top_n=5
)
query_engine = index.as_query_engine(
    similarity_top_k=20,
    node_postprocessors=[reranker],
    response_mode="tree_summarize"
)

response = query_engine.query("Explain the backup recovery procedure")
print(response)
for node in response.source_nodes:
    print(f"  [{node.score:.3f}] {node.metadata['file_name']}")

Direct API: Pinecone + OpenAI

import openai
from pinecone import Pinecone

client = openai.OpenAI()
pc = Pinecone(api_key="your-api-key")
index = pc.Index("rag-production")

def embed(text: str, input_type: str = "search_query") -> list[float]:
    """Generate embedding for text."""
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=1536  # Matryoshka shortening for cost savings
    )
    return response.data[0].embedding

def retrieve(query: str, top_k: int = 10, filters: dict = None) -> list[dict]:
    """Retrieve relevant chunks from Pinecone."""
    query_vector = embed(query)
    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True,
        filter=filters  # e.g., {"source": "engineering-docs", "year": {"$gte": 2025}}
    )
    return [
        {"text": m.metadata["text"], "source": m.metadata["source"], "score": m.score}
        for m in results.matches
    ]

def generate(query: str, contexts: list[dict]) -> str:
    """Generate answer grounded in retrieved context."""
    context_str = "\n\n---\n\n".join(
        f"[Source: {c['source']}] {c['text']}" for c in contexts
    )
    response = client.chat.completions.create(
        model="gpt-4o",
        temperature=0,
        messages=[
            {"role": "system", "content": f"""Answer based on the context below.
Cite sources using [Source: ...] format.
If context is insufficient, say "I don't have enough information."

Context:
{context_str}"""},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    chunks = []
    for chunk in response:
        if chunk.choices[0].delta.content:
            chunks.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end="", flush=True)
    return "".join(chunks)

# RAG query
query = "What are the SLA requirements for the payment service?"
contexts = retrieve(query, top_k=10, filters={"team": "payments"})
answer = generate(query, contexts)

pgvector with Python (Direct SQL)

import psycopg2
import openai
import json

client = openai.OpenAI()
conn = psycopg2.connect("postgresql://user:pass@localhost:5432/ragdb")

# Setup: create table with vector column
with conn.cursor() as cur:
    cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
    cur.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id SERIAL PRIMARY KEY,
            content TEXT NOT NULL,
            source VARCHAR(255),
            embedding vector(1536),
            metadata JSONB DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ DEFAULT NOW()
        )
    """)
    cur.execute("""
        CREATE INDEX IF NOT EXISTS docs_embedding_idx
        ON documents USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 200)
    """)
    conn.commit()

def ingest(content: str, source: str, metadata: dict = None):
    """Embed and store a document chunk."""
    resp = client.embeddings.create(
        model="text-embedding-3-large", input=content, dimensions=1536
    )
    embedding = resp.data[0].embedding
    with conn.cursor() as cur:
        cur.execute(
            """INSERT INTO documents (content, source, embedding, metadata)
               VALUES (%s, %s, %s::vector, %s::jsonb)""",
            (content, source, str(embedding), json.dumps(metadata or {}))
        )
    conn.commit()

def search(query: str, top_k: int = 5, source_filter: str = None) -> list:
    """Hybrid search: vector similarity + optional source filter."""
    q_emb = client.embeddings.create(
        model="text-embedding-3-large", input=query, dimensions=1536
    ).data[0].embedding

    sql = """
        SELECT content, source,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        WHERE 1=1
    """
    params = [str(q_emb)]
    if source_filter:
        sql += " AND source = %s"
        params.append(source_filter)
    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params.extend([str(q_emb), top_k])

    with conn.cursor() as cur:
        cur.execute(sql, params)
        return [
            {"content": r[0], "source": r[1], "score": r[2]}
            for r in cur.fetchall()
        ]