État du Workflow

L'état du Workflow permet de partager des valeurs entre les étapes sans les faire transiter par les schémas inputSchema et outputSchema de chacune d'elles. Il est utile pour suivre l'avancement, accumuler des résultats ou partager une configuration dans l'ensemble du Workflow.

État et entrées/sorties des étapes

Il est important de comprendre la différence entre l'état et les entrées/sorties des étapes :

• Entrées/sorties des étapes : les données circulent séquentiellement entre les étapes. Chaque étape reçoit la sortie de l'étape précédente dans inputData et renvoie une sortie destinée à l'étape suivante.
• État : un espace de stockage partagé que toutes les étapes peuvent lire et mettre à jour au moyen de state et setState. L'état persiste pendant toute l'exécution du Workflow, y compris au cours des cycles de suspension et de reprise.

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

Définir les schémas d'état

Définissez un stateSchema à la fois sur le Workflow et sur chaque étape. Le stateSchema du Workflow est le schéma principal qui contient toutes les valeurs d'état possibles, tandis que chaque étape ne déclare que le sous-ensemble dont elle a besoin :

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

Définir l'état initial

Transmettez initialState au démarrage d'une exécution de Workflow pour définir les valeurs initiales :

const run = await workflow.createRun()

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

L'objet initialState doit respecter la structure définie dans le stateSchema du Workflow.

Persistance de l'état entre suspension et reprise

L'état persiste automatiquement au fil des cycles de suspension et de reprise. Lorsqu'un Workflow est suspendu puis repris, toutes les mises à jour de l'état effectuées avant la suspension sont conservées :

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

État dans les Workflows imbriqués

Avec des Workflows imbriqués, l'état se propage du parent à l'enfant. Les modifications apportées par le Workflow parent avant l'appel d'un Workflow imbriqué sont visibles par les étapes de ce dernier :

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

Voir aussi

• Présentation des Workflows
• Suspendre et reprendre
• Classe Step
• Classe Workflow