stackbone.rag

Parse, chunk, embed, store and retrieve documents on top of the agent's own Postgres. The RAG schema lives in the same database as stackbone.database and shares the same connection pool, so an ingest and a write to your own tables can happen inside one transaction. You reach it through the ambient stackbone client — the same handle on every surface, from an agent tool or a durable workflow step.

import { stackbone } from '@stackbone/sdk';

const result = await stackbone.rag.retrieve({
  text: 'how do I reset my password?',
  topK: 5,
});

stackbone.rag runs identically across local development, self-host and cloud, because it sits on the agent's own Postgres — its only dependency is the DATABASE_URL the runtime injects. There is no separate hosted RAG service to provision.

Mental model

stackbone.rag is a thin façade over four canonical tables — rag_collections, rag_documents, rag_chunks and rag_jobs, in the managed stackbone_platform schema — that Stackbone provisions in your agent's Postgres for you. Three properties follow from this and shape the rest of the page:

• One database. The tables live in the agent's Postgres (DATABASE_URL), in the stackbone_platform schema, next to your own tables. There is no separate "RAG store".
• One pool. stackbone.rag reuses the pool that stackbone.database builds, so an agent that touches both surfaces opens exactly one pool and can wrap RAG operations and your own writes in the same stackbone.database.transaction(...) callback.
• Already there. The schema is provisioned for you on every install — you never run a migration to create it, and there is no ad-hoc DDL on first call. You only read and write through stackbone.rag.

Embeddings are computed inside the agent process through the same OpenRouter-compatible endpoint that stackbone.ai uses. The default model is openai/text-embedding-3-small (1536 dims), frozen for schema stability — the dimensionality is part of the canonical schema, so changing it would mean re-ingesting everything.

Setup

The RAG schema is provisioned for you — there is nothing to install. The four tables are present in your agent's database from the first boot, so you can call stackbone.rag straight away. (An operator can even create collections and ingest documents from the Studio UI before your code ever reads them back — see Managed ingestion from Studio.)

Configure embeddings (optional)

The default embedding model is openai/text-embedding-3-small (1536 dims), fixed for schema stability. It is read from your project config; you only need to set it if you want a different embedding-capable model. The SDK reads it from your agent.yaml:

rag:
  embeddingModel: openai/text-embedding-3-small # default

• embeddingModel — any model id the OpenRouter-compatible embeddings endpoint resolves. Changing the model on an existing collection requires re-ingestion, because the vector dimension is fixed by the schema.

You can also pass model explicitly on any ingest/retrieve call to override the configured default per request. Leave it unset and the default above applies.

Where RAG calls live

stackbone.rag is the ambient handle — you import it once and call it from wherever your code runs. Two places matter for an agent:

• From an agent tool. Retrieve inside a tool call and return the hits, so the model can use them to answer. This is how a deep agent grounds its replies in your documents. See LangChain's tool docs.
• From a durable workflow step. Ingest is I/O that can be slow and can fail, so put it inside a 'use step' — the Workflow SDK runs each step once, persists its result and auto-retries on failure, which is exactly what you want for a parse → chunk → embed → store pipeline.

The pure helpers parse and chunk touch nothing external, so you can also call them from a one-off script or a test.

Ingestion

Pure helpers — parse and chunk

stackbone.rag.parse(...) and stackbone.rag.chunk(...) are pure helpers. They never touch the database and never call the embedding provider, so you can use them anywhere — in a tool, a step, a script or a test — to iterate on chunking before committing to ingestion:

import { stackbone } from '@stackbone/sdk';

// `parse` accepts the document bytes (or a plain string / Blob), not a URL —
// fetch the source yourself first.
const bytes = await fetch('https://example.com/post').then((r) => r.arrayBuffer());
const text = await stackbone.rag.parse(bytes);

const chunks = stackbone.rag.chunk(text, { size: 512, overlap: 64 });

parse accepts a string, a Blob, a Uint8Array or an ArrayBuffer. Markdown and plain text come back verbatim; PDFs are flattened to one string. The reader is text-only (no OCR): a scanned, image-only PDF has no text layer and is rejected with a clear error rather than ingested as an empty document.

Synchronous ingest — stackbone.rag.ingest

For one-off ingestion (a setup script, a small upload handler) call ingest directly. The promise resolves once every chunk has been embedded and persisted:

import { stackbone } from '@stackbone/sdk';

const bytes = await fetch('https://example.com/post').then((r) => r.arrayBuffer());
const text = await stackbone.rag.parse(bytes);

const result = await stackbone.rag.ingest({
  id: 'post-123',
  collection: 'help-center',
  chunks: stackbone.rag.chunk(text),
  metadata: { source: 'help-center', url: 'https://example.com/post' },
});

if (result.error) {
  // SdkError — see "Errors" below.
  throw new Error(`${result.error.code}: ${result.error.message}`);
}

console.log(`Ingested ${result.data.chunks} chunks for ${result.data.id}.`);

When you pass string chunks without a model, the SDK embeds them with your configured default. Re-ingesting with the same id replaces all chunks for that document atomically — useful for periodic re-crawls.

You can pass onProgress to observe per-chunk progress without paying for the SQL job writer:

await stackbone.rag.ingest({
  id: 'post-123',
  collection: 'help-center',
  chunks,
  onProgress: (event) => console.log(event.type, event),
});

Ingest inside a durable workflow step

A real ingest is slow and can fail mid-flight. Put it in a workflow 'use step' so it runs once, persists its result and auto-retries on transient failures:

import { stackbone, z } from '@stackbone/sdk';

export const inputSchema = z.object({ url: z.string().url(), id: z.string() });
export const outputSchema = z.object({ chunks: z.number() });

export async function indexDocsWorkflow(input: z.infer<typeof inputSchema>) {
  'use workflow';
  return ingestStep(input.url, input.id);
}

async function ingestStep(url: string, id: string) {
  'use step'; // runs once, persisted, retried on failure — keep it idempotent
  const bytes = await fetch(url).then((r) => r.arrayBuffer());
  const text = await stackbone.rag.parse(bytes);

  const result = await stackbone.rag.ingest({
    id,
    collection: 'help-center',
    chunks: stackbone.rag.chunk(text),
    metadata: { source: 'help-center', url },
  });
  if (result.error) throw new Error(result.error.code);
  return { chunks: result.data.chunks };
}

Durable ingest in one call — ingestDocuments

The step above wires the parse → chunk → embed → store pipeline by hand. When you just want to ingest a document durably from a workflow and don't need to customise the pipeline, call ingestDocuments from @stackbone/sdk/workflow. It runs the built-in ingest workflow as a durable sub-workflow and resolves once the document is stored, returning { documentId, chunks }:

import { ingestDocuments } from '@stackbone/sdk/workflow';

export async function indexDocsWorkflow(input: { markdown: string }) {
  'use workflow';

  // Hand it the text — the helper stages it to storage and ingests it durably.
  const { documentId, chunks } = await ingestDocuments({
    collection: 'docs',
    content: input.markdown,
    contentType: 'text/markdown',
  });

  return { documentId, chunks };
}

It takes one of two input shapes:

• { content } — inline text or bytes. The helper stages it to the agent's storage for you (in a durable, replay-safe step keyed by the content hash) and then ingests it.
• { storageKey } — a file already staged under the reserved rag/ prefix (for example one an operator uploaded from Studio). The helper ingests it directly, so a large file never travels in the workflow input.

Both must be called from a workflow body. If your workspace ships its own version of the ingest workflow, ingestDocuments uses it automatically. For a fully custom pipeline, call stackbone.rag.ingest inside your own step, as above.

Asynchronous ingest — stackbone.rag.ingestAsync

For a long-running ingest you want to observe outside a workflow (a webhook handler that must return in seconds, a backfill that takes minutes) call ingestAsync. It allocates a rag_jobs row synchronously, returns the job id immediately, and exposes both a streaming channel of progress events and the final Result:

import { stackbone } from '@stackbone/sdk';

const handle = await stackbone.rag.ingestAsync({
  id: 'post-123',
  collection: 'help-center',
  chunks,
});

if (handle.error) throw new Error(handle.error.code);

// You now have a stable job id — hand it back to the caller and return early.
const { jobId } = handle.data;

// Drain progress events into your own logs / stream when you want to follow it.
for await (const event of handle.data.events) {
  switch (event.type) {
    case 'started':
      console.log('rag.ingest.started', event);
      break;
    case 'progress':
      console.log('rag.ingest.progress', event);
      break;
    case 'completed':
      console.log('rag.ingest.completed', event);
      break;
    case 'failed':
      console.error('rag.ingest.failed', event);
      break;
  }
}

rag_jobs is the observability surface. Studio's RAG explorer reads it; during local development GET /api/rag/jobs returns it; cancelling a job from Studio (or via POST /api/rag/jobs/:jobId/cancel) flips the row's status, which the worker observes between chunk batches and exits with rag_ingest_cancelled.

Retrieval

stackbone.rag.retrieve is the only read API. Pass a text query to let the SDK embed it for you. The most common place to call it is inside an agent tool, returning the hits so the model can ground its answer:

import { tool } from '@langchain/core/tools';
import { stackbone, z } from '@stackbone/sdk';

const searchDocs = tool(
  async ({ query }: { query: string }) => {
    const result = await stackbone.rag.retrieve({ text: query, topK: 5 });
    if (result.error) throw new Error(result.error.code);

    // Return the hits to the model so it can answer from them.
    return result.data.map((hit) => ({
      score: hit.score,
      content: hit.content,
    }));
  },
  {
    name: 'search_docs',
    description: 'Search the help-center knowledge base before answering a question.',
    schema: z.object({ query: z.string() }),
  },
);

topK defaults to 5. Each hit is { id, chunkIdx, content?, metadata?, score }, ordered by descending cosine similarity (score is in [0, 1], 1 = identical).

Filtering on metadata

Pass filter to scope the search to documents whose metadata matches a JSON sub-object (matched server-side via jsonb @> $1):

const result = await stackbone.rag.retrieve({
  text: 'pricing tiers',
  filter: { source: 'help-center', locale: 'en' },
  topK: 10,
});

V1 supports exact-match equality on simple paths. Range queries and jsonpath operators are out of scope for now.

Precomputed embeddings

If you already have an embedding (e.g. from your own embedder or a cached vector), pass it directly:

await stackbone.rag.retrieve({
  embedding: cachedQueryVector, // number[]
  topK: 5,
});

The pipeline skips the auto-embed step entirely.

Deletion

Two atomic deletion APIs:

// Delete one or many documents by id (re-ingest replaces; this removes).
await stackbone.rag.delete(['post-123', 'post-124']);

// Delete every chunk whose metadata matches a sub-object.
await stackbone.rag.deleteWhere({ source: 'help-center' });

Both run inside a single SQL statement and cascade through to chunks via the foreign key. Pass { namespace } to scope either call.

Cross-table transactions

Because RAG and your own tables share the pool, you can ingest a document and update your own documents table in the same transaction. If anything throws, both roll back together. This works the same from a workflow step or an agent tool — here it is inside a step:

import { stackbone } from '@stackbone/sdk';
import { eq } from '@stackbone/sdk/db';
import { documents } from '../src/schema';

async function reindexStep(id: string, parsedText: string) {
  'use step';
  await stackbone.database.transaction(async (tx) => {
    // Your own row — uses the tx-scoped Drizzle handle.
    await tx.update(documents).set({ ragIndexedAt: new Date() }).where(eq(documents.id, id));

    // RAG ingest — runs against the same connection.
    const result = await stackbone.rag.ingest({
      id,
      collection: 'help-center',
      chunks: stackbone.rag.chunk(parsedText),
    });
    if (result.error) throw new Error(result.error.code);
  });
}

RAG and your own tables share one connection pool, so they can participate in the same transaction. That single-pool guarantee is what makes this atomic boundary possible — there is no second connection to leave half-committed.

Inspecting RAG data

Two places see the same rows:

• During local development — stackbone dev exposes GET /api/rag/collections, POST /api/rag/collections/:name/query, GET /api/rag/jobs?status=... and POST /api/rag/jobs/:jobId/cancel. Studio's RAG explorer is the primary consumer.
• Studio's RAG explorer — gated on the rag.basic capability the agent advertises in its contract handshake. If the agent does not advertise it, Studio hides the explorer entirely instead of showing a half-broken view. The explorer is read-write: a human can create a collection, upload a file and ingest it from the screen — see Managed ingestion from Studio below.

Managed ingestion from Studio

stackbone.rag.ingest is the code path: your agent calls it at runtime. There is also a human path — anyone who installed your agent (and, locally, you as the creator running stackbone dev) can feed a collection straight from the Studio screen, with no ingest code to write. Both paths land in the same rag_* tables, so the agent retrieves manually-uploaded chunks with stackbone.rag.retrieve() exactly as if your code had ingested them — there is no second-class "uploaded by hand" flag.

What a person can do from the screen:

• Create an empty collection with a name and an optional description, before uploading anything.
• Upload a file (.txt, .md or .pdf) into a collection. The upload returns immediately with a job id; the parse → chunk → embed → store work runs in the background, so the browser never freezes on a large PDF.
• Watch the job move through queued → running → succeeded | failed in the Jobs tab, with per-chunk progress. A failed job can be retried without re-uploading; a running job can be cancelled (it stops with rag_ingest_cancelled, the same status your code sees).
• Delete a single document (its chunks cascade) or delete a whole collection (documents + chunks cascade).
• Download the original file that was uploaded — the source is kept, linked to its document row.

Two contracts your agent should be aware of:

• Same embedding model as your agent. Manual uploads are embedded with the model your agent uses to search (the configured embeddingModel, default openai/text-embedding-3-small). That is what makes a hand-uploaded document findable by stackbone.rag.retrieve() — if the two used different models the agent would never match it.
• Gated on the schema existing. The upload UI only appears once the rag_* tables exist — i.e. once your agent actually uses stackbone.rag. An agent that never touches RAG shows the usual empty state, with no way to feed a RAG nothing reads. Studio never creates or migrates the schema itself; authoring the schema stays with the platform.

The original files uploaded this way are kept in object storage under a reserved, read-only rag/ prefix — see stackbone.storage.

Custom pgvector columns

If you need a vector column on your own table — e.g. an embedding field on documents you maintain by hand — vector is re-exported from @stackbone/sdk/db:

import { pgTable, text, vector } from '@stackbone/sdk/db';

export const documents = pgTable('documents', {
  id: text('id').primaryKey(),
  content: text('content').notNull(),
  embedding: vector('embedding', { dimensions: 1536 }).notNull(),
});

The first migration that introduces a vector(...) column emits a CREATE EXTENSION IF NOT EXISTS vector automatically; the RAG installer also ensures the extension exists, so the order in which you add custom and platform vector columns does not matter.

Errors

Every stackbone.rag.* call returns Result<T> — check .error and branch on error.code:

The contract gate adds contract_version_unsupported, capability_unavailable, contract_unreachable and contract_malformed. Every gated stackbone.rag method requires the rag.basic capability — see overview.

Pure helpers (parse, chunk) never hit the database or the embedding provider and are intentionally exempt from the gate.

Internal subpaths

@stackbone/sdk exposes one low-level subpath behind the RAG pipeline. It exists for the SDK's own use — creators usually do not import it directly. It is listed here so the surface is discoverable when something low-level is needed.

This subpath is internal to the platform contract — it may change shape between SDK majors.

Where to go next

• stackbone.database — the same pool RAG runs on, for your own tables and migrations.
• Agents & sessions — where a retrieval tool lives, and how an agent grounds answers in your documents.
• Workflows — where a long ingest belongs, as a step that persists and auto-retries.
• @stackbone/sdk overview — the rest of the surfaces: storage, AI, approval, secrets, config.