Build a RAG Search Pipeline: A Practical Guide

You have a language model and a collection of private documents. The model does not know what is in those documents, and fine-tuning would be slow and expensive. The standard fix is retrieval-augmented generation: find the relevant context at query time, pass it to the model, and get an answer grounded in your actual data.

The gap between a working prototype and a reliable pipeline is almost always in the details. How you chunk documents, what metadata you store, which similarity metric you use, and whether you measure retrieval quality at all. This guide walks through each of those decisions with working Python code, ending with a pipeline you can drop into a FastAPI service.

What you will build#

By the end of this guide you will have a working RAG search pipeline that:

• Loads and chunks documents using LangChain's RecursiveCharacterTextSplitter
• Embeds chunks with OpenAI's text-embedding-3-small model
• Stores vectors in PostgreSQL with the pgvector extension
• Retrieves the top-k most relevant chunks for a query
• Passes context to an LLM and returns a grounded answer
• Measures retrieval precision using a simple offline evaluation loop

The code targets Python 3.11+ and uses async-compatible patterns so it can be embedded inside a FastAPI service.

Prerequisites you need first#

You do not need to run a separate vector database. pgvector is a PostgreSQL extension that adds a vector column type and HNSW / IVFFlat indexes directly inside Postgres, which reduces infrastructure footprint significantly (pgvector on GitHub).

Architecture overview#

A RAG system has two separate data flows that share one vector store. Ingestion turns documents into searchable vectors. Retrieval answers a single query at request time. Both read and write the same pgvector table, but they run on different schedules.

Keeping ingestion and retrieval separate matters for production. Ingestion runs on a schedule or as a background job, while retrieval runs synchronously with user requests. If you mix them, re-ingestion latency leaks into query latency.

The original RAG paper by Lewis et al. (NeurIPS 2020, arXiv:2005.11401) frames this as combining parametric memory (the model's weights) with non-parametric memory (the retrieval index). The retrieval index is what you control, and it is the main lever for answer quality.

Step-by-step implementation#

-- Run in psql or a migration script
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
    id         SERIAL PRIMARY KEY,
    content    TEXT NOT NULL,
    source     TEXT,
    page       INT,
    embedding  vector(1536)
);

CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops);

from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_and_chunk(text: str, source: str) -> list[dict]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = splitter.split_text(text)
    return [
        {"content": chunk, "source": source, "chunk_index": i}
        for i, chunk in enumerate(chunks)
    ]

import openai
import asyncio

client = openai.AsyncOpenAI()

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

async def embed_in_batches(chunks: list[str], batch_size: int = 100) -> list[list[float]]:
    all_embeddings = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i : i + batch_size]
        embeddings = await embed_chunks(batch)
        all_embeddings.extend(embeddings)
    return all_embeddings

import psycopg2
from pgvector.psycopg2 import register_vector

def get_connection(dsn: str):
    conn = psycopg2.connect(dsn)
    register_vector(conn)
    return conn

def insert_chunks(conn, chunks: list[dict], embeddings: list[list[float]]) -> None:
    with conn.cursor() as cur:
        records = [
            (c["content"], c["source"], c.get("chunk_index"), emb)
            for c, emb in zip(chunks, embeddings)
        ]
        cur.executemany(
            """
            INSERT INTO documents (content, source, page, embedding)
            VALUES (%s, %s, %s, %s)
            """,
            records,
        )
    conn.commit()

async def retrieve(conn, query: str, top_k: int = 5) -> list[dict]:
    query_embedding = await embed_chunks([query])
    vec = query_embedding[0]

    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT content, source, page,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            ORDER BY embedding <=> %s::vector
            LIMIT %s
            """,
            (vec, vec, top_k),
        )
        rows = cur.fetchall()

    return [
        {"content": row[0], "source": row[1], "page": row[2], "similarity": row[3]}
        for row in rows
    ]

async def answer(query: str, context_chunks: list[dict]) -> str:
    context_text = "\n\n---\n\n".join(
        f"[Source: {c['source']}, page {c['page']}]\n{c['content']}"
        for c in context_chunks
    )
    messages = [
        {
            "role": "system",
            "content": (
                "You are a helpful assistant. Answer the question using only the "
                "provided context. If the context does not contain enough information "
                "to answer, say so explicitly."
            ),
        },
        {
            "role": "user",
            "content": f"Context:\n{context_text}\n\nQuestion: {query}",
        },
    ]
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        temperature=0.2,
    )
    return response.choices[0].message.content

Test and verify the result#

Run a quick end-to-end check before building evaluation infrastructure:

import asyncio

DSN = "postgresql://user:password@localhost:5432/ragdb"

async def main():
    conn = get_connection(DSN)

    # Ingest a sample document
    sample_text = """
    pgvector is a PostgreSQL extension for vector similarity search.
    It supports exact and approximate nearest neighbor search using
    HNSW and IVFFlat indexes. Cosine, L2, and inner product distance
    operators are available.
    """
    chunks = load_and_chunk(sample_text, source="sample.txt")
    texts = [c["content"] for c in chunks]
    embeddings = await embed_in_batches(texts)
    insert_chunks(conn, chunks, embeddings)

    # Query
    query = "What index types does pgvector support?"
    results = await retrieve(conn, query, top_k=3)
    response = await answer(query, results)

    print("Answer:", response)
    for r in results:
        print(f"  [{r['similarity']:.3f}] {r['content'][:80]}...")

    conn.close()

asyncio.run(main())

Expected output should include the answer mentioning HNSW and IVFFlat, with similarity scores above 0.80 for a well-matched query.

Common mistakes#

Not measuring retrieval before blaming the model. When answers are wrong or vague, the first question is whether the right chunks were retrieved, not whether the prompt was good. Log every retrieval result and its similarity score during development.

Chunk size set by intuition, not measurement. A 512-character chunk is a starting point. The right chunk size depends on your document type, query length, and the model's context window. Test three or four sizes and measure hit rate against a labeled evaluation set.

Storing embeddings without metadata. A vector without a source, date, and document ID cannot be filtered, audited, or invalidated. Metadata is what lets you remove a document from the index when it becomes outdated.

Using a different model at query time than ingestion time. Embeddings from text-embedding-3-small and text-embedding-ada-002 are incompatible. If you migrate models, you must re-embed everything.

Sending all retrieved chunks to the model regardless of similarity score. Low-scoring chunks add noise. Set a minimum similarity threshold and exclude results below it. Better to pass two relevant chunks than five irrelevant ones.

Production considerations#

Index type and maintenance. HNSW indexes in pgvector require no training data upfront, but consume more memory as they grow. For datasets over five million vectors, budget for dedicated memory on your database instance or evaluate purpose-built vector stores. See the pgvector vs Weaviate vs Qdrant comparison for when a standalone vector database becomes the better choice.

Re-ingestion and deletion. When a source document changes, you need to delete its old chunks and re-ingest. Store the source file path as metadata and use it as a deletion key:

DELETE FROM documents WHERE source = 'contracts/agreement_v2.pdf';

Observability. Log query latency, retrieval count, average similarity score, and token usage per request. A sudden drop in average similarity score often signals that a new document type was ingested with incompatible formatting. See our notes on observability for AI systems for a fuller treatment.

Evaluation at scale. Build a labeled dataset of query–answer pairs for your domain. Measure hit rate (did the correct chunk appear in the top-k results?) and answer faithfulness (does the answer contradict the context?). Automated evaluation frameworks like RAGAS provide both metrics without requiring human review for every run. For a deeper look, read evaluating RAG quality as an experiment.

Rate limits and cost. OpenAI's embedding API has token-per-minute limits. At scale, use batch ingestion with exponential back-off and consider caching embeddings for documents that do not change frequently. The $0.02 per million token rate for text-embedding-3-small makes this economical for most use cases.

Key Takeaways#

A RAG pipeline's quality is determined by three compounding decisions: how you chunk (determines what fits in each retrieval unit), how you embed (determines the geometry of your search space), and how you retrieve (determines whether the right context reaches the model). All three must be correct for answers to be reliable.

pgvector is a practical starting point for teams already using PostgreSQL. It avoids a new infrastructure dependency while supporting production-grade HNSW indexes. When your index grows beyond a few million vectors or your query patterns require advanced filtering, a dedicated vector store becomes worth the operational cost.

Measure retrieval precision before optimizing generation quality. Almost all RAG failures trace back to retrieval, not to the model's reasoning.

FAQ#

How many chunks should I retrieve (top-k)?

Start with 5. If your documents are long and dense, 3–5 focused chunks outperform 10 noisy ones. If your documents are short reference items (FAQs, database records), 8–10 may be appropriate. Measure the tradeoff between context coverage and token cost.

Can I use this pipeline without OpenAI?

Yes. Any embedding model that produces a fixed-dimension dense vector works. Sentence Transformers (via HuggingFace) and Cohere's Embed API are common alternatives. Just ensure chunk_size and chunk_overlap remain appropriate for the token limit of your chosen model.

How do I handle documents in multiple languages?

Multilingual embedding models like text-embedding-3-large or Cohere's multilingual model map different languages into a shared vector space, so cross-language retrieval is possible. Evaluate retrieval quality per language separately, as coverage varies across languages.

When should I switch from pgvector to a standalone vector database?

The main signals are index size (tens of millions of vectors), need for hybrid search (keyword and vector in a single query), or filtering complexity (many metadata dimensions that narrow the search space before ANN runs). See the pinecone vs weaviate vs qdrant comparison for a decision framework.

Does chunk overlap help or hurt performance?

Overlap helps recall at a small storage cost. A 64-character overlap on 512-character chunks adds roughly 12% more storage but recovers sentences split across boundaries. The main risk is duplicate content in the context window if two overlapping chunks from the same passage are both retrieved, so deduplicate on source and chunk index before passing to the model.