How to build a RAG system

A working RAG stack is two pipelines that share an embedding space: a deterministic ingest workflow that writes vectors, and a tool-called retriever that reads them. The hard part is the tenant boundary. If the model can choose whose index to search, you do not have RAG. You have a leak.

This design uses Mastra workflows, Reducto Parse for layout-aware chunking, OpenAI text-embedding-3-small (dense) plus pinecone-sparse-english-v0 (sparse), and a single Pinecone serverless hybrid index partitioned by namespace = organizationId. Identity stays where it already lives: Better Auth organizations and Hono request context, as in Building a Multi-Tenant Backend with Hono, Better Auth, Drizzle, and Postgres RLS. userId is the actor. organizationId is the library.

The write path ends at upsert. Hybrid search and rerank live on the read path — never in the upload line.

1. System shape

Files are the source of truth. Vectors are a derived index. Ingest status is the latest workflow run for that org and path, not a documents table.

Write
  POST /documents
  → S3 orgs/organizationId/docs/filename
  → workflow pinecone-document-ingestion
       loadFile → parseReducto → mapChunks → indexChunks
  → Pinecone index document-chunks, metric=dotproduct

Read (later turn)
  chat → agent
  → tool search-pinecone-documents
  → embed query (same dense + sparse models)
  → hybrid query (alpha-weighted) namespace=organizationId
  → rerank top results (cross-encoder)
  → relevantContext + sources
  → agent

• List files from the filesystem prefix. Ask “is this file searchable yet?” by listing workflow runs with resourceId = organizationId and matching the run’s input path.
• A documents row that can drift from S3 and Pinecone is a third source of truth you will have to reconcile.
• Mastra Workspace search (Workspace + PineconeVector + an embedder) indexes one agent sandbox. Keep it for the agent’s working set. Keep this workflow for the product corpus.
• The upload turn still carries file parts inline. Indexing is async. Retrieval is for subsequent turns, so later turns search the index instead of re-sending the file.
• Persist Reducto’s parse JSON (or its job id) next to the file. You can re-embed without re-parsing when you change embedding models.

Failure: treating Workspace search as the HTTP upload API. It has no membership-checked requestContext and no org-scoped library.

2. Why a workflow for ingest, a tool for retrieve

Mastra’s rule of thumb matches this split: workflows for defined multi-step processes, agents for decisions that use tools. Reducto fits as the parse step inside the ingest workflow — it is a deterministic transform, not a decision, and not a second workflow.

• Populate requestContext the same way on upload and on chat: session → verify membership in the route’s organizationId → requestContext.set("organizationId", organizationId) and requestContext.set("userId", userId).
• Never copy org from a client header, a tool argument, or databaseConfig the model can influence.
• Always-on injection embeds every greeting and burns tokens. Tool-calling keeps retrieval off the default path.

Failure: if the model can choose whose index to search, you have a leak.

3. Embedding contract

Ingest and query must use the same dense model and dimension. Cosine similarity across two embedding spaces is noise. Hybrid adds a second contract: the same sparse model at write and query.

import { ModelRouterEmbeddingModel } from "@mastra/core/llm"
import { embedMany } from "ai"
import { Pinecone } from "@pinecone-database/pinecone"

export const DENSE_MODEL_ID = "openai/text-embedding-3-small"
export const DENSE_DIMENSION = 1536
export const SPARSE_MODEL_ID = "pinecone-sparse-english-v0"

const denseModel = new ModelRouterEmbeddingModel(DENSE_MODEL_ID)
const pc = new Pinecone({ apiKey: process.env.PINECONE_API_KEY! })

export async function embedDense(values: string[]): Promise<number[][]> {
  if (values.length === 0) return []
  const { embeddings } = await embedMany({ model: denseModel, values })
  return embeddings
}

export async function embedSparse(values: string[], inputType: "passage" | "query") {
  if (values.length === 0) return []
  const { data } = await pc.inference.embed({
    model: SPARSE_MODEL_ID,
    inputs: values,
    parameters: { inputType, truncate: "END" },
  })
  return data.map((d) => ({
    indices: d.sparseIndices,
    values: d.sparseValues,
  }))
}

• Mastra’s documented path is ModelRouterEmbeddingModel with a provider/model id for dense. Pinecone hosts the sparse model; call it through the Pinecone SDK.
• A hybrid index stores both vectors on one record and requires metric: "dotproduct". Dense-only indexes in this repo used cosine; switching is a new index and a full re-ingest.
• Dimension is frozen at createIndex. Changing the dense model later means a new index and a full re-ingest.
• text-embedding-3-small caps input at 8191 tokens. Chunk first, then embed.
• Sparse scores are unbounded; dense cosine scores are [-1, 1]. Combine them at query time with an alpha weight (α·dense + (1-α)·sparse). Start at α = 0.75 for natural-language questions. Without weighting, the sparse component dominates.
• Batch embeddings and upserts. Pinecone’s practical ceiling is about 1000 vectors or 2 MB; at 1536-d with text in metadata, stay closer to 100.

Failure: embedMany on an unbounded PDF in one call, or querying hybrid without alpha so exact-match tokens drown semantics.

4. Vector layout: one index, many namespaces

Pinecone offers three tenant strategies. Only namespace = organizationId is acceptable for an org-scoped library.

Per-org indexes (no)
  document-chunks-org-a
  document-chunks-org-b

Shared index + metadata filter (no)
  document-chunks
    filter organizationId=a | organizationId=b

Shared index + namespace (yes)
  document-chunks
    ns=org-a | ns=org-b

• Index per org: isolation is real; createIndex is slow; index-count limits kill you.
• Metadata filter: { organizationId }: one forgotten filter leaks the corpus. Query cost scales with the whole index — the pgvector pattern. Do not port it to Pinecone if namespaces exist.
• Namespace = organizationId: query, upsert, and delete always take namespace. Namespaces are created on first upsert. Pinecone stores them separately, so a missing metadata filter cannot see another tenant. Query cost is 1 RU per GB of the targeted namespace.
• Omitting namespace queries the default namespace, not every tenant. Isolation holds only if you never write that default. Always pass namespace: organizationId.
• Index names: 1–45 characters, lowercase alphanumeric and hyphens, start and end alphanumeric. Use document-chunks, not mastra_document_chunks.

import { Pinecone } from "@pinecone-database/pinecone"

export const PINECONE_INDEX_NAME = "document-chunks"

export const pc = new Pinecone({ apiKey: process.env.PINECONE_API_KEY! })

let indexReady: Promise<void> | undefined

export async function ensurePineconeIndex() {
  if (!indexReady) {
    indexReady = pc
      .createIndex({
        name: PINECONE_INDEX_NAME,
        vectorType: "dense",
        dimension: DENSE_DIMENSION,
        metric: "dotproduct",
        spec: { serverless: { cloud: "aws", region: "us-east-1" } },
      })
      .then(() => undefined)
      .catch((error: unknown) => {
        indexReady = undefined
        throw error
      })
  }
  await indexReady
}

• Hybrid on Pinecone is one dense index with sparse_values on the same record. The linkage is implicit: one upsert, one query, no client-side merge. Separate dense/sparse indexes are only for sparse-only queries or Pinecone-integrated sparse embedding — you do not need them.
• Mastra’s PineconeVector is a dense upsert/query wrapper. Hybrid means using the Pinecone SDK (or wrapping it) so you can set sparse_values. Keep Mastra for the workflow, agent, and requestContext; do not expect @mastra/pinecone to own hybrid.
• Per-vector metadata must be flat. Pinecone does not flatten nested objects. Keep text in metadata or retrieval returns scores with no prose.
• Reducto chunks carry citation metadata. Keep page, bbox, and blockType alongside path and chunkIndex so answers can point at a region, not just a file.
• organizationId does not belong in metadata as the isolation key. The namespace is the tenant. Metadata filters are for relevance inside that tenant (path, later).
• Document identity is path under the org prefix, typically docs/filename. Re-upload of the same path overwrites. Rename is a new document. Two files that collide on filename under docs/ are the same document.
• Flat metadata is { path, filename, chunkIndex, page, bbox, blockType, text, embedText, mediaType }. No nested objects.

Failure: omitting namespace looks like “search everything.” It searches the default namespace.

5. Ingest workflow

File upload and ingest belong in one workflow: pinecone-document-ingestion. “One workflow” is not “one step.” The DAG is four steps. The HTTP route only writes the object and starts the run.

Input matches the upload route so Studio and HTTP share a contract: { path, filename, mediaType }. startAsync is fire-and-forget: it returns { runId } immediately. The route does not wait for status or chunkCount.

POST /documents
  → writeFile(docs/filename)
  → createRun resourceId=organizationId
  → startAsync → runId
  → list runs by resourceId, match input path

pinecone-document-ingestion  (one run)
  loadFile       exists(path) → readFile → upload to Reducto
  parseReducto   variable chunks, embedding_optimized → parse JSON
  mapChunks      chunks → { text, embedText, page, bbox, blockType, chunkIndex }
  indexChunks    deleteMany(path) → embed dense+sparse → upsert

Stay in one workflow because it is one logical operation. Two workflows per file means two runs to join — the documents-table problem as a runs-join. Retries and snapshots are per-run: if upsert flakes after a successful parse, the same run retries indexChunks from the snapshot. It does not start a second pipeline that has to re-derive what the first already knew.

The boundary that matters is the step, not the workflow:

• parseReducto is retry-safe and should not re-parse on every retry. Reducto costs money per parse. Persist the parse JSON (or job id) in the step output — and next to the file — so a retry of indexChunks skips straight to mapChunks when a parse already exists for that path + content hash.
• Do not pass file bytes through step output. Workflow snapshots JSON-serialize that data. Each step reads by path from an already org-rooted filesystem. Only the parse JSON (small, structured) or its reference flows between steps.
• Delete-then-upsert stays inside indexChunks. Splitting delete into its own workflow lets it run without the matching upsert.

What does not belong in this workflow:

• Retrieval. Hybrid query and rerank are a tool on later turns. The upload line ends at upsert.
• Reducto Extract. Schema fill (invoice.total) has different inputs, outputs, and callers. It is a later product, a later workflow.
• Re-embed on model change. A maintenance workflow that reads the cached parse JSON. Share the mapChunks / indexChunks steps — export them and reuse — do not fork the upload-triggered run.

If a step later becomes independently useful or operationally heavy, extract the step (Mastra steps are composable), not a second ingest workflow. Parse that is slow and rate-limited becomes a durable/suspendable step so the same run waits without holding a worker. Per-chunk enrichment becomes a fifth step in the same DAG.

• Use Reducto Parse, not Extract. Parse is the RAG ingest step: pages, blocks, tables, figures, bounding boxes, citation-ready chunks. Extract is schema fill (invoice.total, contract.effectiveDate) — a second product, not the corpus builder.
• Layout-aware parse can also be local Docling. The workflow still treats parse as a deterministic step. That converter is Docling.
• Let Reducto chunk. retrieval.chunking.chunk_mode: "variable" targets semantic boundaries around 1000 characters. Add embedding_optimized: true so tables get an embed string for vectors and a content string for display. Filter Header, Footer, and Page Number blocks.
• Do not run MDocument.chunk() on Reducto output. That pays for structure-aware chunks and then smashes them with a token splitter. MDocument is an adapter here: fromMarkdown (or fromJSON) to hold text + metadata, not to re-chunk.
• Images go through the same Parse path (OCR / vision pass). You do not need a separate vision index unless you later search “find this screenshot” by pixels.
• Persist the Reducto parse JSON (or its job id) next to the file and in the parseReducto step output. Re-embed without re-parsing when you change models. Retry indexChunks without re-paying for Parse.
• getWorkspaceOrganizationId throws if the server forgot to set the org — the same membership-checked value the upload route put on requestContext.
• chunkIndex is the array index. Reducto chunks carry blocks[] with bbox/page; map them yourself into flat metadata.

const result = await reducto.parse.run({
  input: upload.fileId,
  retrieval: {
    chunking: { chunk_mode: "variable", chunk_size: 1000 },
    embedding_optimized: true,
    filter_blocks: ["Header", "Footer", "Page Number"],
  },
})

const chunks = result.result.chunks.map((chunk, chunkIndex) => ({
  chunkIndex,
  text: chunk.content,           // display
  embedText: chunk.embed,        // vectors
  page: chunk.blocks[0]?.bbox.page,
  bbox: chunk.blocks[0]?.bbox,
  blockType: chunk.blocks[0]?.type,
}))

Pinecone has no deleteFilter on upsert. pgvector can replace a document in one call; Pinecone cannot. If a re-upload produces fewer chunks, leftover neighbors survive.

const organizationId = getWorkspaceOrganizationId(requestContext)
const index = pc.Index(PINECONE_INDEX_NAME)

await index.namespace(organizationId).deleteMany({ path: inputData.path })

if (inputData.chunks.length === 0) return { status: "skipped", chunkCount: 0 }

const dense = await embedDense(inputData.chunks.map((c) => c.embedText))
const sparse = await embedSparse(inputData.chunks.map((c) => c.embedText), "passage")

await index.namespace(organizationId).upsert({
  records: inputData.chunks.map((chunk, i) => ({
    id: `${inputData.path}:${chunk.chunkIndex}`,
    values: dense[i],
    sparseValues: sparse[i],
    metadata: {
      path: inputData.path,
      filename: inputData.filename,
      chunkIndex: chunk.chunkIndex,
      page: chunk.page,
      blockType: chunk.blockType,
      text: chunk.text,
      mediaType: inputData.mediaType,
    },
  })),
})

• Delete by path in the namespace first, including on skip, then upsert if there is anything to write. Return { status: "skipped", chunkCount: 0 } or { status: "indexed", chunkCount }.
• Deterministic ids (${path}:${chunkIndex}) overwrite in place. They still do not remove chunkIndex values that no longer exist. Delete-first is the correct overwrite.
• Embed the embedText field, store the text field. They differ when Reducto summarizes tables.
• Batch embed and upsert (~100 at 1536-d with text in metadata).
• Wire the DAG as loadFile → parseReducto → mapChunks → indexChunks, then commit(). Register the store, the workflow, and the agent (with the search tool) on the Mastra instance. One workflow. Four steps.
• Delete then upsert is not atomic. Serialize ingest per { organizationId, path } in the upload handler if concurrent re-uploads matter.
• Serverless search is eventually consistent: a query on the same turn as startAsync can miss the new chunks. That is another reason retrieval is for later turns.

Failure: two concurrent re-uploads of the same path can interleave — delete A, delete B, upsert A, upsert B.

Fire-and-forget from the upload handler with resourceId set to the organization, not the user. resourceId scopes the run to the tenant. It does not join by path. To find the latest ingest for a file, list runs for that resourceId and match input.path.

const run = await pineconeDocumentIngestion.createRun({
  resourceId: organizationId,
})
const { runId } = await run.startAsync({
  inputData: { path, filename, mediaType },
  requestContext,
})

Composing the same DAG with LangGraph

If the team already lives in LangChain/LangGraph, the same DAG composes as a StateGraph. Nodes replace Mastra steps. A checkpointer's thread_id replaces runId as ingest status. Every invariant above still holds — org from the server, delete-first, embed embedText, store text.

ingestion-graph  (one thread per organizationId:path)
  loadFile        exists(path) → readFile → upload to Reducto
                  missing → Command goto deleteVectors { status: "skipped" }
  parseReducto    parse cache hit ? cached JSON : parse.run → persist JSON
  mapChunks       chunks → { text, embedText, page, bbox, blockType, chunkIndex }
  deleteVectors   deleteMany(path) in namespace=organizationId
  planBatches     chunks > 0 → Send upsertBatch × ceil(n / 100)
                  chunks = 0 → finalize
  upsertBatch     embed dense+sparse → upsert batch (parallel per Send)
  finalize        { status, chunkCount } from summed indexedCount

State is the shared memory every node reads. Fields with one writer stay plain zod. Fields written by the fan-out need a reducer — without one, the last parallel batch overwrites the rest.

import { StateSchema, ReducedValue } from "@langchain/langgraph"
import { z } from "zod/v4"

const Chunk = z.object({
  chunkIndex: z.number(),
  text: z.string(),      // display
  embedText: z.string(), // vectors
  page: z.number().optional(),
  blockType: z.string().optional(),
})

export const IngestState = new StateSchema({
  // input — organizationId is assembled by the server at invoke, never by the client
  path: z.string(),
  filename: z.string(),
  mediaType: z.string(),
  organizationId: z.string(),
  // intermediates — one writer each, so plain fields
  fileId: z.string().optional(),
  parseJson: z.unknown().optional(),
  chunks: z.array(Chunk).default(() => []),
  // fan-out accumulator — N upsertBatch workers write it, so it needs a reducer
  indexedCount: new ReducedValue(z.number().default(0), {
    inputSchema: z.number(),
    reducer: (current, n) => current + n,
  }),
  status: z.string().default(""),
  chunkCount: z.number().default(0),
})

export type IngestStateType = typeof IngestState.State

Nodes return partial updates; the graph applies them. loadFile returns Command — state update and routing in one return — so it has no static edge out of it.

import { Command, Send } from "@langchain/langgraph"
import type { IngestStateType } from "./state"

const BATCH_SIZE = 100 // ~100 vectors at 1536-d with text in metadata

export async function loadFile(state: IngestStateType) {
  if (!(await filesystem.exists(state.path))) {
    // skip-with-delete: route straight to the shared delete node
    return new Command({
      update: { status: "skipped", chunks: [] },
      goto: "deleteVectors",
    })
  }
  const file = await filesystem.readFile(state.path)
  const upload = await reducto.upload({ file })
  return new Command({ update: { fileId: upload.fileId }, goto: "parseReducto" })
}

export async function parseReducto(state: IngestStateType) {
  // retryPolicy retries this node; the cache check keeps a retry from re-paying for Parse
  const cached = await readParseCache(state.path)
  if (cached) return { parseJson: cached }

  const result = await reducto.parse.run({
    input: state.fileId!,
    retrieval: {
      chunking: { chunk_mode: "variable", chunk_size: 1000 },
      embedding_optimized: true,
      filter_blocks: ["Header", "Footer", "Page Number"],
    },
  })
  await writeParseCache(state.path, result)
  return { parseJson: result }
}

export async function mapChunks(state: IngestStateType) {
  const chunks = state.parseJson.result.chunks.map((chunk, chunkIndex) => ({
    chunkIndex,
    text: chunk.content,           // display
    embedText: chunk.embed,        // vectors
    page: chunk.blocks[0]?.bbox.page,
    blockType: chunk.blocks[0]?.type,
  }))
  return { chunks }
}

export async function deleteVectors(state: IngestStateType) {
  await pc
    .Index(PINECONE_INDEX_NAME)
    .namespace(state.organizationId)
    .deleteMany({ path: state.path })
  return {}
}

type BatchJob = {
  organizationId: string
  path: string
  filename: string
  mediaType: string
  batch: IngestStateType["chunks"]
  offset: number
}

export function planBatches(state: IngestStateType) {
  if (state.chunks.length === 0) return "finalize"
  const sends: Send[] = []
  for (let offset = 0; offset < state.chunks.length; offset += BATCH_SIZE) {
    sends.push(
      new Send("upsertBatch", {
        organizationId: state.organizationId,
        path: state.path,
        filename: state.filename,
        mediaType: state.mediaType,
        batch: state.chunks.slice(offset, offset + BATCH_SIZE),
        offset,
      }),
    )
  }
  return sends
}

export async function upsertBatch(job: BatchJob) {
  const dense = await embedDense(job.batch.map((c) => c.embedText))
  const sparse = await embedSparse(job.batch.map((c) => c.embedText), "passage")

  await pc
    .Index(PINECONE_INDEX_NAME)
    .namespace(job.organizationId)
    .upsert({
      records: job.batch.map((chunk, i) => ({
        id: `${job.path}:${job.offset + i}`,
        values: dense[i],
        sparseValues: sparse[i],
        metadata: {
          path: job.path,
          filename: job.filename,
          chunkIndex: job.offset + i,
          page: chunk.page,
          blockType: chunk.blockType,
          text: chunk.text,
          mediaType: job.mediaType,
        },
      })),
    })
  return { indexedCount: job.batch.length }
}

export async function finalize(state: IngestStateType) {
  return {
    status: state.indexedCount > 0 ? "indexed" : "skipped",
    chunkCount: state.indexedCount,
  }
}

Wire it. Command destinations are declared with ends. The fan-out is a conditional edge returning Send[]. Paid and flaky nodes get retryPolicy. The checkpointer makes each thread_id a resumable run.

import { StateGraph, START, END, MemorySaver } from "@langchain/langgraph"
import { IngestState } from "./state"

const checkpointer = new MemorySaver() // prod: PostgresSaver on the same Postgres as everything else

export const ingestionGraph = new StateGraph(IngestState)
  .addNode("loadFile", loadFile, { ends: ["parseReducto", "deleteVectors"] })
  .addNode("parseReducto", parseReducto, { retryPolicy: { maxAttempts: 3, initialInterval: 1.0 } })
  .addNode("mapChunks", mapChunks)
  .addNode("deleteVectors", deleteVectors)
  .addNode("upsertBatch", upsertBatch, { retryPolicy: { maxAttempts: 3, initialInterval: 1.0 } })
  .addNode("finalize", finalize)
  .addEdge(START, "loadFile")
  .addEdge("mapChunks", "deleteVectors")
  .addConditionalEdges("deleteVectors", planBatches, ["upsertBatch", "finalize"])
  .addEdge("upsertBatch", "finalize")
  .addEdge("finalize", END)
  .compile({ checkpointer })

The upload handler swaps createRun / startAsync for invoke on a thread named {organizationId}:{path}:

const config = { configurable: { thread_id: `${organizationId}:${path}` } }
void ingestionGraph.invoke({ path, filename, mediaType, organizationId }, config)

// "is this file searchable yet?" — the latest checkpoint is the latest run for that org + path
const { values } = await ingestionGraph.getState(config)

• loadFile routes with Command, so it has no addEdge out of it. Add one anyway and both targets run — the skip path would parse a missing file.
• chunks has no reducer: only mapChunks writes it. indexedCount needs ReducedValue because N parallel upsertBatch workers write it. Match the reducer to the write pattern, not the field type.
• Delete is its own node so the skip path and the index path share exactly one delete. Delete-before-upsert becomes graph topology, not a convention inside a step.
• thread_id = organizationId:path replaces "list runs by resourceId, match input.path." A re-upload resumes the same thread, and concurrent re-uploads of one path serialize on it.
• retryPolicy retries the node, not the paid call inside it — the parse-cache check is what keeps a parseReducto retry from re-paying Reducto.
• MemorySaver dies with the process. Production wants PostgresSaver from @langchain/langgraph-checkpoint-postgres — the same Postgres the rest of the stack already uses. LangSmith replaces Studio for traces.
• Re-embed on model change is the same node functions re-wired (START → mapChunks, reading the cached parse). Extract nodes, not a forked graph.
• Only the workflow swaps. The agent, the search tool, and the Hono routes stay as written — the read path cannot tell which framework wrote the vectors.

Failure: a Send fan-out into a field with no reducer — the last batch's count wins. And organizationId in the invoke input is assembled by the server after the membership check; a client-supplied org is the same leak as a client-supplied namespace.

6. Retrieval

Retrieval is a tool the agent calls, not “embed every user message and prepend hits.” Isolation is applied in execute, not in the tool schema. Hybrid search and rerank live here — never on the upload line.

User question
  → agent (query, optional topK)
  → tool execute
      namespace = organizationId   (server)
      embedDense(query) + embedSparse(query)
      hybrid query (alpha-weighted) dotproduct ANN
      rerank top ~40 → keep ~8
      keep text and score ≥ 0.2
  → relevantContext + sources
  → agent

import { createTool } from "@mastra/core/tools"
import { z } from "zod"

function hybridScoreNorm(dense: number[], sparse: SparseVector, alpha: number) {
  return {
    dense: dense.map((v) => v * alpha),
    sparse: {
      indices: sparse.indices,
      values: sparse.values.map((v) => v * (1 - alpha)),
    },
  }
}

export const searchPineconeDocuments = createTool({
  id: "search-pinecone-documents",
  description:
    "Search this organization's uploaded documents for relevant excerpts.",
  inputSchema: z.object({
    query: z.string().describe("The search query"),
    topK: z.number().int().min(1).max(20).optional(),
  }),
  execute: async (inputData, { requestContext }) => {
    const organizationId = getWorkspaceOrganizationId(requestContext)
    const [denseQ] = await embedDense([inputData.query])
    const [sparseQ] = await embedSparse([inputData.query], "query")
    const { dense, sparse } = hybridScoreNorm(denseQ, sparseQ, 0.75)

    const index = pc.Index(PINECONE_INDEX_NAME)
    const results = await index.namespace(organizationId).query({
      vector: dense,
      sparseVector: sparse,
      topK: 40,
      includeMetadata: true,
    })

    const hits = results.matches
      .map((m) => ({
        id: m.id,
        score: m.score,
        text: String(m.metadata?.text ?? ""),
        path: String(m.metadata?.path ?? ""),
        filename: String(m.metadata?.filename ?? ""),
        page: Number(m.metadata?.page ?? 0),
      }))
      .filter((h) => h.text)

    const reranked = await pc.inference.rerank({
      model: "bge-reranker-v2-m3",
      query: inputData.query,
      documents: hits.map((h) => ({ id: h.id, text: h.text })),
      rankFields: ["text"],
      topN: inputData.topK ?? 8,
      returnDocuments: true,
      parameters: { truncate: "END" },
    })

    const sources = reranked.data
      .map((r) => {
        const hit = hits.find((h) => h.id === r.document.id)
        return hit ? { ...hit, score: r.score } : undefined
      })
      .filter((s): s is NonNullable<typeof s> => Boolean(s && s.score >= 0.2))

    return {
      relevantContext: sources.map((s) => s.text).join("\n\n"),
      sources,
    }
  },
})

Isolation and relevance are different APIs:

• v1 searches all of that organization’s docs. Path filters are a later mention feature. Do not implement tenancy as a metadata filter “just to be consistent with Postgres.”
• Alpha 0.75 leans semantic. Drop toward 0.25 when queries carry SKUs, error codes, or named entities that must match exactly. Evaluate on a labeled set from your own corpus.
• Fetch ~40 candidates, rerank down to 8. Rerank is a query-time cross-encoder (bge-reranker-v2-m3), not an ingest step.
• Reducto’s page and bbox on sources let the UI render citations that point at a region, not just a filename.
• 0.2 is a noise floor after rerank, not calibrated confidence. A successful lookup does not prove the answer is grounded; citations still have to be checked.
• Keep { relevantContext, sources } so a later cutover from Pinecone is a tool swap, not an agent rewrite. If embed returns nothing, return empty sources.
• Document RAG is “what is in the handbook.” Message-history semantic recall is a different index, different tenancy, different tool.

createVectorQueryTool is usable only if the server sets databaseConfig.pinecone.namespace (or a VectorStoreResolver) from the authenticated org before the agent runs, and the tool input schema never includes namespace. It also assumes a dense-only store. For hybrid, a custom tool is the only path.

• Default examples use a static namespace or requestContext.set("databaseConfig", …), which is easy to wire wrong and easy to expose to the model via PINECONE_PROMPT filter syntax.
• For org-scoped user data, derive namespace inside execute from the authenticated principal. A custom tool makes that the only path.
• Do not dump PINECONE_PROMPT into instructions and let the model invent $and / $in filters for tenancy.

Failure: putting namespace on the tool input schema, or letting the model set databaseConfig.

7. Delete, overwrite, and failure modes

Document identity is path. Re-upload replaces the object and replaces vectors for { namespace, path }. Delete must hit both stores, vectors first.

await pc.Index(PINECONE_INDEX_NAME)
  .namespace(organizationId)
  .deleteMany({ path })
await filesystem.deleteFile(path, { force: true })

• If you only delete the object, retrieval keeps serving ghosts.
• If you only delete vectors, the file remains and a later ingest can rebuild. Prefer an orphaned file over a leaked or stale excerpt.
• Skip-with-delete (empty parse, unsupported file, missing file) is the same rule: the current object is the source of truth, so the index must not keep a previous version of that path.

Failure: deleting only the object leaves searchable ghosts in Pinecone.

8. What to leave out until it hurts

Leave these out until hybrid on a flat docs/ prefix actually hurts.

• Separate dense and sparse indexes. Two upserts, two queries, client-side merge. Only worth it for sparse-only queries or Pinecone-integrated sparse embedding. One hybrid index is simpler and correct here.
• A second ingest workflow. Parse is a step, not a pipeline. Splitting parseReducto into its own workflow loses the single-run status and snapshot-retry semantics this design relies on. Extract a composable step if parse gets slow; do not fork the run.
• Reducto Extract. Parse builds the corpus. Extract fills schemas (invoice.total) as a separate product. Do not mix them into the ingest DAG.
• Multimodal image index. Reducto already OCRs images into text chunks. Add pixel embeddings only when users search “find this screenshot.”
• GraphRAG. Follows edges between chunks. For a wiki, maybe. For a flat docs/ prefix, no.
• Always-on injection. Embed every user message. You retrieve on greetings and burn tokens.

9. Invariants

1. Same dense and sparse models on write and read. Same dimension for dense.
2. Namespace is the organization. userId is the actor. Metadata filters are for relevance inside that tenant.
3. Overwrite and skip both start with deleteVectors(path). Pinecone will not do this for you.
4. The agent chooses when to search. The server chooses which org is visible and how signals mix (alpha).
5. Files remain source of truth. Vectors are derived. Reducto parse JSON is a cached intermediate — keep it so re-embed and ingest retries do not re-parse. Workflow runs are ingest status. resourceId is the org; path is matched on the run input.
6. Rerank is query-time. The upload path ends at upsert.
7. Upload and ingest are one workflow, four steps (loadFile → parseReducto → mapChunks → indexChunks). One runId is the status. Do not split parse into a second workflow.

That is the system: a four-step ingest DAG in a single workflow, a hybrid dotproduct index partitioned by organization, and a retriever whose tool schema cannot name another tenant.

Recap Q&A