Streaming

Mastra prend en charge les réponses incrémentielles en temps réel des Agents et des Workflows, ce qui permet aux utilisateurs de voir la sortie à mesure de sa génération plutôt que d’attendre la fin. Cela est utile pour le chat, le contenu long, les Workflows à plusieurs étapes ou toute situation où un retour immédiat est important.

Premiers pas

L’API de streaming de Mastra s’adapte selon la version de votre modèle :

• .stream() : pour les modèles V2, prend en charge AI SDK v5 et les versions ultérieures (LanguageModelV2).
• .streamLegacy() : pour les modèles V1, prend en charge AI SDK v4 (LanguageModelV1).

Streaming avec des Agents

Vous pouvez transmettre une seule chaîne pour des prompts simples, un tableau de chaînes lorsque vous fournissez plusieurs éléments de contexte, ou un tableau d’objets message avec role et content pour contrôler précisément les rôles et les flux de conversation.

Utiliser Agent.stream()

Un textStream divise la réponse en segments à mesure de sa génération, ce qui permet une sortie progressive plutôt qu’une arrivée en bloc. Itérez sur textStream avec une boucle for await pour examiner chaque segment du flux.

const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}

Consultez Agent.stream() pour plus d’informations.

astuce

Pour les Agents qui distribuent des tâches en arrière-plan, utilisez Agent.streamUntilIdle() pour garder le flux ouvert jusqu’à la fin de ces tâches et permettre à l’Agent de répondre à leurs résultats.

Sortie de Agent.stream()

La sortie diffuse en flux la réponse générée par l’Agent.

Of course!
To help you organize your day effectively, I need a bit more information.
Here are some questions to consider:
...

Propriétés du flux d’Agent

Un flux d’Agent donne accès aux propriétés de réponse suivantes :

• stream.textStream : un flux lisible qui émet des segments de texte.
• stream.text : une promesse qui se résout en réponse textuelle complète.
• stream.finishReason : la raison pour laquelle l’Agent a arrêté le streaming.
• stream.usage : les informations d’utilisation des tokens.

Compatibilité avec AI SDK v5+

AI SDK v5 et les versions ultérieures utilisent LanguageModelV2 pour les fournisseurs de modèles. Si vous obtenez une erreur indiquant que vous utilisez un modèle AI SDK v4, vous devrez mettre à niveau votre package de modèle vers la prochaine version majeure.

Pour l’intégration avec AI SDK v5+, utilisez l’utilitaire toAISdkV5Stream() de @mastra/ai-sdk afin de convertir les flux Mastra vers un format compatible avec AI SDK :

import { toAISdkV5Stream } from '@mastra/ai-sdk'

const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

// Convert to AI SDK v5+ compatible stream
const aiSDKStream = toAISdkV5Stream(stream, { from: 'agent' })

Pour convertir les messages au format AI SDK v5+, utilisez l’utilitaire toAISdkV5Messages() de @mastra/ai-sdk/ui :

import { toAISdkV5Messages } from '@mastra/ai-sdk/ui'

const messages = [{ role: 'user', content: 'Hello' }]
const aiSDKMessages = toAISdkV5Messages(messages)

Streaming avec des Workflows

Le streaming depuis un Workflow retourne une séquence d’événements structurés décrivant le cycle de vie de l’exécution, plutôt que des segments de texte incrémentiels. Ce format fondé sur des événements permet de suivre la progression du Workflow et d’y répondre en temps réel une fois une exécution créée avec .createRun().

Utiliser Run.stream()

La méthode stream() retourne directement un ReadableStream d’événements.

const run = await testWorkflow.createRun()

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

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

Consultez Run.stream() pour plus d’informations.

Sortie de Run.stream()

La structure de l’événement comprend runId et from au niveau supérieur, ce qui facilite l’identification et le suivi des exécutions de Workflow sans devoir examiner la charge utile.

{
  type: 'workflow-start',
  runId: '1eeaf01a-d2bf-4e3f-8d1b-027795ccd3df',
  from: 'WORKFLOW',
  payload: {
    stepName: 'step-1',
    args: { value: 'initial data' },
    stepCallId: '8e15e618-be0e-4215-a5d6-08e58c152068',
    startedAt: 1755121710066,
    status: 'running'
  }
}

Propriétés du flux de Workflow

Un flux de Workflow donne accès aux propriétés de réponse suivantes :

• stream.status : le statut de l’exécution du Workflow.
• stream.result : le résultat de l’exécution du Workflow.
• stream.usage : l’utilisation totale des tokens par l’exécution du Workflow.

Le streaming des Agents ou des Workflows offre une visibilité en temps réel sur la sortie du LLM ou sur le statut d’une exécution de Workflow. Transmettez ce retour directement à l’utilisateur, ou utilisez-le dans une application pour afficher l’évolution du statut du Workflow.

Les événements émis par les Agents ou les Workflows représentent différentes phases de génération et d’exécution, comme le démarrage d’une exécution, la production de texte ou l’invocation d’un Tool.

Types d’événements

Vous trouverez ci-dessous la liste complète des événements émis par .stream(). Selon que vous diffusez le flux depuis un Agent ou un Workflow, seul un sous-ensemble de ces événements se produit :

• start : marque le début d’une exécution d’Agent ou de Workflow.
• step-start : indique qu’une étape de Workflow a commencé son exécution.
• text-delta : segments de texte incrémentiels à mesure de leur génération par le LLM.
• tool-call : lorsque l’Agent décide d’utiliser un Tool, avec le nom et les arguments du Tool.
• tool-result : le résultat retourné par l’exécution du Tool.
• step-finish : confirme qu’une étape précise est complètement terminée et peut inclure des métadonnées, comme la raison de fin de cette étape.
• finish : lorsque l’Agent ou le Workflow se termine, avec les statistiques d’utilisation.

Inspecter les flux d’Agent

Itérez sur stream avec une boucle for await pour examiner tous les segments d’événements émis.

const testAgent = mastra.getAgent('testAgent')

const stream = await testAgent.stream([{ role: 'user', content: 'Help me organize my day' }])

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

Consultez Agent.stream() pour plus d’informations.

Exemple de sortie d’Agent

Voici un exemple d’événements qui peuvent être émis. Chaque événement comprend toujours un type et peut inclure des champs supplémentaires, comme from et payload.

{
  type: 'start',
  from: 'AGENT',
  // ..
}
{
  type: 'step-start',
  from: 'AGENT',
  payload: {
    messageId: 'msg-cdUrkirvXw8A6oE4t5lzDuxi',
    // ...
  }
}
{
  type: 'tool-call',
  from: 'AGENT',
  payload: {
    toolCallId: 'call_jbhi3s1qvR6Aqt9axCfTBMsA',
    toolName: 'testTool'
    // ..
  }
}

API Writer

L’API writer est partagée par les Tools et les étapes de Workflow. Consultez la documentation des Tools et Workflows pour des exemples propres à chaque fonctionnalité.

Agent utilisant un Tool

Le streaming d’Agent peut être combiné avec des appels de Tools, ce qui permet d’écrire les sorties des Tools directement dans la réponse diffusée de l’Agent. L’activité des Tools apparaît ainsi comme partie intégrante de l’interaction.

import { Agent } from '@mastra/core/agent'
import { testTool } from '../tools/test-tool'

export const testAgent = new Agent({
  id: 'test-agent',
  name: 'Test Agent',
  instructions: 'You are a weather agent.',
  model: 'openai/gpt-5.6-sol',
  tools: { testTool },
})

Utiliser context.writer

L’objet context.writer est disponible dans la fonction execute() d’un Tool et peut émettre des événements, données ou valeurs personnalisés dans le flux actif. Les Tools utilisent ces événements pour fournir des résultats intermédiaires ou des mises à jour de statut pendant l’exécution.

attention

Vous devez utiliser await avec l’appel à writer.write() ; sinon, vous verrouillerez le flux et obtiendrez l’erreur WritableStream is locked.

import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.write({
      type: 'custom-event',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.write({
      type: 'custom-event',
      status: 'success',
    })

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

Vous pouvez également utiliser writer.custom() pour émettre des segments de flux au niveau supérieur. Cela est utile lors de l’intégration avec des frameworks d’interface.

import { createTool } from '@mastra/core/tools'

export const testTool = createTool({
  execute: async (inputData, context) => {
    const { value } = inputData

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'pending',
    })

    const response = await fetch()

    await context?.writer?.custom({
      type: 'data-tool-progress',
      status: 'success',
    })

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

Segments de données transitoires

Par défaut, les segments data-* émis avec writer.custom() sont persistés dans le stockage comme partie de l’historique des messages. Pour les segments nécessaires uniquement pendant le streaming en direct, tels que les mises à jour de progression ou les sorties de journal détaillées, définissez transient: true afin d’ignorer la persistance dans le stockage. Les segments transitoires sont toujours diffusés au client en temps réel, mais ne sont pas enregistrés dans la base de données.

await context?.writer?.custom({
  type: 'data-build-log',
  data: { line: 'Compiling module 3 of 12...' },
  transient: true,
})

Utilisez des segments transitoires lorsque les données sont volumineuses ou très fréquentes et ne sont pertinentes que pendant la session active. Après l’actualisation de la page, les segments transitoires ne sont plus disponibles. Seuls la valeur de retour du Tool et les segments non transitoires sont chargés depuis le stockage.

Utiliser l’argument writer

L’argument writer est transmis à la fonction execute d’une étape de Workflow et peut émettre des événements, données ou valeurs personnalisés dans le flux actif. Les étapes de Workflow utilisent ces événements pour fournir des résultats intermédiaires ou des mises à jour de statut pendant l’exécution.

attention

Vous devez utiliser await avec l’appel à writer.write(...) ; sinon, vous verrouillerez le flux et obtiendrez l’erreur WritableStream is locked.

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

export const testStep = createStep({
  execute: async ({ inputData, writer }) => {
    const { value } = inputData;

    await writer?.write({
      type: "custom-event",
      status: "pending"
    });

    const response = await fetch(...);

    await writer?.write({
      type: "custom-event",
      status: "success"
    });

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