Workflow 수업

그만큼Workflow클래스를 사용하면 조건부 분기 및 데이터 유효성 검사를 통해 복잡한 작업 시퀀스에 대한 상태 시스템을 만들 수 있습니다.

사용예
사용예에 대한 직접 링크

src/mastra/workflows/test-workflow.ts
import { createWorkflow } from '@mastra/core/workflows'
import { z } from 'zod'

export const workflow = createWorkflow({
  id: 'test-workflow',
  inputSchema: z.object({
    value: z.string(),
  }),
  outputSchema: z.object({
    value: z.string(),
  }),
})

스키마 정의
스키마 정의에 대한 직접 링크

Workflow의 inputSchema와 outputSchema는 Standard JSON Schema를 지원하는 모든 라이브러리로 정의할 수 있습니다. 여기에는 Zod, Valibot, ArkType 등이 포함됩니다.

Zod
Valibot
ArkType

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();

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();

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();

생성자 매개변수
생성자 매개변수에 대한 직접 링크

id:

string

Workflow의 고유 식별자

inputSchema:

StandardJSONSchemaV1

Workflow의 입력 구조를 정의하는 Standard JSON Schema

outputSchema:

StandardJSONSchemaV1

Workflow의 출력 구조를 정의하는 Standard JSON Schema

stateSchema?:

StandardJSONSchemaV1

Workflow 상태를 위한 선택적 Standard JSON Schema입니다. Mastra의 상태 시스템을 사용하면 자동으로 삽입됩니다. 지정하지 않으면 타입은 'any'입니다.

requestContextSchema?:

StandardJSONSchemaV1

요청 컨텍스트 값을 검증하기 위한 Standard JSON Schema입니다. 제공하면 run.start() 시작 시 컨텍스트를 검증하며, 검증에 실패하면 오류가 발생합니다.

schedule?:

WorkflowScheduleConfig | WorkflowScheduleConfig[]

Workflow의 선택적 cron 일정입니다. 단일 설정 또는 여러 주기로 실행할 설정 배열을 받습니다. 이를 설정하면 Workflow가 이벤트 기반 실행 엔진으로 자동 전환됩니다. 사용법은 예약된 Workflow 가이드를 참조하세요.

WorkflowScheduleConfig

id?:

string

일정의 안정적인 식별자입니다. 일정 배열을 전달할 때 필수입니다. 단일 일정 객체를 전달하면 기본값은 Workflow ID입니다.

cron:

string

5개, 6개 또는 7개 부분으로 구성된 cron 표현식입니다. Workflow 생성 시 검증됩니다.

timezone?:

string

IANA 시간대입니다(예: "America/New_York"). 기본값은 호스트의 로컬 시간대입니다. 실행 시간이 서버 로캘에 따라 달라지지 않도록 프로덕션에서는 명시적으로 설정하세요.

inputData?:

TInput

실행될 때마다 Workflow 입력으로 전달되는 페이로드입니다.

initialState?:

TState

실행의 초기 상태입니다.

requestContext?:

Record<string, unknown>

실행에 연결되는 요청 컨텍스트입니다.

metadata?:

Record<string, unknown>

일정 행과 함께 저장되는 임의의 메타데이터입니다.

options?:

WorkflowOptions

Workflow의 선택적 옵션

WorkflowOptions

tracingPolicy?:

TracingPolicy

Workflow의 선택적 추적 정책

validateInputs?:

boolean

Workflow 입력을 검증할지 결정하는 선택적 플래그입니다. Workflow/단계의 입력/재개 데이터에 있는 zodSchemas의 기본값도 적용합니다. 시작/재개 시 입력/재개 데이터 검증에 실패하면 Workflow가 시작/재개되지 않고 대신 오류가 발생합니다. 단계 실행 중 입력 데이터 검증에 실패하면 단계가 실패하고, 이에 따라 Workflow도 실패하며 오류가 반환됩니다.

shouldPersistSnapshot?:

(params: { stepResults: Record<string, StepResult<any, any, any, any>>; workflowStatus: WorkflowRunStatus }) => boolean

Workflow 스냅샷을 저장할지 결정하는 선택적 플래그

pruneSnapshot?:

(params: { snapshot: WorkflowRunState; workflowStatus: WorkflowRunStatus }) => WorkflowRunState

Workflow 스냅샷이 저장되기 직전에 변환하는 선택적 후크입니다. JSON으로 직렬화할 수 있는 데이터를 반환하고 Workflow 재개에 필요한 모든 항목(중단된 단계의 suspendPayloads, suspendedPaths, executionPath 등)을 보존해야 합니다. Agent 실행에서는 스냅샷을 최소화하는 데 내부적으로 사용되며, 사용자 Workflow는 기본적으로 전체 스냅샷을 저장합니다.

onFinish?:

(result: WorkflowFinishCallbackResult) => void | Promise<void>

Workflow가 어떤 상태(success, failed, suspended, tripwire)로든 완료되면 호출되는 콜백입니다. 상태, 출력, 오류 및 단계 결과를 포함한 Workflow 결과를 받습니다. 이 콜백에서 발생한 오류는 포착되어 기록되며 전파되지 않습니다.

WorkflowFinishCallbackResult

status:

WorkflowRunStatus

Workflow 상태: 'success', 'failed', 'suspended' 또는 'tripwire'

result?:

any

Workflow 출력(상태가 'success'일 때)

error?:

SerializedError

오류 세부 정보(상태가 'failed'일 때)

steps:

Record<string, StepResult>

각 단계의 상태와 출력을 포함한 개별 단계 결과

tripwire?:

StepTripwireInfo

트립와이어 정보(상태가 'tripwire'일 때)

runId:

string

이 Workflow 실행의 고유 식별자

workflowId:

string

Workflow의 식별자

resourceId?:

string

선택적 리소스 식별자(실행 생성 시 제공한 경우)

getInitData:

() => any

Workflow에 전달된 초기 입력 데이터를 반환하는 함수

mastra?:

Mastra

Mastra 인스턴스(Workflow가 Mastra에 등록된 경우)

requestContext:

RequestContext

요청 범위의 컨텍스트 데이터

logger:

IMastraLogger

Workflow의 로거 인스턴스

state:

Record<string, any>

Workflow의 현재 상태 객체

onError?:

(errorInfo: WorkflowErrorCallbackInfo) => void | Promise<void>

Workflow가 실패한 경우(failed 또는 tripwire 상태)에만 호출되는 콜백입니다. 오류 세부 정보와 단계 결과를 받습니다. 이 콜백에서 발생한 오류는 포착되어 기록되며 전파되지 않습니다.

WorkflowErrorCallbackInfo

status:

'failed' | 'tripwire'

Workflow 상태('failed' 또는 'tripwire')

error?:

SerializedError

오류 세부 정보

steps:

Record<string, StepResult>

각 단계의 상태와 출력을 포함한 개별 단계 결과

tripwire?:

StepTripwireInfo

트립와이어 정보(상태가 'tripwire'일 때)

runId:

string

이 Workflow 실행의 고유 식별자

workflowId:

string

Workflow의 식별자

resourceId?:

string

선택적 리소스 식별자(실행 생성 시 제공한 경우)

getInitData:

() => any

Workflow에 전달된 초기 입력 데이터를 반환하는 함수

mastra?:

Mastra

Mastra 인스턴스(Workflow가 Mastra에 등록된 경우)

requestContext:

RequestContext

요청 범위의 컨텍스트 데이터

logger:

IMastraLogger

Workflow의 로거 인스턴스

state:

Record<string, any>

Workflow의 현재 상태 객체

초기 상태로 실행 중
초기 상태로 실행 중에 대한 직접 링크

Workflow 실행을 시작할 때 initialState를 전달하여 Workflow 상태의 시작 값을 설정할 수 있습니다.

const run = await workflow.createRun()

const result = await run.start({
  inputData: { value: 'hello' },
  initialState: {
    counter: 0,
    items: [],
  },
})

initialState 객체는 Workflow의 stateSchema에 정의된 구조와 일치해야 합니다. 자세한 내용은 Workflow 상태를 참조하세요.

Workflow 상태
Workflow 상태에 대한 직접 링크

Workflow의 status는 현재 실행 상태를 나타냅니다. 가능한 값은 다음과 같습니다.

success:

string

모든 단계가 유효한 결과 출력과 함께 성공적으로 실행을 완료함

failed:

string

Workflow 실행 중 오류가 발생했으며 오류 세부 정보를 확인할 수 있음

suspended:

string

Workflow 실행이 재개를 기다리며 일시 중지되었고 중단된 단계 정보를 확인할 수 있음

tripwire:

string

프로세서 트립와이어에 의해 Workflow가 종료되었습니다. Workflow 내 Agent 단계가 트립와이어를 트리거할 때 발생합니다(예: 가드레일이 콘텐츠를 차단한 경우). 결과에서 트립와이어 정보를 확인할 수 있습니다.

트립와이어 상태 처리
트립와이어 상태 처리에 대한 직접 링크

Workflow에 트립와이어를 트리거하는 Agent 단계가 포함되어 있으면 Workflow는 status: 'tripwire'를 반환하고 트립와이어 세부 정보를 포함합니다.

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

if (result.status === 'tripwire') {
  console.log('Workflow terminated by tripwire:', result.tripwire?.reason)
  console.log('Processor ID:', result.tripwire?.processorId)
  console.log('Retry requested:', result.tripwire?.retry)
}

이는 예기치 않은 오류를 나타내는 status: 'failed'와 구별됩니다. 트립와이어 상태는 프로세서가 의도적으로 실행을 중단했음을 의미합니다(예: 콘텐츠 조정 목적).

사용예사용예에 대한 직접 링크

스키마 정의스키마 정의에 대한 직접 링크

생성자 매개변수생성자 매개변수에 대한 직접 링크

id:

inputSchema:

outputSchema:

stateSchema?:

requestContextSchema?:

schedule?:

id?:

cron:

timezone?:

inputData?:

initialState?:

requestContext?:

metadata?:

options?:

tracingPolicy?:

validateInputs?:

shouldPersistSnapshot?:

pruneSnapshot?:

onFinish?:

status:

result?:

error?:

steps:

tripwire?:

runId:

workflowId:

resourceId?:

getInitData:

mastra?:

requestContext:

logger:

state:

onError?:

status:

error?:

steps:

tripwire?:

runId:

workflowId:

resourceId?:

getInitData:

mastra?:

requestContext:

logger:

state:

초기 상태로 실행 중초기 상태로 실행 중에 대한 직접 링크

Workflow 상태Workflow 상태에 대한 직접 링크

success:

failed:

suspended:

tripwire:

트립와이어 상태 처리트립와이어 상태 처리에 대한 직접 링크

관련된관련된에 대한 직접 링크

사용예
사용예에 대한 직접 링크

스키마 정의
스키마 정의에 대한 직접 링크

생성자 매개변수
생성자 매개변수에 대한 직접 링크

초기 상태로 실행 중
초기 상태로 실행 중에 대한 직접 링크

Workflow 상태
Workflow 상태에 대한 직접 링크

트립와이어 상태 처리
트립와이어 상태 처리에 대한 직접 링크

관련된
관련된에 대한 직접 링크