Building a RAG pipeline with LangChain and pgvector

What RAG actually solves

Large language models hallucinate. They also have a knowledge cutoff, know nothing about your private data, and return confident-sounding answers even when they are wrong. Retrieval-augmented generation (RAG) addresses all three by splitting the problem in two: first retrieve relevant documents from your own knowledge base, then pass those documents as context to the LLM so its answer is grounded in verifiable source material.

The payoff is measurable. In internal benchmarks across document Q&A tasks, a well-tuned RAG system typically reduces hallucination rates by 40–60% compared to a prompt-only approach, while dramatically cutting the cost of fine-tuning. You get domain-specific accuracy without retraining a model — and you can update the knowledge base at any time without touching the model at all.

This guide builds a complete pipeline end to end: ingest documents, chunk and embed them, store vectors in PostgreSQL with pgvector, retrieve relevant chunks at query time, generate citations-grounded answers with Claude, evaluate retrieval quality, and handle the production concerns that come up once real users start asking unpredictable questions.

Prerequisites

• Node.js 20+
• PostgreSQL 15+ with the pgvector extension
• An Anthropic API key and an OpenAI API key (for embeddings)
• npm install langchain @langchain/openai @langchain/anthropic pg zod

The pipeline runs entirely server-side. All code in this guide is TypeScript executed with ts-node. There is no React or Next.js layer — add that separately once the pipeline is working correctly.

Step 1 — Set up pgvector

pgvector adds a vector column type and three index strategies (exact flat scan, IVFFlat, and HNSW) to PostgreSQL. Install the extension once per database, then create the documents table:

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE documents (
  id          BIGSERIAL PRIMARY KEY,
  source      TEXT NOT NULL,            -- file path, URL, or document ID
  chunk_index INT  NOT NULL,
  content     TEXT NOT NULL,
  embedding   vector(1536),             -- dimension must match your embedding model
  metadata    JSONB DEFAULT '{}'
);

-- HNSW index for fast approximate nearest-neighbour search
-- m=16: number of bidirectional links per node (higher = better recall, more memory)
-- ef_construction=64: size of the dynamic candidate list during construction
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

-- Optional: partial index to exclude deleted/archived chunks
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64)
  WHERE metadata->>'archived' IS DISTINCT FROM 'true';

HNSW (Hierarchical Navigable Small World) trades a small amount of recall for dramatically faster query times compared to a full table scan. At fewer than a few hundred thousand vectors, the accuracy difference is negligible in practice. IVFFlat is a better choice above ten million vectors where memory becomes a constraint.

The metadata JSONB column is valuable from day one. Store the document title, creation date, author, section path, or any other attribute you might want to filter on later. It costs almost nothing to store and saves a separate join.

Step 2 — Chunking strategy

Chunking is the single decision that most affects RAG quality. Too large and you return noisy, unfocused context that confuses the LLM; too small and you lose the surrounding context that gives a sentence its meaning. The right answer depends entirely on your content type:

LangChain's RecursiveCharacterTextSplitter does semantic-aware chunking by splitting on paragraph breaks first, then sentences, then words, then characters — in that priority order. This keeps ideas together rather than splitting mid-sentence.

import { RecursiveCharacterTextSplitter } from "langchain/text_splitter";

const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 600,       // characters, not tokens (rough rule: 1 token ≈ 4 chars)
  chunkOverlap: 80,
  separators: ["\n\n", "\n", ". ", " ", ""],
});

async function chunkDocument(content: string, source: string, metadata: object = {}) {
  const docs = await splitter.createDocuments(
    [content],
    [{ source, ...metadata }]
  );
  return docs.map((doc, i) => ({
    content: doc.pageContent,
    source,
    chunkIndex: i,
    metadata: doc.metadata,
  }));
}

One optimisation that consistently improves retrieval quality: prepend the document title or section heading to each chunk. The embedding for "Authentication tokens expire after 24 hours" and "tokens expire after 24 hours" are meaningfully different — the first will surface more reliably when a user asks about auth token lifetime. Implement this as a simple prefix:

function enrichChunk(chunk: string, documentTitle: string, sectionHeading?: string): string {
  const parts = [documentTitle];
  if (sectionHeading) parts.push(sectionHeading);
  parts.push(chunk);
  return parts.join("\n\n");
}

Another underrated technique: generate a synthetic question for each chunk and embed the question alongside the chunk content. Users ask questions; your content contains answers. The embedding space between "How do I reset my password?" and the corresponding help article paragraph is smaller when the chunk also contains the question form.

Step 3 — Embed and store

OpenAI's text-embedding-3-small (1536 dimensions) is a strong default for English-language RAG. For multilingual content, use text-embedding-3-large or Cohere's embed-multilingual-v3.0. The dimension must match the vector(N) column definition — mixing them causes a silent type error in pgvector.

import { OpenAIEmbeddings } from "@langchain/openai";
import { Pool } from "pg";

const embeddings = new OpenAIEmbeddings({
  model: "text-embedding-3-small",
  batchSize: 512,  // embed up to 512 chunks per API call to reduce round-trips
});

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

interface ChunkInput {
  content: string;
  source: string;
  chunkIndex: number;
  metadata?: object;
}

async function ingestChunks(chunks: ChunkInput[]) {
  const texts = chunks.map((c) => c.content);
  const vectors = await embeddings.embedDocuments(texts);

  const client = await pool.connect();
  try {
    await client.query("BEGIN");

    // Delete existing chunks for these sources to support re-ingestion
    const sources = [...new Set(chunks.map((c) => c.source))];
    await client.query(
      `DELETE FROM documents WHERE source = ANY($1)`,
      [sources]
    );

    for (let i = 0; i < chunks.length; i++) {
      await client.query(
        `INSERT INTO documents (source, chunk_index, content, embedding, metadata)
         VALUES ($1, $2, $3, $4::vector, $5)`,
        [
          chunks[i].source,
          chunks[i].chunkIndex,
          chunks[i].content,
          JSON.stringify(vectors[i]),
          JSON.stringify(chunks[i].metadata ?? {}),
        ]
      );
    }

    await client.query("COMMIT");
    console.log(`Ingested ${chunks.length} chunks from ${sources.join(", ")}`);
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

Always batch inserts inside a single transaction. Committing one row at a time for 10 000 chunks is roughly 100× slower due to fsync overhead on every commit. The DELETE before insert makes re-ingestion idempotent — run it any time source documents change without accumulating stale chunks.

For very large corpora (hundreds of thousands of documents), consider disabling the HNSW index during bulk ingest and rebuilding it afterward:

-- Before bulk ingest
DROP INDEX IF EXISTS documents_embedding_idx;

-- After bulk ingest
CREATE INDEX documents_embedding_idx ON documents
  USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

Building the index once over the full dataset is significantly faster than maintaining it incrementally during insert.

Step 4 — Retrieve relevant chunks

At query time, embed the user's question and find the closest vectors using cosine similarity. pgvector's <=> operator computes cosine distance (1 − similarity), so ordering ascending gives you the closest vectors first:

interface RetrievedChunk {
  id: number;
  source: string;
  content: string;
  similarity: number;
  metadata: Record<string, unknown>;
}

async function retrieveChunks(
  query: string,
  options: { topK?: number; minSimilarity?: number; sourceFilter?: string[] } = {}
): Promise<RetrievedChunk[]> {
  const { topK = 5, minSimilarity = 0.75, sourceFilter } = options;

  const [queryVector] = await embeddings.embedDocuments([query]);

  let sql = `
    SELECT
      id,
      source,
      content,
      metadata,
      1 - (embedding <=> $1::vector) AS similarity
    FROM documents
    WHERE 1 - (embedding <=> $1::vector) > $2
  `;
  const params: unknown[] = [JSON.stringify(queryVector), minSimilarity];

  if (sourceFilter && sourceFilter.length > 0) {
    sql += ` AND source = ANY($3)`;
    params.push(sourceFilter);
  }

  sql += ` ORDER BY embedding <=> $1::vector LIMIT ${topK}`;

  const result = await pool.query<RetrievedChunk>(sql, params);
  return result.rows;
}

The sourceFilter parameter is important in multi-tenant systems where different users should only query their own documents. Pass it explicitly rather than relying on pgvector returning zero results — treat data isolation as a hard constraint enforced in the query, not a soft one enforced by similarity score.

Tuning minSimilarity: Start at 0.75. If answers include clearly irrelevant context, raise it toward 0.80. If the system says "I don't know" too often on queries you know are covered by your documents, lower it toward 0.70. Log every retrieval with its similarity score for the first two weeks in production and plot a histogram — you will quickly see whether your distribution clusters near 0.85 (great) or scatters between 0.60 and 0.95 (chunking or embedding problem).

Step 5 — Generate grounded answers with Claude

Pass the retrieved chunks as numbered context in the system prompt. Explicitly instruct the model to cite source numbers and to say so when it cannot find an answer — Claude follows this instruction reliably when the wording is direct:

import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();

interface AnswerResult {
  answer: string;
  sources: string[];
  retrievedCount: number;
}

async function answerQuestion(query: string): Promise<AnswerResult> {
  const chunks = await retrieveChunks(query, { topK: 6, minSimilarity: 0.72 });

  if (chunks.length === 0) {
    return {
      answer: "I could not find relevant information in the knowledge base to answer that question. Try rephrasing, or ask your administrator to ingest additional documents.",
      sources: [],
      retrievedCount: 0,
    };
  }

  const context = chunks
    .map((c, i) => `[${i + 1}] Source: ${c.source}\n${c.content}`)
    .join("\n\n---\n\n");

  const message = await anthropic.messages.create({
    model: "claude-sonnet-4-6",
    max_tokens: 1500,
    system: `You are a precise technical assistant. Answer questions using only the provided context.
- Always cite sources using bracket notation [1], [2], etc.
- If a claim cannot be supported by the provided context, say so explicitly.
- Do not draw on general knowledge outside the context — users rely on accuracy over completeness.
- If the context contains contradictory information, note the contradiction.`,
    messages: [
      {
        role: "user",
        content: `Context:\n\n${context}\n\nQuestion: ${query}`,
      },
    ],
  });

  const answerText =
    message.content[0].type === "text" ? message.content[0].text : "";
  const uniqueSources = [...new Set(chunks.map((c) => c.source))];

  return {
    answer: answerText,
    sources: uniqueSources,
    retrievedCount: chunks.length,
  };
}

The instruction "do not draw on general knowledge outside the context" is the key guardrail. Without it, Claude will helpfully fill gaps from its training data when no relevant chunk covers the question — which is exactly the hallucination behaviour RAG is supposed to eliminate.

Step 6 — Hybrid search: combining BM25 and vector similarity

Vector search excels at semantic matching but struggles with exact keywords, product names, model numbers, and any term that appears rarely in the training corpus for the embedding model. "Claude 3.5 Sonnet" and "Anthropic's latest reasoning model" may embed very close together — but if a user searches for "claude-sonnet-4-6", the keyword match matters as much as the semantic one.

Hybrid search combines BM25 (keyword) and vector (semantic) scores and merges the two ranked lists with Reciprocal Rank Fusion (RRF):

-- BM25 ranking via PostgreSQL full-text search
SELECT
  id,
  ts_rank_cd(to_tsvector('english', content), query) AS bm25_rank
FROM documents, plainto_tsquery('english', $1) query
WHERE to_tsvector('english', content) @@ query
ORDER BY bm25_rank DESC
LIMIT 20;

function reciprocalRankFusion(
  lists: Array<Array<{ id: number }>>,
  k = 60
): number[] {
  const scores = new Map<number, number>();

  for (const list of lists) {
    list.forEach(({ id }, rank) => {
      scores.set(id, (scores.get(id) ?? 0) + 1 / (k + rank + 1));
    });
  }

  return [...scores.entries()]
    .sort((a, b) => b[1] - a[1])
    .map(([id]) => id);
}

async function hybridRetrieve(query: string, topK = 5): Promise<RetrievedChunk[]> {
  // Run vector and BM25 searches in parallel
  const [vectorResults, bm25Results] = await Promise.all([
    retrieveChunks(query, { topK: 20, minSimilarity: 0.60 }),
    pool
      .query<{ id: number; bm25_rank: number }>(
        `SELECT id, ts_rank_cd(to_tsvector('english', content), q) AS bm25_rank
         FROM documents, plainto_tsquery('english', $1) q
         WHERE to_tsvector('english', content) @@ q
         ORDER BY bm25_rank DESC LIMIT 20`,
        [query]
      )
      .then((r) => r.rows),
  ]);

  const fusedIds = reciprocalRankFusion([
    vectorResults.map((r) => ({ id: r.id })),
    bm25Results.map((r) => ({ id: r.id })),
  ]).slice(0, topK);

  // Fetch the full rows for the fused top-K IDs
  const idList = fusedIds.join(",");
  const result = await pool.query<RetrievedChunk>(
    `SELECT id, source, content, metadata,
            1 - (embedding <=> (SELECT embedding FROM documents WHERE id = $1)::vector) AS similarity
     FROM documents
     WHERE id = ANY(ARRAY[${idList}])`,
    [fusedIds[0]]
  );

  // Re-sort by the fused order
  const byId = new Map(result.rows.map((r) => [r.id, r]));
  return fusedIds.map((id) => byId.get(id)!).filter(Boolean);
}

Hybrid search consistently improves precision by 5–15% over pure vector search at minimal extra latency, since both queries run in parallel. It is particularly impactful for technical documentation where users search for exact function names, error codes, and configuration keys.

Step 7 — Evaluate retrieval quality

A RAG system that is not evaluated will silently degrade. Content changes, new question patterns emerge, and embedding models get updated. Build an evaluation suite from day one.

The two primary metrics are:

Recall@K — of the queries in your test set where a relevant document exists, what fraction does your retrieval find within the top K results? Target ≥ 90% Recall@5 before going to production.

Answer faithfulness — does every factual claim in the generated answer trace to a cited chunk? This is harder to automate but can be approximated by checking that every sentence in the answer contains at least one citation number.

interface EvalCase {
  query: string;
  expectedKeyword: string;   // a word or phrase that must appear in a retrieved chunk
  expectedSource?: string;   // the source document that must be in the results
}

async function evaluateRetrieval(cases: EvalCase[], topK = 5) {
  let keywordHits = 0;
  let sourceHits = 0;
  const misses: string[] = [];

  for (const tc of cases) {
    const chunks = await retrieveChunks(tc.query, { topK });

    const keywordFound = chunks.some((c) =>
      c.content.toLowerCase().includes(tc.expectedKeyword.toLowerCase())
    );
    if (keywordFound) {
      keywordHits++;
    } else {
      misses.push(`MISS keyword: "${tc.query}" → expected "${tc.expectedKeyword}"`);
    }

    if (tc.expectedSource) {
      const sourceFound = chunks.some((c) => c.source === tc.expectedSource);
      if (sourceFound) sourceHits++;
    }
  }

  const total = cases.length;
  console.log(`Recall@${topK} (keyword): ${((keywordHits / total) * 100).toFixed(1)}%`);
  console.log(`Recall@${topK} (source):  ${((sourceHits / cases.filter(c => c.expectedSource).length) * 100).toFixed(1)}%`);
  misses.forEach((m) => console.warn(m));
}

Build your test suite from real questions your users have asked or are likely to ask. Synthetic test cases generated by the LLM itself ("what questions would this document answer?") are a quick way to bootstrap the suite, but supplement them with real-world queries as soon as you have usage data.

Step 8 — Re-ranking retrieved chunks

Vector retrieval gives you a first-pass set of candidates, but the ordering within that set is not always optimal for the LLM. A cross-encoder re-ranker reads each query-chunk pair together and scores their relevance more precisely than the embedding cosine similarity can — at the cost of running N inference calls instead of one.

For most applications, a lightweight re-ranker like Cohere's Rerank API is a pragmatic choice:

import { CohereClient } from "cohere-ai";

const cohere = new CohereClient({ token: process.env.COHERE_API_KEY });

async function rerankChunks(query: string, chunks: RetrievedChunk[], topN = 5) {
  if (chunks.length === 0) return chunks;

  const response = await cohere.rerank({
    model: "rerank-english-v3.0",
    query,
    documents: chunks.map((c) => c.content),
    topN,
  });

  return response.results.map((r) => chunks[r.index]);
}

Plug this between retrieval and generation:

const rawChunks = await hybridRetrieve(query, 15);   // retrieve wider set
const reranked  = await rerankChunks(query, rawChunks, 5);  // rerank to top 5
const answer    = await generateAnswer(query, reranked);

The wider retrieval + reranking pattern reliably outperforms narrow retrieval + no reranking. Fetch 10–20 candidates, rerank to 5. The total latency cost of the reranker is typically 100–200ms, which is acceptable for interactive use cases.

Step 9 — Contextual compression

Even with good chunking, retrieved chunks often contain more text than the LLM needs. Contextual compression extracts only the sentences within a chunk that are actually relevant to the query — reducing token usage and focusing the LLM's attention:

async function compressChunk(query: string, chunkContent: string): Promise<string> {
  const message = await anthropic.messages.create({
    model: "claude-haiku-4-5-20251001",   // use a fast, cheap model for compression
    max_tokens: 300,
    system: "Extract only the sentences from the provided text that are directly relevant to the question. Output only the extracted sentences, verbatim. If no sentences are relevant, output 'NOT_RELEVANT'.",
    messages: [
      {
        role: "user",
        content: `Question: ${query}\n\nText:\n${chunkContent}`,
      },
    ],
  });

  const compressed =
    message.content[0].type === "text" ? message.content[0].text : chunkContent;

  return compressed === "NOT_RELEVANT" ? "" : compressed;
}

Compression is most valuable when chunks are long (800+ words) or when your content has a lot of boilerplate surrounding the key facts. For short, focused chunks it adds latency without much benefit.

Production considerations

Re-embedding on model upgrade. Embeddings from different models — or even different versions of the same model — are not compatible. When you upgrade from text-embedding-3-small to text-embedding-3-large, every row in the documents table must be re-embedded and the vector column redefined. Plan for a background re-indexing job that can run over several hours without blocking reads.

Incremental ingestion. Track a last_ingested_at timestamp per source document (store it in the metadata column or a separate table). On each ingestion run, only process documents modified since that timestamp to avoid re-embedding the entire corpus unnecessarily.

Connection pooling. RAG workloads generate short, bursty queries that exhaust PostgreSQL connection limits quickly. Use PgBouncer in transaction mode in front of your PostgreSQL instance. Set pool_size conservatively — more connections means more memory consumed per connection on the server side.

HNSW index maintenance. HNSW indexes degrade slightly as rows are deleted and reinserted, because deleted nodes leave gaps in the graph. Run REINDEX INDEX CONCURRENTLY on the embedding index weekly during a low-traffic window to rebuild a clean graph.

Monitoring retrieval quality over time. Log every query, every retrieved chunk (with source and similarity score), and every generated answer to a separate analytics table. Set up a weekly job that samples 50 query-answer pairs and grades them for faithfulness. Build a dashboard that tracks Recall@5 and average similarity score over time. Retrieval quality degrades silently — without monitoring you will not know until users complain.

Tenant isolation. If your RAG system is multi-tenant, enforce source-level filtering at the SQL layer on every retrieval query, not just at the application layer. A misconfigured application-layer filter is a support ticket; a SQL-layer filter is a guarantee.

Common failure patterns and fixes

"I don't know" on questions that should be answered. First check: is the relevant content actually ingested? Query directly for the expected keyword with SELECT * FROM documents WHERE content ILIKE '%keyword%'. If it is there, the similarity threshold is too high — lower minSimilarity by 0.05. If it is not there, re-run ingestion.

Answers that mix up information from different documents. Your chunks are too large and contain multiple distinct topics. Reduce chunk size and increase overlap so each chunk covers a single idea.

Answers that are technically correct but miss the point. The query embedding and the answer embedding are similar, but the question is ambiguous. Add a query rewriting step that expands the question into two or three paraphrases and retrieves chunks for all of them.

Slow query times under load. Ensure the HNSW index is being used — run EXPLAIN SELECT ... and look for Index Scan using documents_embedding_idx. If you see a sequential scan, the query planner has decided the index is not worth using, usually because topK is very large or the minSimilarity filter is very loose. Adjust accordingly.

Wrapping up

The core RAG pipeline is small: chunk → embed → store → retrieve → generate. The quality lives in the details: chunking strategy, similarity threshold tuning, hybrid search, contextual compression, re-ranking, and systematic evaluation. Build the evaluation suite before you build the user interface — it is the only reliable way to confirm that a pipeline change actually improved the system rather than just shifted its failure modes.

The pattern scales further than most teams expect from a single PostgreSQL instance. With HNSW indexing and connection pooling, a well-tuned pgvector deployment handles millions of vectors and hundreds of concurrent queries per second — without the operational overhead of a dedicated vector database.