Agent.stream()

La méthode .stream() permet de diffuser en temps réel les réponses d’un agent, avec des fonctionnalités avancées et une grande souplesse de format. Cette méthode accepte des messages et des options de streaming facultatives, offrant une expérience de streaming moderne compatible à la fois avec le format natif de Mastra et avec AI SDK v5+.

Exemple d’utilisation

const stream = await agent.stream('message for agent')

info

Compatibilité des modèles : cette méthode est conçue pour les modèles V2. Les modèles V1 doivent utiliser la méthode .streamLegacy(). Mastra détecte automatiquement la version de votre modèle et génère une erreur en cas d’incompatibilité.

Paramètres

messages:

string | string[] | CoreMessage[] | AiMessageType[] | UIMessageWithMetadata[]

Messages à envoyer à l’agent. Il peut s’agir d’une chaîne unique, d’un tableau de chaînes ou d’objets de message structurés.

options?:

AgentExecutionOptions<Output, Format>

Configuration facultative du processus de streaming.

AgentExecutionOptions<Output, Format>

maxSteps?:

number

Nombre maximal d’étapes à exécuter.

scorers?:

MastraScorers | Record<string, { scorer: MastraScorer['name']; sampling?: ScoringSamplingConfig }>

Évaluateurs à exécuter sur les résultats.

scorer:

string

Nom de l’évaluateur à utiliser.

sampling?:

ScoringSamplingConfig

Configuration de l’échantillonnage de l’évaluateur.

type:

'none' | 'ratio'

Type de stratégie d’échantillonnage. Utilisez 'none' pour désactiver l’échantillonnage ou 'ratio' pour un échantillonnage fondé sur un pourcentage.

rate?:

number

Taux d’échantillonnage (0-1). Requis lorsque le type est 'ratio'.

onIterationComplete?:

(context: IterationCompleteContext) => { continue?: boolean; feedback?: string } | void | Promise<{ continue?: boolean; feedback?: string } | void>

Fonction de rappel appelée à la fin de chaque itération. Utilisez-la pour suivre la progression, fournir un retour afin de guider l’agent ou arrêter l’exécution de manière anticipée. Elle reçoit le contexte de l’itération, notamment le texte actuel, les appels d’outils et le motif de fin.

context.iteration:

number

Numéro de l’itération actuelle (à partir de 1).

context.maxIterations:

number | undefined

Nombre maximal d’itérations autorisées (s’il est défini).

context.text:

string

Réponse textuelle de cette itération.

context.isFinal:

boolean

Indique s’il s’agit de la dernière itération.

context.finishReason:

string

Motif de fin de cette itération (par ex. 'stop', 'length', 'tool-calls').

context.toolCalls:

ToolCall[]

Appels d’outils effectués pendant cette itération.

context.messages:

MastraDBMessage[]

Tous les messages accumulés jusqu’à présent.

return.continue?:

boolean

Définissez cette valeur sur false pour arrêter l’exécution de manière anticipée.

return.feedback?:

string

Message de retour destiné à guider l’itération suivante de l’agent.

isTaskComplete?:

IsTaskCompleteConfig

Configuration de l’évaluation d’achèvement qui vérifie si la tâche est terminée. Utilise les évaluateurs de Mastra pour vérifier automatiquement si la réponse de l’agent satisfait aux critères d’achèvement.

scorers:

MastraScorer[]

Tableau d’évaluateurs qui déterminent si la tâche est achevée. Chaque évaluateur renvoie 0 (échec) ou 1 (réussite).

strategy?:

'all' | 'any'

Stratégie de combinaison des résultats des évaluateurs. 'all' exige la réussite de tous les évaluateurs, tandis que 'any' en exige au moins un.

onComplete?:

(result: IsTaskCompleteRunResult) => void | Promise<void>

Fonction de rappel appelée à la fin de la vérification de l’achèvement de la tâche. Elle reçoit le résultat avec le score de chaque évaluateur.

parallel?:

boolean

Indique si les évaluateurs doivent être exécutés en parallèle.

timeout?:

number

Durée maximale, en millisecondes, d’attente de la fin de tous les évaluateurs.

suppressFeedback?:

boolean

Lorsque cette valeur est true, marque le retour de la vérification d’achèvement afin que les consommateurs puissent le masquer dans la sortie affichée. Ce retour n’est ajouté à la conversation qu’en cas d’échec de la vérification, afin de guider l’itération suivante.

delegation?:

DelegationConfig

Configuration de la délégation aux sous-agents. Utilisez-la pour contrôler et surveiller la délégation de tâches à d’autres agents, notamment pour modifier ou refuser les délégations et fournir un retour afin de guider le superviseur.

onDelegationStart?:

(context: DelegationStartContext) => DelegationStartResult | void | Promise<DelegationStartResult | void>

Appelée avant la délégation à un sous-agent. Utilisez-la pour modifier les paramètres de délégation, refuser entièrement la délégation ou modifier context.requestContext afin d’ajouter des entrées au contexte de requête de l’exécution du sous-agent.

onDelegationComplete?:

(context: DelegationCompleteContext) => { feedback?: string } | void | Promise<{ feedback?: string } | void>

Appelée une fois la délégation à un sous-agent terminée. Le contexte comprend une méthode bail() permettant d’arrêter la suite de l’exécution, et vous pouvez renvoyer { feedback } pour guider l’action suivante du superviseur. Le retour est enregistré dans la mémoire du superviseur sous forme de message de l’assistant.

messageFilter?:

(context: MessageFilterContext) => MastraDBMessage[] | Promise<MastraDBMessage[]>

Fonction de rappel appelée avant la délégation à un sous-agent. Utilisez-la pour filtrer les messages transmis au sous-agent.

tracingContext?:

TracingContext

Contexte de traçage pour la hiérarchie des spans et les métadonnées.

returnScorerData?:

boolean

Indique si des données d’évaluation détaillées doivent être renvoyées dans la réponse.

onChunk?:

(chunk: ChunkType) => Promise<void> | void

Fonction de rappel appelée pour chaque fragment pendant le streaming.

onError?:

({ error }: { error: Error | string }) => Promise<void> | void

Fonction de rappel appelée lorsqu’une erreur survient pendant le streaming.

onAbort?:

(event: any) => Promise<void> | void

Fonction de rappel appelée lorsque le flux est interrompu.

abortSignal?:

AbortSignal

Objet signal qui permet d’interrompre l’exécution de l’agent. Lorsque le signal est interrompu, toutes les opérations en cours prennent fin, y compris les exécutions de sous-agents en cours que l’agent a déléguées.

activeTools?:

Array<keyof ToolSet> | undefined

Tableau des noms d’outils actifs pouvant être utilisés pendant l’exécution.

prepareStep?:

PrepareStepFunction<any>

Fonction de rappel appelée avant chaque étape d’une exécution en plusieurs étapes.

context?:

ModelMessage[]

Messages de contexte supplémentaires à fournir à l’agent.

structuredOutput?:

StructuredOutputOptions<S extends ZodTypeAny = ZodTypeAny>

Options permettant d’affiner la génération de la sortie structurée.

schema:

StandardJSONSchemaV1

Schéma JSON standard définissant la structure de sortie attendue.

model?:

MastraLanguageModel

Modèle de langage à utiliser pour générer la sortie structurée. Lorsqu’il est fourni, il permet à l’agent de répondre en plusieurs étapes avec des appels d’outils, du texte et une sortie structurée.

errorStrategy?:

'strict' | 'warn' | 'fallback'

Stratégie de gestion des erreurs de validation du schéma. 'strict' génère des erreurs, 'warn' consigne des avertissements et 'fallback' utilise des valeurs de repli.

fallbackValue?:

<S extends ZodTypeAny>

Valeur de repli à utiliser lorsque la validation du schéma échoue et que errorStrategy vaut 'fallback'.

instructions?:

string

Instructions supplémentaires pour le modèle de sortie structurée.

jsonPromptInjection?:

boolean | 'system' | 'inline' | 'auto'

Contrôle la façon dont le schéma JSON est transmis au modèle. Définissez cette valeur sur 'auto' pour utiliser la sortie structurée native lorsqu’elle est prise en charge, et injecter le schéma directement dans le prompt dans le cas contraire.

providerOptions?:

ProviderOptions

Options propres au fournisseur transmises à l’agent interne de structuration. Utilisez-les pour contrôler le comportement du modèle, comme l’effort de raisonnement des modèles de réflexion (par ex. { openai: { reasoningEffort: 'low' } }).

outputProcessors?:

Processor[]

Remplace les processeurs de sortie définis sur l’agent. Ces processeurs peuvent modifier ou valider les messages de l’agent avant leur renvoi à l’utilisateur. Ils doivent implémenter l’une des fonctions processOutputResult et processOutputStream, ou les deux.

includeRawChunks?:

boolean

Indique si les fragments bruts doivent être inclus dans la sortie du flux (option non disponible chez tous les fournisseurs de modèles).

inputProcessors?:

Processor[]

Remplace les processeurs d’entrée définis sur l’agent. Ces processeurs peuvent modifier ou valider les messages avant leur traitement par l’agent. Ils doivent implémenter la fonction processInput.

instructions?:

string

Instructions personnalisées qui remplacent les instructions par défaut de l’agent pour cette génération précise. Elles permettent de modifier dynamiquement le comportement de l’agent sans créer une nouvelle instance.

system?:

string | string[] | CoreSystemMessage | SystemModelMessage | CoreSystemMessage[] | SystemModelMessage[]

Messages système personnalisés à inclure dans le prompt. Il peut s’agir d’une chaîne unique, d’un objet message ou d’un tableau contenant l’un ou l’autre. Les messages système apportent du contexte ou des instructions de comportement supplémentaires qui complètent les instructions principales de l’agent.

output?:

Zod schema | JsonSchema7

**Obsolète.** Utilisez structuredOutput sans modèle pour obtenir le même résultat. Définit la structure attendue de la sortie. Peut être un objet JSON Schema ou un schéma Zod.

memory?:

object

Configuration de la mémoire. Il s’agit de la méthode recommandée pour gérer la mémoire.

thread:

string | { id: string; metadata?: Record<string, any>, title?: string }

Thread de conversation, sous forme d’un ID chaîne ou d’un objet comportant un id et des metadata facultatives.

resource:

string

Identifiant de l’utilisateur ou de la ressource associés au thread.

options?:

MemoryConfig

Configuration du comportement de la mémoire, notamment lastMessages, readOnly, semanticRecall, workingMemory et filterIncompleteToolCalls.

onTitleGenerated?:

(title: string) => void | Promise<void>

Fonction de rappel déclenchée de manière asynchrone lorsqu’un titre de thread est généré et conservé dans le stockage. La génération du titre s’effectue en arrière-plan et peut se terminer après la fin du flux. Elle ne se déclenche que lorsque generateTitle est activé dans les options de mémoire et que le thread ne possède pas encore de titre.

onFinish?:

StreamTextOnFinishCallback<any> | StreamObjectOnFinishCallback<OUTPUT>

Fonction de rappel appelée à la fin du streaming. Elle reçoit le résultat final.

onStepFinish?:

StreamTextOnStepFinishCallback<any> | never

Fonction de rappel appelée après chaque étape d’exécution. Elle reçoit les détails de l’étape sous forme de chaîne JSON. Non disponible pour les sorties structurées.

telemetry?:

TelemetrySettings

Paramètres de collecte de la télémétrie OTLP pendant le streaming (hors Tracing).

isEnabled?:

boolean

Active ou désactive la télémétrie. Désactivée par défaut pendant la phase expérimentale.

recordInputs?:

boolean

Active ou désactive l’enregistrement des entrées. Activé par défaut. Vous pouvez désactiver cet enregistrement afin d’éviter de consigner des informations sensibles.

recordOutputs?:

boolean

Active ou désactive l’enregistrement des sorties. Activé par défaut. Vous pouvez désactiver cet enregistrement afin d’éviter de consigner des informations sensibles.

functionId?:

string

Identifiant de cette fonction. Utilisé pour regrouper les données de télémétrie par fonction.

modelSettings?:

CallSettings

Model-specific settings like temperature, maxOutputTokens, topP, etc. These settings control how the language model generates responses.

temperature?:

number

Controls randomness in generation (0-2). Higher values make output more random.

maxOutputTokens?:

number

Maximum number of tokens to generate in the response. Note: Use maxOutputTokens (not maxTokens) as per AI SDK v5 convention.

maxRetries?:

number

Maximum number of retry attempts for failed requests.

topP?:

number

Nucleus sampling parameter (0-1). Controls diversity of generated text.

topK?:

number

Top-k sampling parameter. Limits vocabulary to k most likely tokens.

presencePenalty?:

number

Penalty for token presence (-2 to 2). Reduces repetition.

frequencyPenalty?:

number

Penalty for token frequency (-2 to 2). Reduces repetition of frequent tokens.

stopSequences?:

string[]

Stop sequences. If set, the model will stop generating text when one of the stop sequences is generated.

toolChoice?:

'auto' | 'none' | 'required' | { type: 'tool'; toolName: string }

Contrôle la manière dont l’agent utilise les outils pendant le streaming.

'auto':

string

Laisse le modèle décider s’il doit utiliser des outils (par défaut).

'none':

string

N’utilise aucun outil.

'required':

string

Exige que le modèle utilise au moins un outil.

{ type: 'tool'; toolName: string }:

object

Exige que le modèle utilise un outil précis, désigné par son nom.

toolsets?:

ToolsetsInput

Ensembles d’outils supplémentaires à mettre à la disposition de l’agent pendant le streaming.

clientTools?:

ToolsInput

Outils exécutés côté 'client' de la requête. La définition de ces outils ne comporte aucune fonction execute.

hooks?:

ToolHooks

Hooks propres à l’exécution, lancés avant et après les appels d’outils. Remplace, pour cette exécution, les hooks correspondants définis au niveau de l’agent. beforeToolCall peut renvoyer { proceed: false, output } pour ignorer l’appel d’outil.

savePerStep?:

boolean

Enregistre progressivement les messages après chaque étape terminée du flux (par défaut : false).

requireToolApproval?:

boolean

Lorsque cette valeur est true, tous les appels d’outils exigent une approbation explicite avant leur exécution. Le flux émet des fragments tool-call-approval et se met en pause jusqu’à l’appel de approveToolCall() ou declineToolCall().

autoResumeSuspendedTools?:

boolean

Lorsque cette valeur est true, reprend automatiquement les outils suspendus quand l’utilisateur envoie un nouveau message dans le même thread. L’agent extrait resumeData du message de l’utilisateur selon le resumeSchema de l’outil. Nécessite la configuration de la mémoire.

toolCallConcurrency?:

number

Nombre maximal d’appels d’outils à exécuter simultanément. La valeur par défaut est 1 lorsqu’une approbation peut être requise, et 10 dans le cas contraire.

providerOptions?:

Record<string, Record<string, JSONValue>>

Options supplémentaires propres au fournisseur, transmises au fournisseur LLM sous-jacent. La structure est { providerName: { optionKey: value } }. Par exemple : { openai: { reasoningEffort: 'high' }, anthropic: { maxTokens: 1000 } }.

openai?:

Record<string, JSONValue>

Options propres à OpenAI. Exemple : { reasoningEffort: 'high' }

anthropic?:

Record<string, JSONValue>

Options propres à Anthropic. Exemple : { maxTokens: 1000 }

google?:

Record<string, JSONValue>

Options propres à Google. Exemple : { safetySettings: [...] }

[providerName]?:

Record<string, JSONValue>

Autres options propres au fournisseur. La clé correspond au nom du fournisseur et la valeur est un enregistrement d’options propres à celui-ci.

runId?:

string

ID unique de cette exécution de génération. Utile pour le suivi et le débogage.

requestContext?:

RequestContext

Contexte de requête destiné à l’injection de dépendances et aux informations contextuelles.

tracingContext?:

TracingContext

Contexte de traçage permettant de créer des spans enfants et d’ajouter des métadonnées. Injecté automatiquement lors de l’utilisation du système de traçage de Mastra.

currentSpan?:

Span

Span actuel servant à créer des spans enfants et à ajouter des métadonnées. Utilisez-le pour créer des spans enfants personnalisés ou mettre à jour les attributs d’un span pendant l’exécution.

tracingOptions?:

TracingOptions

Options de configuration du traçage.

metadata?:

Record<string, any>

Métadonnées à ajouter au span racine de la trace. Utiles pour ajouter des attributs personnalisés comme des ID utilisateur, des ID de session ou des indicateurs de fonctionnalité.

requestContextKeys?:

string[]

Clés RequestContext supplémentaires à extraire en tant que métadonnées de cette trace. Prend en charge la notation par points pour les valeurs imbriquées (par ex. 'user.id').

traceId?:

string

ID de trace à utiliser pour cette exécution (1 à 32 caractères hexadécimaux). Lorsqu’il est fourni, cette trace appartient à la trace indiquée.

parentSpanId?:

string

ID du span parent à utiliser pour cette exécution (1 à 16 caractères hexadécimaux). Lorsqu’il est fourni, le span racine est créé comme enfant de ce span.

tags?:

string[]

Tags à appliquer à cette trace. Libellés sous forme de chaînes permettant de classer et de filtrer les traces.

versions?:

VersionOverrides

Remplacements de version propres à chaque invocation pour la délégation aux sous-agents. Ils sont fusionnés avec les versions définies au niveau de l’instance Mastra et propagés automatiquement dans les appels de sous-agents via requestContext. Nécessite le package editor. Consultez la gestion des versions dans Editor.

VersionOverrides

agents?:

Record<string, VersionSelector>

Correspondance entre les ID d’agents et leurs sélecteurs de version.

VersionSelector

versionId?:

string

Cible une version précise par son ID.

status?:

'draft' | 'published'

Cible la dernière version possédant ce statut de publication.

untilIdle?:

boolean | { maxIdleMs?: number }

Lorsque cette option est définie, maintient le flux ouvert pendant les continuations de tâches en arrière-plan. L’agent invoque automatiquement le LLM de nouveau lorsque les tâches en arrière-plan se terminent, et diffuse les tours de continuation dans le même fullStream. Transmettez true pour utiliser les paramètres par défaut (délai d’inactivité de 5 min), ou un objet avec maxIdleMs pour les configurer. Nécessite la mémoire. Remplace la méthode autonome streamUntilIdle().

maxIdleMs?:

number

Ferme le flux externe après ce nombre de millisecondes d’inactivité entre les tours. Le minuteur ne s’exécute que lorsque le composant d’encapsulation se trouve entre deux tours. Valeur par défaut : 5 minutes.

Valeurs renvoyées

stream:

MastraModelOutput<Output>

Renvoie une instance MastraModelOutput qui donne accès à la sortie diffusée en streaming.

traceId?:

string

ID de trace associé à cette exécution lorsque le traçage est activé. Utilisez-le pour corréler les journaux et déboguer le flux d’exécution.

spanId?:

string

ID du span racine associé à cette exécution lorsque le traçage est activé. Utilisez-le pour les recherches et les corrélations au niveau du span.

Exemple d’utilisation détaillé

Format Mastra (par défaut)

index.ts

import { stepCountIs } from 'ai-v5'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// Access text stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

// or access full stream
for await (const chunk of stream.fullStream) {
  console.log(chunk)
}

// Get full text after streaming
const fullText = await stream.text

Format AI SDK v5+

Pour utiliser le flux avec AI SDK v5 (et les versions ultérieures), vous pouvez le convertir à l’aide de notre fonction utilitaire toAISdkStream.

index.ts

import { stepCountIs, createUIMessageStreamResponse } from 'ai'
import { toAISdkStream } from '@mastra/ai-sdk'

const stream = await agent.stream('Tell me a story', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
})

// In an API route for frontend integration
return createUIMessageStreamResponse({
  stream: toAISdkStream(stream, { from: 'agent' }),
})

Utilisation des fonctions de rappel

Toutes les fonctions de rappel sont désormais disponibles en tant que propriétés de premier niveau, pour une utilisation plus claire de l’API.

index.ts

const stream = await agent.stream('Tell me a story', {
  onFinish: result => {
    console.log('Streaming finished:', result)
  },
  onStepFinish: step => {
    console.log('Step completed:', step)
  },
  onChunk: chunk => {
    console.log('Received chunk:', chunk)
  },
  onError: ({ error }) => {
    console.error('Streaming error:', error)
  },
  onAbort: event => {
    console.log('Stream aborted:', event)
  },
})

// Process the stream
for await (const chunk of stream.textStream) {
  console.log(chunk)
}

Exemple avancé avec des options

index.ts

import { z } from 'zod'
import { stepCountIs } from 'ai'

await agent.stream('message for agent', {
  stopWhen: stepCountIs(3), // Stop after 3 steps
  modelSettings: {
    temperature: 0.7,
  },
  memory: {
    thread: 'user-123',
    resource: 'test-app',
  },
  toolChoice: 'auto',
  // Structured output with better DX
  structuredOutput: {
    schema: z.object({
      sentiment: z.enum(['positive', 'negative', 'neutral']),
      confidence: z.number(),
    }),
    model: 'openai/gpt-5.6-sol',
    errorStrategy: 'warn',
  },
  // Output processors for streaming response validation
  outputProcessors: [
    new ModerationProcessor({ model: 'openrouter/openai/gpt-oss-safeguard-20b' }),
    new BatchPartsProcessor({ maxBatchSize: 3, maxWaitTime: 100 }),
  ],
})

Transport WebSocket de Responses

Activez le streaming WebSocket de Responses avec les options du fournisseur. Cela ne s’applique qu’aux appels en streaming et est pris en charge pour les modèles OpenAI directs ainsi que pour les déploiements Responses d’Azure OpenAI. Si le streaming WebSocket n’est pas disponible, Mastra utilise le streaming HTTP comme solution de repli. Par défaut, Mastra ferme le WebSocket à la fin du flux.

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket', // 'websocket' | 'fetch' | 'auto'
      websocket: {
        url: 'wss://api.openai.com/v1/responses',
        closeOnFinish: true, // default
      },
    },
  },
})

Pour Azure OpenAI, configurez la passerelle avec useResponsesAPI: true, puis utilisez providerOptions.azure.transport.

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    azure: {
      transport: 'websocket',
      store: false,
      websocket: { closeOnFinish: true },
    },
  },
})

Pour garder la connexion ouverte après la fin du flux, définissez closeOnFinish: false et fermez-la manuellement.

index.ts

const stream = await agent.stream('Hello', {
  providerOptions: {
    openai: {
      transport: 'websocket',
      websocket: { closeOnFinish: false },
    },
  },
})

// Later, when you're done with the connection:
stream.transport?.close()

Les connexions WebSocket de Responses traitent une seule réponse à la fois. Mastra rejette les requêtes de continuation qui se chevauchent et incluent previous_response_id sur le même transport WebSocket. Attendez la fin du flux actif avant d’envoyer le tour suivant de la chaîne de réponses.

Voir aussi

• Génération de réponses
• Streaming des réponses
• Approbation des agents