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