Mastra Workflows — TypeScript Agent Orchestration

简介

Mastra Workflows 是 Mastra 的类型化编排原语 —— 用 .step().then().branch().parallel().while() 串多步 agent 流水线，TypeScript 从第一个输入推断到最后一个输出，整条链类型安全。适合需要可读、可测、可观测工作流的生产 TypeScript agent。需要 Mastra 0.5+ 和 Node 20+。装机时间 5 分钟。

定义 workflow

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

const fetchData = new Step({
  id: "fetch",
  inputSchema: z.object({ url: z.string().url() }),
  outputSchema: z.object({ markdown: z.string() }),
  execute: async ({ context }) => {
    const res = await fetch(context.inputData.url);
    return { markdown: await res.text() };
  },
});

const summarize = new Step({
  id: "summarize",
  inputSchema: z.object({ markdown: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
  execute: async ({ context, mastra }) => {
    const agent = mastra.getAgent("summarizer");
    const result = await agent.generate(`Summarize: ${context.inputData.markdown}`);
    return { summary: result.text };
  },
});

const workflow = new Workflow({
  name: "fetch-and-summarize",
  triggerSchema: z.object({ url: z.string().url() }),
}).step(fetchData).then(summarize).commit();

分支

.then(classify)
.branch([
  [async ({ context }) => context.results.classify.kind === "bug",
   fixerStep],
  [async ({ context }) => context.results.classify.kind === "feature",
   plannerStep],
])

跑 + 流式

const { runId, start } = workflow.createRun();

start({ triggerData: { url: "https://example.com/blog" } });

// 订阅 step 事件
workflow.watch(runId, (event) => {
  console.log(event.type, event.payload);
});

每个 step 的输入输出会 checkpoint —— 下游 step 失败时从最后一个成功 step 重启。

FAQ

Q: Mastra Workflows 跟 LangGraph 比？ A: LangGraph 是 Python 优先 + 图 DSL。Mastra Workflows 是 TypeScript 优先 + 流式链 API。按语言偏好和生态选；解决同一类问题。

Q: Mastra 免费吗？ A: 免费 —— Elastic License 2.0 开源。可自托管。Mastra Cloud（托管部署 + 可观测）付费。

Q: workflow 能调 workflow 吗？ A: 能 —— step(otherWorkflow) 把一个 workflow 当成子 step。共享子流程（比如 validateAndCleanData 在多个流水线复用）有用。