D
Dev SOPKnowledge Base
Search
← All topics

RAG Pipeline: Chunking, Embeddings, Vector Search, and Retrieval

Production RAG pipeline — document chunking strategy (500 tokens, 50 overlap), Voyage AI embeddings, pgvector in Supabase, hybrid search (semantic + BM25 with RRF), reranking, and augmented generation with citations.

ragembeddingsvector-dbsupabasepgvectorretrievalai
Agent trigger phrases: RAG pipeline · embeddings · vector search · pgvector · semantic search · Voyage AI · document chunking · hybrid search · retrieval augmented generation

Overview

RAG (Retrieval-Augmented Generation) grounds LLM responses in your documents. Pipeline: chunk → embed → store → search → augment → generate.

Embedding Model Selection

| Model | Provider | Dimensions | Use Case | |-------|---------|-----------|----------| | voyage-3 | Voyage AI | 1024 | General documents (best quality) | | voyage-3-lite | Voyage AI | 512 | High volume, cost-sensitive | | text-embedding-3-small | OpenAI | 1536 | Fallback, widely supported | | text-embedding-3-large | OpenAI | 3072 | Maximum quality (high cost) |

Voyage AI: 200M free tokens/month. Preferred over OpenAI for cost and quality balance.

Document Chunking

from typing import Generator

def chunk_document(
    text: str,
    chunk_size: int = 500,     # tokens approximate
    overlap: int = 50,
    metadata: dict = None
) -> Generator[dict, None, None]:
    """Chunk with overlap for context continuity."""
    words = text.split()
    metadata = metadata or {}

    i = 0
    chunk_index = 0
    while i < len(words):
        end = min(i + chunk_size, len(words))
        chunk_text = " ".join(words[i:end])

        yield {
            "content": chunk_text,
            "chunk_index": chunk_index,
            "word_start": i,
            "word_end": end,
            **metadata,
        }

        chunk_index += 1
        i = end - overlap  # move back by overlap amount

chunks = list(chunk_document(document_text, chunk_size=500, overlap=50))

Voyage AI Embeddings

import voyageai

client = voyageai.Client(api_key=os.environ["VOYAGE_API_KEY"])

def embed_texts(texts: list[str], input_type: str = "document") -> list[list[float]]:
    """
    input_type:
    - "document" for content being stored
    - "query" for search queries
    """
    result = client.embed(
        texts,
        model="voyage-3",
        input_type=input_type,
    )
    return result.embeddings

# Embed query differently from documents
query_embedding = embed_texts(["What is the refund policy?"], input_type="query")[0]
doc_embeddings = embed_texts([chunk["content"] for chunk in chunks], input_type="document")

Supabase pgvector Setup

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Documents table
CREATE TABLE documents (
  id BIGSERIAL PRIMARY KEY,
  source TEXT NOT NULL,
  chunk_index INTEGER NOT NULL,
  content TEXT NOT NULL,
  embedding vector(1024),  -- match Voyage-3 dimensions
  metadata JSONB DEFAULT '{}',
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Vector similarity index (HNSW for fast approximate search)
CREATE INDEX documents_embedding_idx
  ON documents USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

-- Full-text search index for BM25
CREATE INDEX documents_content_fts_idx
  ON documents USING gin(to_tsvector('english', content));

-- RLS
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

Storing Embeddings

import psycopg2
import json

def store_chunks(chunks: list[dict], embeddings: list[list[float]]):
    conn = psycopg2.connect(os.environ["DATABASE_URL"])

    with conn.cursor() as cur:
        for chunk, embedding in zip(chunks, embeddings):
            cur.execute(
                """
                INSERT INTO documents (source, chunk_index, content, embedding, metadata)
                VALUES (%s, %s, %s, %s::vector, %s)
                """,
                (
                    chunk["source"],
                    chunk["chunk_index"],
                    chunk["content"],
                    embedding,
                    json.dumps(chunk.get("metadata", {})),
                )
            )

    conn.commit()
    conn.close()

Semantic Search

def semantic_search(query: str, limit: int = 5) -> list[dict]:
    query_embedding = embed_texts([query], input_type="query")[0]

    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT content, source, chunk_index, metadata,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            ORDER BY embedding <=> %s::vector
            LIMIT %s
            """,
            (query_embedding, query_embedding, limit)
        )
        rows = cur.fetchall()

    conn.close()
    return [
        {"content": r[0], "source": r[1], "chunk_index": r[2], "metadata": r[3], "score": r[4]}
        for r in rows
    ]

Hybrid Search (Semantic + BM25 with RRF)

def hybrid_search(query: str, limit: int = 5, semantic_weight: float = 0.7) -> list[dict]:
    """Combine semantic and keyword search with Reciprocal Rank Fusion."""
    query_embedding = embed_texts([query], input_type="query")[0]

    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    with conn.cursor() as cur:
        cur.execute(
            """
            WITH semantic AS (
                SELECT id, content, source,
                       ROW_NUMBER() OVER (ORDER BY embedding <=> %s::vector) AS rank
                FROM documents
                ORDER BY embedding <=> %s::vector
                LIMIT 20
            ),
            keyword AS (
                SELECT id, content, source,
                       ROW_NUMBER() OVER (ORDER BY ts_rank(to_tsvector('english', content), plainto_tsquery('english', %s)) DESC) AS rank
                FROM documents
                WHERE to_tsvector('english', content) @@ plainto_tsquery('english', %s)
                LIMIT 20
            )
            SELECT
                COALESCE(s.id, k.id) AS id,
                COALESCE(s.content, k.content) AS content,
                COALESCE(s.source, k.source) AS source,
                (COALESCE(1.0/(60 + s.rank), 0) * %s + COALESCE(1.0/(60 + k.rank), 0) * %s) AS rrf_score
            FROM semantic s
            FULL OUTER JOIN keyword k ON s.id = k.id
            ORDER BY rrf_score DESC
            LIMIT %s
            """,
            (query_embedding, query_embedding, query, query, semantic_weight, 1 - semantic_weight, limit)
        )
        rows = cur.fetchall()

    conn.close()
    return [{"id": r[0], "content": r[1], "source": r[2], "score": r[3]} for r in rows]

Augmented Generation

import anthropic

def rag_query(question: str) -> str:
    # Retrieve
    chunks = hybrid_search(question, limit=5)

    # Format context
    context = "\n\n---\n\n".join(
        f"[Source: {c['source']}]\n{c['content']}"
        for c in chunks
    )

    # Generate
    client = anthropic.Anthropic()
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2_000,
        system=f"""Answer questions using ONLY the provided context. 
If the answer isn't in the context, say so clearly.
Cite sources with [Source: filename].

Context:
{context}""",
        messages=[{"role": "user", "content": question}],
    )

    return response.content[0].text

Embedding Update Strategy

When content changes, re-embed only the changed chunks:

import hashlib

def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()[:16]

# Store hash with each chunk
# On update: compare hash → only re-embed changed chunks
# This keeps embedding costs minimal on incremental ingestion