Mastra 深度解析：Gatsby 团队打造的 TypeScript AI Agent 框架——从 Agent 构建到 Workflow 编排的完整实战指南

引言：为什么 TypeScript 开发者需要一个专属的 AI Agent 框架？

2026 年，AI Agent 已经从概念验证走向了生产落地。但如果你是一个 TypeScript 开发者，你会发现一个尴尬的现实：Python 生态有 LangChain、CrewAI、AutoGen 等成熟框架，而 TypeScript 生态却长期缺乏一个「全家桶」级别的解决方案。

你可能用过 Vercel AI SDK 来做模型调用，但它更偏向于「SDK」而非「框架」——它帮你解决了「怎么调模型」的问题，但没有解决「怎么构建 Agent、怎么编排 Workflow、怎么做 RAG、怎么管理记忆」这些更高层的问题。

Mastra 就是为了解决这个问题而生的。

由 Gatsby 团队（没错，就是那个做静态网站生成器的 Gatsby）打造，Mastra 是一个全功能的 TypeScript AI Agent 框架。它不是又一个「套壳 SDK」，而是一个从底层模型路由到上层 Agent 编排的完整解决方案。截至 2026 年 7 月，Mastra 在 GitHub 上已经积累了超过 16,000 次提交，被 Replit、Fireworks、Medusa、SoftBank 等知名公司在生产环境中使用。

本文将从架构设计、核心模块、代码实战三个维度，深入解析 Mastra 的技术细节，帮助你判断它是否适合你的下一个 AI 项目。

一、Mastra 的整体架构：模块化设计的哲学

Mastra 的架构可以用一句话概括：一个核心运行时 + 六大功能模块。

1.1 核心运行时（Mastra Core）

@mastra/core 是整个框架的基石。它提供了：

• Mastra 实例：全局应用容器，负责注册 Agent、Workflow、Tool 等组件
• 模型路由器（Model Router）：统一的 LLM 调用接口，支持 40+ 模型提供商
• 可观测性基础设施：内置的日志、追踪、评估系统

import { Mastra } from '@mastra/core'
import { weatherAgent } from './agents/weather-agent'
import { dataWorkflow } from './workflows/data-workflow'

export const mastra = new Mastra({
  agents: { weatherAgent },
  workflows: { dataWorkflow },
  // 可选：配置日志、存储等
})

这个设计的精妙之处在于：Mastra 实例是一个依赖注入容器。你注册的所有 Agent、Workflow、Tool 都通过它来访问共享资源（记忆、存储、日志等），而不是各自为政。

1.2 六大功能模块

这种模块化设计意味着你可以按需引入：如果你只需要 Agent + Tools，不需要安装 RAG 相关的依赖；如果你不需要语音功能，Voice 模块完全不会出现在你的 bundle 中。

二、Agent 系统：让 LLM 学会「做事」

2.1 Agent 的本质

在 Mastra 中，Agent 不是一个简单的「聊天机器人」。它是一个具有自主决策能力的系统——给定一个目标，它能自己决定调用哪些工具、调用多少次、何时停止。

这与 Workflow 形成了鲜明的对比：

• Agent：适合开放性任务，步骤不预先确定，由 LLM 动态决策
• Workflow：适合确定性任务，步骤预先定义，控制流明确

2.2 创建一个 Agent

import { Agent } from '@mastra/core/agent'
import { weatherTool } from '../tools/weather-tool'
import { searchTool } from '../tools/search-tool'

export const assistantAgent = new Agent({
  id: 'assistant-agent',
  name: 'Smart Assistant',
  instructions: `
    你是一个智能助手，能够帮助用户完成各种任务。
    
    核心能力：
    - 查询天气信息
    - 搜索网络内容
    - 回答技术问题
    
    行为准则：
    - 优先使用工具获取实时数据
    - 回答要简洁但信息量充足
    - 不确定时坦诚告知
  `,
  model: 'openai/gpt-5.5',
  tools: { weatherTool, searchTool },
})

几个关键设计决策：

1. instructions 替代传统的 system prompt

Mastra 把 instructions 设计为 Agent 的「灵魂」。它不仅仅是 system prompt，而是 Agent 的行为规范。你可以在其中定义角色、能力边界、行为准则、输出格式等。

2. model 使用 provider/model 格式

这是 Mastra 的模型路由器格式。openai/gpt-5.5 意味着使用 OpenAI 提供的 gpt-5.5 模型。框架会自动根据 provider 查找对应的环境变量（如 OPENAI_API_KEY）。

这种设计的好处是：你可以在不修改代码的情况下切换模型提供商。把 openai/gpt-5.5 改成 anthropic/claude-sonnet-4-6，Agent 的行为逻辑完全不变。

3. Tools 通过对象注入

Tools 以 { toolId: toolInstance } 的形式传入。Agent 可以根据需要自主决定调用哪个工具。

2.3 Agent 的调用方式

Mastra 提供了两种调用 Agent 的方式：

// 方式一：完整响应（等所有 token 生成完毕）
const response = await agent.generate('北京今天天气怎么样？')
console.log(response.text)           // 文本响应
console.log(response.toolCalls)      // 工具调用记录
console.log(response.toolResults)    // 工具返回结果

// 方式二：流式响应（实时输出 token）
const stream = await agent.stream('帮我分析一下这段代码...')
for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}

2.4 Agent 的内部循环

当你调用 agent.generate() 时，Mastra 内部执行的是一个推理-行动循环（ReAct Loop）：

1. 推理：LLM 分析用户输入，决定是否需要调用工具
2. 行动：如果需要，调用指定的 Tool 并获取结果
3. 观察：将 Tool 的结果反馈给 LLM
4. 循环：LLM 继续推理，直到得出最终答案或触发停止条件

这个循环的最大迭代次数可以通过 maxSteps 参数控制：

const response = await agent.generate('帮我完成这个复杂任务', {
  maxSteps: 10,  // 最多执行 10 轮推理-行动循环
})

2.5 多 Agent 协作

Mastra 支持 Agent 之间的嵌套调用。一个 Agent 可以把另一个 Agent 当作 Tool 来使用：

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

// 专业 Agent：代码审查
const codeReviewAgent = new Agent({
  id: 'code-reviewer',
  name: 'Code Reviewer',
  instructions: '你是一个代码审查专家，专注于代码质量、安全性和最佳实践。',
  model: 'anthropic/claude-sonnet-4-6',
})

// 专业 Agent：文档生成
const docsAgent = new Agent({
  id: 'docs-generator',
  name: 'Documentation Generator',
  instructions: '你是一个技术文档专家，能够生成清晰、结构化的 API 文档。',
  model: 'openai/gpt-5.5',
})

// 协调 Agent：把专业 Agent 当作 Tool
const orchestratorAgent = new Agent({
  id: 'orchestrator',
  name: 'Project Orchestrator',
  instructions: `
    你是一个项目协调者。当用户提交代码时：
    1. 先调用 codeReviewAgent 进行代码审查
    2. 根据审查结果，调用 docsAgent 生成文档
    3. 汇总两个 Agent 的输出，给出综合报告
  `,
  model: 'openai/gpt-5.5',
  tools: {
    codeReview: codeReviewAgent,  // Agent 作为 Tool
    docsGenerator: docsAgent,
  },
})

这种设计模式让 Mastra 天然支持多 Agent 系统，而不需要额外的编排层。

三、Tool 系统：给 Agent 装上「手脚」

3.1 Tool 的定义

在 Mastra 中，Tool 不是一个普通的函数——它是一个带有 Schema 验证的结构化组件：

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

export const databaseQueryTool = createTool({
  id: 'query-database',
  description: '执行 SQL 查询并返回结果。只支持 SELECT 语句。',
  inputSchema: z.object({
    query: z.string().describe('SQL SELECT 查询语句'),
    database: z.string().default('main').describe('目标数据库名称'),
    limit: z.number().max(1000).default(100).describe('最大返回行数'),
  }),
  outputSchema: z.object({
    rows: z.array(z.record(z.unknown())),
    rowCount: z.number(),
    executionTime: z.number(),
  }),
  execute: async ({ query, database, limit }, context) => {
    // context 包含请求上下文、追踪信息、中止信号等
    const startTime = Date.now()
    
    // 执行数据库查询
    const rows = await db.query(query, { database, limit })
    
    return {
      rows,
      rowCount: rows.length,
      executionTime: Date.now() - startTime,
    }
  },
})

几个关键点：

1. createTool() 是必须的

Mastra 明确要求使用 createTool() 工厂函数来定义 Tool，而不是用普通对象。这是因为 createTool() 会做 Schema 验证、错误处理、可观测性注入等工作。直接使用普通对象定义的 Tool 会静默失败。

2. inputSchema 和 outputSchema 使用 Zod

Schema 不仅用于运行时验证，还会被传递给 LLM 作为 Tool 的参数说明。这意味着你在 z.object() 中写的 .describe() 会直接影响 LLM 对 Tool 的理解和使用方式。

3. execute 的第二个参数是上下文

context 对象包含：

• requestContext：请求级别的上下文信息
• tracingContext：OpenTelemetry 追踪上下文
• abortSignal：用于取消长时间运行的 Tool

3.2 Tool 的高级用法

动态 Tool：根据运行时条件生成 Tool

const dynamicTool = createTool({
  id: 'dynamic-api-call',
  description: '调用外部 API',
  inputSchema: z.object({
    endpoint: z.string(),
    method: z.enum(['GET', 'POST', 'PUT', 'DELETE']),
    body: z.record(z.unknown()).optional(),
  }),
  execute: async ({ endpoint, method, body }, { requestContext }) => {
    // 根据用户权限动态决定可用的 API
    const userPermissions = requestContext?.permissions ?? []
    
    if (!userPermissions.includes(endpoint)) {
      throw new Error(`无权访问 ${endpoint}`)
    }
    
    const response = await fetch(endpoint, {
      method,
      body: body ? JSON.stringify(body) : undefined,
      headers: {
        'Authorization': `Bearer ${requestContext?.token}`,
      },
    })
    
    return response.json()
  },
})

四、Workflow 系统：确定性的流程编排

4.1 为什么需要 Workflow？

Agent 适合开放性任务，但很多业务场景需要的是确定性控制流——你知道每一步做什么，只是需要把它们串起来。这时候 Workflow 就是更好的选择。

Mastra 的 Workflow 系统支持：

• 顺序执行：.then() 链式调用
• 条件分支：.branch() 根据条件走不同路径
• 并行执行：.parallel() 同时执行多个步骤
• 暂停与恢复：在任意节点等待用户输入或审批
• 状态管理：跨步骤共享数据
• 错误处理：重试、降级、回滚

4.2 创建一个 Workflow

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

// 步骤 1：数据采集
const fetchDataTask = createStep({
  id: 'fetch-data',
  inputSchema: z.object({
    source: z.string(),
    dateRange: z.object({
      start: z.string(),
      end: z.string(),
    }),
  }),
  outputSchema: z.object({
    data: z.array(z.record(z.unknown())),
    count: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { source, dateRange } = inputData
    const data = await fetchFromDataSource(source, dateRange)
    return { data, count: data.length }
  },
})

// 步骤 2：数据清洗
const cleanDataTask = createStep({
  id: 'clean-data',
  inputSchema: z.object({
    data: z.array(z.record(z.unknown())),
    count: z.number(),
  }),
  outputSchema: z.object({
    cleanedData: z.array(z.record(z.unknown())),
    removedCount: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { data } = inputData
    const cleaned = data.filter(row => isValid(row))
    return {
      cleanedData: cleaned,
      removedCount: data.length - cleaned.length,
    }
  },
})

// 步骤 3：AI 分析
const analyzeDataTask = createStep({
  id: 'analyze-data',
  inputSchema: z.object({
    cleanedData: z.array(z.record(z.unknown())),
    removedCount: z.number(),
  }),
  outputSchema: z.object({
    summary: z.string(),
    insights: z.array(z.string()),
    confidence: z.number(),
  }),
  execute: async ({ inputData }) => {
    // 这里可以调用 LLM 进行分析
    const analysis = await llm.analyze(inputData.cleanedData)
    return analysis
  },
})

// 组装 Workflow
export const dataPipeline = createWorkflow({
  id: 'data-pipeline',
  inputSchema: z.object({
    source: z.string(),
    dateRange: z.object({
      start: z.string(),
      end: z.string(),
    }),
  }),
  outputSchema: z.object({
    summary: z.string(),
    insights: z.array(z.string()),
    confidence: z.number(),
  }),
})
  .then(fetchDataTask)
  .then(cleanDataTask)
  .then(analyzeDataTask)
  .commit()

4.3 条件分支

const routeByType = createStep({
  id: 'route-by-type',
  inputSchema: z.object({
    data: z.record(z.unknown()),
    type: z.enum(['text', 'image', 'video']),
  }),
  outputSchema: z.object({
    result: z.string(),
  }),
  execute: async ({ inputData }) => {
    // 根据类型路由到不同的处理逻辑
    switch (inputData.type) {
      case 'text':
        return { result: await processText(inputData.data) }
      case 'image':
        return { result: await processImage(inputData.data) }
      case 'video':
        return { result: await processVideo(inputData.data) }
    }
  },
})

const pipeline = createWorkflow({
  id: 'multi-type-pipeline',
  inputSchema: z.object({
    data: z.record(z.unknown()),
    type: z.enum(['text', 'image', 'video']),
  }),
  outputSchema: z.object({
    result: z.string(),
  }),
})
  .then(routeByType)
  .commit()

4.4 并行执行

const pipeline = createWorkflow({
  id: 'parallel-pipeline',
  inputSchema: z.object({
    text: z.string(),
  }),
  outputSchema: z.object({
    sentiment: z.string(),
    summary: z.string(),
    keywords: z.array(z.string()),
  }),
})
  .parallel([
    sentimentAnalysisStep,
    summarizationStep,
    keywordExtractionStep,
  ])
  .then(mergeResultsStep)
  .commit()

4.5 暂停与恢复（Human-in-the-Loop）

这是 Mastra Workflow 最强大的特性之一。你可以在 Workflow 的任意节点暂停，等待人工审批或输入，然后恢复执行：

const approvalStep = createStep({
  id: 'human-approval',
  inputSchema: z.object({
    proposal: z.string(),
    riskLevel: z.enum(['low', 'medium', 'high']),
  }),
  outputSchema: z.object({
    approved: z.boolean(),
    feedback: z.string(),
  }),
  suspend: async ({ inputData }) => {
    // 发送审批通知
    await sendSlackNotification({
      message: `需要审批：${inputData.proposal}`,
      riskLevel: inputData.riskLevel,
    })
  },
  resume: async ({ inputData, suspendData }) => {
    // 用户在 UI 中审批后，resume 被调用
    return {
      approved: suspendData.approved,
      feedback: suspendData.feedback,
    }
  },
  execute: async ({ inputData }) => {
    // 正常执行逻辑
    return { approved: true, feedback: '' }
  },
})

五、RAG 系统：让 Agent 拥有「长期记忆」

5.1 RAG 的基本流程

Mastra 的 RAG 系统遵循标准的「分块 → 嵌入 → 存储 → 检索」流程：

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

// 1. 文档加载与分块
const doc = MDocument.fromText(`
  Mastra 是一个 TypeScript AI Agent 框架...
  // 更多文档内容
`)

const chunks = await doc.chunk({
  strategy: 'recursive',  // 递归分块策略
  size: 512,              // 每块最大 512 token
  overlap: 50,            // 相邻块重叠 50 token
})

// 2. 生成嵌入向量
import { ModelRouterEmbeddingModel } from '@mastra/core/llm'

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

// 3. 存储到向量数据库
const pgVector = new PgVector({
  id: 'pg-vector',
  connectionString: process.env.POSTGRES_CONNECTION_STRING,
})

await pgVector.upsert({
  indexName: 'knowledge-base',
  vectors: embeddings,
  metadata: chunks.map(chunk => ({
    text: chunk.text,
    source: chunk.metadata?.source,
  })),
})

// 4. 语义检索
const queryEmbedding = await embed('什么是 Mastra 的 Agent 系统？')
const results = await pgVector.query({
  indexName: 'knowledge-base',
  queryVector: queryEmbedding,
  topK: 5,
  filter: { source: 'documentation' },  // 可选：元数据过滤
})

5.2 支持的向量数据库

Mastra 支持多种向量数据库后端：

5.3 分块策略

Mastra 提供了多种分块策略：

• recursive：递归分块，按照段落 → 句子 → 词的层级尝试分割
• sliding window：滑动窗口，固定大小 + 重叠
• semantic：语义分块，根据语义相似度自动分割

// 递归分块（推荐）
const chunks = await doc.chunk({
  strategy: 'recursive',
  size: 512,
  overlap: 50,
})

// 语义分块（更智能但更慢）
const chunks = await doc.chunk({
  strategy: 'semantic',
  threshold: 0.8,  // 语义相似度阈值
})

六、Voice 系统：让 Agent 「开口说话」

6.1 统一的语音接口

Mastra 的 Voice 系统提供了一个统一的接口，支持 TTS（文字转语音）、STT（语音转文字）和 STS（实时语音对话）：

import { Agent } from '@mastra/core/agent'
import { OpenAIVoice } from '@mastra/voice-openai'

const voiceAgent = new Agent({
  id: 'voice-assistant',
  name: 'Voice Assistant',
  instructions: '你是一个语音助手，用简洁自然的口语风格回答问题。',
  model: 'openai/gpt-5.5',
  voice: new OpenAIVoice(),
})

// TTS：把文字转成语音
const { text } = await voiceAgent.generate('今天天气怎么样？')
const audioStream = await voiceAgent.voice.speak(text, {
  speaker: 'default',
  responseFormat: 'wav',
})

// STT：把语音转成文字
const transcription = await voiceAgent.voice.listen(audioStream)

6.2 支持的语音提供商

七、可观测性：生产环境的「仪表盘」

7.1 内置的追踪系统

Mastra 内置了基于 OpenTelemetry 的追踪系统，可以监控 Agent 的每一步推理、每次 Tool 调用、每次模型调用的延迟和成本：

import { Mastra } from '@mastra/core'

export const mastra = new Mastra({
  agents: { myAgent },
  observability: {
    // 配置追踪导出
    serviceName: 'my-ai-app',
    exporter: 'otlp',  // 导出到 Jaeger、Grafana Tempo 等
  },
})

7.2 Studio：可视化调试工具

Mastra 提供了一个名为 Studio 的可视化调试工具：

• Agent 调试：实时查看 Agent 的推理过程、Tool 调用记录
• Workflow 可视化：以图形方式展示 Workflow 的执行流程
• 时间旅行：Workflow 执行完成后，可以回放任意步骤
• 性能分析：查看每个步骤的延迟、token 消耗、成本

八、与竞品对比

8.1 Mastra vs LangChain (TypeScript)

8.2 Mastra vs Vercel AI SDK

8.3 Mastra vs AutoGen / CrewAI

九、生产部署与最佳实践

9.1 项目结构

my-mastra-app/
├── src/
│   ├── mastra/
│   │   ├── index.ts          # Mastra 实例入口
│   │   ├── agents/
│   │   │   ├── assistant.ts
│   │   │   └── code-reviewer.ts
│   │   ├── tools/
│   │   │   ├── weather.ts
│   │   │   └── database.ts
│   │   ├── workflows/
│   │   │   ├── data-pipeline.ts
│   │   │   └── approval-flow.ts
│   │   └── rag/
│   │       ├── documents.ts
│   │       └── embeddings.ts
│   └── api/
│       └── chat.ts           # API 路由
├── .env                      # API Keys
├── package.json
└── tsconfig.json

9.2 部署选项

Mastra 支持多种部署方式：

• Serverless：Vercel、Cloudflare Workers、AWS Lambda
• 容器化：Docker + Kubernetes
• 传统服务器：Node.js 进程
• 托管服务：Mastra Cloud（官方托管平台）

9.3 环境变量配置

# 模型提供商 API Keys
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AI...

# 向量数据库
POSTGRES_CONNECTION_STRING=postgresql://...

# 可观测性
OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318

十、总结与展望

10.1 Mastra 的核心优势

1. TypeScript 原生：不是从 Python 移植的，是真正为 TypeScript 生态设计的
2. 全家桶方案：Agent + Workflow + RAG + Voice + 可观测性，一站式解决
3. 生产就绪：内置部署、监控、评估，不需要额外的工程投入
4. 模块化设计：按需引入，不会让项目变得臃肿
5. 活跃的社区：16,000+ 次提交，Gatsby 团队背书

10.2 适用场景

• Web 全栈团队：已经有 Next.js / React 项目，想快速集成 AI 能力
• 企业级应用：需要生产级的 Agent 系统，包含监控、审计、合规
• 多模态应用：需要同时支持文字、语音交互
• 数据管道：需要把 AI 嵌入到现有的数据处理流程中

10.3 不适用场景

• 纯 Python 团队：如果你的团队全是 Python 开发者，LangChain 可能更合适
• 纯模型调用：如果你只需要调用 LLM API，不需要 Agent / Workflow，Vercel AI SDK 足够
• 研究用途：如果你在做 AI 研究，需要快速实验各种 Agent 架构，AutoGen 更灵活

10.4 未来展望

Mastra 正在快速迭代。从 GitHub 的提交记录来看，团队正在重点推进：

• MCP 协议支持：让 Agent 可以调用外部 MCP Server 的工具
• A2A 协议：Agent-to-Agent 通信标准
• 更丰富的 Workflow 原语：支持更复杂的控制流模式
• 企业级功能：多租户、权限管理、审计日志

对于 TypeScript 开发者来说，Mastra 可能是 2026 年最值得关注的 AI Agent 框架。它不是最灵活的，但可能是最适合「把 AI 交付到生产环境」的那个。

项目地址：https://github.com/mastra-ai/mastra

官方文档：https://mastra.ai/docs

快速开始：npm create mastra@latest