如何搭建一套 RAG 系統

能用的 RAG stack 是兩條共用 embedding space 的 pipelines：一條寫入 vectors 的 deterministic ingest workflow，以及一條讀取它們的 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. 系統形狀

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，會變成你必須對帳的第三個 source of truth。
• Mastra Workspace search（Workspace + PineconeVector + embedder）為一個 agent sandbox 建立 index。把 Workspace 留給 agent 的 working set。把這個 workflow 留給產品 corpus。
• Upload 那個 turn 仍然把 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 檢查的 requestContext，也沒有 org-scoped library。

2. 為什麼 ingest 用 workflow、retrieve 用 tool

Mastra 的經驗法則對上這個切割：workflows 處理已定義的 multi-step 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 配置：一個 index、多個 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 檢查的值。
• 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 與失敗模式

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. 在痛之前先別做的東西

在扁平 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