Crux
CookbookRAG

Semantic RAG with Upstash

Index a doc corpus into Upstash Vector, compose a retrieval recipe, and answer questions with citations.

This recipe sets up a complete production-style RAG flow: ingest a directory of docs, chunk + embed + write into Upstash Vector, compose a named retrieval recipe, then answer questions with grounded citations. Hybrid mode gives you both semantic and keyword recall, while recipe steps improve the final evidence before it reaches the prompt.

Primitives used

  • @use-crux/ingest for document loading
  • @use-crux/core/indexing: indexer() for chunk + embed + write, corpus() for incremental sync
  • @use-crux/core/retrieval: retriever() and retrievalRecipe() for query-time evidence selection
  • @use-crux/convex and @use-crux/upstash: convexRecordStore() for JSON records and upstashVectorStore() for managed vector search
  • embedding(): one dense + one sparse embedding

When to reach for this pattern

  • You need a production-ready RAG without managing vector infrastructure yourself
  • Your corpus is document-shaped: Markdown, HTML, files
  • Both semantic and keyword matching matter (most real corpora benefit from hybrid)

Full code

lib/rag/embedding.ts

import { embedding } from "@use-crux/core/embedding";
import { openai } from "@ai-sdk/openai";

export const dense = embedding({
  kind: "dense",
  name: "text-embedding-3-small",
  dimensions: 1536,
  maxInputTokens: 8191,
  embed: async (inputs) => {
    const { embedMany } = await import("ai");
    const { embeddings } = await embedMany({
      model: openai.embedding("text-embedding-3-small"),
      values: inputs.map((input) => {
        if (input.type !== "text") throw new Error("text-only provider invariant");
        return input.text;
      }),
    });
    return { embeddings };
  },
});

export const sparse = embedding({
  kind: "sparse",
  name: "bm25",
  maxInputTokens: 8191,
  embed: async (texts) => ({ embeddings: texts.map(toSparseVector) }),
});

lib/rag/storage.ts

import { convexRecordStore } from "@use-crux/convex";
import { upstashVectorStore } from "@use-crux/upstash";
import { Index } from "@upstash/vector";

export const records = convexRecordStore({
  component: components.crux,
  ctx,
});

export const vectors = upstashVectorStore({
  index: new Index({
    url: process.env.UPSTASH_VECTOR_URL!,
    token: process.env.UPSTASH_VECTOR_TOKEN!,
  }),
  namespace: "product-docs",
});

lib/rag/index-docs.ts

import {
  chunker,
  corpus,
  indexer,
  indexingPipeline,
  transform,
} from "@use-crux/core/indexing";
import { filesSource } from "@use-crux/ingest/files";

import { dense, sparse } from "./embedding";
import { records, vectors } from "./storage";

export const docsIndexer = indexer({
  id: "product-docs",
  namespace: "product-docs",
  records,
  vectors,
  dense,
  sparse,
  cache: true,
  pipeline: indexingPipeline({
    documents: [
      transform.document({
        name: "tag-source",
        version: "1",
        run(document) {
          return {
            ...document,
            metadata: { ...document.metadata, corpus: "product-docs" },
          };
        },
      }),
    ],
    chunker: chunker.structured({ tableRowsPerChunk: 25 }),
  }),
});

export const docsCorpus = corpus({
  id: "product-docs",
  namespace: "product-docs",
  records,
  indexer: docsIndexer,
});

export async function syncDocs() {
  const source = filesSource(
    { directory: "./docs", recursive: true },
    { namespace: "product-docs" },
  );
  return docsCorpus.sync(source.load(), {
    sourceSet: "complete",
    stale: "delete",
  });
}

Run pnpm tsx lib/rag/index-docs.ts (or invoke syncDocs()) when content changes. Because this uses corpus.sync(), unchanged files are skipped, changed files are reindexed, and deleted files are removed when the folder scan is complete.

lib/rag/retriever.ts

import {
  retrievalRecipe,
  retrieve,
  rerank,
  retriever,
} from "@use-crux/core/retrieval";

import { dense, sparse } from "./embedding";
import { retrievalModel } from "./retrieval-model";
import { records, vectors } from "./storage";

export const docs = retriever({
  id: "product-docs",
  namespace: "product-docs",
  records,
  vectors,
  dense,
  sparse,
  search: { mode: "hybrid", limit: 5, fusion: "rrf" },
});

export const docsAnswer = retrievalRecipe({
  id: "product-docs-answer",
  retriever: docs,
  model: retrievalModel,
  steps: [
    retrieve({ limit: 20 }),
    rerank({
      engine: judgeReranker({ name: "docs-judge", model: retrievalModel }),
      topK: 5,
    }),
  ],
});

const { hits, trace } = await docsAnswer.retrieveWithTrace(
  "enterprise SSO setup",
);

Use the plain docs retriever when hybrid retrieval is enough. Use docsAnswer when you need traceable recipe steps such as reranking, parent expansion, compression, or federation.

lib/rag/answer.ts

import { prompt } from "@use-crux/core";
import { generate } from "@use-crux/ai";
import { grounding, citationSchema } from "@use-crux/core/citations";
import { openai } from "@ai-sdk/openai";
import { z } from "zod";

import { docsAnswer } from "./retriever";

const groundedDocs = grounding({
  id: "product-docs",
  retriever: docsAnswer.asRetriever(),
  query: ({ input }) => input.question,
  limit: 5,
  citations: {
    required: true,
    quotes: "required",
  },
});

const answerPrompt = prompt({
  id: "docs-answer",
  use: [groundedDocs],
  input: z.object({ question: z.string() }),
  output: z.object({
    answer: z.string(),
    citations: z.array(citationSchema),
  }),
  system: `Answer using only the provided retrieved docs. Cite the chunks you used.
     If the docs don't contain the answer, say so.`,
  prompt: ({ input }) => input.question,
});

export async function answer(question: string) {
  const result = await generate(answerPrompt, {
    model: openai("gpt-4o"),
    input: { question },
  });
  return result.object;
}

How it works

  1. Sync deliberately, query many times. Syncing is expensive (pipeline work + embedding + writes); querying is cheap. corpus.sync() makes repeated syncs safe by skipping unchanged sources and tracking failures.
  2. Hybrid retrieval combines two strengths. Dense embeddings catch semantic similarity, while sparse embeddings catch exact terms and rare keywords. RRF (Reciprocal Rank Fusion) merges those rankings without needing manual score normalization.
  3. The recipe sharpens the final shortlist. Retrieval gets candidate chunks. Recipe steps can rerank, expand parents, compress evidence, and produce a structured trace.
  4. grounding() runs retrieval at prompt-resolve time. Every call to answer() re-runs retrieval with the user's question, injects the allowed evidence, and contributes a citation constraint.
  5. Structured citations are validated. Each citation captures sourceId, chunkId, and optional quotes. Crux validates that cited chunks came from the retrieved hit set, whose source contains the allowed URL/path attribution.
  6. The indexing ledger explains sync behavior. Each source record stores transform, chunking, cache, duration, and chunk-count records. In devtools, CLI/TUI, and OTel this is the difference between "RAG feels stale" and knowing exactly which source changed.

Variations

Sync one changed source

When a single doc changes and your job only sees that one file, use a partial source set. Crux will update that source without assuming missing sources were deleted.

await docsCorpus.sync(
  fileSource("./docs/changed.md", { namespace: "product-docs" }).load(),
  { sourceSet: "partial" },
);

Citations as URLs

If your docs come from URLs (urlsSource), the hit carries source.url. Override the grounding renderer to format the evidence with clickable labels:

const urlGrounding = grounding({
  id: "url-docs",
  retriever: docs,
  query: ({ input }) => input.question,
  render: ({ hits }) =>
    hits
      .map(
        (hit, index) =>
          `[${index + 1}] ${hit.parent?.title ?? hit.source.id} - ${hit.source.url ?? hit.chunkId}\n${hit.content}`,
      )
      .join("\n\n"),
  citations: {
    required: true,
    quotes: "optional",
  },
});

Where to next

Local files RAG

The same retrieval stack with an in-memory local corpus.

Retrieval guide

The full RAG architecture: embeddings, ingestion, indexing, retrieval.

Episodic recall

Auto-recall past chat turns instead of indexed docs.

@use-crux/upstash reference

Upstash VectorStore and Redis RecordStore APIs.

Local Files RAG

Index a local docs directory, compose a retrieval recipe, and answer questions with citations.

Episodic recall

Recall relevant past chat events with the episodes memory block.

On this page

Primitives usedWhen to reach for this patternFull codelib/rag/embedding.tslib/rag/storage.tslib/rag/index-docs.tslib/rag/retriever.tslib/rag/answer.tsHow it worksVariationsSync one changed sourceCitations as URLsWhere to next