如何搭建一套 RAG 系统

一套能跑的 RAG stack 不是「embed 文本然后祈祷」。它是两条共享 embedding space 的 pipelines：一条deterministic ingest workflow 负责写 vectors，一条tool-called retriever 负责读。难的是 tenant boundary。若 model 能选择搜谁的 index，你就没有 RAG。你有的是泄漏。

这套设计用 Mastra workflows、MDocument chunking、OpenAI text-embedding-3-small（1536-d），以及按 namespace = organizationId 分区的 Pinecone serverless index。Identity 与 membership 留在它们本来就在的地方：Better Auth organizations 与 Hono request context，见用 Hono、Better Auth、Drizzle 与 Postgres RLS 打造 Multi-Tenant 后端。userId 是 actor。organizationId 是图书馆。同一 org 的 members 共享一份 corpus。

1. System shape

Files 是 source of truth。Vectors 是派生 index。Ingest status 是该 org 与 path 的最新 workflow run，不是一张 documents table。

从 filesystem prefix 列出 files。问「这个 file 能搜了吗？」就列出 resourceId = organizationId 的 workflow runs，再匹配 run 的 input path。失败的 snapshot 就是失败的 ingest。这两件事都不需要 documents row 才能知道；而一张能跟 S3 与 Pinecone 漂移的 row，是你日后必须 reconcile 的第三份 source of truth。

Mastra Workspace search（Workspace + PineconeVector + embedder）索引的是一个 agent sandbox。这对 coding agent 的本地 files 是正确的 primitive。对 HTTP upload API、org-scoped library、以及做过 membership check 的 requestContext，它是错误的 primitive。Workspace 留给 agent 的 working set。这套 workflow 留给产品 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

Upload 那一轮仍把 file parts 内联在 chat message 上。Indexing 是 async。Retrieval 是后续 turns 对 org corpus 的搜索。这个拆分是故意的：用户可以这一轮就问刚附上的 PDF，哪怕 Pinecone 还没有 chunks。后面的 turns 不该再把整个 file 寄一遍。它们搜 index。

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

关注点	Primitive	原因
Load → extract → chunk → embed → upsert	`createWorkflow`	固定 DAG、retries、run snapshots、skip vs fail
「User asked about the handbook」	`createTool` on an agent	Open-ended；「hi」时跳过 retrieval
Tenant key	`requestContext`, never tool args	LLM 绝不能自己挑 namespace

Mastra 的经验法则对上这个拆分：workflows 做已定义的多步 processes，agents 做会用 tools 的决策。

Upload 与 chat 用同一套方式填 requestContext：session → 验证 route 里 organizationId 的 membership → requestContext.set("organizationId", organizationId) 与 requestContext.set("userId", userId)。绝不要从 client header、tool argument，或 model 能影响的 databaseConfig 复制 org。

3. Embedding contract

Ingest 与 query 必须用同一个 model 与 dimension。跨两个 embedding spaces 做 cosine similarity 是噪声。Mastra 文档里的路径是带 provider/model id 的 ModelRouterEmbeddingModel。

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 是 cosine。Dimension 在 createIndex 时冻结。以后换 model 意味着新 index 加一次完整 re-ingest。

text-embedding-3-small 把 input 上限卡在 8191 tokens。先 chunk，再 embed。不要一次 embedMany 一份无界 PDF。Batch embeddings 与 upserts（Pinecone 实务上限大约是每次 request 1000 vectors 或 2 MB；1536-d 且 metadata 带 text 时，靠近 100 更稳）。

4. Vector layout: one index, many namespaces

Pinecone 提供三种 tenant 策略。对 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 是真的；createIndex 很慢；index-count limits 会弄死你。
Metadata filter: { organizationId }： 漏掉一次 filter 就泄漏整份 corpus。Query cost 还随整个 index 缩放。这是 pgvector 模式。有 namespaces 就不要把它搬到 Pinecone。
Namespace = organizationId： query、upsert 与 delete 始终带 namespace。Namespaces 在第一次 upsert 时创建。Pinecone 分开存放，所以漏掉 metadata filter 也看不见另一个 tenant。Query cost 是目标 namespace 每 GB 1 RU，不是整个 index。

省略 namespace 不会搜遍每个 tenant。它查的是 default namespace。Isolation 成立的前提是你从不往那个 default namespace 写。始终传 namespace: organizationId。

Index names：1–45 个字符，小写字母数字与连字符，开头结尾必须是字母数字。不要下划线，不要点。用 document-chunks，不要用 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
}

@mastra/pinecone 上的 createIndex 把 409 当成「already exists」，然后检查 dimension。上面的 latch 避免重复那次 round trip。

每个 vector 的 metadata 必须是flat。Pinecone 不支持 nested objects（它不会帮你 flatten）。把 text 放进 metadata，否则 retrieval 只返回没有散文的 scores。maxSize: 512 的 token chunks 能待在 40 KB metadata 上限内。

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

organizationId 不该作为 isolation key 放进 metadata。Namespace 就是 tenant。Metadata filters 用来在该 tenant 内部做 relevance（path，以后）。

Document identity 是 org prefix 下的 path，通常是 docs/filename。同一 path 再 upload 会覆盖。Rename 是一份新 document。docs/ 下两个 filename 撞车的 files 是同一份 document。

5. Ingest workflow

Input 与 upload route 对齐，让 Studio 与 HTTP 共享同一份 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 是 fire-and-forget。它立刻返回 { runId }。Route 不等 status 或 chunkCount。之后再 poll 或 list。

Load, extract, and chunk

跳过 images。PDFs 走 text extractor。其余都是 UTF-8。空的 extract 是 skipped，不是 throw。Skip 仍然删除该 path 上既有的 vectors：用同名 image 替换 PDF，绝不能让旧 handbook 还能搜到。

MDocument.chunk 的 maxSize 是字符数，除非你用 strategy: "token"。用 tokens，让 size 对上 embedding model。

不要把 file bytes 经 step output 传递。Workflow snapshots 会 JSON-serialize 那些数据。每一步都从已经 org-rooted 的 filesystem 按 path 读取。

若 server 忘了设 org，getWorkspaceOrganizationId 会 throw。那就是 upload route 放到 requestContext 上、已经做过 membership check 的同一个值。

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。Images、空 text、缺失 files 都变成 skipped。Index step 仍会为该 path 做 delete。

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。chunkIndex 是 array index。MDocument chunks 是 { text, metadata } — 它们不会帮你带上 chunkIndex。

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

Media type 对得上时用 fromMarkdown / fromHTML / fromJSON，让 separators 跟着结构走，而不是跟着原始字符。

Index step: Pinecone has no `deleteFilter` on upsert

pgvector 可以一次 call 替换一份 document：

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

Pinecone 不行。若 re-upload 产生更少 chunks，剩下的 neighbors 会活下来。在 namespace 里先按 path delete，包括 skip 时，然后有东西可写再 upsert。

Deterministic ids（${path}:${chunkIndex}）会原地覆盖。它们仍不会删掉已经不存在的 chunkIndex 值。先 delete 才是正确的 overwrite。传入 ids，让 snippet 对上这段散文。

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 与 upsert。Pinecone 实务上限大约是 1000 vectors 或 2 MB；1536-d 且 metadata 带 text 时，100 更稳。

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 再 upsert 不是 atomic。同一 path 的两次并发 re-uploads 会交错（delete A、delete B、upsert A、upsert B — 或更糟，B 的 delete 之后再 upsert A）。若这很重要，在 upload handler 里按 { organizationId, path } serialize ingest。Serverless search 也是 eventually consistent：与 startAsync 同一轮的 query 可能错过新 chunks。这是 retrieval 留给后续 turns、当前 turn 用 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()

在 Mastra instance 上注册 store、workflow，以及带 search tool 的 agent：

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

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

从 upload handler fire-and-forget，把 resourceId 设成 organization，不是 user。resourceId 把 run 限定在 tenant。它不按 path join。要找某个 file 的最新 ingest，列出该 resourceId 的 runs，再匹配 input.path。

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

6. Retrieval

Retrieval 不是「把每条 user message 都 embed，再把 hits 前置」。它是 agent 调用的 tool。Isolation 加在 execute 里，不加在 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 从 requestContext 取 organizationId。Tool schema 不能点名另一个 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

旋钮	Pinecone mechanism	由谁控制
Tenant	`namespace: organizationId`	Server，始终
Which file	`filter: { path }`	Optional；v1 省略
Neighbor count	`topK`	Agent，上限 1–20
Weak matches	`minScore` 0.2	Server constant

v1 搜索该 organization 的全部 docs。Path filters 是以后的 mention 功能。不要「为了跟 Postgres 一致」把 tenancy 做成 metadata filter。

0.2 是 noise floor，不是校准过的 confidence。对这个 model，不相关的 pairs 常常就停在附近；不要把它当成「典型范围」。topK: 8 是一份 context budget：8 × ~512-token 的 overlapping chunks。Pinecone search 是 ANN；你可能漏掉在 exact kNN 里排第 9 的 chunk。对 org library 这可以接受。

Why not `createVectorQueryTool`

Mastra 可以这样绑定 Pinecone：

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

若 server 在 agent 跑之前，从已认证的 org 设定 databaseConfig.pinecone.namespace（或一个 VectorStoreResolver），且 tool input schema 从不包含 namespace，这个 helper 能用。默认 examples 用静态 namespace 或 requestContext.set("databaseConfig", …)，很容易接错，也很容易经由 PINECONE_PROMPT filter syntax 暴露给 model。

对 org-scoped 的用户数据，在 execute 里从已认证的 principal 推导 namespace，并且不要把它放进 tool input schema。Custom tool 让那成为唯一路径。

保持 output shape { relevantContext, sources }，这样以后从 pgvector cutover 是换 tool，不是重写 agent。不要把 PINECONE_PROMPT 倒进 instructions，让 model 自己发明 $and / $in filters 来做 tenancy。

7. Delete, overwrite, and failure modes

Document identity 是 path。Re-upload 替换 object，并替换 { namespace, path } 的 vectors。Delete 必须打到两边 stores，先打 vectors：

await pinecone.deleteVectors({
  indexName: PINECONE_INDEX_NAME,
  namespace: organizationId,
  filter: { path },
})
await filesystem.deleteFile(path, { force: true })

只删 object，retrieval 会继续端出幽灵。只删 vectors，file 还在，之后的 ingest 可以重建。宁可留下孤立 file，也不要泄漏或过期 excerpt。

Skip-with-delete（空 extract、images、缺失 file）是同一条规则：当前 object 是 source of truth，所以 index 不能留下该 path 的上一版。

8. What to leave out until it hurts

Hybrid sparse + dense。 需要 metric: "dotproduct"，以及 upsert 和 query 都带 sparse vectors。对精确 tokens（SKUs、error codes）有帮助。散文用 dense cosine 就够。以后的 hybrid index 是新 index，不是你在 document-chunks 上拨的一个 setting。

Re-rank。 先取 topK=30，再用 rerankWithScorer 收到 8。ANN 返回「同一话题、错误段落」时有帮助。多一跳 cross-encoder（或 agent scorer）。

GraphRAG。 跟着 chunks 之间的 edges 走。Wiki 也许可以。扁平的 docs/ prefix，不要。

Always-on injection。 每条 user message 都 embed。打招呼也会 retrieve，烧掉 tokens。Tool-calling 让 retrieval 离开默认路径。

9. Invariants

Write 与 read 用同一个 embedding model 与 dimension。
Namespace 是 organization。userId 是 actor。Metadata filters 用于该 tenant 内部的 relevance。
Overwrite 与 skip 都以 deleteVectors(path) 开头。Pinecone 不会替你做这件事。
Agent 选择何时搜索。Server 选择哪个 org 可见。
Files 仍是 source of truth。Vectors 是派生的。Workflow runs 是 ingest status。resourceId 是 org；path 在 run input 上匹配。

这就是整套系统：四步 ingest DAG、按 organization 分区的 cosine index，以及一份 tool schema 无法点名另一个 tenant 的 retriever。

阅读下一篇笔记

人生之诗