Mastra 开发（04）：Mastra 的图编排

前三篇把单个 Agent 和工具跑通了：模型拿着一套工具，自己决定下一步做什么。但生产里另一类任务恰恰相反——步骤顺序是人定的，每步何时执行、失败了怎么办、中间要不要停下来等真人审批，都得写死在代码里。这类确定性流程就是 Workflow 的主场。本篇是系列第 4 篇，对应 Agent 系列第 6～8 篇的 LangGraph StateGraph：同样的图编排问题，看 Mastra 用 TypeScript 原生的方式怎么回答。全文按"分工 → 步骤定义 → 数据流 → 编排原语 → 人机协同 → 错误处理 → 实战"组织，最后照例留一个练习。

1、Agent 与 Workflow 的分工

先回答"什么时候上 Workflow"。Agent 的工具循环本质是一个隐式循环：模型看到工具列表，自己决定调哪个、调几次、何时停。这种开放式决策适合探索型任务——调研一个话题、排查一个 bug，路径本身不可预知，交给模型是对的。

代价是可控性。你无法在编译期保证"生成之后必须经过审核才能发布"，也无法让流程在第 3 步挂起八小时等审批再继续。而这些恰好是生产流水线的硬需求：

• 顺序固定：先抓取、再清洗、再生成、最后入库
• 要挂起：等审批、等人工复核、等外部事件到达
• 要重试：外部 API 抖动，按策略退避重试
• 要审计：每步输入输出可查，执行路径可复现

判断标准很朴素：下一步谁说了算。由模型说了算，用 Agent；由代码说了算，用 Workflow。两者是组合关系而非替代——Workflow 的一个 step 里完全可以跑一个 Agent，把确定性骨架和开放性关节拼在一起，骨架管"必须先审后发"，关节管"这一稿怎么写"。本篇实战就是这么干的。

还有一层工程红利：workflow 注册进 new Mastra({ workflows: {...} }) 之后，mastra dev 的 Studio 会把整张图可视化——节点、连边、每次运行走到哪一步，全部摊开在界面上。Agent 的隐式循环要靠翻 trace 还原路径，Workflow 的路径天生就是一张看得见的图，这是"确定性"送的可观测性。

2、createStep 步骤定义

Workflow 的最小构件是 step，用 createStep 定义。最小示例如下：

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

const fetchSourceStep = createStep({
  id: 'fetch-source',
  inputSchema: z.object({ url: z.string() }),
  outputSchema: z.object({ markdown: z.string(), chars: z.number() }),
  execute: async ({ inputData }) => {
    const res = await fetch(inputData.url);
    const markdown = await res.text();
    return { markdown, chars: markdown.length };
  },
});

四个字段各司其职：id 是步骤标识；inputSchema、outputSchema 用 Zod 声明进出的数据形状；execute 写业务逻辑，入参解构出 inputData——它已被 schema 校验并推导好类型，inputData.url 就是 string。和第 2 篇的 createTool 摆在一起看，形似而神不同：

维度	createTool	createStep
服务对象	模型，模型决定何时调用	代码，编排关系写死
description	必须写清，模型靠它决策	不需要，没人看说明书
execute 入参	{ context }	{ inputData }
触发时机	工具循环里模型说了算	上一步完成即触发

最大的差异是 createStep 没有 description：步骤不是被模型"选中"的，是被代码排定的。去掉模型这层不确定性，换来的是类型检查贯穿全流程。还有一点工程上的好处：step 是普通值，定义一次就能在多个 workflow 里复用，测试时也能单独调用，粒度比"塞进某个 Agent 的工具列表"友好得多。

3、显式数据流模型

步骤定义好了怎么连？答案是链式编排加类型咬合，完整示例如下：

import { createWorkflow } from '@mastra/core/workflows';

const summarizeStep = createStep({
  id: 'summarize',
  inputSchema: z.object({ markdown: z.string(), chars: z.number() }),
  outputSchema: z.object({ summary: z.string() }),
  execute: async ({ inputData }) => ({ summary: `共 ${inputData.chars} 字` }),
});

const pipeline = createWorkflow({
  id: 'fetch-pipeline',
  inputSchema: z.object({ url: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
})
  .then(fetchSourceStep)
  .then(summarizeStep)
  .commit();

const run = await pipeline.createRun();
const result = await run.start({ inputData: { url: 'https://example.com' } });

这套代码有三层语义。其一，上一步的输出就是下一步的输入：fetchSourceStep 的 outputSchema 与 summarizeStep 的 inputSchema 逐字段咬合，不匹配编辑器直接标红——schema 即契约，且在编译期检查。其二，.commit() 把链定型成一张固定的图，之后可以无数次 createRun()，每次运行互不干扰。其三，run.start() 的返回值是按 status 区分的联合类型：success 时取 result.result，failed 时取 result.error，suspended 的处理见第 5 节。

相邻两步形状不一致时不硬凑 schema，而是用 .map() 显式变换、.unmap() 取回变换前的数据——变换本身也要过类型检查。最典型的场景在并行之后：parallel 的下游收到的是数组，想回到单个对象，就在中间垫一层 .map() 把数组归并成下一步要的形状。这是与 LangGraph 哲学分野的地方：

维度	LangGraph	Mastra
状态载体	全图共享 State 对象	步骤间显式输入输出
并发写	需声明 reducer 合并策略	parallel 自动收进数组
契约检查	主要靠运行时校验	编译期类型推导
改字段影响	全图排查谁读了它	定位相邻两步

LangGraph 是黑板模型：节点读写同一个 State，并发写同一字段要靠 reducer 声明合并策略，字段一多心智负担在"谁动了我的 state"。Mastra 是管道模型：数据穿隧道，每步只声明吃什么吐什么。管道的短板是不适合处处要读的全局变量——1.x 为此补了可选的 workflow state：createWorkflow 里加 stateSchema，execute 里解构 state 与 setState，挂起恢复后保留，嵌套子流程也能继承。默认走管道、确有全局诉求再开 state，这是官方Workflow 文档的顺序，也是我的建议。

注意：Workflow API 在 1.x 线上仍在快速演进，本文以 @mastra/core 1.79 为基准，动手前建议对照官方文档。

4、四大编排原语

then 串行已见，剩下三个原语各给最小代码。

// 条件分支：条件是普通函数，返回 true 的分支执行
reviewFlow.branch([
  [({ approved }) => approved, publishStep],
  [({ approved }) => !approved, rejectStep],
]);

// 并行：所有分支拿到同一份输入，结果按声明顺序收进数组
pipeline.parallel(summarizeStep, translateStep, extractTagsStep);

// 循环：先执行后判断，条件成立即停
pipeline.dountil(reviseStep, (output) => output.approved);

branch 的条件函数收到上一步输出，实战里习惯写成互补对，兜底就不会漏；若所有条件都不命中，这一层不派发任何分支，所以互补条件值得当成纪律来写。parallel 是 fan-out：三个分支同时开跑，下游步骤的 inputSchema 把结果声明为数组即可——不需要 reducer，合并逻辑显式写在下游。dountil 是循环原语，注意它和 dowhile 都是先执行再判断、至少跑一轮；条件函数第二个参数是已循环轮数 loopCount，熔断就靠它，练习会用到。另有两个轻量原语提一句：sleep(ms) 与 sleepUntil(date) 是等待步骤，适合"轮询前先歇五分钟"这类定时场景；.foreach() 对数组输入逐项处理，concurrency 默认 1、调大即并发。完整清单见控制流文档。

5、人机协同中断恢复

审批是 Workflow 相对 Agent 最硬的优势：流程可以挂起，等人操作后原位继续。挂起步骤的定义如下：

const approveStep = createStep({
  id: 'human-approve',
  inputSchema: z.object({ draft: z.string() }),
  outputSchema: z.object({ approved: z.boolean() }),
  execute: async ({ inputData, resumeData, suspend, bail }) => {
    if (!resumeData) {
      // 首轮：挂起，resumeData 为 undefined
      await suspend({ draft: inputData.draft, note: '请审批' });
      return { approved: false }; // 满足类型检查，实际不会执行到
    }
    if (resumeData.reject) {
      await bail(); // 打回：整个 workflow 以 failed 终止
    }
    return { approved: resumeData.approved };
  },
});

suspend 从 execute 上下文解构而来：调用后流程在此冻结，run.start() 正常返回，result.status 为 'suspended'，result.suspendPayload 带着你给审批人看的数据。审批结果用 resume 灌回：

const result = await run.start({ inputData: { draft: '……' } });

if (result.status === 'suspended') {
  // 真实场景：把 suspendPayload 推给审批人，隔天再恢复
  const final = await run.resume({
    resumeAs: 'human-approve', // 只有一个挂起步骤时可省略
    resumeData: { approved: true },
  });
}

恢复后该步骤从头重跑 execute，这次 resumeData 有值，走下半段逻辑。关键认知是：挂起的是运行记录，不是进程——runId 持久化在 storage 里，跨进程、隔天、服务器重启后恢复都行，配好 LibSQL 或 Postgres 存储后端即可长期挂着等。调试也方便：Studio 的 Playground 已支持直接触发挂起并手动恢复，不用先写审批界面就能把人机协同跑通。对照 LangGraph 的 interrupt（Agent 系列第 11 篇讲过）：那边靠 checkpointer 存档加 Command(resume=...) 恢复，Mastra 把同一件事收进 suspend 与 resume 两个原语，语义更窄也更直白。bail() 是拒绝通道：审批人打回，整条流程立刻以 failed 收场。更多模式（多轮交互、按 URL 远程恢复）见人机协同文档。

6、错误处理与重试

step 抛出异常，流程立刻中断：result.status === 'failed'，result.error 带堆栈，默认不重试。瞬时故障想自动扛过去，用 try 配置，写在链上即可：

pipeline
  .then(fetchSourceStep.try({ attempts: 3, delayMs: 500 }))
  .then(summarizeStep)

attempts 是总尝试次数，delayMs 是重试间隔，默认 1 次、不等待；也支持函数形式，按错误对象动态决定次数。等价写法是在 createStep 配置里直接放 tryConfig。要点是重试是编排能力，不侵入业务代码：execute 里不写 while 和退避，换一个流程复用同一个 step 时，重试策略跟着编排走而不是跟着步骤走。除自动重试外还有两类兜底：给 createWorkflow 配 onStepStatusChange 之类的生命周期回调，可以在步骤状态翻转时发告警；Studio 的运行视图会直接标出失败步骤与重试轨迹，排障不用翻日志。另一类"再来一次"是业务级的——生成质量不达标重写——那不是异常，是循环，用 dountil 表达。两类机制别混用，细节见错误处理文档。

7、写作审稿修订实战

Agent 系列第 8 篇用 StateGraph 做过"写作-审稿-修订"循环，同款任务换 Mastra 重写，正好检验前六节。两个 Agent（写手 writer、审稿人 reviewer，定义沿用第 2 篇写法），三个 step，修订加复审打包成子流程循环。先看前两个步骤：

const draftSchema = z.object({ draft: z.string() });
const reviewSchema = z.object({ approved: z.boolean(), comments: z.string() });

const draftStep = createStep({
  id: 'draft',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: draftSchema,
  execute: async ({ inputData }) => {
    const { object } = await writer.generate(
      `就主题「${inputData.topic}」写一段 300 字技术导语`,
      { output: draftSchema },
    );
    return object;
  },
});

const reviewStep = createStep({
  id: 'review',
  inputSchema: draftSchema,
  outputSchema: reviewSchema,
  execute: async ({ inputData }) => {
    const { object } = await reviewer.generate(
      `审阅这段导语能否发布：\n${inputData.draft}`,
      { output: reviewSchema },
    );
    return object;
  },
});

Agent 进 step 就是普通函数调用：generate 配 output 选项拿到结构化的 object（写法见结构化输出文档），返回值恰好就是 step 的 outputSchema——同一份 Zod schema 从模型输出贯穿到流程编排。修订步骤同构，只是输入换成审稿意见：

const reviseStep = createStep({
  id: 'revise',
  inputSchema: reviewSchema,
  outputSchema: draftSchema,
  execute: async ({ inputData }) => {
    const { object } = await writer.generate(
      `按审稿意见重写导语：${inputData.comments}`,
      { output: draftSchema },
    );
    return object;
  },
});

reviseStep 吃 reviewSchema 吐 draftSchema，正好把审稿结论翻译回新的草稿。有了它，循环体和主流程就能组装起来：

// 子流程：按意见修订一轮并复审，输入输出同为 reviewSchema
const reviseOnce = createWorkflow({
  id: 'revise-once',
  inputSchema: reviewSchema,
  outputSchema: reviewSchema,
})
  .then(reviseStep)
  .then(reviewStep)
  .commit();

const writingWorkflow = createWorkflow({
  id: 'writing-review-loop',
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: reviewSchema,
})
  .then(draftStep)
  .then(reviewStep)
  .dountil(reviseOnce, ({ approved }) => approved)
  .commit();

两个细节。其一，workflow 本身可以当 step 用：reviseOnce 嵌进 dountil，循环体是"修订加复审"两步——这正是 LangGraph 里"条件边指回上游节点"的等价写法，差别是 Mastra 要求循环体收敛成一个可复用单元。其二，dountil 先跑再判：首审已通过时也会多修订一轮，演示可接受，生产可在循环前用 branch 拦一道。运行与取结果：

const run = await writingWorkflow.createRun();
const result = await run.start({ inputData: { topic: 'Mastra Workflow' } });

console.log(result.status);  // success / failed / suspended
console.log(result.result);  // 最新一轮审稿结论

长流程不想干等，把 start 换成 run.stream() 即可逐事件拿进度：走到哪个 step、每轮审稿结论如何，边跑边出。对照 Agent 系列第 8 篇：那边是节点加条件边，状态在共享 State 里流转；这边是链加循环原语，数据在 schema 间咬合。行数相近，但这份代码从输入到输出全程有类型兜底——改一个字段名，编辑器立刻告诉你断在哪两步之间。

8、总结

回顾全文：Workflow 管确定性流程、Agent 管开放式决策，两者组合而非替代；createStep 用 Zod 声明输入输出，没有 description，因为不需要模型做选择；数据流是显式管道，上一步输出即下一步输入，map 补桥、stateSchema 兜全局；then、branch、parallel、dountil 覆盖串行、条件、并行、循环；suspend 与 resume 挂起的是运行记录，跨进程恢复不丢；try 让重试跟着编排走。实战把七件事串成了一根全程可类型检查的流水线。

练习：给实战加熔断——最多修订三轮，三轮仍不通过就带最后一轮意见终止。两个提示：dountil 的条件函数第二个参数是 loopCount；若希望终止时同时输出"结论加轮数"，把主流程 outputSchema 换成能携带两者的形状。对于正在把 Agent 能力产品化的工程师，建议挑一段手头最像流水线的胶水代码用 Workflow 重写，再跑 mastra dev 看 Studio 的图视图——哪步慢、哪步挂、哪步在等人，一眼见底。