Classe Workflow
La classe Workflow permet de créer des machines à états pour des séquences complexes d’opérations avec branchement conditionnel et validation des données.
Exemple d’utilisationLien direct vers Exemple d’utilisation
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(),
}),
})
Définir des schémasLien direct vers Définir des schémas
Vous pouvez définir les inputSchema et outputSchema du Workflow avec toute bibliothèque prenant en charge Standard JSON Schema, notamment Zod, Valibot et 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();
Paramètres du constructeurLien direct vers Paramètres du constructeur
id:
string
Identifiant unique du Workflow
inputSchema:
StandardJSONSchemaV1
Standard JSON Schema définissant la structure d’entrée du Workflow
outputSchema:
StandardJSONSchemaV1
Standard JSON Schema définissant la structure de sortie du Workflow
stateSchema?:
StandardJSONSchemaV1
Standard JSON Schema facultatif pour l’état du Workflow. Injecté automatiquement lors de l’utilisation du système d’état de Mastra. S’il n’est pas spécifié, le type est 'any'.
requestContextSchema?:
StandardJSONSchemaV1
Standard JSON Schema pour valider les valeurs du contexte de requête. Lorsqu’il est fourni, le contexte est validé au début de run.start() et une erreur est levée si la validation échoue.
schedule?:
WorkflowScheduleConfig | WorkflowScheduleConfig[]
Planification cron facultative pour le Workflow. Accepte une seule configuration ou un tableau de configurations pour des déclenchements à plusieurs cadences. Sa définition promeut automatiquement le Workflow vers le moteur d’exécution événementiel. Consultez le guide des Workflows planifiés pour l’utilisation.
WorkflowScheduleConfig
id?:
string
Identifiant stable de la planification. Obligatoire lors de la transmission d’un tableau de planifications. Par défaut, l’ID du Workflow est utilisé lorsqu’un seul objet de planification est transmis.
cron:
string
Expression cron à 5, 6 ou 7 parties. Validée lors de la construction du Workflow.
timezone?:
string
Fuseau horaire IANA, par exemple "America/New_York". Par défaut, le fuseau horaire local de l’hôte est utilisé. Définissez-le explicitement en production pour que les heures de déclenchement ne dépendent pas des paramètres régionaux du serveur.
inputData?:
TInput
Payload transmis comme entrée du Workflow à chaque déclenchement.
initialState?:
TState
État initial de l’exécution.
requestContext?:
Record<string, unknown>
Contexte de requête associé à l’exécution.
metadata?:
Record<string, unknown>
Métadonnées arbitraires conservées avec la ligne de planification.
options?:
WorkflowOptions
Options facultatives du Workflow
WorkflowOptions
tracingPolicy?:
TracingPolicy
Politique de traçage facultative pour le Workflow
validateInputs?:
boolean
Indicateur facultatif déterminant s’il faut valider les entrées du Workflow. Il applique également les valeurs par défaut des zodSchemas aux données d’entrée et de reprise du Workflow ou de l’étape. Si la validation des données d’entrée ou de reprise échoue lors du démarrage ou de la reprise, le Workflow ne démarre ni ne reprend et une erreur est levée. Si la validation des données d’entrée échoue pendant l’exécution d’une étape, l’étape échoue, ce qui fait échouer le Workflow et renvoie l’erreur.
shouldPersistSnapshot?:
(params: { stepResults: Record<string, StepResult<any, any, any, any>>; workflowStatus: WorkflowRunStatus }) => boolean
Indicateur facultatif déterminant s’il faut conserver l’instantané du Workflow
pruneSnapshot?:
(params: { snapshot: WorkflowRunState; workflowStatus: WorkflowRunStatus }) => WorkflowRunState
Hook facultatif pour transformer l’instantané du Workflow juste avant sa conservation. Il doit renvoyer des données sérialisables en JSON et conserver tout ce dont le Workflow a besoin pour reprendre (suspendPayloads des étapes suspendues, suspendedPaths, executionPath, etc.). Utilisé en interne par les exécutions d’Agent afin de limiter les instantanés ; les Workflows utilisateur conservent par défaut des instantanés complets.
onFinish?:
(result: WorkflowFinishCallbackResult) => void | Promise<void>
Fonction de rappel invoquée lorsque le Workflow se termine avec n’importe quel état (success, failed, suspended, tripwire). Reçoit le résultat du Workflow, y compris l’état, la sortie, l’erreur et les résultats des étapes. Les erreurs levées dans cette fonction sont interceptées et journalisées, sans être propagées.
WorkflowFinishCallbackResult
status:
WorkflowRunStatus
État du Workflow : 'success', 'failed', 'suspended' ou 'tripwire'
result?:
any
Sortie du Workflow (lorsque status vaut 'success')
error?:
SerializedError
Détails de l’erreur (lorsque status vaut 'failed')
steps:
Record<string, StepResult>
Résultats individuels des étapes avec leur état et leur sortie
tripwire?:
StepTripwireInfo
Informations de tripwire (lorsque status vaut 'tripwire')
runId:
string
Identifiant unique de cette exécution de Workflow
workflowId:
string
Identifiant du Workflow
resourceId?:
string
Identifiant de ressource facultatif (s’il est fourni lors de la création de l’exécution)
getInitData:
() => any
Fonction qui renvoie les données d’entrée initiales transmises au Workflow
mastra?:
Mastra
Instance Mastra (si le Workflow est enregistré auprès de Mastra)
requestContext:
RequestContext
Données de contexte limitées à la requête
logger:
IMastraLogger
Instance de logger du Workflow
state:
Record<string, any>
Objet d’état actuel du Workflow
onError?:
(errorInfo: WorkflowErrorCallbackInfo) => void | Promise<void>
Fonction de rappel invoquée uniquement lorsque le Workflow échoue (état failed ou tripwire). Reçoit les détails de l’erreur et les résultats des étapes. Les erreurs levées dans cette fonction sont interceptées et journalisées, sans être propagées.
WorkflowErrorCallbackInfo
status:
'failed' | 'tripwire'
État du Workflow (soit 'failed', soit 'tripwire')
error?:
SerializedError
Détails de l’erreur
steps:
Record<string, StepResult>
Résultats individuels des étapes avec leur état et leur sortie
tripwire?:
StepTripwireInfo
Informations de tripwire (lorsque status vaut 'tripwire')
runId:
string
Identifiant unique de cette exécution de Workflow
workflowId:
string
Identifiant du Workflow
resourceId?:
string
Identifiant de ressource facultatif (s’il est fourni lors de la création de l’exécution)
getInitData:
() => any
Fonction qui renvoie les données d’entrée initiales transmises au Workflow
mastra?:
Mastra
Instance Mastra (si le Workflow est enregistré auprès de Mastra)
requestContext:
RequestContext
Données de contexte limitées à la requête
logger:
IMastraLogger
Instance de logger du Workflow
state:
Record<string, any>
Objet d’état actuel du Workflow
Exécuter avec un état initialLien direct vers Exécuter avec un état initial
Lors du démarrage d’une exécution de Workflow, vous pouvez transmettre initialState pour définir les valeurs de départ de l’état du Workflow :
const run = await workflow.createRun()
const result = await run.start({
inputData: { value: 'hello' },
initialState: {
counter: 0,
items: [],
},
})
L’objet initialState doit correspondre à la structure définie dans le stateSchema du Workflow. Consultez État du Workflow pour plus de détails.
État du WorkflowLien direct vers État du Workflow
Le status d’un Workflow indique son état d’exécution actuel. Les valeurs possibles sont :
success:
string
Toutes les étapes se sont terminées avec succès, avec une sortie de résultat valide
failed:
string
Le Workflow a rencontré une erreur pendant l’exécution, dont les détails sont disponibles
suspended:
string
L’exécution du Workflow est suspendue en attente de reprise, avec les informations des étapes suspendues
tripwire:
string
Le Workflow a été arrêté par un tripwire de processeur. Cela se produit lorsqu’une étape d’Agent dans le Workflow déclenche un tripwire (par exemple, un contenu a été bloqué par un garde-fou). Les informations de tripwire sont disponibles dans le résultat.
Gérer l’état tripwireLien direct vers Gérer l’état tripwire
Lorsqu’un Workflow contient une étape d’Agent qui déclenche un tripwire, le Workflow renvoie status: 'tripwire' et inclut les détails du 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)
}
Cela diffère de status: 'failed', qui indique une erreur inattendue. Un état tripwire signifie qu’un processeur a intentionnellement arrêté l’exécution (par exemple, pour la modération du contenu).