Workflow 상태

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

상태 대 단계 입력/출력

상태와 단계 입력/출력의 차이를 이해하는 것이 중요합니다.

• 단계 입력/출력: 데이터는 단계 사이를 순차적으로 흐릅니다. 각 단계는 이전 단계의 출력을 자신의 inputData로 받고 다음 단계를 위한 출력을 반환합니다.
• 상태: 모든 단계에서 state와 setState를 통해 읽고 업데이트할 수 있는 공유 저장소입니다. 상태는 일시 중지/재개 주기를 포함해 전체 Workflow 실행 동안 유지됩니다.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({ workflowInput: z.string() }),
  outputSchema: z.object({ step1Output: z.string() }),
  stateSchema: z.object({ sharedCounter: z.number() }),
  execute: async ({ inputData, state, setState }) => {
    // inputData comes from workflow input or previous step's output
    console.log(inputData.workflowInput)

    // state is the shared workflow state
    console.log(state.sharedCounter)

    // Update state for subsequent steps
    await setState({ sharedCounter: state.sharedCounter + 1 })

    // Return output that flows to next step's inputData
    return { step1Output: 'processed' }
  },
})

상태 스키마 정의

Workflow와 개별 단계 모두에 stateSchema를 정의하세요. Workflow의 stateSchema는 가능한 모든 상태 값을 포함하는 마스터 스키마이며, 각 단계는 필요한 하위 집합만 선언합니다.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  stateSchema: z.object({
    processedItems: z.array(z.string()),
  }),
  execute: async ({ inputData, state, setState }) => {
    const { message } = inputData
    const { processedItems } = state

    await setState({
      processedItems: [...processedItems, 'item-1', 'item-2'],
    })

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

const step2 = createStep({
  stateSchema: z.object({
    metadata: z.object({
      processedBy: z.string(),
    }),
  }),
  execute: async ({ inputData, state }) => {
    const { formatted } = inputData
    const { metadata } = state

    return {
      emphasized: `${formatted}!! ${metadata.processedBy}`,
    }
  },
})

export const testWorkflow = createWorkflow({
  stateSchema: z.object({
    processedItems: z.array(z.string()),
    metadata: z.object({
      processedBy: z.string(),
    }),
  }),
})
  .then(step1)
  .then(step2)
  .commit()

초기 상태 설정

Workflow 실행을 시작할 때 initialState를 전달해 시작 값을 설정하세요.

const run = await workflow.createRun()

const result = await run.start({
  inputData: { message: 'Hello' },
  initialState: {
    processedItems: [],
    metadata: { processedBy: 'system' },
  },
})

initialState 객체는 Workflow의 stateSchema에 정의된 구조와 일치해야 합니다.

일시 중지/재개 전반에 걸쳐 상태 지속성

상태는 일시 중지 및 재개 주기 전반에 걸쳐 자동으로 유지됩니다. Workflow가 일시 중단되었다가 나중에 다시 시작되면 일시 중단 전에 수행된 모든 상태 업데이트가 유지됩니다.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({}),
  outputSchema: z.object({}),
  stateSchema: z.object({ count: z.number(), items: z.array(z.string()) }),
  resumeSchema: z.object({ proceed: z.boolean() }),
  execute: async ({ state, setState, suspend, resumeData }) => {
    if (!resumeData) {
      // First run: update state and suspend
      await setState({ count: state.count + 1, items: [...state.items, 'item-1'] })
      await suspend({})
      return {}
    }
    // After resume: state changes are preserved (count: 1, items: ["item-1"])
    return {}
  },
})

중첩된 Workflow의 상태

중첩된 Workflow를 사용하면 상태가 상위에서 하위로 전파됩니다. 중첩된 Workflow를 호출하기 전에 상위 Workflow에서 변경한 내용은 중첩된 Workflow 내부의 단계에 표시됩니다.

src/mastra/workflows/test-workflow.ts

const nestedStep = createStep({
  id: 'nested-step',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
  execute: async ({ state }) => {
    // Receives state modified by parent workflow
    return { result: `Received: ${state.sharedValue}` }
  },
})

const nestedWorkflow = createWorkflow({
  id: 'nested-workflow',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
})
  .then(nestedStep)
  .commit()

const parentStep = createStep({
  id: 'parent-step',
  inputSchema: z.object({}),
  outputSchema: z.object({}),
  stateSchema: z.object({ sharedValue: z.string() }),
  execute: async ({ state, setState }) => {
    // Modify state before nested workflow runs
    await setState({ sharedValue: 'modified-by-parent' })
    return {}
  },
})

const parentWorkflow = createWorkflow({
  id: 'parent-workflow',
  inputSchema: z.object({}),
  outputSchema: z.object({ result: z.string() }),
  stateSchema: z.object({ sharedValue: z.string() }),
})
  .then(parentStep)
  .then(nestedWorkflow)
  .commit()

관련된

• Workflow 개요
• 일시중단 및 재개
• 스텝 클래스
• Workflow 클래스