Voyage dans le temps

Le voyage dans le temps vous permet de réexécuter un Workflow à partir de n’importe quelle étape précise, en utilisant soit les données d’un snapshot stocké, soit le contexte personnalisé que vous fournissez.

Cette fonctionnalité est utile pour déboguer les Workflows ayant échoué, tester des étapes individuelles avec différentes entrées ou récupérer après des erreurs sans réexécuter l’intégralité du Workflow. Vous pouvez également l’utiliser pour exécuter, à partir de n’importe quelle étape précise, un Workflow qui ne l’a encore jamais été.

Fonctionnement du voyage dans le temps

Lorsque vous appelez timeTravel() sur une exécution de Workflow :

1. Le Workflow charge le snapshot existant depuis le stockage, s’il est disponible
2. Les résultats des étapes antérieures à l’étape cible sont reconstruits à partir du snapshot ou du contexte fourni
3. L’exécution commence à l’étape indiquée avec les données d’entrée fournies ou reconstruites
4. Le Workflow poursuit son exécution jusqu’à son terme à partir de ce point

Le voyage dans le temps nécessite un stockage configuré, car il repose sur des snapshots de Workflow persistants.

Utilisation de base

Utilisez run.timeTravel() pour réexécuter un Workflow à partir d’une étape précise :

import { mastra } from './mastra'

const workflow = mastra.getWorkflow('myWorkflow')
const run = await workflow.createRun()

const result = await run.timeTravel({
  step: 'step2',
  inputData: { previousStepResult: 'custom value' },
})

Indiquer l’étape cible

Vous pouvez indiquer l’étape cible au moyen d’une référence d’étape ou d’un identifiant d’étape :

Utiliser une référence d’étape

const result = await run.timeTravel({
  step: step2,
  inputData: { value: 10 },
})

Utiliser un identifiant d’étape

const result = await run.timeTravel({
  step: 'step2',
  inputData: { value: 10 },
})

Étapes de Workflows imbriqués

Pour les étapes situées dans des Workflows imbriqués, utilisez la notation pointée, un tableau d’identifiants d’étape ou un tableau de références d’étape :

// Using dot notation
const result = await run.timeTravel({
  step: 'nestedWorkflow.step3',
  inputData: { value: 10 },
})

// Using array of step IDs
const result = await run.timeTravel({
  step: ['nestedWorkflow', 'step3'],
  inputData: { value: 10 },
})

// Using array of step references
const result = await run.timeTravel({
  step: [nestedWorkflow, step3],
  inputData: { value: 10 },
})

Fournir le contexte d’exécution

Vous pouvez fournir un contexte pour indiquer l’état des étapes précédentes lors d’un voyage dans le temps :

const result = await run.timeTravel({
  step: 'step2',
  context: {
    step1: {
      status: 'success',
      payload: { value: 0 },
      output: { step1Result: 2 },
      startedAt: Date.now(),
      endedAt: Date.now(),
    },
  },
})

L’objet de contexte contient les résultats des étapes indexés par leur identifiant. Chaque résultat d’étape comprend :

• status : état d’exécution de l’étape (success, failed, suspended)
• payload : données d’entrée transmises à l’étape
• output : données de sortie de l’étape pour les étapes réussies
• startedAt : horodatage du démarrage de l’étape
• endedAt : horodatage de la fin de l’étape pour les étapes terminées
• suspendPayload : données transmises à suspend() pour les étapes suspendues
• resumePayload : données transmises à resume() pour les étapes reprises

Réexécuter des Workflows ayant échoué

Le voyage dans le temps est particulièrement utile pour déboguer les exécutions de Workflows ayant échoué et les récupérer :

const workflow = mastra.getWorkflow('myWorkflow')
const run = await workflow.createRun()

// Initial run fails at step2
const failedResult = await run.start({
  inputData: { value: 1 },
})

if (failedResult.status === 'failed') {
  // Re-run from step2 with corrected input
  const recoveredResult = await run.timeTravel({
    step: 'step2',
    inputData: { step1Result: 5 }, // Provide corrected input
  })
}

Voyage dans le temps avec des Workflows suspendus

Vous pouvez voyager dans le temps afin de reprendre un Workflow suspendu depuis une étape antérieure :

const run = await workflow.createRun()

// Start workflow - suspends at promptAgent step
const initialResult = await run.start({
  inputData: { input: 'test' },
})

if (initialResult.status === 'suspended') {
  // Time travel back to an earlier step with resume data
  const result = await run.timeTravel({
    step: 'getUserInput',
    resumeData: {
      userInput: 'corrected input',
    },
  })
}

Diffuser les résultats du voyage dans le temps

Utilisez timeTravelStream() pour recevoir des événements diffusés pendant l’exécution du voyage dans le temps :

const run = await workflow.createRun()

const stream = run.timeTravelStream({
  step: 'step2',
  inputData: { value: 10 },
})

for await (const event of stream.fullStream) {
  console.log(event.type, event.payload)
}

const result = await stream.result

if (result.status === 'success') {
  console.log(result.result)
}

Voyage dans le temps avec un état initial

Vous pouvez fournir un état initial lors du voyage dans le temps afin de définir l’état au niveau du Workflow :

const result = await run.timeTravel({
  step: 'step2',
  inputData: { value: 10 },
  initialState: {
    counter: 5,
    metadata: { source: 'time-travel' },
  },
})

Gestion des erreurs

Le voyage dans le temps lève des erreurs dans certaines situations :

Workflow en cours d’exécution

Vous ne pouvez pas voyager dans le temps vers un Workflow en cours d’exécution :

try {
  await run.timeTravel({ step: 'step2' })
} catch (error) {
  // "This workflow run is still running, cannot time travel"
}

Identifiant d’étape non valide

Le voyage dans le temps lève une erreur si l’étape cible n’existe pas dans le Workflow :

try {
  await run.timeTravel({ step: 'nonExistentStep' })
} catch (error) {
  // "Time travel target step not found in execution graph: 'nonExistentStep'. Verify the step id/path."
}

Données d’entrée non valides

Lorsque validateInputs est activé, le voyage dans le temps valide les données d’entrée par rapport au schéma de l’étape :

try {
  await run.timeTravel({
    step: 'step2',
    inputData: { invalidField: 'value' },
  })
} catch (error) {
  // "Invalid inputData: \n- step1Result: Required"
}

Contexte des Workflows imbriqués

Lorsque vous voyagez dans le temps au sein d’un Workflow imbriqué, vous pouvez fournir un contexte pour les étapes du Workflow parent et celles du Workflow imbriqué :

const result = await run.timeTravel({
  step: 'nestedWorkflow.step3',
  context: {
    step1: {
      status: 'success',
      payload: { value: 0 },
      output: { step1Result: 2 },
      startedAt: Date.now(),
      endedAt: Date.now(),
    },
    nestedWorkflow: {
      status: 'running',
      payload: { step1Result: 2 },
      startedAt: Date.now(),
    },
  },
  nestedStepsContext: {
    nestedWorkflow: {
      step2: {
        status: 'success',
        payload: { step1Result: 2 },
        output: { step2Result: 3 },
        startedAt: Date.now(),
        endedAt: Date.now(),
      },
    },
  },
})

Cas d’utilisation

Déboguer des étapes ayant échoué

Réexécutez une étape ayant échoué avec une entrée identique ou modifiée afin de diagnostiquer les problèmes :

const result = await run.timeTravel({
  step: failedStepId,
  context: originalContext, // Use context from the failed run
})

Tester la logique d’une étape dans une nouvelle exécution de Workflow

Testez des étapes individuelles avec des entrées précises dans une nouvelle exécution de Workflow. Cette approche est utile pour tester la logique d’une étape sans démarrer l’exécution du Workflow depuis le début.

const result = await run.timeTravel({
  step: 'processData',
  inputData: { testData: 'specific test case' },
})

Récupérer après des échecs temporaires

Réexécutez les étapes ayant échoué en raison de problèmes temporaires, comme des erreurs réseau ou des limitations de débit :

// After fixing the external service issue
const result = await run.timeTravel({
  step: 'callExternalApi',
  inputData: savedInputData,
})

Voir aussi

• Snapshots
• Suspendre et reprendre
• Gestion des erreurs
• Flux de contrôle