Step class

Step class 定義 Workflow 內的個別工作單元，並封裝執行邏輯、資料驗證及輸入／輸出處理。 它可接受 Tool 或 Agent 作為參數，並據此自動建立步驟。

使用範例

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  description: 'passes value from input to output',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { value } = inputData
    return {
      value,
    }
  },
})

定義 schema

你可以使用任何支援 Standard JSON Schema 的程式庫，定義步驟的 inputSchema 及 outputSchema，包括 Zod、Valibot 及 ArkType 等程式庫。

Zod

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    formatted: z.string(),
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

const step1 = createStep({
  id: 'step-1',
  inputSchema: toStandardJsonSchema(
    v.object({
      message: v.string(),
    }),
  ),
  outputSchema: toStandardJsonSchema(
    v.object({
      formatted: v.string(),
    }),
  ),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

    return {
      formatted: message.toUpperCase(),
    }
  },
})

從 Agent 建立步驟

你可以直接從 Agent 建立步驟。該步驟會使用 Agent 的名稱作為 ID。

基本 Agent 步驟

src/mastra/workflows/test-workflow.ts

import { testAgent } from '../agents/test-agent'

const agentStep = createStep(testAgent)
// inputSchema: { prompt: string }
// outputSchema: { text: string }

使用結構化輸出的 Agent 步驟

傳入 structuredOutput，讓 Agent 傳回具型別的結構化資料：

src/mastra/workflows/test-workflow.ts

const articleSchema = z.object({
  title: z.string(),
  summary: z.string(),
  tags: z.array(z.string()),
})

const agentStep = createStep(testAgent, {
  structuredOutput: { schema: articleSchema },
})
// inputSchema: { prompt: string }
// outputSchema: { title: string, summary: string, tags: string[] }

Agent 步驟選項

structuredOutput:

{ schema: StandardJSONSchemaV1 }

提供此選項後，Agent 會傳回符合此 schema 的結構化資料，而非純文字。步驟的 outputSchema 會設為所提供的 schema。

onFinish:

(result: AgentResult) => void

Agent 完成生成時呼叫的 callback。

建構函數參數

id:

string

步驟的唯一識別碼

description:

string

說明步驟用途的可選描述

inputSchema:

StandardJSONSchemaV1

定義輸入結構的 Standard JSON Schema

outputSchema:

StandardJSONSchemaV1

定義輸出結構的 Standard JSON Schema

resumeSchema:

StandardJSONSchemaV1

用於恢復步驟的可選 Standard JSON Schema

suspendSchema:

StandardJSONSchemaV1

用於暫停步驟的可選 Standard JSON Schema

stateSchema:

StandardJSONSchemaV1

步驟 state 的可選 Standard JSON Schema。使用 Mastra 的 state 系統時會自動注入。stateSchema 必須是 Workflow stateSchema 的子集。如未指定，型別為 'any'。

requestContextSchema:

StandardJSONSchemaV1

用於驗證 request context 值的 Standard JSON Schema。提供此項後，系統會在步驟的 execute() 執行前驗證 context；如驗證失敗，步驟亦會失敗。

execute:

(params: ExecuteParams) => Promise<any>

包含步驟邏輯的非同步函數

ExecuteParams

inputData:

z.infer<TStepInput>

符合 inputSchema 的輸入資料

resumeData:

z.infer<TResumeSchema>

從暫停狀態恢復步驟時，符合 resumeSchema 的 resume 資料。只在步驟正被恢復時存在。

suspendData:

z.infer<TSuspendSchema>

步驟暫停時原先傳入 suspend() 的 suspend 資料。只在步驟正被恢復，而且之前曾連同資料暫停時存在。

mastra:

Mastra

存取 Mastra 服務（Agent、Tool 等）

getStepResult:

(step: Step | string) => any

存取其他步驟結果的函數

getInitData:

() => any

在任何步驟中存取 Workflow 初始輸入資料的函數

suspend:

(suspendPayload: any, suspendOptions?: { resumeLabel?: string }) => Promise<void>

暫停 Workflow 執行的函數

state:

z.infer<TState>

目前的 Workflow state。包含在所有步驟及 suspend/resume 週期之間持續存在的共用值。其結構由步驟的 stateSchema 定義。

setState:

(state: z.infer<TState>) => void

設定 Workflow state 的函數。透過類似 reducer 的模式注入，例如 'setState({ ...state, ...newState })'

runId:

string

目前的 run ID

requestContext?:

RequestContext

用於依賴注入及 context 資訊的 Request Context。

retryCount?:

number

此特定步驟的重試次數；每次重試步驟時都會自動增加

scorers:

MastraScorers | (({ requestContext }) => MastraScorers | Promise<MastraScorers>)

步驟成功完成後自動執行的 Scorer。每個 Scorer 都會評估步驟本身的輸入及輸出，結果會被儲存並附加至步驟的 trace。請提供 { [name]: { scorer, sampling? } } map，或傳回該 map 的函數。評分會以非同步方式執行，不會阻塞 Workflow。請參閱為步驟輸出評分。

retries:

number

步驟的 execute 函數拋出錯誤時的重試次數。

metadata:

Record<string, any>

用於儲存其他步驟資訊的可選 key-value pair。值必須可序列化（不得包含函數、循環參照等）。

為步驟輸出評分

將 scorers 附加至步驟，即可在該步驟執行時自動評估其輸出，而非只為 Workflow 的最終答案評分。這對多步驟及 RAG Workflow 很有用，因為你可以查看是哪個步驟令品質下降，例如 retrieval 步驟是否在後續步驟進行推理前傳回了相關 chunk。

每個 Scorer 都會接收步驟本身的 input 及 output。步驟成功後，評分會以非同步方式執行，結果則儲存至步驟的 trace。使用 sampling 控制 Scorer 的執行頻率。

以下範例將 Scorer 附加至 retrieval 步驟，令每次執行都會被評分：

src/mastra/workflows/rag-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'
import { retrievalRelevanceScorer } from '../scorers/retrieval-relevance'

const retrievalStep = createStep({
  id: 'retrieval',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ query: z.string(), chunks: z.array(z.string()) }),
  scorers: {
    retrievalRelevance: {
      scorer: retrievalRelevanceScorer(),
      sampling: { type: 'ratio', rate: 1 },
    },
  },
  execute: async ({ inputData }) => {
    const chunks = await retrieve(inputData.query)
    return { query: inputData.query, chunks }
  },
})

將 Scorer 附加至每個要量度的步驟，便可在多步驟 Workflow 中建立逐步評分。由於評分範圍限於單一步驟，因此毋須專用的跨步驟指標，亦可找出品質在哪裏出現變化。

使用 Workflow.agent() 及 Workflow.tool() 加入的 Agent 和 Tool 步驟，其步驟選項亦接受相同的 scorers 選項。

備註

請參閱 Scorer 概覽，了解即時 evaluation 的執行方式及結果的儲存位置；亦可參閱自訂 Scorer，建立自己的 Scorer。

相關內容

• Workflow state
• 控制流程
• 使用 Agent 及 Tool