ステップを用いた並列実行

AIアプリケーションを構築する際、効率を向上させるために複数の独立したタスクを同時に処理する必要があることがよくあります。

制御フローダイアグラム

この例は、各ブランチが独自のデータフローと依存関係を処理しながら、ステップを並行して実行するワークフローの構造を示しています。

こちらが制御フローダイアグラムです：

ステップの作成

ステップを作成し、ワークフローを初期化しましょう。


import { Step, Workflow } from "@mastra/core/workflows";
import { z } from "zod";
 
const stepOne = new Step({
  id: "stepOne",
  execute: async ({ context }) => ({
    doubledValue: context.triggerData.inputValue * 2,
  }),
});
 
const stepTwo = new Step({
  id: "stepTwo",
  execute: async ({ context }) => {
    if (context.steps.stepOne.status !== "success") {
      return { incrementedValue: 0 }
    }
 
    return { incrementedValue: context.steps.stepOne.output.doubledValue + 1 }
  },
});
 
const stepThree = new Step({
  id: "stepThree",
  execute: async ({ context }) => ({
    tripledValue: context.triggerData.inputValue * 3,
  }),
});
 
const stepFour = new Step({
  id: "stepFour",
  execute: async ({ context }) => {
    if (context.steps.stepThree.status !== "success") {
      return { isEven: false }
    }
 
    return { isEven: context.steps.stepThree.output.tripledValue % 2 === 0 }
  },
});
 
const myWorkflow = new Workflow({
  name: "my-workflow",
  triggerSchema: z.object({
    inputValue: z.number(),
  }),
});

ステップの連鎖と並列化

これで、ワークフローにステップを追加できます。.then() メソッドはステップを連鎖するために使用されますが、.step() メソッドはワークフローにステップを追加するために使用されます。


myWorkflow
  .step(stepOne)
    .then(stepTwo) // chain one
  .step(stepThree)
    .then(stepFour) // chain two
  .commit();
 
const { start } = myWorkflow.createRun();
 
const result = await start({ triggerData: { inputValue: 3 } });

GitHubで例を見る