Supabase pgvector plus LangChain for RAG

If you’re building a retrieval-augmented generation system and want PostgreSQL as your vector store, supabase pgvector langchain rag gives you a managed Postgres instance with the pgvector extension pre-installed, plus a Python ecosystem that integrates cleanly with LangChain’s document loaders, text splitters, and retriever abstractions. This walkthrough takes you from a fresh Supabase project to a queryable RAG pipeline you can ship, including the schema decisions, indexing strategy, and retrieval tuning that separate a demo from something that holds up under load.

Step 1: Provision Supabase and enable pgvector

Create a new project at supabase.com. Wait for the database to spin up — usually under two minutes. In the SQL editor, run:

create extension if not exists vector;

Verify it works:

select extname, extversion from pg_extension where extname = 'vector';

You should see vector with a version like 0.5.1 or newer. If the extension isn’t available, your Supabase region may be on an older Postgres image — open a support ticket or choose a different region.

Grab your connection string from Settings → Database → Connection string → Transaction pooler (port 6543). It looks like:

postgresql://postgres.<project-ref>:<password>@aws-0-<region>.pooler.supabase.com:6543/postgres

Store this in your environment as SUPABASE_DB_URL. Do not use the session pooler (port 5432) for application traffic — it’s not designed for sustained connection counts.

Step 2: Design the documents table with hybrid search in mind

A common mistake is storing only the embedding and a text column. Production RAG needs metadata filtering, source attribution, and the ability to re-rank. Create this schema:

create table documents (
  id bigserial primary key,
  content text not null,
  embedding vector(1536),  -- matches text-embedding-3-small
  metadata jsonb not null default '{}',
  created_at timestamptz not null default now()
);

create index on documents using hnsw (embedding vector_cosine_ops)
  with (m = 16, ef_construction = 64);

create index on documents using gin (metadata jsonb_path_ops);

Notes on the indexes:

• HNSW with m=16, ef_construction=64 is a solid default for 1536-dim vectors. Increase ef_construction to 128 if you have >1M rows and can afford the build time.
• The GIN index on metadata with jsonb_path_ops supports @> containment queries (e.g., metadata @> '{"source": "api-docs"}'). It’s smaller than the default GIN opclass and faster for exact key/value lookups.
• We use vector_cosine_ops because OpenAI embeddings are normalized; cosine similarity equals dot product but the operator class makes intent explicit.

Step 3: Install the Python dependencies

pip install langchain langchain-openai langchain-community supabase psycopg2-binary python-dotenv

langchain-community contains the SupabaseVectorStore integration. psycopg2-binary is the driver; if you prefer asyncpg, swap it in and adjust the connection string accordingly.

Step 4: Configure the LangChain vector store wrapper

Create vectorstore.py:

import os
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import SupabaseVectorStore
from supabase import create_client

SUPABASE_URL = os.environ["SUPABASE_URL"]          # https://<project-ref>.supabase.co
SUPABASE_SERVICE_KEY = os.environ["SUPABASE_SERVICE_KEY"]  # service_role key
DB_URL = os.environ["SUPABASE_DB_URL"]             # pooler connection string

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

supabase_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY)

vector_store = SupabaseVectorStore(
    client=supabase_client,
    embedding=embeddings,
    table_name="documents",
    query_name="match_documents",  # we'll create this RPC in Step 5
)

The query_name parameter tells LangChain to call a Postgres function instead of generating raw SQL. This lets us control the search logic — critical for hybrid search and metadata filtering.

Step 5: Create the matching RPC with metadata filters

In the Supabase SQL editor, run:

create or replace function match_documents(
  query_embedding vector(1536),
  match_count int default 10,
  filter jsonb default '{}'
)
returns table (
  id bigint,
  content text,
  metadata jsonb,
  similarity float
)
language sql stable
as $$
  select
    d.id,
    d.content,
    d.metadata,
    1 - (d.embedding <=> query_embedding) as similarity
  from documents d
  where d.metadata @> filter
  order by d.embedding <=> query_embedding
  limit match_count;
$$;

This function:

• Accepts a filter JSONB parameter for metadata containment (@>)
• Returns cosine similarity as 1 - (embedding <=> query) — the <=> operator is cosine distance
• Uses the HNSW index automatically because of the ORDER BY ... <=> pattern

Verify it works:

select * from match_documents(
  query_embedding := (select embedding from documents limit 1),
  match_count := 5,
  filter := '{"source": "api-docs"}'
);

You should get rows back with similarity scores between 0 and 1.

Step 6: Ingest documents with metadata

Create ingest.py:

import os
from pathlib import Path
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    TextLoader,
    PyMuPDFLoader,
    UnstructuredMarkdownLoader,
)
from vectorstore import vector_store

RAW_DIR = Path("data/raw")
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 150

def load_documents():
    docs = []
    for path in RAW_DIR.rglob("*"):
        if path.suffix == ".txt":
            loader = TextLoader(str(path))
        elif path.suffix == ".pdf":
            loader = PyMuPDFLoader(str(path))
        elif path.suffix in (".md", ".markdown"):
            loader = UnstructuredMarkdownLoader(str(path))
        else:
            continue
        loaded = loader.load()
        for doc in loaded:
            doc.metadata["source"] = path.name
            doc.metadata["path"] = str(path.relative_to(RAW_DIR))
        docs.extend(loaded)
    return docs

def main():
    raw_docs = load_documents()
    print(f"Loaded {len(raw_docs)} raw documents")

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", " ", ""],
    )
    chunks = splitter.split_documents(raw_docs)
    print(f"Split into {len(chunks)} chunks")

    # Add a content hash for idempotent re-ingestion
    for chunk in chunks:
        chunk.metadata["content_hash"] = hash(chunk.page_content)

    vector_store.add_documents(chunks)
    print("Ingestion complete")

if __name__ == "__main__":
    main()

Run it:

python ingest.py

Verify in Supabase:

select count(*), count(distinct metadata->>'source') as sources
from documents;

You should see row counts matching your chunk total and at least one source.

Step 7: Build the retrieval chain with hybrid scoring

Pure vector search misses exact matches (error codes, function names, version strings). Add a keyword component using Postgres tsvector. First, add a search column and index:

alter table documents add column if not exists search_vector tsvector
  generated always as (to_tsvector('english', content)) stored;

create index on documents using gin (search_vector);

Update the RPC to blend vector and keyword scores:

create or replace function match_documents(
  query_embedding vector(1536),
  query_text text default '',
  match_count int default 10,
  filter jsonb default '{}',
  vector_weight float default 0.7,
  keyword_weight float default 0.3
)
returns table (
  id bigint,
  content text,
  metadata jsonb,
  similarity float
)
language sql stable
as $$
  with vector_scores as (
    select
      id,
      1 - (embedding <=> query_embedding) as vec_score
    from documents
    where metadata @> filter
  ),
  keyword_scores as (
    select
      id,
      ts_rank_cd(search_vector, plainto_tsquery('english', query_text)) as kw_score
    from documents
    where metadata @> filter
      and query_text <> ''
      and search_vector @@ plainto_tsquery('english', query_text)
  ),
  combined as (
    select
      d.id,
      d.content,
      d.metadata,
      coalesce(vs.vec_score, 0) * vector_weight +
      coalesce(ks.kw_score, 0) * keyword_weight as similarity
    from documents d
    left join vector_scores vs on d.id = vs.id
    left join keyword_scores ks on d.id = ks.id
    where d.metadata @> filter
  )
  select id, content, metadata, similarity
  from combined
  order by similarity desc
  limit match_count;
$$;

The weights (0.7 / 0.3) are a starting point. Tune them per domain — code-heavy corpora often benefit from higher keyword weight.

Update vectorstore.py to pass the query text:

# In vectorstore.py, modify the vector_store initialization:
vector_store = SupabaseVectorStore(
    client=supabase_client,
    embedding=embeddings,
    table_name="documents",
    query_name="match_documents",
)

# Add a helper for hybrid search:
def hybrid_search(query: str, k: int = 8, filter: dict | None = None):
    return vector_store.similarity_search_with_score(
        query=query,
        k=k,
        filter=filter or {},
        # The wrapper passes extra kwargs to the RPC
        query_text=query,
    )

Step 8: Wire up the RAG chain

Create rag_chain.py:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_openai import ChatOpenAI
from vectorstore import vector_store, hybrid_search

SYSTEM_PROMPT = """You are a technical assistant. Answer the user's question using only the provided context.
If the context doesn't contain the answer, say you don't know. Cite sources using [source] notation."""

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def format_docs(docs_with_scores):
    lines = []
    for doc, score in docs_with_scores:
        src = doc.metadata.get("source", "unknown")
        lines.append(f"[{src}] (score: {score:.3f}) {doc.page_content}")
    return "\n\n".join(lines)

retriever = RunnableParallel(
    context=lambda x: format_docs(hybrid_search(x["question"], k=8, filter=x.get("filter"))),
    question=lambda x: x["question"],
)

rag_chain = retriever | prompt | llm

# Example usage:
if __name__ == "__main__":
    result = rag_chain.invoke({
        "question": "How do I configure connection pooling in Supabase?",
        "filter": {"source": "supabase-docs.pdf"}
    })
    print(result.content)

Run it:

python rag_chain.py

You should see a cited answer drawing only from the filtered source.

Step 9: Add observability and guardrails

Production RAG needs three things: latency tracking, retrieval quality signals, and fallback behavior.

Latency tracking

Wrap the retriever:

import time
from langchain_core.runnables import RunnableLambda

def timed_retriever(input_dict):
    start = time.perf_counter()
    result = hybrid_search(input_dict["question"], k=8, filter=input_dict.get("filter"))
    elapsed = time.perf_counter() - start
    print(f"Retrieval took {elapsed*1000:.1f}ms, got {len(result)} docs")
    return {"context": format_docs(result), "question": input_dict["question"]}

retriever = RunnableLambda(timed_retriever)

Retrieval quality: log scores and sources

Modify format_docs to return structured data:

def format_docs(docs_with_scores):
    return [
        {
            "content": doc.page_content,
            "source": doc.metadata.get("source"),
            "score": float(score),
            "metadata": doc.metadata,
        }
        for doc, score in docs_with_scores
    ]

Then log the top-3 scores per query. If the top score is consistently below 0.3, your embeddings or chunking strategy needs work.

Fallback: expand filter or rewrite query

def retrieve_with_fallback(input_dict):
    filter_ = input_dict.get("filter", {})
    results = hybrid_search(input_dict["question"], k=8, filter=filter_)
    
    if not results or results[0][1] < 0.25:
        # Relax filter: drop the most specific key
        relaxed = {k: v for k, v in filter_.items() if k != "section"}
        print(f"Falling back to relaxed filter: {relaxed}")
        results = hybrid_search(input_dict["question"], k=8, filter=relaxed)
    
    return {"context": format_docs(results), "question": input_dict["question"]}

Step 10: Evaluate with a golden set

Create eval_set.jsonl:

{"question": "What is the max request size for Supabase storage?", "answer": "5GB for resumable uploads", "source": "supabase-docs.pdf"}
{"question": "How do I enable RLS on a table?", "answer": "ALTER TABLE ... ENABLE ROW LEVEL SECURITY", "source": "supabase-docs.pdf"}

Evaluation script:

import json
from rag_chain import rag_chain

def evaluate():
    with open("eval_set.jsonl") as f:
        cases = [json.loads(line) for line in f]
    
    for case in cases:
        result = rag_chain.invoke({
            "question": case["question"],
            "filter": {"source": case["source"]}
        })
        print(f"Q: {case['question']}")
        print(f"Expected: {case['answer']}")
        print(f"Got: {result.content[:200]}...")
        print("---")

if __name__ == "__main__":
    evaluate()

Run it weekly. Track:

• Answer correctness (manual or LLM-as-judge)
• Citation accuracy — does the answer reference the right source?
• Retrieval recall — does the golden chunk appear in top-k?

Step 11: Deploy considerations

Connection pooling

Supabase’s transaction pooler (port 6543) handles ~100 concurrent connections per project. For higher throughput, add PgBouncer in front of your application or use the Supabase read replicas (available on Pro plan).

Index maintenance

HNSW indexes degrade with heavy updates. Schedule a weekly REINDEX CONCURRENTLY:

reindex concurrently index documents_embedding_idx;

Run during low-traffic windows. Monitor pg_stat_user_indexes for idx_scan drops.

Cost control

text-embedding-3-small costs ~$0.02/1M tokens. A 100k chunk corpus costs ~$2 to embed once. The ongoing cost is query embeddings — budget ~$0.50/10k queries.

If you’re routing traffic through an inference gateway like n4n.ai, you can swap embedding models without code changes by updating the model name in OpenAIEmbeddings — the gateway handles provider fallback and usage metering automatically.

Verification checklist

Before calling it done, confirm:

1. Ingestion idempotency — run ingest.py twice; row count should not double (the content_hash metadata enables deduplication logic you’ll add).
2. Filter pushdown — EXPLAIN ANALYZE select * from match_documents(...) shows Index Scan using documents_metadata_idx before the vector scan.
3. Hybrid scoring — a query for an exact error code (“ERROR 42P01”) returns that chunk in top-3 even if vector similarity is low.
4. Latency — p95 retrieval + generation under 2s for k=8 on a 50k document corpus.
5. Fallback works — a query with an overly restrictive filter gracefully relaxes and returns results.

What to tune next

• Chunk size/overlap: 1000/150 works for prose. For code, try 500/50 with a code-aware splitter.
• Embedding model: text-embedding-3-large (3072 dim) improves recall ~5-10% at 2x cost and latency. Update the vector(3072) column and re-embed.
• Re-ranking: Add a cross-encoder (e.g., cross-encoder/ms-marco-MiniLM-L-6-v2) over the top-20 results. LangChain’s ContextualCompressionRetriever wraps this cleanly.
• Multi-vector: Store summary embeddings alongside chunk embeddings for multi-hop queries.

The stack — Supabase for storage and metadata filtering, pgvector for ANN, LangChain for orchestration — is boring in the best way. Each component does one thing well, and the integration points are standard SQL and Python. Ship it, measure, iterate.