Supabase as a Vector Database: Building RAG Pipelines with pgvector

-- Run in the Supabase SQL Editor
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id UUID REFERENCES auth.users(id) ON DELETE CASCADE,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);
 
CREATE TABLE document_chunks (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  document_id UUID REFERENCES documents(id) ON DELETE CASCADE,
  user_id UUID REFERENCES auth.users(id) ON DELETE CASCADE,
  chunk_text TEXT NOT NULL,
  embedding VECTOR(1536),  -- text-embedding-3-small dimensions
  chunk_index INTEGER NOT NULL
);
 
-- Index for fast approximate nearest-neighbour search
CREATE INDEX ON document_chunks
  USING ivfflat (embedding vector_cosine_ops)
  WITH (lists = 100);

The IVFFlat index requires at least 50,000 rows to be useful. For smaller datasets, skip the index — pgvector will do exact nearest-neighbour search which is more accurate and fast enough at that scale.

from supabase import create_client
from openai import OpenAI
import os
 
supabase = create_client(os.environ['SUPABASE_URL'], os.environ['SUPABASE_SERVICE_ROLE_KEY'])
openai = OpenAI()
 
def chunk_text(text: str, chunk_size: int = 500) -> list[str]:
    words = text.split()
    return [' '.join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
 
def ingest_document(user_id: str, title: str, content: str):
    # Save document
    doc = supabase.table('documents').insert({
        'user_id': user_id, 'title': title, 'content': content
    }).execute().data[0]
 
    # Chunk and embed
    chunks = chunk_text(content)
    embeddings_res = openai.embeddings.create(
        model='text-embedding-3-small', input=chunks
    )
 
    rows = [{
        'document_id': doc['id'],
        'user_id': user_id,
        'chunk_text': chunk,
        'embedding': emb.embedding,
        'chunk_index': i,
    } for i, (chunk, emb) in enumerate(zip(chunks, embeddings_res.data))]
 
    supabase.table('document_chunks').insert(rows).execute()
    print(f'Ingested {len(rows)} chunks for document {doc["id"]}')

-- Create a SQL function for vector search
CREATE OR REPLACE FUNCTION match_chunks(
  query_embedding VECTOR(1536),
  match_count INT DEFAULT 5,
  filter_user_id UUID DEFAULT NULL
)
RETURNS TABLE (id UUID, chunk_text TEXT, similarity FLOAT)
LANGUAGE SQL STABLE
AS $$
  SELECT
    id,
    chunk_text,
    1 - (embedding <=> query_embedding) AS similarity
  FROM document_chunks
  WHERE (filter_user_id IS NULL OR user_id = filter_user_id)
  ORDER BY embedding <=> query_embedding
  LIMIT match_count;
$$;

def search(user_id: str, query: str, k: int = 5) -> list[dict]:
    # Embed the query
    query_embedding = openai.embeddings.create(
        model='text-embedding-3-small', input=query
    ).data[0].embedding
 
    # Run similarity search
    results = supabase.rpc('match_chunks', {
        'query_embedding': query_embedding,
        'match_count': k,
        'filter_user_id': user_id,
    }).execute()
 
    return results.data

def answer_question(user_id: str, question: str) -> str:
    # Retrieve relevant chunks
    chunks = search(user_id, question, k=5)
    context = '\n\n'.join(c['chunk_text'] for c in chunks)
 
    # Generate answer
    response = openai.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role': 'system', 'content':
             'Answer the question using only the provided context.\n\n'
             f'Context:\n{context}'},
            {'role': 'user', 'content': question},
        ]
    )
    return response.choices[0].message.content