编程 Mastra 深度拆解：当 TypeScript 成为 AI Agent 的一等公民——从工具链、工作流引擎到生产级记忆与 RAG 的工程全貌（2026）

2026-07-18 04:48:17 +0800 CST views 4

Mastra 深度拆解：当 TypeScript 成为 AI Agent 的一等公民——从工具链、工作流引擎到生产级记忆与 RAG 的工程全貌（2026）

2026 年的 GitHub Trending 榜单上，前几名几乎被 AI Agent 项目屠版。但如果你去翻这些项目的技术栈，会发现一个尴尬的事实：绝大多数 Agent 框架是 Python 写的，而真正把 AI 功能跑进生产、服务真实用户的，却是一堆 Next.js、Node、React 的 TypeScript 应用。Mastra 想干的事很简单——让 TypeScript 开发者不用再为了一个 Agent 去维护一套 Python 微服务，在自家已经熟得不能再熟的 JS/TS 栈里，从原型一路写到生产。

一、背景：为什么 2026 年我们还需要一个 TypeScript 原生的 Agent 框架

1.1 一个被忽视的裂缝：框架是 Python 的，产品是 TypeScript 的

过去两年，Agent 生态的叙事几乎被 Python 垄断。LangChain、CrewAI、AutoGen、LangGraph……这些名字你一定听过。它们强大、社区庞大、论文和示例一抓一大把。但当你是一个前端/全栈团队，主栈是 Next.js + Node + Postgres，要做的功能只是「在 SaaS 后台里塞一个能调用内部 API 的智能助手」时，现实会给你两记闷棍：

第一记，是运行时的割裂。Python Agent 服务要单独部署、单独扩缩容、单独监控；前端拿不到类型、拿不到流式、拿不到统一的错误模型。你不得不在两个语言、两套部署、两套可观测体系之间反复横跳。

第二记，是原型到生产的鸿沟。在 Notebook 里跑通一个 ReAct 循环很容易，但生产需要的是：对话状态如何持久化？长上下文怎么不把 token 烧穿？失败怎么重试？人工审核怎么插入？流式怎么对接前端组件？这些恰恰是 Python 框架「示例很美、落地很痛」的地方。

Mastra 的核心判断是：对于一个以 Web 技术栈为主的公司，最贵的不是「会不会写 Agent」，而是「为了写 Agent 额外引入了多少非母语的复杂度」。如果能让 Agent 直接长在你的 TypeScript 代码里，和 React 组件、Next.js Route Handler、Node 服务共享同一套类型、同一套部署、同一套运行时，那么从 demo 到上线的成本会指数级下降。

1.2 Mastra 的赌注：用 Vercel AI SDK 做模型路由，用 TypeScript 做一等公民

Mastra 不是又一个从零造轮子的 LLM 封装。它的底层直接复用 Vercel AI SDK 作为模型路由层，因此天然支持 OpenAI、Anthropic、Google Gemini 等几十个模型供应商。但它往上盖了一整层面向生产的原语：Agent、Tool、Workflow、Memory、RAG、Voice、Channel、Deploy。

它的几个关键设计取向，决定了它和 Python 框架的气质完全不同：

模型用字符串表达。model: 'openai/gpt-5.5'，而不是 import 一个 provider 对象。这意味着你可以在不引入任何 provider 依赖的情况下切换模型，Mastra 自动按 provider 去读对应的环境变量（如 OPENAI_API_KEY）。对配置化、灰度、A/B 测试极其友好。
工具必须是显式声明的。createTool() 强制你写 id、description、inputSchema（Zod）、execute()。纯对象式的 tool 定义会被静默忽略不执行——这个「反人体工程学」的设计，恰恰是为了逼你在编译期就把契约定清楚。
Agent 与 Workflow 分工。Agent 负责「开放性问题、让模型自己决定步骤」；Workflow 负责「流程已知、我要精确控制每一步」。这是 Mastra 架构上最重要的一根脊梁。
TypeScript 原生运行。Node.js 22.18.0 及以上已经能直接跑 .ts 文件，Mastra 顺势把 import './foo.ts' 这种带扩展名的写法变成一等公民，开发体验接近于「写脚本即服务」。

1.3 它适合谁，不适合谁

先泼一盆冷水，避免你冲动选型：

适合：已经是 TS/JS 栈的团队；要做「嵌在自家产品里的 AI 功能」而非「通用大模型平台」；重视从原型到上线的工程闭环；希望 Agent 逻辑和前端共享类型。
不适合：团队主力是 Python 数据/算法工程师；需求是纯离线批处理式的研究管线；你对 LangGraph 那种「状态机画得很爽」的图编排已经很熟且不打算换栈。

换句话说，Mastra 是给「Web 工程师想认真做 Agent」的人准备的，而不是给「Agent 研究者想顺手写个 Web」的人准备的。这个定位差，决定了后面所有的设计取舍。

二、核心概念全景：Mastra 给你的一套生产原语

Mastra 的能力可以拆成一条清晰的纵轴：模型路由 → Agent/Tool → Workflow → Memory → RAG → 评测/语音/渠道 → 部署。我们自上而下逐个拆开。

2.1 Agent 与 Tool：最小的智能单元

一个 Agent 就是「一个 LLM + 一份人设指令（instructions）+ 一组工具（tools）+ 一份记忆（memory）」。它接收开放性问题，自己决定调哪些工具、迭代几轮，直到吐出最终答案。

工具（Tool）是 Agent 伸手够到真实世界的触角。Mastra 里定义一个工具的唯一正确姿势是 createTool()：

// src/mastra/tools/weather-tool.ts
import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const weatherTool = createTool({
  id: 'get-weather',
  description: 'Get current weather for a location',
  inputSchema: z.object({
    location: z.string().describe('City name'),
  }),
  outputSchema: z.object({
    location: z.string(),
    temperatureCelsius: z.number(),
    conditions: z.string(),
  }),
  execute: async ({ inputData }) => {
    // 这里可以查真实天气 API、调内部服务、读数据库
    return {
      location: inputData.location,
      temperatureCelsius: 21,
      conditions: 'sunny',
    }
  },
})

注意 execute 的签名：async ({ inputData }) => {...}。第一个参数是已经按 inputSchema 校验过的输入；它还有第二个可选参数——一个执行上下文对象，里面带着 requestContext、tracingContext、abortSignal 等元数据。这意味着你的工具可以感知「这次调用来自哪个请求」「要不要被取消」，对做超时控制和链路追踪很有用。

然后把它挂到 Agent 上：

// src/mastra/agents/weather-agent.ts
import { Agent } from '@mastra/core/agent'
import { weatherTool } from '../tools/weather-tool.ts'

export const weatherAgent = new Agent({
  id: 'weather-agent',
  name: 'Weather Agent',
  instructions: `
      You are a helpful weather assistant that provides accurate weather information.
      When responding, include humidity, wind, and precipitation.
      Use the weatherTool to fetch current weather data.
  `,
  // 字符串格式的模型路由，无需 import provider
  model: 'openai/gpt-5.5',
  tools: { weatherTool },
})

最后在入口注册并运行。Node 22.18+ 直接跑 TS：

// src/mastra/index.ts
import { Mastra } from '@mastra/core'
import { weatherAgent } from './agents/weather-agent.ts'

export const mastra = new Mastra({
  agents: { weatherAgent },
})

// run.mjs
import { mastra } from './src/mastra/index.ts'

const agent = mastra.getAgentById('weather-agent')
const response = await agent.generate('Weather in SF')
console.log(response.text)

这里有几个工程上很关键的细节，新手极容易踩坑：

模型 ID 的字符串格式是 provider/model，不是 provider:model，也不是传一个 provider 对象。写成 openai:gpt-5.5 或传 openai(model) 都会不工作。Mastra 当前路由表里能直接用的模型 ID 包括：openai/gpt-5.5、openai/gpt-5-mini、anthropic/claude-sonnet-4-6、anthropic/claude-opus-4-7、anthropic/claude-haiku-4-5、google/gemini-2.5-flash 等。
工具必须用 createTool() 声明。execute 收不到输入、或者你随手丢个 { name, fn } 的对象上去，Mastra 会静默地不去执行它——不会报错，但 Agent 永远「忘记」这个工具。这是最阴间的 bug 来源，务必用 createTool。
本地文件 import 要带扩展名（../tools/weather-tool.ts）。因为依赖 Node 原生 TS 运行，不带扩展名会解析失败。

2.2 结构化输出：别再用正则去抠模型的文本了

生产里最该戒掉的坏习惯，就是让模型吐一段自然语言，然后你用正则去抠 JSON。Mastra 的 Agent 支持 structuredOutput，直接让模型按 Zod schema 返对象，结果落在 response.object 上：

import { z } from 'zod'

const response = await agent.generate('帮我规划今天，输出成日程数组。', {
  structuredOutput: {
    schema: z.array(
      z.object({
        name: z.string(),
        activities: z.array(z.string()),
      }),
    ),
  },
})

console.log(response.object)
// [
//   { name: 'Morning Routine', activities: ['Wake up', 'Exercise', 'Breakfast'] },
//   { name: 'Work', activities: ['Check emails', 'Meeting'] },
// ]

流式场景同样支持：stream.fullStream 里会出现 object-result 类型的 chunk，流结束后 stream.object 拿到完整结构。这意味着你前端可以边打字边把结构化结果渲染进卡片，而不是等全部生成完再解析。

观点：结构化输出不是「锦上添花」，而是「生产可用性」的底线。凡是模型输出要进数据库、要驱动 UI、要调下游 API 的地方，一律上 schema。这能把一类最棘手的线上故障（「模型偶尔返回格式不对导致 JSON.parse 崩」）直接消灭在协议层。

2.3 Workflow：当流程已知，把控制权从模型手里拿回来

这是 Mastra 和「纯 Agent 框架」气质分叉的地方。Agent 适合开放问题，但很多业务其实是流程已知的：先取草稿、再做质量检查、再决定发布还是打回、最后落库通知。这类事情让 LLM 自己「想下一步干啥」既慢又不可控。Workflow 用图式 API 把流程写死，每一步都是强 schema 约束的 step：

import { createWorkflow, createStep } from '@mastra/core/workflows'
import { z } from 'zod'

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  execute: async ({ inputData }) => ({ formatted: inputData.message.toUpperCase() }),
})

const step2 = createStep({
  id: 'step-2',
  inputSchema: z.object({ formatted: z.string() }),
  outputSchema: z.object({ emphasized: z.string() }),
  execute: async ({ inputData }) => ({ emphasized: `**${inputData.formatted}**` }),
})

export const testWorkflow = createWorkflow({
  id: 'test-workflow',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ emphasized: z.string() }),
})
  .then(step1)
  .then(step2)
  .commit()

三条铁律先记住：

第一个 step 的 inputSchema 必须等于 workflow 的 inputSchema；
最后一个 step 的 outputSchema 必须等于 workflow 的 outputSchema；
相邻 step 的输出/输入 schema 必须对齐，不对齐就用「输入数据映射（input data mapping）」做转换。

并行：`.parallel()`

多个互不依赖的步骤可以并行跑，全部完成后才进下一步。并行步骤的输出会按 id 聚合成一个对象，下游 step 按 id 取值：

const formatStep = createStep({
  id: 'format-step',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  execute: async ({ inputData }) => ({ formatted: inputData.message.toUpperCase() }),
})

const countStep = createStep({
  id: 'count-step',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ count: z.number() }),
  execute: async ({ inputData }) => ({ count: inputData.message.length }),
})

const combineStep = createStep({
  id: 'combine-step',
  // 必须匹配并行输出的结构：每个 key 是并行 step 的 id
  inputSchema: z.object({
    'format-step': z.object({ formatted: z.string() }),
    'count-step': z.object({ count: z.number() }),
  }),
  outputSchema: z.object({ result: z.string() }),
  execute: async ({ inputData }) => ({
    result: `${inputData['format-step'].formatted} (${inputData['count-step'].count} chars)`,
  }),
})

export const parallelWorkflow = createWorkflow({
  id: 'parallel-output-example',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ result: z.string() }),
})
  .parallel([formatStep, countStep])
  .then(combineStep)
  .commit()

生产级技巧：并行步骤里如果某一个可能失败（比如多个研究 Agent 中有一个 token 过期），别让整个并行块崩。在 step 内部用 try/catch 把它包成「带 failed 标志的成功结果」，下游 step 再过滤掉失败项。这样「部分失败」不会拖垮整体，比把错误往外抛要稳得多。

人工介入（Human-in-the-loop）：`suspend()` / `resume()` / `bail()`

这是 Workflow 区别于「普通函数编排」的杀手锏。某些步骤需要人拍板——发邮件、删数据、发布文章。Mastra 允许 step 在执行中挂起（suspend），把「为什么暂停、需要你做什么」作为 payload 返回，等人类给输入后再 resume 从原 step 继续；人类若拒绝，用 bail() 干净地终止（状态记为 success，不抛错）：

const approvalStep = createStep({
  id: 'approval-step',
  inputSchema: z.object({ userEmail: z.string() }),
  outputSchema: z.object({ output: z.string() }),
  // 恢复时人类会回填的数据形状
  resumeSchema: z.object({ approved: z.boolean() }),
  // 挂起时返回给人类的上下文
  suspendSchema: z.object({ reason: z.string() }),
  execute: async ({ inputData, resumeData, suspend, bail }) => {
    const { userEmail } = inputData
    const { approved } = resumeData ?? {}

    if (approved === false) {
      return bail({ reason: 'User rejected the request.' })
    }
    if (!approved) {
      return await suspend({ reason: 'Human approval required before sending.' })
    }
    return { output: `Email sent to ${userEmail}` }
  },
})

export const sendWorkflow = createWorkflow({
  id: 'send-workflow',
  inputSchema: z.object({ userEmail: z.string() }),
  outputSchema: z.object({ output: z.string() }),
})
  .then(approvalStep)
  .commit()

调用方先 start，发现 status === 'suspended' 就把 suspendPayload 推给前端审核界面；人类点批准/拒绝后，用 run.resume({ step: 'approval-step', resumeData: { approved: true } }) 续跑：

const workflow = mastra.getWorkflow('send-workflow')
const run = await workflow.createRun()

const result = await run.start({ inputData: { userEmail: 'alex@example.com' } })

if (result.status === 'suspended') {
  const suspendStep = result.suspended[0]
  const suspendedPayload = result.steps[suspendStep[0]].suspendPayload
  console.log(suspendedPayload) // { reason: 'Human approval required before sending.' }
}

// 人类审核通过后
await run.resume({ step: 'approval-step', resumeData: { approved: true } })

除了上述，Workflow 还内置了快照（Snapshots）、时间旅行（Time Travel）、错误处理、定时工作流（Scheduled Workflows），并且可以部署到 Inngest 这类托管式 workflow runner 上获得托管基础设施（持久化、自动重试、可观测）。对「跑一次要十几分钟、中途不能丢状态」的长任务，这一套是刚需。

观点：Workflow 的本质，是把「流程的确定性」和「步骤的不确定性」分开。流程由你用 schema 钉死（可测试、可审计、可回放），步骤内部可以调 LLM、可以挂起等人、可以失败重试。一个成熟的 Agent 系统，绝大多数业务逻辑都应该落在 Workflow 里，而不是交给 Agent 自由发挥。Agent 负责「思考」，Workflow 负责「保证事情按规矩发生」。

2.4 Memory：三层记忆，对抗上下文窗口的熵增

没有记忆的 Agent 每轮对话都像金鱼。Mastra 的 Memory 是一套分层系统，解决「多轮一致、长期不炸、成本可控」三个问题：

消息历史（Message History）：最基础的一层，按 resource（用户/实体的稳定 ID）+ thread（某次会话的隔离 ID）把对话存起来。同一个 resource+thread 再来一次，Agent 就「想起来了」。
观察记忆（Observational Memory，推荐）：长对话里原始消息会越堆越多直到撑爆上下文。观察记忆用后台 Agent 把旧消息压缩成稠密观察日志，既保住长期记忆，又把上下文窗口压住。这是 Mastra 记忆层里最聪明的一笔。
工作记忆（Working Memory）：持久化的结构化用户数据，比如名字、偏好、目标。它不像聊天记录那样流动，而像一份一直跟着用户的「档案卡」。
语义回忆（Semantic Recall）：按「意思相近」而不是「关键词相同」去召回历史消息。用户说「上周那个事」，能找回语义相关的过去对话。
多用户线程（Multi-user Threads）：一个 thread 可以由多个用户共享。

Memory 需要存储后端。最小可跑用 @mastra/libsql（:memory: 或文件）：

import { Mastra } from '@mastra/core'
import { LibSQLStore } from '@mastra/libsql'

export const mastra = new Mastra({
  storage: new LibSQLStore({
    id: 'mastra-storage',
    url: ':memory:',
  }),
})

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'

export const memoryAgent = new Agent({
  id: 'memory-agent',
  name: 'Memory Agent',
  model: 'anthropic/claude-haiku-4-5',
  memory: new Memory({
    options: {
      lastMessages: 20,            // 默认保留最近 20 条
      observationalMemory: true,   // 开启观察记忆压缩
    },
  }),
})

使用侧，把 resource/thread 透传进去即可：

await memoryAgent.generate('记住我最喜欢的颜色是蓝色。', {
  memory: { resource: 'user-123', thread: 'conversation-123' },
})

const reply = await memoryAgent.generate('我最喜欢什么颜色？', {
  memory: { resource: 'user-123', thread: 'conversation-123' },
})
// "你最喜欢的颜色是蓝色。"

踩坑提醒：每个 thread 创建后有一个固定的「所有者」resourceId，之后不能改。别拿同一个 thread ID 给不同用户复用，否则查询会报错。这是多租户场景里非常容易翻车的一点。

当记忆总量超过模型上下文上限，Mastra 还有**记忆处理器（Memory Processors）**负责过滤、裁剪、优先级排序，保证塞进窗口的永远是最相关的那部分。

2.5 RAG：把你的私有知识接进模型

RAG 是让模型「基于真实资料回答」而不是「凭权重瞎编」的标准手段。Mastra 的 RAG 提供标准化 API：文档切块 → 生成 embedding → 存入向量库 → 查询时召回相似块。支持的向量库包括 pgvector、Pinecone、Qdrant、MongoDB。

import { embedMany } from 'ai'
import { PgVector } from '@mastra/pg'
import { MDocument } from '@mastra/rag'
import { ModelRouterEmbeddingModel } from '@mastra/core/llm'

// 1. 载入文档
const doc = MDocument.fromText('你的文档正文……')

// 2. 切块（递归策略，512 token 一块，重叠 50 防割裂）
const chunks = await doc.chunk({ strategy: 'recursive', size: 512, overlap: 50 })

// 3. 生成 embedding
const { embeddings } = await embedMany({
  values: chunks.map(c => c.text),
  model: new ModelRouterEmbeddingModel('openai/text-embedding-3-small'),
})

// 4. 存入向量库（以 pgvector 为例）
const pgVector = new PgVector({
  id: 'pg-vector',
  connectionString: process.env.POSTGRES_CONNECTION_STRING,
})
await pgVector.upsert({ indexName: 'embeddings', vectors: embeddings })

// 5. 查询相似块
const results = await pgVector.query({
  indexName: 'embeddings',
  queryVector, // 查询文本的 embedding
  topK: 3,
})

MDocument 支持多种切块策略（递归、滑动窗口等）并可为块附加元数据；召回后还能接 GraphRAG 做图结构检索。RAG 和 Memory 的区别要分清：Memory 记的是「你和这个用户的对话历史」，RAG 取的是「你放在知识库里的静态资料」。一个管「过程」，一个管「背景」。

2.6 部署、语音、渠道：从「能跑」到「能服务」

Mastra 的工程闭环在最后一步收口：

本地开发：npx mastra dev 起一个带 REST API 和 Studio（可视化观测）的本地服务。
独立部署：可以作为一个 standalone server 部署到任意能跑 Node 的地方，对外暴露 Agent/Workflow 的 HTTP 接口；也可以直接在 Next.js Route Handler、Node 服务里 import 使用，和你的业务代码共用进程。
可观测：Studio 里能看每次调用的 token、步骤、记忆状态，调试 Agent 不再靠「print 大法」。
语音（Voice）与渠道（Channels）：内置 TTS/STT、LiveKit 对接；渠道层封装了 Slack、Microsoft Teams、Discord、Telegram、WhatsApp，让同一个 Agent 能长在多个IM里。
协议互通：支持 MCP（模型上下文协议）、A2A（Agent 对 Agent）、ACP，以及 Supervisor Agents（一个主管 Agent 编排多个子 Agent）。这意味着 Mastra 的 Agent 能作为「工具」被别的系统调用，也能反过来调用外部 MCP 工具。

三、架构分析：Mastra 为什么长这样

把上面这些能力摊开看，Mastra 的架构骨架可以归纳成五层：

3.1 包即模块：monorepo 化的能力拼装

Mastra 不是「一个巨石包」，而是按能力拆成独立 npm 包：@mastra/core（骨架）、@mastra/memory、@mastra/libsql、@mastra/pg、@mastra/rag、各 deployers 等。你需要记忆就装 @mastra/memory + 一个存储后端；需要 RAG 就装 @mastra/rag + 一个向量库。按需引入、不强塞。这对生产项目的体积是实打实的友好——你不会为了用个 Agent 把整个向量库依赖都拖进来。

3.2 模型路由层：用字符串消灭 provider 耦合

provider/model 这个字符串设计，是 Mastra 最被低估的工程决策。它带来三个收益：

零 provider 依赖切换：换模型不 import 任何新包，只改字符串。
天然适配配置化/灰度：模型 ID 可以来自环境变量、配置中心，甚至按请求动态决定。
自动凭据绑定：openai/xxx 自动读 OPENAI_API_KEY，anthropic/xxx 自动读 ANTHROPIC_API_KEY，凭据管理不侵入业务代码。

底层它借助 Vercel AI SDK 统一了几十家供应商的调用差异，上层 Mastra 只暴露一个字符串入口。这套「薄路由 + 厚原语」的分层，让 Mastra 既不被某家模型绑架，又不让业务代码感知供应商差异。

3.3 存储抽象：记忆与向量的可插拔底座

Memory 和 RAG 都建立在「存储抽象」之上。LibSQLStore 适合轻量/嵌入式，PgVector 适合已经用 Postgres 的团队（顺便白嫖 pgvector 索引）。存储与计算解耦，意味着你将来从本地 LibSQL 迁到托管 Postgres，业务代码几乎不动——只换一个 store 实现。

3.4 执行引擎：Workflow 的「可暂停、可回放」

Workflow 的执行引擎是 Mastra 区别于「脚本式编排」的核心。.then/.parallel 描述的有向图，在引擎里被拆解成可持久化的执行计划。配合 suspend/resume/bail 和快照，一个跑了一半的工作流可以被序列化、落库、过几分钟甚至几天再被人类输入唤醒续跑——而且能从暂停的那一步继续，而不是从头重来。当这套引擎部署到 Inngest 这类 runner 上时，你还免费获得「崩溃自动重试、步骤幂等、执行可观测」。

3.5 一根脊梁：Agent 负责思考，Workflow 负责保证

回到前面强调的那根脊梁——Mastra 明确区分了「Agent（探索式推理）」和「Workflow（确定性编排）」。这看似平常，实则是对「Agent 到底该怎么进生产」的成熟回答。很多框架的问题在于：什么都让 Agent 自己决定，于是不确定性被放大到整个系统，没法测试、没法审计、没法回放。Mastra 的答案是「能确定的流程就写进 Workflow 用 schema 钉死，只有真正开放的问题才交给 Agent 推理」。这个取舍，是它工程气质最浓的一笔。

四、代码实战：搭一条「博客文章自动审核发布流水线」

光讲概念不够。我们用一个完全贴合本站点（技术博客平台）的场景把上面所有原语串起来：一条「草稿 → 质量评估 → 风格规范检索 → 高风险挂起人工确认 → 发布」的工作流。它同时用到了 Agent（质量评估）、RAG（风格规范库）、Workflow（编排 + 人工介入）、Structured Output（评估结果）、Memory（记住作者偏好）。

4.1 准备：工具与风格规范 RAG

先定义一个「发布到本站」的工具（注意 createTool 的标准姿势）：

// src/mastra/tools/publish-tool.ts
import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

export const publishTool = createTool({
  id: 'publish-to-blog',
  description: '将审核通过的文章发布到技术博客平台，返回文章 ID',
  inputSchema: z.object({
    title: z.string(),
    content: z.string(),
    cid: z.number(),
    tags: z.array(z.string()),
  }),
  outputSchema: z.object({
    articleId: z.string(),
    ok: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    // 真实场景：调用平台发布接口（见 chenxutan-article-publish 流程）
    const articleId = `art_${Date.now()}`
    return { articleId, ok: true }
  },
})

把「本站写作规范」灌进向量库，作为 RAG 召回源：

// src/mastra/rag/style-guide.ts
import { embedMany } from 'ai'
import { PgVector } from '@mastra/pg'
import { MDocument } from '@mastra/rag'
import { ModelRouterEmbeddingModel } from '@mastra/core/llm'

const guideText = `
  标题要有干货感、不标题党；正文需配代码示例；
  结构完整：背景→核心概念→架构→实战→优化→总结；
  避免编造事实，技术点要深入。
`

const doc = MDocument.fromText(guideText)
const chunks = await doc.chunk({ strategy: 'recursive', size: 256, overlap: 32 })
const { embeddings } = await embedMany({
  values: chunks.map(c => c.text),
  model: new ModelRouterEmbeddingModel('openai/text-embedding-3-small'),
})
const pgVector = new PgVector({ id: 'pg-vector', connectionString: process.env.POSTGRES_CONNECTION_STRING! })
await pgVector.upsert({ indexName: 'style-guide', vectors: embeddings })

4.2 质量评估 Agent（结构化输出）

用 structuredOutput 让评估 Agent 吐出可程序化消费的评分，而不是一段「我觉得还行」：

// src/mastra/agents/reviewer-agent.ts
import { Agent } from '@mastra/core/agent'
import { z } from 'zod'

export const reviewerAgent = new Agent({
  id: 'reviewer-agent',
  name: 'Article Reviewer',
  instructions: `
    你是一位严格的技术博客审稿人。给定文章草稿，评估其技术深度、结构完整性、
    事实准确性与可读性。按给定 schema 返回结构化评分，不要返回多余文本。
  `,
  model: 'anthropic/claude-sonnet-4-6',
  // 这里也可以把 style-guide 的召回结果作为上下文传入
})

// 调用时拿结构化结果
const review = await reviewerAgent.generate('评估这篇草稿：' + draft, {
  structuredOutput: {
    schema: z.object({
      score: z.number().describe('0-100 综合分'),
      depth: z.number().describe('技术深度 0-10'),
      structureOk: z.boolean().describe('结构是否完整'),
      issues: z.array(z.string()).describe('主要问题清单'),
      recommendPublish: z.boolean(),
    }),
  },
})
console.log(review.object) // { score: 82, depth: 8, structureOk: true, issues: [...], recommendPublish: true }

4.3 工作流编排 + 人工介入

把「取草稿 → 评估 → 高风险挂起 → 发布」串成 Workflow。分数低于阈值就 suspend 等人拍板：

// src/mastra/workflows/publish-pipeline.ts
import { createWorkflow, createStep } from '@mastra/core/workflows'
import { z } from 'zod'
import { reviewerAgent } from '../agents/reviewer-agent.ts'
import { publishTool } from '../tools/publish-tool.ts'

// 步骤1：质量评估（内部调 Agent，结构化输出）
const reviewStep = createStep({
  id: 'review-step',
  inputSchema: z.object({ draft: z.string(), authorId: z.string() }),
  outputSchema: z.object({
    score: z.number(),
    recommendPublish: z.boolean(),
  }),
  execute: async ({ inputData }) => {
    const r = await reviewerAgent.generate('评估草稿：' + inputData.draft, {
      structuredOutput: {
        schema: z.object({
          score: z.number(),
          recommendPublish: z.boolean(),
        }),
      },
    })
    return { score: r.object.score, recommendPublish: r.object.recommendPublish }
  },
})

// 步骤2：决策 + 高风险人工确认
const decisionStep = createStep({
  id: 'decision-step',
  inputSchema: z.object({ draft: z.string(), authorId: z.string(), score: z.number(), recommendPublish: z.boolean() }),
  outputSchema: z.object({ published: z.boolean(), articleId: z.string().nullable() }),
  resumeSchema: z.object({ approved: z.boolean() }),
  suspendSchema: z.object({ reason: z.string() }),
  execute: async ({ inputData, resumeData, suspend, bail }) => {
    const passed = inputData.recommendPublish && inputData.score >= 75

    // 没过线，且还没收到人工决定 → 挂起等人
    if (!passed && !resumeData) {
      return await suspend({
        reason: `质量分 ${inputData.score} 低于发布阈值 75，需人工确认是否发布。`,
      })
    }
    // 人类明确拒绝
    if (resumeData && resumeData.approved === false) {
      return bail({ reason: '人工拒绝发布。' })
    }
    // 通过或人工批准 → 调用发布工具
    const result = await publishTool.execute!({ inputData: {
      title: '待定标题', content: inputData.draft, cid: 1, tags: ['AI'],
    } })
    return { published: result.ok, articleId: result.articleId }
  },
})

export const publishPipeline = createWorkflow({
  id: 'publish-pipeline',
  inputSchema: z.object({ draft: z.string(), authorId: z.string() }),
  outputSchema: z.object({ published: z.boolean(), articleId: z.string().nullable() }),
})
  .then(reviewStep)
  .then(decisionStep)
  .commit()

调用方：

const wf = mastra.getWorkflow('publish-pipeline')
const run = await wf.createRun()

const res = await run.start({ inputData: { draft, authorId: 'user-123' } })

if (res.status === 'suspended') {
  // 把 res.steps[...].suspendPayload 推到审核后台，等编辑点「通过/拒绝」
  // 通过后：
  await run.resume({ step: 'decision-step', resumeData: { approved: true } })
}

这条流水线把「AI 审稿 + 人工兜底 + 自动发布」收敛进一个可测试、可审计、可暂停的工作流。它恰好解决了本类内容平台最现实的问题：既要靠模型提效，又不能让模型擅自把没审过的东西发出去。

4.4 记忆实战：记住作者偏好

给审稿 Agent 挂上 Memory，让它在多轮协作里「记得」某位作者的常见毛病和历史偏好：

const reviewerWithMemory = new Agent({
  id: 'reviewer-agent',
  name: 'Article Reviewer',
  model: 'anthropic/claude-sonnet-4-6',
  memory: new Memory({ options: { lastMessages: 15, observationalMemory: true } }),
})

await reviewerWithMemory.generate('我这篇喜欢用很多行话，帮我看看是不是太晦涩。', {
  memory: { resource: 'author-123', thread: 'collab-1' },
})

五、性能优化与生产实践清单

把 Mastra 跑进生产，光会写还不够，下面是一份按「性价比」排好序的优化清单。

5.1 长任务上 Workflow Runner，别用内置引擎硬扛

内置执行引擎适合轻量/短流程。一旦工作流要跑几分钟以上、或跨进程/跨重启不能丢状态，就把它部署到 Inngest 这类托管 runner。你白拿：步骤级持久化、崩溃自动重试、执行可观测、背压控制。这比自己手写「状态存 Redis + 定时轮询」稳十倍。

5.2 用 Memory Processors 把上下文成本压住

观察记忆（Observational Memory）默认就帮你压缩旧消息。若还不够，显式配置 lastMessages 控制保留条数，并用 memory processors 在超阈值时裁剪/优先级排序。上下文每少 1k token，就是实打实的钱和延迟。

5.3 RAG 的切块与 embedding 是性价比高地

切块（chunking）：size/overlap 直接决定召回质量。recursive 策略按语义边界切，比盲目按字符切好；overlap 设 10%-15% 能避免知识点被截断在块边缘。
embedding 模型选型：检索质量看模型，成本看维度。高频查询可以先用小维度模型粗排，必要时大维度精排。别默认上最贵那个。
向量库：已经在用 Postgres 就直接 PgVector + pgvector 索引，省一套独立基础设施；超大规模再考虑 Pinecone/Qdrant。

5.4 模型路由做成本分层

Workflow 里不同步骤对「智商」的需求天差地别：草稿打标签用 claude-haiku-4-5/gpt-5-mini 就够了，只有真正需要深度推理的评估步骤才上 claude-opus-4-7/gpt-5.5。因为模型是字符串，你可以在配置里集中管理「每一步用哪个模型」，甚至按流量灰度。这是 Mastra 相比「写死一个 model 对象」最爽的地方。

5.5 永远走结构化输出，永远开流式

凡是模型输出要进系统，上 structuredOutput（Zod schema），把「格式偶发错误」消灭在协议层。
前端交互一律 stream，用 fullStream 的 object-result 把结构化结果渐进渲染，用户体验和成本同时受益。

5.6 给危险操作上 Guardrails 与人工确认

发布、删除、发邮件、扣费——这类「不可逆/有外部影响」的动作，务必走 Workflow 的 suspend/resume/bail 人工介入，或加 Guardrails 校验。把「人类确认」作为工作流的一等节点，比在 Agent 里加一句「请务必谨慎」靠谱一万倍。

5.7 可观测优先

本地 npx mastra dev 的 Studio 不是玩具，是调试主战场。上线后把每次 Agent/Workflow 调用的 token、步骤、记忆状态接进你的监控（Mastra 有 tracing context 可透传）。Agent 系统的 bug 往往不是「崩了」，而是「答得不对」，没观测你连复现都做不到。

六、总结与展望

6.1 Mastra 在 Agent 版图里的位置

把主流框架放在一起看，差异很清晰：

LangChain / LangGraph（Python 为主）：生态最厚、图编排最强，但对 TS 团队是「外来者」。
CrewAI / AutoGen（Python）：多 Agent 协作强，研究/原型友好。
Spring AI（Java）：企业 Java 栈的 AI 原生答案，和 Mastra 定位类似但语言不同。
Mastra（TypeScript）：Web/全栈团队的「母语级」Agent 框架，强在从原型到生产的工程闭环，而不是论文式的能力堆叠。

Mastra 不追求「什么都能做」，它追求「在 TS 栈里把 Agent 做进生产这件事变得不别扭」。这个取舍，恰恰是大量真实团队最缺的那块拼图。

6.2 什么时候不该用 Mastra

说句公道话：

如果你只是想给一个函数套个 LLM 调用，直接用 Vercel AI SDK 就够了，别引入整层框架。
如果团队主力是 Python 算法工程师，Mastra 给你的「母语优势」你享受不到。
如果你要的是离线研究管线、超大规模分布式训练，那本来也不是 Mastra 的场子。

框架是用来降低复杂度的，当它开始增加你的复杂度，就是过度设计。

6.3 展望：Agent 正在变成「基础设施」

2026 年的趋势很明确：Agent 不再是「一个炫酷的 demo」，而是「产品里的一段标准基础设施」。Mastra 押注的几件事——TypeScript 一等公民、模型路由标准化、Workflow 与 Agent 分工、记忆/RAG/人工介入的生产原语、MCP/A2A 互通——都指向同一个方向：让 Agent 像数据库、缓存、消息队列一样，成为后端工程师随手就能接、接了就能上线的标准组件。

AG-UI（Agent 对前端 UI 的协议）、A2A（Agent 互操作）、Durable Execution（持久化执行）会是下一个战场。谁能把「Agent 的确定性」和「LLM 的开放能力」缝得最顺滑，谁就拿到下一代应用架构的门票。Mastra 目前在这条路上的姿态，是值得 TS 团队认真盯着的。

一句话收尾：Mastra 不是又一个「用 TS 重写 LangChain」的跟风者，它真正想解决的是——当你是一个 Web 工程师，想把 Agent 稳稳地写进生产时，那一大半「框架之外、落地之内」的脏活累活。把这部分做扎实了，TypeScript 成为 AI Agent 的一等公民，就不再是口号。

本文基于 Mastra 官方文档（mastra.ai/docs）的公开 API 与 2026 年社区趋势整理，代码示例遵循当前 provider/model 模型路由与 createTool/createStep/createWorkflow 的契约写法。实际落地请以你所使用版本的官方文档为准。