Classe Step

La classe Step définit des unités de travail individuelles au sein d’un Workflow, en encapsulant la logique d’exécution, la validation des données et la gestion des entrées et sorties. Elle peut recevoir un Tool ou un Agent en paramètre afin de créer automatiquement une étape à partir de celui-ci.

Exemple d’utilisation

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  description: 'passes value from input to output',
  inputSchema: z.object({
    value: z.number(),
  }),
  outputSchema: z.object({
    value: z.number(),
  }),
  execute: async ({ inputData }) => {
    const { value } = inputData
    return {
      value,
    }
  },
})

Définir des schémas

Vous pouvez définir les inputSchema et outputSchema de l’étape avec toute bibliothèque prenant en charge Standard JSON Schema, notamment Zod, Valibot et ArkType.

Zod

src/mastra/workflows/test-workflow.ts

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

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    message: z.string(),
  }),
  outputSchema: z.object({
    formatted: z.string(),
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

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

Valibot

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import * as v from 'valibot'
import { toStandardJsonSchema } from '@valibot/to-json-schema'

const step1 = createStep({
  id: 'step-1',
  inputSchema: toStandardJsonSchema(
    v.object({
      message: v.string(),
    }),
  ),
  outputSchema: toStandardJsonSchema(
    v.object({
      formatted: v.string(),
    }),
  ),
  execute: async ({ inputData }) => {
    const { message } = inputData

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

ArkType

src/mastra/workflows/test-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { type } from 'arktype'

const step1 = createStep({
  id: 'step-1',
  inputSchema: type({
    message: 'string',
  }),
  outputSchema: type({
    formatted: 'string',
  }),
  execute: async ({ inputData }) => {
    const { message } = inputData

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

Créer des étapes à partir d’Agents

Vous pouvez créer une étape directement à partir d’un Agent. L’étape utilise le nom de l’Agent comme ID.

Étape d’Agent de base

src/mastra/workflows/test-workflow.ts

import { testAgent } from '../agents/test-agent'

const agentStep = createStep(testAgent)
// inputSchema: { prompt: string }
// outputSchema: { text: string }

Étape d’Agent avec sortie structurée

Transmettez structuredOutput pour que l’Agent renvoie des données structurées typées :

src/mastra/workflows/test-workflow.ts

const articleSchema = z.object({
  title: z.string(),
  summary: z.string(),
  tags: z.array(z.string()),
})

const agentStep = createStep(testAgent, {
  structuredOutput: { schema: articleSchema },
})
// inputSchema: { prompt: string }
// outputSchema: { title: string, summary: string, tags: string[] }

Options d’étape d’Agent

structuredOutput:

{ schema: StandardJSONSchemaV1 }

Lorsqu’il est fourni, l’Agent renvoie des données structurées correspondant à ce schéma au lieu de texte brut. L’outputSchema de l’étape est défini sur le schéma fourni.

onFinish:

(result: AgentResult) => void

Fonction de rappel invoquée lorsque l’Agent termine sa génération.

Paramètres du constructeur

id:

string

Identifiant unique de l’étape

description:

string

Description facultative de l’action effectuée par l’étape

inputSchema:

StandardJSONSchemaV1

Standard JSON Schema définissant la structure d’entrée

outputSchema:

StandardJSONSchemaV1

Standard JSON Schema définissant la structure de sortie

resumeSchema:

StandardJSONSchemaV1

Standard JSON Schema facultatif pour la reprise de l’étape

suspendSchema:

StandardJSONSchemaV1

Standard JSON Schema facultatif pour la suspension de l’étape

stateSchema:

StandardJSONSchemaV1

Standard JSON Schema facultatif pour l’état de l’étape. Injecté automatiquement lors de l’utilisation du système d’état de Mastra. Le stateSchema doit être un sous-ensemble du stateSchema du Workflow. 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é avant l’exécution de execute() de l’étape, ce qui fait échouer l’étape si la validation échoue.

execute:

(params: ExecuteParams) => Promise<any>

Fonction asynchrone contenant la logique de l’étape

ExecuteParams

inputData:

z.infer<TStepInput>

Données d’entrée correspondant à l’inputSchema

resumeData:

z.infer<TResumeSchema>

Données de reprise correspondant au resumeSchema, lors de la reprise de l’étape depuis un état suspendu. N’existent que si l’étape est reprise.

suspendData:

z.infer<TSuspendSchema>

Données de suspension transmises initialement à suspend() lorsque l’étape a été suspendue. N’existent que si l’étape est reprise et qu’elle a été précédemment suspendue avec des données.

mastra:

Mastra

Accès aux services Mastra (Agents, Tools, etc.)

getStepResult:

(step: Step | string) => any

Fonction permettant d’accéder aux résultats des autres étapes

getInitData:

() => any

Fonction permettant d’accéder aux données d’entrée initiales du Workflow depuis n’importe quelle étape

suspend:

(suspendPayload: any, suspendOptions?: { resumeLabel?: string }) => Promise<void>

Fonction permettant de suspendre l’exécution du Workflow

state:

z.infer<TState>

État actuel du Workflow. Contient des valeurs partagées qui persistent entre toutes les étapes et tous les cycles de suspension et reprise. La structure est définie par le stateSchema de l’étape.

setState:

(state: z.infer<TState>) => void

Fonction permettant de définir l’état du Workflow. Injectez-le via un modèle de type reducer, tel que 'setState({ ...state, ...newState })'

runId:

string

ID de l’exécution actuelle

requestContext?:

RequestContext

Request Context pour l’injection de dépendances et les informations contextuelles.

retryCount?:

number

Nombre de nouvelles tentatives pour cette étape spécifique ; il augmente automatiquement à chaque nouvelle tentative de l’étape

scorers:

MastraScorers | (({ requestContext }) => MastraScorers | Promise<MastraScorers>)

Scorers exécutés automatiquement après la fin réussie de l’étape. Chaque scorer évalue l’entrée et la sortie propres à l’étape ; les résultats sont stockés et associés à la trace de l’étape. Fournissez une mappe de { [name]: { scorer, sampling? } } ou une fonction qui en renvoie une. Le scoring s’exécute de manière asynchrone et ne bloque pas le Workflow. Consultez Évaluer la sortie d’une étape.

retries:

number

Nombre de fois où réessayer la fonction execute de l’étape si elle lève une erreur.

metadata:

Record<string, any>

Paires clé-valeur facultatives permettant de stocker des informations supplémentaires sur l’étape. Les valeurs doivent être sérialisables (sans fonctions, références circulaires, etc.).

Évaluer la sortie d’une étape

Associez des scorers à une étape pour évaluer automatiquement la sortie de cette étape à son exécution, au lieu d’évaluer uniquement la réponse finale du Workflow. Cela est utile pour les Workflows à plusieurs étapes et RAG, lorsque vous voulez voir quelle étape a dégradé la qualité, par exemple si une étape de récupération a renvoyé des fragments pertinents avant que les étapes suivantes ne raisonnent à partir d’eux.

Chaque scorer reçoit l’input et l’output propres à l’étape. Le scoring s’exécute de manière asynchrone après la réussite de l’étape, et le résultat est stocké avec la trace de l’étape. Utilisez sampling pour contrôler la fréquence d’exécution d’un scorer.

L’exemple suivant associe un scorer à une étape de récupération afin que chaque exécution soit évaluée :

src/mastra/workflows/rag-workflow.ts

import { createStep } from '@mastra/core/workflows'
import { z } from 'zod'
import { retrievalRelevanceScorer } from '../scorers/retrieval-relevance'

const retrievalStep = createStep({
  id: 'retrieval',
  inputSchema: z.object({ query: z.string() }),
  outputSchema: z.object({ query: z.string(), chunks: z.array(z.string()) }),
  scorers: {
    retrievalRelevance: {
      scorer: retrievalRelevanceScorer(),
      sampling: { type: 'ratio', rate: 1 },
    },
  },
  execute: async ({ inputData }) => {
    const chunks = await retrieve(inputData.query)
    return { query: inputData.query, chunks }
  },
})

Associez un scorer à chaque étape que vous voulez mesurer pour établir des scores par étape dans un Workflow à plusieurs étapes. Comme le scoring est limité à une seule étape, vous n’avez pas besoin d’une métrique dédiée inter-étapes pour voir où la qualité change.

Les étapes d’Agent et de Tool ajoutées avec Workflow.agent() et Workflow.tool() acceptent la même option scorers dans leurs options d’étape.

remarque

Consultez la présentation des Scorers pour savoir comment s’exécutent les évaluations en direct et où les résultats sont stockés, ainsi que les Scorers personnalisés pour créer les vôtres.

Voir aussi

• État du Workflow
• Flux de contrôle
• Utiliser des Agents et des Tools