Interface Processor

L’interface Processor définit le contrat de tous les Processors dans Mastra. Les Processors peuvent implémenter une ou plusieurs méthodes pour gérer différentes étapes du pipeline d’exécution de l’Agent.

Quand les méthodes des Processors s’exécutent

Les méthodes des Processors s’exécutent à différents moments du cycle de vie de l’Agent :

┌────────────────────────────────────────────────────────────────────┐
│                        Agent Execution Flow                        │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  User Input                                                        │
│      │                                                             │
│      ▼                                                             │
│  ┌────────────────────────┐                                        │
│  │  processInput          │  ← Runs ONCE at start                  │
│  └───────────┬────────────┘                                        │
│              │                                                     │
│              ▼                                                     │
│  ┌──────────────────────────────────────────────────────────────┐  │
│  │                        Agentic Loop                          │  │
│  │                                                              │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processInputStep      │  ← Runs at EACH step             │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processLLMRequest     │  ← Before provider call          │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │        LLM Execution ──── API Error? ───┐                    │  │
│  │              │                          │                    │  │
│  │              │              ┌───────────┴──────────┐         │  │
│  │              │              │  processAPIError     │         │  │
│  │              │              └──────────────────────┘         │  │
│  │              │                (retry loops back to LLM)      │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processOutputStream   │  ← Runs on EACH stream chunk     │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processLLMResponse    │  ← After stream completes        │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processOutputStep     │  ← Runs after EACH LLM step      │  │
│  │  └───────────┬────────────┘                                  │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │        Tool Execution (if needed)                            │  │
│  │              │                                               │  │
│  │              ▼                                               │  │
│  │  ┌────────────────────────┐                                  │  │
│  │  │  processToolResult     │  ← Runs per tool, after each     │  │
│  │  └───────────┬────────────┘    tool.execute() returns        │  │
│  │              │                                               │  │
│  │              └──────── Loop back if tools called ────────────│  │
│  │                                                              │  │
│  └──────────────────────────────────────────────────────────────┘  │
│              │                                                     │
│              ▼                                                     │
│  ┌────────────────────────┐                                        │
│  │  processOutputResult   │  ← Runs ONCE after completion          │
│  └────────────────────────┘                                        │
│              │                                                     │
│              ▼                                                     │
│        Final Response                                              │
│                                                                    │
└────────────────────────────────────────────────────────────────────┘

Méthode	Moment d’exécution	Cas d’utilisation
processInput	Une fois au début, avant la boucle agentique	Valider ou transformer l’entrée utilisateur initiale, ajouter du contexte
processInputStep	À chaque étape de la boucle agentique, avant chaque appel au LLM	Transformer les messages entre les étapes, gérer les résultats des Tools
processLLMRequest	Après la conversion de la requête LLM, avant l’appel au Provider	Réécrire le LanguageModelV2Prompt sortant pour l’appel actuel sans persister les modifications
processAPIError	Lorsqu’un appel à l’API du LLM échoue	Examiner les rejets de l’API, modifier éventuellement l’état ou les messages et demander une nouvelle tentative
processOutputStream	À chaque fragment diffusé pendant la réponse du LLM	Filtrer ou modifier le contenu diffusé, détecter des motifs en temps réel
processLLMResponse	Une fois l’étape LLM terminée et les fragments du flux collectés	Capturer ou mettre en cache la réponse complète, exécuter les effets secondaires postérieurs à l’appel associés à processLLMRequest
processOutputStep	Après chaque réponse du LLM, avant l’exécution des Tools	Valider la qualité de la sortie, mettre en œuvre des garde-fous avec nouvelle tentative
processToolResult	Pour chaque Tool, après le retour de tool.execute() et avant l’ajout du résultat à la liste des messages	Rechercher une injection de prompt dans la sortie du Tool, masquer les champs sensibles, interrompre en cas de violation de politique
processOutputResult	Une fois la génération terminée	Post-traiter la réponse finale, journaliser les résultats

Définition de l’interface

interface Processor<TId extends string = string, TTripwireMetadata = unknown> {
  readonly id: TId
  readonly name?: string
  readonly description?: string
  /** Index of this processor in the workflow (set at runtime when combining processors). */
  processorIndex?: number
  /** When true, processOutputStream also receives `data-*` chunks. Default: false. */
  processDataParts?: boolean

  /** Callback invoked when this processor detects a violation, regardless of strategy. */
  onViolation?: (violation: ProcessorViolation) => void | Promise<void>

  processInput?(
    args: ProcessInputArgs<TTripwireMetadata>,
  ): Promise<ProcessInputResult> | ProcessInputResult

  processInputStep?(
    args: ProcessInputStepArgs<TTripwireMetadata>,
  ):
    | Promise<ProcessInputStepResult | MessageList | MastraDBMessage[] | undefined | void>
    | ProcessInputStepResult
    | MessageList
    | MastraDBMessage[]
    | void
    | undefined

  processLLMRequest?(
    args: ProcessLLMRequestArgs<TTripwireMetadata>,
  ): Promise<ProcessLLMRequestResult> | ProcessLLMRequestResult

  processLLMResponse?(
    args: ProcessLLMResponseArgs<TTripwireMetadata>,
  ): Promise<ProcessLLMResponseResult> | ProcessLLMResponseResult

  processAPIError?(
    args: ProcessAPIErrorArgs<TTripwireMetadata>,
  ): Promise<ProcessAPIErrorResult | void> | ProcessAPIErrorResult | void

  processOutputStream?(
    args: ProcessOutputStreamArgs<TTripwireMetadata>,
  ): Promise<ChunkType | null | undefined>

  processOutputStep?(args: ProcessOutputStepArgs<TTripwireMetadata>): ProcessorMessageResult

  processToolResult?(args: ProcessToolResultArgs<TTripwireMetadata>): ProcessorMessageResult

  processOutputResult?(args: ProcessOutputResultArgs<TTripwireMetadata>): ProcessorMessageResult
}

Propriétés

id:

string

Identifiant unique du Processor. Utilisé pour le traçage et le débogage.

name?:

string

Nom d’affichage facultatif du Processor. Utilise id par défaut s’il n’est pas fourni.

description?:

string

Description lisible facultative affichée dans le traçage et Studio.

processorIndex?:

number

Position du Processor dans la liste combinée des Processors. Définie à l’exécution par Mastra lorsque les Processors sont fusionnés avec la Memory, le Workspace et les remplacements propres à chaque appel. Vous ne définissez pas cette valeur vous-même.

processDataParts?:

boolean

Lorsque cette valeur est vraie, la méthode processOutputStream reçoit également les fragments data-* émis par les Tools via writer.custom(). La valeur par défaut est false.

onViolation?:

(violation: ProcessorViolation) => void | Promise<void>

Callback facultatif invoqué lorsque le Processor détecte une violation de politique, quelle que soit la stratégie (blocage ou avertissement). Utilisez-le pour des effets secondaires tels que l’envoi d’alertes, la journalisation dans des systèmes externes ou l’envoi d’e-mails aux utilisateurs. Les erreurs levées par ce callback sont interceptées silencieusement afin de ne pas perturber la logique du Processor. L’objet de violation contient processorId, message et un champ detail propre au Processor.

Arguments de message

La plupart des méthodes des Processors reçoivent à la fois messages et messageList. Ces deux valeurs désignent la même conversation sous-jacente, mais l’exposent différemment.

messages ou messageList

• messages : tableau simple d’objets MastraDBMessage, limité à l’étape actuelle. Pour processInput et processInputStep, il exclut les messages système. Pour processOutputResult et processOutputStep, il inclut la dernière réponse du LLM. Le tableau s’appuie sur messageList ; toute modification sur place du content.parts d’un message est donc visible par les Processors en aval et lors de la persistance.
• messageList : instance active de MessageList qui sous-tend l’exécution. Elle expose des vues filtrées (entrée, réponse, éléments mémorisés, totalité), plusieurs formats de sortie (db, ui, core) et des méthodes permettant de modifier la conversation.

Utilisez messages si vous devez uniquement lire, parcourir avec map ou modifier légèrement les champs des messages de l’étape actuelle. Utilisez messageList lorsque vous devez :

• Lire les messages d’une autre étape, par exemple les messages d’entrée pendant le traitement de la sortie.
• Ajouter, supprimer ou remplacer des messages entiers.
• Convertir les messages dans un autre format, tel que des messages UI ou core pour une API tierce.

messages est toujours dérivé de messageList. Modifier messageList constitue donc la méthode canonique pour ajouter, supprimer ou réordonner des messages. Pour les modifications sur place du contenu d’un message, par exemple la réécriture de content.parts, modifier directement messages revient au même. Si vous renvoyez un nouveau tableau depuis messages, Mastra le réconcilie avec messageList pour l’étape actuelle.

Persistance

Lorsque la Memory est activée, seul le contenu final de messageList, une fois tous les Processors terminés, est persisté dans le stockage. Les deux styles de retour sont équivalents pour la persistance :

• Lorsque vous modifiez directement messageList ou renvoyez la même instance de MessageList, les mutations enregistrées sont appliquées sur place ; la conversation sauvegardée reflète donc vos modifications.
• Lorsque vous renvoyez un MastraDBMessage[] ou { messages, systemMessages }, Mastra réconcilie le tableau renvoyé avec messageList pour l’étape actuelle, en supprimant les messages absents et en remplaçant les messages système.

Renvoyer une autre instance de MessageList constitue une erreur. Modifiez toujours celle qui est transmise à votre Processor.

Lire le texte d’un message

MastraDBMessage.content utilise un objet structuré. Les chaînes ne sont pas prises en charge. La méthode canonique pour lire le texte de l’utilisateur ou de l’assistant consiste à utiliser content.parts :

import type { MastraDBMessage } from '@mastra/core/memory'

function getText(message: MastraDBMessage): string {
  let text = ''

  if (message.content.parts) {
    for (const part of message.content.parts) {
      if (part.type === 'text' && typeof part.text === 'string') {
        text += part.text
      }
    }
  }

  // Fallback for legacy messages that only have the flattened `content` string
  if (!text && typeof message.content.content === 'string') {
    text = message.content.content
  }

  return text
}

Points essentiels :

• message.content.parts est la source principale. Un message peut contenir plusieurs parties, y compris des parties non textuelles telles que des appels de Tools, des résultats de Tools et des fichiers. Filtrez avec part.type === 'text' avant de lire part.text.
• message.content.content est une chaîne aplatie conservée pour la rétrocompatibilité. Utilisez-la uniquement comme solution de repli lorsque parts est vide ou absent.
• message.content n’est jamais une simple chaîne dans MastraDBMessage. Les anciennes structures CoreMessage peuvent être des chaînes, mais les Processors reçoivent toujours un MastraDBMessage.

Méthodes

processInput

Traite les messages d’entrée avant leur envoi au LLM. S’exécute une fois au début de l’exécution de l’Agent.

processInput?(args: ProcessInputArgs): Promise<ProcessInputResult> | ProcessInputResult;

ProcessInputArgs

messages:

MastraDBMessage[]

Messages de l’utilisateur et de l’assistant à traiter (hors messages système).

systemMessages:

CoreMessage[]

Tous les messages système (instructions de l’Agent, contexte de Memory, messages fournis par l’utilisateur). Peuvent être modifiés et renvoyés.

messageList:

MessageList

Instance complète de MessageList pour une gestion avancée des messages.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Lève une erreur TripWire qui arrête l’exécution. Transmettez retry: true pour demander au LLM de réessayer l’étape avec un retour.

retryCount:

number

Nombre de fois où les Processors ont déclenché une nouvelle tentative pour cette génération. Utilisez cette valeur pour limiter les tentatives. Toujours transmise par Mastra ; commence à 0.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, avec des métadonnées d’exécution telles que threadId et resourceId.

ProcessInputResult

La méthode peut renvoyer l’un des trois types suivants :

MastraDBMessage[]:

array

Tableau des messages transformés. Les messages système restent inchangés.

MessageList:

MessageList

La même instance de messageList que celle transmise. Indique que vous l’avez modifiée directement.

{ messages, systemMessages }:

object

Objet contenant à la fois les messages transformés et les messages système modifiés.

---

processInputStep

Traite les messages d’entrée à chaque étape de la boucle agentique, avant leur envoi au LLM. Contrairement à processInput, qui ne s’exécute qu’une fois au début, cette méthode s’exécute à chaque étape, y compris lors de la poursuite des appels de Tools.

processInputStep?<TTripwireMetadata = unknown>(
  args: ProcessInputStepArgs<TTripwireMetadata>,
):
  | Promise<ProcessInputStepResult | MessageList | MastraDBMessage[] | void | undefined>
  | ProcessInputStepResult
  | MessageList
  | MastraDBMessage[]
  | void
  | undefined;

Ordre d’exécution dans la boucle agentique

1. processInput (une fois au début)
2. processInputStep depuis inputProcessors (à chaque étape, avant l’appel au LLM)
3. Callback prepareStep (s’exécute dans le pipeline processInputStep, après inputProcessors)
4. processLLMRequest depuis inputProcessors (après la conversion du prompt, avant l’appel au Provider)
5. Exécution du LLM
6. processOutputStream depuis outputProcessors (à chaque fragment diffusé)
7. processLLMResponse depuis inputProcessors (une fois le flux terminé, en association avec processLLMRequest)
8. processOutputStep depuis outputProcessors (après la réponse du LLM, avant l’exécution des Tools)
9. Exécution des Tools (si nécessaire)
10. Reprise à l’étape 2 si des Tools ont été appelés

ProcessInputStepArgs

messages:

MastraDBMessage[]

Tous les messages, y compris les appels de Tools et les résultats des étapes précédentes (instantané en lecture seule).

messageList:

MessageList

Instance de MessageList permettant de gérer les messages. Peut être modifiée directement ou renvoyée dans le résultat.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0). L’étape 0 correspond à l’appel initial au LLM.

steps:

StepResult[]

Résultats des étapes précédentes, notamment text, toolCalls et toolResults.

systemMessages:

CoreMessage[]

Tous les messages système (instantané en lecture seule). Renvoyez-les dans le résultat pour les remplacer.

model:

MastraLanguageModelV2

Modèle actuellement utilisé. Renvoyez un autre modèle dans le résultat pour en changer.

toolChoice?:

ToolChoice

Paramètre actuel de sélection des Tools ('auto', 'none', 'required' ou Tool spécifique).

activeTools?:

string[]

Noms des Tools actuellement actifs. Renvoyez un tableau filtré pour limiter les Tools.

tools?:

ToolSet

Tools actuellement disponibles pour cette étape. Renvoyez-les dans le résultat pour ajouter ou remplacer des Tools.

providerOptions?:

SharedV2ProviderOptions

Options propres au Provider (par exemple, cacheControl d’Anthropic ou reasoningEffort d’OpenAI).

modelSettings?:

CallSettings

Paramètres du modèle tels que temperature, maxTokens et topP.

structuredOutput?:

StructuredOutputOptions

Configuration de la sortie structurée (schéma, mode de sortie). Renvoyez-la dans le résultat pour la modifier.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Lève une erreur TripWire qui arrête l’exécution. Transmettez retry: true pour demander au LLM de réessayer l’étape avec un retour.

retryCount:

number

Nombre actuel de tentatives provenant de ProcessorContext. Commence à 0 ; utilisez-le pour plafonner les nouvelles tentatives déclenchées par les Processors.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

ProcessInputStepResult

processInputStep peut renvoyer plusieurs formes :

• Objet ProcessInputStepResult : remplacez n’importe quelle combinaison des propriétés ci-dessous pour cette étape (décrites ensuite).
• MessageList : renvoyez la même instance de messageList pour signaler que vous avez modifié les messages sur place.
• MastraDBMessage[] : renvoyez un tableau de messages transformés. Il remplace les messages de l’étape.
• void ou undefined : ne renvoyez rien pour laisser l’étape inchangée.

La forme objet peut renvoyer n’importe quelle combinaison des propriétés suivantes :

model?:

LanguageModelV2 | string

Change le modèle pour cette étape. Peut être une instance de modèle ou un ID de routeur tel que 'openai/gpt-5.5'.

toolChoice?:

ToolChoice

Change le comportement de sélection des Tools pour cette étape.

activeTools?:

string[]

Filtre les Tools disponibles pour cette étape.

tools?:

ToolSet

Remplace ou modifie les Tools pour cette étape. Utilisez la syntaxe spread pour les fusionner : { tools: { ...tools, newTool } }.

messages?:

MastraDBMessage[]

Remplace tous les messages. Ne peut pas être utilisé avec messageList.

messageList?:

MessageList

Renvoie la même instance de messageList (indique que vous l’avez modifiée). Ne peut pas être utilisé avec messages.

systemMessages?:

CoreMessage[]

Remplace tous les messages système pour cette étape uniquement.

providerOptions?:

SharedV2ProviderOptions

Change les options propres au Provider pour cette étape.

modelSettings?:

CallSettings

Change les paramètres du modèle pour cette étape.

structuredOutput?:

StructuredOutputOptions

Change la configuration de la sortie structurée pour cette étape.

Chaînage des Processors

Lorsque plusieurs Processors implémentent processInputStep, ils s’exécutent dans l’ordre et les modifications se propagent dans la chaîne :

Processor 1: receives { model: 'gpt-5.4' } → returns { model: 'gpt-5.4-mini' }
Processor 2: receives { model: 'gpt-5.4-mini' } → returns { toolChoice: 'none' }
Final: model = 'gpt-5.4-mini', toolChoice = 'none'

Isolation des messages système

Les messages système sont réinitialisés à leurs valeurs d’origine au début de chaque étape. Les modifications apportées dans processInputStep n’affectent que l’étape actuelle, pas les étapes suivantes.

Cas d’utilisation

• Changer dynamiquement de modèle selon le numéro d’étape ou le contexte
• Désactiver les Tools après un certain nombre d’étapes
• Ajouter ou remplacer dynamiquement des Tools selon le contexte de la conversation
• Transformer les types de parties de message entre les Providers (par exemple, reasoning → thinking pour Anthropic)
• Modifier les messages selon le numéro d’étape ou le contexte accumulé
• Ajouter des instructions système propres à une étape
• Ajuster les options du Provider à chaque étape (par exemple, le contrôle du cache)
• Modifier le schéma de sortie structurée selon le contexte de l’étape

---

processLLMRequest

Traite la requête LLM finale après la conversion de la MessageList en LanguageModelV2Prompt par Mastra et avant l’appel au Provider. Utilisez cette méthode pour les réécritures temporaires tenant compte du modèle, qui ne doivent affecter que la requête sortante actuelle.

Les modifications du prompt renvoyé sont transmises au modèle uniquement pour l’appel actuel. Elles ne sont pas repersistées dans MessageList, la Memory, l’historique de l’UI ni les appels ultérieurs au Provider.

processLLMRequest?(
  args: ProcessLLMRequestArgs,
): Promise<ProcessLLMRequestResult> | ProcessLLMRequestResult;

ProcessLLMRequestArgs

prompt:

LanguageModelV2Prompt

Prompt de la requête LLM qui sera envoyé au Provider pour cet appel.

model:

MastraLanguageModel

Modèle résolu qui recevra le prompt. Utilisez-le pour limiter les réécritures propres au Provider.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0). L’étape 0 correspond à l’appel initial au LLM.

steps:

StepResult[]

Résultats des étapes précédentes, notamment text, toolCalls et toolResults.

state:

Record<string, unknown>

État propre au Processor, conservé pour tous les appels de méthode au sein de cette requête.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Lève une erreur TripWire qui arrête l’exécution et émet un fragment tripwire.

retryCount:

number

Nombre actuel de tentatives provenant de ProcessorContext. Commence à 0 ; utilisez-le pour plafonner les nouvelles tentatives déclenchées par les Processors.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

writer?:

ProcessorStreamWriter

Writer de flux permettant d’émettre des fragments de données personnalisés pendant la diffusion. Appelez writer.custom() pour émettre un fragment data-*.

abortSignal?:

AbortSignal

Signal permettant d’annuler l’opération.

Valeur de retour

processLLMRequest renvoie ProcessLLMRequestResult, dont le type est { prompt?: LanguageModelV2Prompt } | undefined | void.

• Renvoyez { prompt } pour remplacer le prompt sortant de l’appel actuel au Provider.
• Renvoyez undefined ou void pour transmettre le prompt d’origine sans modification.

Cas d’utilisation

• Supprimer ou restructurer les parties du prompt propres au Provider avant un appel au modèle
• Normaliser les rôles ou le contenu afin de respecter les exigences d’entrée d’un Provider
• Adapter les formats des résultats de Tools lors d’un changement de Provider au milieu de la boucle

---

processLLMResponse

Traite la réponse du LLM une fois l’étape terminée, ou après la relecture d’une réponse mise en cache, et après la collecte des fragments de réponse par les Processors de sortie. Ce hook est associé à processLLMRequest : utilisez processLLMRequest pour stocker un état, tel qu’une clé de cache, avant l’appel au Provider, puis processLLMResponse pour agir sur la réponse terminée, par exemple en l’écrivant dans un cache.

L’objet state est la même instance que celle transmise à processLLMRequest pour la même étape. Les Processors peuvent donc mettre en relation les traitements antérieurs et postérieurs à l’appel.

processLLMResponse?(
  args: ProcessLLMResponseArgs,
): Promise<ProcessLLMResponseResult> | ProcessLLMResponseResult;

ProcessLLMResponseArgs

chunks:

CachedLLMStepChunk[]

Fragments produits par l’appel au LLM, ou relus depuis le cache, pour cette étape, sous une forme épurée ({ type, payload }).

model:

MastraLanguageModel

Modèle qui a produit, ou aurait produit, la réponse.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0).

steps:

StepResult[]

Toutes les étapes terminées jusqu’à présent, y compris l’étape actuelle.

state:

Record<string, unknown>

État propre au Processor partagé avec processLLMRequest pour la même étape. Utilisez-le pour transmettre des données entre les deux hooks, par exemple une clé de cache.

fromCache:

boolean

Lorsque cette valeur est true, la réponse a été relue depuis un cache par l’intermédiaire de processLLMRequest, qui a renvoyé { response }. Les Processors qui écrivent dans un cache doivent ignorer l’écriture lorsque cette valeur est true.

warnings?:

LanguageModelV2CallWarning[]

Avertissements signalés par l’appel au modèle de langage, par exemple des paramètres non pris en charge.

request?:

unknown

Corps de la requête au Provider, lorsqu’il est disponible. Utile pour le traçage.

rawResponse?:

unknown

Réponse brute du Provider, lorsqu’elle est disponible. Utile pour le traçage.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Lève une erreur TripWire qui arrête l’exécution.

retryCount:

number

Nombre actuel de tentatives. Commence à 0 ; utilisez-le pour plafonner les nouvelles tentatives déclenchées par les Processors.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

writer?:

ProcessorStreamWriter

Writer de flux permettant d’émettre des fragments de données personnalisés.

abortSignal?:

AbortSignal

Signal permettant d’annuler l’opération.

Valeur de retour

processLLMResponse renvoie ProcessLLMResponseResult, dont le type est undefined | void. La valeur de retour est réservée à de futures extensions.

Cas d’utilisation

• Écrire les réponses du LLM dans un cache après un appel en direct, en association avec la dérivation de la clé de cache dans processLLMRequest
• Journaliser ou enregistrer la réponse complète à des fins d’analyse
• Déclencher des effets secondaires selon la réponse terminée

---

processAPIError

Gère les erreurs de rejet de l’API du LLM avant qu’elles ne deviennent des erreurs finales. Cette méthode s’exécute lorsque l’appel à l’API échoue avec une erreur non réessayable, telle qu’un code d’état 400 ou 422. Contrairement à processOutputStep, qui s’exécute après les réponses réussies, elle s’exécute lorsque l’API rejette la requête.

Ajoutez les Processors qui implémentent processAPIError au tableau errorProcessors d’un Agent.

Les Processors peuvent examiner l’erreur et modifier la requête, par exemple en ajoutant des messages à la messageList. Renvoyez { retry: true } pour effectuer une nouvelle tentative avec l’état modifié.

processAPIError?(args: ProcessAPIErrorArgs): Promise<ProcessAPIErrorResult | void> | ProcessAPIErrorResult | void;

ProcessAPIErrorArgs

error:

unknown

Erreur survenue pendant l’appel à l’API du LLM.

messages:

MastraDBMessage[]

Tous les messages au moment de l’erreur.

messageList:

MessageList

Instance de MessageList permettant de gérer les messages. Modifiez-la pour changer la requête avant une nouvelle tentative.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0).

steps:

StepResult[]

Toutes les étapes terminées jusqu’à présent.

state:

Record<string, unknown>

État propre au Processor, conservé pour tous les appels de méthode au sein de cette requête.

retryCount:

number

Nombre actuel de tentatives pour les gestionnaires d’erreurs. Utilisez-le pour limiter les nouvelles tentatives.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement.

writer?:

ProcessorStreamWriter

Writer de flux permettant d’émettre des fragments de données personnalisés pendant la diffusion. Appelez writer.custom() pour émettre un fragment data-*.

requestContext?:

RequestContext

Contexte de requête transmis depuis l’appel de l’Agent.

abortSignal?:

AbortSignal

Signal permettant d’annuler l’opération.

ProcessAPIErrorResult

retry:

boolean

Indique s’il faut réessayer l’appel au LLM après l’application des modifications.

Cas d’utilisation

• Gérer les rejets propres à l’API en modifiant la requête puis en réessayant
• Convertir les erreurs non réessayables en erreurs réessayables grâce à des modifications de la requête
• Mettre en œuvre des stratégies de récupération d’erreur propres au modèle

Exemple : récupération d’erreur personnalisée

src/mastra/processors/error-recovery.ts

import { APICallError } from '@ai-sdk/provider'
import type { Processor, ProcessAPIErrorArgs, ProcessAPIErrorResult } from '@mastra/core/processors'

export class ErrorRecoveryProcessor implements Processor {
  id = 'error-recovery'

  processAPIError({
    error,
    messageList,
    retryCount,
  }: ProcessAPIErrorArgs): ProcessAPIErrorResult | void {
    // Only retry once
    if (retryCount > 0) return

    // Check for a specific API error
    if (APICallError.isInstance(error) && error.message.includes('context length exceeded')) {
      // Trim older messages to fit within context
      const messages = messageList.get.all.db()
      if (messages.length > 4) {
        messageList.removeByIds([messages[1]!.id, messages[2]!.id])
        return { retry: true }
      }
    }
  }
}

---

processOutputStream

Traite les fragments de sortie diffusés avec une gestion intégrée de l’état. Permet aux Processors d’accumuler des fragments et de prendre des décisions à partir d’un contexte plus large.

processOutputStream?(args: ProcessOutputStreamArgs): Promise<ChunkType | null | undefined>;

ProcessOutputStreamArgs

part:

ChunkType

Fragment actuel du flux en cours de traitement.

streamParts:

ChunkType[]

Tous les fragments rencontrés jusqu’à présent dans le flux.

state:

Record<string, unknown>

État mutable propre au Processor, conservé pour chaque fragment et chaque appel de méthode au sein d’une même requête. Un nouvel objet d’état est créé pour chaque nouvel appel à generate ou stream.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le flux. Lève une erreur TripWire qui termine le flux et émet un fragment tripwire. Transmettez retry: true pour demander une nouvelle tentative au LLM au lieu de terminer.

retryCount:

number

Nombre actuel de tentatives provenant de ProcessorContext. Commence à 0 ; utilisez-le pour plafonner les nouvelles tentatives déclenchées par les Processors.

messageList?:

MessageList

Instance de MessageList permettant d’accéder à l’historique de la conversation.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

writer?:

ProcessorStreamWriter

Writer de flux permettant de renvoyer des fragments de données personnalisés au client. Appelez writer.custom() pour émettre des fragments typés data-*. Disponible pendant la diffusion.

Valeur de retour

processOutputStream renvoie Promise<ChunkType | null | undefined>.

• Renvoyez le ChunkType pour émettre le fragment. Renvoyez le part d’origine pour l’émettre sans modification, ou un nouveau ChunkType pour émettre un fragment modifié.
• Renvoyez null pour supprimer le fragment. Rien n’est envoyé au Processor suivant ni au client.
• Renvoyez undefined, y compris la valeur undefined implicite d’une instruction return; ou d’une méthode arrivant à sa fin, pour supprimer le fragment. null et undefined se comportent de la même manière.

La suppression d’un fragment n’affecte que ce fragment. Le flux continue et le fragment suivant est toujours traité. Pour arrêter entièrement le flux, appelez abort().

---

processOutputResult

Traite le résultat de sortie complet une fois la diffusion ou la génération terminée.

processOutputResult?(args: ProcessOutputResultArgs): ProcessorMessageResult;

ProcessOutputResultArgs

messages:

MastraDBMessage[]

Messages de réponse générés.

messageList:

MessageList

Instance de MessageList permettant de gérer les messages.

state:

Record<string, unknown>

État propre au Processor, conservé pour tous les appels de méthode au sein de cette requête. Partagé avec processOutputStream et les autres méthodes.

result:

OutputResult

Résultat de génération résolu contenant text (texte accumulé), usage (utilisation des tokens avec inputTokens, outputTokens et totalTokens), finishReason (raison de la fin de la génération) et steps (tous les résultats des étapes LLM, chacun avec toolCalls, toolResults, reasoning, sources, files, etc.).

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Lève une erreur TripWire qui arrête l’exécution et émet un fragment tripwire.

retryCount:

number

Nombre actuel de tentatives provenant de ProcessorContext. Commence à 0 ; utilisez-le pour plafonner les nouvelles tentatives déclenchées par les Processors.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

writer?:

ProcessorStreamWriter

Writer de flux permettant de renvoyer des fragments de données personnalisés au client. Appelez writer.custom() pour émettre des fragments typés data-*. Disponible pendant la diffusion.

---

processOutputStep

Traite la sortie après chaque réponse du LLM dans la boucle agentique, avant l’exécution des Tools. Contrairement à processOutputResult, qui ne s’exécute qu’une fois à la fin, cette méthode s’exécute à chaque étape. Elle est idéale pour mettre en œuvre des garde-fous capables de déclencher de nouvelles tentatives.

processOutputStep?(args: ProcessOutputStepArgs): ProcessorMessageResult;

ProcessOutputStepArgs

messages:

MastraDBMessage[]

Tous les messages, y compris la dernière réponse du LLM.

messageList:

MessageList

Instance de MessageList permettant de gérer les messages.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0).

finishReason?:

string

Raison de fin renvoyée par le LLM (stop, tool-use, length, etc.).

providerMetadata?:

ProviderMetadata

Métadonnées propres au Provider pour l’étape finale, par exemple le traçage des garde-fous AWS Bedrock. Présentes lorsque l’étape du modèle a produit des métadonnées de Provider, y compris pour les blocages du filtre de contenu où steps est vide.

toolCalls?:

ToolCallInfo[]

Appels de Tools effectués pendant cette étape, le cas échéant.

text?:

string

Texte généré pendant cette étape.

usage:

LanguageModelUsage

Utilisation des tokens pour l’étape actuelle (inputTokens, outputTokens, totalTokens).

systemMessages:

CoreMessage[]

Tous les messages système accessibles en lecture et en modification.

steps:

StepResult[]

Toutes les étapes terminées jusqu’à présent, y compris l’étape actuelle.

state:

Record<string, unknown>

État propre au Processor, conservé pour tous les appels de méthode au sein de cette requête. Partagé avec processOutputStream et processOutputResult.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre le traitement. Transmettez retry: true pour demander au LLM de réessayer l’étape.

retryCount:

number

Nombre de fois où les Processors ont déclenché une nouvelle tentative. Utilisez cette valeur pour limiter les tentatives. Toujours transmise par Mastra ; commence à 0.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

Cas d’utilisation

• Mettre en œuvre des garde-fous de qualité capables de demander de nouvelles tentatives
• Valider la sortie du LLM avant l’exécution des Tools
• Ajouter une journalisation ou des métriques propres à chaque étape
• Mettre en œuvre une modération de la sortie avec possibilité de nouvelle tentative

Exemple : garde-fou de qualité avec nouvelle tentative

src/mastra/processors/quality-guardrail.ts

import type { Processor } from '@mastra/core/processors'

export class QualityGuardrail implements Processor {
  id = 'quality-guardrail'

  async processOutputStep({ text, abort, retryCount }) {
    const score = await evaluateResponseQuality(text)

    if (score < 0.7) {
      if (retryCount < 3) {
        // Request retry with feedback for the LLM
        abort('Response quality too low. Please provide more detail.', {
          retry: true,
          metadata: { qualityScore: score },
        })
      } else {
        // Max retries reached, block the response
        abort('Response quality too low after multiple attempts.')
      }
    }

    return []
  }
}

processToolResult

Traite le résultat d’un Tool après le retour de tool.execute() et avant que ce résultat soit ajouté à la liste des messages ou transmis à l’appel suivant au LLM. Cette méthode est symétrique à processOutputStep, qui se déclenche avant l’exécution du Tool. Utilisez-la pour rechercher une injection de prompt dans la sortie du Tool, masquer les champs sensibles ou interrompre l’exécution avec abort('reason', { retry: true }).

Pour remplacer le résultat du Tool, modifiez messageList sur place à l’aide de messageList.updateToolInvocation. Le runtime relit le résultat postérieur au Processor depuis la liste des messages et remplace le fragment de flux du résultat du Tool en aval avant sa mise en file d’attente. Les clients de diffusion voient ainsi la valeur traitée.

Cette méthode ne se déclenche pas lorsque tool.execute() lève une erreur ; elle est appelée uniquement lors des exécutions réussies d’un Tool pour lesquelles un résultat est disponible.

processToolResult?(args: ProcessToolResultArgs): ProcessorMessageResult;

ProcessToolResultArgs

messages:

MastraDBMessage[]

Tous les messages, y compris le message actuel de l’assistant contenant l’appel du Tool.

messageList:

MessageList

Instance de MessageList permettant de gérer les messages. Appelez updateToolInvocation pour remplacer le résultat du Tool par une valeur masquée ou transformée.

stepNumber:

number

Numéro de l’étape actuelle (indexé à partir de 0).

toolName:

string

Nom du Tool qui a été exécuté.

toolCallId:

string

Identifiant unique de cet appel de Tool précis.

args:

unknown

Arguments transmis par le LLM au Tool.

result:

unknown

Valeur renvoyée par le Tool. Pour les Tools exécutés côté client, il s’agit de la sortie de tool.execute() après son passage dans ensureSerializable. Pour les Tools exécutés par le Provider, par exemple web_search d’Anthropic, il s’agit du résultat brut provenant du flux du Provider, qui ne passe pas dans ensureSerializable.

providerExecuted?:

boolean

Indique si ce résultat provient d’un Tool exécuté par le Provider, tel que web_search d’Anthropic. La valeur par défaut est undefined pour les Tools exécutés côté client.

systemMessages:

CoreMessage[]

Tous les messages système accessibles en lecture.

steps:

StepResult[]

Toutes les étapes terminées jusqu’à présent.

state:

Record<string, unknown>

État propre au Processor, conservé pour tous les appels de méthode au sein de cette requête. Partagé avec les autres méthodes du même Processor.

abort:

(reason?: string, options?: { retry?: boolean; metadata?: unknown }) => never

Fonction permettant d’interrompre l’exécution. Transmettez retry: true pour demander au LLM de réessayer l’étape en utilisant le motif d’interruption comme retour.

retryCount:

number

Nombre de fois où les Processors ont déclenché une nouvelle tentative. Commence à 0.

tracingContext?:

TracingContext

Contexte de traçage pour l’observabilité.

requestContext?:

RequestContext

Contexte limité à la requête, accompagné de métadonnées d’exécution.

Cas d’utilisation

• Rechercher une injection de prompt dans la sortie du Tool avant que le LLM ne la voie.
• Masquer les champs sensibles dans les retours des Tools (PII, secrets, identifiants).
• Interrompre l’exécution lorsqu’un Tool renvoie du contenu qui enfreint la politique.
• Journaliser ou instrumenter les retours des Tools à des fins de conformité ou d’audit.

Exemple : masquer les champs sensibles

src/mastra/processors/redact-tool-result.ts

import type { Processor } from '@mastra/core/processors'

export class RedactToolResult implements Processor {
  id = 'redact-tool-result'

  async processToolResult({ toolName, toolCallId, args, result, messageList }) {
    if (toolName !== 'lookup-customer') return

    const redacted = {
      ...(result as Record<string, unknown>),
      ssn: '[REDACTED]',
      email: '[REDACTED]',
    }

    messageList.updateToolInvocation({
      type: 'tool-invocation',
      toolInvocation: {
        state: 'result',
        toolCallId,
        toolName,
        args,
        result: redacted,
      },
    })
  }
}

Exemple : bloquer l’injection de prompt dans la sortie d’un Tool

src/mastra/processors/scan-tool-result.ts

import type { Processor } from '@mastra/core/processors'

export class ScanToolResult implements Processor {
  id = 'scan-tool-result'

  async processToolResult({ result, abort }) {
    const text = typeof result === 'string' ? result : JSON.stringify(result)
    if (containsPromptInjection(text)) {
      abort('blocked by scan-tool-result: suspected prompt injection')
    }
  }
}

function containsPromptInjection(text: string): boolean {
  return /ignore (all )?(previous|prior) instructions/i.test(text)
}

Types de Processors

Mastra fournit des alias de types afin de garantir que les Processors implémentent les méthodes requises :

// Must implement processInput, processInputStep, processLLMRequest, or processLLMResponse (or any combination)
type InputProcessor = Processor &
  (
    | { processInput: required }
    | { processInputStep: required }
    | { processLLMRequest: required }
    | { processLLMResponse: required }
  )

// Must implement processOutputStream, processOutputStep, OR processOutputResult (or any combination)
type OutputProcessor = Processor &
  (
    | { processOutputStream: required }
    | { processOutputStep: required }
    | { processOutputResult: required }
  )

// Must implement processAPIError
type ErrorProcessor = Processor & { processAPIError: required }

Configurez les Processors qui implémentent processAPIError dans errorProcessors :

const agent = new Agent({
  id: 'agent',
  errorProcessors: [new PrefillErrorHandler()],
})

Exemples d’utilisation

Processor d’entrée de base

src/mastra/processors/lowercase.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class LowercaseProcessor implements Processor {
  id = 'lowercase'

  async processInput({ messages }): Promise<MastraDBMessage[]> {
    return messages.map(msg => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map(part =>
          part.type === 'text' ? { ...part, text: part.text.toLowerCase() } : part,
        ),
      },
    }))
  }
}

Processor par étape avec processInputStep

src/mastra/processors/dynamic-model.ts

import type {
  Processor,
  ProcessInputStepArgs,
  ProcessInputStepResult,
} from '@mastra/core/processors'

export class DynamicModelProcessor implements Processor {
  id = 'dynamic-model'

  async processInputStep({
    stepNumber,
    steps,
    toolChoice,
  }: ProcessInputStepArgs): Promise<ProcessInputStepResult> {
    // Use a fast model for initial response
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }

    // Switch to powerful model after tool calls
    if (steps.length > 0 && steps[steps.length - 1].toolCalls?.length) {
      return { model: 'openai/gpt-5.6-sol' }
    }

    // Disable tools after 5 steps to force completion
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }

    return {}
  }
}

Transformation de messages avec processInputStep

src/mastra/processors/reasoning-transformer.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'

export class ReasoningTransformer implements Processor {
  id = 'reasoning-transformer'

  async processInputStep({ messages, messageList }) {
    // Transform reasoning parts to thinking parts at each step
    // This is useful when switching between model providers
    for (const msg of messages) {
      if (msg.role === 'assistant' && msg.content.parts) {
        for (const part of msg.content.parts) {
          if (part.type === 'reasoning') {
            ;(part as any).type = 'thinking'
          }
        }
      }
    }
    return messageList
  }
}

Processor hybride (entrée et sortie)

src/mastra/processors/content-filter.ts

import type { Processor } from '@mastra/core/processors'
import type { MastraDBMessage } from '@mastra/core/memory'
import type { ChunkType } from '@mastra/core/stream'

export class ContentFilter implements Processor {
  id = 'content-filter'
  private blockedWords: string[]

  constructor(blockedWords: string[]) {
    this.blockedWords = blockedWords
  }

  async processInput({ messages, abort }): Promise<MastraDBMessage[]> {
    for (const msg of messages) {
      const text = msg.content.parts
        ?.filter(p => p.type === 'text')
        .map(p => p.text)
        .join(' ')

      if (this.blockedWords.some(word => text?.includes(word))) {
        abort('Blocked content detected in input')
      }
    }
    return messages
  }

  async processOutputStream({ part, abort }): Promise<ChunkType | null> {
    if (part.type === 'text-delta') {
      if (this.blockedWords.some(word => part.payload.text.includes(word))) {
        abort('Blocked content detected in output')
      }
    }
    return part
  }
}

Accumulation du flux avec état

src/mastra/processors/word-counter.ts

import type { Processor } from '@mastra/core/processors'
import type { ChunkType } from '@mastra/core/stream'

export class WordCounter implements Processor {
  id = 'word-counter'

  async processOutputStream({ part, state }): Promise<ChunkType> {
    // Initialize state on first chunk
    if (!state.wordCount) {
      state.wordCount = 0
    }

    // Count words in text chunks
    if (part.type === 'text-delta') {
      const words = part.payload.text.split(/\s+/).filter(Boolean)
      state.wordCount += words.length
    }

    // Log word count on finish
    if (part.type === 'finish') {
      console.log(`Total words: ${state.wordCount}`)
    }

    return part
  }
}

Cycle de vie de l’état

Chaque Processor reçoit un objet state dans processLLMRequest, processLLMResponse, processOutputStream, processOutputStep, processOutputResult et processAPIError. L’état possède trois propriétés importantes :

• Propre au Processor : chaque Processor reçoit son propre objet state, indexé par l’id du Processor. Les Processors dont les ID sont différents ne peuvent ni lire ni remplacer l’état des autres.
• Propre à la requête : un nouvel objet d’état est créé au début de chaque appel à agent.generate() ou agent.stream(). L’état ne fuit ni entre les requêtes ni entre les utilisateurs.
• Partagé entre les méthodes : au sein d’une requête, le même objet state est transmis à processLLMRequest (avant l’appel au Provider), processLLMResponse (une fois l’étape terminée), processOutputStream (pour chaque fragment), processOutputStep (après chaque étape LLM), processOutputResult (une fois à la fin) et processAPIError (lorsqu’un appel au LLM échoue). Par exemple, processLLMRequest peut stocker une clé de cache, puis processLLMResponse peut la relire pour écrire la réponse.

Initialisez les champs de manière défensive lors du premier accès, car state commence comme un objet vide :

import type { Processor } from '@mastra/core/processors'

export class WordCounter implements Processor {
  id = 'word-counter'

  async processOutputStream({ part, state }) {
    state.wordCount ??= 0
    if (part.type === 'text-delta') {
      state.wordCount += part.payload.text.split(/\s+/).filter(Boolean).length
    }
    return part
  }
}

Interruption et fragments tripwire

La fonction abort de chaque méthode lève une erreur TripWire qui arrête le traitement et émet un fragment tripwire dans le flux de sortie. Les clients peuvent détecter ce fragment afin de distinguer une réponse bloquée d’une fin normale.

abort('Blocked content detected', { retry: false, metadata: { category: 'pii' } })

• reason : explication lisible. Apparaît dans tripwire.payload.reason.
• retry : lorsque cette valeur est true, l’Agent réessaie la même étape en utilisant reason comme retour. Les nouvelles tentatives ne s’exécutent que si maxProcessorRetries est défini sur l’Agent ou l’appel ; sinon, la requête est interrompue. Lorsque errorProcessors est configuré, maxProcessorRetries prend par défaut la valeur 10 pour cet appel.
• metadata : données structurées facultatives jointes au fragment tripwire pour les consommateurs en aval.

Le fragment tripwire émis présente la forme suivante :

type TripwireChunk = {
  type: 'tripwire'
  runId: string
  from: 'AGENT'
  payload: {
    reason: string
    retry?: boolean
    metadata?: unknown
    processorId: string
  }
}

Dans les appels sans diffusion (agent.generate()), le résultat expose les mêmes informations dans result.tripwire et result.finishReason === 'other'.

Émettre des fragments de données personnalisés

Les Processors qui ont accès à writer peuvent diffuser des fragments data-* personnalisés vers le client en appelant writer.custom(chunk). Les Tools peuvent faire de même par l’intermédiaire de leur propre writer. C’est le seul moyen pour un Processor d’émettre du contenu en dehors des fragments de texte et de Tool habituels.

await writer.custom({
  type: 'data-moderation',
  runId,
  from: 'AGENT',
  data: { level: 'warn', reason: 'Possibly unsafe' },
})

Lorsque la Memory est configurée, les fragments data-* personnalisés émis depuis processOutputStream ou processOutputResult sont enregistrés comme parties du message de l’assistant. Définissez transient: true sur l’objet fragment pour le diffuser sans l’enregistrer dans la Memory :

await writer.custom({
  type: 'data-progress',
  data: { status: 'Processing' },
  transient: true,
})

Transmettez transient comme propriété du fragment, et non comme deuxième argument de writer.custom(). Le deuxième argument contient les options du writer, telles que messageId.

Par défaut, les Processors ne voient pas les fragments data-* dans processOutputStream, afin de ne pas traiter accidentellement la télémétrie des Tools ou leur propre sortie. Activez explicitement ce comportement en définissant processDataParts: true sur le Processor :

class ModerationCollector implements Processor {
  id = 'moderation-collector'
  processDataParts = true

  async processOutputStream({ part, state }) {
    if (part.type === 'data-moderation') {
      state.warnings ??= []
      state.warnings.push(part.data)
    }
    return part
  }
}

Le type du fragment doit commencer par data- pour être traité comme un fragment de données personnalisé. Renvoyer null ou undefined depuis processOutputStream supprime toujours le fragment ; un Processor peut donc examiner, modifier ou filtrer des données personnalisées de la même manière que des fragments de texte.

Configurer des Processors sur un Agent

Les Processors sont associés à un Agent au moyen de trois tableaux :

import { Agent } from '@mastra/core/agent'
import { PrefillErrorHandler } from '@mastra/core/processors'

const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  model: 'openai/gpt-5',
  instructions: '...',
  inputProcessors: [new ContentFilter(['secret'])],
  outputProcessors: [new WordCounter()],
  errorProcessors: [new PrefillErrorHandler()],
  maxProcessorRetries: 3,
})

• inputProcessors : s’exécute avant le LLM. Reçoit les messages d’entrée.
• outputProcessors : s’exécute pendant ou après la réponse du LLM. Reçoit les fragments ou les messages de sortie.
• errorProcessors : s’exécute lorsque l’appel à l’API du LLM lève une erreur. Reçoit l’erreur brute.

Chaque tableau accepte également une fonction afin que les Processors puissent être construits pour chaque requête à partir de RequestContext :

new Agent({
  id: 'processor-interface-agent',
  inputProcessors: ({ requestContext }) => {
    const blockedWords = requestContext.get('blockedWords') ?? []
    return [new ContentFilter(blockedWords)]
  },
})

Remplacements propres à chaque appel

agent.generate() et agent.stream() acceptent inputProcessors, outputProcessors, errorProcessors et maxProcessorRetries. Lorsqu’un tableau de Processors est défini sur l’appel, il remplace le tableau correspondant configuré sur l’Agent pour cette requête. Les Processors de Memory, Workspace, Skill, Channel et Browser ajoutés automatiquement par Mastra sont toujours conservés et s’exécutent autour de votre tableau.

await agent.stream('Summarize this', {
  outputProcessors: [new StreamFilter()],
  maxProcessorRetries: 5,
})

La valeur maxProcessorRetries transmise lors de l’appel remplace la valeur par défaut de l’Agent. Si aucune n’est définie, les nouvelles tentatives demandées par les Processors sont traitées comme des interruptions.

Liens associés

• Vue d’ensemble des Processors : guide conceptuel des Processors
• Garde-fous : Processors de sécurité et de validation
• Processors de Memory : Processors propres à la Memory