How to build a RAG system

A working RAG stack is not “embed text and hope.” It 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, MDocument chunking, OpenAI text-embedding-3-small (1536-d), and a Pinecone serverless index partitioned by namespace = organizationId. Identity and membership stay where they already live: 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. Members of the same org share one corpus.

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.

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 failed snapshot is a failed ingest. You do not need a documents row to know either of those things, and a 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. That is the right primitive for a coding agent’s local files. It is the wrong primitive for an HTTP upload API, an org-scoped library, and a membership-checked requestContext. Keep Workspace for the agent’s working set. Keep this workflow for the product corpus.

Write
  POST /documents
  → S3 orgs/organizationId/docs/filename
  → workflow pinecone-document-ingestion
  → extract text
  → MDocument.chunk (token)
  → embedMany text-embedding-3-small
  → deleteVectors namespace + path
  → upsert namespace=organizationId
  → Pinecone index document-chunks

Read (later turn)
  chat → agent
  → tool search-pinecone-documents
  → embed query (same model)
  → pinecone.query namespace=organizationId
  → relevantContext + sources
  → agent

The upload turn still carries file parts inline on the chat message. Indexing is async. Retrieval is for subsequent turns against the org corpus. That split is deliberate: the user can ask about the PDF they just attached this turn, before Pinecone has the chunks. Later turns should not re-send the whole file. They search the index.

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.

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.

3. Embedding contract

Ingest and query must use the same model and dimension. Cosine similarity across two embedding spaces is noise. Mastra’s documented path is ModelRouterEmbeddingModel with a provider/model id.

import { ModelRouterEmbeddingModel } from "@mastra/core/llm"
import { embedMany } from "ai"

export const EMBEDDING_MODEL_ID = "openai/text-embedding-3-small"
export const EMBEDDING_DIMENSION = 1536

const embeddingModel = new ModelRouterEmbeddingModel(EMBEDDING_MODEL_ID)

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

Pinecone index metric is cosine. Dimension is frozen at createIndex. Changing the model later means a new index and a full re-ingest.

text-embedding-3-small caps input at 8191 tokens. Chunk first, then embed. Do not embedMany an unbounded PDF in one call. Batch embeddings and upserts (Pinecone’s practical ceiling is about 1000 vectors or 2 MB per request; at 1536-d with text in metadata, stay closer to 100).

4. Vector layout: one index, many namespaces

Pinecone offers three tenant strategies. Only one 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 also scales with the whole index. This is 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, not the whole index.

Omitting namespace does not search every tenant. It queries the default namespace. Isolation holds only if you never write that default namespace. Always pass namespace: organizationId.

Index names: 1–45 characters, lowercase alphanumeric and hyphens, start and end alphanumeric. No underscores, no dots. Use document-chunks, not mastra_document_chunks.

import { PineconeVector } from "@mastra/pinecone"

export const PINECONE_INDEX_NAME = "document-chunks"

export const pinecone = new PineconeVector({
  id: "pinecone",
  apiKey: process.env.PINECONE_API_KEY!,
  cloud: "aws",
  region: "us-east-1",
})

let indexReady: Promise<void> | undefined

export async function ensurePineconeIndex() {
  if (!indexReady) {
    indexReady = pinecone
      .createIndex({
        indexName: PINECONE_INDEX_NAME,
        dimension: EMBEDDING_DIMENSION,
        metric: "cosine",
      })
      .catch((error: unknown) => {
        indexReady = undefined
        throw error
      })
  }
  await indexReady
}

createIndex on @mastra/pinecone treats 409 as “already exists” and then checks dimension. The latch above avoids repeating that round trip.

Per-vector metadata must be flat. Pinecone does not support nested objects (it does not flatten them for you). Keep text in metadata or retrieval returns scores with no prose. Token chunks at maxSize: 512 stay under the 40 KB metadata cap.

type ChunkMetadata = {
  path: string
  filename: string
  chunkIndex: number
  text: string
  mediaType: string
}

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.

5. Ingest workflow

Input matches the upload route so Studio and HTTP share a contract:

import { z } from "zod"

const ingestionInputSchema = z.object({
  path: z.string(), // "docs/handbook.pdf"
  filename: z.string(),
  mediaType: z.string(),
})

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

Background run
  → exists(path) → readFile(path)
  → extractText → MDocument.chunk
  → deleteVectors namespace + filter path
  → upsert ids + vectors + metadata (if chunks)

startAsync is fire-and-forget. It returns { runId } immediately. The route does not wait for status or chunkCount. Poll or list later.

Load, extract, and chunk

Skip images. PDFs go through a text extractor. Everything else is UTF-8. Empty extract is skipped, not thrown. Skip still deletes existing vectors for that path: replacing a PDF with an image of the same name must not leave the old handbook searchable.

MDocument.chunk maxSize is character count unless you use strategy: "token". Use tokens so the size matches the embedding model.

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.

getWorkspaceOrganizationId throws if the server forgot to set the org. That is the same membership-checked value the upload route put on requestContext.

import { RequestContext } from "@mastra/core/request-context"
import { createStep } from "@mastra/core/workflows"
import { MDocument } from "@mastra/rag"
import { z } from "zod"

function getWorkspaceOrganizationId(requestContext: RequestContext): string {
  const organizationId = requestContext.get("organizationId")
  if (typeof organizationId !== "string" || !organizationId) {
    throw new Error("organizationId missing from requestContext")
  }
  return organizationId
}

const loadOutputSchema = ingestionInputSchema.extend({
  skipped: z.boolean().optional(),
  skipReason: z.string().optional(),
})

const loadFile = createStep({
  id: "load-file",
  inputSchema: ingestionInputSchema,
  outputSchema: loadOutputSchema,
  execute: async ({ inputData }) => {
    const exists = await filesystem.exists(inputData.path)
    if (!exists) {
      return {
        ...inputData,
        skipped: true,
        skipReason: "File not found",
      }
    }
    return inputData
  },
})

Extract next. Images, empty text, and missing files all become skipped. The index step still deletes for that path.

const extractOutputSchema = loadOutputSchema.extend({
  text: z.string().optional(),
})

const extractTextStep = createStep({
  id: "extract-text",
  inputSchema: loadOutputSchema,
  outputSchema: extractOutputSchema,
  execute: async ({ inputData }) => {
    if (inputData.skipped) {
      return inputData
    }

    if (inputData.mediaType.startsWith("image/")) {
      return {
        ...inputData,
        skipped: true,
        skipReason: "Images are not indexed",
      }
    }

    const bytes = await filesystem.readFile(inputData.path)
    const text =
      inputData.mediaType === "application/pdf"
        ? await extractPdfText(bytes)
        : new TextDecoder("utf-8").decode(bytes)

    if (!text.trim()) {
      return {
        ...inputData,
        skipped: true,
        skipReason: "Empty extract",
      }
    }

    return { ...inputData, text }
  },
})

Chunk last. chunkIndex is the array index. MDocument chunks are { text, metadata } — they do not carry chunkIndex for you.

const chunkOutputSchema = ingestionInputSchema.extend({
  chunks: z.array(
    z.object({
      text: z.string(),
      chunkIndex: z.number(),
    }),
  ),
  skipped: z.boolean().optional(),
  skipReason: z.string().optional(),
})

const chunkDocument = createStep({
  id: "chunk-document",
  inputSchema: extractOutputSchema,
  outputSchema: chunkOutputSchema,
  execute: async ({ inputData }) => {
    if (inputData.skipped || !inputData.text) {
      return {
        path: inputData.path,
        filename: inputData.filename,
        mediaType: inputData.mediaType,
        chunks: [],
        skipped: true,
        skipReason: inputData.skipReason ?? "No text to chunk",
      }
    }

    const doc =
      inputData.mediaType === "text/markdown"
        ? MDocument.fromMarkdown(inputData.text)
        : inputData.mediaType === "text/html"
          ? MDocument.fromHTML(inputData.text)
          : inputData.mediaType === "application/json"
            ? MDocument.fromJSON(inputData.text)
            : MDocument.fromText(inputData.text)

    const nodes = await doc.chunk({
      strategy: "token",
      maxSize: 512,
      overlap: 50,
      encodingName: "cl100k_base",
    })

    return {
      path: inputData.path,
      filename: inputData.filename,
      mediaType: inputData.mediaType,
      chunks: nodes.map((node, chunkIndex) => ({
        text: node.text,
        chunkIndex,
      })),
    }
  },
})

Use fromMarkdown / fromHTML / fromJSON when the media type matches so separators follow structure instead of raw characters.

Index step: Pinecone has no deleteFilter on upsert

pgvector can replace a document in one call:

await pgVector.upsert({
  indexName,
  vectors: embeddings,
  deleteFilter: { organizationId, path },
  metadata,
})

Pinecone cannot. If a re-upload produces fewer chunks, leftover neighbors survive. Delete by path in the namespace first, including on skip, then upsert if there is anything to write.

Deterministic ids (${path}:${chunkIndex}) overwrite in place. They still do not remove chunkIndex values that no longer exist. Delete-first is the correct overwrite. Pass ids so the snippet matches the prose.

const UPSERT_BATCH_SIZE = 100

const indexOutputSchema = z.object({
  status: z.enum(["indexed", "skipped"]),
  path: z.string(),
  chunkCount: z.number(),
  reason: z.string().optional(),
})

const indexChunks = createStep({
  id: "index-chunks",
  inputSchema: chunkOutputSchema,
  outputSchema: indexOutputSchema,
  execute: async ({ inputData, requestContext }) => {
    const organizationId = getWorkspaceOrganizationId(requestContext)

    await ensurePineconeIndex()

    await pinecone.deleteVectors({
      indexName: PINECONE_INDEX_NAME,
      namespace: organizationId,
      filter: { path: inputData.path },
    })

    if (inputData.skipped || inputData.chunks.length === 0) {
      return {
        status: "skipped" as const,
        path: inputData.path,
        chunkCount: 0,
        reason: inputData.skipReason ?? "No chunks to index",
      }
    }

    await upsertChunkBatches(organizationId, inputData)

    return {
      status: "indexed" as const,
      path: inputData.path,
      chunkCount: inputData.chunks.length,
    }
  },
})

Batch embed and upsert. Pinecone’s practical ceiling is about 1000 vectors or 2 MB; at 1536-d with text in metadata, 100 is safer.

async function upsertChunkBatches(
  organizationId: string,
  inputData: z.infer<typeof chunkOutputSchema>,
) {
  for (let i = 0; i < inputData.chunks.length; i += UPSERT_BATCH_SIZE) {
    const batch = inputData.chunks.slice(i, i + UPSERT_BATCH_SIZE)
    const embeddings = await embedTexts(batch.map((chunk) => chunk.text))

    await pinecone.upsert({
      indexName: PINECONE_INDEX_NAME,
      namespace: organizationId,
      ids: batch.map((chunk) => `${inputData.path}:${chunk.chunkIndex}`),
      vectors: embeddings,
      metadata: batch.map((chunk) => ({
        path: inputData.path,
        filename: inputData.filename,
        chunkIndex: chunk.chunkIndex,
        text: chunk.text,
        mediaType: inputData.mediaType,
      })),
    })
  }
}

Delete then upsert is not atomic. Two concurrent re-uploads of the same path can interleave (delete A, delete B, upsert A, upsert B — or worse, upsert A after B’s delete). Serialize ingest per { organizationId, path } in the upload handler if that matters. Serverless search is also eventually consistent: a query on the same turn as startAsync can miss the new chunks. That is another reason retrieval is for later turns, and the current turn uses inline file parts.

Wiring the DAG

import { createWorkflow } from "@mastra/core/workflows"

export const pineconeDocumentIngestion = createWorkflow({
  id: "pinecone-document-ingestion",
  inputSchema: ingestionInputSchema,
  outputSchema: indexOutputSchema,
})
  .then(loadFile)
  .then(extractTextStep)
  .then(chunkDocument)
  .then(indexChunks)

pineconeDocumentIngestion.commit()

Register the store, the workflow, and the agent (with the search tool) on the Mastra instance:

import { Mastra } from "@mastra/core/mastra"

export const mastra = new Mastra({
  vectors: { pinecone },
  workflows: { pineconeDocumentIngestion },
  agents: { workspaceAgent },
})

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,
})

6. Retrieval

Retrieval is not “embed every user message and prepend hits.” It is a tool the agent calls. Isolation is applied in execute, not in the tool schema.

User question
  → agent (query, optional topK)
  → tool execute
      namespace = organizationId   (server)
      embedTexts(query)
      pinecone.query cosine ANN
      keep text and score ≥ 0.2
  → relevantContext + sources
  → agent

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

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(),
  }),
  outputSchema: z.object({
    relevantContext: z.string(),
    sources: z.array(
      z.object({
        path: z.string(),
        filename: z.string(),
        score: z.number(),
        text: z.string(),
      }),
    ),
  }),
  execute: async (inputData, { requestContext }) => {
    return searchOrgDocuments(inputData, requestContext)
  },
})

execute takes organizationId from requestContext. The tool schema cannot name another tenant.

async function searchOrgDocuments(
  inputData: { query: string; topK?: number },
  requestContext: RequestContext,
) {
  const organizationId = getWorkspaceOrganizationId(requestContext)
  const topK = inputData.topK ?? 8

  await ensurePineconeIndex()

  const [queryVector] = await embedTexts([inputData.query])
  if (!queryVector) {
    return { relevantContext: "", sources: [] }
  }

  const results = await pinecone.query({
    indexName: PINECONE_INDEX_NAME,
    queryVector,
    namespace: organizationId,
    topK,
    includeVector: false,
  })

  const minScore = 0.2
  const sources = results.flatMap((result) => {
    const text =
      typeof result.metadata?.text === "string" ? result.metadata.text : ""
    if (!text || result.score < minScore) return []
    return [
      {
        path: String(result.metadata?.path ?? ""),
        filename: String(result.metadata?.filename ?? ""),
        score: result.score,
        text,
      },
    ]
  })

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

Isolation vs 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.”

0.2 is a noise floor, not calibrated confidence. Unrelated pairs for this model often sit near there; do not treat it as “typical range.” topK: 8 is a context budget: 8 × ~512-token overlapping chunks. Pinecone search is ANN; you can miss a chunk that would rank 9th in exact kNN. That is acceptable for an org library.

Why not createVectorQueryTool

Mastra can bind Pinecone like this:

import { createVectorQueryTool } from "@mastra/rag"
import { ModelRouterEmbeddingModel } from "@mastra/core/llm"

const pineconeQueryTool = createVectorQueryTool({
  vectorStoreName: "pinecone",
  indexName: "document-chunks",
  model: new ModelRouterEmbeddingModel("openai/text-embedding-3-small"),
  databaseConfig: {
    pinecone: { namespace: "production" },
  },
})

That helper is usable 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. The 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 and do not put it on the tool input schema. A custom tool makes that the only path.

Keep the output shape { relevantContext, sources } so a later cutover from pgvector is a tool swap, not an agent rewrite. Do not dump PINECONE_PROMPT into instructions and let the model invent $and / $in filters for tenancy.

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 pinecone.deleteVectors({
  indexName: PINECONE_INDEX_NAME,
  namespace: organizationId,
  filter: { 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 extract, images, 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.

8. What to leave out until it hurts

Hybrid sparse + dense. Requires metric: "dotproduct" and sparse vectors on upsert and query. Helps exact tokens (SKUs, error codes). Dense cosine is enough for prose. A later hybrid index is a new index, not a setting you flip on document-chunks.

Re-rank. Fetch topK=30, then rerankWithScorer down to 8. Helps when ANN returns “same topic, wrong paragraph.” Adds a cross-encoder (or agent scorer) hop.

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. Tool-calling keeps retrieval off the default path.

9. Invariants

1. Same embedding model and dimension on write and read.
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.
5. Files remain source of truth. Vectors are derived. Workflow runs are ingest status. resourceId is the org; path is matched on the run input.

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