技术笔记主流 Agent 框架调研 AI Agent 工作流编排 Mastra TypeScript Next.js

Agent 框架 02：Mastra——面向 TypeScript 生态的现代化 Agent 框架与工作流设计

Asaakii2026-04-222026-10-08

在很长一段时间里，构建 AI Agent 几乎是 Python 生态的专属领地。然而在真实的业务开发中，大量互联网团队的业务主干、前端界面与微服务中台都是基于 Node.js 与 TypeScript 构建的。如果为了引入几个智能体功能而专门架设一套 Python 微服务，团队往往要背负跨语言维护、双重依赖体系以及 RPC 序列化开销等工程包袱。

Mastra 应运而生。它不是将 Python 库进行简单翻译的“二道贩子”，而是针对 TypeScript 开发者的习惯从零设计的现代化 Agent 框架。它将强类型验证库 Zod、基于有向图的 Workflow、企业级 Memory 与原生流式传输深度整合，为全栈工程师提供了一套丝滑的端到端开发体验。

核心设计哲学：类型安全与全栈一体化

Mastra 最显著的标签是**“编译期类型安全”**。它放弃了动态拼接字典的做法，把所有输入、输出与上下文全部纳管在 TypeScript 类型系统下：

  flowchart TD
    Init["顶层单例 new Mastra({ agents, workflows })"] --> RegAgent["Agent 智能体定义"]
    Init --> RegWF["Workflow 有向图工作流"]
    
    subgraph Tooling ["Zod 强类型工具契约"]
        ZodIn["inputSchema: z.object(...)"] --> Exec["execute: async ({ context }) => ..."]
        Exec --> ZodOut["outputSchema: z.object(...)"]
    end
    RegAgent --> Tooling
    
    subgraph Runtime ["多端集成与运行时"]
        Next["Next.js App Router (toDataStreamResponse)"]
        Memory["跨会话持久化 Memory (threadId / resourceId)"]
        Vector["向量检索 RAG (@mastra/vector-*)"]
    end
    Init --> Runtime

通过与 Vercel AI SDK 底层驱动（@ai-sdk/openai 等）的无缝协作，开发者可以在同一个 Monorepo 代码库中完成从前端 UI 交互到后端智能体调度的全流程。

核心概念原语

构件	运行时角色	说明
`Agent`	智能体实体	挂载模型配置、提示词指令、专属工具集与记忆存储的独立执行单元
`createTool`	工具构造器	基于 Zod Schema 声明严格出入参校验的函数封装，运行时自动推导参数类型
`Workflow` / `Step`	有向图工作流	支持串行、条件分支与并行执行的有状态执行图，步骤间输出具备静态类型传递
`Mastra`	容器中枢	项目根实例，统筹注册、发现与管理所有智能体与工作流

安装与工程初始化

在现有的 Node.js / TypeScript 项目中添加核心依赖：

1	pnpm add @mastra/core @ai-sdk/openai zod

最小运行示例

import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import { openai } from "@ai-sdk/openai";

// 1. 声明 Agent
const assistant = new Agent({
  name: "ArchitectureConsultant",
  instructions: "你是一名资深的云原生系统架构师，擅长用清晰的技术逻辑剖析微服务瓶颈。",
  model: openai("gpt-4o-mini"),
});

// 2. 注册进 Mastra 容器
export const mastra = new Mastra({
  agents: { assistant },
});

// 3. 触发生成
async function run() {
  const agent = mastra.getAgent("assistant");
  const response = await agent.generate("简述服务网格（Service Mesh）解决的核心网络痛点。");
  console.log(response.text);
}

run();

基于 Zod 的强类型工具声明

在 Mastra 中，定义工具不仅需要声明执行逻辑，还必须提供输入与输出的双向 Zod Schema。这不仅为大模型提供了准确的 JSON Schema 参数定义，还在本地代码中杜绝了 undefined 字段引发的运行时崩溃：

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

// 创建具备严格出入参约定的生产工具
export const queryHostMetricsTool = createTool({
  id: "query-host-metrics",
  description: "根据主机标识获取服务器实时的 CPU 负载与可用内存",
  inputSchema: z.object({
    hostId: z.string().describe("主机名或实例 ID，例如 'node-k8s-01'"),
  }),
  outputSchema: z.object({
    cpuLoadPercent: z.number(),
    freeMemoryMb: z.number(),
    status: z.enum(["HEALTHY", "WARNING", "CRITICAL"]),
  }),
  execute: async ({ context }) => {
    // context.hostId 具有完整的 TypeScript 类型提示
    const { hostId } = context;
    
    // 模拟拉取监控数据
    return {
      cpuLoadPercent: 88.5,
      freeMemoryMb: 512,
      status: "WARNING",
    };
  },
});

将工具注入 Agent 极其直接：

const opsAgent = new Agent({
  name: "SreAgent",
  instructions: "负责巡检服务器资源指标并在异常时提出扩容预警。",
  model: openai("gpt-4o-mini"),
  tools: { queryHostMetricsTool },
});

有向图工作流（Workflow）：确定性流程编排

并非所有业务都适合让大模型完全自主“自由发挥”。Mastra 提供了图工作流（Workflow），允许将确定性的前置处理、大模型审查与后置落库组装为有序管道：

import { Workflow, Step } from "@mastra/core/workflows";
import { z } from "zod";

// 步骤 1: 数据拉取
const fetchLogStep = new Step({
  id: "fetch-logs",
  outputSchema: z.object({ rawLogs: z.string() }),
  execute: async ({ context }) => {
    const serviceName = context.triggerData.serviceName as string;
    return { rawLogs: `[ERROR] Connection reset by peer at service ${serviceName}` };
  },
});

// 步骤 2: 数据提炼（静态感知上一步返回值）
const diagnoseStep = new Step({
  id: "diagnose",
  outputSchema: z.object({ conclusion: z.string() }),
  execute: async ({ context }) => {
    const prev = context.getStepResult<{ rawLogs: string }>("fetch-logs");
    return {
      conclusion: `初步排错意见: 发现下游网络连接重置 -> ${prev?.rawLogs}`,
    };
  },
});

// 组装并提交工作流
export const incidentWorkflow = new Workflow({
  name: "incident-triage-pipeline",
  triggerSchema: z.object({ serviceName: z.string() }),
})
  .step(fetchLogStep)
  .then(diagnoseStep)
  .commit();

// 驱动运行
const { start } = incidentWorkflow.createRun();
const workflowResult = await start({ triggerData: { serviceName: "payment-gateway" } });
console.log(workflowResult.results["diagnose"].output.conclusion);

跨会话记忆与 Next.js App Router 全栈集成

在全栈 Web 应用中，智能体通常需要通过 HTTP Server-Sent Events（SSE）向前端页面流式推送文字。Mastra 可以直接内嵌于 Next.js 的路由处理器中：

// app/api/chat/route.ts
import { mastra } from "@/mastra";
import { NextRequest } from "next/server";

export async function POST(req: NextRequest) {
  const { message, userId } = await req.json();
  const agent = mastra.getAgent("assistant");

  // 使用 threadId 实现多用户会话隔离
  const stream = await agent.stream(message, {
    threadId: `thread-user-${userId}`,
    resourceId: "chat-window-main",
  });

  // 直接转化为标准 Web 流式响应
  return stream.toDataStreamResponse();
}