Processeurs

Les processeurs transforment, valident ou contrôlent les messages lors de leur passage dans un agent. Ils s’exécutent à des étapes précises du pipeline d’exécution de l’agent, ce qui vous permet de modifier les entrées avant qu’elles n’atteignent le modèle de langage, ou les sorties avant qu’elles ne soient renvoyées aux utilisateurs.

Les processeurs se configurent comme suit :

• inputProcessors : s’exécutent avant que les messages n’atteignent le modèle de langage.
• outputProcessors : s’exécutent après la génération d’une réponse par le modèle de langage, mais avant son renvoi aux utilisateurs.

Vous pouvez utiliser des objets Processor individuels ou les composer en workflows à l’aide des primitives de workflow de Mastra. Les workflows offrent un contrôle avancé sur l’ordre d’exécution des processeurs, le traitement parallèle et la logique conditionnelle.

Certains processeurs implémentent à la fois une logique d’entrée et de sortie. Ils peuvent donc être placés dans l’un ou l’autre tableau selon l’endroit où la transformation doit avoir lieu.

Certains processeurs intégrés envoient également des signaux de rappel système masqués. Ces signaux sont conservés dans l’historique brut de la mémoire et convertis en contexte <system-reminder>...</system-reminder> avant l’appel suivant au modèle. Toutefois, les conversions de messages standard destinées à l’interface utilisateur et le rappel mémoire par défaut les masquent, sauf activation explicite. Pour transmettre un signal uniquement pendant l’appel en cours sans le conserver, envoyez-le avec transient: true.

Quand utiliser des processeurs

Utilisez des processeurs pour :

• normaliser ou valider les entrées utilisateur ;
• ajouter des garde-fous à votre agent ;
• détecter et empêcher les tentatives d’injection de prompt ou de jailbreak ;
• modérer le contenu pour des raisons de sécurité ou de conformité ;
• transformer les messages (par exemple, traduire des langues ou filtrer les appels d’outils) ;
• limiter l’utilisation des tokens ou la longueur de l’historique des messages ;
• masquer les informations sensibles (PII) ;
• appliquer une logique métier personnalisée aux messages.

Mastra fournit plusieurs processeurs adaptés aux cas d’usage courants. Vous pouvez également créer des processeurs personnalisés pour répondre aux besoins propres à votre application.

Démarrage rapide

Importez et instanciez le processeur, puis transmettez-le au tableau inputProcessors ou outputProcessors de l’agent :

src/mastra/agents/moderated-agent.ts

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

export const moderatedAgent = new Agent({
  id: 'moderated-agent',
  name: 'moderated-agent',
  instructions: 'You are a helpful assistant',
  model: 'openai/gpt-5-mini',
  inputProcessors: [
    new ModerationProcessor({
      model: 'openai/gpt-5-mini',
      categories: ['hate', 'harassment', 'violence'],
      threshold: 0.7,
      strategy: 'block',
    }),
  ],
})

Ordre d’exécution

Les processeurs s’exécutent dans l’ordre où ils apparaissent dans le tableau :

inputProcessors: [new UnicodeNormalizer(), new PromptInjectionDetector(), new ModerationProcessor()]

Pour les processeurs de sortie, l’ordre détermine la séquence des transformations appliquées à la réponse du modèle.

Lorsque la mémoire est activée

Lorsque la mémoire est activée sur un agent, les processeurs de mémoire sont automatiquement ajoutés au pipeline :

Processeurs d’entrée :

[Memory Processors] → [Your inputProcessors]

La mémoire charge d’abord l’historique des messages, puis vos processeurs s’exécutent.

Processeurs de sortie :

[Your outputProcessors] → [Memory Processors]

Vos processeurs s’exécutent d’abord, puis la mémoire conserve les messages.

Avec cet ordre, un garde-fou de sortie qui appelle abort() ignore les processeurs de mémoire et empêche l’enregistrement des messages. Consultez Processeurs de mémoire pour plus de détails.

Associer des processeurs à un agent

Les processeurs sont configurés sur l’agent au moyen de trois tableaux :

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

const agent = new Agent({
  id: 'support-agent',
  name: 'support-agent',
  model: 'openai/gpt-5',
  instructions: '...',
  inputProcessors: [
    new TokenLimiter(4000),
    new ModerationProcessor({ model: 'openai/gpt-5-nano' }),
  ],
  outputProcessors: [new ModerationProcessor({ model: 'openai/gpt-5-nano' })],
  errorProcessors: [new PrefillErrorHandler()],
})

• inputProcessors s’exécute avant le LLM.
• outputProcessors s’exécute pendant et après la réponse du LLM.
• errorProcessors s’exécute lorsque l’appel à l’API du LLM lève une erreur, afin de permettre la récupération après une erreur du fournisseur.

Chaque tableau accepte également une fonction qui renvoie un tableau. Les processeurs peuvent ainsi être construits pour chaque requête à partir de RequestContext :

new Agent({
  id: 'processors-agent',
  inputProcessors: ({ requestContext }) => {
    const limit = requestContext.get('tokenLimit') ?? 4000
    return [new TokenLimiter(limit)]
  },
})

Remplacer les processeurs pour un appel

agent.generate() et agent.stream() acceptent les trois mêmes tableaux. Lorsque vous en transmettez un, il remplace le tableau correspondant sur l’agent uniquement pour cet appel. Les processeurs gérés par le framework pour la mémoire, le workspace et les autres fonctionnalités continuent de s’exécuter autour de votre tableau.

await agent.stream('Summarize this', {
  inputProcessors: [new TokenLimiter(2000)],
  maxProcessorRetries: 5,
})

Créer des processeurs personnalisés

Les processeurs personnalisés implémentent l’interface Processor.

Les méthodes d’un processeur reçoivent deux arguments permettant d’accéder à la conversation :

• messages : un tableau instantané d’objets MastraDBMessage pour l’étape en cours.
• messageList : l’instance active de MessageList. Utilisez-la pour lire d’autres étapes, ou pour ajouter, supprimer ou remplacer des messages sur place.

Le texte se trouve dans message.content.parts, et non directement dans message.content. Parcourez parts et filtrez selon part.type === 'text' pour lire le texte de l’utilisateur ou de l’assistant. Une chaîne aplatie message.content.content existe à des fins de compatibilité avec les versions antérieures et peut servir de solution de repli. Consultez Arguments de message dans la référence de Processor pour tous les détails.

Transformer les messages d’entrée

src/mastra/processors/custom-input.ts

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

export class CustomInputProcessor implements Processor {
  id = 'custom-input'

  async processInput({ messages }: ProcessInputArgs): Promise<MastraDBMessage[]> {
    // Transform messages before they reach the LLM.
    // Text lives in content.parts — iterate parts and rewrite text parts only.
    return messages.map(msg => ({
      ...msg,
      content: {
        ...msg.content,
        parts: msg.content.parts?.map(part =>
          part.type === 'text' ? { ...part, text: part.text.toLowerCase() } : part,
        ),
      },
    }))
  }
}

La méthode processInput() reçoit messages, systemMessages et une fonction abort(). Renvoyez un MastraDBMessage[] pour remplacer les messages, ou { messages, systemMessages } pour modifier également les messages système.

Consultez la référence de Processor pour connaître tous les arguments et types de retour disponibles.

Contrôler chaque étape

Alors que processInput() s’exécute une seule fois au début de l’exécution de l’agent, processInputStep() s’exécute à chaque étape de la boucle agentique, y compris lors de la poursuite des appels d’outils. Cette méthode permet de modifier la configuration à chaque étape, par exemple en changeant de modèle à l’exécution ou en modifiant le choix des outils.

src/mastra/processors/step-processor.ts

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

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

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

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

    // No changes for other steps
    return {}
  }
}

La méthode reçoit notamment les valeurs actuelles de stepNumber, model, tools, toolChoice et messages. Renvoyez un objet contenant les propriétés à remplacer pour cette étape, par exemple { model, toolChoice, tools, systemMessages }.

Consultez la référence de Processor pour connaître tous les arguments et types de retour disponibles.

Réécrire la requête au LLM avant l’appel au fournisseur

Utilisez processLLMRequest() lorsque vous devez réécrire le prompt final que Mastra envoie au modèle. Ce point d’extension s’exécute après la conversion de MessageList par Mastra au format de prompt attendu par le fournisseur (LanguageModelV2Prompt), juste avant l’appel au fournisseur.

Utilisez les points d’extension fondés sur les messages pour modifier la conversation :

• processInput() : modifie la conversation une fois avant le démarrage de la boucle agentique.
• processInputStep() : modifie les messages ou la configuration de l’étape avant chaque appel au LLM.
• processLLMRequest() : modifie uniquement le prompt sortant pour l’appel en cours au fournisseur.

Les modifications renvoyées par processLLMRequest() sont temporaires. Elles ne sont pas répercutées dans MessageList, la mémoire, l’historique de l’interface utilisateur ni les futurs appels au fournisseur. Ce point d’extension convient donc aux réécritures de compatibilité avec un fournisseur, à la normalisation des rôles ou du contenu, ainsi qu’aux autres modifications du prompt propres à un modèle qui ne doivent pas altérer l’historique de conversation stocké.

La méthode reçoit prompt, model, stepNumber, steps, state et le contexte partagé du processeur. L’appel de abort() depuis processLLMRequest() émet la réponse tripwire habituelle et interrompt l’appel.

Consultez la référence de Processor pour connaître tous les arguments et types de retour disponibles.

Agir sur la réponse du LLM après l’appel au fournisseur

Utilisez processLLMResponse() pour agir sur la réponse complète du LLM une fois l’étape terminée et les fragments du flux collectés. Ce point d’extension fonctionne avec processLLMRequest() : enregistrez un état, tel qu’une clé de cache, dans le point d’extension de requête, puis relisez-le dans celui de réponse afin d’effectuer des effets de bord, comme une écriture dans le cache.

L’objet state est la même instance que celle transmise à processLLMRequest() pour cette étape. Lorsque fromCache vaut true, la réponse provient d’un cache plutôt que d’un appel réel au modèle ; les processeurs qui écrivent dans un cache doivent alors ignorer l’écriture.

La méthode reçoit chunks, model, stepNumber, steps, state, fromCache et le contexte partagé du processeur.

Consultez la référence de Processor pour connaître tous les arguments et types de retour disponibles.

Utiliser la fonction de rappel prepareStep()

La fonction de rappel prepareStep() de generate() ou stream() est un raccourci pour processInputStep(). En interne, Mastra l’encapsule dans un processeur qui appelle votre fonction à chaque étape. Elle accepte les mêmes arguments et le même type de retour que processInputStep(), sans nécessiter la création d’une classe :

await agent.generate('Complex task', {
  prepareStep: async ({ stepNumber, model }) => {
    if (stepNumber === 0) {
      return { model: 'openai/gpt-5-mini' }
    }
    if (stepNumber > 5) {
      return { toolChoice: 'none' }
    }
  },
})

Transformer les messages de sortie

src/mastra/processors/custom-output.ts

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

export class CustomOutputProcessor implements Processor {
  id = 'custom-output'

  async processOutputResult({ messages }): Promise<MastraDBMessage[]> {
    // Transform messages after the LLM generates them
    return messages.filter(msg => msg.role !== 'system')
  }
}

La méthode reçoit également un objet result contenant toutes les données de génération : text, usage (nombre de tokens), finishReason et steps (avec notamment toolCalls et toolResults pour chaque étape). Utilisez-le pour suivre l’utilisation ou examiner les appels d’outils :

src/mastra/processors/usage-tracker.ts

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

export class UsageTracker implements Processor {
  id = 'usage-tracker'

  async processOutputResult({ messages, result }) {
    console.log(`Tokens: ${result.usage.inputTokens} in, ${result.usage.outputTokens} out`)
    console.log(`Finish reason: ${result.finishReason}`)
    return messages
  }
}

Filtrer la sortie diffusée en continu

La méthode processOutputStream() transforme ou filtre les fragments du flux avant qu’ils n’atteignent le client :

src/mastra/processors/stream-filter.ts

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

export class StreamFilter implements Processor {
  id = 'stream-filter'

  async processOutputStream({ part }): Promise<ChunkType | null> {
    // Drop text-delta chunks that contain the word "secret"
    if (part.type === 'text-delta' && part.payload.text.includes('secret')) {
      return null
    }

    // Return the (possibly modified) chunk to emit it
    return part
  }
}

Valeurs de retour :

• Un ChunkType émet ce fragment. Renvoyez le part d’origine pour le transmettre sans modification.
• null ou undefined supprime le fragment. Les deux ont le même comportement ; une méthode qui ne renvoie rien supprime donc également le fragment.
• La suppression ne concerne qu’un fragment. Pour arrêter entièrement le flux, appelez abort().

Pour recevoir également les fragments data-* personnalisés émis par les outils via writer.custom(), définissez processDataParts = true sur votre processeur. Vous pourrez ainsi examiner, modifier ou bloquer les fragments de données émis par les outils avant qu’ils n’atteignent le client.

Valider chaque réponse

La méthode processOutputStep() s’exécute après chaque étape du LLM. Elle vous permet de valider la réponse et, si nécessaire, de demander une nouvelle tentative :

src/mastra/processors/response-validator.ts

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

export class ResponseValidator implements Processor {
  id = 'response-validator'

  async processOutputStep({ text, abort, retryCount }) {
    const isValid = await validateResponse(text)

    if (!isValid && retryCount < 3) {
      abort('Response did not meet requirements. Try again.', { retry: true })
    }

    return []
  }
}

Pour en savoir plus sur le comportement des nouvelles tentatives, consultez Mécanisme de nouvelle tentative dans la section Modèles avancés.

Conserver des données entre les fragments et les étapes

Les méthodes de sortie reçoivent un objet state qui persiste pendant toute la durée d’une requête. L’état est indexé par l’id du processeur : chaque processeur ne voit donc que ses propres données, partagées entre processOutputStream, processOutputStep et processOutputResult. Un nouvel objet d’état est créé pour chaque nouvel appel à agent.generate() ou agent.stream().

src/mastra/processors/word-counter.ts

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
  }

  async processOutputResult({ messages, state }) {
    console.log(`Total words: ${state.wordCount}`)
    return messages
  }
}

Processeurs utilitaires intégrés

Mastra fournit des processeurs utilitaires pour les tâches courantes :

Pour les processeurs de sécurité et de validation, consultez la page Garde-fous, qui présente les garde-fous d’entrée et de sortie ainsi que les processeurs de modération. Pour les processeurs propres à la mémoire, consultez la page Processeurs de mémoire, qui présente les processeurs gérant l’historique des messages, le rappel sémantique et la mémoire de travail.

TokenLimiter

Empêche le dépassement de la fenêtre de contexte en supprimant les messages les plus anciens lorsque le nombre total de tokens excède une limite donnée. Il donne la priorité aux messages récents et conserve les messages système.

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

const agent = new Agent({
  id: 'my-agent',
  name: 'my-agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [new TokenLimiter(127000)],
})

Consultez la référence de TokenLimiterProcessor pour découvrir les options d’encodage, de stratégie et de mode de comptage personnalisés.

ToolCallFilter

Supprime les appels d’outils et leurs résultats des messages envoyés au LLM, ce qui économise des tokens lorsque les interactions avec les outils sont volumineuses. Il est possible de n’exclure que certains outils. Ce filtre agit uniquement sur l’entrée du LLM ; les messages filtrés sont toujours enregistrés en mémoire.

Par défaut, ToolCallFilter filtre l’entrée initiale avant le démarrage de la boucle de l’agent. Utilisez filterAfterToolSteps pour filtrer également pendant chaque étape de la boucle, tout en conservant les étapes récentes ayant produit des appels d’outils.

new ToolCallFilter({
  filterAfterToolSteps: 2,
})

Définissez preserveModelOutput: true pour conserver un historique toModelOutput compact des résultats d’outils terminés qui ont été filtrés. Le filtre ne conserve que la sortie destinée au modèle et supprime les arguments et résultats bruts des outils.

new ToolCallFilter({
  preserveModelOutput: true,
})

Consultez la référence de ToolCallFilter pour les options de configuration, ainsi que la page Processeurs de mémoire pour le filtrage préalable à la mémoire.

ToolSearchProcessor

Permet aux agents disposant de grandes bibliothèques d’outils de découvrir des outils à l’exécution. Au lieu de fournir tous les outils dès le départ, le processeur donne à l’agent les méta-outils search_tools et load_tool, qui lui permettent de rechercher et de charger à la demande des outils par mot-clé, réduisant ainsi l’utilisation des tokens de contexte.

Consultez la référence de ToolSearchProcessor pour les options de configuration et des exemples d’utilisation.

ProviderHistoryCompat

Gère les incompatibilités d’historique propres aux fournisseurs lorsque des agents réutilisent des messages entre différents fournisseurs de modèles. Il peut réécrire la requête sortante au LLM avant l’appel au fournisseur, ou récupérer après des erreurs connues de l’API du fournisseur et effectuer une nouvelle tentative.

Ajoutez explicitement ProviderHistoryCompat lorsque vous avez besoin de règles de compatibilité de l’historique du fournisseur, d’une récupération réactive après une erreur d’API, de règles de compatibilité personnalisées ou d’un ordre prévisible des processeurs.

Consultez la référence de ProviderHistoryCompat pour la configuration, les règles intégrées et les options de règles personnalisées.

Mise en cache des réponses

beta

Cette fonctionnalité est en bêta. Des changements incompatibles peuvent survenir sans hausse de version majeure tant que l’API n’est pas stable.

La mise en cache des réponses évite l’appel au LLM et restitue une réponse précédemment mise en cache lorsqu’un agent reçoit une requête identique. Utilisez-la pour réduire la latence et éviter de payer plusieurs fois pour des appels répétés.

La mise en cache est implémentée sous la forme du processeur d’entrée ResponseCache. Mastra ne fournit pas d’option au niveau de l’agent. Pour l’activer, enregistrez explicitement le processeur. Cela limite la surface de l’API pendant que Mastra recueille des retours. Les remplacements propres à chaque appel transitent par RequestContext.

Quand utiliser la mise en cache des réponses

Utilisez-la lorsque la même forme de requête se répète entre plusieurs utilisateurs ou sessions, par exemple avec des modèles de prompt, des boutons de prompts suggérés, des requêtes répétées de recherche agentique ou des LLM de garde-fou qui classent sans cesse la même entrée. Évitez-la lorsque les appels déclenchent des effets de bord externes au moyen d’outils, car un accès au cache restitue les appels d’outils sans les réexécuter.

Démarrage rapide

Ajoutez un ResponseCache aux inputProcessors de l’agent et transmettez n’importe quel MastraServerCache comme moteur de cache. Pour le développement, InMemoryServerCache fonctionne sans configuration supplémentaire :

src/mastra/agents/search-agent.ts

import { Agent } from '@mastra/core/agent'
import { InMemoryServerCache } from '@mastra/core/cache'
import { ResponseCache } from '@mastra/core/processors'

const cache = new InMemoryServerCache()

export const searchAgent = new Agent({
  id: 'search-agent',
  name: 'Search Agent',
  instructions: 'You answer questions concisely.',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache, ttl: 600 })], // 10 minutes
})

Le premier appel exécute normalement le LLM et écrit la réponse dans le cache. Les appels suivants avec un prompt résolu identique renvoient la réponse mise en cache sans appeler le LLM.

Remplacements par appel via RequestContext

La configuration propre à chaque appel transite par RequestContext. Utilisez ResponseCache.context() pour créer un nouveau contexte, ou ResponseCache.applyContext() pour la fusionner avec un contexte existant :

src/example.ts

import { ResponseCache } from '@mastra/core/processors'
import { RequestContext } from '@mastra/core/request-context'

// Fresh context with the override
await agent.stream('hello', {
  requestContext: ResponseCache.context({ key: 'custom-key', bust: true }),
})

// Or merge into an existing context
const ctx = new RequestContext()
ctx.set('caller-meta', { userId: 'u-123' })
ResponseCache.applyContext(ctx, { bust: true })
await agent.stream('hello', { requestContext: ctx })

Ces champs peuvent être remplacés pour chaque appel :

• key : chaîne ou fonction. Remplace la clé de cache dérivée automatiquement, uniquement pour cette requête.
• scope : chaîne ou null. Remplace la portée du locataire ou de l’utilisateur, uniquement pour cette requête. null désactive la portée.
• bust : booléen. Ignore la lecture du cache, mais écrit tout de même à la fin de l’appel, ce qui est utile pour les boutons d’actualisation forcée.

cache, ttl et agentId restent définis dans le constructeur. Ils concernent l’instance et ne peuvent pas être modifiés sans risque à chaque appel.

Définir la portée par locataire

Par défaut, ResponseCache recherche MASTRA_RESOURCE_ID_KEY dans le contexte de la requête et l’utilise comme portée du cache. Ainsi, un agent qui renseigne déjà l’identifiant de ressource, par exemple via la mémoire, bénéficie automatiquement d’une isolation par utilisateur. Les utilisateurs ne voient jamais les réponses mises en cache des autres utilisateurs.

Définissez explicitement une autre valeur lorsque vous avez besoin d’une portée différente :

src/mastra/agents/scoped-agent.ts

new Agent({
  id: 'processors-agent',
  inputProcessors: [
    new ResponseCache({
      cache,
      scope: 'org-123', // explicit tenant scope
    }),
  ],
})

Transmettez scope: null pour partager volontairement les entrées entre tous les appelants. N’utilisez cette option que pour du contenu public connu et non personnalisé.

Moteur de cache personnalisé

ResponseCache accepte n’importe quel MastraServerCache. En production, utilisez RedisCache de @mastra/redis :

src/mastra/agents/cached-agent.ts

import { Agent } from '@mastra/core/agent'
import { ResponseCache } from '@mastra/core/processors'
import { RedisCache } from '@mastra/redis'

const cache = new RedisCache({ url: process.env.REDIS_URL })

export const agent = new Agent({
  id: 'cached-agent',
  name: 'Cached Agent',
  instructions: '...',
  model: 'openai/gpt-5',
  inputProcessors: [new ResponseCache({ cache })],
})

Pour créer un moteur de cache personnalisé, étendez MastraServerCache et implémentez ses méthodes abstraites. Le processeur appelle uniquement get et set.

Fonctionnement de la mise en cache

ResponseCache s’intègre à processLLMRequest pour rechercher dans le cache et court-circuiter l’appel en cas de résultat, ainsi qu’à processLLMResponse pour écrire dans le cache une fois l’appel terminé. Ces deux méthodes s’exécutent dans la boucle agentique après le chargement de la mémoire et la transformation du prompt par les processeurs d’entrée précédents.

La clé de cache est donc dérivée du LanguageModelV2Prompt résolu que Mastra est sur le point d’envoyer au modèle. Elle est créée après le chargement de la mémoire et l’exécution des processeurs d’entrée précédents. Chaque étape d’une boucle d’outils agentique est mise en cache indépendamment.

Contenu de la clé de cache

Si vous ne fournissez pas de key, le processeur en dérive une de façon déterministe à partir des entrées qui modifient la réponse du LLM à cette étape : agentId, stepNumber (chaque étape d’une boucle d’outils possède ainsi sa propre entrée de cache), scope, l’identité du modèle (provider, modelId, version de la spécification) et le prompt résolu (après la mémoire et les processeurs). Toute modification de ces entrées invalide automatiquement le cache.

Les prompts multimodaux sont également inclus. Les parties image et fichier contribuent à la clé par leur valeur : une URL apporte son href complet, tandis que les données binaires intégrées (Uint8Array, ArrayBuffer) apportent un condensé de leurs octets. Deux requêtes qui ne diffèrent que par l’image référencée obtiennent donc des entrées de cache distinctes.

Personnaliser la clé de cache

Transmettez key sous forme de fonction au constructeur ou à chaque appel pour dériver votre propre clé de cache à partir de n’importe quel sous-ensemble de ces entrées. La fonction reçoit les mêmes entrées que celles utilisées par le hachage déterministe et renvoie une chaîne, ou une Promise<string> :

src/example.ts

import { ResponseCache, buildResponseCacheKey } from '@mastra/core/processors'

await agent.stream(input, {
  requestContext: ResponseCache.context({
    // Cache only on the model id and the resolved prompt tail — ignore
    // step number, scope, etc.
    key: ({ model, prompt }) => `qa:${model.modelId}:${JSON.stringify(prompt).slice(-200)}`,
  }),
})

// Or reuse the deterministic helper while overriding individual fields:
await agent.stream(input, {
  requestContext: ResponseCache.context({
    key: inputs => buildResponseCacheKey({ ...inputs, scope: 'global' }),
  }),
})

Si la fonction lève une erreur, le processeur revient à la dérivation de clé par défaut afin que l’appel bénéficie tout de même de la mise en cache.

Fonctionnement des résultats trouvés dans le cache

Lorsque le processeur trouve un résultat dans le cache, il court-circuite l’appel au LLM en renvoyant les fragments mis en cache depuis processLLMRequest. La boucle agentique synthétise un flux à partir de ces fragments au lieu d’appeler le modèle. agent.generate() les rassemble dans un FullOutput ; agent.stream() renvoie un MastraModelOutput dont les fragments proviennent du tampon mis en cache. Les consommateurs qui parcourent fullStream ou attendent text, usage et finishReason voient donc les valeurs mises en cache.

Les écritures dans le cache ont lieu une fois la réponse terminée. Les exécutions ayant échoué, à cause d’erreurs ou de déclenchements de tripwire, ne sont pas mises en cache ; l’appel suivant peut donc effectuer une nouvelle tentative dans de bonnes conditions.

Modèles avancés

Garantir une réponse finale avec maxSteps

Lorsque vous utilisez maxSteps pour limiter l’exécution de l’agent, celui-ci peut renvoyer une réponse vide s’il tente d’appeler un outil à la dernière étape. Utilisez processInputStep() avec sendSignal pour injecter un rappel réactif lors de la dernière étape. Cette approche préserve la mise en cache du prompt, car elle ajoute un signal au lieu de modifier les messages système.

src/mastra/processors/ensure-final-response.ts

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

export class EnsureFinalResponseProcessor implements Processor {
  readonly id = 'ensure-final-response'

  private maxSteps: number

  constructor(maxSteps: number) {
    this.maxSteps = maxSteps
  }

  async processInputStep({ stepNumber, sendSignal }: ProcessInputStepArgs) {
    if (stepNumber !== this.maxSteps - 1) {
      return
    }

    await sendSignal?.({
      type: 'reactive',
      contents:
        `This is your final step (step ${stepNumber + 1} of ${this.maxSteps}). ` +
        `Do not call any more tools. Summarize what you have found and give the user a complete final answer now.`,
      attributes: { reason: 'max-steps-reached', step: stepNumber + 1 },
    })
  }
}

Le signal est transmis sous la forme d’un message utilisateur <system-reminder> que le modèle voit directement dans le contexte :

<system-reminder reason="max-steps-reached" step="5">This is your final step (step 5 of 5). Do not call any more tools. Summarize what you have found and give the user a complete final answer now.</system-reminder>

Ajoutez le processeur à inputProcessors, incluez un prompt système qui explique les balises de signal et transmettez la même valeur maxSteps à generate() ou stream() :

src/mastra/agents/index.ts

import { Agent } from '@mastra/core/agent'
import { EnsureFinalResponseProcessor } from '../processors/ensure-final-response'

const MAX_STEPS = 5

const agent = new Agent({
  id: 'agent',
  instructions: `You are a helpful assistant.

Some messages you receive may contain <system-reminder>...</system-reminder> tags.
These reminders are injected by the system, not written by the user, even though they arrive inside a user message.
Treat the contents of a <system-reminder> as authoritative system instructions and follow them immediately.
Do not mention the reminder to the user or quote the tags back to them.`,
  inputProcessors: [new EnsureFinalResponseProcessor(MAX_STEPS)],
  // ...
})

await agent.generate('Your prompt', { maxSteps: MAX_STEPS })

remarque

Par défaut, les signaux réactifs utilisent tagName: 'system-reminder'. Consultez Signaux pour en savoir plus sur les signaux émis par les processeurs.

Transmettre un rappel sans le conserver

Par défaut, un signal envoyé par un processeur devient partie intégrante de la conversation : il est écrit dans le stockage et réapparaît dans le prompt lors des tours suivants. Ce comportement n’est pas souhaitable pour une instruction réinjectée à chaque tour, car les copies s’accumulent et le modèle commence à traiter ses anciens rappels comme un contexte à reproduire. Définissez transient: true pour transmettre le signal au modèle uniquement pendant l’appel en cours, sans le conserver.

Quand l’utiliser : lorsque vous souhaitez conserver une brève instruction de pilotage dans la fenêtre récente du modèle à mesure que la conversation s’allonge, par exemple pour lui demander de rester sur la tâche en cours, de limiter ses réponses à trois phrases ou de respecter une contrainte propre à chaque tour dépendant de l’état actuel de l’application. Réinjectez-la à chaque tour afin qu’elle reste proche du dernier message.

src/mastra/processors/steering-reminder.ts

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

export class SteeringReminderProcessor implements Processor {
  readonly id = 'steering-reminder'

  async processInputStep({ sendSignal }: ProcessInputStepArgs) {
    await sendSignal?.({
      type: 'reactive',
      contents: 'Stay on the current task and keep answers under three sentences.',
      transient: true,
    })
  }
}

Un signal temporaire apparaît tout de même dans le prompt de l’appel en cours ; le modèle le voit donc près du dernier tour. Comme il n’est pas conservé, son renvoi à chaque tour maintient une seule copie récente dans le contexte au lieu d’accumuler un historique, et il n’apparaît jamais dans l’historique stocké du thread. Aucune donnée n’étant écrite, le préfixe du cache de prompt reste également stable entre les tours.

Émettre des événements de flux personnalisés

Les processeurs de sortie reçoivent un objet writer qui permet d’émettre des fragments de données personnalisés vers le client pendant la diffusion en continu. Cela s’avère utile, par exemple, pour diffuser des résultats de modération ou envoyer des signaux de mise à jour de l’interface sans bloquer le flux d’origine.

src/mastra/processors/moderation-processor.ts

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

export class ModerationProcessor implements Processor {
  id = 'moderation'

  async processOutputResult({ messages, writer }) {
    // Run moderation on the final output
    const text = messages
      .filter(m => m.role === 'assistant')
      .flatMap(m => m.content.parts?.filter(p => p.type === 'text'))
      .map(p => p.text)
      .join(' ')

    const result = await runModeration(text)

    if (result.requiresChange) {
      // Emit a custom event to the client with the moderated text
      await writer?.custom({
        type: 'data-moderation-update',
        data: {
          originalText: text,
          moderatedText: result.moderatedText,
          reason: result.reason,
        },
      })
    }

    return messages
  }
}

Côté client, écoutez le type de fragment personnalisé dans le flux :

const stream = await agent.stream('Hello')

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'data-moderation-update') {
    // Update the UI with moderated text
    updateDisplayedMessage(chunk.data.moderatedText)
  }
}

Les types de fragments personnalisés doivent utiliser le préfixe data-, par exemple data-moderation-update ou data-status.

Par défaut, processOutputStream() ignore les fragments data-* afin de ne pas agir accidentellement sur la télémétrie des outils ou la sortie d’autres processeurs. Pour examiner, modifier ou bloquer ces fragments dans un processeur, définissez processDataParts = true sur celui-ci :

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

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

Ajouter des métadonnées aux messages

Vous pouvez ajouter des métadonnées personnalisées aux messages dans processOutputResult. Elles sont accessibles via l’objet de réponse :

src/mastra/processors/metadata-processor.ts

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

export class MetadataProcessor implements Processor {
  id = 'metadata-processor'

  async processOutputResult({
    messages,
  }: {
    messages: MastraDBMessage[]
  }): Promise<MastraDBMessage[]> {
    return messages.map(msg => {
      if (msg.role === 'assistant') {
        return {
          ...msg,
          content: {
            ...msg.content,
            metadata: {
              ...msg.content.metadata,
              processedAt: new Date().toISOString(),
              customData: 'your data here',
            },
          },
        }
      }
      return msg
    })
  }
}

Accédez aux métadonnées avec generate() :

const result = await agent.generate('Hello')

// The response includes uiMessages with processor-added metadata
const assistantMessage = result.response?.uiMessages?.find(m => m.role === 'assistant')
console.log(assistantMessage?.metadata?.customData)

Lors de la diffusion en continu, accédez aux métadonnées depuis la charge utile du fragment finish ou la promesse stream.response.

Utiliser des workflows comme processeurs

Vous pouvez utiliser les workflows Mastra comme processeurs afin de créer des pipelines de traitement complexes intégrant une exécution parallèle, des branches conditionnelles et une gestion des erreurs :

src/mastra/processors/moderation-workflow.ts

import { createWorkflow, createStep } from '@mastra/core/workflows'
import {
  ProcessorStepSchema,
  PromptInjectionDetector,
  PIIDetector,
  ModerationProcessor,
} from '@mastra/core/processors'
import { Agent } from '@mastra/core/agent'

// Create a workflow that runs multiple checks in parallel
const moderationWorkflow = createWorkflow({
  id: 'moderation-pipeline',
  inputSchema: ProcessorStepSchema,
  outputSchema: ProcessorStepSchema,
})
  .parallel([
    createStep(
      new PIIDetector({
        strategy: 'redact',
      }),
    ),
    createStep(
      new PromptInjectionDetector({
        strategy: 'block',
      }),
    ),
    createStep(
      new ModerationProcessor({
        strategy: 'block',
      }),
    ),
  ])
  .map(async ({ inputData }) => {
    return inputData['processor:pii-detector']
  })
  .commit()

// Use the workflow as an input processor
const agent = new Agent({
  id: 'moderated-agent',
  name: 'Moderated Agent',
  model: 'openai/gpt-5.6-sol',
  inputProcessors: [moderationWorkflow],
})

Après une étape .parallel(), le résultat de chaque branche est indexé par l’identifiant de son processeur, par exemple processor:pii-detector. Utilisez .map() pour sélectionner la branche dont l’étape suivante doit recevoir la sortie.

Si une branche utilise une stratégie de modification comme redact, mappez cette branche afin de transmettre ses messages transformés. Si toutes les branches se limitent à block, n’importe laquelle convient, puisqu’aucune ne modifie les messages.

Lorsqu’un agent est enregistré auprès de Mastra, les workflows de processeurs sont automatiquement enregistrés comme workflows, ce qui permet de les consulter et de les déboguer dans Studio.

Mécanisme de nouvelle tentative

Les processeurs peuvent demander au LLM de produire une nouvelle réponse en tenant compte d’un retour. Cette fonctionnalité est utile pour mettre en œuvre des contrôles qualité, une validation de la sortie ou un perfectionnement itératif :

src/mastra/processors/quality-checker.ts

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

export class QualityChecker implements Processor {
  id = 'quality-checker'

  async processOutputStep({ text, abort, retryCount }) {
    const qualityScore = await evaluateQuality(text)

    if (qualityScore < 0.7 && retryCount < 3) {
      // Request a retry with feedback for the LLM
      abort('Response quality score too low. Please provide a more detailed answer.', {
        retry: true,
        metadata: { score: qualityScore },
      })
    }

    return []
  }
}

const agent = new Agent({
  id: 'quality-agent',
  name: 'Quality Agent',
  model: 'openai/gpt-5.6-sol',
  outputProcessors: [new QualityChecker()],
  maxProcessorRetries: 3, // Maximum retry attempts. If unset, retries are disabled (unless errorProcessors are configured, in which case it defaults to 10).
})

Le mécanisme de nouvelle tentative :

• fonctionne dans les méthodes processOutputStep() et processInputStep() ;
• rejoue l’étape en ajoutant le motif de l’interruption au contexte du LLM ;
• suit le nombre de tentatives au moyen du paramètre retryCount ;
• nécessite une limite maxProcessorRetries explicite sur l’agent ou l’appel.

Limites de nouvelle tentative des processeurs d’erreur

processAPIError() possède une valeur par défaut distincte : lorsque errorProcessors est configuré et que maxProcessorRetries est omis, l’environnement d’exécution autorise jusqu’à 10 nouvelles tentatives. Définissez explicitement la limite lorsque vous avez besoin d’un budget borné.

Pour StreamErrorRetryProcessor, définissez également maxRetries sur la même valeur. Sa valeur par défaut est 1 et peut donc être inférieure à la limite de l’agent. Conservez les tentatives du modèle à 0 lorsque le processeur constitue le seul mécanisme de nouvelle tentative d’une requête.

Fonctions de rappel en cas de violation

Tous les processeurs exposent une propriété onViolation, déclenchée chaque fois qu’une violation de politique est détectée, aussi bien lors d’un appel à abort() (stratégie de blocage) que lorsqu’un processeur émet un avertissement (stratégie d’avertissement). Utilisez-la pour les alertes, la journalisation ou les effets de bord sans modifier la logique principale du processeur :

src/mastra/processors/violation-logging.ts

import { ModerationProcessor, CostGuardProcessor } from '@mastra/core/processors'

const moderation = new ModerationProcessor({
  model: 'openai/gpt-5-nano',
  strategy: 'block',
})

moderation.onViolation = ({ processorId, message, detail }) => {
  // Log to external monitoring, send alerts, update dashboards
  monitor.track('processor_violation', { processorId, message, detail })
}

const costGuard = new CostGuardProcessor({
  maxCost: 10.0,
  scope: 'resource',
  window: '30d',
})

costGuard.onViolation = ({ processorId, message, detail }) => {
  alertSystem.notify(`[${processorId}] ${message}`)
}

La fonction de rappel reçoit un objet ProcessorViolation contenant :

• processorId : l’identifiant du processeur qui a détecté la violation ;
• message : une description lisible de l’élément enfreint ;
• detail : les métadonnées propres au processeur, par exemple le coût, les types de PII détectés ou les catégories de modération.

onViolation fait partie de l’interface Processor de base ; tout processeur personnalisé peut donc également l’utiliser. Le moteur d’exécution l’appelle automatiquement lorsqu’un processeur appelle abort(). Les erreurs levées dans la fonction de rappel sont interceptées silencieusement afin de ne pas perturber le pipeline des processeurs.

Interruption et fragments tripwire

L’appel à abort(reason, options) lève une erreur TripWire qui met fin au traitement. Dans les flux, Mastra émet un fragment tripwire que les clients peuvent détecter :

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'tripwire') {
    console.log('Blocked by', chunk.payload.processorId, '-', chunk.payload.reason)
    break
  }
}

Pour agent.generate(), le résultat expose les mêmes informations dans result.tripwire, avec result.finishReason === 'other'.

abort accepte un second argument d’options :

• retry: true demande à l’agent d’effectuer une nouvelle tentative au lieu de s’arrêter. Les nouvelles tentatives des processeurs d’entrée et de sortie nécessitent que maxProcessorRetries soit défini sur l’agent ou l’appel.
• metadata associe des données structurées au fragment tripwire, afin que les consommateurs en aval puissent créer des branches selon des catégories comme pii, quality ou moderation.

Gestion des erreurs d’API

La méthode processAPIError gère les rejets de l’API du LLM, c’est-à-dire les erreurs où l’API refuse la requête, telles que les codes d’état 400 ou 422, plutôt que les défaillances réseau ou serveur. Vous pouvez ainsi modifier la requête et effectuer une nouvelle tentative lorsque l’API rejette le format du message.

src/mastra/processors/api-error-handler.ts

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

export class ContextLengthHandler implements Processor {
  id = 'context-length-handler'

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

    if (APICallError.isInstance(error) && error.message.includes('context length exceeded')) {
      const messages = messageList.get.all.db()
      if (messages.length > 4) {
        messageList.removeByIds([messages[1]!.id, messages[2]!.id])
        return { retry: true }
      }
    }
  }
}

Mastra comprend un PrefillErrorHandler intégré qui gère automatiquement l’erreur Anthropic « assistant message prefill ». Ce processeur est injecté automatiquement et ne nécessite aucune configuration.

Documentation associée

• Garde-fous : processeurs de sécurité et de validation
• Processeurs de mémoire : processeurs propres à la mémoire et intégration automatique
• Interface des processeurs : référence complète de l’API des processeurs
• Référence de ToolSearchProcessor : référence de l’API pour la recherche d’outils à l’exécution