Workflow 개요

Workflow를 사용하면 단일 Agent의 추론에 의존하지 않고 명확하고 구조화된 단계를 사용하여 복잡한 작업 순서를 정의할 수 있습니다. 작업이 분할되는 방식, 작업 간에 데이터가 이동하는 방식, 그리고 언제 실행될 것인지에 대한 완전한 제어권을 제공합니다. Workflow는 기본적으로 내장된 실행 엔진을 사용하여 실행되거나 다음 위치에 배포될 수 있습니다.workflow runners관리형 인프라를 위한 Ingest와 같습니다.

AI Agent의 경우:달리다npx mastra api workflow run start test-workflow '{"inputData":{"message":"Hello, world!"}}' 하여 임시 스크립트를 작성하는 대신 이 Workflow를 직접 시작합니다. 이 명령을 사용하려면 Mastra 서버가 실행 중이어야 합니다. 다음을 사용하여 로컬 서버를 시작하세요: npx mastra dev, 또는 연결 가능한 서버의 기본 URL을 다음과 함께 전달합니다.--url. 달리다npx mastra api workflow run start --schema다른 입력을 구성하기 전에. 마스트라의 스킬을 설치하세요.npx skills add mastra-ai/skills --skill mastra완전한 API CLI 검색, 타겟팅, 스키마, 인증 및 오류 처리 지침을 제공합니다.

Workflow를 사용해야 하는 경우

명확하게 정의되어 있고 특정 실행 순서가 있는 여러 단계가 포함된 작업에 대해 Workflow를 사용하세요. 이를 통해 단계 간 데이터 흐름 및 변환 방식과 각 단계에서 호출되는 기본 요소를 세밀하게 제어할 수 있습니다.

핵심 원칙

Mastra Workflow는 다음 원칙을 사용하여 작동합니다.

• 정의steps with createStep, specifying input/output schemas and business logic.
• 식자steps with createWorkflow to define the execution flow.
• 달리기workflows 하여 전체 시퀀스를 실행하며, 일시 중단, 재개, 결과 스트리밍을 기본적으로 지원합니다.

Workflow 단계 만들기

단계는 Workflow의 구성 요소입니다. 다음을 사용하여 단계를 만듭니다.createStep() with inputSchema and outputSchema 하여 허용하고 반환하는 데이터를 정의합니다. 두 스키마 모두 Standard JSON Schema (Zod, Valibot, ArkType, etc.).

그만큼execute 함수는 단계가 수행하는 작업을 정의합니다. 이 함수를 사용하여 코드베이스의 함수, 외부 APIs, 에이전트 또는 도구를 호출합니다.

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(),
    }
  },
})

방문하다Step for a full list of configuration options.

Agent 및 Tool 사용

Workflow 단계에서는 등록된 Agent를 호출하거나 Tool을 직접 가져오고 실행할 수도 있습니다.Using Tools page for more information.

Workflow 만들기

다음을 사용하여 Workflow를 만듭니다.createWorkflow() with inputSchema and outputSchema 하여 허용하고 반환하는 데이터를 정의합니다. 두 스키마 모두 Standard JSON Schema (Zod, Valibot, ArkType, etc.). Add steps using .then() and complete the workflow with .commit().

Zod

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    output: z.string()
  })
})
  .then(step1)
  .commit();

Valibot

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import * as v from "valibot";
import { toStandardJsonSchema } from "@valibot/to-json-schema";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: toStandardJsonSchema(v.object({
    message: v.string()
  })),
  outputSchema: toStandardJsonSchema(v.object({
    output: v.string()
  }))
})
  .then(step1)
  .commit();

ArkType

src/mastra/workflows/test-workflow.ts

import { createWorkflow, createStep } from "@mastra/core/workflows";
import { type } from "arktype";

const step1 = createStep({...});

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: type({
    message: "string"
  }),
  outputSchema: type({
    output: "string"
  })
})
  .then(step1)
  .commit();

방문하다Workflow Class for a full list of configuration options.

제어 흐름 이해

Workflow는 다양한 방법을 사용하여 구성할 수 있습니다. 선택한 방법에 따라 각 단계의 스키마 구성 방식이 결정됩니다. 방문Control Flow page for more information.

사진관

열려 있는Studio and select a workflow from the Workflows tab.

• 그래프 보기: 중앙 패널에서는 Workflow의 단계와 실행 흐름을 시각화합니다.
• 입력 양식: 오른쪽 사이드바는 Workflow의 양식을 생성합니다.inputSchema. Fill it in and start the run.
• 실시간 상태: 실행 중에 그래프는 각 단계의 상태를 실시간으로 업데이트합니다. 사이드바에는 Workflow의 입력, 출력, 상태 및 로그가 표시됩니다.
• 시간여행: 실행이 완료된 후 개별 단계를 재생하여 검사하거나 다시 시도합니다.

Workflow 상태

Workflow 상태를 사용하면 모든 단계의 inputSchema 및 OutputSchema를 통과하지 않고도 여러 단계에서 값을 공유할 수 있습니다. 상태를 사용하여 진행 상황을 추적하거나 결과를 누적하거나 전체 Workflow에서 구성을 공유할 수 있습니다.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ message: z.string() }),
  outputSchema: z.object({ formatted: z.string() }),
  stateSchema: z.object({ counter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // Read from state
    console.log(state.counter)

    // Update state for subsequent steps
    setState({ ...state, counter: state.counter + 1 })

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

방문하다Workflow State 에서 상태 스키마, 초기 상태, 일시 중단/재개 간 영속성, 중첩 Workflow에 관한 전체 문서를 확인하세요.

단계로서의 Workflow

Workflow를 단계로 사용하여 더 큰 컴포지션 내에서 해당 논리를 재사용합니다. 입력과 출력은 다음에 설명된 것과 동일한 스키마 규칙을 따릅니다.Core principles.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({...});
const step2 = createStep({...});

const childWorkflow = createWorkflow({
  id: "child-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(step1)
  .then(step2)
  .commit();

export const testWorkflow = createWorkflow({
  id: "test-workflow",
  inputSchema: z.object({
    message: z.string()
  }),
  outputSchema: z.object({
    emphasized: z.string()
  })
})
  .then(childWorkflow)
  .commit();

Workflow 복제

다음을 사용하여 Workflow 복제cloneWorkflow() 하여 해당 로직을 재사용하면서 새 ID로 별도 추적할 수 있습니다. 각 복제본은 독립적으로 실행되며 로그와 Observability 도구에 별개의 Workflow로 표시됩니다.

src/mastra/workflows/test-workflow.ts

import { cloneWorkflow } from "@mastra/core/workflows";

const step1 = createStep({...});

const parentWorkflow = createWorkflow({...})
const clonedWorkflow = cloneWorkflow(parentWorkflow, { id: "cloned-workflow" });

export const testWorkflow = createWorkflow({...})
  .then(step1)
  .then(clonedWorkflow)
  .commit();

Workflow 등록

애플리케이션 전체에서 사용할 수 있도록 Mastra 인스턴스에 Workflow를 등록하세요. 등록되면 Agent나 Tool에서 호출할 수 있으며 로깅 및 관찰 기능과 같은 공유 리소스에 액세스할 수 있습니다.

src/mastra/index.ts

import { Mastra } from '@mastra/core/mastra'
import { testWorkflow } from './workflows/test-workflow'

export const mastra = new Mastra({
  workflows: { testWorkflow },
})

Workflow 참조

Agent, Tool, Mastra Client 또는 명령줄에서 Workflow를 실행할 수 있습니다. 전화로 참조를 얻으십시오.getWorkflow() on your mastra or mastraClient instance, depending on your setup:

const testWorkflow = mastra.getWorkflow('testWorkflow')

:::infomastra.getWorkflow() 가 직접 가져오기보다 권장되는 이유는 두 가지입니다:

1. Mastra 인스턴스 구성(로거, 원격 측정, 스토리지, 등록된 Agent 및 벡터 저장소)에 대한 액세스를 제공합니다.
2. Workflow 입력 및 출력 스키마에 대한 전체 TypeScript 유형 추론을 제공합니다.

사용getWorkflow() with the workflow's registration key (the key used when adding it to Mastra). While getWorkflowById() is available for retrieving workflows by their id 속성은 동일한 수준의 타입 추론을 제공하지 않습니다. :::

Workflow 실행

Workflow는 두 가지 모드로 실행될 수 있습니다. 시작은 반환하기 전에 모든 단계가 완료될 때까지 기다리는 반면, 스트림은 실행 중에 이벤트를 내보냅니다. 사용 사례에 맞는 접근 방식을 선택하세요. 최종 결과만 필요할 때 시작하고 진행 상황을 모니터링하거나 단계가 완료될 때 작업을 트리거하려는 경우 스트리밍하세요.

.start()

다음을 사용하여 Workflow 실행 인스턴스를 만듭니다.createRun(), then call .start() with inputData matching the workflow's inputSchema. Workflow는 모든 단계를 실행하고 최종 결과를 반환합니다.

const run = await testWorkflow.createRun()

const result = await run.start({
  inputData: {
    message: 'Hello world',
  },
})

if (result.status === 'success') {
  console.log(result.result)
}

.stream()

다음을 사용하여 Workflow 실행 인스턴스를 만듭니다..createRun(), then call .stream() with inputData matching the workflow's inputSchema. Iterate over fullStream to track progress, then await result to get the final workflow result.

const run = await testWorkflow.createRun()

const stream = run.stream({
  inputData: {
    message: 'Hello world',
  },
})

for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get the final result (same type as run.start())
const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Workflow 결과 유형

둘 다run.start() and stream.result return a discriminated union based on the status property, which can be success, failed, suspended, tripwire, or paused. You can always safely access result.status, result.input, result.steps, and optionally result.state regardless of the status.

또한 상태에 따라 다양한 속성을 사용할 수 있습니다.

상태	고유한 속성	설명
success	result	The workflow's output data
failed	error	The error that caused the failure
tripwire	tripwire	Contains reason, retry?, metadata?, processorId?
suspended	suspendPayload, suspended	일시 중단 데이터 및 일시 중단된 단계 경로의 배열
paused	(none)	Only common properties available

상태별 속성에 액세스하려면status first:

const result = await run.start({ inputData: { message: 'Hello world' } })

if (result.status === 'success') {
  console.log(result.result) // Only available when status is "success"
} else if (result.status === 'failed') {
  console.log(result.error.message)
} else if (result.status === 'suspended') {
  console.log(result.suspendPayload)
}

Workflow 출력

다음은 성공적인 Workflow 결과의 예입니다.input, steps, and result properties:

{
  "status": "success",
  "steps": {
    "step-1": {
      "status": "success",
      "payload": {
        "message": "Hello world"
      },
      "output": {
        "formatted": "HELLO WORLD"
      }
    },
    "step-2": {
      "status": "success",
      "payload": {
        "formatted": "HELLO WORLD"
      },
      "output": {
        "emphasized": "HELLO WORLD!!!"
      }
    }
  },
  "input": {
    "message": "Hello world"
  },
  "result": {
    "emphasized": "HELLO WORLD!!!"
  }
}

스트리밍

일반용writer API usage, see Streaming.

Workflow 스트림 페이로드 검사

스트림에 기록된 이벤트는 방출된 청크에 포함됩니다. 이벤트 유형, 중간 값 또는 단계별 데이터와 같은 사용자 정의 필드에 액세스하려면 이러한 청크를 검사하세요.

const testWorkflow = mastra.getWorkflow('testWorkflow')

const run = await testWorkflow.createRun()

const stream = await run.stream({
  inputData: {
    value: 'initial data',
  },
})

for await (const chunk of stream) {
  console.log(chunk)
}

if (result!.status === 'suspended') {
  // if the workflow is suspended, we can resume it with the resumeStream method
  const resumedStream = await run.resumeStream({
    resumeData: { value: 'resume data' },
  })

  for await (const chunk of resumedStream) {
    console.log(chunk)
  }
}

중단된 Workflow 스트림 재개

어떤 이유로든 Workflow 스트림이 닫히거나 중단된 경우 다음을 사용하여 재개할 수 있습니다.resumeStream method. This will return a new ReadableStream 를 사용하여 Workflow 이벤트를 관찰할 수 있습니다.

const newStream = await run.resumeStream()

for await (const chunk of newStream) {
  console.log(chunk)
}

Agent를 사용하는 Workflow

Agent의 파이프textStream to the workflow step's writer. 이는 부분 출력을 스트리밍하며, Mastra는 에이전트의 사용량을 Workflow 실행에 자동으로 집계합니다.

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

export const testStep = createStep({
  execute: async ({ inputData, mastra, writer }) => {
    const { city } = inputData

    const testAgent = mastra?.getAgent('testAgent')
    const stream = await testAgent?.stream(`What is the weather in ${city}?`)

    await stream!.textStream.pipeTo(writer!)

    return {
      value: await stream!.text,
    }
  },
})

활성 Workflow 실행을 다시 시작하는 중

Workflow 실행에서 서버와의 연결이 끊어지면 마지막 활성 단계에서 다시 시작할 수 있습니다. 이는 서버 연결이 끊어졌을 때 계속 실행될 수 있는 장기 실행 Workflow에 유용합니다. Workflow 실행을 다시 시작하면 마지막 활성 단계에서 실행이 다시 시작되고 Workflow도 거기에서 계속됩니다.

다음을 사용하여 Workflow의 모든 활성 Workflow 실행을 다시 시작합니다.restartAllActiveWorkflowRuns()

사용restartAllActiveWorkflowRuns() 하여 Workflow의 활성 실행을 모두 다시 시작합니다. 이를 사용하면 각 실행을 수동으로 순회하며 다시 시작할 필요 없이 Workflow의 모든 활성 실행을 다시 시작할 수 있습니다.

workflow.restartAllActiveWorkflowRuns()

다음을 사용하여 활성 Workflow 실행을 다시 시작합니다.restart()

사용restart() 하여 마지막 활성 단계부터 활성 Workflow 실행을 다시 시작합니다. 실행이 마지막 활성 단계부터 재개되고 Workflow는 그 지점부터 계속됩니다.

const run = await workflow.createRun()

const result = await run.start({ inputData: { value: 'initial data' } })

const restartedResult = await run.restart()

활성 Workflow 실행 식별

Workflow 실행이 활성화되면status of running or waiting. You can check the workflow's status to confirm it's active, and use active to identify the active workflow run.

const activeRuns = await workflow.listActiveWorkflowRuns()
if (activeRuns.runs.length > 0) {
  console.log(activeRuns.runs)
}

노트

로컬 마스터 서버를 실행하는 경우 서버가 시작되면 모든 활성 Workflow 실행이 자동으로 다시 시작됩니다.

사용RequestContext

사용RequestContext 하여 요청별 값에 접근합니다. 이를 통해 요청의 맥락에 따라 동작을 조건부로 조정할 수 있습니다.

src/mastra/workflows/test-workflow.ts

export type UserTier = {
  'user-tier': 'enterprise' | 'pro'
}

const step1 = createStep({
  execute: async ({ requestContext }) => {
    const userTier = requestContext.get('user-tier') as UserTier['user-tier']

    const maxResults = userTier === 'enterprise' ? 1000 : 50

    return { maxResults }
  },
})

방문하다Request Context for more information.

팁

유형이 안전한 요청 컨텍스트 스키마 유효성 검사에 대해서는 다음을 참조하세요.Schema Validation.

관련된

Workflow를 자세히 살펴보려면 다음을 참조하세요.Workflow Guide. 여기서는 실용적인 예제를 통해 핵심 개념을 단계별로 살펴봅니다.

• Workflow 상태
• 제어 흐름
• 일시중단 및 재개
• 오류 처리
• 노동자: 전용 백그라운드 프로세스에서 Workflow 실행 실행
• 📹 Agentic workflows with Mastra workshop