ステップによる並列実行

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

フロー制御図

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

フロー制御図は次のとおりです：

ステップの作成

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


import { LegacyStep, LegacyWorkflow } from "@mastra/core/workflows/legacy";
import { z } from "zod";
 
const stepOne = new LegacyStep({
  id: "stepOne",
  execute: async ({ context }) => ({
    doubledValue: context.triggerData.inputValue * 2,
  }),
});
 
const stepTwo = new LegacyStep({
  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 LegacyStep({
  id: "stepThree",
  execute: async ({ context }) => ({
    tripledValue: context.triggerData.inputValue * 3,
  }),
});
 
const stepFour = new LegacyStep({
  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 LegacyWorkflow({
  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で例を見る