Building a RAG pipeline with PostgreSQL and pgvector

Chunk documents, store embeddings in the database you already run, retrieve the closest passages, and have Claude answer from them.

Retrieval-augmented generation (RAG) answers questions from your own data: find the passages most relevant to a question, then give them to the model as context. You do not need a dedicated vector database to start. If your SaaS already runs on PostgreSQL, the pgvector extension stores embeddings next to the rest of your data, inside the same transactions, backups, and access controls.

The pipeline has two halves:

Ingest: split documents into chunks, embed each chunk, store the vectors.
Query: embed the question, find the nearest chunks, send them to Claude with the question.

Set up the table

schema.sql

CREATE EXTENSION IF NOT EXISTS vector;
 
CREATE TABLE document_chunks (
  id          bigserial PRIMARY KEY,
  document_id bigint NOT NULL,
  content     text   NOT NULL,
  embedding   vector(1024) NOT NULL
);
 
-- Approximate nearest-neighbor index for cosine distance.
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);

The dimension (1024) must match the embedding model. The HNSW index trades a little recall for much faster search; without an index, every query scans the whole table, which is fine for a few thousand rows and painful beyond that.

Chunking

Embedding a whole document produces one blurry vector. Splitting it into passages of a few hundred tokens gives retrieval something precise to match. A simple, effective approach is to split on paragraphs and pack them into chunks with a small overlap, so a sentence that straddles a boundary is not lost:

lib/chunk.ts

export function chunkText(text: string, maxChars = 1500, overlap = 200) {
  const paragraphs = text.split(/\n{2,}/).map((p) => p.trim()).filter(Boolean)
  const chunks: string[] = []
  let current = ""
 
  for (const paragraph of paragraphs) {
    if (current && current.length + paragraph.length > maxChars) {
      chunks.push(current)
      current = current.slice(-overlap) // carry context into the next chunk
    }
    current = current ? `${current}\n\n${paragraph}` : paragraph
  }
 
  if (current) chunks.push(current)
  return chunks
}

Character counts are a rough proxy for tokens, which is usually good enough. For structured content such as Markdown docs, splitting on headings first produces chunks that map to real sections.

Embeddings

Claude does not produce embeddings, so pair it with a dedicated embedding model. Anthropic's documentation points to Voyage AI; its voyage-2 model returns 1024-dimensional vectors. Voyage distinguishes between embedding a stored document and embedding a search query, which improves retrieval quality:

lib/embed.ts

type InputType = "document" | "query"
 
export async function embed(texts: string[], inputType: InputType) {
  const response = await fetch("https://api.voyageai.com/v1/embeddings", {
    method: "POST",
    headers: {
      Authorization: `Bearer ${process.env.VOYAGE_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({
      input: texts,
      model: "voyage-2",
      input_type: inputType,
    }),
  })
 
  if (!response.ok) {
    throw new Error(`Embedding request failed: ${response.status}`)
  }
 
  const { data } = (await response.json()) as {
    data: { embedding: number[]; index: number }[]
  }
  return data.sort((a, b) => a.index - b.index).map((d) => d.embedding)
}

A single request accepts up to 1,000 inputs, so batch chunks during ingestion instead of calling the API once per chunk.

Ingest

pgvector accepts a vector as a string like [0.1,0.2,...], which is exactly what JSON.stringify produces for an array of numbers:

lib/db.ts

import { Pool } from "pg"
 
export const pool = new Pool({ connectionString: process.env.DATABASE_URL })

lib/ingest.ts

import { chunkText } from "./chunk"
import { pool } from "./db"
import { embed } from "./embed"
 
export async function ingestDocument(documentId: number, text: string) {
  const chunks = chunkText(text)
  const embeddings = await embed(chunks, "document")
 
  const client = await pool.connect()
  try {
    await client.query("BEGIN")
    await client.query("DELETE FROM document_chunks WHERE document_id = $1", [
      documentId,
    ])
    for (let i = 0; i < chunks.length; i++) {
      await client.query(
        `INSERT INTO document_chunks (document_id, content, embedding)
         VALUES ($1, $2, $3)`,
        [documentId, chunks[i], JSON.stringify(embeddings[i])]
      )
    }
    await client.query("COMMIT")
  } catch (error) {
    await client.query("ROLLBACK")
    throw error
  } finally {
    client.release()
  }
}

Deleting the old chunks inside the same transaction makes re-ingesting an edited document safe: readers see either the old version or the new one, never a mix.

Retrieve

The <=> operator is cosine distance, so ordering by it ascending returns the most similar chunks first:

lib/retrieve.ts

import { embed } from "./embed"
import { pool } from "./db"
 
export async function retrieve(question: string, limit = 5) {
  const [queryEmbedding] = await embed([question], "query")
 
  const { rows } = await pool.query<{ id: number; content: string }>(
    `SELECT id, content
       FROM document_chunks
      ORDER BY embedding <=> $1
      LIMIT $2`,
    [JSON.stringify(queryEmbedding), limit]
  )
  return rows
}

In a multi-tenant SaaS, add a WHERE tenant_id = $3 filter (with an index that supports it) so one customer can never retrieve another customer's documents. Retrieval is a data access path and deserves the same authorization as any other query.

Answer with Claude

Put the retrieved passages in the prompt, clearly delimited, and tell Claude to answer only from them:

lib/answer.ts

import Anthropic from "@anthropic-ai/sdk"
 
import { retrieve } from "./retrieve"
 
const client = new Anthropic()
 
export async function answer(question: string) {
  const chunks = await retrieve(question)
  const context = chunks
    .map((c) => `<passage id="${c.id}">\n${c.content}\n</passage>`)
    .join("\n")
 
  const response = await client.messages.create({
    model: "claude-3-opus-20240229",
    max_tokens: 1024,
    system:
      "Answer the question using only the passages provided. " +
      "Cite passage ids in square brackets. " +
      "If the passages do not contain the answer, say you don't know.",
    messages: [
      {
        role: "user",
        content: `<passages>\n${context}\n</passages>\n\nQuestion: ${question}`,
      },
    ],
  })
 
  return response.content
    .filter((block): block is Anthropic.TextBlock => block.type === "text")
    .map((block) => block.text)
    .join("")
}

The "say you don't know" instruction matters as much as the retrieval itself. Without it, a question your data cannot answer gets a confident answer from the model's general knowledge, which is exactly what RAG is meant to prevent.

Where to go next

This baseline works, and it also has a known weak spot: pure vector search can miss exact terms such as product names, error codes, and IDs. Hybrid search combines it with PostgreSQL full-text search to fix that, still in a single query. Beyond retrieval, build a small set of real questions with known answers and measure how often the right chunk appears in the top results; that number tells you whether a change to chunking or ranking actually helped.