How to set up a RAG pipeline

Overview

Retrieval-augmented generation gives a model access to a private corpus without fine-tuning: split documents into chunks, embed them, store the vectors, retrieve the closest chunks for each question, and answer from them. This guide builds the smallest working version on pgvector. Strategy lives elsewhere: rag, rag-chunking, rag-retrieval.

Prerequisites

• Python 3.10 or newer and pip install "psycopg[binary]" pgvector voyageai langchain-text-splitters anthropic.
• A Postgres database with pgvector; see configure-pgvector-on-neon.
• VOYAGE_API_KEY, ANTHROPIC_API_KEY, and DATABASE_URL set in the environment.

Steps

1. Chunk

from langchain_text_splitters import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)  # characters
chunks = splitter.split_text(open("docs/handbook.md").read())

chunk_size counts characters here, so 2,000 is roughly 500 tokens. Treat the size as a starting point and tune it against retrieval tests (rag-chunking).

2. Embed

import voyageai
 
vo = voyageai.Client()   # reads VOYAGE_API_KEY
MODEL = "voyage-4"       # 1024 dimensions by default
 
def embed(texts, input_type):
    out = []
    for i in range(0, len(texts), 128):
        out += vo.embed(texts[i:i + 128], model=MODEL, input_type=input_type).embeddings
    return out
 
vectors = embed(chunks, "document")

Use input_type="document" when indexing and "query" when searching; mixing them lowers recall. voyage-3 is a legacy model. Model choices: embeddings.

3. Store

CREATE EXTENSION IF NOT EXISTS vector;
CREATE TABLE chunks (
  id bigserial PRIMARY KEY, source text NOT NULL, content text NOT NULL,
  embedding vector(1024) NOT NULL
);
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);

import os, psycopg
from pgvector import Vector
from pgvector.psycopg import register_vector
 
conn = psycopg.connect(os.environ["DATABASE_URL"], autocommit=True)
register_vector(conn)
with conn.cursor() as cur:
    cur.executemany(
        "INSERT INTO chunks (source, content, embedding) VALUES (%s, %s, %s)",
        [(f"handbook.md#{i}", c, Vector(v)) for i, (c, v) in enumerate(zip(chunks, vectors))],
    )

The vector dimension must equal the model’s output. Wrap every embedding in Vector(...) (or pass a NumPy array): a plain Python list is sent as double precision[], which inserts through an assignment cast but fails in the <=> query.

4. Retrieve

def retrieve(question, k=5):
    q = embed([question], "query")[0]
    return conn.execute(
        "SELECT source, content FROM chunks ORDER BY embedding <=> %s LIMIT %s", (Vector(q), k)
    ).fetchall()

Add a reranker once basic retrieval works; see rag-reranking.

5. Answer with citations

Put the sources in the user turn inside tags, with instructions in the system prompt.

import anthropic
 
client = anthropic.Anthropic()
 
def answer(question):
    docs = "\n".join(f'<source id="{s}">{c}</source>' for s, c in retrieve(question))
    msg = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1024,
        system="Answer only from the provided sources and cite the source id in brackets. "
               "If the sources do not contain the answer, say so.",
        messages=[{"role": "user", "content": f"<sources>\n{docs}\n</sources>\n\nQuestion: {question}"}],
    )
    return next(b.text for b in msg.content if b.type == "text")

Citation formats and the API’s native citation support are covered in rag-citations.

Verify it worked

psql "$DATABASE_URL" -c "SELECT count(*) FROM chunks;"
python -c "from pipeline import answer; print(answer('What is the onboarding process?'))"

The count equals the number of chunks, and the answer cites at least one handbook.md#N id. Test retrieval alone with questions whose answers you know, and check the right chunk is in the top five before blaming the model.

Common errors

• Irrelevant chunks. A wrong input_type, oversized chunks, or no overlap; fix retrieval before changing the prompt.
• operator does not exist: vector <=> double precision[]. The query embedding was passed as a plain list; wrap it in Vector(...) and call register_vector(conn). If the error names vector as a missing type, the extension is not enabled in this database.
• different vector dimensions. The column does not match the embedding model.
• The model answers beyond the sources. Keep the “answer only from sources” instruction and a refusal path.

FAQ

What chunk size should I start with for RAG?

Start at 2,000 characters with 200 characters of overlap, roughly 500 tokens, and tune it against retrieval tests.

Why does my pgvector query fail with operator does not exist?

The query embedding was passed as a plain Python list, which is sent as double precision[]; wrap it in Vector(...) and call register_vector(conn).

Do I use the same Voyage input_type for documents and queries?

No; use input_type="document" when indexing and "query" when searching, because mixing them lowers recall.

Should I fix the prompt or the retrieval when RAG answers are wrong?

Fix retrieval first: check that the right chunk is in the top five for questions whose answers you know before blaming the model.

Related

• rag
• rag-chunking
• rag-retrieval
• embeddings
• rag-citations
• rag-reranking
• configure-pgvector-on-neon
• pinecone-vs-pgvector