rag-core-engine

Framework-agnostic RAG (Retrieval-Augmented Generation) engine: crawl, chunk, embed, store, retrieve, answer — plus general-purpose LLM tasks (summarize, extract structured data) over the same documents, for when you want something other than chat. It's plain TypeScript with no dependency on any web framework, HTTP server, or CMS — everything here works the same whether you call it from a script, an Express route, a NestJS service, or the payload-plugin-rag-chatbot adapter built on top of it.

This doc is about using rag-core-engine on its own, outside of Payload.

Install

Install it like any npm package:

npm install rag-core-engine pg

pg is a peer-ish runtime dependency you provide yourself — core never creates its own Pool, you pass one in (see Vector store).

Requirements:

• Node.js LTS, ESM ("type": "module" in your package.json, or .mjs/dynamic import())
• A Postgres database with the pgvector extension installable (CREATE EXTENSION vector — core runs this for you on first use)
• API keys for whichever embedding/LLM providers you use (core never reads process.env itself — you always pass keys in explicitly)

Design boundary

core intentionally has no knowledge of HTTP, routing, or any specific backend framework. It exposes plain async functions and small interfaces (EmbeddingProvider, LLMProvider, VectorStore). Every provider is injected by the caller — nothing is hardcoded to Anthropic/Voyage/Postgres except the reference provider implementations core ships, which you're free to swap out. This is what makes it safe to drop into Express, NestJS, a CLI script, a queue worker, or anywhere else Node.js runs.

Core concepts

| Type | What it is | | --- | --- | | Source | A normalized ingestion origin — a crawled URL, uploaded file, or raw text, reduced to plain text | | Chunk | A bounded slice of a Source, ready to embed | | EmbeddedChunk | A Chunk plus its embedding vector | | EmbeddingProvider | { dimensions, embed(texts: string[]): Promise<number[][]> } | | LLMProvider | { generate(question, context, history?), complete(prompt, options?), extract<T>(prompt, schema, options?) } — generate is RAG chat; complete/extract are general-purpose (see Beyond chat) | | VectorStore | { init, upsert, deleteBySourceId, query(embedding, k, options?: { filter? }) } — see Metadata & filtering | | ConversationStore | { createConversation, getHistory, appendMessage } — optional, for multi-turn chat (see Conversation memory) |

Full definitions: src/types.ts.

Quick start (plain Node.js script)

import {
  crawlAndIndex,
  retrieveAndAnswer,
  voyageEmbeddingProvider,
  anthropicLLMProvider,
  pgVectorStore,
} from "rag-core-engine";
import { Pool } from "pg";

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

const embeddingProvider = voyageEmbeddingProvider({ apiKey: process.env.VOYAGE_API_KEY! });
const llmProvider = anthropicLLMProvider({ apiKey: process.env.ANTHROPIC_API_KEY! });
const vectorStore = pgVectorStore({ pool });

// 1. Crawl a site and index it
await crawlAndIndex("https://example.com", {
  embeddingProvider,
  vectorStore,
  crawlOptions: { maxDepth: 2, maxPages: 100 },
});

// 2. Ask a question grounded in what was indexed
const result = await retrieveAndAnswer("What does this site do?", {
  embeddingProvider,
  vectorStore,
  llmProvider,
  topK: 5,
});

console.log(result.answer);
console.log(result.sources.map((s) => s.sourceRef));

Other pipeline entry points, all in src/pipeline.ts:

import { indexFile, indexText } from "rag-core-engine";

// Index an uploaded file (pdf, docx, markdown, plain text).
// PDFs are extracted via Claude, not a parsing library — see "PDF extraction" below —
// so a PDF requires `pdfExtraction`; other file types ignore it.
await indexFile(
  { filename: "handbook.pdf", buffer: fileBuffer, mimeType: "application/pdf" },
  { embeddingProvider, vectorStore, pdfExtraction: { apiKey: process.env.ANTHROPIC_API_KEY! } },
);

// Index raw text directly (e.g. from your own DB/CMS content)
await indexText(
  { id: "faq-1", content: "Refunds are processed within 5 business days." },
  { embeddingProvider, vectorStore },
);

Using it in Express

Build providers once (e.g. at app startup) and reuse them across requests — they're stateless aside from the pooled pg.Pool.

// rag.ts
import { Pool } from "pg";
import {
  voyageEmbeddingProvider,
  anthropicLLMProvider,
  pgVectorStore,
} from "rag-core-engine";

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

export const embeddingProvider = voyageEmbeddingProvider({ apiKey: process.env.VOYAGE_API_KEY! });
export const llmProvider = anthropicLLMProvider({ apiKey: process.env.ANTHROPIC_API_KEY! });
export const vectorStore = pgVectorStore({ pool });

// app.ts
import express from "express";
import { retrieveAndAnswer, crawlAndIndex } from "rag-core-engine";
import { embeddingProvider, llmProvider, vectorStore } from "./rag.js";

const app = express();
app.use(express.json());

app.post("/chat", async (req, res) => {
  const { question } = req.body;
  if (!question) return res.status(400).json({ error: "`question` is required" });

  const result = await retrieveAndAnswer(question, { embeddingProvider, vectorStore, llmProvider });
  res.json({ answer: result.answer, sources: result.sources.map((s) => s.sourceRef) });
});

app.post("/crawl", async (req, res) => {
  const { siteUrl } = req.body;
  const result = await crawlAndIndex(siteUrl, { embeddingProvider, vectorStore });
  res.json(result);
});

app.listen(3000);

Using it in NestJS

Wrap the providers in an injectable service so Nest's DI container owns their lifecycle; inject ConfigService for API keys instead of reading process.env directly.

// rag.service.ts
import { Injectable, OnModuleDestroy } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import { Pool } from "pg";
import {
  voyageEmbeddingProvider,
  anthropicLLMProvider,
  pgVectorStore,
  retrieveAndAnswer,
  crawlAndIndex,
  type EmbeddingProvider,
  type LLMProvider,
  type VectorStore,
} from "rag-core-engine";

@Injectable()
export class RagService implements OnModuleDestroy {
  private readonly pool: Pool;
  private readonly embeddingProvider: EmbeddingProvider;
  private readonly llmProvider: LLMProvider;
  private readonly vectorStore: VectorStore;

  constructor(config: ConfigService) {
    this.pool = new Pool({ connectionString: config.getOrThrow("DATABASE_URL") });
    this.embeddingProvider = voyageEmbeddingProvider({ apiKey: config.getOrThrow("VOYAGE_API_KEY") });
    this.llmProvider = anthropicLLMProvider({ apiKey: config.getOrThrow("ANTHROPIC_API_KEY") });
    this.vectorStore = pgVectorStore({ pool: this.pool });
  }

  async ask(question: string) {
    return retrieveAndAnswer(question, {
      embeddingProvider: this.embeddingProvider,
      vectorStore: this.vectorStore,
      llmProvider: this.llmProvider,
    });
  }

  async crawl(siteUrl: string) {
    return crawlAndIndex(siteUrl, {
      embeddingProvider: this.embeddingProvider,
      vectorStore: this.vectorStore,
    });
  }

  async onModuleDestroy() {
    await this.pool.end();
  }
}

// rag.controller.ts
import { Body, Controller, Post } from "@nestjs/common";
import { RagService } from "./rag.service.js";

@Controller("rag")
export class RagController {
  constructor(private readonly rag: RagService) {}

  @Post("chat")
  chat(@Body("question") question: string) {
    return this.rag.ask(question);
  }

  @Post("crawl")
  crawl(@Body("siteUrl") siteUrl: string) {
    return this.rag.crawl(siteUrl);
  }
}

// rag.module.ts
import { Module } from "@nestjs/common";
import { RagService } from "./rag.service.js";
import { RagController } from "./rag.controller.js";

@Module({
  providers: [RagService],
  controllers: [RagController],
})
export class RagModule {}

Chunking strategy

The indexing pipeline (crawlAndIndex, indexFile, indexText) chunks each source with semantic chunking by default (see src/semanticChunk.ts): sentences are embedded and a chunk boundary is placed wherever the embedding distance between consecutive sentences spikes above a percentile threshold — i.e. wherever the topic actually shifts — rather than at a fixed character count. This costs two rounds of embedding calls per source (the sentence-boundary pass, then the final per-chunk embedding), in exchange for chunks that stay on-topic.

Tune it via chunkOptions (all optional, defaults shown):

await crawlAndIndex("https://example.com", {
  embeddingProvider,
  vectorStore,
  chunkOptions: {
    bufferSize: 1, // neighbor sentences included when embedding each sentence, for more context-aware boundary detection
    breakpointPercentile: 95, // lower = more/smaller chunks; higher = fewer/larger chunks
    minChunkChars: 200, // don't split below this size even at a strong breakpoint
    maxChunkChars: 2000, // hard ceiling — falls back to a word-boundary split above this
  },
});

A plain fixed-size chunker (chunk(), size + overlap, no embedding calls) is still exported for cases where the extra embedding cost isn't worth it, but it isn't used by the pipeline by default:

import { chunk } from "rag-core-engine";

const chunks = chunk(source, { chunkSize: 1000, chunkOverlap: 150 });

Progress reporting

crawlAndIndex/indexFile/indexText all accept an optional onProgress callback (via IndexPipelineOptions) — useful for a progress UI on a crawl that can take a while:

await crawlAndIndex("https://example.com", {
  embeddingProvider,
  vectorStore,
  onProgress: (event) => {
    if (event.phase === "crawling") {
      console.log(`Crawling… ${event.pagesFound} pages found so far`);
    } else {
      console.log(`Indexing… ${event.completed} of ${event.total}`);
    }
  },
});

Two phases, since a crawl fully fetches every page before any indexing starts (see src/crawler.ts/src/pipeline.ts): "crawling" fires once per page as it's fetched, with a running count and no known total (the frontier changes as new links are discovered — bounded above by crawlOptions.maxPages, but a smaller site finishes well short of it); "indexing" fires once per source once the crawl is done and chunking/embedding/storing begins, with a known total. indexFile/ indexText only ever fire a single "indexing" event ({ completed: 1, total: 1 }).

PDF extraction

PDFs are extracted by handing them to Claude directly (via Anthropic's native document support), not a parsing library — see src/providers/pdfExtraction.ts. This reads scanned pages and complex layouts the way a human would, where a mechanical text extractor would garble them. DOCX/TXT/MD still use plain extraction (mammoth for DOCX) — no API call, no extra cost.

Because of this, indexFile/ingestFile need an Anthropic API key whenever the file is a PDF:

import { ingestFile } from "rag-core-engine";

const source = await ingestFile(
  { filename: "handbook.pdf", buffer, mimeType: "application/pdf" },
  { pdf: { apiKey: process.env.ANTHROPIC_API_KEY! } },
);

Omitting pdf/pdfExtraction when the file is a PDF throws immediately with a clear error rather than silently failing.

Conversation memory

retrieveAndAnswer is single-shot by default — no history, no persistence. To ground follow-up questions in prior turns, supply a ConversationStore implementation and let the pipeline manage history for you:

const result = await retrieveAndAnswer("What about the second point?", {
  embeddingProvider,
  vectorStore,
  llmProvider,
  conversationStore: myConversationStore,
  conversationId: previousResult?.conversationId, // omit on the first turn — one is created for you
  historyLimit: 5, // prior turns fed back into the prompt; default 5
});

console.log(result.conversationId); // pass this back in on the next turn

core ships no default ConversationStore implementation — unlike VectorStore, there's no framework-agnostic default that makes sense here, since it's just structured storage with no natural "reference" backend. Implement the interface against whatever you already have (a table, Redis, even an in-memory Map for a script):

import type { ConversationStore } from "rag-core-engine";

const myConversationStore: ConversationStore = {
  async createConversation() { /* insert a row, return its id */ },
  async getHistory(conversationId, limit) { /* return up to `limit` prior messages, oldest first */ },
  async appendMessage(conversationId, message) { /* insert { role, content } */ },
};

If nothing relevant is indexed yet (vectorStore.query returns no matches), retrieveAndAnswer returns a canned message (override via noContextMessage) without calling the LLM at all — no wasted API call for an unanswerable question.

Persona / system prompt

anthropicLLMProvider accepts an optional persona — a short description of who's answering, inserted into the system prompt's identity line. Defaults to a neutral, site-agnostic identity so it works out of the box; override it to give the chatbot a voice matching your site:

const llmProvider = anthropicLLMProvider({
  apiKey: process.env.ANTHROPIC_API_KEY!,
  persona: "a friendly, knowledgeable member of the Acme Corp team",
});

The rest of the prompt (formatting rules — when to use bullets vs. plain sentences, no [n] citation markers, keep answers tight, don't fabricate) is fixed and not currently configurable beyond persona. See src/providers/llm.ts for the full template, or implement your own LLMProvider (below) for full control.

Beyond chat: summarization & extraction

generate() is shaped specifically for RAG chat (a question, retrieved context chunks, conversation history, and a fixed system prompt). Every LLMProvider — including the bundled anthropicLLMProvider — also exposes two general-purpose methods that aren't tied to that shape at all, so the same provider/API key you already configured for chat can run other LLM tasks over your own documents:

interface LLMProvider {
  generate(question, context, history?): Promise<string>;   // RAG chat
  complete(prompt, options?): Promise<string>;               // freeform text
  extract<T>(prompt, schema: ZodType<T>, options?): Promise<T>; // structured
}

complete — freeform generation, e.g. summarization. prompt is used as-is (no context/history wrapping); steer the task with options.system:

const summary = await llmProvider.complete(documentText, {
  system: "Summarize the following document in 3 bullet points.",
  maxTokens: 300,
});

extract — same idea, but the result is validated against a Zod schema and returned typed, instead of hand-parsing JSON out of a text response:

import { z } from "zod";

const InvoiceSchema = z.object({
  vendor: z.string(),
  totalAmount: z.number(),
  dueDate: z.string(),
  lineItems: z.array(z.object({ description: z.string(), amount: z.number() })),
});

const invoice = await llmProvider.extract(documentText, InvoiceSchema, {
  system: "Extract the invoice details from this document.",
});
// invoice: { vendor: string; totalAmount: number; dueDate: string; lineItems: {...}[] }

Neither method touches the vector store or embedding provider — chunk/ semanticChunk/ingestFile/ingestText are still useful upstream (to get plain text out of a PDF/DOCX, or to split a long document before summarizing each piece), but retrieval-and-generation is only one thing you can build on core's primitives. A document-analysis pipeline that never indexes anything into a VectorStore at all is a perfectly normal way to use this package.

Vector store

The shipped pgVectorStore (see src/store/pgvector.ts) takes an existing pg.Pool you construct and manage — it never reads connection details from the environment. It lazily creates its table (default rag_chatbot_chunks, override with tableName), the pgvector extension, and an hnsw index the first time init() runs (which every pipeline function calls for you). It's hnsw, not ivfflat: ivfflat's index is trained by clustering whatever data exists in the table at CREATE INDEX time — since init() always runs on an empty table (it's the very first thing that happens), that index would be trained on zero rows, and later approximate searches can silently miss most (or all) of what gets inserted afterward, especially at the row counts typical of initial testing. hnsw builds incrementally as rows are inserted, with no equivalent footgun — if you're implementing your own VectorStore against pgvector directly, use hnsw too.

If you're running Postgres via Payload's postgresAdapter in dev mode (push-based schema sync), this table needs to be excluded from that sync — see the plugin README for why and how, even if you're not using the plugin itself.

Swap in a different backend by implementing the VectorStore interface yourself — nothing else in core needs to change:

import type { VectorStore } from "rag-core-engine";

const myVectorStore: VectorStore = {
  async init(dimensions) { /* ... */ },
  async upsert(chunks) { /* ... */ },
  async deleteBySourceId(sourceId) { /* ... */ },
  async query(embedding, k, options) { /* apply options?.filter, if you support it */ },
};

Metadata & filtering

Every Source/FileInput/TextInput accepts an optional metadata: Record<string, unknown>, which is copied onto every Chunk derived from it and stored alongside its embedding. VectorStore.query's third argument can then filter on it — useful any time you're indexing more than one "kind" of thing into the same store and need to narrow a search before ranking by similarity: candidate résumés by role/seniority, support articles by product, listings by category, and so on.

import { z } from "zod";

// Index a résumé with structured metadata pulled out by extract() —
// see "Beyond chat" above — attached for later filtering.
const CvSchema = z.object({
  category: z.string(),       // e.g. "backend", "design"
  yearsExperience: z.number(),
});
const parsed = await llmProvider.extract(resumeText, CvSchema, {
  system: "Classify this résumé's role category and total years of experience.",
});

await indexText(
  { id: "candidate-42", content: resumeText, metadata: parsed },
  { embeddingProvider, vectorStore },
);

// Later: find backend candidates with 3+ years, ranked by similarity to a job description
const [jobEmbedding] = await embeddingProvider.embed([jobDescriptionText]);
const matches = await vectorStore.query(jobEmbedding, 10, {
  filter: { category: "backend", yearsExperience: { gte: 3 } },
});

MetadataFilter fields are AND-ed together; each field is either a bare value (shorthand for { eq: value }) or an operators object:

| Operator | Meaning | Value type | | --- | --- | --- | | eq | equals | string \| number \| boolean | | ne | not equals | string \| number \| boolean | | gt / gte / lt / lte | numeric comparison | number | | in | one of | (string \| number \| boolean)[] |

There's no OR across top-level fields and no nested/grouped conditions — this covers "narrow the candidate pool, then rank by similarity," not a general query language. A chunk with no metadata, or missing the filtered field, never matches a filter on that field.

pgVectorStore implements this with a metadata JSONB column (added via ALTER TABLE ... ADD COLUMN IF NOT EXISTS on init(), so it backfills tables created before this existed) and a GIN index for filtered lookups. Implementing your own VectorStore? options?.filter is optional to support — a store that doesn't implement it can just ignore the parameter, at the cost of filter silently doing nothing rather than narrowing results.

Custom embedding / LLM providers

Same pattern — implement the interface, pass it in wherever a provider is expected:

import type { EmbeddingProvider, LLMProvider } from "rag-core-engine";

const openaiEmbeddingProvider: EmbeddingProvider = {
  dimensions: 1536,
  async embed(texts) {
    // call your embedding API, return one vector per input text
  },
};

const myLLMProvider: LLMProvider = {
  async generate(question, context, history = []) {
    // call your LLM, grounded in `context` (EmbeddedChunk[]) and optionally
    // `history` (ConversationMessage[], prior turns — empty unless the
    // caller passed a conversationStore to retrieveAndAnswer)
    return "...";
  },
};

Reference implementations for both live in src/providers/embedding.ts and src/providers/llm.ts if you want a starting point.

Provider registries

Constructing a provider directly (voyageEmbeddingProvider({...}), as above) is all you need if your app only ever uses one embedding provider, one LLM, and one vector store. If instead you want the choice of provider to be a runtime/config decision — e.g. a consuming app picks it from an env var, or offers multiple backends to different tenants — core also exposes three name → factory registries, pre-populated with its own bundled providers:

import { embeddingProviders, llmProviders, vectorStores } from "rag-core-engine";

embeddingProviders.create("voyage", { apiKey: process.env.VOYAGE_API_KEY });
llmProviders.create("anthropic", { apiKey: process.env.ANTHROPIC_API_KEY });
vectorStores.create("pgvector", { pool });

Register your own provider under a new name — no fork of core required — and it's selectable the same way:

import { embeddingProviders } from "rag-core-engine";
import type { EmbeddingProvider } from "rag-core-engine";

embeddingProviders.register("openai", (config: { apiKey: string; model?: string }): EmbeddingProvider => ({
  dimensions: 1536,
  async embed(texts) {
    // call OpenAI's embeddings API, return one vector per input text
  },
}));

embeddingProviders.create("openai", { apiKey: process.env.OPENAI_API_KEY });

ProviderRegistry (also exported) is the generic class behind all three — use it directly if you want the same by-name pattern for something of your own outside these three slots. See payload-plugin-rag-chatbot's README for how that package uses this to select providers by name from plugin options / environment variables, including a per-slot apiKey/provider env var fallback convention you may want to mirror in your own app.

Configuration reference

Every option across every provider and pipeline function, in one place — each links back to where it's used above. All are optional except where noted; core never reads process.env itself, so every value here has to be passed explicitly.

voyageEmbeddingProvider(options: VoyageEmbeddingOptions)

| Option | Type | Required | Default | | --- | --- | --- | --- | | apiKey | string | yes | — | | model | string | no | "voyage-3" | | dimensions | number | no | 1024 — must match what model actually returns | | baseUrl | string | no | "https://api.voyageai.com/v1" | | retryOnRateLimit | boolean | no | false — retry a 429 with exponential backoff (20s base, doubling, capped at 90s, up to 8 attempts) instead of throwing immediately. Useful for accounts on Voyage's unbilled tier (3 requests/minute) | | onRateLimited | (attempt: number, delayMs: number) => void | no | Called right before each retry wait, e.g. for logging |

anthropicLLMProvider(options: AnthropicLLMOptions)

| Option | Type | Required | Default | | --- | --- | --- | --- | | apiKey | string | yes | — | | model | string | no | "claude-opus-5" | | maxTokens | number | no | unset (AI SDK/model default) — also the fallback for complete/extract calls that don't pass their own maxTokens | | persona | string | no | a neutral, site-agnostic identity — see Persona / system prompt |

Per-call options for complete(prompt, options?) / extract(prompt, schema, options?) — see Beyond chat — are an LLMTaskOptions: { system?: string; maxTokens?: number }, both optional.

extractPdfText(buffer, options: PdfExtractionOptions) — also accepted as pdfExtraction/pdf on IndexPipelineOptions/ingestFile

| Option | Type | Required | Default | | --- | --- | --- | --- | | apiKey | string | yes | — | | model | string | no | "claude-opus-5" | | maxTokens | number | no | 8192 |

pgVectorStore(options: PgVectorStoreOptions)

| Option | Type | Required | Default | | --- | --- | --- | --- | | pool | pg.Pool | yes | — you construct and manage it | | tableName | string | no | "rag_chatbot_chunks" |

CrawlOptions (crawlAndIndex's crawlOptions / crawlSite's second argument)

| Option | Type | Default | | --- | --- | --- | | maxDepth | number | 3 | | maxPages | number | 200 | | userAgent | string | "rag-core-engine/0.1 (+https://github.com/)" |

SemanticChunkOptions (chunkOptions on IndexPipelineOptions — see Chunking strategy)

| Option | Type | Default | | --- | --- | --- | | bufferSize | number | 1 | | breakpointPercentile | number | 95 | | minChunkChars | number | 200 | | maxChunkChars | number | 2000 |

ChunkOptions (fixed-size chunk() — not used by the pipeline by default)

| Option | Type | Default | | --- | --- | --- | | chunkSize | number | 1000 | | chunkOverlap | number | 150 |

RetrieveAndAnswerOptions (beyond the required embeddingProvider/vectorStore/llmProvider)

| Option | Type | Default | | --- | --- | --- | | topK | number | 5 | | filter | MetadataFilter | none — no narrowing | See Metadata & filtering. | | historyLimit | number | 5 — only relevant when conversationStore is supplied | | conversationStore | ConversationStore | none — single-shot Q&A, no history | | conversationId | string | none — auto-created when conversationStore is supplied without one | | noContextMessage | string | a built-in "nothing indexed yet" message |

API reference

All exports come from a single entry point — import { ... } from "rag-core-engine", never deep imports into dist/ or src/. See src/index.ts for the full list:

• Pipeline: crawlAndIndex, indexFile, indexText, retrieveAndAnswer
• Building blocks: crawlSite, ingestFile, ingestText, chunk (fixed-size), semanticChunk (pipeline default)
• Reference providers: voyageEmbeddingProvider, anthropicLLMProvider, extractPdfText, pgVectorStore
• Provider registries (see Provider registries): embeddingProviders, llmProviders, vectorStores, ProviderRegistry
• Types: everything in src/types.ts (including ConversationMessage, ConversationStore, SemanticChunkOptions, LLMTaskOptions — the options type for complete/extract — MetadataFilter/MetadataFilterOperators/MetadataPrimitive/VectorQueryOptions — see Metadata & filtering — and CrawlProgressEvent — see Progress reporting), plus each provider's options type (VoyageEmbeddingOptions, AnthropicLLMOptions, PdfExtractionOptions, PgVectorStoreOptions, IndexPipelineOptions, RetrieveAndAnswerOptions, ProviderFactory)

Development

pnpm --filter rag-core-engine build       # compile src/ -> dist/
pnpm --filter rag-core-engine typecheck
pnpm --filter rag-core-engine test

See SRS.md for the full requirements/design spec, including the constraints that keep this package portable (no framework imports, no direct process.env reads, providers always injected).