Agents durables

Ajouté dans : @mastra/core@1.45.0

attention

Les agents durables sont actuellement en bêta. Les API pourront évoluer dans de prochaines versions.

Un agent durable encapsule un Agent standard afin d’exécuter la boucle agentique dans un workflow. Les événements transitent par PubSub, ce qui permet à un client de se déconnecter puis de se reconnecter sans perdre de fragments. L’état de l’exécution est conservé et résiste donc aux redémarrages du processus.

Quand utiliser les agents durables

Utilisez un agent durable dans l’un des cas suivants :

• Le client peut perdre la connexion et se reconnecter en cours de flux (appareil mobile, réseau instable, appels prolongés).
• La boucle agentique peut durer plus longtemps qu’une seule requête HTTP (recherche en arrière-plan, utilisation d’outils en plusieurs étapes).
• Vous avez besoin d’une API d’observation et de reconnexion permettant à un second client de reprendre un flux lancé par un premier client.
• Vous souhaitez une exécution propulsée par Inngest, avec mémorisation des étapes, nouvelles tentatives et supervision.

Pour les appels de courte durée limités à une requête, durant lesquels le client reste connecté, un Agent standard utilisant stream() ou generate() est plus simple.

Démarrage rapide

Encapsulez un agent existant avec createDurableAgent() depuis @mastra/core/agent/durable :

src/mastra/agents/researcher.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'researcher',
  name: 'Researcher',
  instructions: 'You research topics thoroughly.',
  model: 'openai/gpt-5.6-sol',
})

export const durableResearcher = createDurableAgent({ agent })

Enregistrez l’agent durable auprès de Mastra, puis appelez stream() :

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { durableResearcher } from './agents/researcher'

const mastra = new Mastra({
  agents: { durableResearcher },
})

const { output, runId, cleanup } = await durableResearcher.stream(
  'Research quantum computing advances in 2025',
)

for await (const chunk of output.fullStream) {
  // Process each chunk as it arrives
}

// Release PubSub subscriptions and clear the run from the registry.
// If you skip this, an automatic cleanup timer fires after the stream ends.
cleanup()

Le runId renvoyé identifie l’exécution. Transmettez-le à observe() pour vous reconnecter depuis un autre client. Consultez la référence de DurableAgent pour découvrir l’ensemble des options de configuration et des méthodes de l’API.

Fonctionnement

Un agent durable ajoute trois couches à un agent standard :

1. Exécution du workflow : stream() sérialise les messages et les options dans l’entrée d’un workflow, puis déclenche la boucle agentique au sein d’un workflow durable. Le workflow exécute la même boucle que Agent.stream(), mais chaque étape peut être mémorisée et rejouée.

2. Diffusion en continu avec PubSub : pendant l’exécution de la boucle, les fragments sont publiés dans un sujet PubSub associé à l’identifiant d’exécution. L’appelant s’abonne à ce sujet et transmet les fragments à un ReadableStream. S’il se déconnecte puis se reconnecte, les fragments manqués sont rejoués depuis le cache.

3. Couche de cache : un cache facultatif (en mémoire par défaut, ou Redis ou un autre système en production) stocke les événements publiés afin qu’un abonné tardif puisse rattraper son retard.

Variantes d’exécution

Mastra fournit trois fonctions de fabrique qui créent des agents durables. Elles se distinguent par leur mode d’exécution du workflow :

Fabrique	Package	Cas d’usage idéal
createDurableAgent()	@mastra/core	Développement local et serveurs à processus unique. Vous obtenez un flux que vous pouvez attendre directement.
createEventedAgent()	@mastra/core	Exécution en arrière-plan. Le workflow démarre sans bloquer et vous consommez les fragments via PubSub.
createInngestAgent()	@mastra/inngest	Déploiements en production. Inngest ajoute la mémorisation des étapes, les nouvelles tentatives et un tableau de bord de supervision.

Toutes trois renvoient un objet que vous enregistrez auprès de Mastra comme un agent standard. createDurableAgent() et createEventedAgent() renvoient des instances de classes qui étendent Agent. createInngestAgent() renvoie un objet reposant sur un Proxy, qui transmet les méthodes d’Agent à l’agent sous-jacent.

Dans le processus avec createDurableAgent()

Encapsulez votre agent et appelez stream(). Vous obtenez un DurableAgentStreamResult dans le même processus. Aucune infrastructure externe n’est nécessaire : c’est donc le moyen le plus rapide de commencer.

src/mastra/agents/durable.ts

import { Agent } from '@mastra/core/agent'
import { createDurableAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'helper',
  instructions: 'You are a helpful assistant.',
  model: 'openai/gpt-5.6-sol',
})

export const durableHelper = createDurableAgent({ agent })

Lancement sans attente avec createEventedAgent()

Le workflow démarre en arrière-plan sans bloquer l’appelant. Vous continuez à recevoir les fragments via PubSub ; stream() renvoie donc un résultat que vous pouvez consommer. Le gestionnaire HTTP qui a déclenché l’exécution n’a pas besoin d’attendre la fin du workflow :

src/mastra/agents/evented.ts

import { Agent } from '@mastra/core/agent'
import { createEventedAgent } from '@mastra/core/agent/durable'

const agent = new Agent({
  id: 'writer',
  instructions: 'You write articles.',
  model: 'openai/gpt-5.6-sol',
})

export const eventedWriter = createEventedAgent({ agent })

Propulsé par Inngest avec createInngestAgent()

Exécutez le workflow sur la plateforme Inngest. Chaque appel d’outil devient une étape mémorisée qu’Inngest peut retenter indépendamment, et vous disposez d’un tableau de bord pour superviser les exécutions :

src/mastra/agents/inngest.ts

import { Agent } from '@mastra/core/agent'
import { createInngestAgent } from '@mastra/inngest'
import { Inngest } from 'inngest'

const inngest = new Inngest({ id: 'my-app' })

const agent = new Agent({
  id: 'analyst',
  instructions: 'You analyze data.',
  model: 'openai/gpt-5.6-sol',
})

export const inngestAnalyst = createInngestAgent({ agent, inngest })

Consultez la référence de createInngestAgent() pour découvrir l’API complète, notamment les options propres à Inngest telles que la configuration de PubSub et du cache.

Flux pouvant être repris

Les agents durables prennent en charge les flux pouvant être repris grâce à PubSub et à un cache d’événements. Lorsqu’un client se déconnecte en cours de flux, le cache continue de stocker les événements. Ce même client peut se reconnecter en appelant observe() avec le runId :

src/server/reconnect.ts

const { output, cleanup } = await durableResearcher.observe(runId)

for await (const chunk of output.fullStream) {
  // Chunks from the run, including any missed while disconnected
}

cleanup()

createDurableAgent() et createEventedAgent() utilisent par défaut un cache en mémoire, ce qui signifie que la reprise des flux fonctionne au sein d’un seul processus. En production, fournissez un système de cache persistant (par exemple Redis) afin que les événements mis en cache résistent aux redémarrages du processus :

src/mastra/agents/durable-with-cache.ts

import { createDurableAgent } from '@mastra/core/agent/durable'
import { RedisServerCache } from '@mastra/redis'
import Redis from 'ioredis'

const cache = new RedisServerCache({ client: new Redis('redis://localhost:6379') })

export const durableAgent = createDurableAgent({
  agent,
  cache,
})

createInngestAgent() n’active pas le cache par défaut. Transmettez une option cache ou enregistrez l’agent auprès d’une instance de Mastra dont le serverCache est configuré afin d’activer la reprise des flux.

Diffusion en continu avec des tâches en arrière-plan

Les agents durables prennent en charge la même option untilIdle que les agents standards. Lorsque untilIdle est défini, stream() maintient la connexion ouverte pendant les reprises des tâches en arrière-plan, jusqu’à ce que l’agent soit inactif :

src/mastra/run.ts

const { output, cleanup } = await durableAgent.stream('Research and summarize the topic', {
  untilIdle: true,
  memory: { thread: 'thread-1', resource: 'user-1' },
})

for await (const chunk of output.fullStream) {
  // Chunks from the initial turn AND any follow-up turns triggered by
  // background task completions
}

cleanup()

Transmettez { maxIdleMs } pour personnaliser le délai d’inactivité (5 minutes par défaut) :

await durableAgent.stream('Research topic', {
  untilIdle: { maxIdleMs: 30_000 },
  memory: { thread: 'thread-1', resource: 'user-1' },
})

Consultez Tâches en arrière-plan pour accéder au guide complet, notamment à la configuration, aux sous-agents ainsi qu’à la suspension et à la reprise.

Nettoyage

Chaque appel à stream() et à observe() renvoie une fonction cleanup. Son appel désabonne de PubSub et supprime l’exécution du registre interne. Si vous oubliez de l’appeler, un minuteur automatique se déclenche après la fin du flux ; toutefois, appeler vous-même cleanup() libère immédiatement les ressources.

Approbation des outils

Les agents durables prennent en charge l’approbation des outils (avec intervention humaine). Lorsqu’un appel d’outil nécessite une approbation, le workflow se suspend, appelle la fonction de rappel onSuspended et attend que l’appelant le reprenne avec resume() :

src/mastra/run.ts

const { output, runId, cleanup } = await durableAgent.stream('Delete the old records', {
  requireToolApproval: true,
  onSuspended: ({ toolCallId, toolName, args }) => {
    // Notify the user and ask for approval
  },
})

Après l’approbation, reprenez l’exécution suspendue :

await durableAgent.resume(runId, { approved: true })

Récupération après un arrêt brutal

Si le processus serveur s’arrête brutalement pendant l’exécution d’un agent durable, celle-ci conserve le statut running dans le stockage, sans nouvelle tentative automatique. Au prochain démarrage du serveur, vous pouvez relancer ces exécutions orphelines afin qu’elles reprennent là où elles s’étaient arrêtées.

Récupération automatique

Définissez recovery.durableAgents sur 'auto' dans la configuration de Mastra. Au démarrage, le système de déploiement appelle recoverAllDurableAgents() juste après avoir redémarré les exécutions actives des workflows :

src/mastra/index.ts

export const mastra = new Mastra({
  agents: { myAgent: durableAgent },
  storage: new PostgresStore({ connectionString: process.env.DATABASE_URL! }),
  recovery: { durableAgents: 'auto' },
})

Au démarrage, ce mécanisme détecte tous les agents durables enregistrés dont des exécutions sont bloquées avec le statut running, puis les relance depuis le dernier instantané conservé.

attention

La récupération réexécute la boucle agentique depuis le dernier instantané, ce qui relance les appels au LLM (avec un coût réel) et les appels d’outils. Assurez-vous que vos outils sont idempotents avant d’activer la récupération automatique.

Récupération manuelle

Si vous avez besoin d’un contrôle plus fin, par exemple pour conditionner la récupération à une élection de leader ou pour l’exécuter selon un calendrier, appelez directement les méthodes :

// Recover all durable agents
const result = await mastra.recoverAllDurableAgents()
console.log(`Recovered ${result.recovered} runs (${result.succeeded} ok, ${result.failed} failed)`)

// Recover a specific agent
const agentResult = await durableAgent.recoverActiveRuns()

// Recover a single known run
await durableAgent.recoverActiveRuns({ runId: 'run-abc-123' })

Déploiements multi-instances

Mastra ne fournit pas encore de bail distribué ni de verrou distribué. Dans les déploiements à plusieurs réplicas, chaque réplica démarrant avec recovery.durableAgents: 'auto' tentera simultanément de récupérer les mêmes exécutions. Pour le moment, conditionnez la récupération à votre propre élection de leader ou lancez-la depuis un seul réplica.

Ressources connexes

• Référence de DurableAgent
• Référence de createInngestAgent()
• Tâches en arrière-plan
• Guide de déploiement avec Inngest
• Présentation des agents
• Présentation des processus de travail