Gestion des erreurs

Les workflows Mastra permettent de gérer les erreurs grâce à la vérification du statut du résultat après l’exécution, à des politiques de nouvelle tentative pour les défaillances temporaires et à des fonctions de rappel du cycle de vie pour centraliser la journalisation des erreurs ou les alertes.

Gérer les résultats d’un workflow

Lorsque vous exécutez un workflow, l’objet de résultat contient son statut ainsi que les éventuelles erreurs survenues.

Vérifier le statut du résultat

src/run-workflow.ts

import { mastra } from './mastra'

const workflow = mastra.getWorkflow('myWorkflow')
const run = await workflow.createRun()
const result = await run.start({ inputData: { value: 'test' } })

switch (result.status) {
  case 'success':
    console.log('Workflow completed:', result.result)
    break
  case 'failed':
    console.error('Workflow failed:', result.error)
    break
  case 'suspended':
    console.log('Workflow suspended, waiting for resume')
    break
}

Structure de l’objet de résultat

L’objet de résultat contient les éléments suivants :

• status - Le statut du workflow : 'success', 'failed', 'suspended' ou 'tripwire'
• result - La sortie du workflow (lorsque le statut est 'success')
• error - Les détails de l’erreur (lorsque le statut est 'failed')
• steps - Les résultats de chaque étape, avec leur statut et leur sortie

Accéder aux résultats des étapes

Vous pouvez examiner le résultat de chaque étape pour déterminer où une défaillance s’est produite :

src/run-workflow.ts

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

if (result.status === 'failed') {
  // Find which step failed
  for (const [stepId, stepResult] of Object.entries(result.steps)) {
    if (stepResult.status === 'failed') {
      console.error(`Step ${stepId} failed:`, stepResult.error)
    }
  }
}

Fonctions de rappel du cycle de vie

Lorsque vous devez gérer la fin d’un workflow sans attendre son résultat, par exemple pour des tâches en arrière-plan, des workflows lancés sans attendre leur résultat ou une journalisation centralisée, vous pouvez utiliser des fonctions de rappel du cycle de vie.

onFinish

Appelé lorsqu’un workflow se termine, quel que soit son statut (succès, échec, suspension ou tripwire) :

src/mastra/workflows/order-workflow.ts

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

const orderWorkflow = createWorkflow({
  id: 'order-processing',
  inputSchema: z.object({ orderId: z.string() }),
  outputSchema: z.object({ orderId: z.string(), status: z.string() }),
  options: {
    onFinish: async result => {
      if (result.status === 'success') {
        await db.updateOrderStatus(result.result.orderId, result.status)
      }
      await analytics.track('workflow_completed', {
        workflowId: 'order-processing',
        status: result.status,
      })
    },
  },
})

La fonction de rappel onFinish reçoit :

• status - Le statut du workflow
• result - La sortie du workflow (en cas de succès)
• error - Les détails de l’erreur (en cas d’échec)
• steps - Les résultats de chaque étape
• tripwire - Les informations sur le tripwire (si le statut est 'tripwire')
• runId - L’identifiant unique de cette exécution du workflow
• workflowId - L’identifiant du workflow
• resourceId - L’identifiant facultatif de la ressource (s’il a été fourni lors de la création de l’exécution)
• getInitData<any>() - La fonction qui renvoie les données d’entrée initiales
• mastra - L’instance Mastra (si le workflow est enregistré auprès de Mastra)
• requestContext - Les données de contexte propres à la requête
• logger - L’instance de journalisation du workflow
• state - L’objet représentant l’état actuel du workflow

onError

Appelé uniquement lorsqu’un workflow échoue (le statut est 'failed' ou 'tripwire') :

src/mastra/workflows/payment-workflow.ts

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

const paymentWorkflow = createWorkflow({
  id: 'payment-processing',
  inputSchema: z.object({ amount: z.number() }),
  outputSchema: z.object({ transactionId: z.string() }),
  options: {
    onError: async errorInfo => {
      await alertService.notify({
        channel: 'payments-alerts',
        message: `Payment workflow failed: ${errorInfo.error?.message}`,
      })
      await errorTracker.capture(errorInfo.error)
    },
  },
})

La fonction de rappel onError reçoit :

• status - Soit 'failed', soit 'tripwire'
• error - Les détails de l’erreur
• steps - Les résultats de chaque étape
• tripwire - Les informations sur le tripwire (si le statut est 'tripwire')
• runId - L’identifiant unique de cette exécution du workflow
• workflowId - L’identifiant du workflow
• resourceId - L’identifiant facultatif de la ressource (s’il a été fourni lors de la création de l’exécution)
• getInitData<any>() - La fonction qui renvoie les données d’entrée initiales
• mastra - L’instance Mastra (si le workflow est enregistré auprès de Mastra)
• requestContext - Les données de contexte propres à la requête
• logger - L’instance de journalisation du workflow
• state - L’objet représentant l’état actuel du workflow

Utiliser les deux fonctions de rappel

Vous pouvez utiliser les deux fonctions de rappel conjointement :

src/mastra/workflows/pipeline-workflow.ts

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

const pipelineWorkflow = createWorkflow({
  id: 'data-pipeline',
  inputSchema: z.object({ source: z.string() }),
  outputSchema: z.object({ recordsProcessed: z.number() }),
  options: {
    onFinish: async result => {
      // Always log completion
      await logger.info('Pipeline completed', { status: result.status })
    },
    onError: async errorInfo => {
      // Alert on failures
      await pagerDuty.alert('Data pipeline failed', errorInfo.error)
    },
  },
})

Gestion des erreurs dans les fonctions de rappel

Les erreurs levées dans les fonctions de rappel sont interceptées et journalisées. Elles n’affectent pas le résultat du workflow et ne provoquent pas son échec. Ainsi, les problèmes rencontrés dans ces fonctions n’interrompent pas vos workflows en production.

options: {
  onFinish: async (result) => {
    // If this throws, it's logged but the workflow result is unchanged
    await externalService.notify(result);
  },
}

Nouvelles tentatives

Mastra propose un mécanisme de nouvelle tentative pour les workflows ou les étapes qui échouent à cause d’erreurs temporaires, par exemple lorsque des étapes interagissent avec des services externes ou des ressources susceptibles d’être momentanément indisponibles.

Au niveau du workflow avec retryConfig

Vous pouvez configurer les nouvelles tentatives au niveau du workflow. Cette configuration s’applique alors à toutes ses étapes :

src/mastra/workflows/test-workflow.ts

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

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

export const testWorkflow = createWorkflow({
  retryConfig: {
    attempts: 5,
    delay: 2000
  }
})
  .then(step1)
  .commit();

Au niveau d’une étape avec retries

Vous pouvez configurer les nouvelles tentatives de chaque étape à l’aide de la propriété retries. Pour l’étape concernée, cette propriété remplace la configuration des nouvelles tentatives définie au niveau du workflow :

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  execute: async () => {
    const response = await fetch('example-url')

    if (!response.ok) {
      throw new Error('Error')
    }

    return {
      value: '',
    }
  },
  retries: 3,
})

Branchement conditionnel

Vous pouvez utiliser une logique conditionnelle pour créer d’autres chemins d’exécution selon la réussite ou l’échec des étapes précédentes :

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  execute: async () => {
    try {
      const response = await fetch('example-url');

      if (!response.ok) {
        throw new Error('error');
      }

      return {
        status: "ok"
      };
    } catch (error) {
      return {
        status: "error"
      };
    }
  }
});

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

export const testWorkflow = createWorkflow({})
  .then(step1)
  .branch([
    [async ({ inputData: { status } }) => status === "ok", step2],
    [async ({ inputData: { status } }) => status === "error", fallback]
  ])
  .commit();

Vérifier les résultats des étapes précédentes

Utilisez getStepResult() pour examiner les résultats d’une étape précédente.

src/mastra/workflows/test-workflow.ts

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

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

const step2 = createStep({
  execute: async ({ getStepResult }) => {
    const step1Result = getStepResult(step1);

    return {
      value: ""
    };
  }
});

Quitter prématurément avec bail()

Utilisez bail() dans une étape pour quitter prématurément avec un résultat réussi. La charge utile fournie est alors renvoyée comme sortie de l’étape, et l’exécution du workflow prend fin.

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step1',
  execute: async ({ bail }) => {
    return bail({ result: 'bailed' });
  },
  inputSchema: z.object({ value: z.string() }),
  outputSchema: z.object({ result: z.string() }),
});

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

Quitter prématurément avec Error()

Utilisez throw new Error() dans une étape pour quitter avec une erreur.

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step1',
  execute: async () => {
    throw new Error('error');
  },
  inputSchema: z.object({ value: z.string() }),
  outputSchema: z.object({ result: z.string() }),
});

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

## Surveiller les erreurs avec stream()

Vous pouvez détecter les erreurs des workflows à l’aide de stream :

src/test-workflow.ts

import { mastra } from '../src/mastra'

const workflow = mastra.getWorkflow('testWorkflow')

const run = await workflow.createRun()

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

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

Voir aussi

• Flux de contrôle
• Suspendre et reprendre
• Voyage dans le temps
• Intervention humaine