如何搭建一套 RAG 系统

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

这套设计用 Mastra workflows、用 Reducto Parse 做 layout-aware chunking、OpenAI text-embedding-3-small（dense）加 pinecone-sparse-english-v0（sparse），以及一个按 namespace = organizationId 分区的 Pinecone serverless hybrid index。Identity 留在它本来就在的地方：Better Auth organizations 与 Hono request context，见 用 Hono、Better Auth、Drizzle 与 Postgres RLS 打造 Multi-Tenant 后端。userId 是 actor。organizationId 是图书馆。

Write path 到 upsert 为止。Hybrid search 与 rerank 活在 read path——绝不在 upload 这条线上。

1. System shape

Files 是 source of truth。Vectors 是派生 index。Ingest status 是该 org 与 path 的最新 workflow run，不是一张 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

• 从 filesystem prefix 列出 files。问「这个 file 能搜了吗？」就列出 resourceId = organizationId 的 workflow runs，再匹配 run 的 input path。
• 一张能跟 S3 与 Pinecone 漂移的 documents row，是你日后必须 reconcile 的第三份 source of truth。
• Mastra Workspace search（Workspace + PineconeVector + embedder）索引的是一个 agent sandbox。留给 agent 的 working set。这套 workflow 留给产品 corpus。
• Upload 那一轮仍把 file parts 内联带上。Indexing 是 async。Retrieval 是后续 turns，这样后面的 turns 搜 index，而不是再寄一遍 file。
• 把 Reducto 的 parse JSON（或它的 job id）留在 file 旁边。换 embedding models 时可以 re-embed 而不必 re-parse。

失败： 把 Workspace search 当成 HTTP upload API。它没有做过 membership check 的 requestContext，也没有 org-scoped library。

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

Mastra 的经验法则对上这个拆分：workflows 做已定义的多步 processes，agents 做会用 tools 的决策。Reducto 作为 ingest workflow 里的 parse step 很合适——它是 deterministic transform，不是决策，也不是第二条 workflow。

• Upload 与 chat 用同一套方式填 requestContext：session → 验证 route 里 organizationId 的 membership → requestContext.set("organizationId", organizationId) 与 requestContext.set("userId", userId)。
• 绝不要从 client header、tool argument，或 model 能影响的 databaseConfig 复制 org。
• Always-on injection 会 embed 每一句问候并烧掉 tokens。Tool-calling 把 retrieval 留在 default path 之外。

失败： 若 model 能选择搜谁的 index，你就有泄漏。

3. Embedding contract

Ingest 与 query 必须用同一个 dense model 与 dimension。跨两个 embedding spaces 做 cosine similarity 是噪声。Hybrid 加了第二份契约：write 与 query 用同一个 sparse model。

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 文档里的路径是带 provider/model id 的 ModelRouterEmbeddingModel，用于 dense。Sparse model 由 Pinecone 托管；通过 Pinecone SDK 调用。
• Hybrid index 把两种 vectors 存在同一条 record 上，要求 metric: "dotproduct"。本 repo 里 dense-only 的 indexes 用的是 cosine；切换意味着新 index 加一次完整 re-ingest。
• Dimension 在 createIndex 时冻结。以后换 dense model 意味着新 index 加一次完整 re-ingest。
• text-embedding-3-small 把 input 上限卡在 8191 tokens。先 chunk，再 embed。
• Sparse scores 无界；dense cosine scores 是 [-1, 1]。在 query 时用 alpha 权重（α·dense + (1-α)·sparse）把它们组合起来。自然语言问题从 α = 0.75 开始。不加权，sparse 部分会压过一切。
• Batch embeddings 与 upserts。Pinecone 实务上限大约是 1000 vectors 或 2 MB；1536-d 且 metadata 带 text 时，靠近 100 更稳。

失败： 一次 embedMany 一份无界 PDF，或者做 hybrid query 却不加 alpha，让 exact-match tokens 淹没语义。

4. Vector layout: one index, many namespaces

Pinecone 提供三种 tenant 策略。对 org-scoped library，只有 namespace = organizationId 可接受。

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 存在，不要把它 port 到 Pinecone。
• Namespace = organizationId： query、upsert 与 delete 总是带 namespace。Namespaces 在第一次 upsert 时被创建。Pinecone 把它们分开存，所以漏掉的 metadata filter 看不见另一个 tenant。Query cost 是目标 namespace 每 GB 1 RU。
• 省略 namespace 查的是 default namespace，不是每个 tenant。Isolation 成立的前提是你从不写入那个 default。永远传 namespace: organizationId。
• Index names：1–45 个字符，小写 alphanumeric 与 hyphens，开头和结尾是 alphanumeric。用 document-chunks，不要用 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
}

• Pinecone 上的 hybrid 是一个 dense index，在同一条 record 上带 sparse_values。Linkage 是隐式的：一次 upsert、一次 query、不需要 client-side merge。Separate dense/sparse indexes 只用于 sparse-only queries 或 Pinecone-integrated sparse embedding——这里用不上。
• Mastra 的 PineconeVector 是 dense 的 upsert/query wrapper。Hybrid 意味着用 Pinecone SDK（或包一层）才能设 sparse_values。把 workflow、agent 与 requestContext 留给 Mastra；别指望 @mastra/pinecone 来管 hybrid。
• 每个 vector 的 metadata 必须是 flat 的。Pinecone 不会 flatten nested objects。把 text 留在 metadata 里，否则 retrieval 只返回没有散文的 scores。
• Reducto 的 chunks 带 citation metadata。把 page、bbox 与 blockType 跟 path、chunkIndex 放在一起，答案才能指到某个区域，而不只是某个 file。
• organizationId 不属于 metadata 里的 isolation key。Namespace 就是 tenant。Metadata filters 是该 tenant 内部 的相关性（path，以及以后）。
• Document identity 是 org prefix 下的 path，通常是 docs/filename。同一 path 再 upload 会覆盖。Rename 是一份新 document。docs/ 下两个撞上同一 filename 的 files 是同一份 document。
• Flat metadata 是 { path, filename, chunkIndex, page, bbox, blockType, text, embedText, mediaType }。没有 nested objects。

失败： 省略 namespace 看起来像「搜所有东西」。它搜的是 default namespace。

5. Ingest workflow

File upload 与 ingest 属于 一条 workflow：pinecone-document-ingestion。「一条 workflow」不是「一步」。DAG 是四步。HTTP route 只写 object 并 start 这次 run。

Input 对齐 upload route，这样 Studio 与 HTTP 共享一份契约：{ path, filename, mediaType }。startAsync 是 fire-and-forget：它立刻返回 { runId }。Route 不等 status 或 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

留在一条 workflow，因为它是 一次逻辑操作。每个 file 两条 workflows 意味着要 join 两次 runs——documents table 问题变成 runs-join。Retries 与 snapshots 是 per-run 的：若 upsert 在 parse 成功之后 flake，同一条 run 从 snapshot retry indexChunks。它不会再开一条 pipeline，去重新推导第一条已经知道的东西。

真正要紧的边界是 step，不是 workflow：

• parseReducto 要 retry-safe，不该每次 retry 都再 parse。Reducto 按 parse 收费。把 parse JSON（或 job id）留在 step output——以及 file 旁边——这样 retry indexChunks 时，若该 path + content hash 已经有 parse，就直接跳到 mapChunks。
• 不要把 file bytes 传过 step output。Workflow snapshots 会把那份 data JSON-serialize。每一步按 path 从已经 org-rooted 的 filesystem 读。步骤之间只流 parse JSON（小、有结构）或它的引用。
• Delete-then-upsert 留在 indexChunks 里。把 delete 拆成自己的 workflow，会让它在没有 matching upsert 的情况下跑起来。

不属于 这条 workflow 的东西：

• Retrieval。Hybrid query 与 rerank 是后续 turns 上的 tool。Upload 这条线到 upsert 为止。
• Reducto Extract。Schema fill（invoice.total）有不同的 inputs、outputs 与 callers。那是以后的产品、以后的 workflow。
• 换 model 时的 re-embed。一条读缓存 parse JSON 的 maintenance workflow。共享 mapChunks / indexChunks steps——export 它们再 reuse——不要 fork 由 upload 触发的那次 run。

若某一步以后独立有用或运维上变重，抽出 step（Mastra steps 可组合），而不是第二条 ingest workflow。又慢又被 rate-limit 的 parse 变成 durable/suspendable step，让同一条 run 等着，不必占着 worker。Per-chunk enrichment 变成同一 DAG 里的第五步。

• 用 Reducto 的 Parse，不是 Extract。Parse 是 RAG ingest 这一步：pages、blocks、tables、figures、bounding boxes、citation-ready chunks。Extract 是 schema fill（invoice.total、contract.effectiveDate）——那是第二个产品，不是 corpus builder。
• Layout-aware parse 也可以是本地的 Docling。Workflow 仍把 parse 当 deterministic step。那份 converter 是 Docling。
• 让 Reducto 来 chunk。retrieval.chunking.chunk_mode: "variable" 以 1000 字符左右的 semantic boundaries 为目标。加上 embedding_optimized: true，tables 会得到一条给 vectors 用的 embed 字符串和一条给 display 用的 content 字符串。过滤 Header、Footer 与 Page Number blocks。
• 不要对 Reducto 的输出再跑 MDocument.chunk()。那是先花钱买 structure-aware chunks，再用 token splitter 砸烂。MDocument 在这里是 adapter：用 fromMarkdown（或 fromJSON）装 text + metadata，而不是 re-chunk。
• Images 走同一条 Parse 路径（OCR / vision pass）。除非以后要靠像素搜「找到这张截图」，否则不需要单独的 vision index。
• 把 Reducto parse JSON（或它的 job id）留在 file 旁边 以及 parseReducto 的 step output 里。换 models 时 re-embed 而不 re-parse。Retry indexChunks 时不必再为 Parse 付钱。
• 若 server 忘了设 org，getWorkspaceOrganizationId 会 throw——就是 upload route 放进 requestContext 的那份做过 membership check 的值。
• chunkIndex 是 array index。Reducto 的 chunks 带 blocks[]，里面有 bbox/page；自己把它们 map 成 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 的 upsert 没有 deleteFilter。pgvector 可以一次 call 替换一份 document；Pinecone 不能。若再 upload 产生更少的 chunks，剩下的 neighbors 会活下来。

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

• 在 namespace 里 先 按 path delete，包括 skip 时，然后若有东西可写再 upsert。返回 { status: "skipped", chunkCount: 0 } 或 { status: "indexed", chunkCount }。
• Deterministic ids（${path}:${chunkIndex}）会原地覆盖。它们仍不会移除已经不存在的 chunkIndex 值。Delete-first 才是正确的 overwrite。
• Embed 用 embedText 字段，存 text 字段。当 Reducto 给 tables 做 summary 时，两者会不同。
• Batch embed 与 upsert（1536-d 且 metadata 带 text 时约 100）。
• 把 DAG 接成 loadFile → parseReducto → mapChunks → indexChunks，然后 commit()。在 Mastra instance 上注册 store、workflow，以及带 search tool 的 agent。一条 workflow。四步。
• Delete 然后 upsert 不是 atomic。若 concurrent re-uploads 重要，在 upload handler 里按 { organizationId, path } serialize ingest。
• Serverless search 是 eventually consistent：与 startAsync 同一 turn 的 query 可能错过新 chunks。这也是 retrieval 留给后续 turns 的另一个原因。

失败： 同一 path 的两次 concurrent re-uploads 可能交错——delete A、delete B、upsert A、upsert B。

从 upload handler fire-and-forget，把 resourceId 设成 organization，不是 user。resourceId 把 run scope 到 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,
})

用 LangGraph 组合同一条 DAG

如果团队本来就在 LangChain/LangGraph 上，同一条 DAG 可以组合成一个 StateGraph。Nodes 取代 Mastra steps。Checkpointer 的 thread_id 取代 runId 成为 ingest status。以上每条 invariant 照样成立——org 来自 server、delete-first、embed 用 embedText、存 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 是每个 node 共享的 memory。只有一个 writer 的字段维持 plain zod。会被 fan-out 写入的字段需要 reducer——没有的话，最后一个 parallel batch 会覆盖其他全部。

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 返回 partial updates；graph 负责套用。loadFile 返回 Command——state update 与 routing 合并成一次 return——所以它没有对外的 static edge。

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

接线。Command 的目的地用 ends 声明。Fan-out 是一条返回 Send[] 的 conditional edge。要付费、会 flake 的 nodes 加上 retryPolicy。Checkpointer 让每个 thread_id 成为可 resume 的 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 })

Upload handler 把 createRun / startAsync 换成在名为 {organizationId}:{path} 的 thread 上 invoke：

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 用 Command 路由，所以它没有对外的 addEdge。若照样加一条，两个目标都会跑——skip path 会去 parse 一个不存在的 file。
• chunks 不需要 reducer：只有 mapChunks 写它。indexedCount 需要 ReducedValue，因为 N 个并行的 upsertBatch workers 都写它。Reducer 对应的是写入 pattern，不是字段类型。
• Delete 自成一个 node，让 skip path 与 index path 共用同一份 delete。Delete-before-upsert 变成 graph topology，不是 step 内的约定。
• thread_id = organizationId:path 取代「列出该 resourceId 的 runs，再匹配 input.path」。重新 upload 会 resume 同一条 thread；同一 path 的 concurrent re-uploads 在它上面序列化。
• retryPolicy retry 的是 node，不是里面那个付费 call——parse cache 检查才是让 parseReducto retry 不必再向 Reducto 付费的东西。
• MemorySaver 随 process 死掉。Production 用 @langchain/langgraph-checkpoint-postgres 的 PostgresSaver——就是整个 stack 本来在用的同一个 Postgres。LangSmith 取代 Studio 看 traces。
• 换 model 时的 re-embed 是同一批 node functions 重新接线（START → mapChunks，读缓存的 parse）。抽换 nodes，不是 fork 一条 graph。
• 只有 workflow 被换掉。Agent、search tool 与 Hono routes 照旧——read path 分不出 vectors 是哪个 framework 写的。

失败： Send fan-out 写进没有 reducer 的字段——最后一批的 count 获胜。而 invoke input 里的 organizationId 是 server 在 membership 检查之后组装的；由 client 提供 org 等同由 client 提供 namespace，是同一种泄漏。

6. Retrieval

Retrieval 是 agent 调用的 tool，不是「embed 每条 user message 再 prepend hits」。Isolation 套在 execute 里，不在 tool schema 里。Hybrid search 与 rerank 活在这里——绝不在 upload 这条线上。

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 与 relevance 是不同的 APIs：

• v1 搜的是 该 organization 的全部 docs。Path filters 是以后的 mention feature。不要把 tenancy 做成 metadata filter，「只为了跟 Postgres 保持一致」。
• Alpha 0.75 偏语义。当 query 带 SKUs、error codes，或必须 exact match 的 named entities 时，往 0.25 降。用你自己 corpus 里带标注的集合来 evaluate。
• 先取约 40 个 candidates，再 rerank 到 8。Rerank 是 query 时的 cross-encoder（bge-reranker-v2-m3），不是 ingest step。
• sources 上 Reducto 的 page 与 bbox 让 UI 能把 citations 渲染到某个区域，而不只是某个 filename。
• 0.2 是 rerank 之后的 noise floor，不是校准过的 confidence。一次成功的 lookup 不能证明答案是 grounded 的；citations 仍必须被检查。
• 保持 { relevantContext, sources }，这样以后从 Pinecone cutover 是换 tool，不是改写 agent。若 embed 什么都不返回，就返回 empty sources。
• Document RAG 是「handbook 里有什么」。Message-history semantic recall 是不同的 index、不同的 tenancy、不同的 tool。

createVectorQueryTool 只有在 server 于 agent 跑起来 之前，从已认证的 org 设置 databaseConfig.pinecone.namespace（或一个 VectorStoreResolver），并且 tool input schema 从不包含 namespace 时才可用。它也假设 store 是 dense-only。对 hybrid，custom tool 是唯一路径。

• 默认 examples 用 static namespace 或 requestContext.set("databaseConfig", …)，很容易接错，也很容易通过 PINECONE_PROMPT filter syntax 暴露给 model。
• 对 org-scoped user data，在 execute 里从已认证的 principal 派生 namespace。Custom tool 让那成为唯一路径。
• 不要把 PINECONE_PROMPT 倒进 instructions，让 model 为 tenancy 发明 $and / $in filters。

失败： 把 namespace 放进 tool input schema，或让 model 设置 databaseConfig。

7. Delete, overwrite, and failure modes

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

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

• 若你只删 object，retrieval 会继续端出 ghosts。
• 若你只删 vectors，file 仍在，以后的 ingest 可以重建。Prefer 一份 orphaned file，而不是泄漏或过期的 excerpt。
• Skip-with-delete（empty parse、unsupported file、missing file）是同一条规则：当前 object 是 source of truth，所以 index 不得保留该 path 的上一版。

失败： 只删 object，会在 Pinecone 里留下可搜的 ghosts。

8. What to leave out until it hurts

在扁平 docs/ prefix 上的 hybrid 真正痛之前，先把这些留在外面。

• Separate dense 与 sparse indexes。 两次 upserts、两次 queries、client-side merge。只有在需要 sparse-only queries 或 Pinecone-integrated sparse embedding 时才值得。一个 hybrid index 更简单，在这里也是对的。
• 第二条 ingest workflow。 Parse 是一步，不是一条 pipeline。把 parseReducto 拆成自己的 workflow，会丢掉这套设计依赖的 single-run status 与 snapshot-retry。Parse 变慢就抽出可组合的 step；不要 fork 这次 run。
• Reducto Extract。 Parse 建 corpus。Extract 填 schemas（invoice.total），是另一个产品。不要把它们混进 ingest DAG。
• Multimodal image index。 Reducto 已经把 images OCR 成 text chunks。只有当用户搜「找到这张截图」时，才加 pixel embeddings。
• GraphRAG。 跟随 chunks 之间的 edges。对 wiki，也许。对扁平 docs/ prefix，不必。
• Always-on injection。 Embed 每条 user message。你会在问候上 retrieval，并烧掉 tokens。

9. Invariants

1. Write 与 read 用同一个 dense 与 sparse models。Dense 用同一个 dimension。
2. Namespace 是 organization。userId 是 actor。Metadata filters 是该 tenant 内部的相关性。
3. Overwrite 与 skip 都以 deleteVectors(path) 开始。Pinecone 不会替你做这件事。
4. Agent 选择何时搜。Server 选择哪个 org 可见，以及 signals 如何混合（alpha）。
5. Files 仍是 source of truth。Vectors 是派生的。Reducto parse JSON 是缓存的中间产物——留着它，re-embed 与 ingest retries 就不必 re-parse。Workflow runs 是 ingest status。resourceId 是 org；path 在 run input 上匹配。
6. Rerank 是 query-time。Upload path 到 upsert 为止。
7. Upload 与 ingest 是一条 workflow、四步（loadFile → parseReducto → mapChunks → indexChunks）。一个 runId 就是 status。不要把 parse 拆成第二条 workflow。

这就是这套系统：一条 workflow 里的四步 ingest DAG、一份按 organization 分区的 hybrid dotproduct index，以及一份 tool schema 不能点名另一个 tenant 的 retriever。

Recap Q&A